feat(engine): add cli-agent telemetry hub and session state machine (U3)

Pure engine code (no HTTP) for the CLI agent executor:

- state-machine.ts: authoritative per-session state machine implementing the
  HTD diagram (starting→ready→busy⇄waitingOnInput→done; done→busy follow-up;
  dead-classification choice → killed/userExited/authFailed/resuming; resume cap
  of 2 with backoff → needsAttention). Positive completion is distinct from idle
  (idle never produces done); inactivity stall backstop re-armed by output/
  telemetry events (no fixed turn timeout); termination classification helper for
  all five paths; per-turn latches reset between turns. Persists every transition
  via CliSessionStore (the transient `resuming` machine state maps onto the U1
  `dead` store enum) and exposes a throttled `onStateChange` subscription for the
  later SSE bridge — no dashboard imports.

- telemetry-hub.ts: in-process ingestion contract (ingest(sessionId, event)) for
  the U17 route and log-tailing adapters. Mints high-entropy per-session hook
  tokens (issueToken/validateToken/invalidate); rebuilds the registry only from
  live sessions in CliSessionStore so stale tokens for non-live sessions never
  validate; a token validates only for its own session. Bounds everything
  ingested: per-event size caps, per-turn count caps (lifecycle events exempt),
  ANSI/control stripping before pattern matching, and secret redaction that
  survives chunk boundaries via a held-back carry window (redactSecrets from
  @fusion/core).

Tests: 35 new (state-machine.test.ts, telemetry-hub.test.ts) covering AE1/AE2,
stall backstop, all termination paths, resume caps, token registry, two-turn
latch reset, oversized capping, ANSI stripping, and cross-chunk redaction.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

packages/engine/src/cli-agent/__tests__/state-machine.test.ts

@@ -0,0 +1,297 @@
+import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
+import { CliSessionStore } from "@fusion/core";
+import { Database } from "@fusion/core";
+import { mkdtempSync } from "node:fs";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+import { rm } from "node:fs/promises";
+import {
+  CliSessionStateMachine,
+  classifyTermination,
+  isResumeEligible,
+  looksLikeAuthFailure,
+  InvalidCliTransitionError,
+  type CliStateChange,
+} from "../state-machine.js";
+
+describe("CliSessionStateMachine", () => {
+  let tmpDir: string;
+  let fusionDir: string;
+  let db: Database;
+  let store: CliSessionStore;
+
+  beforeEach(() => {
+    tmpDir = mkdtempSync(join(tmpdir(), "kb-cli-sm-test-"));
+    fusionDir = join(tmpDir, ".fusion");
+    db = new Database(fusionDir, { inMemory: true });
+    db.init();
+    store = new CliSessionStore(fusionDir, db);
+    vi.useRealTimers();
+  });
+
+  afterEach(async () => {
+    db.close();
+    await rm(tmpDir, { recursive: true, force: true });
+  });
+
+  function seedSession(overrides: Record<string, unknown> = {}): string {
+    const s = store.createSession({
+      purpose: "execute",
+      projectId: "proj",
+      adapterId: "claude-code",
+      ...overrides,
+    });
+    return s.id;
+  }
+
+  function makeMachine(
+    sessionId: string,
+    opts: Partial<ConstructorParameters<typeof CliSessionStateMachine>[0]> = {},
+  ): CliSessionStateMachine {
+    return new CliSessionStateMachine({ sessionId, store, ...opts });
+  }
+
+  // ── AE1: native done advances; idle never does ───────────────────────────
+
+  it("AE1: positive done signal advances busy → done; persists", () => {
+    const id = seedSession();
+    const m = makeMachine(id);
+    m.markReady();
+    m.injectPrompt();
+    expect(m.getState()).toBe("busy");
+    m.signalDone();
+    expect(m.getState()).toBe("done");
+    expect(store.getSession(id)?.agentState).toBe("done");
+    expect(store.getSession(id)?.terminationReason).toBe("completed");
+  });
+
+  it("AE1: idle / output progress NEVER advances to done", () => {
+    const id = seedSession();
+    const m = makeMachine(id);
+    m.markReady();
+    m.injectPrompt();
+    m.signalOutputProgress();
+    m.signalOutputProgress();
+    expect(m.getState()).toBe("busy"); // never done
+  });
+
+  // ── AE2: permission prompt → waitingOnInput, no advance/fail ──────────────
+
+  it("AE2: waitingOnInput holds state (neither advances nor fails) and is reversible", () => {
+    const id = seedSession();
+    const m = makeMachine(id);
+    m.markReady();
+    m.injectPrompt();
+    m.signalWaitingOnInput();
+    expect(m.getState()).toBe("waitingOnInput");
+    expect(store.getSession(id)?.agentState).toBe("waitingOnInput");
+    m.signalBusy(); // user answered
+    expect(m.getState()).toBe("busy");
+  });
+
+  // ── Stall backstop ───────────────────────────────────────────────────────
+
+  it("stall backstop fires on a quiet busy turn past threshold → needsAttention", () => {
+    vi.useFakeTimers();
+    const id = seedSession();
+    const m = makeMachine(id, { stallThresholdMs: 1000 });
+    m.markReady();
+    m.injectPrompt();
+    vi.advanceTimersByTime(1000);
+    expect(m.getState()).toBe("needsAttention");
+  });
+
+  it("stall backstop NEVER fires on a streaming session (re-armed by output)", () => {
+    vi.useFakeTimers();
+    const id = seedSession();
+    const m = makeMachine(id, { stallThresholdMs: 1000 });
+    m.markReady();
+    m.injectPrompt();
+    for (let i = 0; i < 5; i++) {
+      vi.advanceTimersByTime(900);
+      m.signalOutputProgress(); // re-arm
+    }
+    vi.advanceTimersByTime(900);
+    expect(m.getState()).toBe("busy");
+  });
+
+  it("stall backstop suppressed while waitingOnInput", () => {
+    vi.useFakeTimers();
+    const id = seedSession();
+    const m = makeMachine(id, { stallThresholdMs: 1000 });
+    m.markReady();
+    m.injectPrompt();
+    m.signalWaitingOnInput();
+    vi.advanceTimersByTime(5000);
+    expect(m.getState()).toBe("waitingOnInput"); // no backstop while waiting
+  });
+
+  // ── Termination classification — all five paths ──────────────────────────
+
+  it("classifies clean exit-0 mid-task → userExited", () => {
+    expect(classifyTermination({ exitCode: 0, hadDone: false })).toBe("userExited");
+  });
+
+  it("classifies SIGKILL-from-cancel → killed (no resume)", () => {
+    const reason = classifyTermination({ cancelled: true, signal: "SIGKILL" });
+    expect(reason).toBe("killed");
+    expect(isResumeEligible(reason)).toBe(false);
+  });
+
+  it("classifies nonzero exit → crashed (resume-eligible)", () => {
+    const reason = classifyTermination({ exitCode: 1 });
+    expect(reason).toBe("crashed");
+    expect(isResumeEligible(reason)).toBe(true);
+  });
+
+  it("classifies credential-failure pattern → authFailed", () => {
+    expect(
+      classifyTermination({ exitCode: 1, recentOutput: "Error: Invalid API key" }),
+    ).toBe("authFailed");
+    expect(looksLikeAuthFailure("authentication failed")).toBe(true);
+    expect(looksLikeAuthFailure("all good")).toBe(false);
+  });
+
+  it("classifies found-dead-on-restart → engineDeath (resume-eligible)", () => {
+    const reason = classifyTermination({ foundDeadOnRestart: true });
+    expect(reason).toBe("engineDeath");
+    expect(isResumeEligible(reason)).toBe(true);
+  });
+
+  it("processEnded(crashed) routes busy → resuming", () => {
+    const id = seedSession();
+    const m = makeMachine(id);
+    m.markReady();
+    m.injectPrompt();
+    const reason = m.processEnded({ exitCode: 1 });
+    expect(reason).toBe("crashed");
+    expect(m.getState()).toBe("resuming");
+    expect(store.getSession(id)?.terminationReason).toBe("crashed");
+  });
+
+  it("processEnded(killed) lands on dead with killed reason", () => {
+    const id = seedSession();
+    const m = makeMachine(id);
+    m.markReady();
+    m.injectPrompt();
+    const reason = m.processEnded({ cancelled: true });
+    expect(reason).toBe("killed");
+    expect(m.getState()).toBe("dead");
+  });
+
+  // ── Resume caps ──────────────────────────────────────────────────────────
+
+  it("resume cap: two failures → needsAttention, third never attempted", () => {
+    const id = seedSession();
+    const m = makeMachine(id);
+    m.markReady();
+    m.injectPrompt();
+    m.processEnded({ exitCode: 1 }); // → resuming
+    m.recordResumeResult(false); // attempt 1 fails
+    expect(m.getState()).toBe("resuming");
+    expect(m.getResumeAttempts()).toBe(1);
+    m.recordResumeResult(false); // attempt 2 fails → cap
+    expect(m.getState()).toBe("needsAttention");
+    expect(m.getResumeAttempts()).toBe(2);
+    // No third attempt possible (not in resuming).
+    expect(() => m.recordResumeResult(false)).toThrow(InvalidCliTransitionError);
+  });
+
+  it("resume success returns to busy and resets attempts", () => {
+    const id = seedSession();
+    const m = makeMachine(id);
+    m.markReady();
+    m.injectPrompt();
+    m.processEnded({ exitCode: 1 });
+    m.recordResumeResult(false); // 1 fail
+    m.recordResumeResult(true); // succeed
+    expect(m.getState()).toBe("busy");
+    expect(m.getResumeAttempts()).toBe(0);
+  });
+
+  it("resume backoff metadata grows per attempt", () => {
+    const id = seedSession();
+    const changes: CliStateChange[] = [];
+    const m = makeMachine(id, { resumeBackoffBaseMs: 100, maxResumeAttempts: 5 });
+    m.onStateChange((c) => changes.push(c));
+    m.markReady();
+    m.injectPrompt();
+    m.processEnded({ exitCode: 1 });
+    m.recordResumeResult(false); // attempt 1 → backoff 100
+    m.recordResumeResult(false); // attempt 2 → backoff 200
+    const backoffs = changes.filter((c) => c.resumeBackoffMs != null).map((c) => c.resumeBackoffMs);
+    expect(backoffs).toEqual([100, 200]);
+  });
+
+  // ── Follow-up + per-turn latch reset ─────────────────────────────────────
+
+  it("done → busy follow-up resets per-turn done latch (two turns one handler)", () => {
+    vi.useFakeTimers();
+    const id = seedSession();
+    const m = makeMachine(id, { stallThresholdMs: 1000 });
+    m.markReady();
+    m.injectPrompt();
+    m.signalDone();
+    expect(m.getState()).toBe("done");
+    // Second turn through the same handler: follow-up re-arms a fresh turn.
+    m.followUp();
+    expect(m.getState()).toBe("busy");
+    // The new turn's stall watchdog is fresh (latch reset) — a quiet turn trips it.
+    vi.advanceTimersByTime(1000);
+    expect(m.getState()).toBe("needsAttention");
+  });
+
+  // ── needsAttention escalation ────────────────────────────────────────────
+
+  it("userExited dead landing can escalate to needsAttention preserving reason", () => {
+    const id = seedSession();
+    const m = makeMachine(id);
+    m.markReady();
+    m.injectPrompt();
+    m.processEnded({ exitCode: 0 }); // userExited → dead
+    expect(m.getState()).toBe("dead");
+    m.escalateToNeedsAttention();
+    expect(m.getState()).toBe("needsAttention");
+    expect(store.getSession(id)?.terminationReason).toBe("userExited");
+  });
+
+  // ── Throttled emission ───────────────────────────────────────────────────
+
+  it("throttled onStateChange coalesces rapid transitions", () => {
+    vi.useFakeTimers();
+    const id = seedSession();
+    let nowMs = 0;
+    const changes: CliStateChange[] = [];
+    const m = makeMachine(id, {
+      stateChangeThrottleMs: 100,
+      now: () => nowMs,
+    });
+    m.onStateChange((c) => changes.push(c));
+    m.markReady(); // emits immediately (first)
+    m.injectPrompt(); // within window → coalesced
+    m.signalWaitingOnInput(); // within window → coalesced
+    expect(changes.length).toBe(1);
+    nowMs = 100;
+    vi.advanceTimersByTime(100);
+    // The latest coalesced change is delivered at the window edge.
+    expect(changes.length).toBe(2);
+    expect(changes[1].state).toBe("waitingOnInput");
+  });
+
+  // ── Rebuild from persisted record ────────────────────────────────────────
+
+  it("rebuilds state from the persisted record on construction", () => {
+    const id = seedSession({ agentState: "busy" });
+    const m = makeMachine(id);
+    expect(m.getState()).toBe("busy");
+  });
+
+  // ── Invalid transitions guarded ──────────────────────────────────────────
+
+  it("rejects illegal transitions", () => {
+    const id = seedSession();
+    const m = makeMachine(id); // starting
+    expect(() => m.signalDone()).toThrow(InvalidCliTransitionError);
+    expect(() => m.followUp()).toThrow(InvalidCliTransitionError);
+  });
+});

packages/engine/src/cli-agent/__tests__/telemetry-hub.test.ts

@@ -0,0 +1,221 @@
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import { CliSessionStore, Database } from "@fusion/core";
+import { mkdtempSync } from "node:fs";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+import { rm } from "node:fs/promises";
+import { TelemetryHub, stripAnsiControl } from "../telemetry-hub.js";
+
+describe("TelemetryHub", () => {
+  let tmpDir: string;
+  let fusionDir: string;
+  let db: Database;
+  let store: CliSessionStore;
+
+  beforeEach(() => {
+    tmpDir = mkdtempSync(join(tmpdir(), "kb-cli-hub-test-"));
+    fusionDir = join(tmpDir, ".fusion");
+    db = new Database(fusionDir, { inMemory: true });
+    db.init();
+    store = new CliSessionStore(fusionDir, db);
+  });
+
+  afterEach(async () => {
+    db.close();
+    await rm(tmpDir, { recursive: true, force: true });
+  });
+
+  function seed(overrides: Record<string, unknown> = {}): string {
+    return store.createSession({
+      purpose: "execute",
+      projectId: "proj",
+      adapterId: "claude-code",
+      ...overrides,
+    }).id;
+  }
+
+  // ── Token registry ─────────────────────────────────────────────────────────
+
+  it("token validates only for its own session", () => {
+    const a = seed({ agentState: "busy" });
+    const b = seed({ agentState: "busy" });
+    const hub = new TelemetryHub({ store });
+    const tokenA = hub.issueToken(a);
+    const tokenB = hub.issueToken(b);
+    expect(hub.validateToken(a, tokenA)).toBe(true);
+    expect(hub.validateToken(b, tokenB)).toBe(true);
+    // Forged completion: session A presenting B's token → rejected.
+    expect(hub.validateToken(a, tokenB)).toBe(false);
+    expect(hub.validateToken(b, tokenA)).toBe(false);
+  });
+
+  it("tokens are high-entropy and unique per session", () => {
+    const a = seed({ agentState: "busy" });
+    const b = seed({ agentState: "busy" });
+    const hub = new TelemetryHub({ store });
+    const tokenA = hub.issueToken(a);
+    const tokenB = hub.issueToken(b);
+    expect(tokenA).toHaveLength(64); // 32 bytes → 64 hex
+    expect(tokenA).not.toEqual(tokenB);
+    expect(hub.validateToken(a, "deadbeef")).toBe(false);
+    expect(hub.validateToken(a, null)).toBe(false);
+  });
+
+  it("invalidate revokes the token after session end", () => {
+    const a = seed({ agentState: "busy" });
+    const hub = new TelemetryHub({ store });
+    const tokenA = hub.issueToken(a);
+    expect(hub.validateToken(a, tokenA)).toBe(true);
+    hub.invalidate(a);
+    expect(hub.validateToken(a, tokenA)).toBe(false);
+    expect(hub.hasSession(a)).toBe(false);
+  });
+
+  it("rebuilds only from live sessions; non-live sessions never validate after restart", () => {
+    const live = seed({ agentState: "busy" });
+    const dead = seed({ agentState: "done", terminationReason: "completed" });
+    // First hub mints a token for the dead-in-future session while it was live...
+    const hub1 = new TelemetryHub({ store });
+    const staleToken = hub1.issueToken(dead);
+    expect(hub1.validateToken(dead, staleToken)).toBe(true);
+
+    // Simulate restart: a fresh hub rebuilds from the store. `dead` is no longer
+    // live, so its on-disk-era token is not reconstituted.
+    const hub2 = new TelemetryHub({ store });
+    expect(hub2.hasSession(live)).toBe(true);
+    expect(hub2.hasSession(dead)).toBe(false);
+    expect(hub2.validateToken(dead, staleToken)).toBe(false);
+  });
+
+  // ── Ingestion → state routing ────────────────────────────────────────────
+
+  it("sessionStart drives starting → ready; native session id captured", () => {
+    const a = seed(); // starting
+    const hub = new TelemetryHub({ store });
+    hub.issueToken(a);
+    hub.ingest(a, { kind: "sessionStart", payload: { nativeSessionId: "claude-xyz" } });
+    expect(hub.getStateMachine(a)?.getState()).toBe("ready");
+    expect(store.getSession(a)?.nativeSessionId).toBe("claude-xyz");
+  });
+
+  it("native done advances to done; idle/output never does", () => {
+    const a = seed({ agentState: "busy" });
+    const hub = new TelemetryHub({ store });
+    hub.issueToken(a);
+    hub.ingest(a, { kind: "outputProgress", payload: { text: "thinking..." } });
+    hub.ingest(a, { kind: "toolActivity" });
+    expect(hub.getStateMachine(a)?.getState()).toBe("busy"); // never done from activity
+    hub.ingest(a, { kind: "done" });
+    expect(hub.getStateMachine(a)?.getState()).toBe("done");
+  });
+
+  it("AE2: waitingOnInput dispatches notification, state does not advance/fail", () => {
+    const a = seed({ agentState: "busy" });
+    const dispatched: unknown[] = [];
+    const hub = new TelemetryHub({
+      store,
+      onNotification: (info) => dispatched.push(info),
+    });
+    hub.issueToken(a);
+    hub.ingest(a, {
+      kind: "waitingOnInput",
+      payload: { notification: { type: "permission", tool: "Bash" } },
+    });
+    expect(hub.getStateMachine(a)?.getState()).toBe("waitingOnInput");
+    expect(dispatched).toHaveLength(1);
+    expect(dispatched[0]).toMatchObject({
+      sessionId: a,
+      notification: { type: "permission", tool: "Bash" },
+    });
+  });
+
+  it("ingest on unknown / non-live session is a no-op, not a crash", () => {
+    const hub = new TelemetryHub({ store });
+    expect(() => hub.ingest("nope", { kind: "done" })).not.toThrow();
+    expect(hub.ingest("nope", { kind: "done" })).toBeUndefined();
+  });
+
+  // ── Two turns through one handler: latch reset ───────────────────────────
+
+  it("per-turn event budget resets on a new busy turn", () => {
+    const a = seed({ agentState: "busy" });
+    const hub = new TelemetryHub({ store, maxEventsPerTurn: 2 });
+    hub.issueToken(a);
+    // Turn 1: budget = 2. Third event dropped.
+    expect(hub.ingest(a, { kind: "outputProgress", payload: { text: "a" } })).toBeDefined();
+    expect(hub.ingest(a, { kind: "outputProgress", payload: { text: "b" } })).toBeDefined();
+    expect(hub.ingest(a, { kind: "outputProgress", payload: { text: "c" } })).toBeUndefined();
+    // A `busy` event begins a fresh turn → budget resets (the busy event itself
+    // consumes one slot, then there is room again).
+    hub.ingest(a, { kind: "busy" });
+    expect(hub.ingest(a, { kind: "outputProgress", payload: { text: "d" } })).toBeDefined();
+  });
+
+  // ── Bounding: oversized event capped ─────────────────────────────────────
+
+  it("oversized event text is capped", () => {
+    const a = seed({ agentState: "busy" });
+    const hub = new TelemetryHub({ store, maxEventChars: 50, chunkCarryChars: 0 });
+    hub.issueToken(a);
+    // Plain prose (no secret-looking runs) so redaction doesn't collapse it
+    // before the size cap is exercised.
+    const big = "lorem ipsum ".repeat(2000);
+    const out = hub.ingest(a, { kind: "outputProgress", payload: { text: big } });
+    expect(out?.text?.length).toBe(50);
+    expect(out?.truncated).toBe(true);
+  });
+
+  // ── ANSI noise stripped before pattern matching ──────────────────────────
+
+  it("strips ANSI / control sequences before pattern matching", () => {
+    expect(stripAnsiControl("do\x1b[1mne\x1b[0m")).toBe("done");
+    expect(stripAnsiControl("clean")).toBe("clean");
+    const a = seed({ agentState: "busy" });
+    const hub = new TelemetryHub({ store, chunkCarryChars: 0 });
+    hub.issueToken(a);
+    const out = hub.ingest(a, {
+      kind: "transcript",
+      payload: { text: "\x1b[32mhello\x1b[0m \x1b[1mworld\x1b[0m" },
+    });
+    expect(out?.text).toBe("hello world");
+  });
+
+  // ── Secret redaction (incl. cross-chunk boundary) ────────────────────────
+
+  it("redacts secrets within a single chunk", () => {
+    const a = seed({ agentState: "busy" });
+    const hub = new TelemetryHub({ store, chunkCarryChars: 0 });
+    hub.issueToken(a);
+    const out = hub.ingest(a, {
+      kind: "transcript",
+      payload: { text: "export API_KEY=sk-abcdef0123456789abcdef0123" },
+    });
+    expect(out?.text).not.toContain("sk-abcdef0123456789abcdef0123");
+    expect(out?.text).toContain("[REDACTED]");
+  });
+
+  it("redacts a secret spanning a chunk boundary", () => {
+    const a = seed({ agentState: "busy" });
+    // Generous carry so the boundary prefix is held and joined with the next chunk.
+    const hub = new TelemetryHub({ store, chunkCarryChars: 64 });
+    hub.issueToken(a);
+    // Prefix "token=" arrives in chunk 1 (held in carry), value in chunk 2.
+    const out1 = hub.ingest(a, { kind: "transcript", payload: { text: "the token=" } });
+    const out2 = hub.ingest(a, {
+      kind: "transcript",
+      payload: { text: "sk-abcdef0123456789abcdef0123 done" },
+    });
+    const combined = (out1?.text ?? "") + (out2?.text ?? "") + (hub.flush(a) ?? "");
+    expect(combined).not.toContain("sk-abcdef0123456789abcdef0123");
+    expect(combined).toContain("[REDACTED]");
+  });
+
+  it("flush emits the held tail redacted on session end", () => {
+    const a = seed({ agentState: "busy" });
+    const hub = new TelemetryHub({ store, chunkCarryChars: 64 });
+    hub.issueToken(a);
+    hub.ingest(a, { kind: "transcript", payload: { text: "trailing secret=sk-zzzz0123456789abcd0123" } });
+    const flushed = hub.flush(a) ?? "";
+    expect(flushed).not.toContain("sk-zzzz0123456789abcd0123");
+  });
+});

packages/engine/src/cli-agent/state-machine.ts

@@ -0,0 +1,532 @@
+/**
+ * CliSessionStateMachine — authoritative per-session agent state machine
+ * (CLI Agent Executor, U3).
+ *
+ * Implements the HTD state diagram exactly:
+ *
+ *   [*] → starting
+ *   starting → ready            (readiness detected)
+ *   ready → busy                (prompt injected)
+ *   busy → waitingOnInput       (permission / question signal)
+ *   waitingOnInput → busy       (user answers)
+ *   busy → done                 (POSITIVE completion signal — idle NEVER does this)
+ *   done → busy                 (follow-up; resume first if the PTY was reaped)
+ *   busy → dead                 (PTY end / engine death)
+ *   waitingOnInput → dead       (PTY end / engine death)
+ *   dead → {killed|userExited|authFailed|resuming}   (classification choice)
+ *   resuming → busy             (native resume ok)
+ *   resuming → needsAttention   (2 attempts exhausted)
+ *   userExited → needsAttention (advance / retry / cancel prompt)
+ *   authFailed → needsAttention (re-authenticate message)
+ *
+ * Key behaviors (KTD — completion gating, termination taxonomy, stall backstop):
+ * - Positive completion is distinct from idleness. `signalDone()` advances to
+ *   `done`; output progress / idleness NEVER advance to done.
+ * - Stall backstop: no output progress past a configurable threshold WITHOUT a
+ *   done/waiting signal → needsAttention. The inactivity watchdog is re-armed by
+ *   any telemetry/output event (no fixed turn timeout); `waitingOnInput`
+ *   suppresses it (expected idleness).
+ * - Termination classification helper maps the manner of a PTY end onto the
+ *   taxonomy (killed / userExited / crashed / authFailed / engineDeath).
+ * - Resume attempt cap = 2 with backoff metadata; exhaustion → needsAttention.
+ * - Per-turn latches/budgets reset between turns (a new busy turn re-arms the
+ *   completion latch so a second turn through one handler is tracked cleanly).
+ *
+ * Persistence + observability:
+ * - Every transition persists through `CliSessionStore.updateSession` (state +
+ *   terminationReason + resumeAttempts written atomically by the store).
+ * - A throttled `onStateChange` callback is exposed for the SSE bridge to
+ *   subscribe to later. This module NEVER imports dashboard code.
+ */
+
+import type {
+  CliAgentState,
+  CliAutonomyPosture,
+  CliSessionStore,
+  CliTerminationReason,
+} from "@fusion/core";
+
+// ── Public types ───────────────────────────────────────────────────────────
+
+/**
+ * The machine's own state space. This is the U1 `CliAgentState` plus the
+ * transient HTD `"resuming"` sub-state, which is NOT a persisted store enum
+ * (U1's union has no `resuming`). When persisting, `resuming` maps onto the
+ * `dead` store state while the resume-eligible termination reason
+ * (crashed / engineDeath) carries the recovery intent. Surfaces that subscribe
+ * to `onStateChange` see the richer machine state so the SSE bridge can render
+ * "resuming…" without a schema change.
+ */
+export type CliMachineState = CliAgentState | "resuming";
+
+/** Map a machine state onto the persisted U1 store enum. */
+export function toPersistedState(state: CliMachineState): CliAgentState {
+  return state === "resuming" ? "dead" : state;
+}
+
+/** A throttled state-change notification handed to subscribers (e.g. the SSE bridge). */
+export interface CliStateChange {
+  sessionId: string;
+  /** The machine state moved into (may be the transient `resuming`). */
+  state: CliMachineState;
+  /** Termination reason when relevant (set on dead-classification transitions). */
+  terminationReason: CliTerminationReason | null;
+  /** Resume attempt count at the time of the change. */
+  resumeAttempts: number;
+  /** Backoff (ms) to wait before the next resume attempt, when resuming. */
+  resumeBackoffMs?: number;
+  /** ISO timestamp of the change. */
+  at: string;
+}
+
+export type CliStateChangeListener = (change: CliStateChange) => void;
+
+/**
+ * How a PTY ended, as observed by the manager / restart sweep. Fed into the
+ * classification helper to derive the termination taxonomy.
+ */
+export interface CliProcessEndInfo {
+  /** Whether the engine itself died and found the session dead on restart. */
+  foundDeadOnRestart?: boolean;
+  /** Whether the end was a deliberate hard cancel (SIGKILL-from-cancel). */
+  cancelled?: boolean;
+  /** Process exit code (0 = clean). Undefined when killed by signal. */
+  exitCode?: number | null;
+  /** Signal that terminated the process, if any (e.g. "SIGKILL"). */
+  signal?: string | number | null;
+  /**
+   * Recent (ANSI-stripped) output, scanned for a credential-failure pattern.
+   * Supplied by the caller (the hub strips ANSI before pattern matching).
+   */
+  recentOutput?: string;
+  /** Whether the session had observed a positive `done` before the end. */
+  hadDone?: boolean;
+}
+
+export interface CliStateMachineOptions {
+  sessionId: string;
+  store: CliSessionStore;
+  /** Autonomy posture (supplies maxResumeAttempts override). */
+  posture?: CliAutonomyPosture | null;
+  /**
+   * Inactivity / stall threshold (ms). If no output progress and no done/waiting
+   * signal arrives within this window of a busy turn, the backstop fires
+   * (needsAttention). Default 5 minutes.
+   */
+  stallThresholdMs?: number;
+  /** Max resume attempts before giving up. Default 2 (KTD). */
+  maxResumeAttempts?: number;
+  /** Base backoff (ms) for resume attempts; doubled per attempt. Default 1000. */
+  resumeBackoffBaseMs?: number;
+  /** Throttle window (ms) for `onStateChange`. Default 0 (emit every change). */
+  stateChangeThrottleMs?: number;
+  /** Clock injection for tests. */
+  now?: () => number;
+  /**
+   * Timer scheduler injection for tests (fake timers). Returns a cancel handle.
+   * Defaults to setTimeout/clearTimeout.
+   */
+  setTimer?: (fn: () => void, ms: number) => unknown;
+  clearTimer?: (handle: unknown) => void;
+}
+
+/**
+ * Default credential-failure detector. Scans (already ANSI-stripped) recent
+ * output for common auth-rejection phrasing.
+ */
+const AUTH_FAILURE_PATTERN =
+  /\b(authentication failed|invalid api key|unauthorized|401 unauthorized|not authenticated|please (?:re-?)?(?:login|log in|authenticate)|credential[s]? (?:rejected|invalid|expired)|your session has expired|token (?:expired|invalid|revoked))\b/i;
+
+/** Detect a credential-failure pattern in recent (ANSI-stripped) output. */
+export function looksLikeAuthFailure(recentOutput: string | undefined): boolean {
+  if (!recentOutput) return false;
+  return AUTH_FAILURE_PATTERN.test(recentOutput);
+}
+
+/**
+ * Classify a PTY end onto the termination taxonomy (KTD). Pure — no side effects.
+ *
+ * - found-dead-on-restart                → engineDeath
+ * - SIGKILL-from-cancel / hard cancel    → killed
+ * - credential-failure in recent output  → authFailed
+ * - clean exit-0 mid-task (no done)      → userExited
+ * - nonzero exit / killed by signal      → crashed
+ * - any exit AFTER a positive done       → completed
+ */
+export function classifyTermination(info: CliProcessEndInfo): CliTerminationReason {
+  if (info.foundDeadOnRestart) return "engineDeath";
+  if (info.cancelled) return "killed";
+  if (looksLikeAuthFailure(info.recentOutput)) return "authFailed";
+  if (info.hadDone) return "completed";
+  // Killed by a signal (no clean exit) → crashed.
+  if (info.signal != null && info.signal !== 0) return "crashed";
+  if (info.exitCode === 0) return "userExited";
+  // Any nonzero / unknown exit code → crashed.
+  return "crashed";
+}
+
+/** Resume-eligible termination reasons (KTD): only crash / engine death auto-resume. */
+export function isResumeEligible(reason: CliTerminationReason): boolean {
+  return reason === "crashed" || reason === "engineDeath";
+}
+
+/** Error thrown when a transition is attempted from an incompatible state. */
+export class InvalidCliTransitionError extends Error {
+  readonly code = "INVALID_CLI_TRANSITION";
+  constructor(
+    public readonly from: CliMachineState,
+    public readonly intent: string,
+  ) {
+    super(`Invalid CLI session transition: cannot ${intent} from state "${from}"`);
+    this.name = "InvalidCliTransitionError";
+  }
+}
+
+// ── State machine ──────────────────────────────────────────────────────────
+
+export class CliSessionStateMachine {
+  readonly sessionId: string;
+  private readonly store: CliSessionStore;
+  private readonly stallThresholdMs: number;
+  private readonly maxResumeAttempts: number;
+  private readonly resumeBackoffBaseMs: number;
+  private readonly throttleMs: number;
+  private readonly now: () => number;
+  private readonly setTimer: (fn: () => void, ms: number) => unknown;
+  private readonly clearTimer: (handle: unknown) => void;
+
+  private state: CliMachineState;
+  private terminationReason: CliTerminationReason | null = null;
+  private resumeAttempts = 0;
+
+  /** Per-turn latch: has a positive done fired in the current busy turn. */
+  private doneLatched = false;
+  /** Per-turn latch: has waiting-on-input fired in the current busy turn. */
+  private waitingLatched = false;
+
+  private stallTimer: unknown = null;
+  private listeners = new Set<CliStateChangeListener>();
+
+  // Throttle bookkeeping.
+  private lastEmitAt = 0;
+  private pendingEmit: CliStateChange | null = null;
+  private throttleTimer: unknown = null;
+
+  constructor(opts: CliStateMachineOptions) {
+    this.sessionId = opts.sessionId;
+    this.store = opts.store;
+    this.stallThresholdMs = opts.stallThresholdMs ?? 5 * 60_000;
+    this.maxResumeAttempts =
+      opts.maxResumeAttempts ??
+      (typeof opts.posture?.maxResumeAttempts === "number"
+        ? opts.posture.maxResumeAttempts
+        : 2);
+    this.resumeBackoffBaseMs = opts.resumeBackoffBaseMs ?? 1000;
+    this.throttleMs = opts.stateChangeThrottleMs ?? 0;
+    this.now = opts.now ?? (() => Date.now());
+    this.setTimer =
+      opts.setTimer ?? ((fn, ms) => setTimeout(fn, ms) as unknown);
+    this.clearTimer =
+      opts.clearTimer ?? ((h) => clearTimeout(h as ReturnType<typeof setTimeout>));
+
+    // Seed from the persisted record so a rebuilt machine reflects reality.
+    const existing = this.store.getSession(this.sessionId);
+    this.state = existing?.agentState ?? "starting";
+    this.terminationReason = existing?.terminationReason ?? null;
+    this.resumeAttempts = existing?.resumeAttempts ?? 0;
+  }
+
+  // ── Observation ──────────────────────────────────────────────────────────
+
+  getState(): CliMachineState {
+    return this.state;
+  }
+
+  getTerminationReason(): CliTerminationReason | null {
+    return this.terminationReason;
+  }
+
+  getResumeAttempts(): number {
+    return this.resumeAttempts;
+  }
+
+  /** Subscribe to throttled state changes. Returns an unsubscribe handle. */
+  onStateChange(listener: CliStateChangeListener): () => void {
+    this.listeners.add(listener);
+    return () => this.listeners.delete(listener);
+  }
+
+  // ── Transitions (HTD diagram) ────────────────────────────────────────────
+
+  /** starting → ready (readiness detected). */
+  markReady(): void {
+    if (this.state !== "starting") {
+      throw new InvalidCliTransitionError(this.state, "markReady");
+    }
+    this.transition("ready");
+  }
+
+  /**
+   * ready → busy (prompt injected) and done → busy (follow-up).
+   * Begins a new turn: per-turn latches reset, stall watchdog armed.
+   */
+  injectPrompt(): void {
+    if (this.state !== "ready" && this.state !== "done" && this.state !== "resuming") {
+      throw new InvalidCliTransitionError(this.state, "injectPrompt");
+    }
+    this.beginTurn();
+    this.transition("busy");
+  }
+
+  /** done → busy (follow-up). Alias for injectPrompt from the done state. */
+  followUp(): void {
+    if (this.state !== "done") {
+      throw new InvalidCliTransitionError(this.state, "followUp");
+    }
+    this.beginTurn();
+    this.transition("busy");
+  }
+
+  /**
+   * Output progress / activity. Re-arms the inactivity watchdog. NEVER advances
+   * state — idleness and activity are both gated away from `done`.
+   */
+  signalOutputProgress(): void {
+    if (this.state === "busy") {
+      this.armStallWatchdog();
+    }
+  }
+
+  /**
+   * busy → waitingOnInput (permission / question signal). Suppresses the stall
+   * watchdog (expected idleness). Does NOT advance the pipeline or fail.
+   */
+  signalWaitingOnInput(): void {
+    if (this.state === "waitingOnInput") return; // idempotent
+    if (this.state !== "busy") {
+      throw new InvalidCliTransitionError(this.state, "signalWaitingOnInput");
+    }
+    this.waitingLatched = true;
+    this.clearStallWatchdog();
+    this.transition("waitingOnInput");
+  }
+
+  /** waitingOnInput → busy (user answered). Re-arms the watchdog. */
+  signalBusy(): void {
+    if (this.state === "busy") {
+      this.armStallWatchdog();
+      return;
+    }
+    if (this.state !== "waitingOnInput") {
+      throw new InvalidCliTransitionError(this.state, "signalBusy");
+    }
+    this.armStallWatchdog();
+    this.transition("busy");
+  }
+
+  /**
+   * busy → done (POSITIVE completion signal). This is the ONLY path to `done`.
+   * Idle / output progress never reach here.
+   */
+  signalDone(): void {
+    if (this.state === "done") return; // idempotent
+    if (this.state !== "busy" && this.state !== "waitingOnInput") {
+      throw new InvalidCliTransitionError(this.state, "signalDone");
+    }
+    this.doneLatched = true;
+    this.clearStallWatchdog();
+    this.transition("done", "completed");
+  }
+
+  /**
+   * busy/waitingOnInput → dead, then classify. Provide the observed end info;
+   * the taxonomy is derived by `classifyTermination`. After classification:
+   * - killed / userExited / authFailed → terminal (userExited/authFailed will be
+   *   surfaced as needsAttention by the caller's escalation, but the recorded
+   *   reason stays precise — `escalateToNeedsAttention` moves the state).
+   * - crashed / engineDeath → `resuming` (caller drives resume attempts).
+   * - completed → done.
+   *
+   * @returns the classified termination reason.
+   */
+  processEnded(info: CliProcessEndInfo): CliTerminationReason {
+    // dead is reachable from any active state.
+    this.clearStallWatchdog();
+    const reason = classifyTermination({ ...info, hadDone: info.hadDone ?? this.doneLatched });
+    this.terminationReason = reason;
+
+    if (reason === "completed") {
+      this.transition("done", "completed");
+      return reason;
+    }
+    if (isResumeEligible(reason)) {
+      this.transition("resuming", reason);
+      return reason;
+    }
+    // killed / userExited / authFailed are recorded on a `dead` landing; the
+    // diagram's killed → [*] is terminal, while userExited / authFailed escalate
+    // to needsAttention via escalateToNeedsAttention().
+    this.transition("dead", reason);
+    return reason;
+  }
+
+  /**
+   * Record a resume attempt result.
+   * - success → busy (a fresh turn).
+   * - failure → another `resuming` with backoff, until the cap (2) is hit, then
+   *   needsAttention. The third attempt is never made.
+   */
+  recordResumeResult(success: boolean): void {
+    if (this.state !== "resuming") {
+      throw new InvalidCliTransitionError(this.state, "recordResumeResult");
+    }
+    if (success) {
+      this.resumeAttempts = 0;
+      this.beginTurn();
+      this.transition("busy");
+      return;
+    }
+    this.resumeAttempts += 1;
+    if (this.resumeAttempts >= this.maxResumeAttempts) {
+      this.transition("needsAttention");
+      return;
+    }
+    // Stay in resuming with backoff metadata so the coordinator schedules a retry.
+    const backoff = this.resumeBackoffBaseMs * 2 ** (this.resumeAttempts - 1);
+    this.persistAndEmit("resuming", this.terminationReason, backoff);
+  }
+
+  /** Backoff (ms) the coordinator should wait before the next resume attempt. */
+  nextResumeBackoffMs(): number {
+    return this.resumeBackoffBaseMs * 2 ** this.resumeAttempts;
+  }
+
+  /**
+   * Escalate the current dead/auth/userExit landing to needsAttention (the
+   * userExited → needsAttention and authFailed → needsAttention edges). The
+   * recorded terminationReason is preserved.
+   */
+  escalateToNeedsAttention(): void {
+    if (
+      this.state !== "dead" &&
+      this.state !== "resuming" &&
+      this.state !== "busy" &&
+      this.state !== "waitingOnInput"
+    ) {
+      throw new InvalidCliTransitionError(this.state, "escalateToNeedsAttention");
+    }
+    this.clearStallWatchdog();
+    this.transition("needsAttention");
+  }
+
+  /** Force-dispose: cancel timers and drop listeners. */
+  dispose(): void {
+    this.clearStallWatchdog();
+    if (this.throttleTimer != null) {
+      this.clearTimer(this.throttleTimer);
+      this.throttleTimer = null;
+    }
+    this.listeners.clear();
+  }
+
+  // ── Per-turn latches / stall watchdog ────────────────────────────────────
+
+  private beginTurn(): void {
+    // Reset per-turn latches and budgets between turns (KTD).
+    this.doneLatched = false;
+    this.waitingLatched = false;
+    this.armStallWatchdog();
+  }
+
+  private armStallWatchdog(): void {
+    this.clearStallWatchdog();
+    this.stallTimer = this.setTimer(() => {
+      this.onStall();
+    }, this.stallThresholdMs);
+  }
+
+  private clearStallWatchdog(): void {
+    if (this.stallTimer != null) {
+      this.clearTimer(this.stallTimer);
+      this.stallTimer = null;
+    }
+  }
+
+  /**
+   * Stall backstop: quiet busy turn past the threshold with no done/waiting
+   * signal → needsAttention. Never fires from waitingOnInput (cleared) and never
+   * from a streaming session (re-armed by output progress).
+   */
+  private onStall(): void {
+    this.stallTimer = null;
+    if (this.state !== "busy") return;
+    if (this.doneLatched || this.waitingLatched) return;
+    this.transition("needsAttention");
+  }
+
+  // ── Persistence + throttled emit ─────────────────────────────────────────
+
+  private transition(next: CliMachineState, reason?: CliTerminationReason): void {
+    this.state = next;
+    if (reason !== undefined) this.terminationReason = reason;
+    if (next === "busy" || next === "ready") {
+      // Live again: clear any stale termination reason.
+      this.terminationReason = null;
+    }
+    this.persistAndEmit(next, this.terminationReason);
+  }
+
+  private persistAndEmit(
+    next: CliMachineState,
+    reason: CliTerminationReason | null,
+    resumeBackoffMs?: number,
+  ): void {
+    // Persist the U1 store enum (resuming → dead); the machine state and the
+    // resume-eligible reason carry the recovery intent for surfaces.
+    this.store.updateSession(this.sessionId, {
+      agentState: toPersistedState(next),
+      terminationReason: reason,
+      resumeAttempts: this.resumeAttempts,
+    });
+    const change: CliStateChange = {
+      sessionId: this.sessionId,
+      state: next,
+      terminationReason: reason,
+      resumeAttempts: this.resumeAttempts,
+      ...(resumeBackoffMs !== undefined ? { resumeBackoffMs } : {}),
+      at: new Date(this.now()).toISOString(),
+    };
+    this.emitThrottled(change);
+  }
+
+  private emitThrottled(change: CliStateChange): void {
+    if (this.throttleMs <= 0) {
+      this.deliver(change);
+      return;
+    }
+    // Leading-edge: when no throttle window is open, deliver immediately and
+    // open a window. Subsequent changes within the window coalesce into a single
+    // trailing emit of the latest change at the window edge.
+    if (this.throttleTimer == null) {
+      this.deliver(change);
+      this.throttleTimer = this.setTimer(() => {
+        this.throttleTimer = null;
+        if (this.pendingEmit) {
+          const pending = this.pendingEmit;
+          this.pendingEmit = null;
+          this.deliver(pending);
+        }
+      }, this.throttleMs);
+      return;
+    }
+    // Within an open window → coalesce (keep only the latest).
+    this.pendingEmit = change;
+  }
+
+  private deliver(change: CliStateChange): void {
+    this.lastEmitAt = this.now();
+    for (const listener of this.listeners) {
+      listener(change);
+    }
+  }
+}

packages/engine/src/cli-agent/telemetry-hub.ts

@@ -0,0 +1,387 @@
+/**
+ * TelemetryHub — in-process telemetry ingestion + per-session token registry
+ * (CLI Agent Executor, U3).
+ *
+ * The hub is the single in-process sink for normalized telemetry events about a
+ * CLI agent session. It is consumed later by:
+ * - the dashboard hook route (U17), which forwards validated hook POSTs, and
+ * - log-tailing adapters (Codex rollout, Pi JSONL) that synthesize events.
+ *
+ * The engine has NO HTTP server — this module is pure engine code. It performs
+ * NO networking; it only validates tokens and ingests already-delivered events.
+ *
+ * Responsibilities (KTD — telemetry tiering, completion gating, security):
+ * - Token registry: mint a high-entropy per-session hook token at spawn
+ *   (`issueToken`), validate it scoped to its own session (`validateToken`), and
+ *   invalidate it on session end (`invalidate`). On construction the registry is
+ *   rebuilt ONLY from sessions still live in `CliSessionStore`, so stale on-disk
+ *   tokens for non-live sessions never validate after an engine restart.
+ *   A forged completion using another session's token is rejected because a
+ *   token validates only for the session it was issued to.
+ * - Normalization + bounding: per-event payload size caps, per-turn event count
+ *   caps, ANSI / control stripping BEFORE any pattern matching, and secret
+ *   redaction that survives chunk boundaries (a token split across two chunks is
+ *   still caught — uses `redactSecrets` from @fusion/core on the joined tail).
+ * - Routing: maps a normalized event onto the session's state machine
+ *   (sessionStart→ready, busy→signalBusy, waitingOnInput→signalWaitingOnInput +
+ *   notification dispatch, done→signalDone, outputProgress→signalOutputProgress).
+ *   Idle/output NEVER advances to done — that gating lives in the state machine.
+ */
+
+import { randomBytes } from "node:crypto";
+import { redactSecrets, type CliSessionStore } from "@fusion/core";
+import { CliSessionStateMachine } from "./state-machine.js";
+
+// ── Constants (bounding rules) ───────────────────────────────────────────────
+
+/** Max retained text per ingested event after stripping (bytes/chars). */
+export const DEFAULT_MAX_EVENT_CHARS = 64 * 1024;
+/** Max events accepted per turn before further events are dropped (count cap). */
+export const DEFAULT_MAX_EVENTS_PER_TURN = 5000;
+/**
+ * Carry-over window kept across chunks so a secret straddling a chunk boundary is
+ * still redacted (prefix in chunk N, value in chunk N+1).
+ */
+export const DEFAULT_CHUNK_CARRY_CHARS = 256;
+
+/** A live session that the hub considers "live" when rebuilding tokens. */
+const LIVE_STATES = new Set(["starting", "ready", "busy", "waitingOnInput", "resuming"]);
+
+// ── Event contract ───────────────────────────────────────────────────────────
+
+/** Normalized telemetry event kinds the hub understands. */
+export type TelemetryEventKind =
+  | "sessionStart"
+  | "busy"
+  | "waitingOnInput"
+  | "done"
+  | "toolActivity"
+  | "outputProgress"
+  | "transcript";
+
+/** A normalized telemetry event. `payload` is event-specific, free-form, bounded. */
+export interface TelemetryEvent {
+  kind: TelemetryEventKind;
+  payload?: Record<string, unknown> & {
+    /** Raw text chunk (output / transcript) — stripped + redacted on ingest. */
+    text?: string;
+    /** Native session id reported by the CLI (e.g. Claude `session_id`). */
+    nativeSessionId?: string;
+    /** Notification context for a waitingOnInput event (permission/question). */
+    notification?: Record<string, unknown>;
+  };
+}
+
+/** The sanitized form of an event after ingest bounding/stripping/redaction. */
+export interface SanitizedTelemetryEvent {
+  kind: TelemetryEventKind;
+  /** Sanitized text (ANSI/control stripped, secret-redacted, size-capped). */
+  text?: string;
+  nativeSessionId?: string;
+  notification?: Record<string, unknown>;
+  /** True when the event text was truncated by the size cap. */
+  truncated?: boolean;
+}
+
+/** Dispatch invoked when a waitingOnInput event is ingested (banner/notify). */
+export type NotificationDispatch = (info: {
+  sessionId: string;
+  notification: Record<string, unknown> | undefined;
+}) => void;
+
+export interface TelemetryHubOptions {
+  store: CliSessionStore;
+  /** Notification dispatch for waiting-on-input events (per node config). */
+  onNotification?: NotificationDispatch;
+  /** Per-event text cap. */
+  maxEventChars?: number;
+  /** Per-turn event count cap. */
+  maxEventsPerTurn?: number;
+  /** Cross-chunk carry-over window for boundary-spanning secret redaction. */
+  chunkCarryChars?: number;
+  /** Token byte length (high entropy). Default 32 bytes → 64 hex chars. */
+  tokenBytes?: number;
+  /** Factory for a session's state machine (test injection). */
+  createStateMachine?: (sessionId: string) => CliSessionStateMachine;
+}
+
+// ── ANSI / control stripping ─────────────────────────────────────────────────
+
+const ESC = "\\u001b";
+// OSC: ESC ] ... terminated by BEL () or ST (ESC \). Strip first — it
+// carries ';' the CSI pattern would otherwise eat into.
+const OSC_PATTERN = new RegExp(`${ESC}\\][\\s\\S]*?(?:\\u0007|${ESC}\\\\)`, "g");
+// CSI (ESC [ params intermediates final) + other 2-char ESC sequences.
+const ANSI_PATTERN = new RegExp(`${ESC}\\[[0-?]*[ -/]*[@-~]|${ESC}[@-Z\\\\-_]`, "g");
+// Remaining lone C0 controls (except \t \n \r) and DEL.
+// eslint-disable-next-line no-control-regex
+const C0_PATTERN = /[\u0000-\u0008\u000b\u000c\u000e-\u001f\u007f]/g;
+
+/** Strip ANSI escape sequences and stray control chars from text. */
+export function stripAnsiControl(text: string): string {
+  return text.replace(OSC_PATTERN, "").replace(ANSI_PATTERN, "").replace(C0_PATTERN, "");
+}
+
+// ── Per-session telemetry state ──────────────────────────────────────────────
+
+interface SessionTelemetry {
+  token: string;
+  machine: CliSessionStateMachine;
+  /** Event count in the current turn (reset on a new busy turn). */
+  turnEventCount: number;
+  /** Tail of the previous chunk's text, kept for boundary-spanning redaction. */
+  carry: string;
+}
+
+// ── Hub ──────────────────────────────────────────────────────────────────────
+
+export class TelemetryHub {
+  private readonly store: CliSessionStore;
+  private readonly onNotification?: NotificationDispatch;
+  private readonly maxEventChars: number;
+  private readonly maxEventsPerTurn: number;
+  private readonly chunkCarryChars: number;
+  private readonly tokenBytes: number;
+  private readonly createStateMachine: (sessionId: string) => CliSessionStateMachine;
+
+  /** token → sessionId reverse index (validates token-belongs-to-session). */
+  private readonly tokenToSession = new Map<string, string>();
+  private readonly sessions = new Map<string, SessionTelemetry>();
+
+  constructor(opts: TelemetryHubOptions) {
+    this.store = opts.store;
+    this.onNotification = opts.onNotification;
+    this.maxEventChars = opts.maxEventChars ?? DEFAULT_MAX_EVENT_CHARS;
+    this.maxEventsPerTurn = opts.maxEventsPerTurn ?? DEFAULT_MAX_EVENTS_PER_TURN;
+    this.chunkCarryChars = opts.chunkCarryChars ?? DEFAULT_CHUNK_CARRY_CHARS;
+    this.tokenBytes = opts.tokenBytes ?? 32;
+    this.createStateMachine =
+      opts.createStateMachine ??
+      ((sessionId) => new CliSessionStateMachine({ sessionId, store: this.store }));
+
+    this.rebuildFromLiveSessions();
+  }
+
+  /**
+   * Rebuild the per-session registry from sessions still live in the store. Stale
+   * tokens for non-live sessions are NOT recreated — only sessions in a live
+   * state get a fresh token, so a forged POST referencing a dead session's id has
+   * no valid token to present. Tokens are NOT persisted, so a restart always
+   * mints fresh ones; an attacker holding an old on-disk token cannot validate.
+   */
+  private rebuildFromLiveSessions(): void {
+    const live = this.store
+      .listSessions()
+      .filter((s) => LIVE_STATES.has(s.agentState));
+    for (const session of live) {
+      this.register(session.id);
+    }
+  }
+
+  /** Whether a session id is currently registered (live) with the hub. */
+  hasSession(sessionId: string): boolean {
+    return this.sessions.has(sessionId);
+  }
+
+  /** Get the state machine for a registered session (for the executor seam). */
+  getStateMachine(sessionId: string): CliSessionStateMachine | undefined {
+    return this.sessions.get(sessionId)?.machine;
+  }
+
+  // ── Token registry ─────────────────────────────────────────────────────────
+
+  /**
+   * Register a session and mint its high-entropy hook token. Idempotent: a second
+   * call returns the existing token (so rebuild + spawn races don't double-mint).
+   */
+  private register(sessionId: string): string {
+    const existing = this.sessions.get(sessionId);
+    if (existing) return existing.token;
+    const token = randomBytes(this.tokenBytes).toString("hex");
+    const machine = this.createStateMachine(sessionId);
+    this.sessions.set(sessionId, { token, machine, turnEventCount: 0, carry: "" });
+    this.tokenToSession.set(token, sessionId);
+    return token;
+  }
+
+  /** Mint (or return) the per-session hook token at spawn. */
+  issueToken(sessionId: string): string {
+    return this.register(sessionId);
+  }
+
+  /**
+   * Validate a token against a specific session. Returns true ONLY when the token
+   * was issued for exactly this session — a valid token for session B never
+   * validates for session A (forged-completion rejection).
+   */
+  validateToken(sessionId: string, token: string | null | undefined): boolean {
+    if (!token) return false;
+    const owner = this.tokenToSession.get(token);
+    if (!owner) return false;
+    return owner === sessionId && this.sessions.has(sessionId);
+  }
+
+  /** Invalidate a session's token (called on session end). */
+  invalidate(sessionId: string): void {
+    const entry = this.sessions.get(sessionId);
+    if (!entry) return;
+    this.tokenToSession.delete(entry.token);
+    entry.machine.dispose();
+    this.sessions.delete(sessionId);
+  }
+
+  /**
+   * Flush any held-back carry tail as a final redacted chunk. Call on session end
+   * (before `invalidate`) so the last bytes — which were held to catch a
+   * boundary-spanning secret — are emitted, still redacted. Returns the flushed
+   * sanitized text, or undefined when there is nothing held / no such session.
+   */
+  flush(sessionId: string): string | undefined {
+    const entry = this.sessions.get(sessionId);
+    if (!entry || entry.carry.length === 0) return undefined;
+    const text = redactSecrets(entry.carry);
+    entry.carry = "";
+    return text;
+  }
+
+  // ── Ingestion ────────────────────────────────────────────────────────────
+
+  /**
+   * Ingest a normalized telemetry event for a session. Token validation is the
+   * caller's responsibility (the route validates before forwarding); ingest is
+   * the in-process bounding + routing seam. An unknown/non-live session is a
+   * no-op (never a crash). Returns the sanitized event for observability, or
+   * undefined when dropped (unknown session or per-turn cap reached).
+   */
+  ingest(sessionId: string, event: TelemetryEvent): SanitizedTelemetryEvent | undefined {
+    const entry = this.sessions.get(sessionId);
+    if (!entry) return undefined; // unknown / non-live session → no-op
+
+    // Lifecycle events (turn boundaries / completion) are never dropped — they
+    // drive the authoritative state machine. The per-turn cap bounds high-volume
+    // activity/text events within a turn (a flood backstop), and resets when a
+    // new turn begins (the `busy` / `sessionStart` route handlers zero it).
+    const isLifecycle =
+      event.kind === "sessionStart" ||
+      event.kind === "busy" ||
+      event.kind === "waitingOnInput" ||
+      event.kind === "done";
+    if (!isLifecycle) {
+      if (entry.turnEventCount >= this.maxEventsPerTurn) {
+        return undefined;
+      }
+      entry.turnEventCount += 1;
+    }
+
+    const sanitized = this.sanitize(entry, event);
+    this.route(entry, sanitized);
+    return sanitized;
+  }
+
+  // ── Sanitization ───────────────────────────────────────────────────────────
+
+  private sanitize(entry: SessionTelemetry, event: TelemetryEvent): SanitizedTelemetryEvent {
+    const out: SanitizedTelemetryEvent = { kind: event.kind };
+    const payload = event.payload ?? {};
+
+    if (typeof payload.nativeSessionId === "string") {
+      out.nativeSessionId = payload.nativeSessionId.slice(0, 256);
+    }
+    if (payload.notification && typeof payload.notification === "object") {
+      out.notification = payload.notification as Record<string, unknown>;
+    }
+
+    if (typeof payload.text === "string") {
+      // 1. Strip ANSI / control BEFORE pattern matching or redaction.
+      const stripped = stripAnsiControl(payload.text);
+      // 2. Redact across chunk boundaries. We hold back a tail window of raw
+      //    (un-redacted) text from each chunk; the held tail is prepended to the
+      //    NEXT chunk before redaction, so a secret whose prefix is in chunk N and
+      //    value is in chunk N+1 is redacted as one string. We emit, for chunk N,
+      //    everything in `carry + chunk` EXCEPT the new held tail.
+      const joined = entry.carry + stripped;
+      const carryLen = entry.carry.length;
+      const newTail = joined.slice(Math.max(carryLen, joined.length - this.chunkCarryChars));
+      const toEmit = joined.slice(0, joined.length - newTail.length);
+      entry.carry = newTail;
+
+      let visible = redactSecrets(toEmit);
+      // 3. Size cap.
+      let truncated = false;
+      if (visible.length > this.maxEventChars) {
+        visible = visible.slice(0, this.maxEventChars);
+        truncated = true;
+      }
+      out.text = visible;
+      if (truncated) out.truncated = true;
+    }
+
+    return out;
+  }
+
+  // ── Routing onto the state machine ──────────────────────────────────────────
+
+  private route(entry: SessionTelemetry, event: SanitizedTelemetryEvent): void {
+    const machine = entry.machine;
+    // Capture native session id whenever reported.
+    if (event.nativeSessionId) {
+      const current = this.store.getSession(entry.machine.sessionId);
+      if (current && current.nativeSessionId !== event.nativeSessionId) {
+        this.store.updateSession(entry.machine.sessionId, {
+          nativeSessionId: event.nativeSessionId,
+        });
+      }
+    }
+
+    switch (event.kind) {
+      case "sessionStart": {
+        if (machine.getState() === "starting") machine.markReady();
+        break;
+      }
+      case "busy": {
+        entry.turnEventCount = 0; // new turn → reset per-turn budget
+        safeMachineCall(() => machine.signalBusy());
+        break;
+      }
+      case "waitingOnInput": {
+        safeMachineCall(() => machine.signalWaitingOnInput());
+        // Notification dispatch is invoked per node config; it never advances or
+        // fails the state (AE2).
+        this.onNotification?.({
+          sessionId: machine.sessionId,
+          notification: event.notification,
+        });
+        break;
+      }
+      case "done": {
+        // POSITIVE completion only. Idle / output progress never reach here.
+        safeMachineCall(() => machine.signalDone());
+        break;
+      }
+      case "toolActivity":
+      case "transcript":
+      case "outputProgress": {
+        // Activity re-arms the inactivity watchdog but NEVER advances to done.
+        machine.signalOutputProgress();
+        break;
+      }
+    }
+  }
+}
+
+/**
+ * State-machine calls can throw InvalidCliTransitionError when a stray event
+ * arrives in a state that doesn't accept it (e.g. a `busy` event after `done`).
+ * Telemetry is best-effort: swallow the transition error rather than crash
+ * ingest — the authoritative state simply doesn't move.
+ */
+function safeMachineCall(fn: () => void): void {
+  try {
+    fn();
+  } catch (err) {
+    if (err instanceof Error && (err as { code?: string }).code === "INVALID_CLI_TRANSITION") {
+      return;
+    }
+    throw err;
+  }
+}