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
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename

### Fixed

- Occupancy takes one dry-episode shot when the parent settles idle even if the
live fleet 1→0 edge was never observed.
- Dry-fleet transcript and `/status` report the outcome tally only
(`2 done, 1 failed`). They no longer claim `nothing running` when the
parent may still continue.
Expand Down
187 changes: 187 additions & 0 deletions src/tui/runtime-bridge.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1718,6 +1718,193 @@ describe("fleet-dry open-task drive (CL-7540)", () => {
);
});

test("already-dry settle drives once even without a live 1→0 fleet event", async () => {
await withTestRenderer(
async (h) => {
const shell = createAppShell(h.renderer, {
terminal: { columns: 80, rows: 24 },
wireKeys: false,
run: "idle",
});
const port = createRecordingPort();
const bridge = attachSessionBridge(shell, port);
try {
const prompt =
"The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)\n";
let drives = 0;
bridge.setDryOpenTaskDriver(() => {
drives += 1;
bridge.beginSystemContinuation(prompt);
return true;
});
bridge.submit("dispatch workers", "immediate");
bridge.handle({ type: "fleet", running: 0 });
expect(drives).toBe(0);
expect(shell.session.run).toBe("busy");
settleToollessTurn(bridge);
expect(drives).toBe(1);
expect(shell.session.run).toBe("busy");
settleToollessTurn(bridge);
expect(drives).toBe(1);
} finally {
bridge.dispose();
shell.dispose();
}
},
{ width: 80, height: 24 },
);
});

test("a no-op dry settle does not eat the shot for a later missed 1→0 with open tasks", async () => {
await withTestRenderer(
async (h) => {
const shell = createAppShell(h.renderer, {
terminal: { columns: 80, rows: 24 },
wireKeys: false,
run: "idle",
});
const port = createRecordingPort();
const bridge = attachSessionBridge(shell, port);
try {
const prompt =
"The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)\n";
let openTasks = false;
let drives = 0;
bridge.setDryOpenTaskDriver(() => {
if (!openTasks) return false;
drives += 1;
bridge.beginSystemContinuation(prompt);
return true;
});
bridge.submit("first turn, no todos", "immediate");
bridge.handle({ type: "fleet", running: 0 });
settleToollessTurn(bridge);
expect(drives).toBe(0);
expect(shell.session.run).toBe("idle");

openTasks = true;
bridge.submit("dispatch workers", "immediate");
bridge.handle({ type: "fleet", running: 0 });
expect(drives).toBe(0);
expect(shell.session.run).toBe("busy");
settleToollessTurn(bridge);
expect(drives).toBe(1);
expect(shell.session.run).toBe("busy");
settleToollessTurn(bridge);
expect(drives).toBe(1);
} finally {
bridge.dispose();
shell.dispose();
}
},
{ width: 80, height: 24 },
);
});

test("a new live lane resets the dry-episode latch for one more occupancy shot", async () => {
await withTestRenderer(
async (h) => {
const shell = createAppShell(h.renderer, {
terminal: { columns: 80, rows: 24 },
wireKeys: false,
run: "idle",
});
const port = createRecordingPort();
const bridge = attachSessionBridge(shell, port);
try {
const prompt =
"The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)\n";
let drives = 0;
bridge.setDryOpenTaskDriver(() => {
drives += 1;
bridge.beginSystemContinuation(prompt);
return true;
});
bridge.submit("dispatch workers", "immediate");
settleToollessTurn(bridge);
expect(drives).toBe(1);
expect(shell.session.run).toBe("busy");
settleToollessTurn(bridge);
expect(drives).toBe(1);
expect(shell.session.run).toBe("idle");

bridge.submit("dispatch more workers", "immediate");
bridge.handle({ type: "fleet", running: 1 });
settleToollessTurn(bridge);
expect(drives).toBe(1);
expect(shell.session.run).toBe("busy");
bridge.handle({ type: "fleet", running: 0 });
expect(drives).toBe(2);
expect(shell.session.run).toBe("busy");
} finally {
bridge.dispose();
shell.dispose();
}
},
{ width: 80, height: 24 },
);
});

test("parent still processing does not take the occupancy shot", async () => {
await withTestRenderer(
async (h) => {
const shell = createAppShell(h.renderer, {
terminal: { columns: 80, rows: 24 },
wireKeys: false,
run: "idle",
});
const port = createRecordingPort();
const bridge = attachSessionBridge(shell, port);
try {
let drives = 0;
bridge.setDryOpenTaskDriver(() => {
drives += 1;
return true;
});
bridge.submit("dispatch workers", "immediate");
bridge.handle({ type: "fleet", running: 0 });
expect(drives).toBe(0);
expect(shell.session.run).toBe("busy");
expect(bridge.turn.isProcessing).toBe(true);
} finally {
bridge.dispose();
shell.dispose();
}
},
{ width: 80, height: 24 },
);
});

test("live fleet still blocks the occupancy drive", async () => {
await withTestRenderer(
async (h) => {
const shell = createAppShell(h.renderer, {
terminal: { columns: 80, rows: 24 },
wireKeys: false,
run: "idle",
});
const port = createRecordingPort();
const bridge = attachSessionBridge(shell, port);
try {
let drives = 0;
bridge.setDryOpenTaskDriver(() => {
drives += 1;
return true;
});
bridge.submit("dispatch workers", "immediate");
bridge.handle({ type: "fleet", running: 1 });
settleToollessTurn(bridge);
expect(drives).toBe(0);
expect(shell.session.run).toBe("busy");
} finally {
bridge.dispose();
shell.dispose();
}
},
{ width: 80, height: 24 },
);
});

test("occupancy send failure re-arms, clears continuation hold, and drains follow-ups", async () => {
await withTestRenderer(
async (h) => {
Expand Down
52 changes: 26 additions & 26 deletions src/tui/runtime-bridge.ts
Original file line number Diff line number Diff line change
Expand Up @@ -252,15 +252,15 @@ export interface SessionBridge {
beginSystemContinuation: (text: string) => void;
/**
* Occupancy send failed after beginSystemContinuation. Drop the occupancy
* echo so a later matching inbound is not swallowed, re-arm the dry-open
* echo so a later matching inbound is not swallowed, re-arm the dry-episode
* latch, drop the continuation hold, and idle so follow-ups can drain and a
* later settle can take another occupancy shot.
*/
abortSystemContinuation: () => void;
/**
* Occupancy owner for dry+open continuation. Called once from
* settleRunToIdle when a latched fleet-dry edge is still dry. Return true
* if a continuation was sent (run stays busy).
* Occupancy owner for dry+open continuation. Called once per dry episode
* from settleRunToIdle when the fleet is dry. Return true if a continuation
* was sent (run stays busy).
*/
setDryOpenTaskDriver: (driver: (() => boolean) | undefined) => void;
}
Expand Down Expand Up @@ -425,18 +425,19 @@ export interface BridgeBag {
*/
flushPendingAskWake: (() => void) | null;
/**
* One deferred occupancy shot for the last live-fleet → 0 edge. Consumed on
* settle so a missed wentDry while the parent was processing still drives
* once, and a later settle cannot loop.
* One occupancy shot per dry episode. Reset when a live lane starts. Consumed
* only when the driver actually sends a continuation — a no-op (no open
* tasks) must not eat the shot, or a later missed 1→0 with leftover tasks
* never drives. A later settle after a true drive cannot loop.
*/
pendingDryOpenDrive: boolean;
droveOpenTasksThisDry: boolean;
/**
* beginSystemContinuation re-armed the turn during the previous cycle's
* settle. Late connector.reply from that cycle must not settle this one
* until its own inference.start arrives.
*/
awaitingContinuationInference: boolean;
/** Occupancy driver: collect+send when settle takes the deferred dry shot. */
/** Occupancy driver: collect+send when settle takes a dry-episode shot. */
dryOpenTaskDriver: (() => boolean) | undefined;
/** Last prompt actually sent — replay source for the quota auto-retry. */
lastSentMessage: string;
Expand Down Expand Up @@ -1011,19 +1012,17 @@ function drainLiveSteersAtBoundary(shell: AppShell, bag: BridgeBag): void {
}
}

function occupancyHold(bag: BridgeBag, runBusy: boolean): boolean {
if (bag.liveFleet > 0 || bag.awaitingContinuationInference) return true;
return runBusy && bag.pendingDryOpenDrive;
function occupancyHold(bag: BridgeBag): boolean {
return bag.liveFleet > 0 || bag.awaitingContinuationInference;
}

/**
* Release the run to idle and drain everything queued — but only at true
* session-idle. A live fleet holds the run busy after the parent turn settles
* (idle-with-fleet): Enter upgrades to a new primary turn during the hold and
* follow-ups keep waiting; the fleet event landing at zero re-enters here to
* release the hold. A latched dry edge with open tasks takes one occupancy
* shot here instead of idling, so a wentDry missed while processing cannot
* disagree with settle.
* release the hold. A dry fleet takes one occupancy shot here per dry episode
* instead of idling, even if the live 1→0 edge was never observed.
*/
function settleRunToIdle(shell: AppShell, bag: BridgeBag): void {
if (shell.session.run !== "busy") return;
Expand All @@ -1041,15 +1040,19 @@ function settleRunToIdle(shell: AppShell, bag: BridgeBag): void {
bag.flushPendingAskWake?.();
return;
}
if (bag.pendingDryOpenDrive) {
bag.pendingDryOpenDrive = false;
if (!bag.droveOpenTasksThisDry) {
let driven = false;
try {
driven = bag.dryOpenTaskDriver?.() === true;
} catch {
driven = false;
}
if (driven) return;
// Consume only on a real continuation. A false/no-op leaves the latch
// open so a later missed-edge settle with open tasks can still fire.
if (driven) {
bag.droveOpenTasksThisDry = true;
return;
}
}
shell.session = setRunState(shell.session, "idle");
bag.awaitingContinuationInference = false;
Expand All @@ -1074,12 +1077,9 @@ function applyInbound(
// queued follow-ups drain now. While the parent is still working the
// count just updates — the ordinary turn settle does the draining.
if (event.type === "fleet") {
const previous = bag.liveFleet;
bag.liveFleet = event.running;
if (event.running > 0) {
bag.pendingDryOpenDrive = false;
} else if (previous > 0) {
bag.pendingDryOpenDrive = true;
bag.droveOpenTasksThisDry = false;
}
if (event.running === 0 && !bag.turn.isProcessing) {
settleRunToIdle(shell, bag);
Expand Down Expand Up @@ -1185,7 +1185,7 @@ export function attachSessionBridge(
pendingAskWake: new Map(),
deliveredAskWake: new Map(),
flushPendingAskWake: null,
pendingDryOpenDrive: false,
droveOpenTasksThisDry: false,
awaitingContinuationInference: false,
dryOpenTaskDriver: undefined,
lastSentMessage: "",
Expand Down Expand Up @@ -1292,7 +1292,7 @@ export function attachSessionBridge(
currentToolName: turn.currentToolName,
streamingType: turn.streamingType,
nowMs,
sessionActive: occupancyHold(bag, shell.session.run === "busy"),
sessionActive: occupancyHold(bag),
};
const fleet = fleetProgress(bag.agentSessions, nowMs);
const label = resolveTurnLabel(input, isStalled, fleet);
Expand Down Expand Up @@ -1573,7 +1573,7 @@ export function attachSessionBridge(
bag.liveFleet = 0;
bag.pendingAskWake.clear();
bag.deliveredAskWake.clear();
bag.pendingDryOpenDrive = false;
bag.droveOpenTasksThisDry = false;
bag.awaitingContinuationInference = false;
bag.pendingRowUpdates.clear();
paintChrome(shell);
Expand Down Expand Up @@ -1746,7 +1746,7 @@ export function attachSessionBridge(
}
}
bag.awaitingContinuationInference = false;
bag.pendingDryOpenDrive = true;
bag.droveOpenTasksThisDry = false;
bag.lastSentMessage = "";
flushOpenRow(shell, bag);
bag.turnThinking = null;
Expand Down
Loading