diff --git a/doc/observability.md b/doc/observability.md index 7876651e0e..ff80fe858e 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -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 diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 495346d310..ff1d7150c4 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -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 diff --git a/server/src/services/native-runtime/native-run-trace.test.ts b/server/src/services/native-runtime/native-run-trace.test.ts index a195cc28d6..0320f0f465 100644 --- a/server/src/services/native-runtime/native-run-trace.test.ts +++ b/server/src/services/native-runtime/native-run-trace.test.ts @@ -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)?.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((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)?.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)?.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(); + }); +}); diff --git a/server/src/services/native-runtime/native-run-trace.ts b/server/src/services/native-runtime/native-run-trace.ts index faef91b5f3..3fccc1c9b5 100644 --- a/server/src/services/native-runtime/native-run-trace.ts +++ b/server/src/services/native-runtime/native-run-trace.ts @@ -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>(); + const defer = (work: () => Promise): void => { + if (ended) return; + const pending: Promise = Promise.resolve() + .then(work) + .catch(() => { + // Diagnostics cannot change runner control flow. + }) + .finally(() => { + pendingDiagnostics.delete(pending); + }); + pendingDiagnostics.add(pending); + }; + const finish = async (outcome: NativeRunSpanOutcome): Promise => { + 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, }; } diff --git a/server/src/services/native-runtime/native-run-wake-anchor.integration.test.ts b/server/src/services/native-runtime/native-run-wake-anchor.integration.test.ts new file mode 100644 index 0000000000..19da74a06c --- /dev/null +++ b/server/src/services/native-runtime/native-run-wake-anchor.integration.test.ts @@ -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> | null = null; + let db!: ReturnType; + let locker!: ReturnType; + 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((resolve) => { release = resolve; }); + let locked!: () => void; + const lockHeld = new Promise((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); +}); diff --git a/server/src/services/native-runtime/native-run-wake-anchor.test.ts b/server/src/services/native-runtime/native-run-wake-anchor.test.ts new file mode 100644 index 0000000000..fd81317aa9 --- /dev/null +++ b/server/src/services/native-runtime/native-run-wake-anchor.test.ts @@ -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, 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) => 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(() => {}); + 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(); + }); +}); diff --git a/server/src/services/native-runtime/native-run-wake-anchor.ts b/server/src/services/native-runtime/native-run-wake-anchor.ts new file mode 100644 index 0000000000..c0d7e64f30 --- /dev/null +++ b/server/src/services/native-runtime/native-run-wake-anchor.ts @@ -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 { + const boundedMs = Math.max(1, Math.floor(timeoutMs)); + let timer: ReturnType | undefined; + const timeout = new Promise((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 { + 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; + } +} diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 09108ef78c..94f79b08a9 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -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 => 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((resolve) => { answerRead = resolve; })); + const event = (eventType: string, payload: Record) => ({ + 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"], diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index c18f97a92c..4e12dbc710 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -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}); - await trace.record({ - name: "task.time_to_first_token", - parentName: "task.provider_session", - startedAtMs: trace.traceStartedAtMs, - endedAtMs: Date.now(), - attributes: { eventKind: textKind }, + // 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: ttftStart.startedAtMs, + endedAtMs: textAt, + attributes: { eventKind: textKind, anchor: ttftStart.anchor }, + }); }); } await observeAccountingEvent(event);