Attribute failed turns from runtime source events

Author: TheGreatAxios

docs/TELEMETRY.md

@@ -114,11 +114,11 @@ Pre-progress operator aborts settle with `status=cancelled` and
 keeps a worker resumable settles with `status=interrupted` and the same
 `stop_reason=cancelled`; terminal events never report a still-running status.
 
-A deterministic representative fixture uses 10 parent turns with 80 parent tool
-calls and 4 workers totaling 24 turns and 96 tool calls. The former per-call and
-worker-generation shape is 218 billable events; the default aggregate shape is
-18 (10 generations plus 4 start/end pairs), a 91.7% reduction. This is a test
-fixture, not a claim about production PostHog traffic.
+A deterministic synthetic fixture captures 10 parent generations, 80 parent
+tool spans, and 4 worker start/end pairs. The comparable former shape is 98
+billable events; the default aggregate shape is 18 (10 generations plus 8
+start/end events), removing 80 of 98 events, or 81.6%. This is synthetic test
+evidence, not a claim about production PostHog traffic.
 
 Successful `$ai_generation` events may be sampled with
 `CORBITS_TELEMETRY_GENERATION_SAMPLE_RATE` (a float in `0`–`1`, default `1.0`
@@ -165,10 +165,11 @@ Terminal generation settlement belongs to `src/session/run-sink.ts`.
 `inference.error` records only a pending attempt failure: retry success or a
 completed message run discards it. `inference.done` settles success and only
 then applies successful-generation sampling. A failed `message.run.ended`
-settles an unresolved turn once as an unsampled terminal failure, attributed to
-the provider/model snapshot from the latest `inference.start`. Therefore a
-parent turn emits at most one terminal `$ai_generation`, including retry and
-failover paths.
+settles an unresolved turn once as an unsampled terminal failure. Attribution
+uses the latest `inference.usage` source, the first lifecycle payload carrying
+the runtime-resolved provider/model pair for an attempt; it does not infer
+fallback from the externally selected source. Therefore a parent turn emits at
+most one terminal `$ai_generation`, including retry and failover paths.
 
 ## What's never collected
 

src/session/run-sink.test.ts

@@ -150,11 +150,11 @@ describe("createRunSink", () => {
       flush: async () => {},
       discard: () => {},
     };
-    let source = { provider: "provider-a", model: "model-a" };
+    const selectedSource = { provider: "provider-a", model: "model-a" };
     const observer = createTurnObserver({
       telemetry: () => telemetry,
       getSessionId: () => "session-1",
-      getSource: () => source,
+      getSource: () => selectedSource,
     });
     const runSink = createRunSink({
       emitter: new EventEmitter(),
@@ -164,10 +164,14 @@ describe("createRunSink", () => {
 
     runSink.sink(event("inference.start", { model: "model-a" }));
     runSink.sink(event("inference.error", { error: { message: "attempt a failed" } }));
-    source = { provider: "provider-b", model: "model-b" };
     runSink.sink(event("inference.start", { model: "model-b" }));
+    runSink.sink(
+      event("inference.usage", {
+        usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, thinking: 0 },
+        source: { sourceId: "fallback", provider: "provider-b", model: "model-b" },
+      }),
+    );
     runSink.sink(event("inference.error", { error: { message: "attempt b failed" } }));
-    source = { provider: "provider-a", model: "model-a" };
     runSink.sink(
       event("message.run.ended", {
         messageRunId: "run-1",

src/session/run-sink.ts

@@ -1,6 +1,6 @@
 import type { EventEmitter } from "node:events";
 import type { ReactorEmittedEvent } from "@intx/inference";
-import type { TokenUsage } from "@intx/types/runtime";
+import type { LastCycleSource, TokenUsage } from "@intx/types/runtime";
 import { createPerfReactorObserver } from "../perf/reactor-spans.js";
 import { onTurnBoundary } from "../agent/reactor-events.js";
 import { createTurnContextCollector, type LifecycleHookManager, type RunSummary } from "./hooks.js";
@@ -26,6 +26,10 @@ export interface RunSinkArgs {
   // current count (the in-flight turn that has not completed yet) — used to
   // stamp parent_trace_id on subagent_end while tools still run.
   onTurnStarted?: (info: { turnIndex: number }) => void;
+  // inference.usage is the first attempt event carrying the runtime-resolved
+  // provider/model pair. It remains authoritative even when the selected source
+  // outside the reactor has not changed during fallback.
+  onTurnSourceObserved?: (info: { turnIndex: number; source: LastCycleSource }) => void;
   // Continues a resumed session's persisted run.json turn count instead of
   // restarting the collector at zero.
   initialTurnCount?: number;
@@ -94,6 +98,7 @@ export function createRunSink(args: RunSinkArgs): RunSink {
     onTurnComplete,
     onTurnFailed,
     onTurnStarted,
+    onTurnSourceObserved,
     initialTurnCount,
     onTurnBoundarySnapshot,
   } = args;
@@ -147,6 +152,12 @@ export function createRunSink(args: RunSinkArgs): RunSink {
       turnInFlight = true;
       onTurnStarted?.({ turnIndex: turnCollector.getTurnCount() });
     }
+    if (event.type === "inference.usage") {
+      onTurnSourceObserved?.({
+        turnIndex: turnCollector.getTurnCount(),
+        source: event.data.source,
+      });
+    }
 
     if (event.type === "reactor.done") {
       runCompleted = true;

src/telemetry/ai-observability.test.ts

@@ -146,21 +146,28 @@ describe("turnTraceId", () => {
 });
 
 describe("representative fleet event volume", () => {
-  test("reduces deterministic billable events by at least 80 percent", () => {
-    const parentTurns = 10;
-    const parentToolCalls = 80;
-    const workers = 4;
-    const workerTurns = 24;
-    const workerToolCalls = 96;
-    const oldBillableEvents =
-      parentTurns + parentToolCalls + workerTurns + workerToolCalls + workers * 2;
-    const newBillableEvents = parentTurns + workers * 2;
-
-    expect({ oldBillableEvents, newBillableEvents }).toEqual({
-      oldBillableEvents: 218,
-      newBillableEvents: 18,
-    });
-    expect(1 - newBillableEvents / oldBillableEvents).toBeGreaterThanOrEqual(0.8);
+  test("reduces deterministic synthetic billable events by at least 80 percent", () => {
+    const captureFixture = (includeToolSpans: boolean): string[] => {
+      const captured: string[] = [];
+      for (let turn = 0; turn < 10; turn++) captured.push("$ai_generation");
+      if (includeToolSpans) {
+        for (let toolCall = 0; toolCall < 80; toolCall++) captured.push("$ai_span");
+      }
+      for (let worker = 0; worker < 4; worker++) {
+        captured.push("subagent_start", "subagent_end");
+      }
+      return captured;
+    };
+
+    const oldCaptured = captureFixture(true);
+    const newCaptured = captureFixture(false);
+    const reduction = 1 - newCaptured.length / oldCaptured.length;
+
+    expect(oldCaptured).toHaveLength(98);
+    expect(newCaptured).toHaveLength(18);
+    expect(oldCaptured.length - newCaptured.length).toBe(80);
+    expect(reduction).toBeCloseTo(80 / 98, 6);
+    expect(reduction).toBeGreaterThanOrEqual(0.8);
   });
 });
 

src/telemetry/ai-observability.ts

@@ -229,15 +229,18 @@ export interface CreateTurnObserverOptions {
 // one place instead of at each call site.
 export function createTurnObserver(options: CreateTurnObserverOptions): {
   onTurnStarted: (info: { turnIndex: number }) => void;
+  onTurnSourceObserved: (info: { turnIndex: number; source: TurnSource }) => void;
   onTurnComplete: (ctx: TurnContext) => void;
   onTurnFailed: (info: { turnIndex: number; error: string }) => void;
 } {
   let latestAttemptSource: TurnSource | undefined;
   return {
     onTurnStarted: (info) => {
-      latestAttemptSource = { ...options.getSource() };
       noteCurrentTurnTraceId(turnTraceId(options.getSessionId(), info.turnIndex));
     },
+    onTurnSourceObserved: (info) => {
+      latestAttemptSource = { ...info.source };
+    },
     onTurnComplete: (ctx) => {
       latestAttemptSource = undefined;
       clearCurrentTurnTraceId();