Skip to content

Commit 2e00ef9

Browse files
committed
fix(tui): claim the primary turn for occupancy before ask wake
Mailbox mail collect is async, so isProcessing is still false while occupancy already owns the slot. A shared mail-then-wake guard uses the latch claim so ask-wake cannot send on that stack.
1 parent 37c3c1f commit 2e00ef9

5 files changed

Lines changed: 257 additions & 26 deletions

File tree

‎src/subagent/mailbox-mail-drive.test.ts‎

Lines changed: 45 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -132,7 +132,7 @@ describe("driveMailboxMail", () => {
132132
}),
133133
).toBe(false);
134134
expect(
135-
await driveMailboxMail({
135+
driveMailboxMail({
136136
parentProcessing: false,
137137
mailbox: mapMailbox(new Map()),
138138
lanes: [],
@@ -315,7 +315,7 @@ describe("driveMailboxMail", () => {
315315
["live", { status: "running" }],
316316
]);
317317
expect(
318-
await driveMailboxMail({
318+
driveMailboxMail({
319319
parentProcessing: false,
320320
mailbox: mapMailbox(records),
321321
lanes: [],
@@ -329,6 +329,28 @@ describe("driveMailboxMail", () => {
329329
).toBe(false);
330330
});
331331

332+
test("does not begin if the parent starts processing during collect", async () => {
333+
const records = new Map<string, FleetDryMailboxRecord>([
334+
["w1", { status: "done", report: "ok" }],
335+
]);
336+
let processing = false;
337+
const driven = driveMailboxMail({
338+
parentProcessing: false,
339+
isParentProcessing: () => processing,
340+
mailbox: mapMailbox(records),
341+
lanes: [],
342+
beginSystemContinuation: () => {
343+
throw new Error("must not begin");
344+
},
345+
send: () => {
346+
throw new Error("must not send");
347+
},
348+
});
349+
processing = true;
350+
expect(await driven).toBe(false);
351+
expect(records.get("w1")?.collected).not.toBe(true);
352+
});
353+
332354
test("not-delivered send leaves the wake retryable", async () => {
333355
const records = new Map<string, FleetDryMailboxRecord>([
334356
["w1", { status: "done", report: "ok" }],
@@ -395,8 +417,10 @@ describe("latchMailboxMailDrive", () => {
395417
}),
396418
);
397419
expect(driver()).toBe(true);
420+
expect(driver.claimed()).toBe(true);
398421
expect(driver()).toBe(false);
399422
expect(driver()).toBe(false);
423+
expect(driver.claimed()).toBe(true);
400424
await sent;
401425
expect(sends).toHaveLength(1);
402426
expect(sends[0]).toContain(MAILBOX_MAIL_WAKE_PREFIX);
@@ -410,9 +434,28 @@ describe("latchMailboxMailDrive", () => {
410434
});
411435
expect(driver()).toBe(false);
412436
expect(driver()).toBe(false);
437+
expect(driver.claimed()).toBe(false);
413438
expect(calls).toBe(2);
414439
});
415440

441+
test("an empty mailbox does not claim the occupancy slot", () => {
442+
const driver = latchMailboxMailDrive(() =>
443+
driveMailboxMail({
444+
parentProcessing: false,
445+
mailbox: mapMailbox(new Map()),
446+
lanes: [],
447+
beginSystemContinuation: () => {
448+
throw new Error("must not begin");
449+
},
450+
send: () => {
451+
throw new Error("must not send");
452+
},
453+
}),
454+
);
455+
expect(driver()).toBe(false);
456+
expect(driver.claimed()).toBe(false);
457+
});
458+
416459
test("after the in-flight drive settles, a new terminal can send", async () => {
417460
const records = new Map<string, FleetDryMailboxRecord>([
418461
["first", { status: "done", report: "one" }],

‎src/subagent/mailbox-mail-drive.ts‎

Lines changed: 46 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -109,16 +109,41 @@ export function buildMailboxMailPrompt(
109109
return [mailboxMailWakeLine(), JSON.stringify(reports)].join("\n");
110110
}
111111

112+
function hasDeliverableMailboxMail(
113+
mailbox: FleetDryMailbox | undefined,
114+
): boolean {
115+
if (mailbox === undefined) return false;
116+
const delivering = deliveringByMailbox.get(mailbox);
117+
for (const id of mailbox.ids()) {
118+
if (delivering?.has(id)) continue;
119+
const record = mailbox.peek(id);
120+
if (record === undefined || record.collected === true) continue;
121+
if (isLiveWaitStatus(record.status)) continue;
122+
return true;
123+
}
124+
return false;
125+
}
126+
127+
function parentIsProcessing(args: {
128+
parentProcessing: boolean;
129+
isParentProcessing?: () => boolean;
130+
}): boolean {
131+
return args.isParentProcessing?.() ?? args.parentProcessing;
132+
}
133+
112134
export function driveMailboxMail(args: {
113135
parentProcessing: boolean;
136+
/** Live check after awaited collect; occupancy must not start a turn already claimed. */
137+
isParentProcessing?: () => boolean;
114138
mailbox: FleetDryMailbox | undefined;
115139
lanes: readonly FleetDryLane[];
116140
writeBlob?: FleetDryBlobWriter;
117141
beginSystemContinuation: (prompt: string) => void;
118142
send: (prompt: string) => unknown;
119143
onSendFailure?: () => void;
120144
}): boolean | Promise<boolean> {
121-
if (args.parentProcessing) return false;
145+
if (parentIsProcessing(args)) return false;
146+
if (!hasDeliverableMailboxMail(args.mailbox)) return false;
122147
return driveMailboxMailAfterCollect(args);
123148
}
124149

@@ -138,6 +163,7 @@ async function driveMailboxMailAfterCollect(
138163
)
139164
).filter((report) => !delivering.has(report.agent_id));
140165
if (reports.length === 0) return false;
166+
if (parentIsProcessing(args)) return false;
141167
const ids = reports.map((report) => report.agent_id);
142168
for (const id of ids) delivering.add(id);
143169
const prompt = buildMailboxMailPrompt(reports);
@@ -169,12 +195,28 @@ async function driveMailboxMailAfterCollect(
169195
* mail. `driveMailboxMail` does not mark the parent busy until after an
170196
* awaited collect, so overlapping flushes would each call `send()` and fill
171197
* the agent's depth-16 queue. Hold one drive until that promise settles.
198+
*
199+
* `flush()` is true only when this call starts a drive. `claimed()` is true
200+
* while that drive owns the slot — ask-wake admission uses it so a wake
201+
* cannot send on the same stack, or on a later stack before collect lands.
172202
*/
203+
export type MailboxMailFlush = (() => boolean) & {
204+
readonly claimed: () => boolean;
205+
};
206+
207+
export function mailboxMailDriveClaimed(
208+
flush: (() => boolean) | undefined,
209+
): boolean {
210+
if (flush === undefined) return false;
211+
const claimed = (flush as MailboxMailFlush).claimed;
212+
return typeof claimed === "function" && claimed() === true;
213+
}
214+
173215
export function latchMailboxMailDrive(
174216
drive: () => boolean | Promise<boolean>,
175-
): () => boolean {
217+
): MailboxMailFlush {
176218
let inFlight = false;
177-
return () => {
219+
const flush = (): boolean => {
178220
if (inFlight) return false;
179221
const driven = drive();
180222
if (driven === false) return false;
@@ -184,4 +226,5 @@ export function latchMailboxMailDrive(
184226
});
185227
return true;
186228
};
229+
return Object.assign(flush, { claimed: () => inFlight });
187230
}

‎src/tui/agent-ask-wake.test.ts‎

Lines changed: 133 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import { createLiveSessionPort } from "./live-session-port";
44
import { createAppShell } from "./shell/index";
55
import { withTestRenderer } from "./harness";
66
import type { PendingAskWake } from "../subagent/fleet-report.js";
7+
import { latchMailboxMailDrive } from "../subagent/mailbox-mail-drive.js";
78
import { classifySubmission, createSubmitHandler } from "./runner/submit.js";
89
import {
910
createDeliveryGeneration,
@@ -33,7 +34,7 @@ async function withWakeBridge(
3334
run: (
3435
bridge: ReturnType<typeof attachSessionBridge>,
3536
sends: string[],
36-
) => void,
37+
) => void | Promise<void>,
3738
) {
3839
await withTestRenderer(
3940
async (h) => {
@@ -55,7 +56,7 @@ async function withWakeBridge(
5556
}),
5657
);
5758
try {
58-
run(bridge, sends);
59+
await run(bridge, sends);
5960
} finally {
6061
bridge.dispose();
6162
shell.dispose();
@@ -65,6 +66,22 @@ async function withWakeBridge(
6566
);
6667
}
6768

69+
function latchAsyncMailboxMail(
70+
bridge: ReturnType<typeof attachSessionBridge>,
71+
order: string[],
72+
hold: Promise<void>,
73+
) {
74+
return latchMailboxMailDrive(() => {
75+
order.push("mail");
76+
return (async () => {
77+
await hold;
78+
order.push("begin");
79+
bridge.beginSystemContinuation("mailbox occupancy");
80+
return true;
81+
})();
82+
});
83+
}
84+
6885
for (const action of [
6986
"retry",
7087
"interrupt",
@@ -485,6 +502,120 @@ describe("agent ask wake delivery", () => {
485502
});
486503
});
487504

505+
test("async mailbox drive on gate close claims the slot before wake (CL-8061)", async () => {
506+
await withWakeBridge(async (bridge, sends) => {
507+
const order: string[] = [];
508+
let release: () => void = () => undefined;
509+
const hold = new Promise<void>((resolve) => {
510+
release = resolve;
511+
});
512+
bridge.setOnAskWakeSent(() => {
513+
order.push("wake");
514+
});
515+
bridge.setMailboxMailDriver(latchAsyncMailboxMail(bridge, order, hold));
516+
bridge.gateOpened();
517+
bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] });
518+
expect(sends).toEqual([]);
519+
bridge.gateClosed();
520+
expect(order).toEqual(["mail"]);
521+
expect(sends).toEqual([]);
522+
expect(bridge.turn.isProcessing).toBe(false);
523+
release();
524+
await hold;
525+
await Promise.resolve();
526+
await Promise.resolve();
527+
expect(order).toEqual(["mail", "begin"]);
528+
expect(bridge.turn.isProcessing).toBe(true);
529+
expect(sends).toEqual([]);
530+
});
531+
});
532+
533+
test("async mailbox drive on idle-with-fleet settle claims the slot before wake (CL-8061)", async () => {
534+
await withWakeBridge(async (bridge, sends) => {
535+
const order: string[] = [];
536+
let release: () => void = () => undefined;
537+
const hold = new Promise<void>((resolve) => {
538+
release = resolve;
539+
});
540+
bridge.setOnAskWakeSent(() => {
541+
order.push("wake");
542+
});
543+
bridge.setMailboxMailDriver(latchAsyncMailboxMail(bridge, order, hold));
544+
bridge.handle({ type: "inference.start", data: {} });
545+
bridge.handle({ type: "fleet", running: 1 });
546+
bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] });
547+
expect(sends).toEqual([]);
548+
bridge.handle({ type: "inference.done", data: {} });
549+
expect(order).toEqual(["mail"]);
550+
expect(sends).toEqual([]);
551+
expect(bridge.turn.isProcessing).toBe(false);
552+
release();
553+
await hold;
554+
await Promise.resolve();
555+
await Promise.resolve();
556+
expect(order).toEqual(["mail", "begin"]);
557+
expect(bridge.turn.isProcessing).toBe(true);
558+
expect(sends).toEqual([]);
559+
});
560+
});
561+
562+
test("releaseRunToIdle flushes mailbox mail before the ask wake (CL-8061)", async () => {
563+
await withWakeBridge((bridge, sends) => {
564+
const order: string[] = [];
565+
bridge.setOnAskWakeSent(() => {
566+
order.push("wake");
567+
});
568+
bridge.setMailboxMailDriver(() => {
569+
order.push("mail");
570+
return false;
571+
});
572+
bridge.handle({ type: "inference.start", data: {} });
573+
bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] });
574+
expect(sends).toEqual([]);
575+
bridge.handle({ type: "inference.done", data: {} });
576+
expect(order).toEqual(["mail", "wake"]);
577+
expect(sends).toHaveLength(1);
578+
});
579+
});
580+
581+
test("releaseRunToIdle suppresses the wake when mail takes the turn (CL-8061)", async () => {
582+
await withWakeBridge((bridge, sends) => {
583+
const order: string[] = [];
584+
bridge.setOnAskWakeSent(() => {
585+
order.push("wake");
586+
});
587+
bridge.setMailboxMailDriver(() => {
588+
order.push("mail");
589+
bridge.beginSystemContinuation("mailbox occupancy");
590+
return true;
591+
});
592+
bridge.handle({ type: "inference.start", data: {} });
593+
bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] });
594+
expect(sends).toEqual([]);
595+
bridge.handle({ type: "inference.done", data: {} });
596+
expect(order).toEqual(["mail"]);
597+
expect(sends).toEqual([]);
598+
});
599+
});
600+
601+
test("idle parent subscribe does not flush wake before mailbox mail (CL-8061)", async () => {
602+
await withWakeBridge((bridge, sends) => {
603+
const order: string[] = [];
604+
bridge.setOnAskWakeSent(() => {
605+
order.push("wake");
606+
});
607+
bridge.setMailboxMailDriver(() => {
608+
order.push("mail");
609+
bridge.beginSystemContinuation("mailbox occupancy");
610+
return true;
611+
});
612+
bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] });
613+
expect(order).toEqual(["mail"]);
614+
expect(sends).toEqual([]);
615+
expect(bridge.turn.isProcessing).toBe(true);
616+
});
617+
});
618+
488619
test("clearQueuedDelivery drops stashed wakes and resets delivered identities", async () => {
489620
await withWakeBridge((bridge, sends) => {
490621
bridge.handle({ type: "inference.start", data: {} });

‎src/tui/runner/wiring.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -362,6 +362,7 @@ export function wirePostStartup(
362362
const storage = state.currentStorage;
363363
return driveMailboxMail({
364364
parentProcessing: sessionBridge.turn.isProcessing,
365+
isParentProcessing: () => sessionBridge.turn.isProcessing,
365366
mailbox: services.toolset.fleetRecords,
366367
lanes: services.subAgentSessions.list(),
367368
...(storage !== null

0 commit comments

Comments
 (0)