fix(tui): land compact continue on the replacement agent

Author: TheGreatAxios

docs/ARCHITECTURE.md

@@ -192,7 +192,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 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.
+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. A hop superseded because interrupt already bumped generation and enqueued a rebuild is re-queued onto the replacement agent rather than consumed on the outgoing liveAgent. 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/subagent/ask-director.test.ts

@@ -5,11 +5,11 @@ import {
   ASK_DIRECTOR_MAX_QUESTIONS,
   commitAskDirector,
   createAskDirectorState,
+  createDeferredContinuation,
   evaluateAskDirector,
   handleAskDirector,
   releaseAskDirector,
   resetAskDirectorTurn,
-  skipStallContinuationWhileAskPending,
 } from "./ask-director.js";
 
 describe("evaluateAskDirector", () => {
@@ -207,18 +207,43 @@ describe("evaluateAskDirector", () => {
 describe("requestContinuation stall skip", () => {
   test("is a no-op while askDirectorState.pending", () => {
     const state = createAskDirectorState();
+    const latch = createDeferredContinuation();
     let delivered = 0;
-    skipStallContinuationWhileAskPending(state, () => {
+    const deliver = () => {
       delivered += 1;
-    });
+    };
+    latch.request(state, deliver);
     expect(delivered).toBe(1);
 
     expect(evaluateAskDirector({ question: "which file?", state }).ok).toBe(
       true,
     );
-    skipStallContinuationWhileAskPending(state, () => {
+    latch.request(state, deliver);
+    expect(delivered).toBe(1);
+  });
+
+  test("ask-pending skip defers compact continuation until the ask releases", () => {
+    const state = createAskDirectorState();
+    const latch = createDeferredContinuation();
+    let delivered = 0;
+    const deliver = () => {
       delivered += 1;
-    });
+    };
+
+    expect(evaluateAskDirector({ question: "which file?", state }).ok).toBe(
+      true,
+    );
+    latch.request(state, deliver);
+    expect(delivered).toBe(0);
+
+    latch.flush(state, deliver);
+    expect(delivered).toBe(0);
+
+    releaseAskDirector(state);
+    latch.flush(state, deliver);
+    expect(delivered).toBe(1);
+
+    latch.flush(state, deliver);
     expect(delivered).toBe(1);
   });
 });

src/subagent/ask-director.ts

@@ -89,15 +89,30 @@ export function resetAskDirectorTurn(state: AskDirectorState): void {
 }
 
 /**
- * Waiting on the spawning director is work, not silence. A stall ping
- * would otherwise salvage a blocked ask_director as stalled.
+ * Compact continuation shares the stall ping channel. Skipping while
+ * ask_director is parked must not drop the only continue: remember the skip
+ * and flush after the ask releases.
  */
-export function skipStallContinuationWhileAskPending(
-  state: AskDirectorState,
-  deliver: () => void,
-): void {
-  if (state.pending) return;
-  deliver();
+export function createDeferredContinuation(): {
+  request: (state: AskDirectorState, deliver: () => void) => void;
+  flush: (state: AskDirectorState, deliver: () => void) => void;
+} {
+  let deferred = false;
+  return {
+    request(state, deliver) {
+      if (state.pending) {
+        deferred = true;
+        return;
+      }
+      deferred = false;
+      deliver();
+    },
+    flush(state, deliver) {
+      if (!deferred || state.pending) return;
+      deferred = false;
+      deliver();
+    },
+  };
 }
 
 export async function handleAskDirector(args: {

src/subagent/run.ts

@@ -152,9 +152,9 @@ import {
   ASK_DIRECTOR_MAX_BYTES,
   ASK_DIRECTOR_MAX_QUESTIONS,
   createAskDirectorState,
+  createDeferredContinuation,
   handleAskDirector,
   resetAskDirectorTurn,
-  skipStallContinuationWhileAskPending,
 } from "./ask-director.js";
 import {
   abortError,
@@ -625,6 +625,8 @@ async function runSubAgentInner(
   let turnToken = params.tier === "leaf" ? generateSessionId() : undefined;
   const submitResultState = createSubmitResultState();
   const askDirectorState = createAskDirectorState();
+  const compactContinue = createDeferredContinuation();
+  let deliverCompactContinue = (): void => undefined;
   const spawnRegistry = createSubAgentSpawnRegistryPlugin();
   // Assigned inside the try once the agent handle exists; before that (or after
   // close) a completion is dropped, matching the continuation contract.
@@ -862,12 +864,16 @@ async function runSubAgentInner(
                 "Record the question under Blockers and finish with the markdown report envelope."
               );
             }
-            return handleAskDirector({
-              question: rawArgs.question,
-              state: askDirectorState,
-              port,
-              signal,
-            });
+            try {
+              return await handleAskDirector({
+                question: rawArgs.question,
+                state: askDirectorState,
+                port,
+                signal,
+              });
+            } finally {
+              compactContinue.flush(askDirectorState, deliverCompactContinue);
+            }
           },
         }),
       ];
@@ -1038,13 +1044,14 @@ async function runSubAgentInner(
       ReturnType<typeof createAgentWithLiveToolDispatch>
     > | null = null;
     const requestContinuation = (): void => {
-      skipStallContinuationWhileAskPending(askDirectorState, () => {
-        try {
-          agentHandle?.deliver(buildCompactionContinuationMessage());
-        } catch {
-          // Agent may be closing; a dropped continuation is harmless.
-        }
-      });
+      compactContinue.request(askDirectorState, deliverCompactContinue);
+    };
+    deliverCompactContinue = (): void => {
+      try {
+        agentHandle?.deliver(buildCompactionContinuationMessage());
+      } catch {
+        // Agent may be closing; a dropped continuation is harmless.
+      }
     };
 
     // modelFamilyPolicy is resolved above at the skill mount so the

src/tui/deliver-agent-message.test.ts

@@ -1,10 +1,15 @@
 import { describe, expect, test } from "bun:test";
 import { AgentClosedError } from "@intx/agent";
 import {
+  createDeliveryGeneration,
+  createSessionOperationQueue,
   deliverAgentMessage,
   deliveryResultNotice,
+  enqueueCompactionContinuationHop,
   runGenerationGuardedDeliver,
+  settleCompactionContinuationHop,
 } from "./delivery-queue.js";
+import { startInterruptRebuild } from "./runner/exit.js";
 
 describe("deliverAgentMessage", () => {
   test("reports session-unavailable without calling deliver when rebuild failed", async () => {
@@ -125,6 +130,101 @@ describe("runGenerationGuardedDeliver", () => {
   });
 });
 
+describe("settleCompactionContinuationHop", () => {
+  test("a superseded hop requeues instead of delivering to the outgoing agent", async () => {
+    let generation = 1;
+    const stillCurrent = () => generation === 1;
+    let delivered = 0;
+    let requeued = 0;
+    generation = 2;
+    const result = await settleCompactionContinuationHop({
+      stillCurrent,
+      deliver: async () => {
+        delivered += 1;
+        return { status: "accepted" as const };
+      },
+      onSuperseded: () => {
+        requeued += 1;
+      },
+    });
+    expect(result).toEqual({
+      status: "not-delivered",
+      reason: "superseded",
+      detail: "session identity changed before delivery",
+    });
+    expect(delivered).toBe(0);
+    expect(requeued).toBe(1);
+  });
+
+  test("a closed hop re-issues the continue once", async () => {
+    let attempts = 0;
+    const result = await settleCompactionContinuationHop({
+      stillCurrent: () => true,
+      deliver: async () => {
+        attempts += 1;
+        if (attempts === 1) {
+          return {
+            status: "not-delivered" as const,
+            reason: "agent-closed" as const,
+            detail: "agent is closed",
+          };
+        }
+        return { status: "accepted" as const };
+      },
+      onSuperseded: () => {
+        throw new Error("closed hop must not requeue as superseded");
+      },
+    });
+    expect(result).toEqual({ status: "accepted" });
+    expect(attempts).toBe(2);
+  });
+
+  test("a current hop delivers once and does not re-issue", async () => {
+    let runs = 0;
+    const result = await settleCompactionContinuationHop({
+      stillCurrent: () => true,
+      deliver: async () => {
+        runs += 1;
+        return { status: "accepted" as const };
+      },
+      onSuperseded: () => {
+        throw new Error("current hop must not requeue");
+      },
+    });
+    expect(result).toEqual({ status: "accepted" });
+    expect(runs).toBe(1);
+  });
+
+  test("continuation enqueued then interrupt rebuild queued lands on the replacement agent", async () => {
+    const { enqueue, awaitTail } = createSessionOperationQueue();
+    const deliveryGeneration = createDeliveryGeneration();
+    let agent = "outgoing";
+    const deliveredTo: string[] = [];
+
+    enqueueCompactionContinuationHop({
+      enqueue,
+      captureGeneration: () => deliveryGeneration.capture(),
+      deliver: async () => {
+        deliveredTo.push(agent);
+        return { status: "accepted" as const };
+      },
+      onResult: () => undefined,
+    });
+
+    startInterruptRebuild({
+      deliveryGeneration,
+      markSendAborted: () => undefined,
+      enqueue,
+      rebuild: async () => {
+        agent = "replacement";
+      },
+    });
+
+    await awaitTail();
+    expect(deliveredTo).toEqual(["replacement"]);
+  });
+});
+
 describe("deliveryResultNotice", () => {
   test("closed restored and deferred copy is actionable", () => {
     const closed = {

src/tui/delivery-queue.ts

@@ -104,6 +104,68 @@ export async function runGenerationGuardedDeliver(options: {
   return options.run();
 }
 
+function compactionContinuationIsSuperseded(
+  result: AgentDeliveryResult,
+): boolean {
+  return result.status === "not-delivered" && result.reason === "superseded";
+}
+
+/**
+ * Compact continuation is consume-once at the stream gate. Interrupt rebuild
+ * bumps generation and enqueues a rebuild behind this hop, so a stale hop
+ * must re-queue rather than retry liveAgent in this op. A closed hop still
+ * retries once here.
+ */
+export async function settleCompactionContinuationHop(options: {
+  stillCurrent: () => boolean;
+  deliver: () => Promise<AgentDeliveryResult>;
+  onSuperseded: () => void;
+}): Promise<AgentDeliveryResult> {
+  const first = await runGenerationGuardedDeliver({
+    stillCurrent: options.stillCurrent,
+    onStale: () => ({
+      status: "not-delivered",
+      reason: "superseded",
+      detail: "session identity changed before delivery",
+    }),
+    run: options.deliver,
+  });
+  if (compactionContinuationIsSuperseded(first)) {
+    options.onSuperseded();
+    return first;
+  }
+  if (first.status === "not-delivered" && first.reason === "agent-closed") {
+    return options.deliver();
+  }
+  return first;
+}
+
+/**
+ * Enqueue a compact-continue hop. If interrupt already bumped generation and
+ * queued a rebuild behind this hop, the continue is scheduled again after
+ * that rebuild instead of retrying liveAgent in the same op.
+ */
+export function enqueueCompactionContinuationHop(options: {
+  enqueue: (op: () => Promise<void>) => unknown;
+  captureGeneration: () => () => boolean;
+  deliver: () => Promise<AgentDeliveryResult>;
+  onResult: (result: AgentDeliveryResult) => void;
+}): void {
+  const schedule = (): void => {
+    const stillCurrent = options.captureGeneration();
+    void options.enqueue(async () => {
+      const result = await settleCompactionContinuationHop({
+        stillCurrent,
+        deliver: options.deliver,
+        onSuperseded: schedule,
+      });
+      if (compactionContinuationIsSuperseded(result)) return;
+      options.onResult(result);
+    });
+  };
+  schedule();
+}
+
 /** Operator-facing copy for a settled delivery that did not accept. */
 export function deliveryResultNotice(
   result: Exclude<AgentDeliveryResult, { status: "accepted" }>,

src/tui/runner/exit.ts

@@ -325,11 +325,13 @@ export async function createRunLifecycle(
       // the same message the old requestContinuation closure delivered,
       // through the serial op queue like every other deliver. Each emission
       // is answered once: a replayed duplicate of an already-answered
-      // emission is ignored instead of re-delivered.
+      // emission is ignored instead of re-delivered. A hop superseded by
+      // interrupt rebuild (generation bump + rebuild already queued) is
+      // re-queued onto the replacement agent so consume-once cannot land on
+      // the outgoing liveAgent.
       if (continuationGate.shouldDeliver(event.seq)) {
-        const targetAgent = liveAgent(state);
-        state.enqueueAgentDeliver?.(() =>
-          targetAgent.deliver(buildCompactionContinuationMessage()),
+        state.enqueueCompactionContinuation?.(() =>
+          liveAgent(state).deliver(buildCompactionContinuationMessage()),
         );
       }
     }