Make the reactor.done and inference.done distinction impossible to get wrong

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
 

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");

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);
     }
 

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);
+  });
+});

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 = <E extends { type: string }>(
+  event: E,
+): event is Extract<E, { type: "inference.done" }> => event.type === "inference.done";
+
+/** True when `event` is reactor shutdown — fires once, at the end of the run. */
+export const onReactorShutdown = <E extends { type: string }>(
+  event: E,
+): event is Extract<E, { type: "reactor.done" }> => event.type === "reactor.done";

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<string, unknown> | 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);

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<typeof block, { type: "tool_call" }> => block.type === "tool_call")
           .map((block): ToolCall => ({

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") {

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;
     }

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);

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;

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<RunnerHost>
   // 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)
 

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()