diff --git a/src/agents/embedded-agent-runner/run/lane-controller.group-wait.test.ts b/src/agents/embedded-agent-runner/run/lane-controller.group-wait.test.ts new file mode 100644 index 000000000000..35af340797a4 --- /dev/null +++ b/src/agents/embedded-agent-runner/run/lane-controller.group-wait.test.ts @@ -0,0 +1,138 @@ +/** + * Group-blocked waits must be visible to the cron setup watchdog. + * + * Round-4 review (fiducian-spencer-001) asked for the 8-cron / hook-holding- + * reserve / 8th-cron-waiting regression asserting no false setup timeout. The + * chain spans three files: + * + * lane-controller.noteLaneWaitIfBusy -- emits onLaneWait({waiting:true}) + * -> timer-job-runner.noteLaneState -- maps it to the watchdog + * -> agent-watchdog.noteLaneWait() -- sets waitingForLane, clears timeout + * -> agent-watchdog:98 -- suppresses the setup timeout + * + * The watchdog end is already covered by agent-watchdog.test.ts. The link the + * capacity-group change introduced is the FIRST one, and it is the one that can + * silently fail: a group-blocked lane looks idle to a lane-local view, so the + * predicate returns false, no wait is ever reported, and a healthy run queued + * behind group capacity takes a false setup timeout. + * + * These tests drive the real predicate with real snapshots from a real group. + */ +import { afterEach, beforeEach, describe, expect, test } from "vitest"; +import { + clearCommandLaneGroup, + enqueueCommandInLane, + getCommandLaneSnapshot, + resetAllLanes, + setCommandLaneConcurrency, + setCommandLaneGroup, +} from "../../../process/command-queue.js"; +import { shouldNoteLaneWait } from "./lane-controller.js"; + +const CRON = "cron-nested"; +const HOOK = "hook-dispatch"; +const GROUP = "cron-hooks"; + +function gate() { + let release!: () => void; + const promise = new Promise((resolve) => { + release = resolve; + }); + return { promise, release }; +} + +async function settle(): Promise { + for (let i = 0; i < 5; i++) { + await new Promise((resolve) => setTimeout(resolve, 0)); + } +} + +beforeEach(() => { + resetAllLanes(); + clearCommandLaneGroup(GROUP); + setCommandLaneConcurrency(CRON, 8); + setCommandLaneConcurrency(HOOK, 1); + setCommandLaneGroup(GROUP, { + budget: 8, + members: [CRON, HOOK], + reservations: { [HOOK]: 1 }, + }); +}); + +afterEach(() => { + clearCommandLaneGroup(GROUP); + resetAllLanes(); +}); + +describe("group-blocked lane waits are reported", () => { + test("8th cron run blocked by the hook's reserve reports a wait", async () => { + // 7 cron active; the 8th slot is the hook's hard reservation. + const gates = Array.from({ length: 7 }, () => gate()); + const runs = gates.map((g) => enqueueCommandInLane(CRON, async () => await g.promise)); + await settle(); + + const snapshot = getCommandLaneSnapshot(CRON); + // This is the state that defeats a lane-local predicate: under its own + // maxConcurrent, nothing queued, yet unable to start. + expect(snapshot.activeCount).toBe(7); + expect(snapshot.maxConcurrent).toBe(8); + expect(snapshot.queuedCount).toBe(0); + expect(snapshot.queuedCount > 0 || snapshot.activeCount >= snapshot.maxConcurrent).toBe(false); + + // ...and the predicate must still report the wait, or the watchdog never + // suppresses its setup timeout and the run fails spuriously. + expect(shouldNoteLaneWait(snapshot)).toBe(true); + + for (const g of gates) g.release(); + await Promise.all(runs); + }); + + test("a hook blocked by group budget reports a wait", async () => { + // Fill the group entirely, including the hook's own reserved slot. + const cronGates = Array.from({ length: 7 }, () => gate()); + const cronRuns = cronGates.map((g) => enqueueCommandInLane(CRON, async () => await g.promise)); + const hookGate = gate(); + const hookRun = enqueueCommandInLane(HOOK, async () => await hookGate.promise); + await settle(); + + // A second hook cannot start: lane is one-wide AND the group is full. + expect(shouldNoteLaneWait(getCommandLaneSnapshot(HOOK))).toBe(true); + + hookGate.release(); + await hookRun; + for (const g of cronGates) g.release(); + await Promise.all(cronRuns); + }); + + test("no wait is reported when the lane can start immediately", async () => { + // The negative control. Without it, a predicate hardcoded to `true` would + // pass both tests above. + expect(shouldNoteLaneWait(getCommandLaneSnapshot(CRON))).toBe(false); + expect(shouldNoteLaneWait(getCommandLaneSnapshot(HOOK))).toBe(false); + + const g = gate(); + const run = enqueueCommandInLane(CRON, async () => await g.promise); + await settle(); + // One active out of eight: still admits, still no wait. + expect(shouldNoteLaneWait(getCommandLaneSnapshot(CRON))).toBe(false); + + g.release(); + await run; + }); + + test("waits are still reported for ordinary lane-local saturation", async () => { + // The pre-existing behaviour must survive the predicate change. + clearCommandLaneGroup(GROUP); + setCommandLaneConcurrency("ungrouped", 1); + const g = gate(); + const run = enqueueCommandInLane("ungrouped", async () => await g.promise); + await settle(); + + const snapshot = getCommandLaneSnapshot("ungrouped"); + expect(snapshot.activeCount).toBe(1); + expect(shouldNoteLaneWait(snapshot)).toBe(true); + + g.release(); + await run; + }); +}); diff --git a/src/agents/embedded-agent-runner/run/lane-controller.ts b/src/agents/embedded-agent-runner/run/lane-controller.ts index 832e58acc920..204e9f0c271d 100644 --- a/src/agents/embedded-agent-runner/run/lane-controller.ts +++ b/src/agents/embedded-agent-runner/run/lane-controller.ts @@ -6,6 +6,7 @@ import { withAgentRunLifecycleGeneration, } from "../../../infra/agent-events.js"; import { enqueueCommandInLane, getCommandLaneSnapshot } from "../../../process/command-queue.js"; +import type { CommandLaneSnapshot } from "../../../process/command-queue.js"; import type { CommandQueueEnqueueOptions } from "../../../process/command-queue.types.js"; import { withSessionPlacementTurnAdmission } from "../../session-placement-admission.js"; import type { EmbeddedAgentRunResult } from "../types.js"; @@ -18,6 +19,35 @@ import { import type { RunEmbeddedAgentParams } from "./params.js"; import { assertAgentHarnessRunAdmission } from "./session-bootstrap.js"; +/** + * Whether a run about to enter `lane` is going to wait rather than start now. + * + * Called BEFORE enqueue, so it must answer from the lane's current admission + * state — `queuedCount` is 0 at this point in the common case. + * + * `blockedBy` is the only term that can see a GROUP-imposed wait: a member + * blocked by group budget or a sibling's hard reservation has + * `activeCount < maxConcurrent` and typically `queuedCount === 0`, so both + * lane-local terms are false while the task genuinely cannot start. + * + * Missing that wait is not merely an observability gap. `cron/service/ + * agent-watchdog.ts` suppresses the cron setup timeout only while + * `waitingForLane` is true, and that flag is set from this signal via + * `timer-job-runner.ts` -> `noteLaneWait()`. A group wait that goes unreported + * therefore produces a FALSE setup timeout for a run that is healthy and simply + * queued behind capacity. + * + * Exported for test: the chain from group-blocked lane to timeout suppression + * spans three files, and this is the link the capacity-group change introduced. + */ +export function shouldNoteLaneWait(snapshot: CommandLaneSnapshot): boolean { + return ( + snapshot.queuedCount > 0 || + snapshot.activeCount >= snapshot.maxConcurrent || + snapshot.blockedBy != null + ); +} + type LaneParams = RunEmbeddedAgentParams & { sessionFile: string; }; @@ -90,19 +120,7 @@ export function createEmbeddedRunLaneController(opti return; } const snapshot = getCommandLaneSnapshot(lane); - // `blockedBy` is the only signal that can see a GROUP-imposed wait. A member - // blocked by group budget or a sibling's reservation has - // `activeCount < maxConcurrent` and can have `queuedCount === 0`, so the two - // lane-local terms below are both false while the task genuinely cannot - // start. Missing that wait is not merely an observability gap: it defeats - // the setup-timeout suppression in cron/service/agent-watchdog.ts, which - // engages only while `waitingForLane` is true — so a run waiting on group - // capacity would take a false setup timeout. - if ( - snapshot.queuedCount > 0 || - snapshot.activeCount >= snapshot.maxConcurrent || - snapshot.blockedBy != null - ) { + if (shouldNoteLaneWait(snapshot)) { params.onLaneWait({ waitMs: 0, queuedAhead: snapshot.queuedCount + snapshot.activeCount,