diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 161edb0bae5..217e785025a 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -68,6 +68,10 @@ Every `run.performance.span` payload also carries `wallStartedAtMs`, the span's start time in epoch milliseconds (since 2026-11-09). Span rows can be written late through deferred diagnostics, so readers take a span's start from this field (or from the trace start plus `startOffsetMs`), never from the row's write time. +The preparation spans of a native run (`heartbeat.queue`, `heartbeat.prepare_*`, +`environment.*`) are always written this way: in the order they ended, while the +provider starts, and before the run's trace finishes. They can appear after the +first Runner events of the run. `run.performance.span` also records `task.time_to_first_token` once per native dispatch at the first newly committed deliverable text: assistant or reasoning 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 1cfb22cddcd..bd448269e37 100644 --- a/server/src/services/native-runtime/native-run-trace.test.ts +++ b/server/src/services/native-runtime/native-run-trace.test.ts @@ -595,6 +595,44 @@ describe("deferred diagnostics", () => { expect(names.indexOf("task.time_to_first_token")).toBeLessThan(names.indexOf("task.provider_session.measured")); }); + it("records spans without waiting on deferred run-log rows, then writes them in order before finish", async () => { + const { traceContext, spans } = createRecordingTraceContext(); + const written: unknown[] = []; + let release!: () => void; + const gate = new Promise((resolve) => { release = resolve; }); + const write = vi.fn(async (event: AdapterRuntimeEvent) => { + await gate; + written.push((event.payload as Record)?.span); + }); + const trace = createNativeRunTrace({ runId: "deferred-rows", startedAtMs: 1_000, traceContext, onEvent: write }); + trace.start("task.prepare", { parentName: "task.provider_session", startedAtMs: 1_000 }); + const names = ["heartbeat.queue", "heartbeat.prepare_before_environment", "heartbeat.prepare_after_environment"]; + for (const [index, name] of names.entries()) { + await trace.record({ name, parentName: "task.prepare", startedAtMs: 1_000 + index, endedAtMs: 1_100 + index }, { deferRunLog: true }); + } + // The spans exist under their parent while every row write is still blocked. + expect(spans.filter((span) => span.parentName === "task.prepare").map((span) => span.name)).toEqual(names); + expect(spans.filter((span) => names.includes(span.name)).every((span) => span.endedAtMs !== null)).toBe(true); + expect(written).toEqual([]); + expect(write.mock.calls.length).toBeLessThanOrEqual(1); + const finished = trace.finish("ok"); + release(); + await finished; + // Rows keep the order their spans ended in and land before the root span closes. + expect(written.slice(0, 3)).toEqual(names); + expect(written.indexOf("task.provider_session.measured")).toBeGreaterThan(2); + }); + + it("drops deferred run-log rows requested after the trace finished", async () => { + const write = vi.fn(async () => {}); + const trace = createNativeRunTrace({ runId: "deferred-rows-late", traceContext: createRecordingTraceContext().traceContext, onEvent: write }); + await trace.finish("ok"); + write.mockClear(); + await trace.record({ name: "heartbeat.queue", startedAtMs: 1_000, endedAtMs: 1_100 }, { deferRunLog: true }); + await new Promise((resolve) => setImmediate(resolve)); + expect(write).not.toHaveBeenCalled(); + }); + 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"); }); diff --git a/server/src/services/native-runtime/native-run-trace.ts b/server/src/services/native-runtime/native-run-trace.ts index 9224fd9dfee..6267e985f32 100644 --- a/server/src/services/native-runtime/native-run-trace.ts +++ b/server/src/services/native-runtime/native-run-trace.ts @@ -531,6 +531,8 @@ export function createNativeRunTrace(input: { endedAtMs?: number; outcome?: NativeRunSpanOutcome; attributes?: Record; + /** Write the run-log row after the caller continues; `finish` settles it. */ + deferRunLog?: boolean; } = {}, ): Promise => { const state = scopeState(scope); @@ -550,14 +552,18 @@ export function createNativeRunTrace(input: { }); if (outcome === "failed") { try { state.span.setStatus({ code: 2 }); } catch {} } try { state.span.end(endedAtMs); } catch {} - if (state.persistRunLog) await emitCompletedSpanEvent({ - name: state.name, - parentName: state.parentName, - startedAtMs: state.startedAtMs, - endedAtMs, - outcome, - attributes: state.attributes, - }); + if (state.persistRunLog) { + const completed: NativeRunHistoricalSpan = { + name: state.name, + parentName: state.parentName, + startedAtMs: state.startedAtMs, + endedAtMs, + outcome, + attributes: state.attributes, + }; + if (options.deferRunLog) deferRunLogRow(completed); + else await emitCompletedSpanEvent(completed); + } checkSpanBudget(state.name, state.startedAtMs, endedAtMs); }; @@ -596,7 +602,7 @@ export function createNativeRunTrace(input: { ); }; - const record = async (span: NativeRunHistoricalSpan, options: { persistRunLog?: boolean } = {}): Promise => { + const record = async (span: NativeRunHistoricalSpan, options: { persistRunLog?: boolean; deferRunLog?: boolean } = {}): Promise => { const scope = start(span.name, { parentName: span.parentName, startedAtMs: span.startedAtMs, @@ -606,6 +612,7 @@ export function createNativeRunTrace(input: { await end(scope, { endedAtMs: span.endedAtMs, outcome: span.outcome, + deferRunLog: options.deferRunLog, }); }; @@ -648,6 +655,16 @@ export function createNativeRunTrace(input: { pendingDiagnostics.add(pending); }; + // Deferred run-log rows are written one at a time, in the order their spans + // ended, so the sink sees the same sequence as the awaited path. + let runLogRows: Promise = Promise.resolve(); + const deferRunLogRow = (span: NativeRunHistoricalSpan): void => { + if (ended) return; + const written = runLogRows.then(() => emitCompletedSpanEvent(span)); + runLogRows = written; + defer(() => written); + }; + const finish = async (outcome: NativeRunSpanOutcome): Promise => { if (ended) return; try { input.onFinish?.(); } catch { /* Diagnostics cannot change runner control flow. */ } diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 729845d8a14..b6af62a30fc 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -2510,7 +2510,9 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); backend: { descriptor, openSession }, preparationSpans: [{ name: "heartbeat.prepare_after_environment", startedAtMs: Date.now(), endedAtMs: Date.now() }], onEvent: async event => { - if (event.payload?.span === "heartbeat.prepare_after_environment") { + // Preparation span rows are written in the background, so hold the + // controller on the awaited environment receipt that precedes them. + if (event.eventType === "run.startup.reuse" && event.payload?.layer === "environment") { prepared(); await continuePreparation; } diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index f0e9a0e335f..25eba55d801 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -9773,6 +9773,8 @@ async function executePaperclipNativeSessionAttempt( startedAtMs: environmentStartedAtMs, }) : null; + // The spans are recorded now; their run-log rows are written in the + // background so the provider start does not wait for one insert per span. await timeController("span_persistence", async () => { for (const span of preparationSpans) { const rootMilestone = isNativeRunRootHistoricalSpan(span.name); @@ -9783,10 +9785,10 @@ async function executePaperclipNativeSessionAttempt( : environmentSpans.includes(span) ? "environment.startup" : "task.prepare", - }); + }, { deferRunLog: true }); } if (environmentScope) { - await trace.end(environmentScope, { endedAtMs: environmentEndedAtMs }); + await trace.end(environmentScope, { endedAtMs: environmentEndedAtMs, deferRunLog: true }); } }); let retainedTransition: VerifiedWarmTransitionBinding | undefined;