diff --git a/docs/TELEMETRY.md b/docs/TELEMETRY.md index f9d4269cf..e833799c0 100644 --- a/docs/TELEMETRY.md +++ b/docs/TELEMETRY.md @@ -11,22 +11,22 @@ env kill switches (see Intentional feedback below). Each event carries a small set of properties: -| Event | When | 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` | -| `$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`, `$ai_cache_read_input_tokens`, `$ai_cache_creation_input_tokens`, `$ai_reasoning_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 (shared product-event path) | `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` | -| `subagent_start` | A `task` dispatch begins | `agent_name` | -| `subagent_end` | A `task` dispatch finishes | `agent_name`, `status`, `duration_ms` | -| `permission_prompt` | An approval prompt is answered (or abandoned) | `decision`, `permission_kind` | -| `compaction` | The compactor actually folds turns away | `mode`, `duration_ms`, `turns_before`, `turns_after` | -| `crash` | A fatal error reaches the process-level handler | `kind`, `error_class` | -| `auth_failure` | A provider rejects the stored credentials | `auth_provider` | -| `survey sent` | User submits intentional feedback via `/feedback` | `$survey_id`, `$survey_response`, `$survey_questions`, `turn_trace_id` | +| Event | When | 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` | +| `$ai_generation` | Once per completed turn (may be sampled); always on turn failure | `$ai_trace_id`, `$ai_provider`, `$ai_model`, `$ai_input_tokens`, `$ai_output_tokens`, `$ai_latency`, `$ai_is_error`, `$ai_error`, `$ai_cache_read_input_tokens`, `$ai_cache_creation_input_tokens`, `$ai_reasoning_tokens`, `tool_call_count`, `tool_error_count`, `subagent_call_count` | +| `$ai_span` | Opt-in only — once per top-level tool call when `CORBITS_TELEMETRY_AI_SPANS` is set | `$ai_trace_id`, `$ai_span_id`, `$ai_parent_id`, `$ai_span_name`, `$ai_is_error` | +| `slash_command` | A slash command is dispatched (shared product-event path) | `command_name` | +| `skill_used` | `use_skill` loads a skill that resolved | (none beyond common properties) | +| `plugin_loaded` | First successful load of a plugin identity in this process | `origin` | +| `subagent_start` | A `task` / fleet dispatch begins | `agent_name` | +| `subagent_end` | A `task` / fleet dispatch finishes | `agent_name`, `status`, `duration_ms`, `model`, `turn_count`, `input_tokens`, `output_tokens`, `cache_read_tokens`, `cache_write_tokens`, `reasoning_tokens`, `tool_call_count`, `tool_error_count`, `stop_reason`, `parent_trace_id` | +| `permission_prompt` | An approval prompt is answered (or abandoned) | `decision`, `permission_kind` | +| `compaction` | The compactor actually folds turns away | `mode`, `duration_ms`, `turns_before`, `turns_after` | +| `crash` | A fatal error reaches the process-level handler | `kind`, `error_class` | +| `auth_failure` | A provider rejects the stored credentials | `auth_provider` | +| `survey sent` | User submits intentional feedback via `/feedback` | `$survey_id`, `$survey_response`, `$survey_questions`, `turn_trace_id` | `compaction` is deliberately silent on the runs where the compactor decides there is nothing to compact — an event that also fires on no-ops makes its own @@ -36,7 +36,11 @@ Common properties attached to every event: a random installation UUID (`distinct_id`), `session_id`, `$app_version` (PostHog's standard Version property, the running package version), `service_version` (same value, kept for existing custom-property dashboards), `os_type`, `os_arch`, and a -`schema_version` for forward compatibility. +`schema_version` for forward compatibility. Ambient product and AI events also +carry `$process_person_profile: false` so PostHog treats them as anonymous +(batch capture otherwise defaults to identified processing). Intentional +`survey sent` omits that flag so `/feedback` can still join a person profile +if one is ever created. Approximate country-level location is derived server-side by PostHog from the request IP; no location data is collected by the client. @@ -62,10 +66,16 @@ So none of them are transmitted. Each is matched against a fixed list of names this project itself ships and reported as that name, or as `custom` when it matches nothing — with `mcp` as its own bucket for `permission_kind`, so the share of prompts driven by MCP stays visible without the server key coming -with it. `skill_used` and `plugin_loaded` go further: there is no first-party -list of skills or plugins to match against, so `skill_used` carries no name at -all and `plugin_loaded` carries only `origin`, the discovery tier -(`repo`, `user`, `project`, `path`). +with it. `agent_name` on `subagent_*` is the same pattern: first-party +director ids from `DIRECTOR_IDS` (and the legacy `worker` alias) are reported +by id; project-defined or marketplace profile ids become `custom`. +`skill_used` and `plugin_loaded` go further: there is no first-party list of +skills or plugins to match against, so `skill_used` carries no name at all and +`plugin_loaded` carries only `origin`, the discovery tier (`repo`, `user`, +`project`, `path`). Enabled telemetry reports the same plugin identity at most +once per runtime reporter, including across reloads. Disabled/no-op loads do not +consume that identity, so enabling telemetry later can report the first real +load. `error_class` is bucketed the same way: only the error types defined by the language are reported by name, because an error subclass defined in @@ -84,21 +94,48 @@ 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. +`$ai_generation` and (optionally) `$ai_span` are the 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. + +**Default volume shape (CL-6816):** each completed primary turn emits **one** +`$ai_generation` with tool/subagent aggregates folded onto it +(`tool_call_count`, `tool_error_count`, `subagent_call_count`). Per-call +`$ai_span` events are **off by default**. Set `CORBITS_TELEMETRY_AI_SPANS` to +a truthy value (`1`, `true`, …) to restore per-call spans for debugging. +Leaf `runSubAgent` workers do not emit `$ai_*`; worker rollups travel on +`subagent_end` instead. Both TUI and exec install the same turn observer, so a +worker ending during an active parent turn carries that turn's `parent_trace_id`. +Pre-progress operator aborts settle with `status=cancelled` and +`stop_reason=cancelled` even when the worker promise rejects. An interrupt that +keeps a worker resumable settles with `status=interrupted` and the same +`stop_reason=cancelled`; terminal events never report a still-running status. + +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` += 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 -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. +of its `$ai_span`s (when spans are enabled) 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. @@ -124,11 +161,19 @@ 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. +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. Attribution +uses the latest `inference.usage` source, the first lifecycle payload carrying +the runtime-resolved provider/model pair for an attempt. Each `inference.start` +clears that authoritative source and records the newly attempted model. If a +fallback fails before usage exposes its source, telemetry retains that actual +model but uses the fixed `unknown` provider/source bucket rather than attributing +it to the previously selected provider. When the attempted model still matches +the selected source, that full source remains valid. Therefore a parent turn +emits at most one terminal `$ai_generation`, including retry and failover paths. ## What's never collected @@ -159,6 +204,8 @@ is off (`settings.telemetry.enabled === false` or the Telemetry toggle Off), because the operator typed the text for that purpose. Hard env kill switches still win — `DO_NOT_TRACK=1` or `CORBITS_TELEMETRY=0/false/off/no` block `/feedback` as well. Sending also requires an installation id and API key. +Unlike ambient events, `survey sent` does not stamp `$process_person_profile: +false`, so the response can join a person profile if one is ever created. Survey id / question id are **baked into the client** (Corbits team survey `Corbits Code Feedback`). Same trust class as the public PostHog project key — diff --git a/src/exec/runner.ts b/src/exec/runner.ts index 1e19592f0..cd6c21d06 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -62,6 +62,7 @@ import type { ApprovalOutcome, PermissionRequest } from "../permission/types.js" import { createAgentToolset, type AgentToolset, type OperatorResult } from "../agent/tools.js"; import { createAgentWithLiveToolDispatch } from "../agent/live-tool-dispatch.js"; import { liveTelemetry } from "../telemetry/singleton.js"; +import { createTurnObserver } from "../telemetry/ai-observability.js"; import { collectToolPlugins, resolveToolPlugins } from "../plugins/tool-plugins.js"; import { expandExistingPluginMembers, @@ -657,9 +658,15 @@ export async function runExec(config: Config): Promise { const hookManager = createLifecycleHookManager({ hooks: await discoverLifecycleHooks(hookDirectories(config.cwd)), }); + const turnObserver = createTurnObserver({ + telemetry: () => liveTelemetry, + getSessionId: () => sessionId, + getSource: () => liveSource, + }); const liveSink = createRunSink({ emitter, hookManager, + ...turnObserver, onTurnBoundarySnapshot: () => { void persist("running"); }, diff --git a/src/plugins/loader.ts b/src/plugins/loader.ts index f0ac29d4e..f44f5b768 100644 --- a/src/plugins/loader.ts +++ b/src/plugins/loader.ts @@ -10,7 +10,10 @@ import type { CommandPlugin } from "../tui/commands/registry.js"; import { pathIsInsideOrEqual } from "../util/path-contain.js"; import { parsePluginManifest, type PluginManifest } from "./manifest.js"; import { NOOP_TELEMETRY, type Telemetry } from "../telemetry/index.js"; +import type { PluginLoadReporter } from "../telemetry/product-events.js"; +import { runtimePluginLoadReporter } from "../telemetry/singleton.js"; import { loadDataOnlyPlugin } from "./data-only.js"; + import { resolvePluginWarningHandler, stderrPluginWarning, @@ -125,6 +128,7 @@ export async function loadPluginEntry( diagnostics?: PluginLoadDiagnostics; origin?: PluginOrigin; telemetry?: Telemetry; + pluginLoadReporter?: PluginLoadReporter; } = {}, ): Promise { const cwd = opts.cwd ?? process.cwd(); @@ -139,6 +143,7 @@ export async function loadPluginEntry( ); const origin = opts.origin; const telemetry = opts.telemetry ?? NOOP_TELEMETRY; + const reportPluginLoaded = opts.pluginLoadReporter ?? runtimePluginLoadReporter; let target = entryPath; let pluginDir = entryPath; try { @@ -173,7 +178,9 @@ export async function loadPluginEntry( mod.origin = origin; mod.pluginPath = resolve(entryPath); } - if (origin !== undefined) telemetry.capture("plugin_loaded", { origin }); + if (origin !== undefined) { + reportPluginLoaded(telemetry, origin, resolve(entryPath)); + } return mod; } return null; @@ -234,7 +241,9 @@ export async function loadPluginEntry( result.origin = origin; result.pluginPath = resolve(pluginDir); } - if (origin !== undefined) telemetry.capture("plugin_loaded", { origin }); + if (origin !== undefined) { + reportPluginLoaded(telemetry, origin, resolve(pluginDir)); + } return result; } catch (err) { // Route through the same sink as skill/load warnings so a diagnostics @@ -848,7 +857,9 @@ export async function discoverClaudeInstalledPlugins( name: plugin.manifest.name === plugin.manifest.id ? idFromKey : plugin.manifest.name, }; } - opts.telemetry?.capture("plugin_loaded", { origin: "user" }); + if (opts.telemetry !== undefined) { + runtimePluginLoadReporter(opts.telemetry, "user", resolve(d)); + } results.push(plugin); } } diff --git a/src/session/run-sink.test.ts b/src/session/run-sink.test.ts index f23da7ab9..1e52e5587 100644 --- a/src/session/run-sink.test.ts +++ b/src/session/run-sink.test.ts @@ -3,6 +3,8 @@ import { describe, expect, test } from "bun:test"; import type { ReactorEmittedEvent } from "@intx/inference"; import { createRunSink } from "./run-sink.js"; import type { LifecycleHookStatus } from "./hooks.js"; +import { createTurnObserver } from "../telemetry/ai-observability.js"; +import type { Telemetry } from "../telemetry/index.js"; function event(type: string, data: unknown): ReactorEmittedEvent { return { type, seq: 1, data } as ReactorEmittedEvent; @@ -23,6 +25,42 @@ const enabledHook: LifecycleHookStatus = { enabled: true, }; +function attributionHarness(selectedSource = { provider: "provider-a", model: "model-a" }) { + const captured: { event: string; properties: Record }[] = []; + const telemetry: Telemetry = { + enabled: true, + installationId: "test", + capture: (capturedEvent, properties = {}) => { + captured.push({ event: capturedEvent, properties }); + }, + captureIntentional: () => false, + flush: async () => {}, + discard: () => {}, + }; + const observer = createTurnObserver({ + telemetry: () => telemetry, + getSessionId: () => "session-1", + getSource: () => selectedSource, + }); + const runSink = createRunSink({ + emitter: new EventEmitter(), + hookManager: stubHookManager([]), + ...observer, + }); + return { captured, runSink }; +} + +function failMessageRun(runSink: ReturnType): void { + runSink.sink(event("inference.error", { error: { message: "attempt failed" } })); + runSink.sink( + event("message.run.ended", { + messageRunId: "run-1", + messageId: "message-1", + status: "failed", + }), + ); +} + describe("createRunSink", () => { test("allocates no turn collector when no lifecycle hooks are configured", () => { const runSink = createRunSink({ @@ -112,7 +150,7 @@ describe("createRunSink", () => { expect(runSink.getTurnCount()).toBe(1); }); - test("reports the in-flight turn to onTurnFailed when a turn errors instead of completing", () => { + test("settles a pending inference failure only when the message run fails", () => { const failures: { turnIndex: number; error: string }[] = []; const runSink = createRunSink({ emitter: new EventEmitter(), @@ -122,26 +160,121 @@ describe("createRunSink", () => { runSink.sink(event("inference.start", {})); runSink.sink(event("inference.error", { error: { message: "429 rate limit" } })); + expect(failures).toEqual([]); + + runSink.sink( + event("message.run.ended", { + messageRunId: "run-1", + messageId: "message-1", + status: "failed", + error: { message: "reactor gave up", kind: "inference_error" }, + }), + ); 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", () => { + test("uses unknown attribution when a fallback model fails before usage", () => { + const { captured, runSink } = attributionHarness(); + + runSink.sink(event("inference.start", { model: "model-b" })); + failMessageRun(runSink); + + expect(captured).toHaveLength(1); + expect(captured[0]?.event).toBe("$ai_generation"); + expect(captured[0]?.properties).toMatchObject({ + $ai_provider: "unknown", + $ai_model: "model-b", + $ai_is_error: true, + }); + }); + + test("uses the selected source when its model fails before usage", () => { + const { captured, runSink } = attributionHarness(); + + runSink.sink(event("inference.start", { model: "model-a" })); + failMessageRun(runSink); + + expect(captured).toHaveLength(1); + expect(captured[0]?.properties).toMatchObject({ + $ai_provider: "provider-a", + $ai_model: "model-a", + $ai_is_error: true, + }); + }); + + test("uses authoritative usage attribution for a failed fallback", () => { + const { captured, runSink } = attributionHarness(); + + 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" }, + }), + ); + failMessageRun(runSink); + + expect(captured).toHaveLength(1); + expect(captured[0]?.properties).toMatchObject({ + $ai_provider: "provider-b", + $ai_model: "model-b", + $ai_is_error: true, + }); + }); + + test("does not leak authoritative source attribution across retry attempts", () => { + const { captured, runSink } = attributionHarness(); + + runSink.sink(event("inference.start", { model: "model-a" })); + runSink.sink( + event("inference.usage", { + usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, thinking: 0 }, + source: { sourceId: "selected", provider: "provider-a", model: "model-a" }, + }), + ); + runSink.sink(event("inference.error", { error: { message: "retry" } })); + runSink.sink(event("inference.start", { model: "model-b" })); + failMessageRun(runSink); + + expect(captured).toHaveLength(1); + expect(captured[0]?.properties).toMatchObject({ + $ai_provider: "unknown", + $ai_model: "model-b", + $ai_is_error: true, + }); + }); + + test("discards a recoverable inference failure after retry success", () => { 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.error", { error: { message: "429 rate limit" } })); - runSink.sink(event("reactor.error", { error: "reactor gave up" })); + runSink.sink(event("inference.error", { error: { message: "retry me" } })); + 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("message.run.ended", { + messageRunId: "run-1", + messageId: "message-1", + status: "completed", + }), + ); - expect(failures).toEqual([{ turnIndex: 0, error: "429 rate limit" }]); + expect(completions).toEqual([0]); + expect(failures).toEqual([]); }); test("reports no failure for a turn that already completed", () => { @@ -168,9 +301,7 @@ describe("createRunSink", () => { 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", () => { + test("settles the latest failed retry exactly once", () => { const failures: { turnIndex: number; error: string }[] = []; const runSink = createRunSink({ emitter: new EventEmitter(), @@ -182,11 +313,18 @@ describe("createRunSink", () => { runSink.sink(event("inference.error", { error: { message: "500 upstream" } })); runSink.sink(event("inference.start", {})); runSink.sink(event("inference.error", { error: { message: "500 upstream again" } })); + runSink.sink( + event("message.run.ended", { + messageRunId: "run-1", + messageId: "message-1", + status: "failed", + }), + ); - expect(failures).toEqual([{ turnIndex: 0, error: "500 upstream" }]); + expect(failures).toEqual([{ turnIndex: 0, error: "500 upstream again" }]); }); - test("still reports a failure after reset clears the latch", () => { + test("reset discards an unresolved pending failure", () => { const failures: { turnIndex: number; error: string }[] = []; const runSink = createRunSink({ emitter: new EventEmitter(), @@ -199,11 +337,15 @@ describe("createRunSink", () => { runSink.reset(); runSink.sink(event("inference.start", {})); runSink.sink(event("inference.error", { error: { message: "second session" } })); + runSink.sink( + event("message.run.ended", { + messageRunId: "run-2", + messageId: "message-2", + status: "failed", + }), + ); - expect(failures).toEqual([ - { turnIndex: 0, error: "first session" }, - { turnIndex: 0, error: "second session" }, - ]); + expect(failures).toEqual([{ turnIndex: 0, error: "second session" }]); }); test("seeds the turn count from a resumed session's prior turnsUsed", () => { diff --git a/src/session/run-sink.ts b/src/session/run-sink.ts index 4843d297c..068b5ba02 100644 --- a/src/session/run-sink.ts +++ b/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"; @@ -22,9 +22,17 @@ 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 for every inference attempt. The model comes from inference.start, + // while the turn index is the collector's current in-flight turn count. + onTurnStarted?: (info: { turnIndex: number; model: string }) => 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; + // 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 +96,8 @@ export function createRunSink(args: RunSinkArgs): RunSink { hookManager, onTurnComplete, onTurnFailed, + onTurnStarted, + onTurnSourceObserved, initialTurnCount, onTurnBoundarySnapshot, } = args; @@ -119,28 +129,19 @@ 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? + // A provider failure is only an attempt failure until the enclosing message + // run settles. Retried attempts reuse the same turn index, so emitting at + // inference.error would create a terminal generation for a recoverable retry. 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; + let pendingInferenceError: string | undefined; // Always-on local PerfTrace: not gated by lifecycle hooks. const perfObserver = createPerfReactorObserver(); - function reportTurnFailure(error: string): void { + function settleTurnFailure(error: string): void { if (!turnInFlight) return; turnInFlight = false; - const turnIndex = turnCollector.getTurnCount(); - if (turnIndex === failedTurnIndex) return; - failedTurnIndex = turnIndex; - onTurnFailed?.({ turnIndex, error }); + pendingInferenceError = undefined; + onTurnFailed?.({ turnIndex: turnCollector.getTurnCount(), error }); } const sink = (event: ReactorEmittedEvent): void => { @@ -148,7 +149,15 @@ export function createRunSink(args: RunSinkArgs): RunSink { perfObserver.observe(event); if (event.type === "inference.start") { turnInFlight = true; + onTurnStarted?.({ turnIndex: turnCollector.getTurnCount(), model: event.data.model }); + } + if (event.type === "inference.usage") { + onTurnSourceObserved?.({ + turnIndex: turnCollector.getTurnCount(), + source: event.data.source, + }); } + if (event.type === "reactor.done") { runCompleted = true; // Terminal success clears any earlier transient inference error. @@ -159,18 +168,25 @@ export function createRunSink(args: RunSinkArgs): RunSink { // would mark a recovered successful send as failed. if (onTurnBoundary(event)) { turnInFlight = false; + pendingInferenceError = undefined; 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 } }; + pendingInferenceError = data.error.message; runError = data.error.message; - reportTurnFailure(data.error.message); + } + if (event.type === "message.run.ended") { + if (event.data.status === "failed" && turnInFlight) { + settleTurnFailure(pendingInferenceError ?? event.data.error?.message ?? "Inference failed"); + } else { + pendingInferenceError = undefined; + } } emitter.emit("event", event); }; @@ -188,7 +204,7 @@ export function createRunSink(args: RunSinkArgs): RunSink { runCompleted = false; runError = undefined; turnInFlight = false; - failedTurnIndex = null; + pendingInferenceError = undefined; turnCollector = createCollector(); perfObserver.reset(); }, diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index 38c16548e..820a84d90 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -70,11 +70,16 @@ import type { RunSubAgentResult, SubAgentProvider, SubAgentSandboxDeps, + SubAgentRunSettlement, } from "./types.js"; import { cleanupSubAgentWorktree, createSubAgentWorktree, WorktreeError } from "./worktree.js"; 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 { SPAWN_AGENT_TOOL_NAME } from "./tool-taxonomy.js"; + import { formatSubAgentTaskAuthFailureMessage } from "./inference-auth-failure.js"; import { isSubAgentCancelError } from "./dispose.js"; @@ -261,7 +266,7 @@ const SpawnAgentArgs = type({ }); export const spawnAgentToolDefinition: ToolDefinition = { - name: "spawn_agent", + name: SPAWN_AGENT_TOOL_NAME, description: "Start a worker agent and return IMMEDIATELY with its agent_id — this never blocks on the worker's completion. Same brief fields as task() (description/prompt/context/goals/intent/success_criteria/do_not/report_focus); pass agent= a director id or intent= (one of explore|implement|review|plan|general). Fire several spawn_agent calls in one turn to start workers in parallel, then use wait_agents to collect them. task() is the deprecated fused spawn+wait fallback for a single blocking worker.", inputSchema: { @@ -536,8 +541,40 @@ 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 settlement: Readonly | undefined; + let endFinalized = false; + const finalizeEnd = (setupFailed = false): void => { + if (endFinalized) return; + endFinalized = true; + const terminalSession = deps.sessions.get(session.id); + const status = + terminalSession?.status === "cancelled" + ? "cancelled" + : terminalSession?.lifecycleStatus === "interrupted" + ? "interrupted" + : (terminalSession?.status ?? "completed"); + captureSubagentEnd(telemetry, { + agentName, + status: setupFailed ? "failed" : status, + durationMs: Date.now() - startedAt, + model: settlement?.model ?? provider.model, + stopReason: setupFailed ? "setup_error" : (settlement?.terminal_reason ?? "error"), + rollup: settlement ?? { + turn_count: 0, + input_tokens: 0, + output_tokens: 0, + cache_read_tokens: 0, + cache_write_tokens: 0, + reasoning_tokens: 0, + tool_call_count: 0, + tool_error_count: 0, + }, + ...(parentTraceId !== undefined ? { parentTraceId } : {}), + }); + }; const childCtl = new AbortController(); deps.sessions.registerCancel(session.id, () => { @@ -567,6 +604,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { : `sub-agent worktree setup failed: ${err instanceof Error ? err.message : String(err)}`; deps.fleetRecords.reject(session.id, message); deps.sessions.fail(session.id, message); + finalizeEnd(true); return fleetResult(call.id, `Error: ${message}`); } } @@ -642,6 +680,9 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { ...(reportFocus !== undefined && reportFocus.length > 0 ? { reportFocus } : {}), signal: childCtl.signal, onEvent, + onRunSettled: (summary) => { + settlement = summary; + }, ...(deps.onProgress !== undefined ? { onProgress: deps.onProgress } : {}), ...(resolved.capabilities !== undefined ? { capabilities: resolved.capabilities } : {}), systemPromptRole: resolved.systemPromptRole, @@ -697,6 +738,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { // that never saw interrupt_agent (or raced it) cannot hang. if (result.interrupted === true) { keepWorktreeAlive = true; + deps.sessions.interruptOne(session.id); deps.fleetRecords.interrupt(session.id, result.report); return; } @@ -740,11 +782,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { deps.sessions.fail(session.id, failReason); }) .finally(() => { - telemetry.capture("subagent_end", { - agent_name: agentName, - status: deps.sessions.get(session.id)?.status ?? "completed", - duration_ms: Date.now() - startedAt, - }); + finalizeEnd(); if (!keepWorktreeAlive) void reclaimWorktree(); }); diff --git a/src/subagent/index.ts b/src/subagent/index.ts index 816bb8c28..24d96378e 100644 --- a/src/subagent/index.ts +++ b/src/subagent/index.ts @@ -88,6 +88,7 @@ export type { RunSubAgentResult, SubAgentProvider, SubAgentSandboxDeps, + SubAgentTelemetryRollup, } from "./types.js"; export { diff --git a/src/subagent/run-settlement.test.ts b/src/subagent/run-settlement.test.ts new file mode 100644 index 000000000..33981b675 --- /dev/null +++ b/src/subagent/run-settlement.test.ts @@ -0,0 +1,184 @@ +import { expect, test } from "bun:test"; +import { mkdtemp } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import type { ReactorEmittedEvent } from "@intx/inference"; + +import { withMockedModuleDuring } from "../../tests/helpers/mock-module.js"; +import { createPermissionGate } from "../permission/gate.js"; +import type { SubAgentRunSettlement } from "./types.js"; + +const permissionGate = createPermissionGate({ + approvals: [], + interactive: false, + skipPermissions: true, +}); + +test("rejected workers settle prior rollups with the latest observed model", async () => { + const cwd = await mkdtemp(join(tmpdir(), "corbits-run-settlement-")); + const originalError = new Error("worker failed after prior activity"); + let settlement: Readonly | undefined; + + const caught = await withMockedModuleDuring( + import.meta.resolve("../agent/live-tool-dispatch.js"), + (real: typeof import("../agent/live-tool-dispatch.js")) => ({ + ...real, + createAgentWithLiveToolDispatch: async () => ({ + send: async () => { + await new Promise((resolve) => setTimeout(resolve, 20)); + throw originalError; + }, + stream: () => + (async function* (): AsyncGenerator { + yield { + type: "tool.start", + seq: 1, + data: { call: { id: "call-1", name: "read_file", arguments: {} } }, + } as ReactorEmittedEvent; + yield { + type: "tool.done", + seq: 2, + data: { + call: { id: "call-1", name: "read_file", arguments: {} }, + result: { callId: "call-1", content: "failed", isError: true }, + }, + } as ReactorEmittedEvent; + yield { + type: "inference.done", + seq: 3, + data: { + turn: { role: "assistant", content: [], model: "backup-model", timestamp: 0 }, + usage: { + input: 11, + output: 7, + cacheRead: 3, + cacheWrite: 2, + thinking: 5, + }, + source: { + sourceId: "backup-source", + provider: "backup", + model: "backup-model", + }, + }, + } as ReactorEmittedEvent; + yield { + type: "inference.start", + seq: 4, + data: { model: "terminal-model" }, + }; + })(), + deliver: () => {}, + close: async () => {}, + setSource: () => {}, + setSources: () => {}, + history: async () => [], + checkpoints: async () => [], + readAt: async () => [], + blobReader: {}, + }), + }), + async () => { + const { runSubAgent } = await import("./run.js"); + try { + await runSubAgent({ + cwd, + workdirBase: join(cwd, ".ctx"), + permissionGate, + provider: { + providerName: "initial", + baseURL: "http://localhost", + model: "initial-model", + }, + description: "settlement probe", + prompt: "do work then fail", + onRunSettled: (summary) => { + settlement = summary; + }, + }); + } catch (error) { + return error; + } + throw new Error("expected runSubAgent to reject"); + }, + ); + + expect(caught).toBe(originalError); + expect(settlement).toMatchObject({ + turn_count: 1, + input_tokens: 11, + output_tokens: 7, + cache_read_tokens: 3, + cache_write_tokens: 2, + reasoning_tokens: 5, + tool_call_count: 1, + tool_error_count: 1, + error_count: 1, + model: "terminal-model", + terminal_reason: "error", + }); + expect(Object.isFrozen(settlement)).toBe(true); +}); + +test("pre-progress cancellation settles as cancelled without changing rejection", async () => { + const cwd = await mkdtemp(join(tmpdir(), "corbits-run-cancelled-")); + const controller = new AbortController(); + const originalError = new DOMException("operator cancelled", "AbortError"); + controller.abort(originalError); + let settlement: Readonly | undefined; + + const caught = await withMockedModuleDuring( + import.meta.resolve("../agent/live-tool-dispatch.js"), + (real: typeof import("../agent/live-tool-dispatch.js")) => ({ + ...real, + createAgentWithLiveToolDispatch: async () => ({ + send: async () => { + throw new Error("send must not start after cancellation"); + }, + stream: () => (async function* () {})(), + deliver: () => {}, + close: async () => {}, + setSource: () => {}, + setSources: () => {}, + history: async () => [], + checkpoints: async () => [], + readAt: async () => [], + blobReader: {}, + }), + }), + async () => { + const { runSubAgent } = await import("./run.js"); + try { + await runSubAgent({ + cwd, + workdirBase: join(cwd, ".ctx"), + permissionGate, + provider: { + providerName: "initial", + baseURL: "http://localhost", + model: "initial-model", + }, + description: "cancelled settlement probe", + prompt: "do not start", + signal: controller.signal, + onRunSettled: (summary) => { + settlement = summary; + }, + }); + } catch (error) { + return error; + } + throw new Error("expected runSubAgent to reject"); + }, + ); + + expect(caught).toBe(originalError); + expect(settlement).toMatchObject({ + turn_count: 0, + error_count: 1, + model: "initial-model", + terminal_reason: "cancelled", + }); + expect(Object.isFrozen(settlement)).toBe(true); +}); diff --git a/src/subagent/run.ts b/src/subagent/run.ts index 7e614a0df..beb860e27 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -74,7 +74,9 @@ import { gatherEnvironment } from "../agent/environment.js"; import { generateSessionId } from "../session/index.js"; import { consumeStream } from "../session/stream-consumer.js"; import { createCycleTextRecorder } from "../session/stream-journal.js"; +import { onTurnBoundary } from "../agent/reactor-events.js"; import { refreshInferenceSourceBundle } from "./refresh-inference-source.js"; + import type { CapabilityFilter } from "../agent/profiles.js"; import type { Settings } from "../config/settings.js"; import { toolWatchdogFromSettings } from "../config/settings.js"; @@ -127,7 +129,14 @@ import { createSendInputTool, } from "./lifecycle-tools.js"; import { createSubAgentSessionStore } from "./session-store.js"; -import type { RunSubAgentParams, RunSubAgentResult, SubAgentProvider } from "./types.js"; +import type { + RunSubAgentParams, + RunSubAgentResult, + SubAgentProvider, + SubAgentTelemetryRollup, + SubAgentTerminalReason, +} from "./types.js"; + import type { TaskIntent } from "./report.js"; import { runWithSubAgentIdentity } from "./identity-context.js"; @@ -353,6 +362,51 @@ export function createCodexProxyRunTool(posixTools: CodexProxyToolRunner): Codex // gets its own posix tool instances and its own git-backed context store so // the two loops never trample each other's state. export async function runSubAgent(params: RunSubAgentParams): Promise { + const startedAt = Date.now(); + const telemetryRollup: SubAgentTelemetryRollup = { + turn_count: 0, + input_tokens: 0, + output_tokens: 0, + cache_read_tokens: 0, + cache_write_tokens: 0, + reasoning_tokens: 0, + tool_call_count: 0, + tool_error_count: 0, + }; + let terminalReason: SubAgentTerminalReason = "error"; + let errorCount = 0; + const settlementState = { latestModel: params.provider.model }; + + try { + const result = await runSubAgentInner(params, telemetryRollup, settlementState); + terminalReason = result.stopReason ?? "complete"; + return result; + } catch (error) { + errorCount = 1; + if (isSubAgentCancelError(error, params.signal)) terminalReason = "cancelled"; + throw error; + } finally { + try { + params.onRunSettled?.( + Object.freeze({ + ...telemetryRollup, + error_count: errorCount, + duration_ms: Date.now() - startedAt, + model: settlementState.latestModel, + terminal_reason: terminalReason, + }), + ); + } catch { + // Settlement is observational and must not change the run's outcome. + } + } +} + +async function runSubAgentInner( + params: RunSubAgentParams, + telemetryRollup: SubAgentTelemetryRollup, + settlementState: { latestModel: string }, +): Promise { await seedPricingMetadataFromCache({ cachePath: defaultPricingCachePath(), }); @@ -837,6 +891,10 @@ export async function runSubAgent(params: RunSubAgentParams): Promise ({ + ...result, + telemetry: { ...telemetryRollup }, + }); // Watch the streamed text of the in-flight cycle so a salvage on // cancel/deadline has the cycle's tail as its payload, even though no // turn boundary has completed yet to carry it. @@ -845,6 +903,7 @@ export async function runSubAgent(params: RunSubAgentParams): Promise }[] = []; + const telemetry: Telemetry = { + enabled: true, + installationId: "test", + capture: (event, properties = {}) => events.push({ event, properties }), + captureIntentional: () => false, + flush: async () => {}, + discard: () => {}, + }; + return { telemetry, events }; +} + const tempDirs: string[] = []; afterEach(async () => { @@ -53,6 +67,14 @@ async function pathExists(path: string): Promise { } } +async function waitFor(predicate: () => boolean | Promise): Promise { + for (let attempt = 0; attempt < 500; attempt++) { + if (await predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 1)); + } + throw new Error("condition was not reached"); +} + function deferred(): { promise: Promise; resolve: (v: T) => void; @@ -108,17 +130,20 @@ describe("spawn_agent worktree isolation", () => { tempDirs.push(workdirBase); let ran = false; + const { telemetry, events } = telemetryCapture(); + const sessions = createSubAgentSessionStore(); const tool = createSpawnAgentTool({ permissionGate: testPermissionGate, cwd: notARepo, getWorkdirBase: () => workdirBase, provider, useWorktree: true, + telemetry, run: async () => { ran = true; return { report: "no" }; }, - sessions: createSubAgentSessionStore(), + sessions, fleetRecords: createFleetRecords(), }); if (tool.kind !== "full") throw new Error("expected full tool"); @@ -132,6 +157,81 @@ describe("spawn_agent worktree isolation", () => { ); expect(result.isError).toBe(true); expect(ran).toBe(false); + expect(sessions.list()).toHaveLength(1); + expect(sessions.list()[0]?.status).toBe("failed"); + expect(events.filter((event) => event.event === "subagent_start")).toHaveLength(1); + const ends = events.filter((event) => event.event === "subagent_end"); + expect(ends).toHaveLength(1); + expect(ends[0]?.properties).toMatchObject({ + status: "failed", + stop_reason: "setup_error", + model: "test-model", + turn_count: 0, + input_tokens: 0, + output_tokens: 0, + cache_read_tokens: 0, + cache_write_tokens: 0, + reasoning_tokens: 0, + tool_call_count: 0, + tool_error_count: 0, + }); + expect(typeof ends[0]?.properties.duration_ms).toBe("number"); + }); + + test("pairs pre-progress cancellation with a cancelled terminal event", async () => { + const repo = await makeRepo(); + tempDirs.push(repo); + const { telemetry, events } = telemetryCapture(); + const sessions = createSubAgentSessionStore(); + const tool = createSpawnAgentTool({ + permissionGate: testPermissionGate, + cwd: repo, + getWorkdirBase: () => repo, + provider, + telemetry, + run: async (params) => { + params.onRunSettled?.({ + turn_count: 0, + input_tokens: 0, + output_tokens: 0, + cache_read_tokens: 0, + cache_write_tokens: 0, + reasoning_tokens: 0, + tool_call_count: 0, + tool_error_count: 0, + error_count: 1, + duration_ms: 1, + model: "test-model", + terminal_reason: "cancelled", + }); + const error = new Error("aborted"); + error.name = "AbortError"; + throw error; + }, + sessions, + fleetRecords: createFleetRecords(), + }); + if (tool.kind !== "full") throw new Error("expected full tool"); + + const result = await tool.handler( + { + id: "cancelled-spawn", + name: "spawn_agent", + arguments: { description: "cancelled", prompt: "Do the work", intent: "explore" }, + }, + new AbortController().signal, + ); + + expect(result.isError).not.toBe(true); + await waitFor(() => events.some((event) => event.event === "subagent_end")); + expect(sessions.list()[0]?.status).toBe("cancelled"); + expect(events.filter((event) => event.event === "subagent_start")).toHaveLength(1); + const ends = events.filter((event) => event.event === "subagent_end"); + expect(ends).toHaveLength(1); + expect(ends[0]?.properties).toMatchObject({ + status: "cancelled", + stop_reason: "cancelled", + }); }); test("defers worktree cleanup while the session is retained for followup", async () => { @@ -193,6 +293,9 @@ describe("spawn_agent worktree isolation", () => { const settle = deferred(); let workerCwd: string | undefined; + let settlementCount = 0; + let settlementWasFrozen = false; + const { telemetry, events } = telemetryCapture(); const sessions = createSubAgentSessionStore(); const tool = createSpawnAgentTool({ permissionGate: testPermissionGate, @@ -200,6 +303,7 @@ describe("spawn_agent worktree isolation", () => { getWorkdirBase: () => workdirBase, provider, useWorktree: true, + telemetry, run: async (params) => { workerCwd = params.cwd; params.onAgentReady?.({ @@ -208,7 +312,25 @@ describe("spawn_agent worktree isolation", () => { followup: async () => "", deliver: () => {}, }); - return settle.promise; + const result = await settle.promise; + const summary = Object.freeze({ + turn_count: 0, + input_tokens: 0, + output_tokens: 0, + cache_read_tokens: 0, + cache_write_tokens: 0, + reasoning_tokens: 0, + tool_call_count: 0, + tool_error_count: 0, + error_count: 0, + duration_ms: 1, + model: "test-model", + terminal_reason: "cancelled" as const, + }); + settlementCount += 1; + settlementWasFrozen = Object.isFrozen(summary); + params.onRunSettled?.(summary); + return result; }, sessions, fleetRecords: createFleetRecords(), @@ -227,10 +349,20 @@ describe("spawn_agent worktree isolation", () => { settle.resolve({ report: "## Summary\nStopped.\n## Findings\npartial\n## Blockers\ninterrupted\n## Paths\n", + stopReason: "cancelled", interrupted: true, }); - await new Promise((resolve) => setTimeout(resolve, 50)); + await waitFor(() => events.some((event) => event.event === "subagent_end")); + expect(settlementCount).toBe(1); + expect(settlementWasFrozen).toBe(true); + expect(sessions.get(agentId)?.lifecycleStatus).toBe("interrupted"); + const ends = events.filter((event) => event.event === "subagent_end"); + expect(ends).toHaveLength(1); + expect(ends[0]?.properties).toMatchObject({ + status: "interrupted", + stop_reason: "cancelled", + }); expect(workerCwd).toBeDefined(); expect(await pathExists(workerCwd!)).toBe(true); @@ -269,9 +401,11 @@ describe("spawn_agent worktree isolation", () => { }, new AbortController().signal, ); - await new Promise((resolve) => setTimeout(resolve, 50)); + await waitFor(() => workerCwd !== undefined); + if (workerCwd === undefined) throw new Error("worker cwd was not captured"); + const completedWorkerCwd = workerCwd; + await waitFor(async () => !(await pathExists(completedWorkerCwd))); - expect(workerCwd).toBeDefined(); - expect(await pathExists(workerCwd!)).toBe(false); + expect(await pathExists(completedWorkerCwd)).toBe(false); }); }); diff --git a/src/subagent/task-tool-worktree.test.ts b/src/subagent/task-tool-worktree.test.ts index 9d58a7740..f52e30535 100644 --- a/src/subagent/task-tool-worktree.test.ts +++ b/src/subagent/task-tool-worktree.test.ts @@ -8,6 +8,7 @@ import { promisify } from "node:util"; import { createTaskTool } from "./task-tool.js"; import type { RunSubAgentParams } from "./types.js"; import { createPermissionGate } from "../permission/gate.js"; +import type { Telemetry } from "../telemetry/index.js"; const run = promisify(execFile); @@ -23,6 +24,19 @@ const provider = { model: "test-model", }; +function telemetryCapture() { + const events: { event: string; properties: Record }[] = []; + const telemetry: Telemetry = { + enabled: true, + installationId: "test", + capture: (event, properties = {}) => events.push({ event, properties }), + captureIntentional: () => false, + flush: async () => {}, + discard: () => {}, + }; + return { telemetry, events }; +} + async function callTask( tool: ReturnType, args: Record, @@ -122,12 +136,14 @@ describe("createTaskTool worktree isolation", () => { tempDirs.push(workdirBase); let ran = false; + const { telemetry, events } = telemetryCapture(); const tool = createTaskTool({ permissionGate: testPermissionGate, cwd: notARepo, getWorkdirBase: () => workdirBase, provider, useWorktree: true, + telemetry, run: async () => { ran = true; return { report: "done" }; @@ -143,6 +159,70 @@ describe("createTaskTool worktree isolation", () => { expect(result).toContain("Error:"); expect(result).toContain("not inside a git repository"); expect(ran).toBe(false); + expect(events.filter((event) => event.event === "subagent_start")).toHaveLength(1); + const ends = events.filter((event) => event.event === "subagent_end"); + expect(ends).toHaveLength(1); + expect(ends[0]?.properties).toMatchObject({ + status: "failed", + stop_reason: "setup_error", + model: "test-model", + turn_count: 0, + input_tokens: 0, + output_tokens: 0, + cache_read_tokens: 0, + cache_write_tokens: 0, + reasoning_tokens: 0, + tool_call_count: 0, + tool_error_count: 0, + }); + expect(typeof ends[0]?.properties.duration_ms).toBe("number"); + }); + + test("pairs pre-progress cancellation with a cancelled terminal event", async () => { + const repo = await makeRepo(); + tempDirs.push(repo); + const { telemetry, events } = telemetryCapture(); + const tool = createTaskTool({ + permissionGate: testPermissionGate, + cwd: repo, + getWorkdirBase: () => repo, + provider, + telemetry, + run: async (params) => { + params.onRunSettled?.({ + turn_count: 0, + input_tokens: 0, + output_tokens: 0, + cache_read_tokens: 0, + cache_write_tokens: 0, + reasoning_tokens: 0, + tool_call_count: 0, + tool_error_count: 0, + error_count: 1, + duration_ms: 1, + model: "test-model", + terminal_reason: "cancelled", + }); + const error = new Error("aborted"); + error.name = "AbortError"; + throw error; + }, + }); + + const result = await callTask(tool, { + description: "cancelled job", + prompt: "Do the work", + intent: "explore", + }); + + expect(result).toContain("cancelled by operator"); + expect(events.filter((event) => event.event === "subagent_start")).toHaveLength(1); + const ends = events.filter((event) => event.event === "subagent_end"); + expect(ends).toHaveLength(1); + expect(ends[0]?.properties).toMatchObject({ + status: "cancelled", + stop_reason: "cancelled", + }); }); test("preserves a worktree the sub-agent left dirty, with a notice in the report", async () => { diff --git a/src/subagent/task-tool.ts b/src/subagent/task-tool.ts index f7b37662a..d64a24bb0 100644 --- a/src/subagent/task-tool.ts +++ b/src/subagent/task-tool.ts @@ -57,6 +57,10 @@ import { end, start } from "../perf/index.js"; 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 { SPAWN_AGENT_TOOL_NAME, TASK_TOOL_NAME } from "./tool-taxonomy.js"; + import { join } from "node:path"; import type { NestedDispatchDeps, @@ -64,6 +68,8 @@ import type { RunSubAgentResult, SubAgentProvider, SubAgentSandboxDeps, + SubAgentTelemetryRollup, + SubAgentTerminalReason, } from "./types.js"; const log = getLogger([LOG_NAMESPACE_ROOT, "subagent", "task-tool"]); @@ -87,7 +93,7 @@ export const TaskToolArgs = type({ // much still routes through it — but new work should reach for the split // verbs first. export const taskToolDefinition: ToolDefinition = { - name: "task", + name: TASK_TOOL_NAME, description: 'Deprecated: prefer spawn_agent + wait_agents for new call sites (this fused blocking form is kept for compatibility). Spawn a sub-agent (a short-lived child agent) for one self-contained job. This is not a checklist item — use manage_tasks for your own work list. The sub-agent has the full file, search, and shell toolset, uses this session\'s permission gate (saved grants and auto mode when eligible; you may be prompted for other consequential actions), and returns a structured report (Summary / Findings / Blockers / Paths). Use it to parallelize exploration ("map every caller of X") or hand off a well-scoped implementation so your own context stays focused. Fire several task calls in one turn to run sub-agents in parallel. When launching multiple agents with the same profile, assign each a distinct lens in description and prompt so they do not duplicate work. The sub-agent cannot ask you questions. Depending on dispatch configuration it either shares your working tree directly, or runs isolated in its own git worktree snapshotted from your last commit — in the isolated case, any uncommitted or untracked changes in your working tree are excluded. Write a clear brief: context = durable background; prompt = actionable goal; goals = optional manage_tasks seeds. Prefer the typed spawn contract so workers finish without thrashing: intent (explore|implement|review|plan|general), success_criteria (done-when checklist), do_not (scope fence), report_focus (what Findings must cover).', inputSchema: { @@ -326,7 +332,7 @@ async function runTaskViaFleet(input: { const started = await spawn.handler( { id: input.callId, - name: "spawn_agent", + name: SPAWN_AGENT_TOOL_NAME, arguments: { description: input.description, prompt: input.prompt, @@ -922,14 +928,18 @@ 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; + let endStopReason: SubAgentTerminalReason | "setup_error" | undefined; + let endModel: string | undefined; + try { if (deps.useWorktree === true) { const worktreePath = join(deps.getWorkdirBase(), "worktrees", generateSessionId()); @@ -946,6 +956,18 @@ export function createTaskTool(deps: TaskToolDeps): AgentTool { err instanceof WorktreeError ? err.message : `sub-agent worktree setup failed: ${err instanceof Error ? err.message : String(err)}`; + subagentStatus = "failed"; + endStopReason = "setup_error"; + endRollup = { + turn_count: 0, + input_tokens: 0, + output_tokens: 0, + cache_read_tokens: 0, + cache_write_tokens: 0, + reasoning_tokens: 0, + tool_call_count: 0, + tool_error_count: 0, + }; briefLedger.release(fingerprint); if (session !== undefined) deps.sessions?.fail(session.id, message); signal.removeEventListener("abort", onParentAbort); @@ -1009,6 +1031,11 @@ export function createTaskTool(deps: TaskToolDeps): AgentTool { ...(reportFocus !== undefined && reportFocus.length > 0 ? { reportFocus } : {}), signal: childCtl.signal, onEvent, + onRunSettled: (summary) => { + endRollup = summary; + endStopReason = summary.terminal_reason; + endModel = summary.model; + }, ...(deps.onProgress !== undefined ? { onProgress: deps.onProgress } : {}), ...(capabilities !== undefined ? { capabilities } : {}), ...(systemPromptRole !== undefined ? { systemPromptRole } : {}), @@ -1031,6 +1058,7 @@ export function createTaskTool(deps: TaskToolDeps): AgentTool { const result = await run(params); // Operator cancel may race after run resolves. Keep strip status cancelled // when requested, but never discard a returned body (including salvage). + const wasCancelled = childCtl.signal.aborted || (session !== undefined && deps.sessions?.get(session.id)?.status === "cancelled"); @@ -1119,10 +1147,14 @@ export function createTaskTool(deps: TaskToolDeps): AgentTool { } finally { activeLanes.delete(call.id); end(subagentSpanId); - telemetry.capture("subagent_end", { - agent_name: agentName, + captureSubagentEnd(telemetry, { + agentName, status: subagentStatus, - duration_ms: Date.now() - subagentStartedAt, + durationMs: Date.now() - subagentStartedAt, + ...(endModel !== undefined ? { model: endModel } : { model: provider.model }), + ...(endStopReason !== undefined ? { stopReason: endStopReason } : {}), + ...(endRollup !== undefined ? { rollup: endRollup } : {}), + ...(parentTraceId !== undefined ? { parentTraceId } : {}), }); } }, diff --git a/src/subagent/tool-taxonomy.ts b/src/subagent/tool-taxonomy.ts new file mode 100644 index 000000000..5cb399574 --- /dev/null +++ b/src/subagent/tool-taxonomy.ts @@ -0,0 +1,8 @@ +export const TASK_TOOL_NAME = "task"; +export const SPAWN_AGENT_TOOL_NAME = "spawn_agent"; + +const SUBAGENT_TOOL_NAMES = new Set([TASK_TOOL_NAME, SPAWN_AGENT_TOOL_NAME]); + +export function isSubagentToolName(toolName: string): boolean { + return SUBAGENT_TOOL_NAMES.has(toolName); +} diff --git a/src/subagent/types.ts b/src/subagent/types.ts index b0abc88c4..e8089fd9f 100644 --- a/src/subagent/types.ts +++ b/src/subagent/types.ts @@ -108,6 +108,7 @@ export type RunSubAgentParams = { signal?: AbortSignal; onEvent?: (event: ReactorEmittedEvent) => void; onProgress?: (info: { description: string; toolName: string }) => void; + onRunSettled?: (summary: Readonly) => void; capabilities?: CapabilityFilter; systemPromptRole?: string; /** Resolved closed-director id (e.g. "critic") when the worker is one. Structured gate key — prefer over persona-string matching in systemPromptRole. */ @@ -188,6 +189,26 @@ export type RunSubAgentParams = { } & SubAgentSandboxDeps; /** runSubAgent's result: the parent-facing report plus, when force-stopped, the structured reason why — classify outcomes from `stopReason`, not by parsing `report`. */ +export interface SubAgentTelemetryRollup { + turn_count: number; + input_tokens: number; + output_tokens: number; + cache_read_tokens: number; + cache_write_tokens: number; + reasoning_tokens: number; + tool_call_count: number; + tool_error_count: number; +} + +export type SubAgentTerminalReason = ForcedStopReason | "complete" | "error"; + +export interface SubAgentRunSettlement extends SubAgentTelemetryRollup { + error_count: number; + duration_ms: number; + model: string; + terminal_reason: SubAgentTerminalReason; +} + export interface RunSubAgentResult { report: string; stopReason?: ForcedStopReason; @@ -207,4 +228,9 @@ export interface RunSubAgentResult { * already transitioned the session to "interrupted" synchronously. */ interrupted?: boolean; + /** + * Counts accumulated during the run for ambient `subagent_end` telemetry. + * Never includes prompts, paths, or free-text ids. + */ + telemetry?: SubAgentTelemetryRollup; } diff --git a/src/telemetry/ai-observability.test.ts b/src/telemetry/ai-observability.test.ts index f6796af9f..f9c338370 100644 --- a/src/telemetry/ai-observability.test.ts +++ b/src/telemetry/ai-observability.test.ts @@ -1,8 +1,10 @@ -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, classifyErrorKind, classifySpanKind, createTurnObserver, @@ -11,10 +13,15 @@ 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 }[]; @@ -75,7 +82,7 @@ function fakeTurnContext(overrides: Partial = {}): TurnContext { } as TurnContext; } -const emitOptions = { sessionId: SESSION_ID, subagentToolName: SUBAGENT_TOOL_NAME }; +const emitOptions = { sessionId: SESSION_ID, env: {} }; const FAILED_TURN_SOURCE = { provider: "openai-compatible", model: "model-x" }; describe("secondsFromMs", () => { @@ -87,13 +94,14 @@ describe("secondsFromMs", () => { }); describe("classifySpanKind", () => { - test("classifies the subagent tool as subagent_call", () => { - expect(classifySpanKind(SUBAGENT_TOOL_NAME, SUBAGENT_TOOL_NAME)).toBe("subagent_call"); + test("classifies both subagent dispatch tools as subagent_call", () => { + expect(classifySpanKind("task")).toBe("subagent_call"); + expect(classifySpanKind("spawn_agent")).toBe("subagent_call"); }); test("classifies every other tool as tool_call, regardless of name", () => { - expect(classifySpanKind("read_file", SUBAGENT_TOOL_NAME)).toBe("tool_call"); - expect(classifySpanKind("mcp__acme__fetch_secret", SUBAGENT_TOOL_NAME)).toBe("tool_call"); + expect(classifySpanKind("read_file")).toBe("tool_call"); + expect(classifySpanKind("mcp__acme__fetch_secret")).toBe("tool_call"); }); }); @@ -137,6 +145,42 @@ describe("turnTraceId", () => { }); }); +describe("representative fleet event volume", () => { + 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); + }); +}); + +describe("aggregateToolCalls", () => { + test("counts tool calls, subagent calls, and errors separately", () => { + expect(aggregateToolCalls(fakeTurnContext())).toEqual({ + tool_call_count: 1, + tool_error_count: 1, + subagent_call_count: 1, + }); + }); +}); + describe("createTurnObserver", () => { // Regression: the trace id must be built from whatever session id is live // at emission. A call site that captured it once would keep filing turns @@ -149,7 +193,6 @@ describe("createTurnObserver", () => { telemetry: () => telemetry, getSessionId: () => sessionId, getSource: () => ({ provider: "openai-compatible", model: "model-x" }), - subagentToolName: SUBAGENT_TOOL_NAME, }); observer.onTurnComplete(fakeTurnContext({ turnIndex: 0, toolCalls: [], toolResults: [] })); @@ -167,7 +210,6 @@ describe("createTurnObserver", () => { telemetry: () => telemetry, getSessionId: () => sessionId, getSource: () => ({ provider: "openai-compatible", model: "model-x" }), - subagentToolName: SUBAGENT_TOOL_NAME, }); observer.onTurnFailed({ turnIndex: 0, error: "boom" }); @@ -187,7 +229,6 @@ describe("createTurnObserver", () => { telemetry: () => telemetry, getSessionId: () => SESSION_ID, getSource: () => source, - subagentToolName: SUBAGENT_TOOL_NAME, }); observer.onTurnFailed({ turnIndex: 0, error: "429 rate limit" }); @@ -199,14 +240,44 @@ 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" }), + }); + + observer.onTurnStarted({ turnIndex: 2, model: "model-x" }); + expect(getCurrentTurnTraceId()).toBe(`${SESSION_ID}:turn:2`); + observer.onTurnComplete(fakeTurnContext({ turnIndex: 2, toolCalls: [], toolResults: [] })); + expect(getCurrentTurnTraceId()).toBeUndefined(); + }); }); describe("emitAiObservability", () => { - test("emits one $ai_generation and one $ai_span per tool call", () => { + test("emits one $ai_generation with aggregates and zero $ai_span by default", () => { const { telemetry, captured } = fakeTelemetry(); emitAiObservability(telemetry, fakeTurnContext(), emitOptions); + expect(captured.length).toBe(1); + expect(captured[0]?.event).toBe("$ai_generation"); + expect(captured[0]?.properties.tool_call_count).toBe(1); + expect(captured[0]?.properties.tool_error_count).toBe(1); + expect(captured[0]?.properties.subagent_call_count).toBe(1); + expect(captured.filter((c) => c.event === "$ai_span")).toHaveLength(0); + }); + + test("restores per-call $ai_span when CORBITS_TELEMETRY_AI_SPANS is truthy", () => { + const { telemetry, captured } = fakeTelemetry(); + + emitAiObservability(telemetry, fakeTurnContext(), { + ...emitOptions, + env: { CORBITS_TELEMETRY_AI_SPANS: "1" }, + }); + expect(captured.length).toBe(3); expect(captured[0]?.event).toBe("$ai_generation"); expect(captured[1]?.event).toBe("$ai_span"); @@ -261,7 +332,10 @@ describe("emitAiObservability", () => { const { telemetry, captured } = fakeTelemetry(); const ctx = fakeTurnContext(); - emitAiObservability(telemetry, ctx, emitOptions); + emitAiObservability(telemetry, ctx, { + ...emitOptions, + env: { CORBITS_TELEMETRY_AI_SPANS: "1" }, + }); const traceId = turnTraceId(SESSION_ID, ctx.turnIndex); const generation = captured.find((c) => c.event === "$ai_generation"); @@ -279,7 +353,10 @@ describe("emitAiObservability", () => { test("names the span by fixed enum, never the raw tool name", () => { const { telemetry, captured } = fakeTelemetry(); - emitAiObservability(telemetry, fakeTurnContext(), emitOptions); + emitAiObservability(telemetry, fakeTurnContext(), { + ...emitOptions, + env: { CORBITS_TELEMETRY_AI_SPANS: "true" }, + }); const spans = captured.filter((c) => c.event === "$ai_span"); expect(spans[0]?.properties.$ai_span_name).toBe("tool_call"); @@ -293,7 +370,10 @@ describe("emitAiObservability", () => { test("propagates tool error state onto the span without the result content", () => { const { telemetry, captured } = fakeTelemetry(); - emitAiObservability(telemetry, fakeTurnContext(), emitOptions); + emitAiObservability(telemetry, fakeTurnContext(), { + ...emitOptions, + env: { CORBITS_TELEMETRY_AI_SPANS: "1" }, + }); const spans = captured.filter((c) => c.event === "$ai_span"); expect(spans[0]?.properties.$ai_is_error).toBe(false); @@ -303,7 +383,10 @@ describe("emitAiObservability", () => { test("never leaks prompt text, tool arguments, tool results, or file paths", () => { const { telemetry, captured } = fakeTelemetry(); - emitAiObservability(telemetry, fakeTurnContext(), emitOptions); + emitAiObservability(telemetry, fakeTurnContext(), { + ...emitOptions, + env: { CORBITS_TELEMETRY_AI_SPANS: "1" }, + }); const serialized = JSON.stringify(captured); expect(serialized).not.toContain("secret-project"); @@ -314,6 +397,38 @@ describe("emitAiObservability", () => { expect(serialized).not.toContain("here is the plan"); expect(serialized).not.toContain("/Users/attacker"); }); + + test("samples successful generations when CORBITS_TELEMETRY_GENERATION_SAMPLE_RATE is below 1", () => { + const { telemetry, captured } = fakeTelemetry(); + + emitAiObservability(telemetry, fakeTurnContext(), { + ...emitOptions, + env: { CORBITS_TELEMETRY_GENERATION_SAMPLE_RATE: "0.5" }, + random: () => 0.9, + }); + + expect(captured).toHaveLength(0); + }); + + test("always keeps a successful generation when the sample roll is under the rate", () => { + const { telemetry, captured } = fakeTelemetry(); + + emitAiObservability(telemetry, fakeTurnContext(), { + ...emitOptions, + env: { CORBITS_TELEMETRY_GENERATION_SAMPLE_RATE: "0.5" }, + random: () => 0.1, + }); + + 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", () => { @@ -350,21 +465,36 @@ describe("emitAiTurnFailure", () => { expect(captured[0]?.properties.$ai_model).toBe("model-x"); }); - test("never leaks the raw provider error message", () => { + test("errored generations always ship regardless of sample rate", () => { + const { telemetry, captured } = fakeTelemetry(); + + // emitAiTurnFailure has no sample gate — confirm directly. + emitAiTurnFailure(telemetry, { + sessionId: SESSION_ID, + turnIndex: 1, + source: FAILED_TURN_SOURCE, + error: "boom", + }); + + expect(captured).toHaveLength(1); + expect(captured[0]?.properties.$ai_is_error).toBe(true); + }); + + test("never forwards the provider's error message", () => { const { telemetry, captured } = fakeTelemetry(); + const message = + "POST https://api.acme.internal/v1/chat failed: prompt contained /Users/attacker/secret.md"; emitAiTurnFailure(telemetry, { sessionId: SESSION_ID, turnIndex: 7, source: FAILED_TURN_SOURCE, - error: - "429 rate limit on https://api.internal.acme.corp/v1/chat while reading /Users/attacker/secret-project/plan.md: XYZ-SECRET-123", + error: message, }); const serialized = JSON.stringify(captured); - expect(serialized).not.toContain("acme.corp"); + expect(serialized).not.toContain("acme.internal"); expect(serialized).not.toContain("/Users/attacker"); - expect(serialized).not.toContain("secret-project"); - expect(serialized).not.toContain("XYZ-SECRET-123"); + expect(serialized).not.toContain(message); }); }); diff --git a/src/telemetry/ai-observability.ts b/src/telemetry/ai-observability.ts index 8b1ca3742..d8707ac3b 100644 --- a/src/telemetry/ai-observability.ts +++ b/src/telemetry/ai-observability.ts @@ -1,12 +1,28 @@ -// Emits PostHog AI observability events ($ai_generation, $ai_span) from a -// completed turn, in the same privacy mode as product telemetry: only ids, -// enums, and counts ever leave the process. TurnContext already carries tool -// call arguments and results for lifecycle hooks — this module reads only +// Emits PostHog AI observability events ($ai_generation, optionally $ai_span) +// from a completed turn, in the same privacy mode as product telemetry: only +// ids, enums, and counts ever leave the process. TurnContext already carries +// tool call arguments and results for lifecycle hooks — this module reads only // the scalar/id fields off it and never the content fields. +// +// Default volume shape (CL-6816): one $ai_generation per turn with tool/subagent +// aggregates folded onto it. Per-call $ai_span events are opt-in via +// CORBITS_TELEMETRY_AI_SPANS for debugging. import type { TurnContext } from "../session/hooks.js"; -import { noteLastTurnTraceId } from "./feedback.js"; -import type { AiErrorKind, AiSpanKind, Telemetry } from "./index.js"; +import { isSubagentToolName } from "../subagent/tool-taxonomy.js"; +import { + noteLastTurnTraceId, + noteCurrentTurnTraceId, + clearCurrentTurnTraceId, +} from "./feedback.js"; + +import { + aiSpansEnabled, + generationSampleRate, + type AiErrorKind, + type AiSpanKind, + type Telemetry, +} from "./index.js"; // PostHog reports latency in seconds as a float; the runtime measures every // duration in milliseconds. Reporting milliseconds under the seconds-typed @@ -29,8 +45,8 @@ export function turnTraceId(sessionId: string, turnIndex: number): string { // (e.g. the subagent task tool's registered name) rather than reaching into // subagent internals, so this module has no dependency on tool // implementations beyond the one identifier it needs to classify. -export function classifySpanKind(toolName: string, subagentToolName: string): AiSpanKind { - return toolName === subagentToolName ? "subagent_call" : "tool_call"; +export function classifySpanKind(toolName: string): AiSpanKind { + return isSubagentToolName(toolName) ? "subagent_call" : "tool_call"; } // Word-bounded so a status code is only read where one was actually written. @@ -58,16 +74,50 @@ export function classifyErrorKind(message: string): AiErrorKind { export interface EmitAiObservabilityOptions { // The runtime's per-session id, which scopes the trace id. sessionId: string; - // Name of the tool that spawns a sub-agent, used to classify that call's - // span kind as "subagent_call" instead of the generic "tool_call". - subagentToolName: string; + /** Override process.env for tests. */ + env?: NodeJS.ProcessEnv; + /** Override Math.random for generation sampling tests. */ + random?: () => number; +} + +export interface ToolCallAggregates { + tool_call_count: number; + tool_error_count: number; + subagent_call_count: number; +} + +export function aggregateToolCalls( + ctx: Pick, +): ToolCallAggregates { + const resultsByCallId = new Map(ctx.toolResults.map((result) => [result.callId, result])); + let tool_call_count = 0; + let tool_error_count = 0; + let subagent_call_count = 0; + for (const call of ctx.toolCalls) { + const kind = classifySpanKind(call.name); + if (kind === "subagent_call") { + subagent_call_count += 1; + } else { + tool_call_count += 1; + } + if (resultsByCallId.get(call.id)?.isError === true) { + tool_error_count += 1; + } + } + return { tool_call_count, tool_error_count, subagent_call_count }; +} + +function shouldSampleSuccessfulGeneration(env: NodeJS.ProcessEnv, random: () => number): boolean { + const rate = generationSampleRate(env); + if (rate >= 1) return true; + if (rate <= 0) return false; + return random() < rate; } // Called once per completed turn. Emits one $ai_generation for the model -// call, then one $ai_span per tool call in the turn, all sharing the turn's -// $ai_trace_id. The trace is flat by construction: every span's -// $ai_parent_id is the trace id, which PostHog accepts, and TurnContext only -// exposes top-level tool calls so there is no nesting to describe. PostHog +// call with tool/subagent aggregates. Per-call $ai_span events are emitted +// only when CORBITS_TELEMETRY_AI_SPANS is truthy. The trace is flat by +// construction: every span's $ai_parent_id is the trace id. PostHog // synthesises the trace itself from these children, so no $ai_trace event is // emitted. export function emitAiObservability( @@ -75,11 +125,21 @@ export function emitAiObservability( ctx: TurnContext, options: EmitAiObservabilityOptions, ): void { + const env = options.env ?? process.env; + const random = options.random ?? Math.random; const traceId = turnTraceId(options.sessionId, ctx.turnIndex); // Remember for intentional /feedback linking (works even when ambient capture // is a no-op because this call still computes the id). noteLastTurnTraceId(traceId); + 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); + telemetry.capture("$ai_generation", { $ai_trace_id: traceId, // The canonical provider kind, never ctx.source.sourceId: sourceId is the @@ -94,8 +154,13 @@ export function emitAiObservability( $ai_cache_read_input_tokens: ctx.usage.cacheRead, $ai_cache_creation_input_tokens: ctx.usage.cacheWrite, $ai_reasoning_tokens: ctx.usage.thinking, + tool_call_count: aggregates.tool_call_count, + tool_error_count: aggregates.tool_error_count, + subagent_call_count: aggregates.subagent_call_count, }); + if (!aiSpansEnabled(env)) return; + const resultsByCallId = new Map(ctx.toolResults.map((result) => [result.callId, result])); for (const call of ctx.toolCalls) { @@ -106,7 +171,7 @@ export function emitAiObservability( // send: it identifies the call within the trace and nothing else. $ai_span_id: call.id, $ai_parent_id: traceId, - $ai_span_name: classifySpanKind(call.name, options.subagentToolName), + $ai_span_name: classifySpanKind(call.name), $ai_is_error: result?.isError === true, }); } @@ -135,6 +200,7 @@ export interface EmitAiTurnFailureOptions { // observability earns its keep and where a completion-only emitter is // silent. Emits the $ai_generation the turn never got to emit, marked as an // error, with no token counts or latency because the turn produced none. +// Errored generations always ship (no sampling). export function emitAiTurnFailure(telemetry: Telemetry, options: EmitAiTurnFailureOptions): void { telemetry.capture("$ai_generation", { $ai_trace_id: turnTraceId(options.sessionId, options.turnIndex), @@ -153,30 +219,68 @@ export interface CreateTurnObserverOptions { // session id in this same process, and a trace id built from a captured // one would file the new session's turns under the old session's traces. getSessionId: () => string; - // The source the next inference will run against, which is the best - // available attribution for a turn that failed before producing one. + // The currently selected source. It is safe failure attribution only when + // its model matches the model named by the latest inference.start. getSource: () => TurnSource; - subagentToolName: string; } -// Binds the emitters to the live session and source, giving the run sink two -// plain callbacks and keeping the "read it now, do not capture it" rule in -// one place instead of at each call site. +const UNKNOWN_INFERENCE_PROVIDER = "unknown"; + +function failedAttemptSource( + attemptedModel: string | undefined, + observedSource: TurnSource | undefined, + selectedSource: TurnSource, +): TurnSource { + if (observedSource !== undefined) return observedSource; + if (attemptedModel === undefined || attemptedModel === selectedSource.model) { + return selectedSource; + } + return { provider: UNKNOWN_INFERENCE_PROVIDER, model: attemptedModel }; +} + +// Binds the emitters to the live session and source, 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; model: string }) => void; + onTurnSourceObserved: (info: { turnIndex: number; source: TurnSource }) => void; onTurnComplete: (ctx: TurnContext) => void; onTurnFailed: (info: { turnIndex: number; error: string }) => void; } { + let latestAttemptModel: string | undefined; + let latestAttemptSource: TurnSource | undefined; + + function clearAttempt(): void { + latestAttemptModel = undefined; + latestAttemptSource = undefined; + } + return { + onTurnStarted: (info) => { + latestAttemptModel = info.model; + latestAttemptSource = undefined; + noteCurrentTurnTraceId(turnTraceId(options.getSessionId(), info.turnIndex)); + }, + onTurnSourceObserved: (info) => { + latestAttemptSource = { ...info.source }; + }, onTurnComplete: (ctx) => { + clearAttempt(); + clearCurrentTurnTraceId(); emitAiObservability(options.telemetry(), ctx, { sessionId: options.getSessionId(), - subagentToolName: options.subagentToolName, }); }, onTurnFailed: (info) => { + clearCurrentTurnTraceId(); + const source = failedAttemptSource( + latestAttemptModel, + latestAttemptSource, + options.getSource(), + ); + clearAttempt(); emitAiTurnFailure(options.telemetry(), { sessionId: options.getSessionId(), - source: options.getSource(), + source, ...info, }); }, diff --git a/src/telemetry/classify.ts b/src/telemetry/classify.ts index baf6b434f..00b58fe3c 100644 --- a/src/telemetry/classify.ts +++ b/src/telemetry/classify.ts @@ -10,6 +10,7 @@ // it matches nothing. What leaves the process is a first-party enum: the fact // that something unrecognised was used, never what it was called. +import { DIRECTOR_IDS } from "../agent/directors/types.js"; import { isMcpToolName } from "../mcp/tool-name.js"; const CUSTOM = "custom"; @@ -60,9 +61,10 @@ const BUILT_IN_COMMAND_NAMES: ReadonlySet = new Set([ "status", ]); -// The one agent label the runtime supplies itself; every other profile id -// comes from a project or plugin directory. -const BUILT_IN_AGENT_NAME = "worker"; +// First-party director ids from the closed fleet package, plus the legacy +// "worker" label the runtime still supplies as a fallback. Project/plugin +// profile ids are never reported by name. +const BUILT_IN_AGENT_NAMES: ReadonlySet = new Set([...DIRECTOR_IDS, "worker"]); // Error constructors defined by the language. A subclass name is application // or plugin code and can be as identifying as any other author-chosen string. @@ -90,7 +92,7 @@ export function classifyCommandName(commandName: string): string { } export function classifyAgentName(agentName: string): string { - return agentName === BUILT_IN_AGENT_NAME ? BUILT_IN_AGENT_NAME : CUSTOM; + return BUILT_IN_AGENT_NAMES.has(agentName) ? agentName : CUSTOM; } export function classifyErrorClass(error: unknown): string { diff --git a/src/telemetry/feedback.ts b/src/telemetry/feedback.ts index 93a6c409c..c52900cd7 100644 --- a/src/telemetry/feedback.ts +++ b/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; } diff --git a/src/telemetry/index.ts b/src/telemetry/index.ts index b62ccb606..e477fe5cf 100644 --- a/src/telemetry/index.ts +++ b/src/telemetry/index.ts @@ -12,8 +12,11 @@ const DEFAULT_POSTHOG_API_KEY = "phc_BWpXcEx3XBH2EiuNi3fXrdzfgnfbVe4WbVyfR8r5KbL const TELEMETRY_HOST_ENV = `${ENV_PREFIX}TELEMETRY_HOST`; const TELEMETRY_KEY_ENV = `${ENV_PREFIX}TELEMETRY_KEY`; export const TELEMETRY_ENV = `${ENV_PREFIX}TELEMETRY`; +export const TELEMETRY_AI_SPANS_ENV = `${ENV_PREFIX}TELEMETRY_AI_SPANS`; +export const TELEMETRY_GENERATION_SAMPLE_RATE_ENV = `${ENV_PREFIX}TELEMETRY_GENERATION_SAMPLE_RATE`; export const POSTHOG_HOST = process.env[TELEMETRY_HOST_ENV] ?? DEFAULT_POSTHOG_HOST; + export const POSTHOG_API_KEY = process.env[TELEMETRY_KEY_ENV] ?? DEFAULT_POSTHOG_API_KEY; // Upper bound on how long flush() may hold up process exit; anything still @@ -128,7 +131,13 @@ const EVENT_PROPERTY_ALLOWLIST: Record = { "$ai_cache_read_input_tokens", "$ai_cache_creation_input_tokens", "$ai_reasoning_tokens", + // Aggregates folded from per-call spans (CL-6816). Custom properties — + // PostHog LLM cost views still only query the $ai_*-prefixed fields above. + "tool_call_count", + "tool_error_count", + "subagent_call_count", ], + // The trace is flat: every span's $ai_parent_id is the turn's // $ai_trace_id. PostHog documents $ai_parent_id as accepting a trace id or // another span id, so a flat trace is legal and it is all the runtime can @@ -147,7 +156,23 @@ const EVENT_PROPERTY_ALLOWLIST: Record = { // author-chosen free text and is not sent. plugin_loaded: ["origin"], subagent_start: ["agent_name"], - subagent_end: ["agent_name", "status", "duration_ms"], + subagent_end: [ + "agent_name", + "status", + "duration_ms", + "model", + "turn_count", + "input_tokens", + "output_tokens", + "cache_read_tokens", + "cache_write_tokens", + "reasoning_tokens", + "tool_call_count", + "tool_error_count", + "stop_reason", + "parent_trace_id", + ], + permission_prompt: ["decision", "permission_kind"], compaction: ["mode", "duration_ms", "turns_before", "turns_after"], crash: ["kind", "error_class"], @@ -164,11 +189,32 @@ const FALSY_ENV_FLAG_VALUES = new Set(["", "0", "false", "off", "no"]); // Trimmed so .env files and shell scripts that produce " 0" or "false\n" // still count as an opt-out — opt-out parsing must fail toward disabled. -function truthyEnvFlag(value: string | undefined): boolean { +export function truthyEnvFlag(value: string | undefined): boolean { if (value === undefined) return false; return !FALSY_ENV_FLAG_VALUES.has(value.trim().toLowerCase()); } +/** Opt-in debug: emit per-call `$ai_span` events alongside generation aggregates. */ +export function aiSpansEnabled(env: NodeJS.ProcessEnv = process.env): boolean { + return truthyEnvFlag(env[TELEMETRY_AI_SPANS_ENV]); +} + +/** + * Sample rate for successful `$ai_generation` events (0–1). Default 1.0 (no + * drop). Errors (`$ai_is_error: true`), `crash`, and `auth_failure` always ship. + */ +export function generationSampleRate(env: NodeJS.ProcessEnv = process.env): number { + const raw = env[TELEMETRY_GENERATION_SAMPLE_RATE_ENV]; + if (raw === undefined) return 1; + 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)); +} + // Env kills win over everything and require no settings at all — callers use // this to skip settings writes (installationId generation) entirely. // CORBITS_TELEMETRY set to any falsy value ("0", "false", "off", "") @@ -340,7 +386,11 @@ export function createTelemetry(options: CreateTelemetryOptions): Telemetry { return running; } - function enqueue(event: TelemetryEvent, properties?: Record): void { + function enqueue( + event: TelemetryEvent, + properties: Record | undefined, + mode: "ambient" | "intentional", + ): void { // Own-property only: `in` walks Object.prototype, so capture("toString") // or capture("constructor") would clear the guard this exists to be and // hand allowedProperties a function where it expects an allowlist array. @@ -351,6 +401,12 @@ export function createTelemetry(options: CreateTelemetryOptions): Telemetry { timestamp: new Date().toISOString(), properties: { ...allowedProperties(event, properties), + // Ambient product/AI events are anonymous: PostHog's batch API + // defaults to identified processing, so stamp this explicitly + // (https://posthog.com/docs/data/anonymous-vs-identified-events). + // Intentional /feedback may stay identified so a survey can join a + // person profile if one is ever created. + ...(mode === "ambient" ? { $process_person_profile: false } : {}), // PostHog's built-in Version breakdown reads $app_version; without it // every event buckets as "Other". service_version is the same value // kept for dashboards that already filter on the custom property. @@ -384,7 +440,7 @@ export function createTelemetry(options: CreateTelemetryOptions): Telemetry { function capture(event: TelemetryEvent, properties?: Record): void { if (!enabled) return; - enqueue(event, properties); + enqueue(event, properties, "ambient"); } function captureIntentional( @@ -395,7 +451,7 @@ export function createTelemetry(options: CreateTelemetryOptions): Telemetry { // must never ride the ambient-bypass path. if (event !== "survey sent") return false; if (!intentionalEnabled) return false; - enqueue(event, properties); + enqueue(event, properties, "intentional"); return true; } diff --git a/src/telemetry/product-events.ts b/src/telemetry/product-events.ts index 37ae53cb4..c8e56ef65 100644 --- a/src/telemetry/product-events.ts +++ b/src/telemetry/product-events.ts @@ -1,6 +1,7 @@ // Shared product-event emitters that every surface (TUI, exec, future // headless) must call so dashboards are not silently TUI-only. +import type { SubAgentTelemetryRollup, SubAgentTerminalReason } from "../subagent/types.js"; import type { Telemetry } from "./index.js"; import { classifyCommandName } from "./classify.js"; @@ -10,3 +11,63 @@ export function captureSlashCommand(telemetry: Telemetry, commandName: string): command_name: classifyCommandName(commandName), }); } + +export interface CaptureSubagentEndArgs { + agentName: string; + status: string; + durationMs: number; + /** Canonical model id from the provider, never a free-text source label. */ + model?: string; + stopReason?: SubAgentTerminalReason | "setup_error"; + rollup?: SubAgentTelemetryRollup; + /** + * 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 { + const props: Record = { + agent_name: args.agentName, + status: args.status, + duration_ms: args.durationMs, + }; + if (args.model !== undefined && args.model.length > 0) { + props.model = args.model; + } + if (args.stopReason !== undefined) { + props.stop_reason = args.stopReason; + } + 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; + props.input_tokens = args.rollup.input_tokens; + props.output_tokens = args.rollup.output_tokens; + props.cache_read_tokens = args.rollup.cache_read_tokens; + props.cache_write_tokens = args.rollup.cache_write_tokens; + props.reasoning_tokens = args.rollup.reasoning_tokens; + props.tool_call_count = args.rollup.tool_call_count; + props.tool_error_count = args.rollup.tool_error_count; + } + return props; +} + +/** Emit `subagent_end` with rollup fields when available. */ +export function captureSubagentEnd(telemetry: Telemetry, args: CaptureSubagentEndArgs): void { + telemetry.capture("subagent_end", buildSubagentEndProperties(args)); +} + +export type PluginLoadReporter = (telemetry: Telemetry, origin: string, identity: string) => void; + +export function createPluginLoadReporter(): PluginLoadReporter { + const loadedPluginIdentities = new Set(); + return (telemetry, origin, identity) => { + if (!telemetry.enabled || identity.length === 0 || loadedPluginIdentities.has(identity)) return; + telemetry.capture("plugin_loaded", { origin }); + loadedPluginIdentities.add(identity); + }; +} diff --git a/src/telemetry/singleton.ts b/src/telemetry/singleton.ts index 6777ff4f9..4697e0cc2 100644 --- a/src/telemetry/singleton.ts +++ b/src/telemetry/singleton.ts @@ -1,4 +1,5 @@ import { NOOP_TELEMETRY, type Telemetry } from "./index.js"; +import { createPluginLoadReporter } from "./product-events.js"; // Process-wide telemetry handle. index.ts constructs the real instance once // at startup; runner.ts and the /settings Telemetry tab read it from here rather @@ -7,6 +8,8 @@ import { NOOP_TELEMETRY, type Telemetry } from "./index.js"; // tests) never throws. let instance: Telemetry = NOOP_TELEMETRY; +export const runtimePluginLoadReporter = createPluginLoadReporter(); + export function setTelemetry(telemetry: Telemetry): void { instance = telemetry; } diff --git a/src/tui/runner.ts b/src/tui/runner.ts index 008a494c5..16f343442 100644 --- a/src/tui/runner.ts +++ b/src/tui/runner.ts @@ -151,7 +151,6 @@ import { FLEET_STALL_POLL_MS, liveFleetCount, observeFleet, - taskToolDefinition, } from "../subagent/index.js"; import type { ContextStore, @@ -1584,13 +1583,14 @@ export async function runTUI(initialConfig: Config): Promise { telemetry: getTelemetry, getSessionId: () => sessionId, getSource: () => liveSource, - subagentToolName: taskToolDefinition.name, }); const runSink = createRunSink({ emitter, hookManager, initialTurnCount: resumeSeed.turnsUsed, + onTurnStarted: turnObserver.onTurnStarted, + onTurnSourceObserved: turnObserver.onTurnSourceObserved, onTurnComplete: turnObserver.onTurnComplete, onTurnFailed: turnObserver.onTurnFailed, // persistRunSnapshot is defined below but not invoked until the stream diff --git a/tests/unit/telemetry-product-events.test.ts b/tests/unit/telemetry-product-events.test.ts index 7d254bb4d..06b43b1f7 100644 --- a/tests/unit/telemetry-product-events.test.ts +++ b/tests/unit/telemetry-product-events.test.ts @@ -22,8 +22,17 @@ import { classifyPermissionKind, } from "../../src/telemetry/classify.js"; -import { createTelemetry, type Telemetry } from "../../src/telemetry/index.js"; -import { captureSlashCommand } from "../../src/telemetry/product-events.js"; +import { createTelemetry, NOOP_TELEMETRY, type Telemetry } from "../../src/telemetry/index.js"; +import { + buildSubagentEndProperties, + captureSlashCommand, + createPluginLoadReporter, +} from "../../src/telemetry/product-events.js"; +import { + noteCurrentTurnTraceId, + noteLastTurnTraceId, + resetFeedbackStateForTests, +} from "../../src/telemetry/feedback.js"; import { captureAuthFailure, classifyAgentSendFailure } from "../../src/tui/session-chrome.js"; interface BatchBody { @@ -59,6 +68,7 @@ function harness(): { const tempDirs: string[] = []; afterEach(async () => { + resetFeedbackStateForTests(); while (tempDirs.length > 0) { await rm(tempDirs.pop()!, { recursive: true, force: true }); } @@ -170,7 +180,12 @@ test("plugin_loaded carries only the discovery origin, never the manifest id", a "---\ndescription: ship it\n---\n\nShip.\n", ); - const mod = await loadPluginEntry(pluginDir, { cwd: root, origin: "project", telemetry }); + const mod = await loadPluginEntry(pluginDir, { + cwd: root, + origin: "project", + telemetry, + pluginLoadReporter: createPluginLoadReporter(), + }); expect(mod).not.toBeNull(); const [event] = await events(); @@ -180,6 +195,46 @@ test("plugin_loaded carries only the discovery origin, never the manifest id", a expect(await wire()).not.toContain("acmecorp"); }); +test("disabled plugin reporting does not consume the enabled dedupe identity", async () => { + const { telemetry, events } = harness(); + const reporter = createPluginLoadReporter(); + + reporter(NOOP_TELEMETRY, "project", "/plugin/acme"); + reporter(telemetry, "project", "/plugin/acme"); + reporter(telemetry, "project", "/plugin/acme"); + + expect((await events()).filter((event) => event.event === "plugin_loaded")).toHaveLength(1); +}); + +test("plugin_loaded emits once per plugin identity in-process", async () => { + const { telemetry, events } = harness(); + const root = await tempDir("corbits-plugin-dedupe-"); + const pluginDir = join(root, "plugin"); + await mkdir(join(pluginDir, "commands"), { recursive: true }); + await writeFile( + join(pluginDir, "plugin.json"), + JSON.stringify({ + id: "acmecorp/internal-tools", + name: "acmecorp internal tools", + version: "1.0.0", + }), + ); + await writeFile( + join(pluginDir, "commands", "ship.md"), + "---\ndescription: ship it\n---\n\nShip.\n", + ); + + const pluginLoadReporter = createPluginLoadReporter(); + const options = { cwd: root, origin: "project" as const, telemetry, pluginLoadReporter }; + const first = await loadPluginEntry(pluginDir, options); + const second = await loadPluginEntry(pluginDir, options); + expect(first).not.toBeNull(); + expect(second).not.toBeNull(); + + const captured = await events(); + expect(captured.filter((e) => e.event === "plugin_loaded")).toHaveLength(1); +}); + // --------------------------------------------------------------------------- // 4. agent_name — agent profiles are user-definable per project // --------------------------------------------------------------------------- @@ -197,7 +252,23 @@ test('subagent events bucket a project-defined profile id to "custom"', async () profiles: [ { id: "acmecorp-release-captain", description: "release", systemPromptRole: "release" }, ], - run: async () => ({ report: "done" }), + run: async (params) => { + params.onRunSettled?.({ + turn_count: 2, + input_tokens: 10, + output_tokens: 5, + cache_read_tokens: 1, + cache_write_tokens: 0, + reasoning_tokens: 0, + tool_call_count: 3, + tool_error_count: 1, + error_count: 0, + duration_ms: 10, + model: "test-model", + terminal_reason: "deadline", + }); + return { report: "done", stopReason: "deadline" }; + }, telemetry, }); if (tool.kind !== "full") throw new Error(`expected full tool, got ${tool.kind}`); @@ -216,15 +287,130 @@ test('subagent events bucket a project-defined profile id to "custom"', async () expect(names).toContain("subagent_end"); for (const event of captured) { expect(event.properties.agent_name).toBe("custom"); - // Sub-agents run in this process on the same session id, so a parent id - // would only ever restate session_id. expect(event.properties.parent_session_id).toBeUndefined(); } + const end = captured.find((e) => e.event === "subagent_end"); + expect(end?.properties.model).toBe("test-model"); + expect(end?.properties.turn_count).toBe(2); + expect(end?.properties.tool_call_count).toBe(3); + expect(end?.properties.tool_error_count).toBe(1); + expect(end?.properties.input_tokens).toBe(10); + expect(end?.properties.output_tokens).toBe(5); + expect(end?.properties.stop_reason).toBe("deadline"); + // No in-flight parent turn was noted — omit rather than invent. + expect(end?.properties.parent_trace_id).toBeUndefined(); expect(await wire()).not.toContain("acmecorp"); }); -test("the built-in worker label is reported by name", () => { +test("subagent_end parent_trace_id is the in-flight turn at spawn, not the last completed turn", async () => { + const { telemetry, events } = harness(); + const cwd = await tempDir("corbits-parent-trace-"); + const gate = createPermissionGate({ approvals: [], interactive: false, skipPermissions: true }); + + // Completed turn 0 is already "last" — spawn happens during turn 1. + noteLastTurnTraceId("sess:turn:0"); + noteCurrentTurnTraceId("sess:turn:1"); + + const tool = createTaskTool({ + cwd, + getWorkdirBase: () => cwd, + permissionGate: gate, + provider: { providerName: "test-provider", baseURL: "http://localhost", model: "test-model" }, + profiles: [ + { id: "acmecorp-release-captain", description: "release", systemPromptRole: "release" }, + ], + run: async () => ({ report: "done" }), + telemetry, + }); + if (tool.kind !== "full") throw new Error(`expected full tool, got ${tool.kind}`); + await tool.handler( + { + id: "call-1", + name: "task", + arguments: { description: "Ship", prompt: "Ship it", agent: "acmecorp-release-captain" }, + }, + new AbortController().signal, + ); + + const end = (await events()).find((e) => e.event === "subagent_end"); + expect(end).toBeDefined(); + expect(end?.properties.parent_trace_id).toBe("sess:turn:1"); + expect(end?.properties.parent_trace_id).not.toBe("sess:turn:0"); +}); + +test("buildSubagentEndProperties shapes rollup fields and omits empty parentTraceId", () => { + const withRollup = buildSubagentEndProperties({ + agentName: "builder", + status: "completed", + durationMs: 42, + model: "gpt-test", + stopReason: "deadline", + parentTraceId: "sess:turn:3", + rollup: { + turn_count: 4, + input_tokens: 100, + output_tokens: 50, + cache_read_tokens: 2, + cache_write_tokens: 1, + reasoning_tokens: 3, + tool_call_count: 7, + tool_error_count: 1, + }, + }); + expect(withRollup).toEqual({ + agent_name: "builder", + status: "completed", + duration_ms: 42, + model: "gpt-test", + stop_reason: "deadline", + parent_trace_id: "sess:turn:3", + turn_count: 4, + input_tokens: 100, + output_tokens: 50, + cache_read_tokens: 2, + cache_write_tokens: 1, + reasoning_tokens: 3, + tool_call_count: 7, + tool_error_count: 1, + }); + + const bare = buildSubagentEndProperties({ + agentName: "custom", + status: "failed", + durationMs: 1, + parentTraceId: "", + }); + expect(bare.parent_trace_id).toBeUndefined(); + expect(bare.turn_count).toBeUndefined(); + // Must not invent a parent from last-completed feedback state. + noteLastTurnTraceId("sess:turn:99"); + expect( + buildSubagentEndProperties({ + agentName: "custom", + status: "completed", + durationMs: 1, + }).parent_trace_id, + ).toBeUndefined(); +}); + +test("first-party director ids are reported by name; unknown profiles stay custom", () => { expect(classifyAgentName("worker")).toBe("worker"); + expect(classifyAgentName("builder")).toBe("builder"); + expect(classifyAgentName("skywalker")).toBe("skywalker"); + expect(classifyAgentName("greybeard")).toBe("greybeard"); + expect(classifyAgentName("explorer")).toBe("explorer"); + expect(classifyAgentName("counsel")).toBe("counsel"); + expect(classifyAgentName("critic")).toBe("critic"); + expect(classifyAgentName("intern")).toBe("intern"); + expect(classifyAgentName("tester")).toBe("tester"); + expect(classifyAgentName("testsmith")).toBe("testsmith"); + expect(classifyAgentName("shakespeare")).toBe("shakespeare"); + expect(classifyAgentName("rand")).toBe("rand"); + expect(classifyAgentName("draper")).toBe("draper"); + expect(classifyAgentName("emil")).toBe("emil"); + expect(classifyAgentName("gaasbot")).toBe("gaasbot"); + expect(classifyAgentName("bruckheimer")).toBe("bruckheimer"); + expect(classifyAgentName("neckbeard")).toBe("neckbeard"); expect(classifyAgentName("acmecorp-release-captain")).toBe("custom"); }); diff --git a/tests/unit/telemetry.test.ts b/tests/unit/telemetry.test.ts index 61c50bc2b..f038a881d 100644 --- a/tests/unit/telemetry.test.ts +++ b/tests/unit/telemetry.test.ts @@ -262,6 +262,7 @@ test("capture payload shape includes distinct_id and common props, with no clien expect(body.properties.distinct_id).toBe("my-install-id"); expect(typeof body.timestamp).toBe("string"); expect(body.properties.$geoip_disable).toBeUndefined(); + expect(body.properties.$process_person_profile).toBe(false); expect(body.properties.schema_version).toBe(1); expect(typeof body.properties.service_version).toBe("string"); expect(body.properties.$app_version).toBe(body.properties.service_version); @@ -271,6 +272,28 @@ test("capture payload shape includes distinct_id and common props, with no clien expect(body.properties.os_arch).toBe(process.arch); }); +test("intentional survey capture omits anonymous person-processing flag", async () => { + const { impl, events } = recordingFetch(); + const telemetry = createTelemetry({ + settings: settingsWith("my-install-id", false), + env: {}, + fetchFn: impl, + apiKey: "test-key", + }); + expect(telemetry.enabled).toBe(false); + expect( + telemetry.captureIntentional("survey sent", { + $survey_id: "s", + $survey_response: "hello", + $survey_questions: [], + }), + ).toBe(true); + await telemetry.flush(); + const body = events()[0]!; + expect(body.event).toBe("survey sent"); + expect(body.properties.$process_person_profile).toBeUndefined(); +}); + test("flush resolves after pending captures settle", async () => { let resolveFetch: (() => void) | undefined; const impl = (() =>