diff --git a/CHANGELOG.md b/CHANGELOG.md index 5da1cf80..2d1771ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,10 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename `advance_workflow` is gone. Already-complete and not-current ids are acknowledged without advancing. The unused `autoAdvance` workflow field is removed. +- `resume_agent(target, message)` starts the next turn on a retained completed + or interrupted worker and returns immediately. `wait_agents` collects the + reply. `send_input` steers only an in-flight running turn. Closed workers + stay closed. ### Fixed diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 2ff7a4a2..6e73c41b 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -130,7 +130,7 @@ The ChatDirector counts consecutive assistant turns that contain tool calls and #### Sub-agent stall management -`SubAgentDirector` tracks `lastActivityAt`, updated on every real `inference.done` and `tool.done`. Directors are pure `decide(event, ...)` functions with no timer of their own and the reactor has no proactive "idle" event, so a genuinely silent leaf (e.g. parked on a long-running background command with nothing else to do) produces no event for the director to react to. `runSubAgent` (`src/subagent/index.ts`) arms an external interval, at `subAgentStallTimeoutMs`, that pings the same content-less continuation channel the compaction governor uses to re-enter an idle reactor (`requestContinuation`). The director only acts on a ping if the elapsed time since `lastActivityAt` has crossed the timeout — a ping delivered while a tool call is still executing simply queues until that cycle finishes, so "no pending harness-tracked work" falls out of when the check can run at all rather than needing separate bookkeeping. The first stall past the timeout gets one continuation nudge (asking the leaf to check on the background work or report status); a second **consecutive** stall (no activity since that nudge) escalates to the existing salvage path, returning a `stalled` `forcedStopReport` with the same structured shape (summary/findings/blockers) as `turn-budget` and `cancelled`. Any real activity between pings resets the streak, so a leaf that is genuinely working through a slow single turn is never penalized. After the leaf has already replied with a terminal report (complete envelope or salvage), further empty continuations — idle-compact meter sync or stall pings — return `wait` instead of falling through to `DefaultDirector.infer`; only a non-empty parent message (`followup_task` / `send_input`) re-opens the brief. +`SubAgentDirector` tracks `lastActivityAt`, updated on every real `inference.done` and `tool.done`. Directors are pure `decide(event, ...)` functions with no timer of their own and the reactor has no proactive "idle" event, so a genuinely silent leaf (e.g. parked on a long-running background command with nothing else to do) produces no event for the director to react to. `runSubAgent` (`src/subagent/index.ts`) arms an external interval, at `subAgentStallTimeoutMs`, that pings the same content-less continuation channel the compaction governor uses to re-enter an idle reactor (`requestContinuation`). The director only acts on a ping if the elapsed time since `lastActivityAt` has crossed the timeout — a ping delivered while a tool call is still executing simply queues until that cycle finishes, so "no pending harness-tracked work" falls out of when the check can run at all rather than needing separate bookkeeping. The first stall past the timeout gets one continuation nudge (asking the leaf to check on the background work or report status); a second **consecutive** stall (no activity since that nudge) escalates to the existing salvage path, returning a `stalled` `forcedStopReport` with the same structured shape (summary/findings/blockers) as `turn-budget` and `cancelled`. Any real activity between pings resets the streak, so a leaf that is genuinely working through a slow single turn is never penalized. After the leaf has already replied with a terminal report (complete envelope or salvage), further empty continuations — idle-compact meter sync or stall pings — return `wait` instead of falling through to `DefaultDirector.infer`; only a non-empty parent message (`resume_agent` / `send_input`) re-opens the brief. **Intervention log**: every stop and nudge is appended as one JSONL record to `interventions.jsonl` in the firing leaf's trace dir (`src/subagent/intervention-log.ts`), carrying the trigger's measured value beside the threshold it crossed, the provider/model/family it fired on, and the run state at that moment (turns used vs budget, tool calls, read/edit counts). A refused parent re-dispatch is recorded on the parent side, where no leaf run exists to record it. The parent also appends one `outcome` record per completed dispatch — the salvage kind `classifyBriefSalvage` assigned, or a clean-complete marker, plus the dispatch count — so the log carries dispatch outcomes as well as interventions, and a stop record can later be read alongside what the dispatch it touched actually produced. Writes are fire-and-forget and swallow their own errors — a diagnostic must not be able to fail a run. `scripts/intervention-forensics.ts` aggregates these across local sessions: per-intervention counts by model family, the measured-value distribution against the threshold, two context columns (stops that fired on runs which had already edited files; stops that fired before half the turn budget was spent — neither is a measured false-positive rate, since either is equally consistent with a correct stop or a wrong one), and outcome counts by kind. This exists because every threshold in this tree was set by judgment and four of those judgments were later reverted — a threshold change is expected to cite this data (CL-6938). @@ -218,8 +218,8 @@ Every director package carries a required `tier: SubagentTier` field (`src/agent Enforcement is runtime code at the existing tool-mount point, not prompt wording — this is the fix for four prior mechanisms (`writePaths`, `report.requiredSections`, a `--config` comment, the thrash matcher) that were documented-as-enforced while enforcing nothing: -- **Mount-time gate — live today, and fails closed.** `task-tool.ts` resolves the caller's tier at dispatch time — a closed director's `DirectorPackage.tier` — and forwards it as `RunSubAgentParams.orchestratorTier`. `runSubAgent` (`src/subagent/run.ts`) then calls `assertTierMayMountFleetVerb(tier, toolName)` (`src/subagent/authority.ts`) before installing fleet verbs, treating a **missing** `orchestratorTier` as `"leaf"` — deny, not skip. This is the case that matters most: a project-local or plugin `AgentProfile` with `orchestrator: true` is outside the closed director set and is **not** trusted with fleet verbs just because `orchestrator: true` is set — there is no profile-level opt-in today, so the mount always throws `FleetAuthorityError` for a profile-sourced orchestrator. `FLEET_VERBS` in `authority.ts` names the live verbs (`task`, `spawn_agent`, `wait_agents`, `list_agents`, `send_input`, `interrupt_agent`, `close_agent`, `resume_agent`, `followup_task`, `read_agent_trace`, `search_agents`) so every mount site inherits the same gate. `list_agents` is the non-blocking mailbox-scoped list of this install's own `spawn_agent` workers (same scope as `wait_agents`); nested orchestrators may mount it. Fleet discovery (`search_agents`) remains Tier 1 only. -- **Subtree authority — wired for addressing verbs.** `assertCanTargetAgent(actor, targetId, nodes)` implements the "root owns its tree; a child manages only its own descendants" rule over the `{id, parentSessionId}` shape `SubAgentSessionStore` already tracks. Production call sites: `read_agent_trace`, `send_input`, `interrupt_agent`, `close_agent`, `resume_agent`, and `followup_task`. Nested mounts pass `{actorId, tier, getNodes}` from `run.ts`; a missing `actorId` fails closed. Tier-1 primary omits authority and stays unrestricted. `spawn_agent` records `parentSessionId` on nested workers so `close_agent`'s descendant walk can see them. `wait_agents` with omitted targets waits only on that caller's own `fleetRecords`, not every running session in the shared store. `list_agents` reports that same mailbox without blocking. `interrupt_agent` / `send_input` with `interrupt:true` terminalize the wait mailbox immediately; the soft-interrupt wait path collects so a later followup cannot resurrect an already-observed interrupt. `close_agent` also terminalizes the wait mailbox before teardown. +- **Mount-time gate — live today, and fails closed.** `task-tool.ts` resolves the caller's tier at dispatch time — a closed director's `DirectorPackage.tier` — and forwards it as `RunSubAgentParams.orchestratorTier`. `runSubAgent` (`src/subagent/run.ts`) then calls `assertTierMayMountFleetVerb(tier, toolName)` (`src/subagent/authority.ts`) before installing fleet verbs, treating a **missing** `orchestratorTier` as `"leaf"` — deny, not skip. This is the case that matters most: a project-local or plugin `AgentProfile` with `orchestrator: true` is outside the closed director set and is **not** trusted with fleet verbs just because `orchestrator: true` is set — there is no profile-level opt-in today, so the mount always throws `FleetAuthorityError` for a profile-sourced orchestrator. `FLEET_VERBS` in `authority.ts` names the live verbs (`task`, `spawn_agent`, `wait_agents`, `list_agents`, `send_input`, `interrupt_agent`, `close_agent`, `resume_agent`, `read_agent_trace`, `search_agents`) so every mount site inherits the same gate. `list_agents` is the non-blocking mailbox-scoped list of this install's own `spawn_agent` workers (same scope as `wait_agents`); nested orchestrators may mount it. Fleet discovery (`search_agents`) remains Tier 1 only. +- **Subtree authority — wired for addressing verbs.** `assertCanTargetAgent(actor, targetId, nodes)` implements the "root owns its tree; a child manages only its own descendants" rule over the `{id, parentSessionId}` shape `SubAgentSessionStore` already tracks. Production call sites: `read_agent_trace`, `send_input`, `interrupt_agent`, `close_agent`, and `resume_agent`. Nested mounts pass `{actorId, tier, getNodes}` from `run.ts`; a missing `actorId` fails closed. Tier-1 primary omits authority and stays unrestricted. `spawn_agent` records `parentSessionId` on nested workers so `close_agent`'s descendant walk can see them. `wait_agents` with omitted targets waits only on that caller's own `fleetRecords`, not every running session in the shared store. `list_agents` reports that same mailbox without blocking. `interrupt_agent` / `send_input` with `interrupt:true` terminalize the wait mailbox immediately; the soft-interrupt wait path collects so a later followup cannot resurrect an already-observed interrupt. `close_agent` also terminalizes the wait mailbox before teardown. - `task()` remains the deprecated fused spawn+wait fallback. `spawn_agent` + `wait_agents` is the supported parallel path. The tier check still gates which packages may mount any fleet verb. #### Closed director fleet (`src/agent/directors/`) diff --git a/src/agent/fleet-verbs-mount.test.ts b/src/agent/fleet-verbs-mount.test.ts index ac027483..f40a54a4 100644 --- a/src/agent/fleet-verbs-mount.test.ts +++ b/src/agent/fleet-verbs-mount.test.ts @@ -17,7 +17,6 @@ const FLEET_VERBS = [ "close_agent", "resume_agent", "interrupt_agent", - "followup_task", "send_input", ] as const; diff --git a/src/agent/tool-search.test.ts b/src/agent/tool-search.test.ts index d03baa27..4ca0c237 100644 --- a/src/agent/tool-search.test.ts +++ b/src/agent/tool-search.test.ts @@ -92,7 +92,6 @@ describe("createToolIndex", () => { "close_agent", "resume_agent", "interrupt_agent", - "followup_task", "send_input", ] as const) { expect(CORE_TOOL_NAMES).toContain(name); @@ -243,7 +242,6 @@ describe("advertisedTools", () => { "close_agent", "resume_agent", "interrupt_agent", - "followup_task", "send_input", ] as const) { expect(prefix).toContain(name); diff --git a/src/agent/tool-search.ts b/src/agent/tool-search.ts index acb1856f..ec6681e6 100644 --- a/src/agent/tool-search.ts +++ b/src/agent/tool-search.ts @@ -49,7 +49,6 @@ export const CORE_TOOL_NAMES: readonly string[] = [ "close_agent", "resume_agent", "interrupt_agent", - "followup_task", "send_input", ]; @@ -62,7 +61,6 @@ const ORCHESTRATOR_ONLY_TOOL_NAMES: readonly string[] = [ "close_agent", "resume_agent", "interrupt_agent", - "followup_task", "send_input", ]; diff --git a/src/agent/tools.ts b/src/agent/tools.ts index a28cd6fd..dcb3bc03 100644 --- a/src/agent/tools.ts +++ b/src/agent/tools.ts @@ -51,7 +51,6 @@ import { createCloseAgentTool, createResumeAgentTool, createInterruptAgentTool, - createFollowupTaskTool, createSendInputTool, } from "../subagent/lifecycle-tools.js"; import { parseManageTasksArgs } from "./tasks.js"; @@ -398,9 +397,8 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise { // the store itself had already evicted and released the earliest // session, because a retained session shared the 20-item display cap // with every other finished session — exactly the shipped defect this - // ticket fixes (resume_agent/followup_task failed with a bare + // ticket fixes (resume_agent failed with a bare // "not_found" past 20 spawned workers, blaming the caller for nothing). // Open retained sessions now have their own cap (`maxRetained`, default // 50), so 25 of them all stay resumable; fleetRecords/wait_agents is diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index b52881b9..26d18aed 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -704,7 +704,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { // store a bounded close for close_agent to call later. // Worktree cleanup is deferred until that close when the session // stays alive for followup (agentRetained / interrupt keep-alive) — - // matching run.ts's persisting gate so followup_task does not hit a + // matching run.ts's persisting gate so resume_agent does not hit a // removed cwd. persist: deps.persist !== false, onAgentReady: ({ close, interrupt, followup, deliver }) => { @@ -802,7 +802,7 @@ interface WaitAgentsDeps { function isSoftInterrupted( session: ReturnType, ): session is NonNullable> { - // interrupt_agent keeps strip status "running" so followup_task can reuse + // interrupt_agent keeps strip status "running" so resume_agent can reuse // the session. cancel() also sets lifecycleStatus "interrupted" but flips // status to "cancelled" — that path still owes wait_agents a salvage // report via fleetRecords, so it is not wait-terminal on its own. diff --git a/src/subagent/authority.test.ts b/src/subagent/authority.test.ts index 01cc486d..3bcf5b25 100644 --- a/src/subagent/authority.test.ts +++ b/src/subagent/authority.test.ts @@ -14,11 +14,11 @@ describe("assertTierMayMountFleetVerb", () => { // The reusable-session verbs are gated the same way. expect(() => assertTierMayMountFleetVerb("leaf", "close_agent")).toThrow(FleetAuthorityError); expect(() => assertTierMayMountFleetVerb("leaf", "resume_agent")).toThrow(FleetAuthorityError); - // Interrupt_agent / followup_task are gated the same way. + // Interrupt_agent / send_input are gated the same way. expect(() => assertTierMayMountFleetVerb("leaf", "interrupt_agent")).toThrow( FleetAuthorityError, ); - expect(() => assertTierMayMountFleetVerb("leaf", "followup_task")).toThrow(FleetAuthorityError); + expect(() => assertTierMayMountFleetVerb("leaf", "send_input")).toThrow(FleetAuthorityError); }); test("leaves may still mount non-fleet tools", () => { diff --git a/src/subagent/authority.ts b/src/subagent/authority.ts index 33fbfbd4..4cec891e 100644 --- a/src/subagent/authority.ts +++ b/src/subagent/authority.ts @@ -6,7 +6,7 @@ * * - assertTierMayMountFleetVerb: a Tier 3 leaf may never mount a fleet verb * (task, spawn_agent, wait_agents, list_agents, interrupt_agent, close_agent, - * resume_agent, followup_task, send_input, read_agent_trace, search_agents). + * resume_agent, send_input, read_agent_trace, search_agents). * Fleet *discovery* of the director catalog (search_agents) is Tier 1 only * (CL-7051). list_agents is not catalog discovery — it lists this install's * own spawn_agent workers, the same scoped mailbox wait_agents uses, so @@ -38,7 +38,6 @@ export const FLEET_VERBS = new Set([ "close_agent", "resume_agent", "read_agent_trace", - "followup_task", ]); /** @@ -114,7 +113,7 @@ function isDescendant( * always fails closed here too. * * Production call sites: `read_agent_trace`, `send_input`, `interrupt_agent`, - * `close_agent`, `resume_agent`, and `followup_task` (nested mounts pass + * `close_agent`, and `resume_agent` (nested mounts pass * authority from run.ts; Tier-1 primary omits it and stays unrestricted). */ export function assertCanTargetAgent( diff --git a/src/subagent/followup-live-agent.test.ts b/src/subagent/followup-live-agent.test.ts index 0472b9b2..522df76a 100644 --- a/src/subagent/followup-live-agent.test.ts +++ b/src/subagent/followup-live-agent.test.ts @@ -1,10 +1,10 @@ /** * Regression guard: lifecycle-tools.test.ts proves interrupt_agent / - * followup_task behave correctly against *fake registered closures* at the + * resume_agent behave correctly against *fake registered closures* at the * tool/store layer — it never exercises run.ts's real wiring, where * `followup` calls `agent!.send()` on the same live agent object created by * `createAgentWithLiveToolDispatch`. A future refactor could make - * `followup_task` rebuild the agent instead of reusing it (exactly the + * `resume_agent` rebuild the agent instead of reusing it (exactly the * regression this feature exists to prevent — a rebuilt agent means the * worker re-reads the codebase from scratch) without failing any existing * test. @@ -39,32 +39,41 @@ async function tmpCwd(): Promise { * to prove reuse: `sendLog` accumulates every message across BOTH the * original send and the later followup send, and rejects like the real * `Agent.send`'s documented `signal` option when its signal fires. */ -function createStubAgent() { +function createStubAgent(opts?: { hangFromSend?: number }) { const sendLog: string[] = []; + const abortedSends: boolean[] = []; return { sendLog, - async send(content: string, opts?: { signal?: AbortSignal }) { + abortedSends, + async send(content: string, optsSend?: { signal?: AbortSignal }) { sendLog.push(content); + const index = sendLog.length - 1; + abortedSends[index] = false; return await new Promise((resolve, reject) => { - if (opts?.signal?.aborted === true) { - reject(opts.signal.reason instanceof Error ? opts.signal.reason : new Error("aborted")); + const abort = (reason: unknown) => { + abortedSends[index] = true; + reject(reason instanceof Error ? reason : new Error("aborted")); + }; + if (optsSend?.signal?.aborted === true) { + abort(optsSend.signal.reason); return; } - const timer = setTimeout( - () => - resolve({ - reply: `reply #${sendLog.length}`, - turn: { role: "assistant", content: [] }, - }), - 20, - ); - opts?.signal?.addEventListener( + const hang = opts?.hangFromSend !== undefined && sendLog.length >= opts.hangFromSend; + const timer = hang + ? undefined + : setTimeout( + () => + resolve({ + reply: `reply #${sendLog.length}`, + turn: { role: "assistant", content: [] }, + }), + 20, + ); + optsSend?.signal?.addEventListener( "abort", () => { - clearTimeout(timer); - reject( - opts.signal!.reason instanceof Error ? opts.signal!.reason : new Error("aborted"), - ); + if (timer !== undefined) clearTimeout(timer); + abort(optsSend.signal!.reason); }, { once: true }, ); @@ -82,7 +91,7 @@ function createStubAgent() { }; } -describe("interrupt_agent / followup_task reuse the same live agent", () => { +describe("interrupt_agent / resume_agent reuse the same live agent", () => { test("followup after interrupt sends into the SAME agent instance — not a rebuilt one", async () => { const cwd = await tmpCwd(); let constructions = 0; @@ -157,4 +166,75 @@ describe("interrupt_agent / followup_task reuse the same live agent", () => { expect(capturedAgent!.sendLog[1]).toBe("do X instead, not what the original prompt said"); expect(outcome.reply).toBe("reply #2"); }); + + test("interrupt_agent aborts the resumed followup agent.send", async () => { + const cwd = await tmpCwd(); + let constructions = 0; + let capturedAgent: ReturnType | undefined; + + const outcome = await withMockedModuleDuring( + import.meta.resolve("../agent/live-tool-dispatch.js"), + (real: typeof import("../agent/live-tool-dispatch.js")) => ({ + ...real, + createAgentWithLiveToolDispatch: async () => { + constructions++; + const stub = createStubAgent({ hangFromSend: 2 }); + capturedAgent = stub; + return stub as unknown as Awaited< + ReturnType + >; + }, + }), + async () => { + const { runSubAgent } = await import("./run.js"); + + let handles: + | { + close: (ms?: number) => Promise; + interrupt: () => void; + followup: (message: string) => Promise; + } + | undefined; + + const params: RunSubAgentParams = { + cwd, + workdirBase: join(cwd, ".ctx"), + permissionGate: testPermissionGate, + provider: { providerName: "test", baseURL: "http://localhost", model: "test-model" }, + description: "live-agent followup interrupt probe", + prompt: "finish the first turn", + persist: true, + onAgentReady: (h) => { + handles = h; + }, + }; + + const first = await runSubAgent(params); + if (handles === undefined) throw new Error("onAgentReady never fired"); + + const followupPromise = handles.followup("now do the second turn"); + for (let i = 0; i < 500 && (capturedAgent?.sendLog.length ?? 0) < 2; i++) { + await new Promise((resolve) => setTimeout(resolve, 1)); + } + handles.interrupt(); + const followup = await followupPromise.then( + (reply) => ({ ok: true as const, reply }), + (err: unknown) => ({ + ok: false as const, + message: err instanceof Error ? err.message : String(err), + }), + ); + return { first, followup }; + }, + ); + + expect(outcome.first.agentRetained).toBe(true); + expect(constructions).toBe(1); + expect(capturedAgent?.sendLog.length).toBe(2); + expect(capturedAgent?.sendLog[1]).toBe("now do the second turn"); + expect(capturedAgent?.abortedSends[1]).toBe(true); + expect(outcome.followup.ok).toBe(false); + if (outcome.followup.ok) throw new Error("expected followup send to abort"); + expect(outcome.followup.message).toContain("interrupted by interrupt_agent"); + }); }); diff --git a/src/subagent/lifecycle-tools.test.ts b/src/subagent/lifecycle-tools.test.ts index 34f3a088..8f38e403 100644 --- a/src/subagent/lifecycle-tools.test.ts +++ b/src/subagent/lifecycle-tools.test.ts @@ -4,10 +4,9 @@ import { createCloseAgentTool, createResumeAgentTool, createInterruptAgentTool, - createFollowupTaskTool, createSendInputTool, } from "./lifecycle-tools.js"; -import { createFleetRecords } from "./agent-fleet.js"; +import { createFleetRecords, createWaitAgentsTool } from "./agent-fleet.js"; import { createSubAgentSessionStore } from "./session-store.js"; async function callTool( @@ -15,8 +14,8 @@ async function callTool( | ReturnType | ReturnType | ReturnType - | ReturnType - | ReturnType, + | ReturnType + | ReturnType, args: Record, ): Promise> { if (tool.kind !== "full") throw new Error(`expected full tool, got ${tool.kind}`); @@ -89,34 +88,66 @@ describe("close_agent", () => { }); describe("resume_agent", () => { - test("resumes a retained completed session and rejects a non-retained one", async () => { + test("starts the next turn on a completed retained session and returns immediately", async () => { const sessions = createSubAgentSessionStore(); + const fleetRecords = createFleetRecords(); const retained = sessions.start({ description: "d", agentId: "a", brief: "b", retained: true }); + const history: string[] = ["first task"]; + let finish: (reply: string) => void = () => {}; + sessions.registerFollowup( + retained.id, + (message: string) => + new Promise((resolve) => { + history.push(message); + finish = resolve; + }), + ); sessions.complete(retained.id, "## Summary\nDone."); const notRetained = sessions.start({ description: "d2", agentId: "a", brief: "b" }); sessions.complete(notRetained.id, "## Summary\nDone."); - const resumeAgent = createResumeAgentTool({ sessions }); + const resumeAgent = createResumeAgentTool({ sessions, fleetRecords }); - const ok = await callTool(resumeAgent, { target: retained.id }); + const started = Date.now(); + const ok = await callTool(resumeAgent, { target: retained.id, message: "now do task two" }); + expect(Date.now() - started).toBeLessThan(1000); expect(ok.status).toBe("running"); + expect(sessions.get(retained.id)?.status).toBe("running"); expect(sessions.get(retained.id)?.lifecycleStatus).toBe("running"); + expect(history).toEqual(["first task", "now do task two"]); + + sessions.registerDeliver(retained.id, () => {}); + sessions.registerInterrupt(retained.id, () => {}); + const sendInput = createSendInputTool({ sessions }); + const steered = await callTool(sendInput, { target: retained.id, message: "steer" }); + expect(steered).toEqual({ agent_id: retained.id, status: "running" }); + const interrupted = await callTool(createInterruptAgentTool({ sessions, fleetRecords }), { + target: retained.id, + }); + expect(interrupted.status).toBe("interrupted"); + + finish("done, history now 2 turns"); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(sessions.get(retained.id)?.lifecycleStatus).toBe("completed"); + expect(sessions.get(retained.id)?.id).toBe(retained.id); + expect(sessions.get(retained.id)?.report).toBe("done, history now 2 turns"); - const rawResult = await (async () => { - if (resumeAgent.kind !== "full") throw new Error("expected full tool"); - return resumeAgent.handler( - { id: "call-x", name: "resume_agent", arguments: { target: notRetained.id } }, - new AbortController().signal, - ); - })(); - expect(rawResult.isError).toBe(true); + if (resumeAgent.kind !== "full") throw new Error("expected full tool"); + const rejected = await resumeAgent.handler( + { + id: "call-x", + name: "resume_agent", + arguments: { target: notRetained.id, message: "more" }, + }, + new AbortController().signal, + ); + expect(rejected.isError).toBe(true); }); -}); -describe("interrupt_agent / followup_task", () => { - test("interrupt then followup keeps prior context — the worker does not re-read from scratch", async () => { + test("resumes an interrupted retained session without calling close()", async () => { const sessions = createSubAgentSessionStore(); + const fleetRecords = createFleetRecords(); const worker = sessions.start({ description: "worker", agentId: "a", @@ -125,11 +156,12 @@ describe("interrupt_agent / followup_task", () => { }); sessions.markRunning(worker.id); - // Simulates the live agent's own message history (what run.ts's - // `followup`/`interrupt` closures actually close over) — a shared array, - // not something recreated per call. const history: string[] = ["read src/index.ts", "found the bug on line 12"]; let interruptFired = false; + let closeCalls = 0; + sessions.registerClose(worker.id, async () => { + closeCalls++; + }); sessions.registerInterrupt(worker.id, () => { interruptFired = true; }); @@ -138,112 +170,136 @@ describe("interrupt_agent / followup_task", () => { return `Applying fix given ${history.length} prior turns of context.`; }); - const interruptAgent = createInterruptAgentTool({ - sessions, - fleetRecords: createFleetRecords(), - }); - const followupTask = createFollowupTaskTool({ sessions }); + const interruptAgent = createInterruptAgentTool({ sessions, fleetRecords }); + const resumeAgent = createResumeAgentTool({ sessions, fleetRecords }); const interruptResult = await callTool(interruptAgent, { target: worker.id }); expect(interruptResult.status).toBe("interrupted"); expect(interruptFired).toBe(true); + expect(closeCalls).toBe(0); expect(sessions.get(worker.id)?.lifecycleStatus).toBe("interrupted"); - const followupResult = await callTool(followupTask, { + const started = Date.now(); + const resumeResult = await callTool(resumeAgent, { target: worker.id, message: "actually fix line 12 directly, not line 20", }); - expect(followupResult.status).toBe("completed"); - - // The load-bearing assertion: the worker's own history object still - // holds the turns that predate the interrupt, plus the new one appended - // in place — not a fresh array the followup started from empty. + expect(Date.now() - started).toBeLessThan(1000); + expect(resumeResult.status).toBe("running"); + expect(closeCalls).toBe(0); expect(history).toEqual([ "read src/index.ts", "found the bug on line 12", "actually fix line 12 directly, not line 20", ]); - expect(history.length).toBe(3); + + await new Promise((resolve) => setTimeout(resolve, 0)); expect(sessions.get(worker.id)?.lifecycleStatus).toBe("completed"); - expect(sessions.get(worker.id)?.report).toBe(followupResult.reply as string); + expect(sessions.get(worker.id)?.report).toContain("Applying fix given 3 prior turns"); }); - test("followup_task on a completed retained worker reuses its existing session, not a fresh one", async () => { + test("rejects a closed session and a concurrent resume of a running turn", async () => { const sessions = createSubAgentSessionStore(); - const worker = sessions.start({ - description: "worker", + const fleetRecords = createFleetRecords(); + const closed = sessions.start({ + description: "closed", agentId: "a", brief: "b", retained: true, }); - const history: string[] = ["did the first task"]; - sessions.registerFollowup(worker.id, async (message: string) => { - history.push(message); - return `done, history now ${history.length} turns`; - }); - sessions.complete(worker.id, "## Summary\nFirst task done."); + sessions.registerClose(closed.id, async () => {}); + sessions.registerFollowup(closed.id, async () => "should not run"); + sessions.complete(closed.id, "## Summary\nDone."); + const closeAgent = createCloseAgentTool({ sessions, fleetRecords }); + await callTool(closeAgent, { target: closed.id }); - const followupTask = createFollowupTaskTool({ sessions }); - const result = await callTool(followupTask, { target: worker.id, message: "now do task two" }); + const resumeAgent = createResumeAgentTool({ sessions, fleetRecords }); + if (resumeAgent.kind !== "full") throw new Error("expected full tool"); + const closedErr = await resumeAgent.handler( + { id: "c-closed", name: "resume_agent", arguments: { target: closed.id, message: "more" } }, + new AbortController().signal, + ); + expect(closedErr.isError).toBe(true); + expect(String(closedErr.content)).toContain("shutdown"); - expect(result.status).toBe("completed"); - // Same session id throughout — never re-created — and its underlying - // history object grew rather than being replaced. - expect(sessions.get(worker.id)?.id).toBe(worker.id); - expect(history).toEqual(["did the first task", "now do task two"]); + const worker = sessions.start({ + description: "worker", + agentId: "a", + brief: "b", + retained: true, + }); + let finish: (reply: string) => void = () => {}; + sessions.registerFollowup( + worker.id, + () => + new Promise((resolve) => { + finish = resolve; + }), + ); + sessions.complete(worker.id, "## Summary\nDone."); - const nonRetained = sessions.start({ description: "d2", agentId: "a", brief: "b" }); - sessions.complete(nonRetained.id, "## Summary\nDone."); - if (followupTask.kind !== "full") throw new Error("expected full tool"); - const rejected = await followupTask.handler( + const first = await callTool(resumeAgent, { target: worker.id, message: "turn two" }); + expect(first.status).toBe("running"); + const concurrent = await resumeAgent.handler( { - id: "c3", - name: "followup_task", - arguments: { target: nonRetained.id, message: "more work" }, + id: "c-concurrent", + name: "resume_agent", + arguments: { target: worker.id, message: "again" }, }, new AbortController().signal, ); - expect(rejected.isError).toBe(true); + expect(concurrent.isError).toBe(true); + expect(String(concurrent.content)).toContain("running"); + finish("done"); }); - test("an interrupted session is resumable via followup_task and interrupt never touches close()", async () => { + test("wait_agents collects the resumed turn after resume_agent returns", async () => { const sessions = createSubAgentSessionStore(); + const fleetRecords = createFleetRecords(); const worker = sessions.start({ description: "worker", agentId: "a", brief: "b", retained: true, }); - sessions.markRunning(worker.id); - - let closeCalls = 0; - sessions.registerClose(worker.id, async () => { - closeCalls++; - }); - sessions.registerInterrupt(worker.id, () => { - // Real interrupt handle: fires a dedicated signal, never close(). - }); - sessions.registerFollowup(worker.id, async () => "resumed cleanly"); + let finish: (reply: string) => void = () => {}; + sessions.registerFollowup( + worker.id, + () => + new Promise((resolve) => { + finish = resolve; + }), + ); + sessions.complete(worker.id, "first report"); + fleetRecords.register(worker.id); + fleetRecords.resolve(worker.id, "first report"); - const interruptAgent = createInterruptAgentTool({ - sessions, - fleetRecords: createFleetRecords(), - }); - const followupTask = createFollowupTaskTool({ sessions }); + const resumeAgent = createResumeAgentTool({ sessions, fleetRecords }); + const wait = createWaitAgentsTool({ sessions, fleetRecords }); - await callTool(interruptAgent, { target: worker.id }); - expect(closeCalls).toBe(0); + const firstWait = await callTool(wait, { targets: [worker.id], timeout_ms: 1000 }); + expect(firstWait.timed_out).toBe(false); + const firstResults = firstWait.results as { status: string; report?: string }[]; + expect(firstResults[0]!.status).toBe("done"); + expect(firstResults[0]!.report).toBe("first report"); - const followupResult = await callTool(followupTask, { target: worker.id, message: "continue" }); - expect(followupResult.status).toBe("completed"); - expect(closeCalls).toBe(0); - // No lock-strand risk from this path: close() was never invoked, so the - // workdir lock close_agent's bounded teardown would otherwise release - // was never at risk of being held by a wedged close in the first place. - expect(sessions.get(worker.id)?.lifecycleStatus).toBe("completed"); + const started = Date.now(); + const resumed = await callTool(resumeAgent, { target: worker.id, message: "second turn" }); + expect(Date.now() - started).toBeLessThan(1000); + expect(resumed.status).toBe("running"); + + const waiting = callTool(wait, { targets: [worker.id], timeout_ms: 2000 }); + finish("second report"); + const collected = await waiting; + expect(collected.timed_out).toBe(false); + const results = collected.results as { status: string; report?: string }[]; + expect(results[0]!.status).toBe("done"); + expect(results[0]!.report).toBe("second report"); }); +}); - test("interrupt_agent and followup_task fail closed on a non-running / non-retained target", async () => { +describe("interrupt_agent", () => { + test("interrupt_agent fails closed on a non-running target", async () => { const sessions = createSubAgentSessionStore(); const notRunning = sessions.start({ description: "d", agentId: "a", brief: "b" }); sessions.complete(notRunning.id, "## Summary\nDone."); @@ -252,7 +308,6 @@ describe("interrupt_agent / followup_task", () => { sessions, fleetRecords: createFleetRecords(), }); - const followupTask = createFollowupTaskTool({ sessions }); if (interruptAgent.kind !== "full") throw new Error("expected full tool"); const interruptErr = await interruptAgent.handler( @@ -260,14 +315,6 @@ describe("interrupt_agent / followup_task", () => { new AbortController().signal, ); expect(interruptErr.isError).toBe(true); - - if (followupTask.kind !== "full") throw new Error("expected full tool"); - const followupErr = await followupTask.handler( - { id: "c2", name: "followup_task", arguments: { target: notRunning.id, message: "x" } }, - new AbortController().signal, - ); - // Not retained, so followup_task must reject even though it is "completed". - expect(followupErr.isError).toBe(true); }); }); @@ -354,6 +401,73 @@ describe("send_input", () => { expect(sessions.get(missing.id)?.lifecycleStatus).toBe("running"); }); + test("rejects completed, interrupted, and closed sessions — steering is in-flight only", async () => { + const sessions = createSubAgentSessionStore(); + const fleetRecords = createFleetRecords(); + const sendInput = createSendInputTool({ sessions, fleetRecords }); + if (sendInput.kind !== "full") throw new Error("expected full tool"); + + const completed = sessions.start({ + description: "done", + agentId: "a", + brief: "b", + retained: true, + }); + sessions.markRunning(completed.id); + sessions.registerDeliver(completed.id, () => { + throw new Error("must not deliver to a completed session"); + }); + sessions.complete(completed.id, "## Summary\nDone."); + const completedErr = await sendInput.handler( + { id: "to-completed", name: "send_input", arguments: { target: completed.id, message: "x" } }, + new AbortController().signal, + ); + expect(completedErr.isError).toBe(true); + + const interrupted = sessions.start({ + description: "paused", + agentId: "a", + brief: "b", + retained: true, + }); + sessions.markRunning(interrupted.id); + sessions.registerInterrupt(interrupted.id, () => {}); + sessions.registerDeliver(interrupted.id, () => { + throw new Error("must not deliver to an interrupted session"); + }); + await callTool(createInterruptAgentTool({ sessions, fleetRecords }), { + target: interrupted.id, + }); + const interruptedErr = await sendInput.handler( + { + id: "to-interrupted", + name: "send_input", + arguments: { target: interrupted.id, message: "x" }, + }, + new AbortController().signal, + ); + expect(interruptedErr.isError).toBe(true); + + const closed = sessions.start({ + description: "closed", + agentId: "a", + brief: "b", + retained: true, + }); + sessions.markRunning(closed.id); + sessions.registerClose(closed.id, async () => {}); + sessions.registerDeliver(closed.id, () => { + throw new Error("must not deliver to a closed session"); + }); + await callTool(createCloseAgentTool({ sessions, fleetRecords }), { target: closed.id }); + const closedErr = await sendInput.handler( + { id: "to-closed", name: "send_input", arguments: { target: closed.id, message: "x" } }, + new AbortController().signal, + ); + expect(closedErr.isError).toBe(true); + expect(sessions.get(closed.id)?.lifecycleStatus).toBe("shutdown"); + }); + test("enforces nested orchestrator descendant authority", async () => { const sessions = createSubAgentSessionStore(); const nested = sessions.start({ @@ -492,7 +606,7 @@ describe("nested lifecycle authority", () => { expect(sessions.get(sibling.id)?.lifecycleStatus).not.toBe("shutdown"); }); - test("followup_task denies a sibling and allows a descendant", async () => { + test("resume_agent denies a sibling and allows a descendant", async () => { const sessions = createSubAgentSessionStore(); const nested = sessions.start({ id: "nested", description: "n", agentId: "a", brief: "b" }); const child = sessions.start({ @@ -514,53 +628,15 @@ describe("nested lifecycle authority", () => { sessions.complete(s.id, "done"); sessions.registerFollowup(s.id, async () => "reply"); } - const followup = createFollowupTaskTool({ - sessions, - authority: nestAuthority(sessions, nested.id), - }); - expect((await callTool(followup, { target: child.id, message: "more" })).status).toBe( - "completed", - ); - if (followup.kind !== "full") throw new Error("expected full tool"); - const denied = await followup.handler( - { - id: "d", - name: "followup_task", - arguments: { target: sibling.id, message: "more" }, - }, - new AbortController().signal, - ); - expect(denied.isError).toBe(true); - }); - - test("resume_agent denies a sibling and allows a descendant", async () => { - const sessions = createSubAgentSessionStore(); - const nested = sessions.start({ id: "nested", description: "n", agentId: "a", brief: "b" }); - const child = sessions.start({ - id: "child", - description: "c", - agentId: "a", - brief: "b", - parentSessionId: nested.id, - retained: true, - }); - const sibling = sessions.start({ - id: "sibling", - description: "s", - agentId: "a", - brief: "b", - retained: true, - }); - sessions.complete(child.id, "done"); - sessions.complete(sibling.id, "done"); const resume = createResumeAgentTool({ sessions, + fleetRecords: createFleetRecords(), authority: nestAuthority(sessions, nested.id), }); - expect((await callTool(resume, { target: child.id })).status).toBe("running"); + expect((await callTool(resume, { target: child.id, message: "more" })).status).toBe("running"); if (resume.kind !== "full") throw new Error("expected full tool"); const denied = await resume.handler( - { id: "d", name: "resume_agent", arguments: { target: sibling.id } }, + { id: "d", name: "resume_agent", arguments: { target: sibling.id, message: "more" } }, new AbortController().signal, ); expect(denied.isError).toBe(true); diff --git a/src/subagent/lifecycle-tools.ts b/src/subagent/lifecycle-tools.ts index a0adbdf8..d50130b3 100644 --- a/src/subagent/lifecycle-tools.ts +++ b/src/subagent/lifecycle-tools.ts @@ -2,13 +2,9 @@ * close_agent / resume_agent: the session-lifecycle half of * reusable worker sessions. spawn_agent/wait_agents start and * collect workers; these two verbs let an orchestrator tear one down on - * purpose (close_agent) or bring a retained one back for further input - * (resume_agent), instead of every session dying the instant its turn ends. - * - * interrupt_agent and followup_task (the verbs that actually push a new - * prompt into a resumed session) are a separate, later change — resume_agent - * here only flips a retained session back to an addressable state; it takes - * no prompt argument. + * purpose (close_agent) or start the next turn on a retained completed + * or interrupted session (resume_agent), returning immediately so + * wait_agents collects. send_input steers an in-flight running turn. */ import { tool } from "@intx/agent"; @@ -58,20 +54,26 @@ export const closeAgentToolDefinition: ToolDefinition = { const ResumeAgentArgs = type({ target: "string", + message: "string", }); export const resumeAgentToolDefinition: ToolDefinition = { name: "resume_agent", description: - "Reopen a retained, completed worker session (one that finished a turn and was never closed) " + - "so it is addressable again. Fails on a session that is still running, was never retained, was " + - "interrupted, or was already closed via close_agent (closing is permanent).", + "Start the next turn on a retained worker that is 'completed' or 'interrupted', reusing its " + + "prior context rather than spawning a fresh worker. Returns immediately with status 'running'; " + + "collect the reply with wait_agents. Fails on a session that is still running, was never " + + "retained, or was already closed via close_agent (closing is permanent).", inputSchema: { type: "object", properties: { - target: { type: "string", description: "agent_id of the session to resume." }, + target: { type: "string", description: "agent_id of the retained session to resume." }, + message: { + type: "string", + description: `The new instruction/message for the worker (non-empty, max ${DEFAULT_MAX_ENTRY_CHARS} characters).`, + }, }, - required: ["target"], + required: ["target", "message"], }, }; @@ -109,7 +111,7 @@ export interface LifecycleAuthority { export interface LifecycleToolDeps { sessions: SubAgentSessionStore; - /** Optional for resume/followup/send_input; close and interrupt require it (see CloseAgentToolDeps / InterruptAgentToolDeps). */ + /** Optional for send_input; close, interrupt, and resume require it. */ fleetRecords?: FleetRecordsHandle; authority?: LifecycleAuthority; } @@ -124,6 +126,11 @@ export type InterruptAgentToolDeps = LifecycleToolDeps & { fleetRecords: FleetRecordsHandle; }; +/** resume_agent registers the next turn on the wait mailbox so wait_agents can collect. */ +export type ResumeAgentToolDeps = LifecycleToolDeps & { + fleetRecords: FleetRecordsHandle; +}; + function gateTarget( deps: LifecycleToolDeps, toolName: string, @@ -197,7 +204,7 @@ export function createCloseAgentTool(deps: CloseAgentToolDeps): AgentTool { }); } -export function createResumeAgentTool(deps: LifecycleToolDeps): AgentTool { +export function createResumeAgentTool(deps: ResumeAgentToolDeps): AgentTool { return tool({ definition: resumeAgentToolDefinition, handler: async (call, _signal): Promise => { @@ -208,7 +215,28 @@ export function createResumeAgentTool(deps: LifecycleToolDeps): AgentTool { const target = parsed.target.trim(); const denied = gateTarget(deps, "resume_agent", target, call.id); if (denied !== undefined) return denied; - const outcome = deps.sessions.resumeOne(target); + const message = parsed.message.trim(); + if (message.length === 0) { + return lifecycleResult(call.id, "Error: resume_agent requires a non-empty message."); + } + if (message.length > DEFAULT_MAX_ENTRY_CHARS) { + return lifecycleResult( + call.id, + `Error: resume_agent message exceeds ${DEFAULT_MAX_ENTRY_CHARS} characters ` + + `(got ${message.length}).`, + ); + } + const outcome = deps.sessions.resumeOne(target, message, { + onStart: () => { + deps.fleetRecords.register(target); + }, + onReply: (reply) => { + deps.fleetRecords.resolve(target, reply); + }, + onFail: (err) => { + deps.fleetRecords.reject(target, err instanceof Error ? err.message : String(err)); + }, + }); if (!outcome.ok) { const hint = outcome.hint !== undefined ? ` ${outcome.hint}` : ""; return lifecycleResult( @@ -233,7 +261,7 @@ export const interruptAgentToolDefinition: ToolDefinition = { "on this id immediately with status 'interrupted'. The worker's in-flight tool call or " + "inference keeps running in the background (there is no way to hard-stop it without tearing the " + "session down); this only stops the caller from waiting on it and marks the session " + - "'interrupted' so followup_task or resume_agent can pick it back up with full prior context. " + + "'interrupted' so resume_agent can pick it back up with full prior context. " + "Fails on a session that is not currently running.", inputSchema: { type: "object", @@ -276,62 +304,6 @@ export function createInterruptAgentTool(deps: InterruptAgentToolDeps): AgentToo }); } -const FollowupTaskArgs = type({ - target: "string", - message: "string", -}); - -export const followupTaskToolDefinition: ToolDefinition = { - name: "followup_task", - description: - "Send new work into an existing retained worker session (one that is 'completed' or " + - "'interrupted'), reusing its prior context and tool outputs rather than starting a fresh worker. " + - "Blocks until the worker replies to this new message, and returns its reply. Fails on a session " + - "that was never retained, is still running, or was closed via close_agent (closing is permanent).", - inputSchema: { - type: "object", - properties: { - target: { type: "string", description: "agent_id of the retained session to resume." }, - message: { type: "string", description: "The new instruction/message for the worker." }, - }, - required: ["target", "message"], - }, -}; - -export function createFollowupTaskTool(deps: LifecycleToolDeps): AgentTool { - return tool({ - definition: followupTaskToolDefinition, - handler: async (call, _signal): Promise => { - const parsed = FollowupTaskArgs(call.arguments); - if (parsed instanceof type.errors) { - return lifecycleResult( - call.id, - `Error: followup_task arguments invalid: ${parsed.summary}`, - ); - } - const target = parsed.target.trim(); - const denied = gateTarget(deps, "followup_task", target, call.id); - if (denied !== undefined) return denied; - const message = parsed.message.trim(); - if (message.length === 0) { - return lifecycleResult(call.id, "Error: followup_task requires a non-empty message."); - } - const outcome = await deps.sessions.followupOne(target, message); - if (!outcome.ok) { - const hint = outcome.hint !== undefined ? ` ${outcome.hint}` : ""; - return lifecycleResult( - call.id, - `Error: cannot send followup to "${target}" (status: ${outcome.status}).${hint}`, - ); - } - return lifecycleResult( - call.id, - JSON.stringify({ agent_id: target, status: "completed", reply: outcome.reply }), - ); - }, - }); -} - const SendInputArgs = type({ target: "string", message: "string", @@ -345,7 +317,7 @@ export const sendInputToolDefinition: ToolDefinition = { "and return immediately without awaiting a reply and without completing wait_agents. " + "With interrupt:true: stop the current turn (same wait-mailbox flip as interrupt_agent) " + "then queue `message` as the next-turn followup without awaiting that reply. Fails on a " + - "session that is not currently running, or when the message is empty / oversize. Nested " + + "session that is not currently running an active turn, or when the message is empty / oversize. Nested " + "orchestrators may only target their own descendants.", inputSchema: { type: "object", diff --git a/src/subagent/nudge-director.ts b/src/subagent/nudge-director.ts index ec0a2c5a..ee8137da 100644 --- a/src/subagent/nudge-director.ts +++ b/src/subagent/nudge-director.ts @@ -111,7 +111,7 @@ export class SubAgentDirector extends DefaultDirector { // salvage), empty continuations from idle-compact / stall must not fall // through to DefaultDirector.infer — that re-opens the brief without a new // parent message (CL-7068). Cleared only by a non-empty parent message - // (followup_task / send_input). + // (resume_agent / send_input). private reportReplied = false; // Stall management: a leaf that goes quiet (e.g. parked on a long-running diff --git a/src/subagent/retain-salvage.test.ts b/src/subagent/retain-salvage.test.ts index 319c37a1..924007f7 100644 --- a/src/subagent/retain-salvage.test.ts +++ b/src/subagent/retain-salvage.test.ts @@ -19,7 +19,7 @@ describe("retained session lifecycle", () => { store.complete(s.id, "Stopped: deadline\n\nPartial work...", { agentRetained: false }); const after = store.get(s.id); console.log("lifecycleStatus:", after?.lifecycleStatus, "retained:", after?.retained); - const outcome = store.resumeOne(s.id); + const outcome = store.resumeOne(s.id, "more"); console.log("resumeOne outcome:", JSON.stringify(outcome)); expect(outcome.ok).toBe(false); }); @@ -44,7 +44,7 @@ describe("retained session lifecycle", () => { // CL-7007: retained completed sessions are no longer bounded by // `maxCompleted` (the TUI display cap) at all — that was CL-7002's fix, - // and it created a new bug: resume_agent/followup_task started failing + // and it created a new bug: resume_agent started failing // with a bare "not_found" once more than `maxCompleted` (default 20) // workers had spawned in a turn, even though every one of them was still // perfectly reusable. Open retained sessions now get their own explicit @@ -83,7 +83,8 @@ describe("retained session lifecycle", () => { // Mirrors agent-fleet's real call: only a clean turnSucceeded completion // sets agentRetained. store.complete(s.id, "done", { agentRetained: true }); - expect(store.resumeOne(s.id).ok).toBe(true); + store.registerFollowup(s.id, async () => "next"); + expect(store.resumeOne(s.id, "more").ok).toBe(true); expect(closed).toBe(false); store.cancelAll("parent stop"); expect(closed).toBe(true); diff --git a/src/subagent/run.ts b/src/subagent/run.ts index 4e36641e..f6a496ff 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -125,7 +125,6 @@ import { createCloseAgentTool, createResumeAgentTool, createInterruptAgentTool, - createFollowupTaskTool, createSendInputTool, } from "./lifecycle-tools.js"; import { createSubAgentSessionStore } from "./session-store.js"; @@ -448,15 +447,18 @@ async function runSubAgentInner( // Set only on the interrupt_agent path (a dedicated signal fired by the // `interrupt` handle below, never runController) — the finally block // skips teardown here too, so the agent and its workdir lock stay live - // for a later followup_task. + // for a later resume_agent. let interruptedKeepAlive = false; // Scoped to this run's `agent.send()` call only. Firing it rejects that // one send's promise (per Agent.send's documented signal option) without // touching agent.close() or runController — the reactor cycle it belongs // to keeps running in the background, exactly as the vendored send-queue - // documents, so a later followup_task's agent.send() simply queues behind + // documents, so a later resume_agent's agent.send() simply queues behind // it rather than racing a half-torn-down session. - const interruptController = new AbortController(); + // Per-turn abort for interrupt_agent. Recreated at the start of each + // followup send so a prior abort cannot immediately reject the next turn, + // and so interrupt_agent can stop a resumed agent.send(). + let interruptController = new AbortController(); // Declared before try (same reasoning as closeOnAbort above): assigned once // requestContinuation/modelFamilyPolicy exist inside the try, but must be // visible to the finally block, which is a sibling scope, not a child. @@ -472,6 +474,10 @@ async function runSubAgentInner( ? resolveSubAgentDeadlineMs(params.deadlineMs, undefined) : undefined; const runController = createSubAgentRunController(params.signal, resolvedDeadlineMs); + const sendAbortSignal = (): AbortSignal => + typeof AbortSignal.any === "function" + ? AbortSignal.any([runController.signal, interruptController.signal]) + : runController.signal; try { const shellDefaultMs = params.shellTimeout?.defaultMs; @@ -582,7 +588,6 @@ async function runSubAgentInner( "close_agent", "resume_agent", "interrupt_agent", - "followup_task", "send_input", ]) { assertTierMayMountFleetVerb(tier, verb); @@ -685,13 +690,16 @@ async function runSubAgentInner( fleetRecords, authority: lifecycleAuthority, }), - createResumeAgentTool({ sessions: fleetSessions, authority: lifecycleAuthority }), + createResumeAgentTool({ + sessions: fleetSessions, + fleetRecords, + authority: lifecycleAuthority, + }), createInterruptAgentTool({ sessions: fleetSessions, fleetRecords, authority: lifecycleAuthority, }), - createFollowupTaskTool({ sessions: fleetSessions, authority: lifecycleAuthority }), createSendInputTool({ sessions: fleetSessions, fleetRecords, @@ -1031,10 +1039,11 @@ async function runSubAgentInner( interruptController.abort(new Error("interrupted by interrupt_agent")); } }; - // followup_task's payoff — call agent.send() again on the same live + // resume_agent's payoff — call agent.send() again on the same live // agent object, reusing full context rather than starting fresh. const followup = async (message: string): Promise => { - const result = await agent!.send(message, { signal: runController.signal }); + interruptController = new AbortController(); + const result = await agent!.send(message, { signal: sendAbortSignal() }); return result.reply.trim().length > 0 ? result.reply.trim() : "Sub-agent finished without a textual result."; @@ -1077,16 +1086,13 @@ async function runSubAgentInner( // treat a pre-send check as permanent. if (runController.signal.aborted) throw abortError(runController.signal); }; + const thisTurnInterrupt = interruptController; try { ensureNotAborted(); // Combine the run's own controller with the dedicated interrupt // signal so either one stops this send() call, while only // runController's abort is wired to closeOnAbort/teardown. - const sendSignal = - typeof AbortSignal.any === "function" - ? AbortSignal.any([runController.signal, interruptController.signal]) - : runController.signal; - const sendOpts = { signal: sendSignal }; + const sendOpts = { signal: sendAbortSignal() }; const fresh = await refreshInferenceSourceBundle( bundle.sources, bundle.defaultSource, @@ -1124,7 +1130,7 @@ async function runSubAgentInner( // interrupt_agent fired its own signal, not runController's — check // that first so an interrupted send doesn't fall into the cancel/ // deadline salvage path or rethrow as a bare AbortError. - if (interruptController.signal.aborted && !runController.signal.aborted) { + if (thisTurnInterrupt.signal.aborted && !runController.signal.aborted) { interruptedKeepAlive = true; const abortedCycleText = await cycleRecorder.dispose("cancelled", { drain: streamPromise }); const tail = salvageFindingsText(accumulatedProse, lastPartialText, abortedCycleText); diff --git a/src/subagent/session-store.test.ts b/src/subagent/session-store.test.ts index bb1675b0..e0d4c325 100644 --- a/src/subagent/session-store.test.ts +++ b/src/subagent/session-store.test.ts @@ -307,25 +307,127 @@ describe("CL-6943 reusable worker sessions", () => { store.markRunning(session.id); expect(store.get(session.id)?.lifecycleStatus).toBe("running"); + store.registerFollowup(session.id, async () => "next turn"); store.complete(session.id, "## Summary\nDone."); expect(store.get(session.id)?.lifecycleStatus).toBe("completed"); expect(store.get(session.id)?.retained).toBe(true); - const outcome = store.resumeOne(session.id); - expect(outcome).toEqual({ ok: true }); + const outcome = store.resumeOne(session.id, "continue"); + expect(outcome).toEqual({ ok: true, status: "running" }); + expect(store.get(session.id)?.status).toBe("running"); expect(store.get(session.id)?.lifecycleStatus).toBe("running"); }); + test("resume-from-completed is a live turn: send_input, interrupt, and appendEvent work", async () => { + let finish: (reply: string) => void = () => {}; + const store = createSubAgentSessionStore(); + const session = store.start({ description: "d", agentId: "a", brief: "b", retained: true }); + store.markRunning(session.id); + const delivered: string[] = []; + store.registerDeliver(session.id, (message) => { + delivered.push(message); + }); + store.registerInterrupt(session.id, () => {}); + store.registerFollowup( + session.id, + () => + new Promise((resolve) => { + finish = resolve; + }), + ); + store.complete(session.id, "## Summary\nDone."); + expect(store.get(session.id)?.status).toBe("done"); + expect(store.get(session.id)?.lifecycleStatus).toBe("completed"); + + expect(store.resumeOne(session.id, "continue")).toEqual({ ok: true, status: "running" }); + expect(store.get(session.id)?.status).toBe("running"); + expect(store.get(session.id)?.lifecycleStatus).toBe("running"); + expect(store.get(session.id)?.finishedAt).toBeUndefined(); + + expect(store.sendInputOne(session.id, "steer mid-turn")).toEqual({ + ok: true, + status: "running", + }); + expect(delivered).toEqual(["steer mid-turn"]); + + store.appendEvent(session.id, startCall(1, "call-1", "grep")); + const afterTool = store.get(session.id); + if (afterTool === undefined) throw new Error("session missing after tool start"); + expect(afterTool.toolNames).toContain("grep"); + expect(afterTool.outstandingTools).toHaveLength(1); + expect(afterTool.currentToolName).toBe("grep"); + expect(agentLaneIsLive(afterTool)).toBe(true); + expect(fleetProgress(store.list(), Date.now()).running).toBe(1); + + expect(store.interruptOne(session.id).ok).toBe(true); + expect(store.get(session.id)?.status).toBe("running"); + expect(store.get(session.id)?.lifecycleStatus).toBe("interrupted"); + + finish("later"); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(store.get(session.id)?.status).toBe("done"); + expect(store.get(session.id)?.lifecycleStatus).toBe("completed"); + }); + + test("rejected followup restores strip status so interrupt_agent fails closed", async () => { + const store = createSubAgentSessionStore(); + const session = store.start({ description: "d", agentId: "a", brief: "b", retained: true }); + store.markRunning(session.id); + store.registerInterrupt(session.id, () => {}); + store.registerFollowup(session.id, async () => { + throw new Error("send failed"); + }); + store.complete(session.id, "## Summary\nDone."); + expect(store.get(session.id)?.status).toBe("done"); + expect(store.get(session.id)?.lifecycleStatus).toBe("completed"); + + expect(store.resumeOne(session.id, "continue")).toEqual({ ok: true, status: "running" }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const after = store.get(session.id); + expect(after?.status).toBe("done"); + expect(after?.lifecycleStatus).toBe("completed"); + expect(store.interruptOne(session.id)).toEqual({ ok: false, status: "completed" }); + }); + + test("interrupt then abort does not overwrite interrupted stamp to completed", async () => { + let rejectFollowup: (err: unknown) => void = () => {}; + const store = createSubAgentSessionStore(); + const session = store.start({ description: "d", agentId: "a", brief: "b", retained: true }); + store.markRunning(session.id); + store.registerInterrupt(session.id, () => {}); + store.registerFollowup( + session.id, + () => + new Promise((_resolve, reject) => { + rejectFollowup = reject; + }), + ); + store.complete(session.id, "## Summary\nDone."); + + expect(store.resumeOne(session.id, "continue")).toEqual({ ok: true, status: "running" }); + expect(store.interruptOne(session.id).ok).toBe(true); + expect(store.get(session.id)?.status).toBe("running"); + expect(store.get(session.id)?.lifecycleStatus).toBe("interrupted"); + + rejectFollowup(new Error("aborted")); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const after = store.get(session.id); + expect(after?.status).toBe("running"); + expect(after?.lifecycleStatus).toBe("interrupted"); + }); + test("resume_agent fails on a session that was never retained", () => { const store = createSubAgentSessionStore(); const session = store.start({ description: "d", agentId: "a", brief: "b" }); store.complete(session.id, "## Summary\nDone."); - expect(store.resumeOne(session.id)).toEqual({ ok: false, status: "completed" }); + expect(store.resumeOne(session.id, "more")).toEqual({ ok: false, status: "completed" }); }); test("resume_agent fails on an unknown id with not_found", () => { const store = createSubAgentSessionStore(); - expect(store.resumeOne("missing")).toEqual({ ok: false, status: "not_found" }); + expect(store.resumeOne("missing", "more")).toEqual({ ok: false, status: "not_found" }); }); test("closeOne is bounded by its deadline when the registered close hangs forever", async () => { @@ -363,13 +465,13 @@ describe("CL-6943 reusable worker sessions", () => { store.registerClose(session.id, async () => {}); store.complete(session.id, "## Summary\nDone."); await store.closeOne(session.id, 1000); - expect(store.resumeOne(session.id)).toEqual({ ok: false, status: "shutdown" }); + expect(store.resumeOne(session.id, "more")).toEqual({ ok: false, status: "shutdown" }); }); // CL-7001 originally folded a retained, still-open session into // maxCompleted (the TUI display cap) with no separate bound at all, - // fixing the unbounded leak but creating a new bug: resume_agent / - // followup_task fail once more than `maxCompleted` (default 20) workers + // fixing the unbounded leak but creating a new bug: resume_agent + // fails once more than `maxCompleted` (default 20) workers // have spawned, even though every one of them is still perfectly // reusable. CL-7007 gives open retained sessions their own cap // (`maxRetained`) instead — this test changed from asserting that @@ -444,7 +546,7 @@ describe("CL-6943 reusable worker sessions", () => { store.complete(s.id, "## Summary\nDone."); } - const outcome = store.resumeOne(retained.id); + const outcome = store.resumeOne(retained.id, "more"); expect(outcome.ok).toBe(false); if (!outcome.ok) { expect(outcome.status).toBe("completed"); @@ -463,8 +565,9 @@ describe("CL-6943 reusable worker sessions", () => { store.markRunning(running.id); // Resume it back to "running" so it is an open, actively-driven session. store.registerClose(running.id, async () => {}); + store.registerFollowup(running.id, () => new Promise(() => {})); store.complete(running.id, "## Summary\nDone."); - store.resumeOne(running.id); + store.resumeOne(running.id, "keep going"); expect(store.get(running.id)?.lifecycleStatus).toBe("running"); for (let i = 0; i < 5; i++) { @@ -628,7 +731,8 @@ describe("interrupt stamps finishedAt once", () => { expect(store.get(session.id)?.finishedAt).toBe(2000); t = 3000; - const pending = store.followupOne(session.id, "keep going"); + const outcome = store.resumeOne(session.id, "keep going"); + expect(outcome).toEqual({ ok: true, status: "running" }); t = 11_000; store.appendEvent(session.id, startCall(1, "call-1", "run_shell")); @@ -641,7 +745,7 @@ describe("interrupt stamps finishedAt once", () => { t = 12_000; finish("done"); - expect(await pending).toEqual({ ok: true, reply: "done" }); + await new Promise((resolve) => setTimeout(resolve, 0)); const terminal = store.list(); expect(terminal[0]?.status).toBe("done"); expect(terminal[0]?.lifecycleStatus).toBe("completed"); diff --git a/src/subagent/session-store.ts b/src/subagent/session-store.ts index 42a84e88..c1a47d76 100644 --- a/src/subagent/session-store.ts +++ b/src/subagent/session-store.ts @@ -126,7 +126,7 @@ export interface SubAgentSessionStoreOptions { // ("completed"/"interrupted" with retained:true) — see maxRetained. maxCompleted?: number; // CL-7007: cap on open retained sessions (spawn_agent workers a caller may - // still resume_agent/followup_task). Sized for fan-out (dozens of + // still resume_agent). Sized for fan-out (dozens of // concurrent workers), independent of maxCompleted's TUI display cap. // Least-recently-used is evicted first; a "running" session is never // evicted regardless of this bound. Non-finite/undefined values (and a @@ -180,15 +180,24 @@ export interface SubAgentSessionStore { // "not_found" for an unknown id without throwing (callers need the status, // not an exception, to report per-target results across a descendant walk). closeOne(id: string, deadlineMs: number): Promise; - // Transitions a retained, still-open ("completed") session back to - // "running" for further input. Fails closed on anything else — a - // "shutdown" session is gone for good (close_agent is permanent), an - // "interrupted" one already tore its agent down, and "running"/ + // Transitions a retained, still-open ("completed" or "interrupted") session + // back to "running", starts the next turn through the registered followup + // handle, and returns immediately. wait_agents collects the reply. Fails + // closed on anything else — a "shutdown" session is gone for good + // (close_agent is permanent), a still-running turn is concurrent, and // "pending_init"/"not_found" have nothing to resume. // CL-7007: a session dropped by pruneRetained still reports its terminal // lifecycleStatus plus `hint` pointing at read_agent_trace — never a bare // "not_found" that reads like a bad id. - resumeOne(id: string): { ok: true } | { ok: false; status: AgentLifecycleStatus; hint?: string }; + resumeOne( + id: string, + message: string, + opts?: { + onStart?: () => void; + onReply?: (reply: string) => void; + onFail?: (error: unknown) => void; + }, + ): { ok: true; status: "running" } | { ok: false; status: AgentLifecycleStatus; hint?: string }; // CL-6997: registers the per-session interrupt/followup handles run.ts // hands back via onAgentReady. Distinct maps from registerClose/closeOne // above (interrupt must never route through close's codepath). @@ -199,16 +208,6 @@ export interface SubAgentSessionStore { // run's promise to settle. Fails closed on anything not currently running // or with no interrupt handle registered (e.g. a session past init). interruptOne(id: string): { ok: true } | { ok: false; status: AgentLifecycleStatus }; - // Sends `message` through the registered followup handle (the same live - // agent, same context) and records the reply as this session's new report - // on success. Fails closed on a session that is not retained or not in a - // resumable state ("completed" or "interrupted"). - followupOne( - id: string, - message: string, - ): Promise< - { ok: true; reply: string } | { ok: false; status: AgentLifecycleStatus; hint?: string } - >; registerDeliver(id: string, deliver: (message: string) => void): void; sendInputOne( id: string, @@ -377,7 +376,7 @@ export function createSubAgentSessionStore( const listeners = new Set<() => void>(); // CL-7007: tombstones for sessions dropped by pruneRetained, keyed by id, // insertion-ordered (Map preserves it) so the oldest can be dropped first - // once MAX_EVICTED_TOMBSTONES is exceeded. Lets resume_agent/followup_task + // once MAX_EVICTED_TOMBSTONES is exceeded. Lets resume_agent // report an actionable terminal status instead of a bare "not_found" for a // session evicted purely to bound retention memory. const evicted = new Map(); @@ -436,7 +435,8 @@ export function createSubAgentSessionStore( content: capText(`Cancelled: ${reason}`, maxEntryChars), }); cancelHandles.delete(session.id); - closeHandles.delete(session.id); + // closeHandles are owned by releaseHandles / closeOne — dropping them + // here would skip teardown for a retained session that is mid-turn. bumpRevision(session.id); pruneCompleted(); }; @@ -537,7 +537,7 @@ export function createSubAgentSessionStore( // candidate (excluded by isOpenRetained requiring "completed"/ // "interrupted"). Handles are released exactly like pruneCompleted's // eviction — sidecars, reactor, and the lock entry are not simply - // forgotten — and a tombstone is kept so resume_agent/followup_task can + // forgotten — and a tombstone is kept so resume_agent can // still report an actionable status afterward instead of "not_found". const pruneRetained = (): void => { const openRetained = [...sessions.values()] @@ -600,19 +600,59 @@ export function createSubAgentSessionStore( // A follow-up turn takes the lane back over: the worker is live again, so // the interrupt's linger stamp must not outlive the new turn. Completion // re-stamps through the caller's own mutate; a rejected turn restores the - // addressable state it started from so followup_task can retry. + // addressable strip state it started from (done, or interrupted linger) so + // resume_agent can retry. interrupt_agent's stamp on this turn wins over + // that restore — do not rewrite interrupted back to completed. const beginFollowupTurn = (id: string): void => { mutate(id, (s) => { + s.status = "running"; s.lifecycleStatus = "running"; delete s.finishedAt; }); }; const endFollowupTurn = (id: string, lifecycleStatus: AgentLifecycleStatus): void => { mutate(id, (s) => { + if (s.lifecycleStatus === "interrupted") { + s.finishedAt = s.finishedAt ?? now(); + return; + } s.lifecycleStatus = lifecycleStatus; s.finishedAt = now(); + s.status = lifecycleStatus === "interrupted" ? "running" : "done"; }); }; + const queueFollowupTurn = ( + id: string, + message: string, + failLifecycle: AgentLifecycleStatus, + opts?: { onReply?: (reply: string) => void; onFail?: (error: unknown) => void }, + ): void => { + const followup = followupHandles.get(id); + if (followup === undefined) return; + beginFollowupTurn(id); + void followup(message) + .then((reply) => { + const still = sessions.get(id); + if (still === undefined || still.lifecycleStatus === "shutdown") return; + mutate(id, (s) => { + s.status = "done"; + s.lifecycleStatus = "completed"; + s.finishedAt = now(); + s.report = reply; + pushEntry(s, { kind: "report", content: capText(reply, maxEntryChars) }); + }); + opts?.onReply?.(reply); + pruneRetained(); + }) + .catch((err: unknown) => { + endFollowupTurn(id, failLifecycle); + opts?.onFail?.(err); + log.error("followup turn failed for {id}: {error}", { + id, + error: err instanceof Error ? err.message : String(err), + }); + }); + }; return { list(): readonly SubAgentSession[] { @@ -983,7 +1023,9 @@ export function createSubAgentSessionStore( ): { ok: true; status: AgentLifecycleStatus } | { ok: false; status: AgentLifecycleStatus } { const session = sessions.get(id); if (session === undefined) return { ok: false, status: "not_found" }; - if (session.status !== "running") return { ok: false, status: session.lifecycleStatus }; + if (session.status !== "running" || session.lifecycleStatus !== "running") { + return { ok: false, status: session.lifecycleStatus }; + } if (opts?.interrupt === true) { const interrupt = interruptHandles.get(id); @@ -992,28 +1034,9 @@ export function createSubAgentSessionStore( return { ok: false, status: session.lifecycleStatus }; } interrupt(); - beginFollowupTurn(id); - void followup(message) - .then((reply) => { - const still = sessions.get(id); - if (still === undefined || still.lifecycleStatus === "shutdown") return; - mutate(id, (s) => { - s.status = "done"; - s.lifecycleStatus = "completed"; - s.finishedAt = now(); - s.report = reply; - pushEntry(s, { kind: "report", content: capText(reply, maxEntryChars) }); - }); - opts.onFollowupReply?.(reply); - pruneRetained(); - }) - .catch((err: unknown) => { - endFollowupTurn(id, "interrupted"); - log.error("send_input followup failed for {id}: {error}", { - id, - error: err instanceof Error ? err.message : String(err), - }); - }); + queueFollowupTurn(id, message, "interrupted", { + ...(opts.onFollowupReply !== undefined ? { onReply: opts.onFollowupReply } : {}), + }); pruneRetained(); return { ok: true, status: "interrupted" }; } @@ -1039,12 +1062,16 @@ export function createSubAgentSessionStore( return { ok: true }; }, - async followupOne( + resumeOne( id: string, message: string, - ): Promise< - { ok: true; reply: string } | { ok: false; status: AgentLifecycleStatus; hint?: string } - > { + opts?: { + onStart?: () => void; + onReply?: (reply: string) => void; + onFail?: (error: unknown) => void; + }, + ): + { ok: true; status: "running" } | { ok: false; status: AgentLifecycleStatus; hint?: string } { const session = sessions.get(id); if (session === undefined) { const tombstone = evicted.get(id); @@ -1062,43 +1089,13 @@ export function createSubAgentSessionStore( const followup = followupHandles.get(id); if (followup === undefined) return { ok: false, status: session.lifecycleStatus }; const priorLifecycle = session.lifecycleStatus; - beginFollowupTurn(id); - let reply: string; - try { - reply = await followup(message); - } catch (err) { - endFollowupTurn(id, priorLifecycle); - throw err; - } - mutate(id, (s) => { - s.status = "done"; - s.lifecycleStatus = "completed"; - s.finishedAt = now(); - s.report = reply; - pushEntry(s, { kind: "report", content: capText(reply, maxEntryChars) }); + queueFollowupTurn(id, message, priorLifecycle, { + ...(opts?.onReply !== undefined ? { onReply: opts.onReply } : {}), + ...(opts?.onFail !== undefined ? { onFail: opts.onFail } : {}), }); + opts?.onStart?.(); pruneRetained(); - return { ok: true, reply }; - }, - - resumeOne( - id: string, - ): { ok: true } | { ok: false; status: AgentLifecycleStatus; hint?: string } { - const session = sessions.get(id); - if (session === undefined) { - const tombstone = evicted.get(id); - if (tombstone !== undefined) { - return { ok: false, status: tombstone.lifecycleStatus, hint: tombstone.hint }; - } - return { ok: false, status: "not_found" }; - } - if (session.lifecycleStatus !== "completed" || session.retained !== true) { - return { ok: false, status: session.lifecycleStatus }; - } - mutate(id, (s) => { - s.lifecycleStatus = "running"; - }); - return { ok: true }; + return { ok: true, status: "running" }; }, cancel(id: string, reason = DEFAULT_CANCEL_REASON): boolean { diff --git a/src/subagent/types.ts b/src/subagent/types.ts index e8089fd9..0167d188 100644 --- a/src/subagent/types.ts +++ b/src/subagent/types.ts @@ -170,7 +170,7 @@ export type RunSubAgentParams = { * `signal` option); this only stops the caller from waiting on it. * - `followup`: sends a new message into the same live agent (same * history, same context store) once the current turn is no longer - * active — this is the resume mechanism `resume_agent`/`followup_task` + * active — this is the resume mechanism `resume_agent` * build on, reusing `agent.send`'s own FIFO send-queue ordering rather * than a second continuation scheme. *