mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-07 06:16:06 -10:00
feat(observability): anchor native time to first token at the accepted wake (#19838)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - When someone messages an agent, the reply must start fast. The durable ownership plan (#19831) makes time to first token (TTFT) a hard invariant. > - Phase 0 of that plan records a TTFT baseline per adapter and environment, measured from the moment the message or wake is accepted. > - Today `task.time_to_first_token` starts at the wake comment, the answered question, or the run row's creation. A wake request that was accepted earlier (deferred, coalesced or retried) is not part of the measurement, and the span does not say which start it used. > - This pull request moves the start back to the earliest accepted input and names it on the span. > - The benefit is that the baseline and later budgets measure what the user waits for, with no new work before `turn.start`. ## Linked Issues or Issue Description Refs #19831 **What is the problem?** `task.time_to_first_token` does not include the time between an accepted wake request and the creation of its run row. It also does not record which input it started from. **What is the expected outcome?** The span starts at the earliest accepted input and carries an `anchor` attribute: `comment`, `question_response`, `wake_request` or `run_created`. **Why does it matter?** Phase 0 of the durable ownership plan sets TTFT budgets from this baseline. A start that skips queue time hides the latency that deferred wakes add. ## What Changed - `resolveNativeTimeToFirstTokenStart` (pure) in `native-run-trace.ts` picks the start and the anchor. `anchor` joins the closed span-attribute allowlist. - `readNativeRunWakeRequestedAtMs` reads `agent_wakeup_requests.requested_at` for the run. It never throws. - `native-session-executor.ts` calls it only after the first deliverable text, so it adds no wait before `turn.start`. The span now ends at the text commit time, not after the trace write. - Updated `doc/run-log-events.md` and `doc/observability.md`. ## Verification - `cd server && npx vitest run src/services/native-runtime/native-run-trace.test.ts src/services/native-runtime/native-run-wake-anchor.test.ts` (26 pass). - `cd server && npx vitest run src/services/native-runtime/native-session-executor.test.ts -t first` (41 pass). The existing TTFT test now asserts the anchor for each ingress. ## Risks - Low risk. This changes only the run-log and OTLP span. It is not Telemetry (AGENTS.md §5.7), and it does not change runtime behavior. - One indexed primary-key read per native dispatch, after the first text. A read failure drops the anchor and keeps the old start. - TTFT values can grow for deferred wakes. That is the intended correction. `startOffsetMs` can be negative when the wake request comes before the trace start. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), via Claude Code with tool use and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [ ] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <[email protected]>
This commit is contained in:
1 parent
84908d176b
commit
e2d490c61b
9 files changed
+423
-12
No files matched your search
+11
-4
@@ -746,10 +746,17 @@ accepted ingress → queue → task.prepare → Runner startup → provider work
|
||||
└────────────────────── task.time_to_first_token ──────────────────────────────────┘
|
||||
```
|
||||
|
||||
The span starts at the native run trace's historical start:
|
||||
the accepted comment or attested question response when present, otherwise run
|
||||
creation from the queue span. Callers without preparation history start at native
|
||||
dispatch. The interval includes queueing, Core preparation, workspace preparation,
|
||||
The span starts at the earliest accepted input. The accepted comment or attested
|
||||
question response is the start only when its span is the earliest preparation span.
|
||||
A comment coalesced into an already-queued run starts after run creation, so that
|
||||
run starts at run creation from the queue span and reports `run_created`. The
|
||||
durable wake request's acceptance time replaces the start when it is earlier
|
||||
(deferred, coalesced and retried wakes). The `anchor` attribute (`comment`, `question_response`,
|
||||
`wake_request`, `run_created`) names which one. The wake request time is read
|
||||
after the first text in detached diagnostic work bounded to 2 s (PostgreSQL
|
||||
cancels a stalled read through a transaction-local `statement_timeout`), so it never adds a
|
||||
wait before `turn.start` and never delays event delivery. Callers without
|
||||
preparation history start at native dispatch and report `run_created`. The interval includes queueing, Core preparation, workspace preparation,
|
||||
Runner startup, provider latency, transport, and durable event ingestion. It ends
|
||||
in Core's commit callback, when the text is readable through the run-events API,
|
||||
using the Core clock rather than the provider event timestamp. Browser delivery
|
||||
|
||||
+13
-4
@@ -68,10 +68,19 @@ instance run log and add no first-party Telemetry or OpenTelemetry export.
|
||||
dispatch at the first newly committed deliverable text: assistant or reasoning
|
||||
`item.delta`/`item.completed`, a redaction tail released at completion, or an
|
||||
accepted run-result summary. Its
|
||||
`durationMs` starts at the native run trace's historical start (accepted ingress,
|
||||
otherwise queued run creation, or dispatch when history is absent) and ends at Core
|
||||
commit callback. `startOffsetMs` is zero, and `eventKind` is `agentMessage` or
|
||||
`reasoning`. It includes preparation and startup, but excludes browser
|
||||
`durationMs` starts at the earliest accepted input. The accepted comment or
|
||||
answered question is the start only when its span is the earliest preparation span.
|
||||
A comment coalesced into an already-queued run starts after run creation, so that
|
||||
run uses queued run creation. The durable wake request's `requested_at` replaces
|
||||
the start when it is earlier (deferred, coalesced and retried wakes). Without
|
||||
preparation history, the start is dispatch and the anchor is `run_created`. It ends at Core commit callback. `anchor` names the start:
|
||||
`comment`, `question_response`, `wake_request` or `run_created`. `startOffsetMs` is
|
||||
zero, or negative when a wake request precedes the run trace. `eventKind` is
|
||||
`agentMessage` or `reasoning`. The wake request time is read after the first text,
|
||||
in detached diagnostic work bounded to 2 s (a
|
||||
transaction-local `statement_timeout`, so PostgreSQL cancels a stalled read), so the measurement adds no wait before
|
||||
`turn.start` and never delays event delivery. The span can appear after later
|
||||
run-log rows; the trace settles it before `task.provider_session.measured`. It includes preparation and startup, but excludes browser
|
||||
delivery/rendering. This measures text availability, not an exact model token.
|
||||
Empty text, item starts, tools, usage, result proposals, and duplicate replay do
|
||||
not qualify. No deliverable text means no timing sample. A recovered dispatch can
|
||||
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
nativeRunPreparationStarts,
|
||||
recordFailedSkillPreparation,
|
||||
recordNativeProviderPerformanceSpans,
|
||||
resolveNativeTimeToFirstTokenStart,
|
||||
NATIVE_RUN_SPAN_EVENT_TYPE,
|
||||
NATIVE_RUN_TRACE_SCHEMA_VERSION,
|
||||
} from "./native-run-trace.js";
|
||||
@@ -504,3 +505,102 @@ it.each(["start", "context", "attribute", "status", "end"])("keeps execution and
|
||||
expect(failedWork).toHaveBeenCalledOnce();
|
||||
for (const span of spans) expect(span.end).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
describe("time to first token anchor", () => {
|
||||
const prep = (name: string, startedAtMs: number) => ({
|
||||
name, parentName: "task.provider_session", startedAtMs, endedAtMs: startedAtMs + 5,
|
||||
});
|
||||
|
||||
it("falls back to run creation when no earlier accepted input is known", () => {
|
||||
expect(resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: 1_000, preparationSpans: [prep("heartbeat.queue", 1_000)], wakeRequestedAtMs: null,
|
||||
})).toEqual({ startedAtMs: 1_000, anchor: "run_created" });
|
||||
});
|
||||
|
||||
it("names the comment or answered question that already anchors the trace", () => {
|
||||
expect(resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: 900, preparationSpans: [prep("comment.to_run_created", 900)], wakeRequestedAtMs: 950,
|
||||
})).toEqual({ startedAtMs: 900, anchor: "comment" });
|
||||
expect(resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: 800, preparationSpans: [prep("question_response.to_run_created", 800)], wakeRequestedAtMs: null,
|
||||
})).toEqual({ startedAtMs: 800, anchor: "question_response" });
|
||||
});
|
||||
|
||||
it("names run creation when a comment was coalesced into an already-queued run", () => {
|
||||
expect(resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: 1_000,
|
||||
preparationSpans: [prep("heartbeat.queue", 1_000), prep("comment.to_run_created", 1_400)],
|
||||
wakeRequestedAtMs: null,
|
||||
})).toEqual({ startedAtMs: 1_000, anchor: "run_created" });
|
||||
});
|
||||
|
||||
it("moves back to an earlier accepted wake request (deferred or coalesced wakes)", () => {
|
||||
expect(resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: 5_000, preparationSpans: [prep("heartbeat.queue", 5_000)], wakeRequestedAtMs: 2_000,
|
||||
})).toEqual({ startedAtMs: 2_000, anchor: "wake_request" });
|
||||
});
|
||||
|
||||
it("ignores invalid wake request times", () => {
|
||||
for (const wakeRequestedAtMs of [Number.NaN, -1, 6_000]) {
|
||||
expect(resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: 5_000, preparationSpans: [], wakeRequestedAtMs,
|
||||
})).toEqual({ startedAtMs: 5_000, anchor: "run_created" });
|
||||
}
|
||||
});
|
||||
|
||||
it("persists the anchor on the run-log span", async () => {
|
||||
const events: AdapterRuntimeEvent[] = [];
|
||||
const trace = createNativeRunTrace({
|
||||
runId: "ttft-run", startedAtMs: 5_000,
|
||||
traceContext: createRecordingTraceContext().traceContext,
|
||||
onEvent: async (event) => { events.push(event); },
|
||||
});
|
||||
const start = resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: trace.traceStartedAtMs, preparationSpans: [], wakeRequestedAtMs: 4_000,
|
||||
});
|
||||
await trace.record({
|
||||
name: "task.time_to_first_token", parentName: "task.provider_session",
|
||||
startedAtMs: start.startedAtMs, endedAtMs: 7_000, attributes: { eventKind: "agentMessage", anchor: start.anchor },
|
||||
});
|
||||
const span = events.find((event) => (event.payload as Record<string, unknown>)?.span === "task.time_to_first_token");
|
||||
expect(span?.payload).toMatchObject({ durationMs: 3_000, startOffsetMs: -1_000, anchor: "wake_request", eventKind: "agentMessage" });
|
||||
});
|
||||
});
|
||||
|
||||
describe("deferred diagnostics", () => {
|
||||
it("never blocks the caller and settles before the trace finishes", async () => {
|
||||
const events: AdapterRuntimeEvent[] = [];
|
||||
const trace = createNativeRunTrace({
|
||||
runId: "deferred-run", startedAtMs: 1_000,
|
||||
traceContext: createRecordingTraceContext().traceContext,
|
||||
onEvent: async (event) => { events.push(event); },
|
||||
});
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>((resolve) => { release = resolve; });
|
||||
trace.defer(async () => {
|
||||
await gate;
|
||||
await trace.record({ name: "task.time_to_first_token", parentName: "task.provider_session", startedAtMs: 1_000, endedAtMs: 2_000 });
|
||||
});
|
||||
// Later work keeps flowing while the deferred read is pending.
|
||||
await trace.record({ name: "response.first_nonempty_text", startedAtMs: 2_000, endedAtMs: 2_000 });
|
||||
const span = (name: string) => events.some((event) => (event.payload as Record<string, unknown>)?.span === name);
|
||||
expect(span("response.first_nonempty_text")).toBe(true);
|
||||
expect(span("task.time_to_first_token")).toBe(false);
|
||||
const finished = trace.finish("ok");
|
||||
release();
|
||||
await finished;
|
||||
const names = events.map((event) => (event.payload as Record<string, unknown>)?.span);
|
||||
expect(names.indexOf("task.time_to_first_token")).toBeGreaterThan(-1);
|
||||
expect(names.indexOf("task.time_to_first_token")).toBeLessThan(names.indexOf("task.provider_session.measured"));
|
||||
});
|
||||
|
||||
it("ignores a failed deferred diagnostic and work deferred after finish", async () => {
|
||||
const trace = createNativeRunTrace({ runId: "deferred-fail", traceContext: createRecordingTraceContext().traceContext });
|
||||
trace.defer(async () => { throw new Error("diagnostic failed"); });
|
||||
await expect(trace.finish("ok")).resolves.toBeUndefined();
|
||||
const late = vi.fn(async () => {});
|
||||
trace.defer(late);
|
||||
await Promise.resolve();
|
||||
expect(late).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -106,6 +106,58 @@ export function buildNativeHeartbeatPreparationSpans(input: {
|
||||
];
|
||||
}
|
||||
|
||||
/** Which accepted input `task.time_to_first_token` is measured from. */
|
||||
export type NativeTimeToFirstTokenAnchor =
|
||||
| "question_response"
|
||||
| "comment"
|
||||
| "wake_request"
|
||||
| "run_created";
|
||||
|
||||
/**
|
||||
* Resolve the start of `task.time_to_first_token`: the earliest accepted input
|
||||
* that caused this run. A question answer or a wake comment is already a root
|
||||
* historical span. Otherwise the run row's creation time is the fallback. A
|
||||
* durable wake request accepted earlier (deferred, coalesced or retried wakes)
|
||||
* moves the start back to that request, so queueing before the run row exists
|
||||
* is part of the measured latency.
|
||||
*
|
||||
* Run-log only. The wake request time is read after the first deliverable
|
||||
* text, never before `turn.start`.
|
||||
*/
|
||||
export function resolveNativeTimeToFirstTokenStart(input: {
|
||||
traceStartedAtMs: number;
|
||||
preparationSpans: readonly NativeRunHistoricalSpan[];
|
||||
wakeRequestedAtMs: number | null;
|
||||
}): { startedAtMs: number; anchor: NativeTimeToFirstTokenAnchor } {
|
||||
const ingress = input.preparationSpans.find(
|
||||
(span) =>
|
||||
span.name === "question_response.to_run_created" ||
|
||||
span.name === "comment.to_run_created",
|
||||
);
|
||||
// The trace starts at the earliest preparation span. A comment coalesced
|
||||
// into an already-queued run starts after run creation, so it names the
|
||||
// start only when it is that earliest span.
|
||||
const ingressIsStart = ingress !== undefined && ingress.startedAtMs <= input.traceStartedAtMs;
|
||||
let anchor: NativeTimeToFirstTokenAnchor =
|
||||
!ingressIsStart
|
||||
? "run_created"
|
||||
: ingress.name === "question_response.to_run_created"
|
||||
? "question_response"
|
||||
: "comment";
|
||||
let startedAtMs = input.traceStartedAtMs;
|
||||
const wakeAt = input.wakeRequestedAtMs;
|
||||
if (
|
||||
wakeAt !== null &&
|
||||
Number.isFinite(wakeAt) &&
|
||||
wakeAt >= 0 &&
|
||||
wakeAt < startedAtMs
|
||||
) {
|
||||
startedAtMs = wakeAt;
|
||||
anchor = "wake_request";
|
||||
}
|
||||
return { startedAtMs, anchor };
|
||||
}
|
||||
|
||||
export function nativeRunPreparationStarts(
|
||||
spans: NativeRunHistoricalSpan[],
|
||||
nowMs: number,
|
||||
@@ -149,6 +201,7 @@ const NOOP_SPAN: SpanHandle = {
|
||||
|
||||
const SAFE_NATIVE_RUN_SPAN_ATTRIBUTE_KEYS = new Set([
|
||||
"adapter",
|
||||
"anchor",
|
||||
"bytesTransferred",
|
||||
"inputBytes", "toolSchemaBytes", "toolCount", "contextReused", "tokenDataAvailable",
|
||||
"connectionOwner",
|
||||
@@ -496,7 +549,28 @@ export function createNativeRunTrace(input: {
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Diagnostic work that must not block the caller (for example a read that
|
||||
* anchors a span). It runs detached; `finish` settles it before the root
|
||||
* span closes. The work must bound its own waits.
|
||||
*/
|
||||
const pendingDiagnostics = new Set<Promise<void>>();
|
||||
const defer = (work: () => Promise<void>): void => {
|
||||
if (ended) return;
|
||||
const pending: Promise<void> = Promise.resolve()
|
||||
.then(work)
|
||||
.catch(() => {
|
||||
// Diagnostics cannot change runner control flow.
|
||||
})
|
||||
.finally(() => {
|
||||
pendingDiagnostics.delete(pending);
|
||||
});
|
||||
pendingDiagnostics.add(pending);
|
||||
};
|
||||
|
||||
const finish = async (outcome: NativeRunSpanOutcome): Promise<void> => {
|
||||
if (ended) return;
|
||||
await Promise.all([...pendingDiagnostics]);
|
||||
if (ended) return;
|
||||
ended = true;
|
||||
const endedAtMs = Date.now();
|
||||
@@ -528,6 +602,7 @@ export function createNativeRunTrace(input: {
|
||||
run,
|
||||
record,
|
||||
measure,
|
||||
defer,
|
||||
finish,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { agentWakeupRequests, agents, companies, createDb, heartbeatRuns } from "@paperclipai/db";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
} from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import { readNativeRunWakeRequestedAtMs } from "./native-run-wake-anchor.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = support.supported ? describe : describe.skip;
|
||||
|
||||
describeEmbeddedPostgres("native run wake anchor against PostgreSQL", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let locker!: ReturnType<typeof createDb>;
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const wakeupRequestId = randomUUID();
|
||||
const runId = randomUUID();
|
||||
const requestedAt = new Date("2026-11-07T10:00:00.000Z");
|
||||
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("paperclip-wake-anchor-");
|
||||
// One connection: a read left running after the timeout would starve the next query.
|
||||
db = createDb(database.connectionString, { maxConnections: 1 });
|
||||
locker = createDb(database.connectionString, { maxConnections: 2 });
|
||||
await db.insert(companies).values({ id: companyId, name: "Anchor", issuePrefix: `A${companyId.slice(0, 6)}` });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Anchor agent", role: "engineer", status: "idle", adapterType: "codex_local" });
|
||||
await db.insert(agentWakeupRequests).values({ id: wakeupRequestId, companyId, agentId, source: "assignment", status: "claimed", requestedAt });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, wakeupRequestId, status: "running" });
|
||||
}, 60_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await database?.cleanup();
|
||||
});
|
||||
|
||||
it("reads requested_at through the run's wake request", async () => {
|
||||
await expect(readNativeRunWakeRequestedAtMs(db, runId)).resolves.toBe(requestedAt.getTime());
|
||||
await expect(readNativeRunWakeRequestedAtMs(db, randomUUID())).resolves.toBeNull();
|
||||
});
|
||||
|
||||
it("cancels a blocked read on the server and returns its connection to the pool", async () => {
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>((resolve) => { release = resolve; });
|
||||
let locked!: () => void;
|
||||
const lockHeld = new Promise<void>((resolve) => { locked = resolve; });
|
||||
const holder = locker.transaction(async (tx) => {
|
||||
await tx.execute(sql`lock table agent_wakeup_requests in access exclusive mode`);
|
||||
locked();
|
||||
await gate;
|
||||
});
|
||||
try {
|
||||
await lockHeld;
|
||||
const startedAt = Date.now();
|
||||
await expect(readNativeRunWakeRequestedAtMs(db, runId, 300)).resolves.toBeNull();
|
||||
expect(Date.now() - startedAt).toBeLessThan(2_000);
|
||||
// The lock is still held. The single pooled connection must already be free,
|
||||
// and PostgreSQL must not still be waiting on the anchor read.
|
||||
const probeStartedAt = Date.now();
|
||||
const [probe] = await db.execute<{ ok: number }>(sql`select 1 as ok`);
|
||||
expect(Number(probe?.ok)).toBe(1);
|
||||
expect(Date.now() - probeStartedAt).toBeLessThan(1_000);
|
||||
const waiting = await locker.execute<{ count: string }>(sql`
|
||||
select count(*)::text as count from pg_stat_activity
|
||||
where state = 'active' and wait_event_type = 'Lock' and query ilike '%agent_wakeup_requests%'`);
|
||||
expect(Number(waiting[0]?.count ?? 0)).toBe(0);
|
||||
} finally {
|
||||
release();
|
||||
await holder;
|
||||
}
|
||||
}, 20_000);
|
||||
});
|
||||
@@ -0,0 +1,39 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
|
||||
import { readNativeRunWakeRequestedAtMs } from "./native-run-wake-anchor.js";
|
||||
|
||||
function fakeDb(result: () => Promise<unknown[]>, statements: string[] = []): Db {
|
||||
const chain = {
|
||||
select: () => chain, from: () => chain, innerJoin: () => chain, where: () => chain,
|
||||
limit: () => result(),
|
||||
execute: async (query: { queryChunks?: unknown[] }) => { statements.push(JSON.stringify(query.queryChunks ?? query)); return []; },
|
||||
};
|
||||
return { transaction: async (fn: (tx: typeof chain) => Promise<unknown>) => fn(chain) } as unknown as Db;
|
||||
}
|
||||
|
||||
describe("readNativeRunWakeRequestedAtMs", () => {
|
||||
it("returns the wake request acceptance time under a transaction-local statement timeout", async () => {
|
||||
const at = new Date("2026-11-07T10:00:00.000Z");
|
||||
const statements: string[] = [];
|
||||
await expect(readNativeRunWakeRequestedAtMs(fakeDb(async () => [{ requestedAt: at }], statements), "run", 250))
|
||||
.resolves.toBe(at.getTime());
|
||||
expect(statements.join(" ")).toContain("statement_timeout");
|
||||
expect(statements.join(" ")).toContain("250ms");
|
||||
});
|
||||
|
||||
it("returns null when the run has no wake request", async () => {
|
||||
await expect(readNativeRunWakeRequestedAtMs(fakeDb(async () => []), "run")).resolves.toBeNull();
|
||||
});
|
||||
|
||||
it("stops waiting at the timeout even before the database answers", async () => {
|
||||
const pending = new Promise<unknown[]>(() => {});
|
||||
const startedAt = Date.now();
|
||||
await expect(readNativeRunWakeRequestedAtMs(fakeDb(() => pending), "run", 20)).resolves.toBeNull();
|
||||
expect(Date.now() - startedAt).toBeLessThan(1_000);
|
||||
});
|
||||
|
||||
it("never throws: a read failure only drops the diagnostic anchor", async () => {
|
||||
await expect(readNativeRunWakeRequestedAtMs(fakeDb(async () => { throw new Error("db down"); }), "run")).resolves.toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,56 @@
|
||||
import { agentWakeupRequests, heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
|
||||
/** A slow database never holds this diagnostic longer than this. */
|
||||
export const NATIVE_RUN_WAKE_ANCHOR_READ_TIMEOUT_MS = 2_000;
|
||||
|
||||
/**
|
||||
* When the durable wake request that created this run was accepted, in epoch
|
||||
* milliseconds. Diagnostic only: it anchors `task.time_to_first_token` to the
|
||||
* accepted input. Callers run it detached after the first deliverable text, so
|
||||
* it never adds a wait before `turn.start` or to event delivery.
|
||||
*
|
||||
* The read is bounded on both sides. PostgreSQL cancels the statement at
|
||||
* `timeoutMs` (a transaction-local `statement_timeout`, which also covers lock
|
||||
* waits), so a stalled read does not keep a shared connection busy after the
|
||||
* caller stops waiting. The caller also stops waiting at `timeoutMs`, which
|
||||
* covers a wait for a pool connection. Any failure returns null.
|
||||
*/
|
||||
export async function readNativeRunWakeRequestedAtMs(
|
||||
db: Db,
|
||||
runId: string,
|
||||
timeoutMs = NATIVE_RUN_WAKE_ANCHOR_READ_TIMEOUT_MS,
|
||||
): Promise<number | null> {
|
||||
const boundedMs = Math.max(1, Math.floor(timeoutMs));
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
const timeout = new Promise<null>((resolve) => {
|
||||
timer = setTimeout(() => resolve(null), boundedMs);
|
||||
timer.unref?.();
|
||||
});
|
||||
try {
|
||||
return await Promise.race([readWakeRequestedAtMs(db, runId, boundedMs), timeout]);
|
||||
} finally {
|
||||
if (timer) clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
async function readWakeRequestedAtMs(db: Db, runId: string, timeoutMs: number): Promise<number | null> {
|
||||
try {
|
||||
return await db.transaction(async (tx) => {
|
||||
await tx.execute(sql`select set_config('statement_timeout', ${`${timeoutMs}ms`}, true)`);
|
||||
const [row] = await tx
|
||||
.select({ requestedAt: agentWakeupRequests.requestedAt })
|
||||
.from(heartbeatRuns)
|
||||
.innerJoin(
|
||||
agentWakeupRequests,
|
||||
eq(agentWakeupRequests.id, heartbeatRuns.wakeupRequestId),
|
||||
)
|
||||
.where(eq(heartbeatRuns.id, runId))
|
||||
.limit(1);
|
||||
const at = row?.requestedAt?.getTime();
|
||||
return typeof at === "number" && Number.isFinite(at) ? at : null;
|
||||
});
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
@@ -139,6 +139,9 @@ vi.mock("../usage-receipts.js", async (importOriginal) => ({
|
||||
createRunUsageRecorder: usageRecorder.create,
|
||||
}));
|
||||
|
||||
const wakeAnchor = vi.hoisted(() => ({ read: vi.fn(async (_db: unknown, _runId: string): Promise<number | null> => null) }));
|
||||
vi.mock("./native-run-wake-anchor.js", () => ({ readNativeRunWakeRequestedAtMs: wakeAnchor.read }));
|
||||
|
||||
vi.mock("./native-usage-delivery.js", () => ({
|
||||
createNativeUsageDelivery: vi.fn(async () => ({ capture: async () => undefined })),
|
||||
}));
|
||||
@@ -8826,7 +8829,9 @@ describe("native resumed preparation timing", () => {
|
||||
const ttft = events.filter((event) => event.payload?.span === "task.time_to_first_token");
|
||||
expect(ttft).toHaveLength(1);
|
||||
expect(ttft[0]?.payload).toMatchObject({ parentSpan: "task.provider_session", startOffsetMs: 0,
|
||||
durationMs: ingress === "native_dispatch" ? 500 : 4_500, eventKind: "agentMessage", outcome: "ok" });
|
||||
durationMs: ingress === "native_dispatch" ? 500 : 4_500, eventKind: "agentMessage", outcome: "ok",
|
||||
anchor: ingress === "comment.to_run_created" ? "comment"
|
||||
: ingress === "question_response.to_run_created" ? "question_response" : "run_created" });
|
||||
expect(events.find((event) => event.payload?.span === "provider.time_to_first_agent_event")?.payload)
|
||||
.toMatchObject({ parentSpan: "agent.turn", eventKind: "agentMessage" });
|
||||
expect(JSON.stringify(ttft)).not.toContain("first text");
|
||||
@@ -8834,6 +8839,41 @@ describe("native resumed preparation timing", () => {
|
||||
},
|
||||
);
|
||||
|
||||
it("anchors first-token timing at an earlier durable wake request without blocking later events", async () => {
|
||||
const startedAtMs = Date.now() - 5_000;
|
||||
const events: AdapterRuntimeEvent[] = [];
|
||||
const clock = vi.spyOn(Date, "now").mockReturnValue(startedAtMs + 4_000);
|
||||
let answerRead!: (value: number) => void;
|
||||
wakeAnchor.read.mockImplementationOnce(() => new Promise<number>((resolve) => { answerRead = resolve; }));
|
||||
const event = (eventType: string, payload: Record<string, unknown>) => ({
|
||||
eventType, payload, itemId: "message", emittedAt: new Date(startedAtMs + 2_000).toISOString(),
|
||||
}) as PrpEvent;
|
||||
state.execute.mockReset().mockImplementationOnce(async () => {
|
||||
for (const candidate of [event("turn.submitted", {}), event("turn.started", {})]) await accountingEvents.committed!(candidate);
|
||||
clock.mockReturnValue(startedAtMs + 4_500);
|
||||
await accountingEvents.committed!(event("item.delta", { kind: "agentMessage", text: "first text" }));
|
||||
// The anchor read is still pending; later events are not held behind it.
|
||||
await accountingEvents.committed!(event("item.delta", { kind: "agentMessage", text: "later text" }));
|
||||
expect(events.some((candidate) => candidate.payload?.span === "task.time_to_first_token")).toBe(false);
|
||||
answerRead(startedAtMs - 2_000);
|
||||
return { result: { summary: "cancelled" }, terminal: { runTerminalState: "cancelled" },
|
||||
turnId: "turn", normalizedSessionId: "session", providerSessionId: null,
|
||||
driverKind: "test", driverVersion: "1", nativeEventCount: 1, highestContiguousSourceSeq: 1 };
|
||||
});
|
||||
try {
|
||||
await executePaperclipNativeSession({
|
||||
db: leaseDb(), execution, runnerInstanceId: "runner",
|
||||
preparationSpans: [{ name: "heartbeat.queue", startedAtMs, endedAtMs: startedAtMs + 100 }],
|
||||
onEvent: async (event) => { events.push(event); },
|
||||
});
|
||||
} finally { clock.mockRestore(); }
|
||||
expect(wakeAnchor.read).toHaveBeenCalledWith(expect.anything(), execution.binding.runId);
|
||||
const ttft = events.filter((event) => event.payload?.span === "task.time_to_first_token");
|
||||
expect(ttft).toHaveLength(1);
|
||||
expect(ttft[0]?.payload).toMatchObject({ parentSpan: "task.provider_session", startOffsetMs: -2_000,
|
||||
durationMs: 6_500, anchor: "wake_request", eventKind: "agentMessage" });
|
||||
});
|
||||
|
||||
it.each([
|
||||
["item.delta", { kind: "reasoning", text: "thinking" }, "reasoning"],
|
||||
["item.completed", { kind: "reasoning", text: "thought" }, "reasoning"],
|
||||
|
||||
@@ -284,11 +284,13 @@ import {
|
||||
isNativeRunRootHistoricalSpan,
|
||||
nativeRunPreparationStarts,
|
||||
recordNativeProviderPerformanceSpans,
|
||||
resolveNativeTimeToFirstTokenStart,
|
||||
type NativeRunHistoricalSpan,
|
||||
type NativeRunSpanScope,
|
||||
type NativeRunTrace,
|
||||
} from "./native-run-trace.js";
|
||||
import { createNativeHarnessBackupStamp } from "./native-harness-backup-stamp.js";
|
||||
import { readNativeRunWakeRequestedAtMs } from "./native-run-wake-anchor.js";
|
||||
import { noteNativeRunnerEvidence } from "./native-runner-evidence-signal.js";
|
||||
import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js";
|
||||
import { registerLiveRunnerGoalController } from "../runner-goal-control-broker.js";
|
||||
@@ -10206,12 +10208,21 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
firstTokenRecorded = true;
|
||||
const textAt = Date.now();
|
||||
await trace.record({name:"response.first_nonempty_text",parentName:"native.session.execute",startedAtMs:textAt,endedAtMs:textAt});
|
||||
// Detached: the anchor read never delays this event, later events,
|
||||
// or turn.start (invariant 8). The trace settles it at finish.
|
||||
trace.defer(async () => {
|
||||
const ttftStart = resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: trace.traceStartedAtMs,
|
||||
preparationSpans,
|
||||
wakeRequestedAtMs: await readNativeRunWakeRequestedAtMs(input.db, input.execution.binding.runId),
|
||||
});
|
||||
await trace.record({
|
||||
name: "task.time_to_first_token",
|
||||
parentName: "task.provider_session",
|
||||
startedAtMs: trace.traceStartedAtMs,
|
||||
endedAtMs: Date.now(),
|
||||
attributes: { eventKind: textKind },
|
||||
startedAtMs: ttftStart.startedAtMs,
|
||||
endedAtMs: textAt,
|
||||
attributes: { eventKind: textKind, anchor: ttftStart.anchor },
|
||||
});
|
||||
});
|
||||
}
|
||||
await observeAccountingEvent(event);
|
||||
|
||||
Reference in new issue
Block a user