Skip to content

Commit 6568833

Browse files
Merge pull request #373 from corbitsdev/cl-5598-make-the-reactordone-and-inferencedone-distinction
2 parents 0978d6b + c8cd470 commit 6568833

14 files changed

Lines changed: 228 additions & 15 deletions

docs/ARCHITECTURE.md

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,20 @@ This repeats until the director emits `capabilities.done()`.
1818

1919
| Event | When it fires |
2020
|---|---|
21-
| `inference.done` | The LLM finished one assistant turn. Carries the full turn content. |
21+
| `inference.done` | The LLM finished one assistant turn. Carries the full turn content. Fires once per turn, every turn — this is the **turn boundary**. |
2222
| `tool.done` | One tool call completed. Carries the result and the original `callId`. |
23+
| `reactor.done` | The reactor loop shut down. Fires once, at the end of the run — not between turns. |
24+
25+
`inference.done` and `reactor.done` read as near-synonyms at a call site but
26+
answer different questions: "did a turn end" versus "did the reactor shut
27+
down." Code that needs either answer should go through the `onTurnBoundary`
28+
/ `onReactorShutdown` guards in `src/agent/reactor-events.ts` rather than
29+
comparing `event.type` to a string directly — naming the question makes the
30+
right thing easier to write than the wrong one.
31+
32+
This is a convention, not an enforced constraint: nothing stops a future
33+
call site from writing `event.type === "reactor.done"` directly instead of
34+
reaching for the guard.
2335

2436
### ReactorActions
2537

src/agent/compaction.ts

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import type {
88
import { compactionThresholdFor, contextTokensFromUsage } from "../provider/context-window.js";
99
import { COMPACTOR_KEEP_RECENT_TURNS, compactorNoOpFloor } from "../session/compactor.js";
1010
import { createContextEstimate, estimateOverheadTokens } from "./context-estimate.js";
11+
import { onTurnBoundary } from "./reactor-events.js";
1112

1213
const COMPACTOR_NAME = "pruning-compactor";
1314
// The exact turn count `createPruningCompactor` (session/compactor.ts) is
@@ -119,7 +120,7 @@ export function createCompactionGovernor(
119120
// compact when it (or the operator's next message) arrives.
120121
function noteIdleTurn(event: ReactorInboundEvent, actions: ReactorAction[]): void {
121122
if (!pending || idlePending || requestContinuation === undefined) return;
122-
if (event.type !== "inference.done") return;
123+
if (!onTurnBoundary(event)) return;
123124
const terminal =
124125
actions.some((a) => a.type === "reply" || a.type === "wait") &&
125126
!actions.some((a) => a.type === "infer" || a.type === "execute_tools");

src/agent/director.ts

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import {
1515
} from "../session/compactor.js";
1616
import type { WorkflowCoordinator } from "../workflows/coordinator.js";
1717
import { createCompactionGovernor, type CompactionGovernor } from "./compaction.js";
18+
import { onTurnBoundary } from "./reactor-events.js";
1819
import { type } from "arktype";
1920
import { applyManageTasks, hasActiveTasks, parseManageTasksArgs, type Task } from "./tasks.js";
2021
import { createCorbitsRetryPolicy } from "./retry-policy.js";
@@ -523,7 +524,7 @@ class ChatDirectorImpl extends DefaultDirector {
523524
this.pendingToolOnlyNudge = false;
524525
this.pausedForToolOnly = false;
525526
}
526-
if (event.type === "inference.done") this.inferenceRecoveries = 0;
527+
if (onTurnBoundary(event)) this.inferenceRecoveries = 0;
527528

528529
if (event.type === "message.received" && this.taskClassifier !== undefined) {
529530
const message = event.message;
@@ -564,7 +565,7 @@ class ChatDirectorImpl extends DefaultDirector {
564565
}
565566
}
566567

567-
if (event.type === "inference.done") {
568+
if (onTurnBoundary(event)) {
568569
this.turnCount++;
569570
const hasToolCalls = event.turn.content.some((b) => b.type === "tool_call");
570571
const hasText = event.turn.content.some(
@@ -679,7 +680,7 @@ class ChatDirectorImpl extends DefaultDirector {
679680
// prefers provider usage when present.
680681
const turns = state.turns ?? [];
681682
this.compaction.syncFromTurns(turns);
682-
if (event.type === "inference.done") {
683+
if (onTurnBoundary(event)) {
683684
this.compaction.noteInferenceDone(event, turns);
684685
}
685686

src/agent/reactor-events.test.ts

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
import { describe, expect, test } from "bun:test";
2+
import type { ReactorEmittedEvent } from "@intx/inference";
3+
import type { ReactorInboundEvent } from "@intx/types/runtime";
4+
import { onReactorShutdown, onTurnBoundary } from "./reactor-events.js";
5+
6+
// Bare `{ type: string }` literals only prove the string comparison works.
7+
// The generic exists so the guards narrow across both `ReactorInboundEvent`
8+
// (the director-facing union, `src/agent/director.ts` / `compaction.ts`) and
9+
// `ReactorEmittedEvent` (the stream-facing union consumers see) without
10+
// redeclaring either union in `reactor-events.ts`. These tests drive real
11+
// members of both unions through the guards so a future change that breaks
12+
// narrowing on either union — e.g. a renamed variant, or the guard's
13+
// signature drifting to accept only one union — fails here instead of
14+
// surfacing as a silent `never` match downstream.
15+
16+
// `reactor.done` is emitted-only: it does not exist on `ReactorInboundEvent`
17+
// at all, so `onReactorShutdown` narrows to `never` for every director-side
18+
// event. That is exactly the distinction the doc and the guard both draw
19+
// ("did a turn end" is a question directors ask; "did the reactor shut
20+
// down" is not), and it is a fact the integration harness cannot exercise
21+
// on its own — the agent stream never hands a `ReactorInboundEvent` to
22+
// application code, only `ReactorEmittedEvent`.
23+
const inboundEvents: ReactorInboundEvent[] = [
24+
{
25+
type: "message.received",
26+
message: { role: "user", content: [{ type: "text", text: "hi" }] },
27+
} as unknown as ReactorInboundEvent,
28+
{
29+
type: "inference.done",
30+
turn: {},
31+
usage: {},
32+
source: {},
33+
} as unknown as ReactorInboundEvent,
34+
{
35+
type: "inference.error",
36+
error: {},
37+
partial: {},
38+
} as unknown as ReactorInboundEvent,
39+
{ type: "tool.done", result: {} } as unknown as ReactorInboundEvent,
40+
{
41+
type: "reactor.gate.cleared",
42+
gateId: "g1",
43+
reason: "resolved",
44+
} as unknown as ReactorInboundEvent,
45+
{ type: "abort", reason: {} } as unknown as ReactorInboundEvent,
46+
];
47+
48+
const emittedEvents: ReactorEmittedEvent[] = [
49+
{
50+
type: "message.received",
51+
seq: 0,
52+
data: { message: { role: "user", content: [{ type: "text", text: "hi" }] } },
53+
} as unknown as ReactorEmittedEvent,
54+
{ type: "inference.done", data: {} } as unknown as ReactorEmittedEvent,
55+
{ type: "reactor.done", data: {} } as unknown as ReactorEmittedEvent,
56+
{ type: "tool.done", data: {} } as unknown as ReactorEmittedEvent,
57+
];
58+
59+
describe("onTurnBoundary", () => {
60+
test("narrows ReactorInboundEvent to exactly the inference.done member", () => {
61+
const matches = inboundEvents.filter(onTurnBoundary);
62+
expect(matches.map((e) => e.type)).toEqual(["inference.done"]);
63+
// Type-level: narrowing must land on the real union member, with its
64+
// real fields, not an unrelated shape — this line fails to compile if
65+
// onTurnBoundary stops narrowing E correctly.
66+
const [narrowed] = matches;
67+
const _turn: unknown = narrowed?.turn;
68+
void _turn;
69+
});
70+
71+
test("narrows ReactorEmittedEvent to exactly the inference.done member", () => {
72+
const matches = emittedEvents.filter(onTurnBoundary);
73+
expect(matches.map((e) => e.type)).toEqual(["inference.done"]);
74+
});
75+
76+
// The property all three shipped defects violated: code that gated a
77+
// turn boundary on `reactor.done` only ever saw it once, at shutdown.
78+
// A multi-turn session must trip this guard once per turn.
79+
test("fires more than once across a multi-turn stream of real events", () => {
80+
const turnEvents: ReactorEmittedEvent[] = [
81+
{ type: "inference.start", data: {} } as unknown as ReactorEmittedEvent,
82+
{ type: "inference.done", data: {} } as unknown as ReactorEmittedEvent,
83+
{ type: "tool.done", data: {} } as unknown as ReactorEmittedEvent,
84+
{ type: "inference.done", data: {} } as unknown as ReactorEmittedEvent,
85+
{ type: "inference.done", data: {} } as unknown as ReactorEmittedEvent,
86+
];
87+
88+
const boundaries = turnEvents.filter(onTurnBoundary);
89+
90+
expect(boundaries.length).toBe(3);
91+
expect(boundaries.length).toBeGreaterThan(1);
92+
});
93+
});
94+
95+
describe("onReactorShutdown", () => {
96+
test("never matches any ReactorInboundEvent member — reactor.done is emitted-only", () => {
97+
const matches = inboundEvents.filter(onReactorShutdown);
98+
expect(matches).toEqual([]);
99+
});
100+
101+
test("narrows ReactorEmittedEvent to exactly the reactor.done member, once per session", () => {
102+
const sessionEvents: ReactorEmittedEvent[] = [
103+
{ type: "inference.done", data: {} } as unknown as ReactorEmittedEvent,
104+
{ type: "inference.done", data: {} } as unknown as ReactorEmittedEvent,
105+
{ type: "inference.done", data: {} } as unknown as ReactorEmittedEvent,
106+
{ type: "reactor.done", data: {} } as unknown as ReactorEmittedEvent,
107+
];
108+
109+
const matches = emittedEvents.filter(onReactorShutdown);
110+
expect(matches.map((e) => e.type)).toEqual(["reactor.done"]);
111+
112+
const shutdowns = sessionEvents.filter(onReactorShutdown);
113+
expect(shutdowns.length).toBe(1);
114+
});
115+
});

src/agent/reactor-events.ts

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
/**
2+
* `inference.done` and `reactor.done` read as near-synonyms at a call site
3+
* but mean opposite things: `inference.done` fires once per turn (the
4+
* boundary code that reacts "between turns" needs), while `reactor.done`
5+
* fires once, at reactor shutdown. Three shipped defects (queued messages
6+
* never dispatching, `run.json`'s `turnsUsed` freezing for a whole session,
7+
* and the shell not returning to idle between turns) all came from code
8+
* keying off `reactor.done` when it meant `inference.done`. These guards
9+
* make the two impossible to confuse: name the question, not the string.
10+
*
11+
* Generic over the event's own type so this narrows both `ReactorInboundEvent`
12+
* (`@intx/types/runtime`) and `ReactorEmittedEvent` (`@intx/inference`)
13+
* call sites without re-declaring the union here.
14+
*/
15+
16+
/** True when `event` is the turn boundary — fires once per turn, every turn. */
17+
export const onTurnBoundary = <E extends { type: string }>(
18+
event: E,
19+
): event is Extract<E, { type: "inference.done" }> => event.type === "inference.done";
20+
21+
/** True when `event` is reactor shutdown — fires once, at the end of the run. */
22+
export const onReactorShutdown = <E extends { type: string }>(
23+
event: E,
24+
): event is Extract<E, { type: "reactor.done" }> => event.type === "reactor.done";

src/perf/reactor-spans.ts

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
*/
2121

2222
import type { ReactorEmittedEvent } from "@intx/inference";
23+
import { onTurnBoundary } from "../agent/reactor-events.js";
2324
import { end, start } from "./index.js";
2425
import {
2526
getActiveTurnId,
@@ -82,7 +83,7 @@ function emptyState(): ObserverState {
8283
}
8384

8485
function toolCallCount(event: ReactorEmittedEvent): number {
85-
if (event.type !== "inference.done") return 0;
86+
if (!onTurnBoundary(event)) return 0;
8687
const data = event.data as {
8788
turn?: { content?: ReadonlyArray<{ type: string }> };
8889
};
@@ -117,7 +118,7 @@ function modelTags(event: ReactorEmittedEvent): Record<string, unknown> | undefi
117118
}
118119
return undefined;
119120
}
120-
if (event.type === "inference.done") {
121+
if (onTurnBoundary(event)) {
121122
const data = event.data as {
122123
source?: { provider?: unknown; model?: unknown };
123124
usage?: { input?: unknown; output?: unknown };
@@ -222,7 +223,7 @@ export function createPerfReactorObserver(): PerfReactorObserver {
222223
return;
223224
}
224225

225-
if (type === "inference.done") {
226+
if (onTurnBoundary(event)) {
226227
const tags = modelTags(event);
227228
closeInferenceTree(tags);
228229
state.pendingTools = toolCallCount(event);

src/session/hooks.ts

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import type {
1212
ToolCall,
1313
ToolResult,
1414
} from "@intx/types/runtime";
15+
import { onTurnBoundary } from "../agent/reactor-events.js";
1516

1617
import { COMMAND_NAME, SETTINGS_DIR_NAME } from "../branding.js";
1718

@@ -236,7 +237,7 @@ export function createTurnContextCollector(
236237
return;
237238
}
238239

239-
if (event.type === "inference.done") {
240+
if (onTurnBoundary(event)) {
240241
const toolCalls = event.data.turn.content
241242
.filter((block): block is Extract<typeof block, { type: "tool_call" }> => block.type === "tool_call")
242243
.map((block): ToolCall => ({

src/session/run-sink.ts

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ import type { EventEmitter } from "node:events";
22
import type { ReactorEmittedEvent } from "@intx/inference";
33
import type { TokenUsage } from "@intx/types/runtime";
44
import { createPerfReactorObserver } from "../perf/reactor-spans.js";
5+
import { onTurnBoundary } from "../agent/reactor-events.js";
56
import {
67
createTurnContextCollector,
78
type LifecycleHookManager,
@@ -112,7 +113,7 @@ export function createRunSink(args: RunSinkArgs): RunSink {
112113
// A completed inference turn supersedes a prior recoverable inference.error
113114
// (ChatDirector retries timeout/retryable/aborted). Leaving the sticky error
114115
// would mark a recovered successful send as failed.
115-
if (event.type === "inference.done") {
116+
if (onTurnBoundary(event)) {
116117
runError = undefined;
117118
}
118119
if (event.type === "reactor.error") {

src/session/stream-journal.ts

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import type { ReactorEmittedEvent } from "@intx/inference";
55
import { getLogger } from "@intx/log";
66

77
import { LOG_NAMESPACE_ROOT } from "../branding.js";
8+
import { onTurnBoundary } from "../agent/reactor-events.js";
89

910
/**
1011
* Partial-output capture for streaming inference cycles.
@@ -98,7 +99,7 @@ export function createCycleTextRecorder(
9899
if (typeof token === "string") cycleText = appendCycleText(cycleText, token);
99100
return;
100101
}
101-
if (event.type === "inference.done") {
102+
if (onTurnBoundary(event)) {
102103
cycleText = "";
103104
return;
104105
}

src/subagent/nudge-director.ts

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ import type {
1414
InferenceOptions,
1515
} from "@intx/types/runtime";
1616
import { createCompactionGovernor, type CompactionGovernor } from "../agent/compaction.js";
17+
import { onTurnBoundary } from "../agent/reactor-events.js";
1718
import {
1819
EMPTY_THRASH_STATE,
1920
nextThrashState,
@@ -137,7 +138,7 @@ export class SubAgentDirector extends DefaultDirector {
137138
// rewrites included). Arming still happens inside noteInferenceDone, which
138139
// prefers provider usage when present.
139140
this.compaction.syncFromTurns(state.turns);
140-
if (event.type === "inference.done") {
141+
if (onTurnBoundary(event)) {
141142
this.lastActivityAt = this.now();
142143
this.consecutiveStalls = 0;
143144
this.compaction.noteInferenceDone(event, state.turns);

0 commit comments

Comments
 (0)