Emit PostHog AI observability events in privacy mode (#417)

* Emit PostHog AI observability spans and generations per turn

PostHog's LLM analytics views query the $ai_-prefixed properties and
nothing else, so the documented names are not ours to choose: an
unprefixed property still arrives on the event but is invisible to every
trace, cost, and latency view. $ai_latency is a duration in seconds,
while the runtime measures milliseconds throughout.

Only ids, enums, and counts leave the process. A tool name and a
provider error message are both free text that routinely embed a local
path or a prompt excerpt, so each is classified into a fixed enum at the
boundary and the original discarded.

The trace is deliberately flat: PostHog accepts a trace id as a span's
parent, and top-level tool calls are all the turn record exposes.

* Retire the inference_turn event in favour of $ai_generation

Both events fired from the same hook with the same payload, so every
turn was reported twice, and only $ai_generation reaches PostHog's LLM
analytics views — inference_turn's numbers were readable in the raw
event stream and nowhere else. This is a deliberate removal of the
duplicate, not an oversight: nothing outside this repository consumed
inference_turn.

Author: TheGreatAxios

docs/TELEMETRY.md

@@ -12,7 +12,8 @@ Each event carries a small set of properties:
 |---|---|---|
 | `cli_start` | Once per used session (see First-run disclosure) | (none beyond common properties) |
 | `session_end` | When a TUI session finishes | `status`, `turn_count`, `duration_ms`, `session_mode`, `exit_reason` |
-| `inference_turn` | Once per completed turn | `provider_id`, `model_id`, `input_tokens`, `output_tokens`, `cache_read_tokens`, `cache_write_tokens`, `thinking_tokens`, `duration_ms` |
+| `$ai_generation` | Once per turn — on completion, and once for a turn that ends in an error instead | `$ai_trace_id`, `$ai_provider`, `$ai_model`, `$ai_input_tokens`, `$ai_output_tokens`, `$ai_latency`, `$ai_is_error`, `$ai_error`, `cache_read_tokens`, `cache_write_tokens`, `thinking_tokens` |
+| `$ai_span` | Once per top-level tool call in a completed turn | `$ai_trace_id`, `$ai_span_id`, `$ai_parent_id`, `$ai_span_name`, `$ai_is_error` |
 | `slash_command` | A slash command is dispatched in the TUI | `command_name` |
 | `skill_used` | `use_skill` loads a skill that resolved | (none beyond common properties) |
 | `plugin_loaded` | A plugin is discovered and loaded at startup | `origin` |
@@ -37,9 +38,9 @@ request IP; no location data is collected by the client.
 Every event is capped to an explicit property allowlist before it leaves the
 process — no other field can ever be attached, even by accident.
 
-`provider_id` is the canonical provider kind resolved by the runtime (e.g.
+`$ai_provider` is the canonical provider kind resolved by the runtime (e.g.
 `openai-compatible`), never the free-text name you gave the provider in
-onboarding or settings. `model_id` is the model identifier exactly as
+onboarding or settings. `$ai_model` is the model identifier exactly as
 configured — it is the one user-entered string that is sent, so do not put
 anything identifying in a model name.
 
@@ -75,6 +76,53 @@ The mapping is `src/telemetry/classify.ts`, and the tests that feed each
 emission site a deliberately identifying name and assert it reaches no part of
 the payload are in `tests/unit/telemetry-product-events.test.ts`.
 
+## AI observability events
+
+`$ai_generation` and `$ai_span` are the two PostHog AI observability events,
+emitted from `src/telemetry/ai-observability.ts`. PostHog's LLM analytics
+views query the `$ai_`-prefixed properties and nothing else, which is why
+these names are not ours to choose. `$ai_latency` is a duration in **seconds**
+as a float, per PostHog's schema — the runtime measures milliseconds and
+converts.
+
+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
+of its `$ai_span`s carry it, and every span's `$ai_parent_id` is that same
+trace id rather than another span. PostHog documents `$ai_parent_id` as
+accepting either a trace id or a span id, so this is a legal trace, and it is
+all the runtime can honestly describe: the turn record only exposes top-level
+tool calls. No `$ai_trace` event is emitted — PostHog synthesises the trace
+from its children.
+
+`$ai_span_id` is the provider-generated opaque tool call id. It identifies
+the call within the trace and carries nothing else.
+
+`$ai_span_name` is one of a fixed enum (`tool_call`, `subagent_call`). The raw
+tool name is never sent: an MCP tool name embeds the server identifier it was
+configured under, which can be a local path.
+
+`$ai_error` is likewise one of a fixed enum (`rate_limit`, `auth`, `timeout`,
+`cancelled`, `inference_failed`). The provider's error message is classified
+into one of these and then discarded — a raw message routinely embeds the
+request URL, a prompt excerpt, or a file path.
+
+The cache and thinking token counts keep unprefixed names because PostHog does
+not publish property names for them in its manual-capture schema; a guessed
+`$ai_` name would land as an unread custom property either way.
+
+Stopping a turn mid-inference is reported, not silent: the runtime aborts the
+in-flight call and classifies the resulting error as `cancelled`, so a stopped
+turn produces the same errored `$ai_generation` as a failed one and is told
+apart by `$ai_error`. A turn that never reaches inference at all — suspended
+at an approval prompt and never resumed — emits nothing, because the runtime
+raises no event for it.
+
+Exactly one `$ai_generation` is ever emitted per turn. A single give-up
+usually surfaces twice at the event stream (the failed inference, then the
+reactor terminating), and a turn that already reported completion is finished;
+`src/session/run-sink.ts` latches on both so neither can double-count a turn
+or append a phantom failure to a successful one.
+
 ## What's never collected
 
 - Prompts, model output, or any conversation content

src/session/run-sink.test.ts

@@ -77,6 +77,98 @@ describe("createRunSink", () => {
     expect(runSink.getTokenUsage()).toEqual({ input: 1, output: 1, cacheRead: 0, cacheWrite: 0, thinking: 0 });
   });
 
+  test("reports the in-flight turn to onTurnFailed when a turn errors instead of completing", () => {
+    const failures: { turnIndex: number; error: string }[] = [];
+    const runSink = createRunSink({
+      emitter: new EventEmitter(),
+      hookManager: stubHookManager([]),
+      onTurnFailed: (info) => failures.push(info),
+    });
+
+    runSink.sink(event("inference.start", {}));
+    runSink.sink(event("inference.error", { error: { message: "429 rate limit" } }));
+
+    expect(failures).toEqual([{ turnIndex: 0, error: "429 rate limit" }]);
+  });
+
+  // Regression: one give-up reaches the sink twice — the director surfaces
+  // the failed inference, then the reactor terminates the run — and reporting
+  // both files two failed turns under a single turn's identity.
+  test("reports one failure per turn across both error paths, not one per error event", () => {
+    const failures: { turnIndex: number; error: string }[] = [];
+    const runSink = createRunSink({
+      emitter: new EventEmitter(),
+      hookManager: stubHookManager([]),
+      onTurnFailed: (info) => failures.push(info),
+    });
+
+    runSink.sink(event("inference.start", {}));
+    runSink.sink(event("inference.error", { error: { message: "429 rate limit" } }));
+    runSink.sink(event("reactor.error", { error: "reactor gave up" }));
+
+    expect(failures).toEqual([{ turnIndex: 0, error: "429 rate limit" }]);
+  });
+
+  test("reports no failure for a turn that already completed", () => {
+    const failures: { turnIndex: number; error: string }[] = [];
+    const completions: number[] = [];
+    const runSink = createRunSink({
+      emitter: new EventEmitter(),
+      hookManager: stubHookManager([]),
+      onTurnComplete: (ctx) => completions.push(ctx.turnIndex),
+      onTurnFailed: (info) => failures.push(info),
+    });
+
+    runSink.sink(event("inference.start", {}));
+    runSink.sink(event("inference.done", {
+      turn: { role: "assistant", content: [], model: "test", timestamp: 0 },
+      usage: { input: 1, output: 1, cacheRead: 0, cacheWrite: 0, thinking: 0 },
+      source: { provider: "test", model: "test" },
+    }));
+    runSink.sink(event("reactor.error", { error: "reactor gave up at shutdown" }));
+
+    expect(completions).toEqual([0]);
+    expect(failures).toEqual([]);
+  });
+
+  // A retry re-enters inference.start under the same turn index, so a second
+  // report would land on the trace id the first one already claimed.
+  test("reports one failure for a turn that fails, retries, and fails again", () => {
+    const failures: { turnIndex: number; error: string }[] = [];
+    const runSink = createRunSink({
+      emitter: new EventEmitter(),
+      hookManager: stubHookManager([]),
+      onTurnFailed: (info) => failures.push(info),
+    });
+
+    runSink.sink(event("inference.start", {}));
+    runSink.sink(event("inference.error", { error: { message: "500 upstream" } }));
+    runSink.sink(event("inference.start", {}));
+    runSink.sink(event("inference.error", { error: { message: "500 upstream again" } }));
+
+    expect(failures).toEqual([{ turnIndex: 0, error: "500 upstream" }]);
+  });
+
+  test("still reports a failure after reset clears the latch", () => {
+    const failures: { turnIndex: number; error: string }[] = [];
+    const runSink = createRunSink({
+      emitter: new EventEmitter(),
+      hookManager: stubHookManager([]),
+      onTurnFailed: (info) => failures.push(info),
+    });
+
+    runSink.sink(event("inference.start", {}));
+    runSink.sink(event("inference.error", { error: { message: "first session" } }));
+    runSink.reset();
+    runSink.sink(event("inference.start", {}));
+    runSink.sink(event("inference.error", { error: { message: "second session" } }));
+
+    expect(failures).toEqual([
+      { turnIndex: 0, error: "first session" },
+      { turnIndex: 0, error: "second session" },
+    ]);
+  });
+
   test("seeds the turn count from a resumed session's prior turnsUsed", () => {
     const runSink = createRunSink({
       emitter: new EventEmitter(),

src/session/run-sink.ts

@@ -20,6 +20,12 @@ export type RunSinkArgs = {
   // turn actually ran against, so consumers report per-turn provider/model
   // even if the live selection changed mid-run.
   onTurnComplete?: (ctx: import("./hooks.js").TurnContext) => void;
+  // Fired at most once per turn, when that turn ends in an error instead of
+  // completing. onTurnComplete only ever sees turns that produced a full
+  // TurnContext, so a consumer relying on it alone goes silent exactly when a
+  // 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;
   // Continues a resumed session's persisted run.json turn count instead of
   // restarting the collector at zero.
   initialTurnCount?: number;
@@ -81,7 +87,8 @@ export function resolveExecRunStatus(args: {
 }
 
 export function createRunSink(args: RunSinkArgs): RunSink {
-  const { emitter, hookManager, onTurnComplete, initialTurnCount, onTurnBoundarySnapshot } = args;
+  const { emitter, hookManager, onTurnComplete, onTurnFailed, initialTurnCount, onTurnBoundarySnapshot } =
+    args;
 
   function hasConfiguredHooks(): boolean {
     return hookManager.getStatuses().length > 0;
@@ -110,12 +117,36 @@ export function createRunSink(args: RunSinkArgs): RunSink {
   let runCompleted = false;
   let runError: string | undefined;
   let turnCollector = createCollector(initialTurnCount);
+  // True between `inference.start` and whichever event settles that turn.
+  // One give-up reaches this sink twice — the director surfaces the failed
+  // inference, then the reactor terminates the run — and a turn that already
+  // completed is finished, so a later shutdown error belongs to no turn at
+  // all. Both cases resolve to the same question: is there a turn in flight
+  // for this error to be about?
+  let turnInFlight = false;
+  // Retries re-enter `inference.start` without advancing the turn count, so
+  // a second failure on a retried turn would report the index a consumer
+  // already recorded a failure for. Consumers key per-turn identity off that
+  // index, which makes a repeat indistinguishable from a duplicate.
+  let failedTurnIndex: number | null = null;
   // Always-on local PerfTrace: not gated by lifecycle hooks.
   let perfObserver = createPerfReactorObserver();
 
+  function reportTurnFailure(error: string): void {
+    if (!turnInFlight) return;
+    turnInFlight = false;
+    const turnIndex = turnCollector.getTurnCount();
+    if (turnIndex === failedTurnIndex) return;
+    failedTurnIndex = turnIndex;
+    onTurnFailed?.({ turnIndex, error });
+  }
+
   const sink = (event: ReactorEmittedEvent): void => {
     turnCollector.observe(event);
     perfObserver.observe(event);
+    if (event.type === "inference.start") {
+      turnInFlight = true;
+    }
     if (event.type === "reactor.done") {
       runCompleted = true;
       // Terminal success clears any earlier transient inference error.
@@ -125,16 +156,19 @@ export function createRunSink(args: RunSinkArgs): RunSink {
     // (ChatDirector retries timeout/retryable/aborted). Leaving the sticky error
     // would mark a recovered successful send as failed.
     if (onTurnBoundary(event)) {
+      turnInFlight = false;
       runError = undefined;
       onTurnBoundarySnapshot?.();
     }
     if (event.type === "reactor.error") {
       const data = event.data as { error: string };
       runError = data.error;
+      reportTurnFailure(data.error);
     }
     if (event.type === "inference.error") {
       const data = event.data as { error: { message: string } };
       runError = data.error.message;
+      reportTurnFailure(data.error.message);
     }
     emitter.emit("event", event);
   };
@@ -151,6 +185,8 @@ export function createRunSink(args: RunSinkArgs): RunSink {
     reset: () => {
       runCompleted = false;
       runError = undefined;
+      turnInFlight = false;
+      failedTurnIndex = null;
       turnCollector = createCollector();
       perfObserver.reset();
     },