diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index a59f823fd..95548271e 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -18,8 +18,20 @@ This repeats until the director emits `capabilities.done()`. | Event | When it fires | |---|---| -| `inference.done` | The LLM finished one assistant turn. Carries the full turn content. | +| `inference.done` | The LLM finished one assistant turn. Carries the full turn content. Fires once per turn, every turn — this is the **turn boundary**. | | `tool.done` | One tool call completed. Carries the result and the original `callId`. | +| `reactor.done` | The reactor loop shut down. Fires once, at the end of the run — not between turns. | + +`inference.done` and `reactor.done` read as near-synonyms at a call site but +answer different questions: "did a turn end" versus "did the reactor shut +down." Code that needs either answer should go through the `onTurnBoundary` +/ `onReactorShutdown` guards in `src/agent/reactor-events.ts` rather than +comparing `event.type` to a string directly — naming the question makes the +right thing easier to write than the wrong one. + +This is a convention, not an enforced constraint: nothing stops a future +call site from writing `event.type === "reactor.done"` directly instead of +reaching for the guard. ### ReactorActions diff --git a/src/agent/compaction.ts b/src/agent/compaction.ts index 1a3677426..c5da0f979 100644 --- a/src/agent/compaction.ts +++ b/src/agent/compaction.ts @@ -8,6 +8,7 @@ import type { import { compactionThresholdFor, contextTokensFromUsage } from "../provider/context-window.js"; import { COMPACTOR_KEEP_RECENT_TURNS, compactorNoOpFloor } from "../session/compactor.js"; import { createContextEstimate, estimateOverheadTokens } from "./context-estimate.js"; +import { onTurnBoundary } from "./reactor-events.js"; const COMPACTOR_NAME = "pruning-compactor"; // The exact turn count `createPruningCompactor` (session/compactor.ts) is @@ -119,7 +120,7 @@ export function createCompactionGovernor( // compact when it (or the operator's next message) arrives. function noteIdleTurn(event: ReactorInboundEvent, actions: ReactorAction[]): void { if (!pending || idlePending || requestContinuation === undefined) return; - if (event.type !== "inference.done") return; + if (!onTurnBoundary(event)) return; const terminal = actions.some((a) => a.type === "reply" || a.type === "wait") && !actions.some((a) => a.type === "infer" || a.type === "execute_tools"); diff --git a/src/agent/director.ts b/src/agent/director.ts index d4cf90ad1..593cfd872 100644 --- a/src/agent/director.ts +++ b/src/agent/director.ts @@ -15,6 +15,7 @@ import { } from "../session/compactor.js"; import type { WorkflowCoordinator } from "../workflows/coordinator.js"; import { createCompactionGovernor, type CompactionGovernor } from "./compaction.js"; +import { onTurnBoundary } from "./reactor-events.js"; import { type } from "arktype"; import { applyManageTasks, hasActiveTasks, parseManageTasksArgs, type Task } from "./tasks.js"; import { createCorbitsRetryPolicy } from "./retry-policy.js"; @@ -523,7 +524,7 @@ class ChatDirectorImpl extends DefaultDirector { this.pendingToolOnlyNudge = false; this.pausedForToolOnly = false; } - if (event.type === "inference.done") this.inferenceRecoveries = 0; + if (onTurnBoundary(event)) this.inferenceRecoveries = 0; if (event.type === "message.received" && this.taskClassifier !== undefined) { const message = event.message; @@ -564,7 +565,7 @@ class ChatDirectorImpl extends DefaultDirector { } } - if (event.type === "inference.done") { + if (onTurnBoundary(event)) { this.turnCount++; const hasToolCalls = event.turn.content.some((b) => b.type === "tool_call"); const hasText = event.turn.content.some( @@ -679,7 +680,7 @@ class ChatDirectorImpl extends DefaultDirector { // prefers provider usage when present. const turns = state.turns ?? []; this.compaction.syncFromTurns(turns); - if (event.type === "inference.done") { + if (onTurnBoundary(event)) { this.compaction.noteInferenceDone(event, turns); } diff --git a/src/agent/reactor-events.test.ts b/src/agent/reactor-events.test.ts new file mode 100644 index 000000000..15612e63e --- /dev/null +++ b/src/agent/reactor-events.test.ts @@ -0,0 +1,115 @@ +import { describe, expect, test } from "bun:test"; +import type { ReactorEmittedEvent } from "@intx/inference"; +import type { ReactorInboundEvent } from "@intx/types/runtime"; +import { onReactorShutdown, onTurnBoundary } from "./reactor-events.js"; + +// Bare `{ type: string }` literals only prove the string comparison works. +// The generic exists so the guards narrow across both `ReactorInboundEvent` +// (the director-facing union, `src/agent/director.ts` / `compaction.ts`) and +// `ReactorEmittedEvent` (the stream-facing union consumers see) without +// redeclaring either union in `reactor-events.ts`. These tests drive real +// members of both unions through the guards so a future change that breaks +// narrowing on either union — e.g. a renamed variant, or the guard's +// signature drifting to accept only one union — fails here instead of +// surfacing as a silent `never` match downstream. + +// `reactor.done` is emitted-only: it does not exist on `ReactorInboundEvent` +// at all, so `onReactorShutdown` narrows to `never` for every director-side +// event. That is exactly the distinction the doc and the guard both draw +// ("did a turn end" is a question directors ask; "did the reactor shut +// down" is not), and it is a fact the integration harness cannot exercise +// on its own — the agent stream never hands a `ReactorInboundEvent` to +// application code, only `ReactorEmittedEvent`. +const inboundEvents: ReactorInboundEvent[] = [ + { + type: "message.received", + message: { role: "user", content: [{ type: "text", text: "hi" }] }, + } as unknown as ReactorInboundEvent, + { + type: "inference.done", + turn: {}, + usage: {}, + source: {}, + } as unknown as ReactorInboundEvent, + { + type: "inference.error", + error: {}, + partial: {}, + } as unknown as ReactorInboundEvent, + { type: "tool.done", result: {} } as unknown as ReactorInboundEvent, + { + type: "reactor.gate.cleared", + gateId: "g1", + reason: "resolved", + } as unknown as ReactorInboundEvent, + { type: "abort", reason: {} } as unknown as ReactorInboundEvent, +]; + +const emittedEvents: ReactorEmittedEvent[] = [ + { + type: "message.received", + seq: 0, + data: { message: { role: "user", content: [{ type: "text", text: "hi" }] } }, + } as unknown as ReactorEmittedEvent, + { type: "inference.done", data: {} } as unknown as ReactorEmittedEvent, + { type: "reactor.done", data: {} } as unknown as ReactorEmittedEvent, + { type: "tool.done", data: {} } as unknown as ReactorEmittedEvent, +]; + +describe("onTurnBoundary", () => { + test("narrows ReactorInboundEvent to exactly the inference.done member", () => { + const matches = inboundEvents.filter(onTurnBoundary); + expect(matches.map((e) => e.type)).toEqual(["inference.done"]); + // Type-level: narrowing must land on the real union member, with its + // real fields, not an unrelated shape — this line fails to compile if + // onTurnBoundary stops narrowing E correctly. + const [narrowed] = matches; + const _turn: unknown = narrowed?.turn; + void _turn; + }); + + test("narrows ReactorEmittedEvent to exactly the inference.done member", () => { + const matches = emittedEvents.filter(onTurnBoundary); + expect(matches.map((e) => e.type)).toEqual(["inference.done"]); + }); + + // The property all three shipped defects violated: code that gated a + // turn boundary on `reactor.done` only ever saw it once, at shutdown. + // A multi-turn session must trip this guard once per turn. + test("fires more than once across a multi-turn stream of real events", () => { + const turnEvents: ReactorEmittedEvent[] = [ + { type: "inference.start", data: {} } as unknown as ReactorEmittedEvent, + { type: "inference.done", data: {} } as unknown as ReactorEmittedEvent, + { type: "tool.done", data: {} } as unknown as ReactorEmittedEvent, + { type: "inference.done", data: {} } as unknown as ReactorEmittedEvent, + { type: "inference.done", data: {} } as unknown as ReactorEmittedEvent, + ]; + + const boundaries = turnEvents.filter(onTurnBoundary); + + expect(boundaries.length).toBe(3); + expect(boundaries.length).toBeGreaterThan(1); + }); +}); + +describe("onReactorShutdown", () => { + test("never matches any ReactorInboundEvent member — reactor.done is emitted-only", () => { + const matches = inboundEvents.filter(onReactorShutdown); + expect(matches).toEqual([]); + }); + + test("narrows ReactorEmittedEvent to exactly the reactor.done member, once per session", () => { + const sessionEvents: ReactorEmittedEvent[] = [ + { type: "inference.done", data: {} } as unknown as ReactorEmittedEvent, + { type: "inference.done", data: {} } as unknown as ReactorEmittedEvent, + { type: "inference.done", data: {} } as unknown as ReactorEmittedEvent, + { type: "reactor.done", data: {} } as unknown as ReactorEmittedEvent, + ]; + + const matches = emittedEvents.filter(onReactorShutdown); + expect(matches.map((e) => e.type)).toEqual(["reactor.done"]); + + const shutdowns = sessionEvents.filter(onReactorShutdown); + expect(shutdowns.length).toBe(1); + }); +}); diff --git a/src/agent/reactor-events.ts b/src/agent/reactor-events.ts new file mode 100644 index 000000000..140392a7b --- /dev/null +++ b/src/agent/reactor-events.ts @@ -0,0 +1,24 @@ +/** + * `inference.done` and `reactor.done` read as near-synonyms at a call site + * but mean opposite things: `inference.done` fires once per turn (the + * boundary code that reacts "between turns" needs), while `reactor.done` + * fires once, at reactor shutdown. Three shipped defects (queued messages + * never dispatching, `run.json`'s `turnsUsed` freezing for a whole session, + * and the shell not returning to idle between turns) all came from code + * keying off `reactor.done` when it meant `inference.done`. These guards + * make the two impossible to confuse: name the question, not the string. + * + * Generic over the event's own type so this narrows both `ReactorInboundEvent` + * (`@intx/types/runtime`) and `ReactorEmittedEvent` (`@intx/inference`) + * call sites without re-declaring the union here. + */ + +/** True when `event` is the turn boundary — fires once per turn, every turn. */ +export const onTurnBoundary = ( + event: E, +): event is Extract => event.type === "inference.done"; + +/** True when `event` is reactor shutdown — fires once, at the end of the run. */ +export const onReactorShutdown = ( + event: E, +): event is Extract => event.type === "reactor.done"; diff --git a/src/perf/reactor-spans.ts b/src/perf/reactor-spans.ts index b15785bb5..dc6884c61 100644 --- a/src/perf/reactor-spans.ts +++ b/src/perf/reactor-spans.ts @@ -20,6 +20,7 @@ */ import type { ReactorEmittedEvent } from "@intx/inference"; +import { onTurnBoundary } from "../agent/reactor-events.js"; import { end, start } from "./index.js"; import { getActiveTurnId, @@ -82,7 +83,7 @@ function emptyState(): ObserverState { } function toolCallCount(event: ReactorEmittedEvent): number { - if (event.type !== "inference.done") return 0; + if (!onTurnBoundary(event)) return 0; const data = event.data as { turn?: { content?: ReadonlyArray<{ type: string }> }; }; @@ -117,7 +118,7 @@ function modelTags(event: ReactorEmittedEvent): Record | undefi } return undefined; } - if (event.type === "inference.done") { + if (onTurnBoundary(event)) { const data = event.data as { source?: { provider?: unknown; model?: unknown }; usage?: { input?: unknown; output?: unknown }; @@ -222,7 +223,7 @@ export function createPerfReactorObserver(): PerfReactorObserver { return; } - if (type === "inference.done") { + if (onTurnBoundary(event)) { const tags = modelTags(event); closeInferenceTree(tags); state.pendingTools = toolCallCount(event); diff --git a/src/session/hooks.ts b/src/session/hooks.ts index d9a19b4aa..e370342cc 100644 --- a/src/session/hooks.ts +++ b/src/session/hooks.ts @@ -12,6 +12,7 @@ import type { ToolCall, ToolResult, } from "@intx/types/runtime"; +import { onTurnBoundary } from "../agent/reactor-events.js"; import { COMMAND_NAME, SETTINGS_DIR_NAME } from "../branding.js"; @@ -236,7 +237,7 @@ export function createTurnContextCollector( return; } - if (event.type === "inference.done") { + if (onTurnBoundary(event)) { const toolCalls = event.data.turn.content .filter((block): block is Extract => block.type === "tool_call") .map((block): ToolCall => ({ diff --git a/src/session/run-sink.ts b/src/session/run-sink.ts index 6f2f63566..0e475f5df 100644 --- a/src/session/run-sink.ts +++ b/src/session/run-sink.ts @@ -2,6 +2,7 @@ import type { EventEmitter } from "node:events"; import type { ReactorEmittedEvent } from "@intx/inference"; import type { TokenUsage } from "@intx/types/runtime"; import { createPerfReactorObserver } from "../perf/reactor-spans.js"; +import { onTurnBoundary } from "../agent/reactor-events.js"; import { createTurnContextCollector, type LifecycleHookManager, @@ -112,7 +113,7 @@ export function createRunSink(args: RunSinkArgs): RunSink { // A completed inference turn supersedes a prior recoverable inference.error // (ChatDirector retries timeout/retryable/aborted). Leaving the sticky error // would mark a recovered successful send as failed. - if (event.type === "inference.done") { + if (onTurnBoundary(event)) { runError = undefined; } if (event.type === "reactor.error") { diff --git a/src/session/stream-journal.ts b/src/session/stream-journal.ts index 58236af8a..a6a6e159c 100644 --- a/src/session/stream-journal.ts +++ b/src/session/stream-journal.ts @@ -5,6 +5,7 @@ import type { ReactorEmittedEvent } from "@intx/inference"; import { getLogger } from "@intx/log"; import { LOG_NAMESPACE_ROOT } from "../branding.js"; +import { onTurnBoundary } from "../agent/reactor-events.js"; /** * Partial-output capture for streaming inference cycles. @@ -98,7 +99,7 @@ export function createCycleTextRecorder( if (typeof token === "string") cycleText = appendCycleText(cycleText, token); return; } - if (event.type === "inference.done") { + if (onTurnBoundary(event)) { cycleText = ""; return; } diff --git a/src/subagent/nudge-director.ts b/src/subagent/nudge-director.ts index dc233fe93..4038af483 100644 --- a/src/subagent/nudge-director.ts +++ b/src/subagent/nudge-director.ts @@ -14,6 +14,7 @@ import type { InferenceOptions, } from "@intx/types/runtime"; import { createCompactionGovernor, type CompactionGovernor } from "../agent/compaction.js"; +import { onTurnBoundary } from "../agent/reactor-events.js"; import { EMPTY_THRASH_STATE, nextThrashState, @@ -137,7 +138,7 @@ export class SubAgentDirector extends DefaultDirector { // rewrites included). Arming still happens inside noteInferenceDone, which // prefers provider usage when present. this.compaction.syncFromTurns(state.turns); - if (event.type === "inference.done") { + if (onTurnBoundary(event)) { this.lastActivityAt = this.now(); this.consecutiveStalls = 0; this.compaction.noteInferenceDone(event, state.turns); diff --git a/src/subagent/stop-policy.ts b/src/subagent/stop-policy.ts index 38c77513a..31abada22 100644 --- a/src/subagent/stop-policy.ts +++ b/src/subagent/stop-policy.ts @@ -4,6 +4,7 @@ */ import type { ReactorEmittedEvent } from "@intx/inference"; +import { onTurnBoundary } from "../agent/reactor-events.js"; import { evaluateThrashStop, type ThrashConfig, @@ -225,7 +226,7 @@ export function lastText(content: ReadonlyArray<{ type: string }>): string { /** Best-effort partial assistant text from a stream event (inference.done). */ export function partialTextFromEvent(event: ReactorEmittedEvent): string | null { - if (event.type !== "inference.done") return null; + if (!onTurnBoundary(event)) return null; // Stream events nest the turn under data (same shape as hooks/renderer). // Guard data.turn so a malformed event cannot throw in the stream sink. const turn = event.data?.turn; diff --git a/src/tui-opentui/runner-host.ts b/src/tui-opentui/runner-host.ts index 888e45d15..c3307579d 100644 --- a/src/tui-opentui/runner-host.ts +++ b/src/tui-opentui/runner-host.ts @@ -28,6 +28,7 @@ import { } from "./model-catalog.js" import type { ItemDescription } from "./shell.js" import { mountProductHost, type ProductHost } from "./product-host.js" +import { onTurnBoundary } from "../agent/reactor-events.js" import { appendStreamRow, clearShellExitHandler, @@ -274,7 +275,7 @@ export async function mountRunnerHost(deps: RunnerHostDeps): Promise // Every completed inference turn changes both cost and context usage; // nothing else needs a fresher read than that. const onCostEvent = (event: { type: string }): void => { - if (event.type === "inference.done") pushCostContext() + if (onTurnBoundary(event)) pushCostContext() } deps.eventEmitter.on("event", onCostEvent) diff --git a/src/tui-opentui/runtime-bridge.ts b/src/tui-opentui/runtime-bridge.ts index 412597941..1da4580f2 100644 --- a/src/tui-opentui/runtime-bridge.ts +++ b/src/tui-opentui/runtime-bridge.ts @@ -33,6 +33,7 @@ import { type AppShell, } from "./shell.js" import { rampFor, rampLine } from "./ramp.js" +import { onTurnBoundary } from "../agent/reactor-events.js" import { resolveRampPhase, resolveTurnLabel, @@ -829,7 +830,7 @@ export function attachSessionBridge( // turn (see turn-state.ts) — the cycle continues, but a boundary still // passed, so a queued message waiting on it should not wait for the // turn's eventual end too. - if (event.type === "inference.done" && bag.turn.activeToolCalls.length > 0) { + if (onTurnBoundary(event) && bag.turn.activeToolCalls.length > 0) { drainAtBoundary(shell, bag) } if (settled) settleRun() diff --git a/tests/integration/reactor-events-guards.test.ts b/tests/integration/reactor-events-guards.test.ts new file mode 100644 index 000000000..dac1e01a7 --- /dev/null +++ b/tests/integration/reactor-events-guards.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, test } from "bun:test"; +import type { ReactorEmittedEvent } from "@intx/inference"; + +import { onTurnBoundary } from "../../src/agent/reactor-events.js"; +import { createPermissionGate } from "../../src/permission/gate.js"; +import { closeIntegrationSession, openIntegrationSession, runUntilDone } from "./harness.js"; + +// The unit tests in `src/agent/reactor-events.test.ts` cover type-level +// narrowing across both event unions and `onReactorShutdown`'s behavior. +// This test asserts the property `onTurnBoundary` exists for — it matches +// exactly once per turn — against a real reactor's real emitted events, +// not a synthetic filtered array of hand-built literals. +// +// `onReactorShutdown` is not exercised here: `@intx/agent`'s `close()` +// clears `stream()` consumers synchronously, before the queued abort that +// produces `reactor.done` is processed, so application code attached via +// `agent.stream()` cannot observe it after `close()` — the same reason +// `src/session/run-sink.ts` snapshots status *before* close instead of +// relying on `reactor.done` to arrive. That gap is covered by the unit +// test's real `ReactorEmittedEvent` / `ReactorInboundEvent` narrowing. +describe("integration — reactor-events guards", () => { + test.serial("onTurnBoundary matches exactly the turn boundary, once per real turn", async () => { + const session = await openIntegrationSession({ + permissionGate: createPermissionGate({ + approvals: [], + interactive: false, + skipPermissions: true, + }), + }); + + try { + session.harness.scenario.replyOnce("anthropic", { + toolCalls: [{ name: "write_file", args: { path: "out.txt", content: "ok\n" } }], + }); + session.harness.scenario.replyOnce("anthropic", { text: "Done." }); + + const { events } = await runUntilDone(session, "Write out.txt with content ok."); + + const turnBoundaries = events.filter(onTurnBoundary); + + // One tool-call turn followed by one final-text turn: exactly two + // inference.done events, despite tool.done and other events on the stream. + expect(turnBoundaries.length).toBe(2); + expect(turnBoundaries.every((e: ReactorEmittedEvent) => e.type === "inference.done")).toBe( + true, + ); + expect(events.some((e) => e.type === "tool.done")).toBe(true); + } finally { + await closeIntegrationSession(session); + } + }); +});