Skip to content

Commit dfd5685

Browse files
committed
fix(tui): bound the watchdog approval resume and report it
The stall watchdog re-ran the approval handler for a parked call without checking whether a settlement already existed. An in-flight settlement was joined, so the attempt hung with the turn, and a handed-over one was ignored while still counting as recovered, which suppressed the interrupt without resuming anything. The handler now reports where a correlation's settlement stands, and the resume only re-presents the approval when none exists. Otherwise it falls back to the existing interrupt. The bridge runs at most one resume at a time, gives it one stall budget before interrupting, and re-checks that the turn is still stalled before a fallback fires. Resume attempts flash a status that names the attempt without claiming the call ran, and record a content-free turn marker for each outcome. The recovery handle is cleared once the original handler settles. Bridge tests cover handled, declined, failed, duplicate-tick and wedged attempts, and unit tests cover the settlement guard.
1 parent efe2bb3 commit dfd5685

6 files changed

Lines changed: 267 additions & 20 deletions

File tree

‎src/session/approval-resume.test.ts‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ import { createExtraDeniedPathMatcher } from "../plugins/secret-guard-plugin.js"
1717
import {
1818
APPROVAL_DROPPED_NOTICE,
1919
createApprovalResume,
20+
createSuspendedApprovalRecovery,
2021
requestFromApprovalSnapshot,
2122
resolveParkedCallIdFromStore,
2223
resolveSuspendedApprovalFromStore,
@@ -489,6 +490,80 @@ describe("persisted suspended approval proof", () => {
489490
});
490491
});
491492

493+
describe("suspended approval recovery", () => {
494+
function recoveryWith(
495+
status: "idle" | "in-flight" | "handed-over",
496+
pending: PendingOperation[] = [operation],
497+
) {
498+
const handle = mock(async (_result: SendResult) => true);
499+
const recovery = createSuspendedApprovalRecovery({
500+
storage: () => storeWith(pending),
501+
resume: { handle, status: () => status },
502+
});
503+
return { recovery, handle };
504+
}
505+
const result = suspension("corr-A", "echo same");
506+
507+
test("re-presents the verified approval once when nothing is settling", async () => {
508+
const { recovery, handle } = recoveryWith("idle");
509+
recovery.capture(result, () => true);
510+
511+
expect(await recovery.tryResumeOnce()).toBe(true);
512+
expect(handle).toHaveBeenCalledTimes(1);
513+
// A second attempt for the same capture takes the fallback.
514+
expect(await recovery.tryResumeOnce()).toBe(false);
515+
expect(handle).toHaveBeenCalledTimes(1);
516+
});
517+
518+
test.each(["in-flight", "handed-over"] as const)(
519+
"never re-enters a %s settlement",
520+
async (status) => {
521+
const { recovery, handle } = recoveryWith(status);
522+
recovery.capture(result, () => true);
523+
524+
expect(await recovery.tryResumeOnce()).toBe(false);
525+
expect(handle).not.toHaveBeenCalled();
526+
},
527+
);
528+
529+
test("a stale generation, a cleared capture or an unproven store never delivers", async () => {
530+
const stale = recoveryWith("idle");
531+
stale.recovery.capture(result, () => false);
532+
expect(await stale.recovery.tryResumeOnce()).toBe(false);
533+
534+
const cleared = recoveryWith("idle");
535+
cleared.recovery.capture(result, () => true);
536+
cleared.recovery.clear();
537+
expect(await cleared.recovery.tryResumeOnce()).toBe(false);
538+
539+
const unproven = recoveryWith("idle", []);
540+
unproven.recovery.capture(result, () => true);
541+
expect(await unproven.recovery.tryResumeOnce()).toBe(false);
542+
543+
for (const { handle } of [stale, cleared, unproven]) {
544+
expect(handle).not.toHaveBeenCalled();
545+
}
546+
});
547+
548+
test("a generation that changes during verification drops the delivery", async () => {
549+
let current = true;
550+
const handle = mock(async (_result: SendResult) => true);
551+
const recovery = createSuspendedApprovalRecovery({
552+
storage: () => ({
553+
load: async () => {
554+
current = false;
555+
return await storeWith([operation]).load();
556+
},
557+
}),
558+
resume: { handle, status: () => "idle" },
559+
});
560+
recovery.capture(result, () => current);
561+
562+
expect(await recovery.tryResumeOnce()).toBe(false);
563+
expect(handle).not.toHaveBeenCalled();
564+
});
565+
});
566+
492567
test("missing or ambiguous stored identity never falls back to an identical history call", async () => {
493568
for (const pending of [[], [operation, operation]]) {
494569
const deliver = mock((_message: InboundMessage): void => undefined);

‎src/session/approval-resume.ts‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,11 @@ export interface ApprovalResume {
5353
* results so callers can forward them unchanged.
5454
*/
5555
handle: (result: SendResult) => Promise<boolean>;
56+
/**
57+
* Where a correlation's settlement stands: `idle` when nothing is gating or
58+
* has been handed over, so a fresh `handle` would open the operator surface.
59+
*/
60+
status: (correlationId: string) => "idle" | "in-flight" | "handed-over";
5661
}
5762

5863
export interface SuspendedApprovalRecovery {
@@ -93,6 +98,11 @@ export function createSuspendedApprovalRecovery(args: {
9398
return false;
9499
}
95100
current.attempted = true;
101+
// Re-running `handle` only helps when no settlement exists for this
102+
// correlation. An in-flight one would be joined (it is the wedge), and a
103+
// handed-over one would be ignored, so both fall back to the abort.
104+
if (args.resume.status(current.result.correlationId) !== "idle")
105+
return false;
96106
const storage = args.storage();
97107
if (storage === undefined) return false;
98108
try {
@@ -466,6 +476,12 @@ export function createApprovalResume(args: {
466476
};
467477

468478
return {
479+
status: (correlationId) =>
480+
handedOver.has(correlationId)
481+
? "handed-over"
482+
: inflight.has(correlationId)
483+
? "in-flight"
484+
: "idle",
469485
handle: (result) => {
470486
if (result.type !== "suspended") return Promise.resolve(false);
471487
const { correlationId } = result;

‎src/tui/runner/submit.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -395,6 +395,8 @@ export function createSubmitPath(
395395
);
396396
}
397397
await services.approvalResume.handle(result);
398+
// Settled: nothing is left for the stall watchdog to re-present.
399+
services.suspendedApprovalRecovery.clear();
398400
});
399401
return { status: "accepted" };
400402
} catch (error) {

‎src/tui/runtime-bridge.ts‎

Lines changed: 46 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ import {
5151
isStalledForDisplay,
5252
shouldAbortForStall,
5353
stallLevel,
54+
STALL_APPROVAL_RESUME_MESSAGE,
5455
STALL_NOTICE_MESSAGE,
5556
STALL_NOTICE_MS,
5657
STALL_RECOVERY_MESSAGE,
@@ -178,12 +179,6 @@ export interface TurnMonitorOptions {
178179
readonly stallNoticeMs?: number;
179180
/** Registers the periodic tick; returns an unsubscribe. */
180181
readonly schedule?: (tick: () => void, intervalMs: number) => () => void;
181-
/**
182-
* Attempts to re-present one already persisted approval suspension. This
183-
* deliberately receives no agent, storage, grant, or tool capability: a
184-
* true result means the correlated approval path owns the continuation.
185-
*/
186-
readonly tryResumeSuspendedApproval?: () => boolean | Promise<boolean>;
187182
}
188183

189184
const DEFAULT_TICK_MS = 250;
@@ -241,8 +236,14 @@ export interface SessionBridge {
241236
gateOpened: () => void;
242237
/** A previously raised gate resolved. */
243238
gateClosed: () => void;
239+
/**
240+
* Registers the attempt to re-present one already persisted approval
241+
* suspension when the stall watchdog fires. It receives no agent, storage,
242+
* grant, or tool capability: a true result means the correlated approval
243+
* path owns the continuation, false falls back to the interrupt.
244+
*/
244245
setSuspendedApprovalRecovery: (
245-
recovery: (() => boolean | Promise<boolean>) | undefined,
246+
recovery: (() => Promise<boolean>) | undefined,
246247
) => void;
247248
dispose: () => void;
248249
/** Current derived turn phase (progress label, stall clock, quota window). */
@@ -519,7 +520,7 @@ export interface BridgeBag {
519520
/** callId→name / delta bookkeeping for production-shaped events. */
520521
mapCtx: StreamMapContext;
521522
disposed: boolean;
522-
suspendedApprovalRecovery: (() => boolean | Promise<boolean>) | undefined;
523+
suspendedApprovalRecovery: (() => Promise<boolean>) | undefined;
523524
turn: TurnState;
524525
/**
525526
* Live fleet-lane count from the last `fleet` event (idle-with-fleet).
@@ -2154,6 +2155,9 @@ export function attachSessionBridge(
21542155
flushOccupancyThenWake();
21552156
};
21562157

2158+
// When the in-flight approval resume began; at most one runs at a time.
2159+
let stallResumeStartedAt: number | undefined;
2160+
21572161
const tick = (): void => {
21582162
if (bag.disposed) return;
21592163
const nowMs = now();
@@ -2224,20 +2228,42 @@ export function attachSessionBridge(
22242228
},
22252229
STALL_RECOVERY_MESSAGE,
22262230
);
2227-
const resumed =
2228-
bag.suspendedApprovalRecovery?.() ??
2229-
monitor?.tryResumeSuspendedApproval?.();
2230-
if (resumed instanceof Promise) {
2231-
void resumed.then(
2232-
(handled) => {
2233-
if (!handled) recover();
2234-
},
2235-
() => recover(),
2236-
);
2231+
if (stallResumeStartedAt !== undefined) {
2232+
// One resume attempt is already out. Give it a single stall budget,
2233+
// then take the abort so a wedged attempt cannot hold the turn open.
2234+
if (nowMs - stallResumeStartedAt < stallArgs.stallTimeoutMs) return;
2235+
stallResumeStartedAt = undefined;
2236+
recordTurnMarker(bag, "stall-resume:timeout");
2237+
recover();
22372238
return;
22382239
}
2239-
if (resumed) return;
2240-
recover();
2240+
const resume = bag.suspendedApprovalRecovery;
2241+
if (resume === undefined) {
2242+
recover();
2243+
return;
2244+
}
2245+
const startedAt = nowMs;
2246+
stallResumeStartedAt = startedAt;
2247+
const settle = (handled: boolean, marker: string): void => {
2248+
// A timed-out or superseded attempt already took its own outcome.
2249+
if (stallResumeStartedAt !== startedAt) return;
2250+
stallResumeStartedAt = undefined;
2251+
recordTurnMarker(bag, marker);
2252+
if (handled) return;
2253+
// Re-evaluate: the turn may have moved on while the attempt ran.
2254+
if (shouldAbortForStall(stallArgsFor(now()))) recover();
2255+
};
2256+
setStatusFlash(shell, STALL_APPROVAL_RESUME_MESSAGE, {
2257+
ttlMs: RUNTIME_FLASH_MS,
2258+
});
2259+
void resume().then(
2260+
(handled) =>
2261+
settle(
2262+
handled,
2263+
handled ? "stall-resume:handled" : "stall-resume:fallback",
2264+
),
2265+
() => settle(false, "stall-resume:error"),
2266+
);
22412267
return;
22422268
}
22432269

‎src/tui/stall-watchdog.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -159,6 +159,13 @@ export const STALL_NOTICE_MESSAGE =
159159
export const STALL_RECOVERY_MESSAGE =
160160
"stopped after no response — send again to retry";
161161

162+
/**
163+
* Shown while the watchdog re-presents an approval that was already parked.
164+
* Names the attempt only: it never claims the parked call ran.
165+
*/
166+
export const STALL_APPROVAL_RESUME_MESSAGE =
167+
"no response while waiting on approval, re-presenting the pending approval";
168+
162169
/**
163170
* Shown once a repeated line aborts the turn. Named as degeneration, not a
164171
* generic failure, so a retry reads as the reasonable next step rather than

‎src/tui/turn-monitor.test.ts‎

Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import { withTestRenderer } from "./harness.js";
1212
import { RUNTIME_FLASH_MS } from "./runtime-notices.js";
1313
import { LIVE_WORD_MS } from "./chrome-state.js";
1414
import {
15+
STALL_APPROVAL_RESUME_MESSAGE,
1516
STALL_NOTICE_MESSAGE,
1617
STALL_RECOVERY_MESSAGE,
1718
} from "./stall-watchdog.js";
@@ -553,6 +554,126 @@ describe("stall watchdog", () => {
553554
});
554555
});
555556

557+
describe("stall recovery of a suspended approval", () => {
558+
const flush = (): Promise<void> => new Promise((r) => setTimeout(r, 0));
559+
560+
// A mid-stream hang past the stall budget: the one shape auto-abort acts on.
561+
function reachStallBudget(t: Harness) {
562+
t.bridge.submit("build it", "immediate");
563+
t.bridge.handle({ type: "inference.text.delta", data: { token: "ok" } });
564+
t.port.clear();
565+
t.advance(1_500);
566+
}
567+
568+
function deferredRecovery() {
569+
let calls = 0;
570+
let settle!: (handled: boolean) => void;
571+
let fail!: (error: Error) => void;
572+
const recovery = (): Promise<boolean> => {
573+
calls += 1;
574+
return new Promise<boolean>((resolve, reject) => {
575+
settle = resolve;
576+
fail = reject;
577+
});
578+
};
579+
return {
580+
recovery,
581+
calls: () => calls,
582+
resolve: (handled: boolean) => settle(handled),
583+
reject: (error: Error) => fail(error),
584+
};
585+
}
586+
587+
test("a handled resume suppresses the interrupt and says what it is doing", async () => {
588+
await withHarness(async (t) => {
589+
const d = deferredRecovery();
590+
t.bridge.setSuspendedApprovalRecovery(d.recovery);
591+
reachStallBudget(t);
592+
593+
t.tick();
594+
expect(d.calls()).toBe(1);
595+
expect(t.shell.statusFlash).toBe(STALL_APPROVAL_RESUME_MESSAGE);
596+
597+
d.resolve(true);
598+
await flush();
599+
expect(t.port.calls).toEqual([]);
600+
});
601+
});
602+
603+
test("a resume that cannot proceed falls back to the one interrupt", async () => {
604+
await withHarness(async (t) => {
605+
const d = deferredRecovery();
606+
t.bridge.setSuspendedApprovalRecovery(d.recovery);
607+
reachStallBudget(t);
608+
609+
t.tick();
610+
d.resolve(false);
611+
await flush();
612+
expect(t.port.calls).toEqual([{ op: "interrupt" }]);
613+
expect(t.shell.statusFlash).toBe(STALL_RECOVERY_MESSAGE);
614+
});
615+
});
616+
617+
test("a resume that throws falls back to the interrupt", async () => {
618+
await withHarness(async (t) => {
619+
const d = deferredRecovery();
620+
t.bridge.setSuspendedApprovalRecovery(d.recovery);
621+
reachStallBudget(t);
622+
623+
t.tick();
624+
d.reject(new Error("boom"));
625+
await flush();
626+
expect(t.port.calls).toEqual([{ op: "interrupt" }]);
627+
});
628+
});
629+
630+
test("duplicate ticks while a resume is out neither re-enter nor interrupt", async () => {
631+
await withHarness(async (t) => {
632+
const d = deferredRecovery();
633+
t.bridge.setSuspendedApprovalRecovery(d.recovery);
634+
reachStallBudget(t);
635+
636+
t.tick();
637+
t.advance(250);
638+
t.tick();
639+
t.advance(250);
640+
t.tick();
641+
expect(d.calls()).toBe(1);
642+
expect(t.port.calls).toEqual([]);
643+
644+
d.resolve(false);
645+
await flush();
646+
expect(t.port.calls).toEqual([{ op: "interrupt" }]);
647+
});
648+
});
649+
650+
test("a wedged resume is bounded by one more stall budget", async () => {
651+
await withHarness(async (t) => {
652+
const d = deferredRecovery();
653+
t.bridge.setSuspendedApprovalRecovery(d.recovery);
654+
reachStallBudget(t);
655+
656+
t.tick();
657+
t.advance(1_000);
658+
t.tick();
659+
expect(t.port.calls).toEqual([{ op: "interrupt" }]);
660+
661+
// The late settle must not interrupt a second time.
662+
d.resolve(false);
663+
await flush();
664+
expect(t.port.calls).toHaveLength(1);
665+
});
666+
});
667+
668+
test("no registered recovery keeps the plain abort", async () => {
669+
await withHarness(async (t) => {
670+
reachStallBudget(t);
671+
t.tick();
672+
expect(t.port.calls).toEqual([{ op: "interrupt" }]);
673+
});
674+
});
675+
});
676+
556677
describe("repetition guard", () => {
557678
test("a slow but progressing turn is never killed", async () => {
558679
await withHarness(async (t) => {

0 commit comments

Comments
 (0)