Skip to content

Commit 35ce63c

Browse files
committed
fix(tui): let occupancy win after a stalled ask-wake abort
A stalled armed wake was re-surfacing before mailbox mail, so occupancy lost the next turn. Share one interrupt path: interrupt the hung inference, then mail, then wake only if still idle. Expiring asks must abort that wake, not merely disarm it.
1 parent d847880 commit 35ce63c

2 files changed

Lines changed: 121 additions & 33 deletions

File tree

‎src/tui/runner/wiring.stall-bound.test.ts‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -148,6 +148,17 @@ describe("stall-bound primary turn (CL-8016)", () => {
148148
expect(bridge.turnMarkers().map((marker) => marker.path)).toContain(
149149
"stall-abort:awaiting-first-token",
150150
);
151+
// ... the hung inference was interrupted before any new deliver ...
152+
const interruptAt = port.calls.findIndex((call) => call.op === "interrupt");
153+
expect(interruptAt).toBeGreaterThanOrEqual(0);
154+
const wakeAfterInterrupt = port.calls
155+
.slice(interruptAt + 1)
156+
.filter(
157+
(call): call is Extract<typeof call, { op: "deliver" }> =>
158+
call.op === "deliver" &&
159+
call.item.text.includes(ASK_DIRECTOR_WAKE_PREFIX),
160+
);
161+
expect(wakeAfterInterrupt).toHaveLength(0);
151162
// ... the queued operator message reached the port ...
152163
expect(
153164
port.calls.some(
@@ -173,6 +184,72 @@ describe("stall-bound primary turn (CL-8016)", () => {
173184
});
174185
});
175186

187+
test("stall monitor abort of an armed wake lets occupancy take the next turn", async () => {
188+
await withTestRenderer(async (h) => {
189+
let nowMs = 4_000_000;
190+
let monitorTick: (() => void) | undefined;
191+
const store = createSubAgentSessionStore({ now: () => nowMs });
192+
parkWorker(store, "a");
193+
const shell = createAppShell(h.renderer, {
194+
terminal: { columns: 80, rows: 24 },
195+
wireKeys: false,
196+
});
197+
const port = createRecordingPort();
198+
const bridge = attachSessionBridge(shell, port, {
199+
now: () => nowMs,
200+
stallTimeoutMs: STALL_TIMEOUT_MS,
201+
schedule: (fn) => {
202+
monitorTick = fn;
203+
return () => {
204+
monitorTick = undefined;
205+
};
206+
},
207+
});
208+
try {
209+
bridge.handle({
210+
type: "agent-ask",
211+
asks: pendingAskSnapshot(store.list(), (id) => {
212+
const ask = store.peekAsk(id);
213+
return ask === undefined ? undefined : ask;
214+
}),
215+
});
216+
expect(bridge.turn.isProcessing).toBe(true);
217+
expect(wakeDeliveries(port)).toHaveLength(1);
218+
219+
let mailDrives = 0;
220+
bridge.setMailboxMailDriver(() => {
221+
if (mailDrives > 0) return false;
222+
mailDrives += 1;
223+
bridge.beginSystemContinuation("mailbox occupancy");
224+
return true;
225+
});
226+
227+
nowMs += STALL_TIMEOUT_MS + 500;
228+
expect(monitorTick).toBeDefined();
229+
monitorTick?.();
230+
231+
// Production abort is the #1095 monitor tick → doInterrupt, not the
232+
// 5s fleet poll. Occupancy must win that next turn; a re-surface wake
233+
// must not start processing first.
234+
const interruptAt = port.calls.findIndex((call) => call.op === "interrupt");
235+
expect(interruptAt).toBeGreaterThanOrEqual(0);
236+
expect(
237+
port.calls.slice(interruptAt + 1).some(
238+
(call) =>
239+
call.op === "deliver" &&
240+
call.item.text.includes(ASK_DIRECTOR_WAKE_PREFIX),
241+
),
242+
).toBe(false);
243+
expect(mailDrives).toBe(1);
244+
expect(wakeDeliveries(port)).toHaveLength(1);
245+
expect(bridge.turn.isProcessing).toBe(true);
246+
expect(bridge.turn.status).toBe("running");
247+
} finally {
248+
bridge.dispose();
249+
}
250+
});
251+
});
252+
176253
test("two parked asks settle exactly once via the ask deadline", async () => {
177254
await withTestRenderer(async (h) => {
178255
let nowMs = 2_000_000;
@@ -238,6 +315,9 @@ describe("stall-bound primary turn (CL-8016)", () => {
238315
expect(store.resolveAsk(worker.sessionId, "late")).toBe(false);
239316
expect(store.hasPendingAsk(worker.sessionId)).toBe(false);
240317
}
318+
// Expiring the asks must also end the silent wake — disarm without
319+
// abort would leave isProcessing hung with nothing left to re-surface.
320+
expect(bridge.turn.isProcessing).toBe(false);
241321
} finally {
242322
bridge.dispose();
243323
}

‎src/tui/runtime-bridge.ts‎

Lines changed: 41 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -311,9 +311,10 @@ export interface SessionBridge {
311311
* Stall bound for a silent ask-wake primary turn (CL-8016). If a wake was
312312
* actually sent (`askWakeTurnArmed`) and `shouldAbortForStall` says the
313313
* turn is silent past the bound (including awaiting-first-token, per
314-
* #1095), un-dedupe the delivered wakes and interrupt so the questions
315-
* re-surface escalated. Returns true when it aborted. Driven by the fleet
316-
* stall poll, which settles deadline-past asks first.
314+
* #1095), interrupt the hung inference, un-dedupe the delivered wakes,
315+
* and hand the next turn to occupancy (mailbox mail) before any re-surface.
316+
* Returns true when it aborted. Driven by the fleet stall poll, which
317+
* settles deadline-past asks first, and by the #1095 monitor tick.
317318
*/
318319
abortStalledWakeTurn: () => boolean;
319320
/**
@@ -1383,9 +1384,9 @@ function applyInbound(
13831384
bag.askWakeResurface.delete(sessionId);
13841385
}
13851386
}
1386-
// A fresh snapshot with nothing pending means every parked question
1387-
// settled or expired: no wake turn is owed, so disarm the bound.
1388-
if (bag.pendingAskWake.size === 0) bag.askWakeTurnArmed = false;
1387+
// A fresh snapshot with nothing pending means no wake is owed, but a
1388+
// still-armed silent turn must stay armed so the stall abort can interrupt
1389+
// it. Disarming here (expire / send_input) would leave isProcessing hung.
13891390
bag.flushPendingAskWake?.();
13901391
return;
13911392
}
@@ -1900,30 +1901,11 @@ export function attachSessionBridge(
19001901
};
19011902

19021903
/**
1903-
* Stall bound for a silent ask-wake primary turn (CL-8016). Only an armed
1904-
* (wake-sent, never settled) turn can match. Silence uses `shouldAbortForStall`
1905-
* — the same #1095 bound that already covers awaiting-first-token — then
1906-
* un-dedupes and interrupts so parked questions re-surface. Mail first so
1907-
* occupancy can take the next turn; the trailing wake flush restates if
1908-
* mail did not start one.
1904+
* Shared abort used by Ctrl+C, the #1095 monitor tick, and the fleet-poll
1905+
* stall bound. Interrupt the hung inference before any new deliver, then
1906+
* occupancy/mailbox first so a re-surface wake cannot steal that turn.
19091907
*/
1910-
const abortStalledWakeTurn = (): boolean => {
1911-
if (bag.disposed || !bag.askWakeTurnArmed) return false;
1912-
if (!shouldAbortForStall(stallArgsFor(now()))) return false;
1913-
const layer = turnStallLayer(bag.turn) ?? "mid-stream";
1914-
unDedupeArmedAskWakes();
1915-
recordTurnMarker(bag, `stall-abort:${layer}`);
1916-
bag.askWakeTurnArmed = false;
1917-
bag.turn = turnStateOnInterrupt(bag.turn, now());
1918-
paintPhase();
1919-
drainAtBoundary(shell, bag);
1920-
flushMailboxMail();
1921-
flushPendingAskWake();
1922-
return true;
1923-
};
1924-
1925-
const doInterrupt = (): void => {
1926-
if (bag.disposed) return;
1908+
const abortInFlightAndHandoff = (): void => {
19271909
closeOpenRow(shell, bag);
19281910
bag.pendingEchoes.length = 0;
19291911
// The stopped attempt is no longer in flight. Expire the error-recovery
@@ -1942,14 +1924,37 @@ export function attachSessionBridge(
19421924
// turn the operator (or the watchdog) deliberately stopped.
19431925
recordLastSent(null);
19441926
bag.awaitingContinuationInference = false;
1945-
// Armed wake that dies here (operator stop or #1095 stall abort) never
1946-
// landed: un-dedupe so flushPendingAskWake restates the questions.
1927+
// Armed wake that dies here (operator stop or stall abort) never
1928+
// landed: un-dedupe so flushPendingAskWake can restate the questions
1929+
// if occupancy does not take the next turn.
19471930
unDedupeArmedAskWakes();
19481931
bag.askWakeTurnArmed = false;
19491932
bag.turn = turnStateOnInterrupt(bag.turn, now());
19501933
paintPhase();
1951-
flushPendingAskWake();
19521934
flushMailboxMail();
1935+
if (!bag.turn.isProcessing) flushPendingAskWake();
1936+
};
1937+
1938+
/**
1939+
* Stall bound for a silent ask-wake primary turn (CL-8016). Only an armed
1940+
* (wake-sent, never settled) turn can match. Silence uses `shouldAbortForStall`
1941+
* — the same #1095 bound that already covers awaiting-first-token — then
1942+
* interrupts the hung inference before any new deliver. Mail first so
1943+
* occupancy can take the next turn; the trailing wake flush restates only
1944+
* if mail did not start one.
1945+
*/
1946+
const abortStalledWakeTurn = (): boolean => {
1947+
if (bag.disposed || !bag.askWakeTurnArmed) return false;
1948+
if (!shouldAbortForStall(stallArgsFor(now()))) return false;
1949+
const layer = turnStallLayer(bag.turn) ?? "mid-stream";
1950+
recordTurnMarker(bag, `stall-abort:${layer}`);
1951+
abortInFlightAndHandoff();
1952+
return true;
1953+
};
1954+
1955+
const doInterrupt = (): void => {
1956+
if (bag.disposed) return;
1957+
abortInFlightAndHandoff();
19531958
};
19541959
const clearQueuedDelivery = (): void => {
19551960
if (bag.disposed) return;
@@ -2052,7 +2057,10 @@ export function attachSessionBridge(
20522057
if (shouldAbortForStall(stallArgs)) {
20532058
applyStallRecovery(
20542059
{
2055-
abort: doInterrupt,
2060+
abort: () => {
2061+
if (abortStalledWakeTurn()) return;
2062+
doInterrupt();
2063+
},
20562064
notify: (message) =>
20572065
setStatusFlash(shell, message, { ttlMs: RUNTIME_FLASH_MS }),
20582066
},

0 commit comments

Comments
 (0)