Add named guards for the turn-boundary vs reactor-shutdown split

inference.done and reactor.done read as near-synonyms at a call
site but mean opposite things: inference.done fires once per turn,
reactor.done fires once at shutdown. Three shipped defects came
from code that needed a turn boundary but keyed off reactor.done
instead. Route every such check through onTurnBoundary /
onReactorShutdown in src/agent/reactor-events.ts so the mistake
can't be reintroduced by a bare string comparison.

Ten call sites converted, including a fifth misuse the original
audit missed: run-sink.ts's sticky-error clear at line ~115 sat
three lines below a legitimate reactor.done shutdown check at
line ~107 and rode along as "already reviewed" on the strength of
its neighbor. That shutdown check, plus renderer.ts, stream-event-
map.ts, and turn-state.ts, remain untouched as genuine shutdown
semantics. The run.json snapshot trigger in tui/runner.ts is also
left alone: it is mid-rework on another branch to move off
reactor.done, so touching it here would collide with that change.

Author: TheGreatAxios

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,45 @@
+import { describe, expect, test } from "bun:test";
+import { onReactorShutdown, onTurnBoundary } from "./reactor-events.js";
+
+describe("onTurnBoundary", () => {
+  test("true only for inference.done", () => {
+    expect(onTurnBoundary({ type: "inference.done" })).toBe(true);
+    expect(onTurnBoundary({ type: "reactor.done" })).toBe(false);
+    expect(onTurnBoundary({ type: "tool.done" })).toBe(false);
+  });
+
+  // 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 session", () => {
+    const turnEvents = [
+      { type: "inference.start" },
+      { type: "inference.done" },
+      { type: "tool.done" },
+      { type: "inference.done" },
+      { type: "inference.done" },
+    ];
+
+    const boundaries = turnEvents.filter((event) => onTurnBoundary(event));
+
+    expect(boundaries.length).toBe(3);
+    expect(boundaries.length).toBeGreaterThan(1);
+  });
+});
+
+describe("onReactorShutdown", () => {
+  test("true only for reactor.done, and fires once per session", () => {
+    const sessionEvents = [
+      { type: "inference.done" },
+      { type: "inference.done" },
+      { type: "inference.done" },
+      { type: "reactor.done" },
+    ];
+
+    expect(onReactorShutdown({ type: "reactor.done" })).toBe(true);
+    expect(onReactorShutdown({ type: "inference.done" })).toBe(false);
+
+    const shutdowns = sessionEvents.filter((event) => onReactorShutdown(event));
+    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;