Add permission.wait and subagent spans for attribution

Author: TheGreatAxios

src/perf/index.test.ts

@@ -243,6 +243,7 @@ describe("sanitizeTags", () => {
       provider_id: "openai",
       model_id: "gpt-5.4",
       transport: "ws",
+      decision: "allow",
       duration_ms: 12.5,
       bytes: 1024,
       payload_bytes: 2048,
@@ -257,6 +258,7 @@ describe("sanitizeTags", () => {
       provider_id: "openai",
       model_id: "gpt-5.4",
       transport: "ws",
+      decision: "allow",
       duration_ms: 12.5,
       bytes: 1024,
       payload_bytes: 2048,
@@ -269,6 +271,12 @@ describe("sanitizeTags", () => {
     });
   });
 
+  test("keeps decision allow/deny and strips free-text decisions", () => {
+    expect(sanitizeTags({ decision: "allow" })).toEqual({ decision: "allow" });
+    expect(sanitizeTags({ decision: "deny" })).toEqual({ decision: "deny" });
+    expect(sanitizeTags({ decision: "maybe" })).toBeUndefined();
+  });
+
   test("strips free-text, paths, prompts, and unknown keys", () => {
     const tags = sanitizeTags({
       prompt: "system: you are a helpful assistant",

src/perf/index.ts

@@ -18,6 +18,7 @@ export {
   OPAQUE_ID_RE,
   ALLOWED_TAG_KEYS,
   type AllowedTagKey,
+  type DecisionKind,
   type PerfTags,
   type TransportKind,
 } from "./sanitize.js";

src/perf/permission-subagent-spans.test.ts

@@ -0,0 +1,248 @@
+/**
+ * CL-5170: permission.wait and subagent spans at the ask gate and task fleet.
+ */
+import { afterEach, describe, expect, test } from "bun:test";
+import type { ReactorEmittedEvent } from "@intx/inference";
+import { createPermissionGate } from "../permission/gate.js";
+import { createTaskTool } from "../subagent/task-tool.js";
+import { clear, snapshot, type PerfSpan } from "./index.js";
+import { createPerfReactorObserver } from "./reactor-spans.js";
+
+afterEach(() => {
+  clear();
+});
+
+function byName(spans: PerfSpan[], name: string): PerfSpan[] {
+  return spans.filter((s) => s.name === name);
+}
+
+function completed(spans: PerfSpan[]): PerfSpan[] {
+  return spans.filter((s) => s.endNs !== undefined);
+}
+
+function event(type: string, data: unknown = {}): ReactorEmittedEvent {
+  return { type, seq: 1, data } as ReactorEmittedEvent;
+}
+
+const shellCall = (command: string) =>
+  ({ id: "c1", name: "run_shell", arguments: { command } }) as const;
+
+const provider = {
+  providerName: "test-provider",
+  baseURL: "http://localhost",
+  model: "test-model",
+};
+
+const skipGate = createPermissionGate({
+  approvals: [],
+  interactive: false,
+  skipPermissions: true,
+});
+
+describe("permission.wait spans", () => {
+  test("records allow decision when operator approves a shell ask", async () => {
+    const gate = createPermissionGate({
+      approvals: [],
+      interactive: true,
+      skipPermissions: false,
+      requestApproval: async () => ({ allow: true }),
+    });
+
+    const verdict = await gate.evaluate(shellCall("curl example.com"));
+    expect(verdict.allowed).toBe(true);
+
+    const waits = byName(completed(snapshot()), "permission.wait");
+    expect(waits).toHaveLength(1);
+    expect(waits[0]!.tags).toEqual({ tool_id: "run_shell", decision: "allow" });
+  });
+
+  test("records deny decision when operator declines", async () => {
+    const gate = createPermissionGate({
+      approvals: [],
+      interactive: true,
+      skipPermissions: false,
+      requestApproval: async () => ({ allow: false, message: "nope" }),
+    });
+
+    const verdict = await gate.evaluate(shellCall("curl example.com"));
+    expect(verdict.allowed).toBe(false);
+
+    const waits = byName(completed(snapshot()), "permission.wait");
+    expect(waits).toHaveLength(1);
+    expect(waits[0]!.tags?.decision).toBe("deny");
+    expect(waits[0]!.tags?.tool_id).toBe("run_shell");
+  });
+
+  test("records path-arg tool ask as permission.wait", async () => {
+    const gate = createPermissionGate({
+      approvals: [],
+      interactive: true,
+      skipPermissions: false,
+      requestApproval: async () => ({ allow: true }),
+    });
+
+    const verdict = await gate.evaluate({
+      id: "c2",
+      name: "write_file",
+      arguments: { path: "src/a.ts", content: "x" },
+    });
+    expect(verdict.allowed).toBe(true);
+
+    const waits = byName(completed(snapshot()), "permission.wait");
+    expect(waits).toHaveLength(1);
+    expect(waits[0]!.tags).toEqual({ tool_id: "write_file", decision: "allow" });
+  });
+
+  test("does not open a span when a grant auto-approves", async () => {
+    let asked = 0;
+    const gate = createPermissionGate({
+      approvals: [{ tool: "run_shell", pattern: "npm *" }],
+      interactive: true,
+      skipPermissions: false,
+      requestApproval: async () => {
+        asked += 1;
+        return { allow: true };
+      },
+    });
+
+    const verdict = await gate.evaluate(shellCall("npm test"));
+    expect(verdict.allowed).toBe(true);
+    expect(asked).toBe(0);
+    expect(byName(snapshot(), "permission.wait")).toHaveLength(0);
+  });
+
+  test("nests under the open turn when a reactor turn is active", async () => {
+    const obs = createPerfReactorObserver();
+    obs.observe(event("inference.start", { model: "m" }));
+    obs.observe(
+      event("inference.done", {
+        turn: {
+          role: "assistant",
+          content: [{ type: "tool_call", id: "t1", name: "run_shell", arguments: {} }],
+          model: "m",
+          timestamp: 0,
+        },
+        usage: { input: 1, output: 1, cacheRead: 0, cacheWrite: 0, thinking: 0 },
+        source: { provider: "p", model: "m" },
+      }),
+    );
+    const turnId = obs.currentTurnId();
+    expect(turnId).not.toBeNull();
+
+    const gate = createPermissionGate({
+      approvals: [],
+      interactive: true,
+      skipPermissions: false,
+      requestApproval: async () => ({ allow: true }),
+    });
+    await gate.evaluate(shellCall("curl x"));
+
+    const wait = byName(completed(snapshot()), "permission.wait")[0]!;
+    expect(wait.parentId).toBe(turnId!);
+
+    obs.reset();
+  });
+});
+
+describe("subagent spans", () => {
+  test("records a completed subagent span around run()", async () => {
+    let runEntered = false;
+    const tool = createTaskTool({
+      permissionGate: skipGate,
+      cwd: "/repo",
+      getWorkdirBase: () => "/repo/.corbits",
+      provider,
+      run: async () => {
+        runEntered = true;
+        // Span must still be open while the child runs.
+        const open = snapshot().filter((s) => s.name === "subagent" && s.endNs === undefined);
+        expect(open).toHaveLength(1);
+        expect(open[0]!.tags?.subagent_id).toBe("call-sa-1");
+        return "## Summary\n\nok\n";
+      },
+    });
+    if (tool.kind !== "full") throw new Error("expected full tool");
+
+    const result = await tool.handler(
+      { id: "call-sa-1", name: "task", arguments: { description: "Job", prompt: "Do it" } },
+      new AbortController().signal,
+    );
+    expect(runEntered).toBe(true);
+    expect(typeof result.content === "string" ? result.content : "").toContain("ok");
+
+    const agents = byName(completed(snapshot()), "subagent");
+    expect(agents).toHaveLength(1);
+    expect(agents[0]!.tags?.subagent_id).toBe("call-sa-1");
+    expect(agents[0]!.endNs).toBeDefined();
+    expect(agents[0]!.endNs! >= agents[0]!.startNs).toBe(true);
+  });
+
+  test("nests under the open turn with turn_id tag for fanout rollup", async () => {
+    const obs = createPerfReactorObserver();
+    obs.observe(event("inference.start", { model: "m" }));
+    obs.observe(
+      event("inference.done", {
+        turn: {
+          role: "assistant",
+          content: [{ type: "tool_call", id: "task-1", name: "task", arguments: {} }],
+          model: "m",
+          timestamp: 0,
+        },
+        usage: { input: 1, output: 1, cacheRead: 0, cacheWrite: 0, thinking: 0 },
+        source: { provider: "p", model: "m" },
+      }),
+    );
+    const turnId = obs.currentTurnId();
+    expect(turnId).not.toBeNull();
+
+    const tool = createTaskTool({
+      permissionGate: skipGate,
+      cwd: "/repo",
+      getWorkdirBase: () => "/repo/.corbits",
+      provider,
+      run: async () => "## Summary\n\nchild done\n",
+    });
+    if (tool.kind !== "full") throw new Error("expected full tool");
+
+    await tool.handler(
+      { id: "call-child", name: "task", arguments: { description: "Child", prompt: "Work" } },
+      new AbortController().signal,
+    );
+
+    const agent = byName(completed(snapshot()), "subagent")[0]!;
+    expect(agent.parentId).toBe(turnId!);
+    expect(agent.tags?.subagent_id).toBe("call-child");
+    expect(agent.tags?.turn_id).toBe(turnId!);
+
+    // Wall time under the child is attributable via parentId (fanout rollup).
+    const turn = byName(snapshot(), "turn").find((s) => s.id === turnId);
+    expect(turn).toBeDefined();
+    expect(agent.startNs >= turn!.startNs).toBe(true);
+
+    obs.reset();
+  });
+
+  test("closes the span when run() rejects", async () => {
+    const tool = createTaskTool({
+      permissionGate: skipGate,
+      cwd: "/repo",
+      getWorkdirBase: () => "/repo/.corbits",
+      provider,
+      run: async () => {
+        throw new Error("boom");
+      },
+    });
+    if (tool.kind !== "full") throw new Error("expected full tool");
+
+    const result = await tool.handler(
+      { id: "call-fail", name: "task", arguments: { description: "Fail", prompt: "Work" } },
+      new AbortController().signal,
+    );
+    expect(typeof result.content === "string" ? result.content : "").toContain("Error:");
+
+    const agents = byName(completed(snapshot()), "subagent");
+    expect(agents).toHaveLength(1);
+    expect(agents[0]!.tags?.subagent_id).toBe("call-fail");
+    expect(agents[0]!.endNs).toBeDefined();
+  });
+});

src/perf/reactor-spans.ts

@@ -7,6 +7,16 @@
  *       inference.ttft   (start → first content-bearing delta)
  *       inference.stream (first delta → inference.done)
  *     tool (per invocation)
+ *     permission.wait   (operator ask; diagnostic nested category — wall time
+ *                        overlaps tool; exclusive attribution already excludes
+ *                        nested categories, so double-count is intentional)
+ *     subagent          (task fleet child wall)
+ *
+ * Single-primary assumption: the session run-sink owns one
+ * `createPerfReactorObserver`. Process-wide `currentTurnId()` is published only
+ * by that primary so permission.wait / subagent can nest outside the observer.
+ * Do not create concurrent observers that also call ensureTurn — they would
+ * overwrite the slot. Tests call `clear()` (and observer `reset()`) between cases.
  */
 
 import type { ReactorEmittedEvent } from "@intx/inference";
@@ -32,8 +42,26 @@ const FIRST_TOKEN_TYPES: ReadonlySet<string> = new Set([
 export type PerfReactorObserver = {
   observe(event: ReactorEmittedEvent): void;
   reset(): void;
+  /** Opaque PerfTrace id of the open turn span, or null when no turn is open. */
+  currentTurnId(): string | null;
 };
 
+/**
+ * Process-wide open-turn id from the most recently active reactor observer.
+ * Permission-wait and subagent spans nest under this when present.
+ * Tests that open turns should clear() PerfTrace and reset observers between cases.
+ */
+let activeTurnId: string | null = null;
+
+/** Current open turn span id (for nesting permission.wait / subagent outside the observer). */
+export function currentTurnId(): string | null {
+  return activeTurnId;
+}
+
+function setActiveTurnId(id: string | null): void {
+  activeTurnId = id;
+}
+
 type ObserverState = {
   turnId: string | null;
   inferenceId: string | null;
@@ -139,10 +167,14 @@ export function createPerfReactorObserver(): PerfReactorObserver {
   /**
    * Single exit for ending a turn: close orphan tool spans, then the turn.
    * Inference tree must already be closed (or will be via abandonTurn).
+   * Clears the process-wide active turn when this observer owns it.
    */
   function closeTurn(): void {
     closeOpenTools();
     endIfOpen(state.turnId);
+    if (state.turnId !== null && activeTurnId === state.turnId) {
+      setActiveTurnId(null);
+    }
     state.turnId = null;
     state.pendingTools = 0;
   }
@@ -156,6 +188,7 @@ export function createPerfReactorObserver(): PerfReactorObserver {
   function ensureTurn(): string {
     if (state.turnId === null) {
       state.turnId = start("turn");
+      setActiveTurnId(state.turnId);
     }
     return state.turnId;
   }
@@ -257,5 +290,9 @@ export function createPerfReactorObserver(): PerfReactorObserver {
     state = emptyState();
   }
 
-  return { observe, reset };
+  return {
+    observe,
+    reset,
+    currentTurnId: () => state.turnId,
+  };
 }