refactor(chat): express compaction continuation as reactor action (#1041)

* feat(chat): express compaction continuation as a reactor emit action

Removes requestContinuation from ChatDirectorOptions. The compaction governor now appends a custom.compaction.continue emit action wherever it previously invoked the closure, and the TUI and exec hosts answer that event by delivering the compaction continuation message.

* fix(chat): validate continuation events before delivering after compact (#1051)

* fix(chat): validate continuation events before delivering after compact

Hosts answer each compaction continuation emit at most once via a seq-keyed consume-once gate, and the director holds unsolicited empty continuations with wait instead of inferring, so forged or replayed events cannot burn billable model turns.

* fix(chat): pin continuation polish leftovers from emit-and-deliver (#1060)

Summary: update the architecture doc to the emit-and-deliver continuation mechanism; collapse idle-turn arming into a single-delivery contract (closure fires and the flag stays false, or no closure and the flag reports the arming) with a guard test; extract the session deliver generation re-check into runGenerationGuardedDeliver with race-pinning tests. Emit shape, guards, and gates unchanged.

Verification: bun test src/tui/deliver-agent-message.test.ts src/agent/compaction.test.ts (38 pass); bun run check (7396 pass, 0 fail).

Author: TheGreatAxios

docs/ARCHITECTURE.md

@@ -177,7 +177,7 @@ When a cycle's input tokens cross a threshold, the director compacts the inferen
 - **Idle (end-of-turn)** — An interactive turn can end with a reply and then sit idle with no tool batch to intercept; the governor requests a continuation at that pause and compacts when it arrives. An operator message that races the continuation still compacts first, then re-enters inference to answer it.
 - **Overflow recovery** — A `context_overflow` inference error would otherwise become a terminal error reply; the governor compacts and retries instead, bounded so a history the compactor cannot shrink does not loop forever. Overflow ignores hysteresis for the compact itself.
 
-The compaction control flow is shaped by a reactor invariant: a `compact` action runs in its own cycle (it cannot be paired with `infer`), and **the reactor delivers no event after a compact cycle**. A director that simply emitted `compact` in place of the follow-up `infer` would leave the loop idle forever — the cause of an earlier stall. Instead the governor, after emitting `compact`, self-delivers a content-less inbound message (a host-supplied `requestContinuation` callback). That message adds no turn (`createInboundTurn` returns `null` for empty content) but re-enters the loop, where the director issues the follow-up `infer` against the freshly truncated history.
+The compaction control flow is shaped by a reactor invariant: a `compact` action runs in its own cycle (it cannot be paired with `infer`), and **the reactor delivers no event after a compact cycle**. A director that simply emitted `compact` in place of the follow-up `infer` would leave the loop idle forever — the cause of an earlier stall. Instead the governor pairs `compact` with an emit action (`custom.compaction.continue`), and the host answers that emission by delivering a content-less inbound message (`buildCompactionContinuationMessage()`) through the serial operation queue, generation-guarded like every other deliver. That message adds no turn (`createInboundTurn` returns `null` for empty content) but re-enters the loop, where the director issues the follow-up `infer` against the freshly truncated history. Each emission is answered at most once (a replayed duplicate of an already-answered emission is ignored), and an unsolicited continuation with neither resume flag set answers `wait` rather than burning a billable inference. The legacy `requestContinuation` closure survives only on the sub-agent path, which delivers directly with no host emit hop; idle arming keeps the two channels exclusive (closure fires and the arming flag stays false, or no closure and the flag reports the arming) so a caller honoring both cannot double-deliver.
 
 Compaction replaces older turns with a structured, workflow-aware summary rather than a stats blob: sections for **What Happened / What We're Doing / Relevant Links / Action Items / Next Steps**, with the active workflow and step woven in so compacting mid-`/build` or mid-`/plan` preserves the contract. The summary is produced by a one-shot model call; on any failure it falls back to a deterministic summary so a compaction cycle never breaks the session.
 

src/agent/compaction.test.ts

@@ -6,7 +6,10 @@ import type {
   ReactorInboundEvent,
   TokenUsage,
 } from "@intx/types/runtime";
-import { createCompactionGovernor } from "./compaction.js";
+import {
+  COMPACTION_CONTINUATION_EVENT,
+  createCompactionGovernor,
+} from "./compaction.js";
 import {
   compactionResumeDeltaFor,
   compactionThresholdFor,
@@ -28,6 +31,11 @@ const capabilities = {
     compactor,
     reason,
   }),
+  emit: (eventType: string, data: unknown) => ({
+    type: "emit",
+    eventType,
+    data,
+  }),
 } as unknown as ReactorCapabilities;
 
 // Distinct, non-zero cacheRead/cacheWrite so a test asserting on the total
@@ -182,15 +190,34 @@ describe("compaction governor", () => {
     ).toBeNull();
   });
 
-  test("stays inert without a continuation channel", () => {
+  test("expresses continuation as an emit action without a closure", () => {
     const governor = createCompactionGovernor(undefined);
     governor.noteInferenceDone(inferenceDone(overThreshold), tenTurns);
+    const actions = governor.interceptActions(
+      toolDone(),
+      inferAction,
+      capabilities,
+    );
+    expect(actions?.some((a) => a.type === "compact")).toBe(true);
+    expect(actions?.some((a) => a.type === "infer")).toBe(false);
     expect(
-      governor.interceptActions(toolDone(), inferAction, capabilities),
-    ).toBeNull();
-    expect(
-      governor.interceptOverflow(overflowError(), capabilities),
-    ).toBeNull();
+      actions?.some(
+        (a) =>
+          a.type === "emit" &&
+          "eventType" in a &&
+          a.eventType === COMPACTION_CONTINUATION_EVENT,
+      ),
+    ).toBe(true);
+    expect(
+      governor
+        .interceptOverflow(overflowError(), capabilities)
+        ?.some(
+          (a) =>
+            a.type === "emit" &&
+            "eventType" in a &&
+            a.eventType === COMPACTION_CONTINUATION_EVENT,
+        ),
+    ).toBe(true);
   });
 
   test("recovers from context overflow a bounded number of times", () => {
@@ -241,6 +268,61 @@ describe("compaction governor", () => {
     ).toBeNull();
   });
 
+  test("an idle over-threshold turn arms an emit continuation without a closure", () => {
+    const governor = createCompactionGovernor(undefined);
+    governor.noteInferenceDone(inferenceDone(overThreshold), tenTurns);
+
+    const terminal: ReactorAction[] = [{ type: "reply", content: "done" }];
+    expect(governor.noteIdleTurn(inferenceDone(overThreshold), terminal)).toBe(
+      true,
+    );
+    // Only arms once even if the idle turn is observed again.
+    expect(governor.noteIdleTurn(inferenceDone(overThreshold), terminal)).toBe(
+      false,
+    );
+
+    const actions = governor.interceptIdleContinuation(
+      emptyMessage(),
+      capabilities,
+    );
+    expect(
+      actions?.some(
+        (a) =>
+          a.type === "compact" &&
+          "reason" in a &&
+          a.reason === "context-threshold",
+      ),
+    ).toBe(true);
+    expect(
+      actions?.some(
+        (a) =>
+          a.type === "emit" &&
+          "eventType" in a &&
+          a.eventType === COMPACTION_CONTINUATION_EVENT,
+      ),
+    ).toBe(true);
+  });
+
+  test("idle arming with a closure fires once and never returns true", () => {
+    // Single-delivery contract: a caller that both installs the legacy closure
+    // and honors the boolean (appending an emit action on true) must still
+    // deliver exactly once. The closure fires; the return stays false.
+    let continuations = 0;
+    const governor = createCompactionGovernor(() => continuations++);
+    governor.noteInferenceDone(inferenceDone(overThreshold), tenTurns);
+
+    const terminal: ReactorAction[] = [{ type: "reply", content: "done" }];
+    expect(governor.noteIdleTurn(inferenceDone(overThreshold), terminal)).toBe(
+      false,
+    );
+    expect(continuations).toBe(1);
+    // A repeated idle turn neither refires nor arms.
+    expect(governor.noteIdleTurn(inferenceDone(overThreshold), terminal)).toBe(
+      false,
+    );
+    expect(continuations).toBe(1);
+  });
+
   // Idle compact with an empty continuation previously left postCompactInfer
   // unset, so resumeAfterCompact never fired and notePostCompact never ran —
   // the Ctx meter stayed on pre-compact lastTurnUsage until the next user turn.

src/agent/compaction.ts

@@ -40,12 +40,26 @@ const MAX_CONSECUTIVE_THRESHOLD_COMPACTS = 2;
 // A compact action runs in its own reactor cycle, after which the reactor
 // idles until the next inbound event. Worker loops (sub-agents, the coding
 // director) have no operator to send that next message, so the governor swaps
-// the post-tool infer for a compact action, asks the host to deliver an empty
-// continuation message, and re-issues the infer when that message arrives.
-// Without a continuation channel the governor stays inert: stalling the loop
-// would be worse than growing the context.
+// the post-tool infer for a compact action, requests a continuation re-entry,
+// and re-issues the infer when that message arrives. Without a continuation
+// channel the governor stays inert: stalling the loop would be worse than
+// growing the context.
 export type CompactionGovernor = ReturnType<typeof createCompactionGovernor>;
 
+// Continuation re-entry expressed as a ReactorAction: the reactor emits this
+// event on the agent stream and the host answers it with
+// buildCompactionContinuationMessage(), the same message the old
+// requestContinuation closure delivered. Subscribers that only care about
+// provider/connector traffic must ignore this event.
+export const COMPACTION_CONTINUATION_EVENT = "custom.compaction.continue";
+
+/** Build the continuation re-entry action for a compacted governor cycle. */
+export function compactionContinuationAction(
+  capabilities: ReactorCapabilities,
+): ReactorAction {
+  return capabilities.emit(COMPACTION_CONTINUATION_EVENT, {});
+}
+
 export function createCompactionGovernor(
   requestContinuation?: () => void,
   systemPrompt = "",
@@ -119,6 +133,20 @@ export function createCompactionGovernor(
     noteCompactIssued();
   }
 
+  // Continuation re-entry for a compact-bearing return. The legacy subagent
+  // path still holds a host closure and drives its stall-ping loop through
+  // it; the chat path holds none and gets an emit action the host answers
+  // with a deliver instead.
+  function continuationActions(
+    capabilities: ReactorCapabilities,
+  ): ReactorAction[] {
+    if (requestContinuation !== undefined) {
+      requestContinuation();
+      return [];
+    }
+    return [compactionContinuationAction(capabilities)];
+  }
+
   function isSpacerEchoTerminal(
     event: ReactorInboundEvent,
     actions: ReactorAction[],
@@ -141,7 +169,6 @@ export function createCompactionGovernor(
     if (event.turn.content.some((block) => block.type === "tool_call")) {
       consecutiveThresholdCompacts = 0;
     }
-    if (requestContinuation === undefined) return;
     syncFromTurns(turns);
     lastModel = event.source?.model;
     const reportedTokens = contextTokensFromUsage(event.usage);
@@ -190,31 +217,43 @@ export function createCompactionGovernor(
     pending = false;
     postCompactInfer = true;
     issueThresholdCompact();
-    requestContinuation?.();
     return [
       ...actions.filter((a) => a.type !== "infer"),
       capabilities.compact(COMPACTOR_NAME, "context-threshold"),
+      ...continuationActions(capabilities),
     ];
   }
 
   // Interactive sessions can end a turn with a reply and then sit idle, so a
   // pending compaction would wait indefinitely for the next tool batch. When
-  // the turn ends without follow-up work, ask the host for a continuation and
+  // the turn ends without follow-up work, request a continuation re-entry and
   // compact when it (or the operator's next message) arrives.
+  //
+  // Single-delivery contract: the closure channel and the boolean return are
+  // mutually exclusive, matching continuationActions below. When a legacy
+  // closure is installed (the sub-agent path) it fires here and this returns
+  // false, so a caller that also honors the return cannot double-deliver.
+  // When no closure is installed (the chat path) nothing fires and the return
+  // reports whether this call newly armed the idle continuation — the caller
+  // must then append compactionContinuationAction to its returned actions.
   function noteIdleTurn(
     event: ReactorInboundEvent,
     actions: ReactorAction[],
-  ): void {
-    if (!pending || idlePending || requestContinuation === undefined) return;
-    if (atThresholdCompactCap()) return;
-    if (!onTurnBoundary(event)) return;
-    if (isSpacerEchoTerminal(event, actions)) return;
+  ): boolean {
+    if (!pending || idlePending) return false;
+    if (atThresholdCompactCap()) return false;
+    if (!onTurnBoundary(event)) return false;
+    if (isSpacerEchoTerminal(event, actions)) return false;
     const terminal =
       actions.some((a) => a.type === "reply" || a.type === "wait") &&
       !actions.some((a) => a.type === "infer" || a.type === "execute_tools");
-    if (!terminal) return;
+    if (!terminal) return false;
     idlePending = true;
-    requestContinuation();
+    if (requestContinuation !== undefined) {
+      requestContinuation();
+      return false;
+    }
+    return true;
   }
 
   function interceptIdleContinuation(
@@ -240,8 +279,10 @@ export function createCompactionGovernor(
       postCompactMeter = true;
     }
     issueThresholdCompact();
-    requestContinuation?.();
-    return [capabilities.compact(COMPACTOR_NAME, "context-threshold")];
+    return [
+      capabilities.compact(COMPACTOR_NAME, "context-threshold"),
+      ...continuationActions(capabilities),
+    ];
   }
 
   // A context-overflow inference error would otherwise terminate the loop
@@ -251,7 +292,6 @@ export function createCompactionGovernor(
     event: ReactorInboundEvent,
     capabilities: ReactorCapabilities,
   ): ReactorAction[] | null {
-    if (requestContinuation === undefined) return null;
     if (
       event.type !== "inference.error" ||
       event.error.category !== "context_overflow"
@@ -263,8 +303,10 @@ export function createCompactionGovernor(
     pending = false;
     postCompactInfer = true;
     noteCompactIssued();
-    requestContinuation();
-    return [capabilities.compact(COMPACTOR_NAME, "context-overflow")];
+    return [
+      capabilities.compact(COMPACTOR_NAME, "context-overflow"),
+      ...continuationActions(capabilities),
+    ];
   }
 
   // After compact, a content-less continuation re-enters decide. "infer" means
@@ -296,6 +338,14 @@ export function createCompactionGovernor(
     usingEstimate = true;
   }
 
+  // True while the governor expects the host to answer a continuation emit.
+  // The post-compact resume flags are consume-on-hit, so an empty
+  // message.received that finds neither set is unsolicited — forged or a
+  // replayed duplicate — and answering it would burn a billable inference.
+  function hasOutstandingContinuation(): boolean {
+    return postCompactInfer || postCompactMeter;
+  }
+
   return {
     get estimatedTokens(): number {
       return estimate.tokens;
@@ -314,5 +364,6 @@ export function createCompactionGovernor(
     interceptIdleContinuation,
     interceptOverflow,
     resumeAfterCompact,
+    hasOutstandingContinuation,
   };
 }

src/agent/director.ts

@@ -21,6 +21,7 @@ import {
 } from "../session/compactor.js";
 import type { WorkflowCoordinator } from "../workflows/coordinator.js";
 import {
+  compactionContinuationAction,
   createCompactionGovernor,
   type CompactionGovernor,
 } from "./compaction.js";
@@ -444,7 +445,6 @@ export interface ChatDirectorOptions {
   inactivityTimeoutMs?: number | undefined;
   totalTimeoutMs?: number | undefined;
   workflowCoordinator?: WorkflowCoordinator | undefined;
-  requestContinuation?: (() => void) | undefined;
   provider?: { providerName: string; model?: string } | undefined;
   /**
    * CL-7918 decisions (both former closures removed, no new env key):
@@ -572,8 +572,10 @@ class ChatDirectorImpl extends DefaultDirector {
     this.totalTimeoutMs = options.totalTimeoutMs;
     this.taskClassifier = options.taskClassifier;
     this.workflowCoordinator = options.workflowCoordinator;
+    // The chat path holds no continuation closure: the governor expresses
+    // continuation as an emit action the host answers with a deliver.
     this.compaction = createCompactionGovernor(
-      options.requestContinuation,
+      undefined,
       composedPrompt,
       toolDefinitions,
     );
@@ -769,6 +771,22 @@ class ChatDirectorImpl extends DefaultDirector {
     const recovery = this.compaction.interceptOverflow(event, capabilities);
     if (recovery !== null) return recovery;
 
+    // A forged or replayed compaction continuation arrives as an empty
+    // message.received with no outstanding compact state (the legit resume
+    // is consumed above). Answering it with infer would burn a billable
+    // model turn and reset the loop-protection budgets below, so hold the
+    // loop instead.
+    if (event.type === "message.received") {
+      const content =
+        typeof event.message.content === "string" ? event.message.content : "";
+      if (
+        content.length === 0 &&
+        !this.compaction.hasOutstandingContinuation()
+      ) {
+        return capabilities.wait();
+      }
+    }
+
     // Only `aborted` (internal-recovery-abort) lands here: the harness's own
     // retry policy already owns `timeout`/`retryable`/`quota_exhausted` and
     // has exhausted its own attempt budget (up to MAX_ATTEMPTS full-context
@@ -1064,13 +1082,21 @@ class ChatDirectorImpl extends DefaultDirector {
       );
     }
 
-    this.compaction.noteIdleTurn(event, baseActions);
+    // Idle arming returns an emit action (continuation as a ReactorAction)
+    // so the host re-enters the loop and the governor can compact on the
+    // continuation's arrival.
+    const idleContinuationArmed = this.compaction.noteIdleTurn(
+      event,
+      baseActions,
+    );
     const compacted = this.compaction.interceptActions(
       event,
       baseActions,
       capabilities,
     );
     if (compacted !== null) return compacted;
+    if (idleContinuationArmed)
+      return [...baseActions, compactionContinuationAction(capabilities)];
 
     // Loop protection takes precedence over workflow/open-task
     // continuation nudges below: those exist to keep a session moving,