From bd87fdcf5c5e9e4589ffa67c100574c9e8a2a40d Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 7 Aug 2026 08:18:20 -0700 Subject: [PATCH 1/3] 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. --- src/agent/compaction.ts | 3 ++- src/agent/director.ts | 7 ++--- src/agent/reactor-events.test.ts | 45 +++++++++++++++++++++++++++++++ src/agent/reactor-events.ts | 24 +++++++++++++++++ src/perf/reactor-spans.ts | 7 ++--- src/session/hooks.ts | 3 ++- src/session/run-sink.ts | 3 ++- src/session/stream-journal.ts | 3 ++- src/subagent/nudge-director.ts | 3 ++- src/subagent/stop-policy.ts | 3 ++- src/tui-opentui/runner-host.ts | 3 ++- src/tui-opentui/runtime-bridge.ts | 3 ++- 12 files changed, 93 insertions(+), 14 deletions(-) create mode 100644 src/agent/reactor-events.test.ts create mode 100644 src/agent/reactor-events.ts 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..c2ae9b14b --- /dev/null +++ b/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); + }); +}); 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() From 920041c0db4c25848402ec4fff452d86d09a2df4 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 7 Aug 2026 08:18:26 -0700 Subject: [PATCH 2/3] Document the reactor.done vs inference.done distinction Names the failure mode in the events table, cites the three defects it caused, and points at the onTurnBoundary / onReactorShutdown guards. States plainly that this is a naming convention rather than an enforced constraint: no lint tooling is configured in this repo to add a restricted-syntax rule, and a type-level fix would require modifying the vendored @intx/types / @intx/inference packages, which is off-limits. --- docs/ARCHITECTURE.md | 27 ++++++++++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index a59f823fd..d39de6ad6 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -18,8 +18,33 @@ 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." Three shipped defects came from code that needed a turn boundary but +keyed off `reactor.done` instead: queued messages never dispatched because +the send-queue drain waited for shutdown; `run.json`'s `turnsUsed` froze for +an entire session because the mid-run snapshot only re-fired on shutdown; +and the shell run state didn't return to idle between turns. Documentation +didn't prevent the second and third instances, so code that needs to ask +"did a turn end" or "did the reactor shut down" 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. Two enforcement routes were considered and both are +out of scope here — a lint rule (`no-restricted-syntax` or similar) would +mean standing up ESLint or Biome from scratch, since neither is configured +anywhere in this repo, disproportionate for a Low-priority cleanup; and a +type-level fix branding `event.type` would require modifying `@intx/types` +or `@intx/inference`, which are vendored and off-limits. Reviewers should +treat a bare `event.type === "reactor.done"` / `"inference.done"` comparison +outside `reactor-events.ts` as a signal to ask why the guard wasn't used. ### ReactorActions From c8cd470dff197e871d3f5c808d3ba9e10cbecf84 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 7 Aug 2026 08:44:17 -0700 Subject: [PATCH 3/3] Trim the reactor-events doc and cover real event unions The ARCHITECTURE.md block explaining inference.done vs reactor.done carried a full rejected-alternatives essay on ESLint and Biome, which reads as noise once someone actually configures a linter. Cut it to the table row, one paragraph naming the distinction, and the note that this is a convention rather than an enforced constraint. The guard tests previously passed bare { type: string } literals, which only proves the string comparison works. Drive real ReactorInboundEvent and ReactorEmittedEvent members through onTurnBoundary and onReactorShutdown instead, so a change that breaks narrowing on either union is caught. Add an integration test that exercises onTurnBoundary against a real reactor run's emitted events; onReactorShutdown's shutdown path is not integration-testable because agent.close() clears stream() consumers before the queued abort event produces reactor.done, so app code can't observe it there either - the unit test covers that guard's correctness instead. --- docs/ARCHITECTURE.md | 23 +--- src/agent/reactor-events.test.ts | 112 ++++++++++++++---- .../integration/reactor-events-guards.test.ts | 52 ++++++++ 3 files changed, 148 insertions(+), 39 deletions(-) create mode 100644 tests/integration/reactor-events-guards.test.ts diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index d39de6ad6..95548271e 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -24,27 +24,14 @@ This repeats until the director emits `capabilities.done()`. `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." Three shipped defects came from code that needed a turn boundary but -keyed off `reactor.done` instead: queued messages never dispatched because -the send-queue drain waited for shutdown; `run.json`'s `turnsUsed` froze for -an entire session because the mid-run snapshot only re-fired on shutdown; -and the shell run state didn't return to idle between turns. Documentation -didn't prevent the second and third instances, so code that needs to ask -"did a turn end" or "did the reactor shut down" 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. +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. Two enforcement routes were considered and both are -out of scope here — a lint rule (`no-restricted-syntax` or similar) would -mean standing up ESLint or Biome from scratch, since neither is configured -anywhere in this repo, disproportionate for a Low-priority cleanup; and a -type-level fix branding `event.type` would require modifying `@intx/types` -or `@intx/inference`, which are vendored and off-limits. Reviewers should -treat a bare `event.type === "reactor.done"` / `"inference.done"` comparison -outside `reactor-events.ts` as a signal to ask why the guard wasn't used. +reaching for the guard. ### ReactorActions diff --git a/src/agent/reactor-events.test.ts b/src/agent/reactor-events.test.ts index c2ae9b14b..15612e63e 100644 --- a/src/agent/reactor-events.test.ts +++ b/src/agent/reactor-events.test.ts @@ -1,26 +1,91 @@ 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("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); + 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 session", () => { - const turnEvents = [ - { type: "inference.start" }, - { type: "inference.done" }, - { type: "tool.done" }, - { type: "inference.done" }, - { type: "inference.done" }, + 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((event) => onTurnBoundary(event)); + const boundaries = turnEvents.filter(onTurnBoundary); expect(boundaries.length).toBe(3); expect(boundaries.length).toBeGreaterThan(1); @@ -28,18 +93,23 @@ describe("onTurnBoundary", () => { }); 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" }, + 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, ]; - expect(onReactorShutdown({ type: "reactor.done" })).toBe(true); - expect(onReactorShutdown({ type: "inference.done" })).toBe(false); + const matches = emittedEvents.filter(onReactorShutdown); + expect(matches.map((e) => e.type)).toEqual(["reactor.done"]); - const shutdowns = sessionEvents.filter((event) => onReactorShutdown(event)); + const shutdowns = sessionEvents.filter(onReactorShutdown); expect(shutdowns.length).toBe(1); }); }); 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); + } + }); +});