From ed4c7ba00603dac0192c3edb37efd3d917d55cd3 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Thu, 4 Jun 2026 23:18:48 -0700 Subject: [PATCH] feat(engine): add cli-agent telemetry hub and session state machine (U3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../cli-agent/__tests__/state-machine.test.ts | 297 ++++++++++ .../cli-agent/__tests__/telemetry-hub.test.ts | 221 ++++++++ .../engine/src/cli-agent/state-machine.ts | 532 ++++++++++++++++++ .../engine/src/cli-agent/telemetry-hub.ts | 387 +++++++++++++ 4 files changed, 1437 insertions(+) create mode 100644 packages/engine/src/cli-agent/__tests__/state-machine.test.ts create mode 100644 packages/engine/src/cli-agent/__tests__/telemetry-hub.test.ts create mode 100644 packages/engine/src/cli-agent/state-machine.ts create mode 100644 packages/engine/src/cli-agent/telemetry-hub.ts diff --git a/packages/engine/src/cli-agent/__tests__/state-machine.test.ts b/packages/engine/src/cli-agent/__tests__/state-machine.test.ts new file mode 100644 index 0000000000..2935515114 --- /dev/null +++ b/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 { + const s = store.createSession({ + purpose: "execute", + projectId: "proj", + adapterId: "claude-code", + ...overrides, + }); + return s.id; + } + + function makeMachine( + sessionId: string, + opts: Partial[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); + }); +}); diff --git a/packages/engine/src/cli-agent/__tests__/telemetry-hub.test.ts b/packages/engine/src/cli-agent/__tests__/telemetry-hub.test.ts new file mode 100644 index 0000000000..bc7851bf11 --- /dev/null +++ b/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 { + 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"); + }); +}); diff --git a/packages/engine/src/cli-agent/state-machine.ts b/packages/engine/src/cli-agent/state-machine.ts new file mode 100644 index 0000000000..9ad6fa6729 --- /dev/null +++ b/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(); + + // 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)); + + // 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); + } + } +} diff --git a/packages/engine/src/cli-agent/telemetry-hub.ts b/packages/engine/src/cli-agent/telemetry-hub.ts new file mode 100644 index 0000000000..9fea4e5250 --- /dev/null +++ b/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 & { + /** 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; + }; +} + +/** 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; + /** 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 | 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(); + private readonly sessions = new Map(); + + 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; + } + + 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; + } +}