Add interrupt_agent / followup_task (CL-6997) (#619)

* Add interrupt_agent / followup_task (CL-6997)

Second half of reusable worker sessions: interrupt_agent stops a
retained worker's current turn while keeping the session and its
context reusable, and followup_task sends new work into a retained
session's existing agent, reusing its prior context and tool outputs.

interrupt_agent fires a signal scoped only to the in-flight
agent.send() call, never close() — it cannot hit the close()-ordering
workdir-lock issue tracked separately (CL-6984). There is no lower-
level stop primitive in the vendored agent for the reactor cycle
itself, so this is an approximation: it stops the caller from waiting,
not the worker's compute, which keeps running in the background until
it finishes naturally.

Both verbs are gated to orchestrator tiers via the existing
FLEET_VERBS / assertTierMayMountFleetVerb mechanism, denied to leaves.

* Add live-agent regression guard for followup_task (CL-6997)

lifecycle-tools.test.ts only exercised interrupt_agent/followup_task
against fake registered closures at the tool/store layer, not run.ts's
real onAgentReady wiring where followup calls agent!.send() on the
same live agent object. Add a test that drives runSubAgent end to end
with createAgentWithLiveToolDispatch replaced by a stub Agent, proving
the followup send after an interrupt lands on the same agent instance
(one construction, one shared message log) rather than a rebuilt one.

Author: TheGreatAxios

CHANGELOG.md

@@ -37,6 +37,19 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename
   operator interrupts it (`interrupt_agent`) rather than the harness enforcing
   a count.
 
+- Added `interrupt_agent({ target })` and `followup_task({ target, message })`,
+  the second half of reusable worker sessions: `interrupt_agent` stops a
+  retained worker's current turn while keeping it and its context alive
+  (distinct from the permanent `close_agent`), and `followup_task` sends new
+  work into a retained worker's existing session, reusing its prior context
+  and tool outputs rather than starting fresh. Both are gated to orchestrator
+  tiers via the existing fleet-verb mechanism, denied to leaves. `interrupt_agent`
+  fires a signal scoped only to the in-flight `agent.send()` call, never
+  `close()`, so it cannot hit the close()-ordering workdir-lock issue tracked
+  separately — the underlying reactor cycle keeps running in the background
+  (there is no lower-level stop primitive for that in the vendored agent), so
+  this is an approximation: it stops the caller from waiting, not the
+  worker's compute.
 - `evaluateSubAgentStop` now always requires the final assistant text; the
   omitted-text branch that unconditionally completed a tool-less turn is
   removed, so every call path gets the `incomplete-report` nudge and salvage

src/subagent/agent-fleet.ts

@@ -449,8 +449,10 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
         // CL-6943: keep the session open after a clean completion, and hand
         // the store a bounded close for close_agent to call later.
         persist: true,
-        onAgentReady: (close) => {
+        onAgentReady: ({ close, interrupt, followup }) => {
           deps.sessions.registerClose(session.id, close);
+          deps.sessions.registerInterrupt(session.id, interrupt);
+          deps.sessions.registerFollowup(session.id, followup);
           deps.sessions.markRunning(session.id);
         },
       };
@@ -466,6 +468,11 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
         .run(params)
         .then((result) => {
           if (childCtl.signal.aborted) return;
+          // CL-6997: interrupt_agent already flipped this session to
+          // "interrupted" synchronously (session-store.interruptOne) — do
+          // not let the settling promise's normal bookkeeping overwrite
+          // that with a "completed" status.
+          if (result.interrupted === true) return;
           deps.fleetRecords.resolve(session.id, result.report);
           // CL-7001: result.agentRetained is only true on run.ts's clean-
           // completion path when persist actually skipped teardown — a

src/subagent/authority.test.ts

@@ -14,6 +14,11 @@ describe("assertTierMayMountFleetVerb", () => {
     // CL-6943: the reusable-session verbs are gated the same way.
     expect(() => assertTierMayMountFleetVerb("leaf", "close_agent")).toThrow(FleetAuthorityError);
     expect(() => assertTierMayMountFleetVerb("leaf", "resume_agent")).toThrow(FleetAuthorityError);
+    // CL-6997: interrupt_agent / followup_task are gated the same way.
+    expect(() => assertTierMayMountFleetVerb("leaf", "interrupt_agent")).toThrow(
+      FleetAuthorityError,
+    );
+    expect(() => assertTierMayMountFleetVerb("leaf", "followup_task")).toThrow(FleetAuthorityError);
   });
 
   test("leaves may still mount non-fleet tools", () => {

src/subagent/followup-live-agent.test.ts

@@ -0,0 +1,160 @@
+/**
+ * CL-6997 regression guard: lifecycle-tools.test.ts proves interrupt_agent /
+ * followup_task 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
+ * regression this feature exists to prevent — a rebuilt agent means the
+ * worker re-reads the codebase from scratch) without failing any existing
+ * test.
+ *
+ * This test drives the real `runSubAgent` (run.ts) end to end with the one
+ * real dependency that would require live inference credentials —
+ * `createAgentWithLiveToolDispatch` — replaced by a stub `Agent`. Everything
+ * else (tool assembly, environment gathering, the dispatch brief, the
+ * onAgentReady wiring, the interrupt/followup closures themselves) is the
+ * genuine run.ts code path.
+ */
+import { describe, expect, test } from "bun:test";
+import { mkdtemp } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import { withMockedModuleDuring } from "../../tests/helpers/mock-module.js";
+import { createPermissionGate } from "../permission/gate.js";
+import type { RunSubAgentParams } from "./types.js";
+
+const testPermissionGate = createPermissionGate({
+  approvals: [],
+  interactive: false,
+  skipPermissions: true,
+});
+
+async function tmpCwd(): Promise<string> {
+  return mkdtemp(join(tmpdir(), "cl6997-live-agent-"));
+}
+
+/** Minimal stand-in for the vendored `Agent` (dist/agent.d.ts), instrumented
+ * 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() {
+  const sendLog: string[] = [];
+  return {
+    sendLog,
+    async send(content: string, opts?: { signal?: AbortSignal }) {
+      sendLog.push(content);
+      return await new Promise((resolve, reject) => {
+        if (opts?.signal?.aborted === true) {
+          reject(opts.signal.reason instanceof Error ? opts.signal.reason : new Error("aborted"));
+          return;
+        }
+        const timer = setTimeout(
+          () =>
+            resolve({
+              reply: `reply #${sendLog.length}`,
+              turn: { role: "assistant", content: [] },
+            }),
+          20,
+        );
+        opts?.signal?.addEventListener(
+          "abort",
+          () => {
+            clearTimeout(timer);
+            reject(
+              opts.signal!.reason instanceof Error ? opts.signal!.reason : new Error("aborted"),
+            );
+          },
+          { once: true },
+        );
+      });
+    },
+    stream: () => (async function* () {})(),
+    deliver: () => {},
+    close: async () => {},
+    setSource: () => {},
+    setSources: () => {},
+    history: async () => [],
+    checkpoints: async () => [],
+    readAt: async () => [],
+    blobReader: {},
+  };
+}
+
+describe("interrupt_agent / followup_task reuse the same live agent (CL-6997)", () => {
+  test("followup after interrupt sends into the SAME agent instance — not a rebuilt one", async () => {
+    const cwd = await tmpCwd();
+    let constructions = 0;
+    let capturedAgent: ReturnType<typeof createStubAgent> | 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();
+          capturedAgent = stub;
+          return stub as unknown as Awaited<
+            ReturnType<typeof real.createAgentWithLiveToolDispatch>
+          >;
+        },
+      }),
+      async () => {
+        const { runSubAgent } = await import("./run.js");
+
+        let handles:
+          | {
+              close: (ms?: number) => Promise<void>;
+              interrupt: () => void;
+              followup: (message: string) => Promise<string>;
+            }
+          | undefined;
+
+        const params: RunSubAgentParams = {
+          cwd,
+          workdirBase: join(cwd, ".ctx"),
+          permissionGate: testPermissionGate,
+          provider: { providerName: "test", baseURL: "http://localhost", model: "test-model" },
+          description: "live-agent reuse probe",
+          prompt: "explore the codebase for the bug",
+          persist: true,
+          onAgentReady: (h) => {
+            handles = h;
+          },
+        };
+
+        const runPromise = runSubAgent(params);
+
+        // onAgentReady fires before agent.send() is awaited; poll briefly
+        // rather than assume a fixed number of ticks.
+        for (let i = 0; i < 500 && handles === undefined; i++) {
+          await new Promise((resolve) => setTimeout(resolve, 1));
+        }
+        if (handles === undefined) throw new Error("onAgentReady never fired");
+
+        handles.interrupt();
+        const interruptedResult = await runPromise;
+
+        const reply = await handles.followup("do X instead, not what the original prompt said");
+        return { interruptedResult, reply };
+      },
+    );
+
+    expect(outcome.interruptedResult.interrupted).toBe(true);
+    // Exactly one agent was ever constructed across the interrupted turn and
+    // the followup — a rebuild would show up here as constructions === 2.
+    expect(constructions).toBe(1);
+    expect(capturedAgent).toBeDefined();
+
+    // The load-bearing assertion: the SAME agent's message log holds both
+    // the original turn's prompt and the followup message, proving the
+    // followup was sent into the same live object rather than a fresh one
+    // with empty history.
+    expect(capturedAgent!.sendLog.length).toBe(2);
+    expect(capturedAgent!.sendLog[0]).toContain("explore the codebase for the bug");
+    expect(capturedAgent!.sendLog[1]).toBe("do X instead, not what the original prompt said");
+    expect(outcome.reply).toBe("reply #2");
+  });
+});

src/subagent/lifecycle-tools.test.ts

@@ -1,10 +1,19 @@
 import { describe, expect, test } from "bun:test";
 
-import { createCloseAgentTool, createResumeAgentTool } from "./lifecycle-tools.js";
+import {
+  createCloseAgentTool,
+  createResumeAgentTool,
+  createInterruptAgentTool,
+  createFollowupTaskTool,
+} from "./lifecycle-tools.js";
 import { createSubAgentSessionStore } from "./session-store.js";
 
 async function callTool(
-  tool: ReturnType<typeof createCloseAgentTool> | ReturnType<typeof createResumeAgentTool>,
+  tool:
+    | ReturnType<typeof createCloseAgentTool>
+    | ReturnType<typeof createResumeAgentTool>
+    | ReturnType<typeof createInterruptAgentTool>
+    | ReturnType<typeof createFollowupTaskTool>,
   args: Record<string, unknown>,
 ): Promise<Record<string, unknown>> {
   if (tool.kind !== "full") throw new Error(`expected full tool, got ${tool.kind}`);
@@ -101,3 +110,151 @@ describe("resume_agent", () => {
     expect(rawResult.isError).toBe(true);
   });
 });
+
+describe("interrupt_agent / followup_task", () => {
+  test("interrupt then followup keeps prior context — the worker does not re-read from scratch", async () => {
+    const sessions = createSubAgentSessionStore();
+    const worker = sessions.start({
+      description: "worker",
+      agentId: "a",
+      brief: "b",
+      retained: true,
+    });
+    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;
+    sessions.registerInterrupt(worker.id, () => {
+      interruptFired = true;
+    });
+    sessions.registerFollowup(worker.id, async (message: string) => {
+      history.push(message);
+      return `Applying fix given ${history.length} prior turns of context.`;
+    });
+
+    const interruptAgent = createInterruptAgentTool({ sessions });
+    const followupTask = createFollowupTaskTool({ sessions });
+
+    const interruptResult = await callTool(interruptAgent, { target: worker.id });
+    expect(interruptResult.status).toBe("interrupted");
+    expect(interruptFired).toBe(true);
+    expect(sessions.get(worker.id)?.lifecycleStatus).toBe("interrupted");
+
+    const followupResult = await callTool(followupTask, {
+      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(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);
+    expect(sessions.get(worker.id)?.lifecycleStatus).toBe("completed");
+    expect(sessions.get(worker.id)?.report).toBe(followupResult.reply as string);
+  });
+
+  test("followup_task on a completed retained worker reuses its existing session, not a fresh one", async () => {
+    const sessions = createSubAgentSessionStore();
+    const worker = sessions.start({
+      description: "worker",
+      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.");
+
+    const followupTask = createFollowupTaskTool({ sessions });
+    const result = await callTool(followupTask, { target: worker.id, message: "now do task two" });
+
+    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 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(
+      {
+        id: "c3",
+        name: "followup_task",
+        arguments: { target: nonRetained.id, message: "more work" },
+      },
+      new AbortController().signal,
+    );
+    expect(rejected.isError).toBe(true);
+  });
+
+  test("an interrupted session is resumable via followup_task and interrupt never touches close()", async () => {
+    const sessions = createSubAgentSessionStore();
+    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");
+
+    const interruptAgent = createInterruptAgentTool({ sessions });
+    const followupTask = createFollowupTaskTool({ sessions });
+
+    await callTool(interruptAgent, { target: worker.id });
+    expect(closeCalls).toBe(0);
+
+    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");
+  });
+
+  test("interrupt_agent and followup_task fail closed on a non-running / non-retained target", async () => {
+    const sessions = createSubAgentSessionStore();
+    const notRunning = sessions.start({ description: "d", agentId: "a", brief: "b" });
+    sessions.complete(notRunning.id, "## Summary\nDone.");
+
+    const interruptAgent = createInterruptAgentTool({ sessions });
+    const followupTask = createFollowupTaskTool({ sessions });
+
+    if (interruptAgent.kind !== "full") throw new Error("expected full tool");
+    const interruptErr = await interruptAgent.handler(
+      { id: "c1", name: "interrupt_agent", arguments: { target: notRunning.id } },
+      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);
+  });
+});