fix(core): show when routing first produces content (#3441)

Co-authored-by: Justin Helmer <justin@coreplane.ai>

Author: justinhelmer

docs/reference/specs/routing-and-config.md

@@ -375,6 +375,7 @@ An accepted non-review operator bind carries its repository separately from the
 | Required saved-context preparation and catalogue reads consume the same operator deadline; attachment preparation cannot renew it before the first model turn. | `[unit]` `src/core/dispatch/operatorReReview.test.ts::required context shares the operator deadline::*` |
 
 | Routing receipts retain bounded phase timing, remaining deadline, provider/model, measured usage and request IDs when reported, redacted rejected questions/proposals and specific proposal validation codes. The phase active at cancellation survives; character-based output counts are labeled estimates. | `[unit]` `src/core/dispatch/operator.test.ts::runOperator — the loop over a scripted model::retains timed rejected proposals and the model call active at the deadline` |
+| A streamed operator completion records its first content time after HTTP headers. A later HTTP attempt clears the prior attempt's marker; observer failure cannot change the answer. The receipt retains no stream content. | `[unit]` `src/core/harness/piAi.test.ts::PiAiProviders — the provider table config.yaml names, built on pi's model library::observes request send and response headers even without a provider request ID`, `src/core/dispatch/route.test.ts::the readers' router is retired::forwards HTTP send and headers observations to the routing recorder`, `src/core/dispatch/operatorTelemetry.test.ts::routing telemetry retention::records the first streamed content after headers and clears it on a later HTTP attempt` |
 | A stalled required-context read retains the phase active at cancellation; a model answer returned after the shared deadline admits no action. Named credentials stay out of telemetry, and late callbacks cannot change sealed evidence. Unicode and step-count bounds preserve the first and terminal phases with explicit loss. | `[unit]` `src/core/dispatch/operator.test.ts::runOperator — the loop over a scripted model::records a stalled context read at cancellation and redacts registry credentials`, `::refuses a valid action returned after the shared deadline`; `src/core/dispatch/operatorTelemetry.test.ts::routing telemetry retention::*` |
 | Each routing tool proposal retains its redacted JSON input and outcome, including all calls in a rejected multi-call answer and every repair. Selection does not claim execution. Grounding helpers retain arguments and returned text. Capped content and omitted evidence are explicit; raw provider bodies and reasoning are excluded. | `[unit]` `src/core/dispatch/operator.test.ts::runOperator — the loop over a scripted model::retains every rejected routing call with its input and validation result`, `::retains helper arguments and results beside the selected action`; `src/core/dispatch/operatorTelemetry.test.ts::routing telemetry retention::*` |
 | Exhausted multi-call recovery uses the ordinary deadline-checked admission path, including recorded context, catalogue and command-fulfillment phases. Late actions and unavailable verdicts admit no execution. | `[unit]` `src/core/dispatch/operator.test.ts::long asks and multi-call answers (issue 2099)::recovered sole actions keep deadline and phase evidence: …`; `src/core/dispatch/operatorFulfillment.test.ts::last-read command correction::recovered commands retain fulfillment evidence and refuse unavailable verdicts: …` |

src/core/dispatch/operatorTelemetry.test.ts

@@ -26,6 +26,35 @@ describe("routing telemetry retention", () => {
     expect(step.responseStatus).toBeUndefined();
   });
 
+  it("records the first streamed content after headers and clears it on a later HTTP attempt", async () => {
+    let clock = 1_000;
+    const recorder = new RoutingRecorder(() => clock, clock, 61_000, new AbortController().signal);
+    await recorder.during("model", async (observe) => {
+      clock = 1_100;
+      observe({ transport: "request_send", transportId: "first" });
+      clock = 1_200;
+      observe({ transport: "response_headers", status: 503 });
+      clock = 1_300;
+      observe({ stream: "first_content" });
+      clock = 1_400;
+      observe({ transport: "request_send", transportId: "second" });
+      clock = 1_500;
+      observe({ transport: "response_headers", status: 200 });
+      clock = 1_700;
+      observe({ stream: "first_content" });
+      clock = 1_800;
+      observe({ stream: "first_content" });
+      return "done";
+    });
+    expect(recorder.snapshot().steps[0]).toMatchObject({
+      transportAttempts: 2,
+      requestSentAt: 1_400,
+      responseHeadersAt: 1_500,
+      firstContentAt: 1_700,
+      finishedAt: 1_800,
+    });
+  });
+
   it.each(["proposed", "selected", "rejected"] as const)(
     "redacts original strings on a %s scripted proposal",
     async (outcome) => {

src/core/dispatch/operatorTelemetry.ts

@@ -107,6 +107,7 @@ export class RoutingRecorder {
         if (value.transportId) step.transportId = safe(value.transportId, 128);
         delete step.requestId;
         delete step.responseHeadersAt;
+        delete step.firstContentAt;
         delete step.responseStatus;
       } else if (
         value.transport === "response_headers" &&
@@ -117,6 +118,8 @@ export class RoutingRecorder {
       ) {
         step.responseHeadersAt = this.now();
         step.responseStatus = value.status;
+      } else if (value.stream === "first_content" && step.responseHeadersAt !== undefined) {
+        step.firstContentAt ??= this.now();
       }
       for (const key of ["provider", "model", "effort", "effortWord", "requestId", "stopReason"] as const) {
         if (value[key] !== undefined) step[key] = safe(value[key], 128);

src/core/dispatch/route.test.ts

@@ -74,6 +74,7 @@ describe("the readers' router is retired", () => {
       complete: async (request) => {
         request.observer?.onRequestStart?.("transport-fixture");
         request.observer?.onResponseHeaders?.(200);
+        request.observer?.onFirstContent?.();
         return { content: [], stopReason: "end_turn" };
       },
     };
@@ -83,6 +84,7 @@ describe("the readers' router is retired", () => {
     );
     expect(observations).toContainEqual({ transport: "request_send", transportId: "transport-fixture" });
     expect(observations).toContainEqual({ transport: "response_headers", status: 200 });
+    expect(observations).toContainEqual({ stream: "first_content" });
   });
 });
 

src/core/dispatch/route.ts

@@ -533,6 +533,7 @@ export function providerStructuredModel(
             observer: {
               onRequestStart: (transportId: string) => observe({ transport: "request_send", transportId }),
               onResponseHeaders: (status: number) => observe({ transport: "response_headers", status }),
+              onFirstContent: () => observe({ stream: "first_content" }),
               onResponse: (requestId: string) => observe({ requestId }),
             },
           }

src/core/harness/piAi.test.ts

@@ -137,14 +137,18 @@ describe("PiAiProviders — the provider table config.yaml names, built on pi's
       observer: {
         onRequestStart: () => events.push("send"),
         onResponseHeaders: (status) => events.push(`headers:${status}`),
+        onFirstContent: () => {
+          events.push("first-content");
+          throw new Error("diagnostic sink failed");
+        },
       },
     });
     await vi.waitFor(() => expect(events).toEqual(["send", "fetch"]));
     returnHeaders(
       new Response(anthropicToolCallStream(), { status: 200, headers: { "content-type": "text/event-stream" } }),
     );
     await expect(completion).resolves.toMatchObject({ stopReason: "tool_use" });
-    expect(events).toEqual(["send", "fetch", "headers:200"]);
+    expect(events).toEqual(["send", "fetch", "headers:200", "first-content"]);
   });
 
   it("observes the provider request ID at response headers without changing the completion", async () => {

src/core/harness/piAi.ts

@@ -260,7 +260,22 @@ export class PiAiProvider implements Provider {
       // Direct API modules take pi's normalized transcript, not the public
       // Context shorthand. Shape tools before moving them into its system message.
       const transcript = normalizeContext(shaped.body as unknown as Context);
-      const message = await this.api$().stream(model, transcript, options).result();
+      const stream = this.api$().stream(model, transcript, options);
+      let firstContent = false;
+      for await (const event of stream) {
+        if (
+          !firstContent &&
+          (event.type === "text_start" || event.type === "thinking_start" || event.type === "toolcall_start")
+        ) {
+          firstContent = true;
+          try {
+            req.observer?.onFirstContent?.();
+          } catch {
+            /* Diagnostics cannot fail a completion. */
+          }
+        }
+      }
+      const message = await stream.result();
       return fromPiMessage(message, this.name, req.tools, failureContext);
     } catch (err) {
       // Pi's `aborted` stop reason is local cancellation, not provider health.

src/core/provider.ts

@@ -66,7 +66,7 @@ export interface CompletionRequest {
    *  it to every breakpoint. Absent → the provider default (`5m`). */
   cacheTtl?: CacheTtl;
   /** Timing hooks for the call's span (docs/reference/specs/tracing.md): a streaming
-   *  provider reports the first token; the span layer stamps the time. A
+   *  provider reports the first content block; the span layer stamps the time. A
    *  provider that cannot observe its stream simply never calls them. */
   observer?: CompletionObserver;
 }
@@ -80,7 +80,7 @@ export interface CompletionObserver {
   onResponse?(requestId: string): void;
   /** One HTTP response, including failures and responses without a request ID. */
   onHttpResponse?(status: number, requestId?: string): void;
-  onFirstToken?(): void;
+  onFirstContent?(): void;
   /** A content block began / ended, by kind (`text`, `thinking`, `redacted_thinking`,
    *  `tool_use`, …) and stream index: the span layer sums a turn's thinking and
    *  writing time from these (docs/reference/specs/tracing.md; live-view item 15). */

src/core/routingTelemetry.ts

@@ -4,6 +4,7 @@ import type { SpanEndEvent } from "./runEvents.js";
 export interface RoutingModelObservation {
   /** Transient provider transport events; the recorder stamps them with its clock. */
   transport?: "request_send" | "response_headers";
+  stream?: "first_content";
   status?: number;
   transportId?: string;
   provider?: string;
@@ -30,7 +31,7 @@ export interface RoutingToolCall {
   result?: string;
 }
 
-export interface RoutingStep extends Omit<RoutingModelObservation, "toolCalls" | "transport" | "status"> {
+export interface RoutingStep extends Omit<RoutingModelObservation, "toolCalls" | "transport" | "stream" | "status"> {
   index: number;
   phase: RoutingPhase;
   startedAt: number;
@@ -42,6 +43,7 @@ export interface RoutingStep extends Omit<RoutingModelObservation, "toolCalls" |
   transportAttempts?: number;
   requestSentAt?: number;
   responseHeadersAt?: number;
+  firstContentAt?: number;
   responseStatus?: number;
   tool?: string;
   question?: string;