Skip to content

Commit 825085f

Browse files
TheGreatAxiosTheGreatAxios
committed
fix(tui): preempt hung continuation hops on interrupt
1 parent 466d3fd commit 825085f

10 files changed

Lines changed: 489 additions & 42 deletions

‎docs/ARCHITECTURE.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -109,7 +109,7 @@ In TUI chat mode there is no completion gate — the session stays open across t
109109
rather than forwarding to a successor. `/clear` and `/new` bump a
110110
delivery generation and call `SessionBridge.clearQueuedDelivery()` so queued
111111
input and deferred recoveries from the previous session cannot enter the new one.
112-
- **Session rotation** — Uses a serial session-operation queue (`createSessionOperationQueue`, not a boolean flag) so rotation, compaction continuation, and `agentProxy.deliver` never race a concurrent rebuild. Each operation chains onto the tail, ensuring in-flight work completes before the agent is torn down.
112+
- **Session rotation** — Uses a serial session-operation queue (`createSessionOperationQueue`, not a boolean flag) so rotation, compaction continuation, and `agentProxy.deliver` never race a concurrent rebuild. Each operation chains onto the tail; continuation hops are preemptible so interrupt and rotation abort a hung deliver instead of waiting forever, while serial rebuild/reload/rotation still finish `close()` before the next op.
113113

114114
### Exec Runner (`src/exec/runner.ts`)
115115

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

Lines changed: 175 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -195,14 +195,16 @@ describe("settleCompactionContinuationHop", () => {
195195
expect(runs).toBe(1);
196196
});
197197

198-
test("continuation enqueued then interrupt rebuild queued lands on the replacement agent", async () => {
199-
const { enqueue, awaitTail } = createSessionOperationQueue();
198+
test("continuation enqueued then interrupt rebuild queued does not auto-deliver to the replacement agent", async () => {
199+
const { enqueue, enqueuePreemptible, abortInFlight, awaitTail } =
200+
createSessionOperationQueue();
200201
const deliveryGeneration = createDeliveryGeneration();
201202
let agent = "outgoing";
202203
const deliveredTo: string[] = [];
204+
const order: string[] = [];
203205

204206
enqueueCompactionContinuationHop({
205-
enqueue,
207+
enqueue: enqueuePreemptible,
206208
captureGeneration: () => deliveryGeneration.capture(),
207209
deliver: async () => {
208210
deliveredTo.push(agent);
@@ -211,17 +213,185 @@ describe("settleCompactionContinuationHop", () => {
211213
onResult: () => undefined,
212214
});
213215

216+
startInterruptRebuild({
217+
deliveryGeneration: {
218+
bump: () => {
219+
order.push("bump");
220+
deliveryGeneration.bump();
221+
},
222+
},
223+
markSendAborted: () => undefined,
224+
abortInFlight: () => {
225+
order.push("abortInFlight");
226+
abortInFlight();
227+
},
228+
enqueue: (op) => {
229+
order.push("enqueue");
230+
return enqueue(op);
231+
},
232+
rebuild: async () => {
233+
agent = "replacement";
234+
},
235+
});
236+
237+
expect(order[0]).toBe("bump");
238+
expect(order.indexOf("abortInFlight")).toBeGreaterThan(
239+
order.indexOf("bump"),
240+
);
241+
expect(order.indexOf("enqueue")).toBeGreaterThan(
242+
order.indexOf("abortInFlight"),
243+
);
244+
245+
await awaitTail();
246+
expect(deliveredTo).not.toContain("outgoing");
247+
expect(deliveredTo).not.toContain("replacement");
248+
});
249+
250+
test("hung hop then startInterruptRebuild runs rebuild without delivering to the outgoing agent", async () => {
251+
const { enqueue, enqueuePreemptible, abortInFlight, awaitTail } =
252+
createSessionOperationQueue();
253+
const deliveryGeneration = createDeliveryGeneration();
254+
let hopStarted = false;
255+
let rebuilt = false;
256+
const deliveredTo: string[] = [];
257+
258+
enqueueCompactionContinuationHop({
259+
enqueue: enqueuePreemptible,
260+
captureGeneration: () => deliveryGeneration.capture(),
261+
deliver: async () => {
262+
hopStarted = true;
263+
await new Promise<void>(() => undefined);
264+
deliveredTo.push("outgoing");
265+
return { status: "accepted" as const };
266+
},
267+
onResult: () => undefined,
268+
});
269+
270+
await Promise.resolve();
271+
await Promise.resolve();
272+
expect(hopStarted).toBe(true);
273+
214274
startInterruptRebuild({
215275
deliveryGeneration,
216276
markSendAborted: () => undefined,
277+
abortInFlight,
217278
enqueue,
218279
rebuild: async () => {
219-
agent = "replacement";
280+
rebuilt = true;
281+
},
282+
});
283+
284+
let timer: ReturnType<typeof setTimeout> | undefined;
285+
try {
286+
await Promise.race([
287+
awaitTail(),
288+
new Promise<never>((_, reject) => {
289+
timer = setTimeout(
290+
() =>
291+
reject(
292+
new Error("rebuild did not run; hung hop parked the tail"),
293+
),
294+
250,
295+
);
296+
}),
297+
]);
298+
} finally {
299+
if (timer !== undefined) clearTimeout(timer);
300+
}
301+
expect(rebuilt).toBe(true);
302+
expect(deliveredTo).toEqual([]);
303+
});
304+
305+
test("stall-abort of a hung continuation does not requeue onto the replacement agent", async () => {
306+
const { enqueue, enqueuePreemptible, abortInFlight, awaitTail } =
307+
createSessionOperationQueue();
308+
const deliveryGeneration = createDeliveryGeneration();
309+
let settleDeliver:
310+
| ((result: {
311+
status: "not-delivered";
312+
reason: "agent-closed";
313+
detail: string;
314+
}) => void)
315+
| undefined;
316+
let deliverCalls = 0;
317+
let hopStarted = false;
318+
let rebuilt = false;
319+
320+
enqueueCompactionContinuationHop({
321+
enqueue: enqueuePreemptible,
322+
captureGeneration: () => deliveryGeneration.capture(),
323+
deliver: async () => {
324+
deliverCalls += 1;
325+
hopStarted = true;
326+
return await new Promise((resolve) => {
327+
settleDeliver = resolve;
328+
});
329+
},
330+
onResult: () => undefined,
331+
});
332+
333+
await Promise.resolve();
334+
await Promise.resolve();
335+
expect(hopStarted).toBe(true);
336+
337+
deliveryGeneration.bump();
338+
abortInFlight();
339+
enqueue(async () => {
340+
rebuilt = true;
341+
});
342+
343+
let timer: ReturnType<typeof setTimeout> | undefined;
344+
try {
345+
await Promise.race([
346+
awaitTail(),
347+
new Promise<never>((_, reject) => {
348+
timer = setTimeout(
349+
() => reject(new Error("stall abort left the tail parked")),
350+
250,
351+
);
352+
}),
353+
]);
354+
} finally {
355+
if (timer !== undefined) clearTimeout(timer);
356+
}
357+
358+
expect(rebuilt).toBe(true);
359+
settleDeliver?.({
360+
status: "not-delivered",
361+
reason: "agent-closed",
362+
detail: "agent is closed",
363+
});
364+
await Promise.resolve();
365+
await Promise.resolve();
366+
expect(deliverCalls).toBe(1);
367+
});
368+
369+
test("two distinct continuation seqs both enqueue", async () => {
370+
const { enqueuePreemptible, awaitTail } = createSessionOperationQueue();
371+
const deliveryGeneration = createDeliveryGeneration();
372+
const delivered: number[] = [];
373+
374+
enqueueCompactionContinuationHop({
375+
enqueue: enqueuePreemptible,
376+
captureGeneration: () => deliveryGeneration.capture(),
377+
deliver: async () => {
378+
delivered.push(1);
379+
return { status: "accepted" as const };
220380
},
381+
onResult: () => undefined,
382+
});
383+
enqueueCompactionContinuationHop({
384+
enqueue: enqueuePreemptible,
385+
captureGeneration: () => deliveryGeneration.capture(),
386+
deliver: async () => {
387+
delivered.push(2);
388+
return { status: "accepted" as const };
389+
},
390+
onResult: () => undefined,
221391
});
222392

223393
await awaitTail();
224-
expect(deliveredTo).toEqual(["replacement"]);
394+
expect(delivered).toEqual([1, 2]);
225395
});
226396
});
227397

‎src/tui/delivery-queue.ts‎

Lines changed: 97 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -112,9 +112,9 @@ function compactionContinuationIsSuperseded(
112112

113113
/**
114114
* Compact continuation is consume-once at the stream gate. Interrupt rebuild
115-
* bumps generation and enqueues a rebuild behind this hop, so a stale hop
116-
* must re-queue rather than retry liveAgent in this op. A closed hop still
117-
* retries once here.
115+
* bumps generation and aborts an in-flight hop so a hung deliver cannot park
116+
* the serial tail. An interrupt-stale or abort-abandoned hop is fail-closed:
117+
* it must not re-queue onto the replacement agent.
118118
*/
119119
export async function settleCompactionContinuationHop(options: {
120120
stillCurrent: () => boolean;
@@ -134,36 +134,38 @@ export async function settleCompactionContinuationHop(options: {
134134
options.onSuperseded();
135135
return first;
136136
}
137+
if (!options.stillCurrent()) {
138+
return {
139+
status: "not-delivered",
140+
reason: "superseded",
141+
detail: "session identity changed before delivery",
142+
};
143+
}
137144
if (first.status === "not-delivered" && first.reason === "agent-closed") {
138145
return options.deliver();
139146
}
140147
return first;
141148
}
142149

143150
/**
144-
* Enqueue a compact-continue hop. If interrupt already bumped generation and
145-
* queued a rebuild behind this hop, the continue is scheduled again after
146-
* that rebuild instead of retrying liveAgent in the same op.
151+
* Enqueue a compact-continue hop on the preemptible lane. Interrupt and stall
152+
* abort drop the hop instead of re-queueing it onto a replacement agent.
147153
*/
148154
export function enqueueCompactionContinuationHop(options: {
149155
enqueue: (op: () => Promise<void>) => unknown;
150156
captureGeneration: () => () => boolean;
151157
deliver: () => Promise<AgentDeliveryResult>;
152158
onResult: (result: AgentDeliveryResult) => void;
153159
}): void {
154-
const schedule = (): void => {
155-
const stillCurrent = options.captureGeneration();
156-
void options.enqueue(async () => {
157-
const result = await settleCompactionContinuationHop({
158-
stillCurrent,
159-
deliver: options.deliver,
160-
onSuperseded: schedule,
161-
});
162-
if (compactionContinuationIsSuperseded(result)) return;
163-
options.onResult(result);
160+
const stillCurrent = options.captureGeneration();
161+
void options.enqueue(async () => {
162+
const result = await settleCompactionContinuationHop({
163+
stillCurrent,
164+
deliver: options.deliver,
165+
onSuperseded: () => undefined,
164166
});
165-
};
166-
schedule();
167+
options.onResult(result);
168+
});
167169
}
168170

169171
/** Operator-facing copy for a settled delivery that did not accept. */
@@ -408,24 +410,97 @@ export function drainSteersOnly(state: SessionQueueState): {
408410

409411
// Serial promise chain for session-scoped operations (reload, interrupt, deliver).
410412
// Each task runs after the previous one settles; failures do not block the tail.
413+
// Continuation hops enqueue as preemptible so interrupt/stall-abort can settle
414+
// the wrapper without waiting forever on Agent.deliver. Rebuild/reload/rotation
415+
// stay serial so close() can finish.
411416

412417
export interface SessionOperationQueue {
413418
/** Enqueue an async operation; returns a promise for this operation's settlement. */
414419
enqueue: (op: () => Promise<void>) => Promise<void>;
420+
/**
421+
* Same tail as `enqueue`, marked preemptible. `abortInFlight` races the
422+
* current op against a captured abort signal so a hung continuation cannot
423+
* park rebuild, send, or quit.
424+
*/
425+
enqueuePreemptible: (op: () => Promise<void>) => Promise<void>;
426+
/**
427+
* Abort the in-flight op only when it is preemptible, then mint a fresh
428+
* controller so a rebuild enqueued after this call is not cancelled.
429+
*/
430+
abortInFlight: () => void;
415431
/** Await the tail of the queue (all prior operations finished or failed). */
416432
awaitTail: () => Promise<void>;
417433
}
418434

435+
type SessionOpKind = "serial" | "preemptible";
436+
437+
function whenAborted(signal: AbortSignal): {
438+
promise: Promise<void>;
439+
dispose: () => void;
440+
} {
441+
let onAbort: (() => void) | undefined;
442+
const promise = new Promise<void>((resolve) => {
443+
if (signal.aborted) {
444+
resolve();
445+
return;
446+
}
447+
onAbort = () => resolve();
448+
signal.addEventListener("abort", onAbort, { once: true });
449+
});
450+
return {
451+
promise,
452+
dispose: () => {
453+
if (onAbort !== undefined) signal.removeEventListener("abort", onAbort);
454+
},
455+
};
456+
}
457+
419458
export function createSessionOperationQueue(): SessionOperationQueue {
420459
let tail: Promise<void> = Promise.resolve();
421-
422-
const enqueue = (op: () => Promise<void>): Promise<void> => {
423-
tail = tail.then(op, op);
460+
let abortController = new AbortController();
461+
let currentKind: SessionOpKind | null = null;
462+
463+
const enqueueKind = (
464+
kind: SessionOpKind,
465+
op: () => Promise<void>,
466+
): Promise<void> => {
467+
const execute = async (): Promise<void> => {
468+
currentKind = kind;
469+
try {
470+
if (kind !== "preemptible") {
471+
await op();
472+
return;
473+
}
474+
const signal = abortController.signal;
475+
const abort = whenAborted(signal);
476+
const running = op();
477+
try {
478+
const outcome = await Promise.race([
479+
running.then(() => "ran" as const),
480+
abort.promise.then(() => "aborted" as const),
481+
]);
482+
if (outcome === "aborted") {
483+
void running.catch(() => undefined);
484+
}
485+
} finally {
486+
abort.dispose();
487+
}
488+
} finally {
489+
currentKind = null;
490+
}
491+
};
492+
tail = tail.then(execute, execute);
424493
return tail;
425494
};
426495

427496
return {
428-
enqueue,
497+
enqueue: (op) => enqueueKind("serial", op),
498+
enqueuePreemptible: (op) => enqueueKind("preemptible", op),
499+
abortInFlight: () => {
500+
if (currentKind !== "preemptible") return;
501+
abortController.abort();
502+
abortController = new AbortController();
503+
},
429504
awaitTail: () => tail.catch(() => undefined),
430505
};
431506
}

0 commit comments

Comments
 (0)