Skip to content

Commit 67b7280

Browse files
committed
fix(tui): re-issue compact continuation after a superseded hop
1 parent 445886d commit 67b7280

8 files changed

Lines changed: 183 additions & 31 deletions

File tree

‎src/subagent/ask-director.test.ts‎

Lines changed: 30 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,11 +5,11 @@ import {
55
ASK_DIRECTOR_MAX_QUESTIONS,
66
commitAskDirector,
77
createAskDirectorState,
8+
createDeferredContinuation,
89
evaluateAskDirector,
910
handleAskDirector,
1011
releaseAskDirector,
1112
resetAskDirectorTurn,
12-
skipStallContinuationWhileAskPending,
1313
} from "./ask-director.js";
1414

1515
describe("evaluateAskDirector", () => {
@@ -207,18 +207,43 @@ describe("evaluateAskDirector", () => {
207207
describe("requestContinuation stall skip", () => {
208208
test("is a no-op while askDirectorState.pending", () => {
209209
const state = createAskDirectorState();
210+
const latch = createDeferredContinuation();
210211
let delivered = 0;
211-
skipStallContinuationWhileAskPending(state, () => {
212+
const deliver = () => {
212213
delivered += 1;
213-
});
214+
};
215+
latch.request(state, deliver);
214216
expect(delivered).toBe(1);
215217

216218
expect(evaluateAskDirector({ question: "which file?", state }).ok).toBe(
217219
true,
218220
);
219-
skipStallContinuationWhileAskPending(state, () => {
221+
latch.request(state, deliver);
222+
expect(delivered).toBe(1);
223+
});
224+
225+
test("ask-pending skip defers compact continuation until the ask releases", () => {
226+
const state = createAskDirectorState();
227+
const latch = createDeferredContinuation();
228+
let delivered = 0;
229+
const deliver = () => {
220230
delivered += 1;
221-
});
231+
};
232+
233+
expect(evaluateAskDirector({ question: "which file?", state }).ok).toBe(
234+
true,
235+
);
236+
latch.request(state, deliver);
237+
expect(delivered).toBe(0);
238+
239+
latch.flush(state, deliver);
240+
expect(delivered).toBe(0);
241+
242+
releaseAskDirector(state);
243+
latch.flush(state, deliver);
244+
expect(delivered).toBe(1);
245+
246+
latch.flush(state, deliver);
222247
expect(delivered).toBe(1);
223248
});
224249
});

‎src/subagent/ask-director.ts‎

Lines changed: 23 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -89,15 +89,30 @@ export function resetAskDirectorTurn(state: AskDirectorState): void {
8989
}
9090

9191
/**
92-
* Waiting on the spawning director is work, not silence. A stall ping
93-
* would otherwise salvage a blocked ask_director as stalled.
92+
* Compact continuation shares the stall ping channel. Skipping while
93+
* ask_director is parked must not drop the only continue: remember the skip
94+
* and flush after the ask releases.
9495
*/
95-
export function skipStallContinuationWhileAskPending(
96-
state: AskDirectorState,
97-
deliver: () => void,
98-
): void {
99-
if (state.pending) return;
100-
deliver();
96+
export function createDeferredContinuation(): {
97+
request: (state: AskDirectorState, deliver: () => void) => void;
98+
flush: (state: AskDirectorState, deliver: () => void) => void;
99+
} {
100+
let deferred = false;
101+
return {
102+
request(state, deliver) {
103+
if (state.pending) {
104+
deferred = true;
105+
return;
106+
}
107+
deferred = false;
108+
deliver();
109+
},
110+
flush(state, deliver) {
111+
if (!deferred || state.pending) return;
112+
deferred = false;
113+
deliver();
114+
},
115+
};
101116
}
102117

103118
export async function handleAskDirector(args: {

‎src/subagent/run.ts‎

Lines changed: 21 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -152,9 +152,9 @@ import {
152152
ASK_DIRECTOR_MAX_BYTES,
153153
ASK_DIRECTOR_MAX_QUESTIONS,
154154
createAskDirectorState,
155+
createDeferredContinuation,
155156
handleAskDirector,
156157
resetAskDirectorTurn,
157-
skipStallContinuationWhileAskPending,
158158
} from "./ask-director.js";
159159
import {
160160
abortError,
@@ -625,6 +625,8 @@ async function runSubAgentInner(
625625
let turnToken = params.tier === "leaf" ? generateSessionId() : undefined;
626626
const submitResultState = createSubmitResultState();
627627
const askDirectorState = createAskDirectorState();
628+
const compactContinue = createDeferredContinuation();
629+
let deliverCompactContinue = (): void => undefined;
628630
const spawnRegistry = createSubAgentSpawnRegistryPlugin();
629631
// Assigned inside the try once the agent handle exists; before that (or after
630632
// close) a completion is dropped, matching the continuation contract.
@@ -862,12 +864,16 @@ async function runSubAgentInner(
862864
"Record the question under Blockers and finish with the markdown report envelope."
863865
);
864866
}
865-
return handleAskDirector({
866-
question: rawArgs.question,
867-
state: askDirectorState,
868-
port,
869-
signal,
870-
});
867+
try {
868+
return await handleAskDirector({
869+
question: rawArgs.question,
870+
state: askDirectorState,
871+
port,
872+
signal,
873+
});
874+
} finally {
875+
compactContinue.flush(askDirectorState, deliverCompactContinue);
876+
}
871877
},
872878
}),
873879
];
@@ -1038,13 +1044,14 @@ async function runSubAgentInner(
10381044
ReturnType<typeof createAgentWithLiveToolDispatch>
10391045
> | null = null;
10401046
const requestContinuation = (): void => {
1041-
skipStallContinuationWhileAskPending(askDirectorState, () => {
1042-
try {
1043-
agentHandle?.deliver(buildCompactionContinuationMessage());
1044-
} catch {
1045-
// Agent may be closing; a dropped continuation is harmless.
1046-
}
1047-
});
1047+
compactContinue.request(askDirectorState, deliverCompactContinue);
1048+
};
1049+
deliverCompactContinue = (): void => {
1050+
try {
1051+
agentHandle?.deliver(buildCompactionContinuationMessage());
1052+
} catch {
1053+
// Agent may be closing; a dropped continuation is harmless.
1054+
}
10481055
};
10491056

10501057
// modelFamilyPolicy is resolved above at the skill mount so the

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

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import {
44
deliverAgentMessage,
55
deliveryResultNotice,
66
runGenerationGuardedDeliver,
7+
settleCompactionContinuationHop,
78
} from "./delivery-queue.js";
89

910
describe("deliverAgentMessage", () => {
@@ -125,6 +126,57 @@ describe("runGenerationGuardedDeliver", () => {
125126
});
126127
});
127128

129+
describe("settleCompactionContinuationHop", () => {
130+
test("a superseded hop re-issues the continue against the live agent", async () => {
131+
let generation = 1;
132+
const stillCurrent = () => generation === 1;
133+
let delivered = 0;
134+
generation = 2;
135+
const result = await settleCompactionContinuationHop({
136+
stillCurrent,
137+
deliver: async () => {
138+
delivered += 1;
139+
return { status: "accepted" as const };
140+
},
141+
});
142+
expect(result).toEqual({ status: "accepted" });
143+
expect(delivered).toBe(1);
144+
});
145+
146+
test("a closed hop re-issues the continue once", async () => {
147+
let attempts = 0;
148+
const result = await settleCompactionContinuationHop({
149+
stillCurrent: () => true,
150+
deliver: async () => {
151+
attempts += 1;
152+
if (attempts === 1) {
153+
return {
154+
status: "not-delivered" as const,
155+
reason: "agent-closed" as const,
156+
detail: "agent is closed",
157+
};
158+
}
159+
return { status: "accepted" as const };
160+
},
161+
});
162+
expect(result).toEqual({ status: "accepted" });
163+
expect(attempts).toBe(2);
164+
});
165+
166+
test("a current hop delivers once and does not re-issue", async () => {
167+
let runs = 0;
168+
const result = await settleCompactionContinuationHop({
169+
stillCurrent: () => true,
170+
deliver: async () => {
171+
runs += 1;
172+
return { status: "accepted" as const };
173+
},
174+
});
175+
expect(result).toEqual({ status: "accepted" });
176+
expect(runs).toBe(1);
177+
});
178+
});
179+
128180
describe("deliveryResultNotice", () => {
129181
test("closed restored and deferred copy is actionable", () => {
130182
const closed = {

‎src/tui/delivery-queue.ts‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,39 @@ export async function runGenerationGuardedDeliver(options: {
104104
return options.run();
105105
}
106106

107+
function compactionContinuationShouldReissue(
108+
result: AgentDeliveryResult,
109+
): boolean {
110+
return (
111+
result.status === "not-delivered" &&
112+
(result.reason === "superseded" || result.reason === "agent-closed")
113+
);
114+
}
115+
116+
/**
117+
* Compact continuation is consume-once at the stream gate, so a superseded or
118+
* closed hop must re-issue in the same serial op — otherwise the only continue
119+
* is dropped and the reactor idles.
120+
*/
121+
export async function settleCompactionContinuationHop(options: {
122+
stillCurrent: () => boolean;
123+
deliver: () => Promise<AgentDeliveryResult>;
124+
}): Promise<AgentDeliveryResult> {
125+
const first = await runGenerationGuardedDeliver({
126+
stillCurrent: options.stillCurrent,
127+
onStale: () => ({
128+
status: "not-delivered",
129+
reason: "superseded",
130+
detail: "session identity changed before delivery",
131+
}),
132+
run: options.deliver,
133+
});
134+
if (compactionContinuationShouldReissue(first)) {
135+
return options.deliver();
136+
}
137+
return first;
138+
}
139+
107140
/** Operator-facing copy for a settled delivery that did not accept. */
108141
export function deliveryResultNotice(
109142
result: Exclude<AgentDeliveryResult, { status: "accepted" }>,

‎src/tui/runner/exit.ts‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -325,11 +325,12 @@ export async function createRunLifecycle(
325325
// the same message the old requestContinuation closure delivered,
326326
// through the serial op queue like every other deliver. Each emission
327327
// is answered once: a replayed duplicate of an already-answered
328-
// emission is ignored instead of re-delivered.
328+
// emission is ignored instead of re-delivered. A superseded or closed
329+
// hop re-issues in the same op so consume-once cannot drop the only
330+
// continue.
329331
if (continuationGate.shouldDeliver(event.seq)) {
330-
const targetAgent = liveAgent(state);
331-
state.enqueueAgentDeliver?.(() =>
332-
targetAgent.deliver(buildCompactionContinuationMessage()),
332+
state.enqueueCompactionContinuation?.(() =>
333+
liveAgent(state).deliver(buildCompactionContinuationMessage()),
333334
);
334335
}
335336
}

‎src/tui/runner/session.ts‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,7 @@ import {
8282
deliverAgentMessage,
8383
deliveryResultNotice,
8484
runGenerationGuardedDeliver,
85+
settleCompactionContinuationHop,
8586
type AgentDeliveryResult,
8687
} from "../delivery-queue.js";
8788
import { createCorrelationAcceptance } from "../correlation-acceptance.js";
@@ -568,6 +569,23 @@ export async function assembleTUISession(
568569
state.systemNotice?.(deliveryResultNotice(result));
569570
});
570571
};
572+
state.enqueueCompactionContinuation = (
573+
deliverToLiveAgent: () => void,
574+
): void => {
575+
const stillCurrent = deliveryGeneration.capture();
576+
void sessionOps.enqueue(async () => {
577+
const result = await settleCompactionContinuationHop({
578+
stillCurrent,
579+
deliver: () =>
580+
deliverAgentMessage({
581+
getFatalBuildError: () => state.fatalBuildError,
582+
deliverToLiveAgent,
583+
}),
584+
});
585+
if (result.status === "accepted") return;
586+
state.systemNotice?.(deliveryResultNotice(result));
587+
});
588+
};
571589

572590
const buildSessionSources = (): LiveSessionSources =>
573591
resolveLiveSessionSources(state.config, state.sessionId);

‎src/tui/runner/state.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -257,6 +257,7 @@ export interface RunnerState {
257257
deliverToLiveAgent: () => void,
258258
onSettle?: (result: AgentDeliveryResult) => void,
259259
) => void;
260+
enqueueCompactionContinuation?: (deliverToLiveAgent: () => void) => void;
260261
reloadIfIdle?: () => void;
261262
systemNotice?: (text: string) => void;
262263
currentAttemptIdentity?: () => InferenceAttemptIdentity;

0 commit comments

Comments
 (0)