Link subagent_end to the in-flight parent turn

Capture parent_trace_id at task/spawn dispatch from the current turn
noted at inference.start, instead of the last completed turn. Treat an
empty generation sample-rate env as unset, and document that sampling
drops opt-in spans with the generation.

Author: TheGreatAxios

docs/TELEMETRY.md

@@ -109,8 +109,11 @@ Leaf `runSubAgent` workers do not emit `$ai_*`; worker rollups travel on
 
 Successful `$ai_generation` events may be sampled with
 `CORBITS_TELEMETRY_GENERATION_SAMPLE_RATE` (a float in `0`–`1`, default `1.0`
-= keep all). Errored generations (`$ai_is_error: true`), `crash`, and
-`auth_failure` always ship regardless of the sample rate.
+= keep all). An empty env value is treated as unset (keep all), not as `0`.
+Errored generations (`$ai_is_error: true`), `crash`, and `auth_failure` always
+ship regardless of the sample rate. When a successful generation is sampled
+out, opt-in `$ai_span`s for that turn are skipped too — a span without its
+parent generation is not useful in PostHog traces.
 
 The trace is **flat**. Every turn gets one `$ai_trace_id` derived from the
 runtime's session id and the turn index; the turn's `$ai_generation` and each

src/session/run-sink.ts

@@ -22,9 +22,14 @@ export interface RunSinkArgs {
   // run goes wrong. The turn index is the collector's current count: the
   // in-flight turn is the one that would have been recorded next.
   onTurnFailed?: (info: { turnIndex: number; error: string }) => void;
+  // Fired when inference for a turn begins. The turn index is the collector's
+  // 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;
   // Continues a resumed session's persisted run.json turn count instead of
   // restarting the collector at zero.
   initialTurnCount?: number;
+
   // Fired at every turn boundary so a caller can persist a mid-run run.json
   // snapshot. `inference.done` is the turn boundary every reactor cycle
   // guarantees; `reactor.done` fires once, at shutdown, and never between
@@ -88,6 +93,7 @@ export function createRunSink(args: RunSinkArgs): RunSink {
     hookManager,
     onTurnComplete,
     onTurnFailed,
+    onTurnStarted,
     initialTurnCount,
     onTurnBoundarySnapshot,
   } = args;
@@ -148,7 +154,9 @@ export function createRunSink(args: RunSinkArgs): RunSink {
     perfObserver.observe(event);
     if (event.type === "inference.start") {
       turnInFlight = true;
+      onTurnStarted?.({ turnIndex: turnCollector.getTurnCount() });
     }
+
     if (event.type === "reactor.done") {
       runCompleted = true;
       // Terminal success clears any earlier transient inference error.

src/subagent/agent-fleet.ts

@@ -75,6 +75,7 @@ import { cleanupSubAgentWorktree, createSubAgentWorktree, WorktreeError } from "
 import { NOOP_TELEMETRY, type Telemetry } from "../telemetry/index.js";
 import { classifyAgentName } from "../telemetry/classify.js";
 import { captureSubagentEnd } from "../telemetry/product-events.js";
+import { getCurrentTurnTraceId } from "../telemetry/feedback.js";
 import type { DirectorPackage } from "../agent/directors/types.js";
 
 import { formatSubAgentTaskAuthFailureMessage } from "./inference-auth-failure.js";
@@ -538,12 +539,11 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
       });
       deps.fleetRecords.register(session.id);
       const agentName = classifyAgentName(resolved.directorId);
+      const parentTraceId = getCurrentTurnTraceId();
       telemetry.capture("subagent_start", { agent_name: agentName });
       const startedAt = Date.now();
       let endResult: RunSubAgentResult | undefined;
 
-
-
       const childCtl = new AbortController();
       deps.sessions.registerCancel(session.id, () => {
         if (!childCtl.signal.aborted) childCtl.abort();
@@ -754,11 +754,12 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
             model: provider.model,
             ...(endResult?.stopReason !== undefined ? { stopReason: endResult.stopReason } : {}),
             ...(endResult?.telemetry !== undefined ? { rollup: endResult.telemetry } : {}),
+            ...(parentTraceId !== undefined ? { parentTraceId } : {}),
           });
+
           if (!keepWorktreeAlive) void reclaimWorktree();
         });
 
-
       return fleetResult(call.id, JSON.stringify({ agent_id: session.id, status: "running" }));
     },
   });

src/subagent/task-tool.ts

@@ -58,6 +58,7 @@ import { currentTurnId } from "../perf/reactor-spans.js";
 import { classifyAgentName } from "../telemetry/classify.js";
 import { NOOP_TELEMETRY, type Telemetry } from "../telemetry/index.js";
 import { captureSubagentEnd } from "../telemetry/product-events.js";
+import { getCurrentTurnTraceId } from "../telemetry/feedback.js";
 
 import { join } from "node:path";
 import type {
@@ -69,7 +70,6 @@ import type {
   SubAgentTelemetryRollup,
 } from "./types.js";
 
-
 const log = getLogger([LOG_NAMESPACE_ROOT, "subagent", "task-tool"]);
 
 export const TaskToolArgs = type({
@@ -926,12 +926,12 @@ export function createTaskTool(deps: TaskToolDeps): AgentTool {
         },
       });
       const subagentStartedAt = Date.now();
-      // Profile ids come from project and plugin directories, so only the
-      // runtime's own "worker" fallback is reportable by name; anything else
-      // is bucketed. Sub-agents run in this process against the same session
-      // id, so there is no parent id worth sending — it would always equal
-      // the session_id already on the payload.
+      // Profile / director ids are classified: first-party DIRECTOR_IDS (and
+      // legacy "worker") report by name; project/plugin profiles become custom.
+      // Capture the in-flight parent turn trace at dispatch — getLastTurnTraceId
+      // would be the previous completed turn while this tool still runs.
       const agentName = classifyAgentName(agentLabel);
+      const parentTraceId = getCurrentTurnTraceId();
       telemetry.capture("subagent_start", { agent_name: agentName });
       let subagentStatus: "completed" | "cancelled" | "failed" = "completed";
       let endRollup: SubAgentTelemetryRollup | undefined;
@@ -1138,7 +1138,9 @@ export function createTaskTool(deps: TaskToolDeps): AgentTool {
           ...(endModel !== undefined ? { model: endModel } : { model: provider.model }),
           ...(endStopReason !== undefined ? { stopReason: endStopReason } : {}),
           ...(endRollup !== undefined ? { rollup: endRollup } : {}),
+          ...(parentTraceId !== undefined ? { parentTraceId } : {}),
         });
+
       }
 
     },

src/telemetry/ai-observability.test.ts

@@ -1,6 +1,7 @@
-import { describe, expect, test } from "bun:test";
 import type { ToolCall, ToolResult } from "@intx/types/runtime";
+import { afterEach, describe, expect, test } from "bun:test";
 import type { Telemetry } from "./index.js";
+import { generationSampleRate } from "./index.js";
 import type { TurnContext } from "../session/hooks.js";
 import {
   aggregateToolCalls,
@@ -12,11 +13,17 @@ import {
   secondsFromMs,
   turnTraceId,
 } from "./ai-observability.js";
+import { getCurrentTurnTraceId, resetFeedbackStateForTests } from "./feedback.js";
 
 const SUBAGENT_TOOL_NAME = "task";
 const SESSION_ID = "0199-parent-session";
 
+afterEach(() => {
+  resetFeedbackStateForTests();
+});
+
 function fakeTelemetry(): {
+
   telemetry: Telemetry;
   captured: { event: string; properties: Record<string, unknown> }[];
 } {
@@ -210,6 +217,21 @@ describe("createTurnObserver", () => {
     expect(captured[1]?.properties.$ai_provider).toBe("codex");
     expect(captured[1]?.properties.$ai_model).toBe("model-y");
   });
+
+  test("onTurnStarted notes the in-flight turn for subagent parent_trace_id", () => {
+    const { telemetry } = fakeTelemetry();
+    const observer = createTurnObserver({
+      telemetry: () => telemetry,
+      getSessionId: () => SESSION_ID,
+      getSource: () => ({ provider: "openai-compatible", model: "model-x" }),
+      subagentToolName: SUBAGENT_TOOL_NAME,
+    });
+
+    observer.onTurnStarted({ turnIndex: 2 });
+    expect(getCurrentTurnTraceId()).toBe(`${SESSION_ID}:turn:2`);
+    observer.onTurnComplete(fakeTurnContext({ turnIndex: 2, toolCalls: [], toolResults: [] }));
+    expect(getCurrentTurnTraceId()).toBeUndefined();
+  });
 });
 
 describe("emitAiObservability", () => {
@@ -378,6 +400,13 @@ describe("emitAiObservability", () => {
     expect(captured).toHaveLength(1);
     expect(captured[0]?.event).toBe("$ai_generation");
   });
+
+  test("empty CORBITS_TELEMETRY_GENERATION_SAMPLE_RATE is treated as unset (1.0)", () => {
+    expect(generationSampleRate({ CORBITS_TELEMETRY_GENERATION_SAMPLE_RATE: "" })).toBe(1);
+    expect(generationSampleRate({ CORBITS_TELEMETRY_GENERATION_SAMPLE_RATE: "  " })).toBe(1);
+    expect(generationSampleRate({})).toBe(1);
+    expect(generationSampleRate({ CORBITS_TELEMETRY_GENERATION_SAMPLE_RATE: "0" })).toBe(0);
+  });
 });
 
 describe("emitAiTurnFailure", () => {

src/telemetry/ai-observability.ts

@@ -9,7 +9,8 @@
 // CORBITS_TELEMETRY_AI_SPANS for debugging.
 
 import type { TurnContext } from "../session/hooks.js";
-import { noteLastTurnTraceId } from "./feedback.js";
+import { noteLastTurnTraceId, noteCurrentTurnTraceId, clearCurrentTurnTraceId } from "./feedback.js";
+
 import {
   aiSpansEnabled,
   generationSampleRate,
@@ -133,7 +134,11 @@ export function emitAiObservability(
   // is a no-op because this call still computes the id).
   noteLastTurnTraceId(traceId);
 
-  if (!shouldSampleSuccessfulGeneration(env, random)) return;
+  if (!shouldSampleSuccessfulGeneration(env, random)) {
+    // Sampling drops the whole turn package, including opt-in `$ai_span`s —
+    // a span without its parent generation is not useful in PostHog traces.
+    return;
+  }
 
   const aggregates = aggregateToolCalls(ctx, options.subagentToolName);
 
@@ -226,17 +231,23 @@ export interface CreateTurnObserverOptions {
 // plain callbacks and keeping the "read it now, do not capture it" rule in
 // one place instead of at each call site.
 export function createTurnObserver(options: CreateTurnObserverOptions): {
+  onTurnStarted: (info: { turnIndex: number }) => void;
   onTurnComplete: (ctx: TurnContext) => void;
   onTurnFailed: (info: { turnIndex: number; error: string }) => void;
 } {
   return {
+    onTurnStarted: (info) => {
+      noteCurrentTurnTraceId(turnTraceId(options.getSessionId(), info.turnIndex));
+    },
     onTurnComplete: (ctx) => {
+      clearCurrentTurnTraceId();
       emitAiObservability(options.telemetry(), ctx, {
         sessionId: options.getSessionId(),
         subagentToolName: options.subagentToolName,
       });
     },
     onTurnFailed: (info) => {
+      clearCurrentTurnTraceId();
       emitAiTurnFailure(options.telemetry(), {
         sessionId: options.getSessionId(),
         source: options.getSource(),

src/telemetry/feedback.ts

@@ -138,6 +138,8 @@ export function feedbackResultMessage(
 
 let feedbackCapturePending = false;
 let lastTurnTraceId: string | undefined;
+/** In-flight primary turn — set at inference.start, cleared when the turn settles. */
+let currentTurnTraceId: string | undefined;
 
 /** Arm after bare `/feedback` so the next non-command submit is treated as feedback. */
 export function armFeedbackCapture(): void {
@@ -169,8 +171,26 @@ export function getLastTurnTraceId(): string | undefined {
   return lastTurnTraceId;
 }
 
+/**
+ * Remember the in-flight turn's `$ai_trace_id` so `subagent_end` can link to the
+ * turn that is still running when `task` / `spawn_agent` dispatch (not the
+ * previous completed turn).
+ */
+export function noteCurrentTurnTraceId(traceId: string): void {
+  if (traceId.length > 0) currentTurnTraceId = traceId;
+}
+
+export function getCurrentTurnTraceId(): string | undefined {
+  return currentTurnTraceId;
+}
+
+export function clearCurrentTurnTraceId(): void {
+  currentTurnTraceId = undefined;
+}
+
 /** Test helper — reset module state between cases. */
 export function resetFeedbackStateForTests(): void {
   feedbackCapturePending = false;
   lastTurnTraceId = undefined;
+  currentTurnTraceId = undefined;
 }

src/telemetry/index.ts

@@ -206,7 +206,11 @@ export function aiSpansEnabled(env: NodeJS.ProcessEnv = process.env): boolean {
 export function generationSampleRate(env: NodeJS.ProcessEnv = process.env): number {
   const raw = env[TELEMETRY_GENERATION_SAMPLE_RATE_ENV];
   if (raw === undefined) return 1;
-  const parsed = Number(raw);
+  const trimmed = raw.trim();
+  // Empty env ("CORBITS_TELEMETRY_GENERATION_SAMPLE_RATE=") is unset, not 0 —
+  // Number("") is 0 and would silently drop every successful generation.
+  if (trimmed.length === 0) return 1;
+  const parsed = Number(trimmed);
   if (!Number.isFinite(parsed)) return 1;
   return Math.min(1, Math.max(0, parsed));
 }

src/telemetry/product-events.ts

@@ -5,7 +5,6 @@ import type { ForcedStopReason } from "../subagent/stop-policy.js";
 import type { SubAgentTelemetryRollup } from "../subagent/types.js";
 import type { Telemetry } from "./index.js";
 import { classifyCommandName } from "./classify.js";
-import { getLastTurnTraceId } from "./feedback.js";
 
 /** Emit slash_command with a classified first-party (or `custom`) name. */
 export function captureSlashCommand(telemetry: Telemetry, commandName: string): void {
@@ -22,16 +21,17 @@ export type CaptureSubagentEndArgs = {
   model?: string;
   stopReason?: ForcedStopReason;
   rollup?: SubAgentTelemetryRollup;
-  /** Override for tests; defaults to the last parent-turn trace id. */
+  /**
+   * Spawn-time parent `$ai_trace_id` (in-flight turn). Callers must capture
+   * this at dispatch — never default to the last *completed* turn.
+   */
   parentTraceId?: string | undefined;
 };
 
 /** Build allowlisted `subagent_end` properties from a finished run. */
 export function buildSubagentEndProperties(
   args: CaptureSubagentEndArgs,
 ): Record<string, unknown> {
-  const parentTraceId =
-    args.parentTraceId !== undefined ? args.parentTraceId : getLastTurnTraceId();
   const props: Record<string, unknown> = {
     agent_name: args.agentName,
     status: args.status,
@@ -43,8 +43,8 @@ export function buildSubagentEndProperties(
   if (args.stopReason !== undefined) {
     props.stop_reason = args.stopReason;
   }
-  if (parentTraceId !== undefined && parentTraceId.length > 0) {
-    props.parent_trace_id = parentTraceId;
+  if (args.parentTraceId !== undefined && args.parentTraceId.length > 0) {
+    props.parent_trace_id = args.parentTraceId;
   }
   if (args.rollup !== undefined) {
     props.turn_count = args.rollup.turn_count;

src/tui/runner.ts

@@ -1591,6 +1591,7 @@ export async function runTUI(initialConfig: Config): Promise<number> {
       emitter,
       hookManager,
       initialTurnCount: resumeSeed.turnsUsed,
+      onTurnStarted: turnObserver.onTurnStarted,
       onTurnComplete: turnObserver.onTurnComplete,
       onTurnFailed: turnObserver.onTurnFailed,
       // persistRunSnapshot is defined below but not invoked until the stream
@@ -1600,6 +1601,7 @@ export async function runTUI(initialConfig: Config): Promise<number> {
       },
     });
 
+
     // MCP servers connected so far, keyed by name so a reconnect after a failure
     // replaces rather than duplicates the entry.
     let connectedMcpServers: ConnectedMcpServer[] = resumeSeed.mcpServers;