Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 90 additions & 0 deletions src/subagent/agent-fleet.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1470,6 +1470,96 @@ describe("interrupt_agent unblocks wait_agents", () => {
expect(fleetRecords.peek(worker.id)?.status).toBe("interrupted");
});

test("completeAfterInterrupt attaches the followup reply to the interrupted lane", async () => {
const sessions = createSubAgentSessionStore();
const fleetRecords = createFleetMailbox(sessions);
const worker = sessions.start({
description: "looping",
agentId: "explorer",
brief: "b",
retained: true,
});
sessions.markRunning(worker.id);
fleetRecords.register(worker.id);
fleetRecords.noteFollowup(worker.id);
fleetRecords.interrupt(worker.id);
fleetRecords.completeAfterInterrupt(worker.id, "followup reply");
// The overlay stays put, but the reply the send_input followup handoff
// passes in must surface to the parent through peek/wait.
expect(fleetRecords.peek(worker.id)?.status).toBe("interrupted");
expect(fleetRecords.peek(worker.id)?.report).toBe("followup reply");
expect(fleetRecords.peek(worker.id)?.collected).toBeUndefined();

// wait_agents delivers the reply exactly once; the bodyHanded guard
// omits it on redelivery so a repeat wait cannot hand the same reply
// out twice.
const wait = createWaitAgentsTool({ sessions, fleetRecords });
const first = await callFleetTool(wait, {
targets: [worker.id],
timeout_ms: 1000,
});
expect(first.timed_out).toBe(false);
const firstResults = first.results as {
agent_id: string;
status: string;
report?: string;
}[];
expect(defined(firstResults[0])).toEqual({
agent_id: worker.id,
status: "interrupted",
report: "followup reply",
});
expect(fleetRecords.peek(worker.id)?.bodyHanded).toBe(true);

const second = await callFleetTool(wait, {
targets: [worker.id],
timeout_ms: 1000,
});
expect(second.timed_out).toBe(false);
const secondResults = second.results as {
agent_id: string;
status: string;
report?: string;
}[];
expect(defined(secondResults[0])).toEqual({
agent_id: worker.id,
status: "interrupted",
});
});

test("tombstone releases a stored followup reply from the lane", () => {
const sessions = createSubAgentSessionStore();
const fleetRecords = createFleetMailbox(sessions);
const worker = sessions.start({
description: "looping",
agentId: "explorer",
brief: "b",
retained: true,
});
sessions.markRunning(worker.id);
fleetRecords.register(worker.id);
fleetRecords.noteFollowup(worker.id);
fleetRecords.interrupt(worker.id);
fleetRecords.completeAfterInterrupt(worker.id, "followup reply");
// Dropping the session makes the lane tombstone; tombstone is permanent
// and the record is kept, so the reply must be released from the
// overlay instead of lingering behind its hidden projection.
sessions.clear();
expect(sessions.get(worker.id)).toBeUndefined();
const snap = fleetRecords.peek(worker.id);
expect(snap?.status).toBe("interrupted");
expect(snap?.tombstoned).toBe(true);
expect(snap?.report).toBeUndefined();
// Heap-truth check: the projection gate alone would hide a retained
// reply; the overlay itself must carry nothing.
const overlay = (
fleetRecords as unknown as {
records: Map<string, { report?: string }>;
}
).records.get(worker.id);
expect(overlay?.report).toBeUndefined();
});

test("rejected send_input followup clears the lane so wait collects interrupted salvage", async () => {
const gate = deferred<RunSubAgentResult>();
const followupGate = deferred<string>();
Expand Down
26 changes: 24 additions & 2 deletions src/subagent/agent-fleet.ts
Original file line number Diff line number Diff line change
Expand Up @@ -171,6 +171,10 @@ interface FleetOverlay {
lastWaitStatus?: WaitJSONStatus;
/** Report/error already copied into a wait or occupancy payload. */
bodyHanded?: boolean;
/** Reply handed to completeAfterInterrupt — the settled follow-up's
* reply, kept on the overlay because an interrupt/close overlay can pin
* the projected status so the session payload alone would hide it. */
report?: string;
tombstoned?: boolean;
hint?: string;
/** Mirror of the session's failure record so it survives eviction. */
Expand Down Expand Up @@ -312,9 +316,11 @@ class FleetMailbox {
/**
* The interrupt followup finished. Clear the followup and queued overlays
* so wait projects the settled session, but keep a close/interrupt overlay —
* a followup reply must not undo interrupt_agent or close_agent.
* a followup reply must not undo interrupt_agent or close_agent. The reply
* itself lands on the overlay so interrupt/close-pinned lanes still hand it
* to the parent; a session payload report keeps precedence.
*/
completeAfterInterrupt(id: string, _report?: string): void {
completeAfterInterrupt(id: string, report?: string): void {
const existing = this.records.get(id);
if (existing === undefined || existing.collected === true) return;
let changed = false;
Expand All @@ -326,6 +332,10 @@ class FleetMailbox {
delete existing.forceQueued;
changed = true;
}
if (report !== undefined && report.length > 0) {
existing.report = report;
changed = true;
}
if (changed) this.sessions?.wake();
}

Expand Down Expand Up @@ -436,6 +446,10 @@ class FleetMailbox {
overlay.tombstoned = true;
overlay.hint = RECOVERY_HINT;
overlay.frozenStatus = this.projectedStatus(id, overlay);
// Tombstone is permanent, so nothing can consume the stored reply
// later — releasing it here keeps the lane's heap holding nothing
// beyond the projection.
delete overlay.report;
if (overlay.pinHeld === true) {
overlay.pinHeld = false;
this.sessions.unpin(id);
Expand All @@ -462,6 +476,11 @@ class FleetMailbox {
...(overlay.bodyHanded === true ? { bodyHanded: true } : {}),
...(overlay.tombstoned === true ? { tombstoned: true } : {}),
...(overlay.hint !== undefined ? { hint: overlay.hint } : {}),
// Tombstone already released the overlay reply (see above); it is
// hidden here too so a pre-existing tombstone never projects a reply.
...(overlay.tombstoned !== true && overlay.report !== undefined
? { report: overlay.report }
: {}),
...(payload?.report !== undefined ? { report: payload.report } : {}),
...(payload?.error !== undefined && status === "failed"
? { error: payload.error }
Expand Down Expand Up @@ -515,6 +534,9 @@ class FleetMailbox {
victim.frozenStatus = this.projectedStatus(victimId, victim);
victim.tombstoned = true;
victim.hint = RECOVERY_HINT;
// Tombstone is permanent, so the stored reply has no later consumer —
// same release as the snapshot tombstone path.
delete victim.report;
if (victim.pinHeld === true) {
victim.pinHeld = false;
this.sessions?.unpin(victimId);
Expand Down
Loading