test(server): reproduce ACP restart after failed process termination

Refs #3256 and #7472. A real ACP subprocess remains alive when the injected termination dependency returns kill-timeout. The following prompt starts a second live subprocess, violating exclusive runtime ownership. This regression test intentionally fails on the inspected contributor commit.

Co-authored-by: Ouyang Xingyuan <galaxyroaming@qq.com>

docs/agent-lifecycle.md

@@ -50,6 +50,15 @@ older cancellation from settling a newer turn. If interruption is rejected or ti
 keeps its active foreground turn and replacement, reload, rewind, and Stop report the failure.
 Accepting new work after an ambiguous interruption would create a split-brain session.
 
+A forced settlement leaves two owners of "is a turn running": the manager has settled the run, while
+the provider session still holds the turn slot it never confirmed ending. `AgentSession.abandonForegroundTurn`
+is the optional seam for that case. The manager calls it after force-canceling, and only then. A
+provider that implements it must make the session usable again, not just clear a local flag: the ACP
+adapter kills the process that owned the turn and drops the slot, and its next `startTurn` resumes the
+durable session in a new process. Clearing the slot while reusing a runtime that may still be working
+would be the split-brain shape this section forbids. Providers that omit the hook keep today's
+behavior, and a rejected release is logged without undoing the settlement.
+
 ## Relationships
 
 Agents can launch other agents via the agent-scoped `create_agent` MCP tool. Agent-scoped creation is always asynchronous and always stamps `paseo.parent-agent-id`, pointing back at the caller. Omit `workspaceId` to use the caller's workspace, or pass an existing workspace ID returned by `create_workspace`. Placement never changes parentage.

packages/server/src/server/agent/agent-manager.test.ts

@@ -1364,6 +1364,7 @@ class McpCapableTestAgentClient extends TestAgentClient {
 
 class ControlledInterruptSession extends TestAgentSession {
   interruptCalled = false;
+  abandonCalls = 0;
 
   constructor(
     config: AgentSessionConfig,
@@ -1384,6 +1385,16 @@ class ControlledInterruptSession extends TestAgentSession {
     this.interruptCalled = true;
     await this.interruptBehavior(this);
   }
+
+  override async abandonForegroundTurn(): Promise<void> {
+    this.abandonCalls += 1;
+    this.pushEvent({
+      type: "turn_canceled",
+      provider: this.provider,
+      reason: "Interrupted",
+      turnId: this.turnId,
+    });
+  }
 }
 
 interface ControlledInterruptFixture {
@@ -2785,6 +2796,91 @@ test("cancelAgentRun succeeds when the provider queues completion before rejecti
   }
 });
 
+test("a force-canceled foreground turn releases the provider turn it never settled", async () => {
+  const fixture = await createControlledInterruptFixture({
+    name: "interrupt-acknowledged-no-settle",
+    agentId: "00000000-0000-4000-8000-000000000307",
+    turnId: "stuck-after-cancel-turn",
+    // The provider answers cancel but never answers session/prompt, so the run stays unsettled
+    // and the manager has to force-cancel its own side of the turn.
+    interrupt: async () => {},
+  });
+
+  try {
+    await fixture.startForegroundRun();
+    expect(fixture.manager.getAgent(fixture.agentId)?.activeForegroundTurnId).toBe(
+      "stuck-after-cancel-turn",
+    );
+
+    await expect(fixture.manager.cancelAgentRun(fixture.agentId)).resolves.toEqual({
+      status: "settled",
+    });
+    expect(fixture.session.abandonCalls).toBe(1);
+    expect(fixture.manager.getAgent(fixture.agentId)).toMatchObject({
+      lifecycle: "idle",
+      activeForegroundTurnId: null,
+    });
+
+    // The session must accept the next prompt: that is the bug this release exists for.
+    await fixture.startForegroundRun();
+    expect(fixture.manager.getAgent(fixture.agentId)?.lifecycle).toBe("running");
+  } finally {
+    await fixture.cleanup();
+  }
+});
+
+test("a settled cancel does not release the provider turn", async () => {
+  let fixture!: ControlledInterruptFixture;
+  fixture = await createControlledInterruptFixture({
+    name: "interrupt-settled",
+    agentId: "00000000-0000-4000-8000-000000000308",
+    turnId: "settled-turn",
+    interrupt: async (session) => {
+      session.pushEvent({
+        type: "turn_canceled",
+        provider: session.provider,
+        reason: "Interrupted",
+        turnId: "settled-turn",
+      });
+    },
+  });
+
+  try {
+    await fixture.startForegroundRun();
+    await expect(fixture.manager.cancelAgentRun(fixture.agentId)).resolves.toEqual({
+      status: "settled",
+    });
+    expect(fixture.session.abandonCalls).toBe(0);
+    expect(fixture.manager.getAgent(fixture.agentId)?.lifecycle).toBe("idle");
+  } finally {
+    await fixture.cleanup();
+  }
+});
+
+test("a provider that cannot release the turn still settles the cancel", async () => {
+  const fixture = await createControlledInterruptFixture({
+    name: "interrupt-release-rejected",
+    agentId: "00000000-0000-4000-8000-000000000309",
+    turnId: "release-rejected-turn",
+    interrupt: async () => {},
+  });
+  let releaseAttempts = 0;
+  fixture.session.abandonForegroundTurn = async () => {
+    releaseAttempts += 1;
+    throw new Error("The provider process did not exit");
+  };
+
+  try {
+    await fixture.startForegroundRun();
+    await expect(fixture.manager.cancelAgentRun(fixture.agentId)).resolves.toEqual({
+      status: "settled",
+    });
+    expect(releaseAttempts).toBe(1);
+  } finally {
+    await fixture.cleanup();
+  }
+});
+
 test("listProviderAvailability uses registered client keys, including custom providers", async () => {
   const customClient: AgentClient = {
     provider: "zai",

packages/server/src/server/agent/agent-manager.ts

@@ -3156,6 +3156,7 @@ export class AgentManager {
         { agentId, turnId: runTurnId, kind: run.kind },
         "cancelAgentRun: acknowledged turn still active after timeout, force-canceling",
       );
+      await this.releaseAbandonedTurn(agent);
       await this.dispatchSessionEvent(agent, {
         type: "turn_canceled",
         provider: agent.provider,
@@ -3168,6 +3169,7 @@ export class AgentManager {
         { agentId, kind: run.kind },
         "cancelAgentRun: acknowledged pending turn still active after timeout, clearing it",
       );
+      await this.releaseAbandonedTurn(agent);
       this.runs.settleForegroundRun(agentId, run.token);
       if (!agent.pendingReplacement) {
         agent.lifecycle = "idle";
@@ -3194,6 +3196,26 @@ export class AgentManager {
     return { status: "settled" };
   }
 
+  /**
+   * After a forced settlement, stop reusing a runtime whose foreground turn it never confirmed
+   * ending. Providers that cannot reconcile omit the hook and keep today's behavior.
+   *
+   * A rejected release leaves the agent unusable, so log it and continue: the manager already
+   * settled the run, and refusing here would report a failure whose only cause is the provider.
+   */
+  private async releaseAbandonedTurn(agent: ActiveManagedAgent): Promise<void> {
+    const abandon = agent.session.abandonForegroundTurn;
+    if (!abandon) return;
+    try {
+      await abandon.call(agent.session);
+    } catch (error) {
+      this.logger.warn(
+        { err: error, agentId: agent.id, provider: agent.provider },
+        "cancelAgentRun: provider could not release the abandoned foreground turn",
+      );
+    }
+  }
+
   private async cancelAgentRunBefore(
     agentId: string,
     action: "reload" | "replace" | "rewind",

packages/server/src/server/agent/agent-sdk-types.ts

@@ -694,6 +694,17 @@ export interface AgentSession {
    * still uncertain.
    */
   interrupt(): Promise<void>;
+  /**
+   * Drop ownership of a foreground turn the provider did not confirm as ended, after `interrupt()`
+   * was acknowledged but the run never settled. The manager calls this only when it has already
+   * force-canceled its own side of that run, so the daemon and the provider must not stay split.
+   *
+   * A provider that implements this must make the session usable again: release the turn slot and
+   * stop trusting the runtime that owned it (kill, replace, or resume it). Reject when the runtime
+   * cannot be released, which keeps the agent in its previous state. Omit it when the provider has
+   * no safe way to reconcile; the manager keeps today's behavior.
+   */
+  abandonForegroundTurn?(): Promise<void>;
   /** Release live runtime resources without archiving or deleting the durable native session. */
   close(): Promise<void>;
   listCommands?(): Promise<AgentSlashCommand[]>;

packages/server/src/server/agent/provider-registry-wrap.test.ts

@@ -22,6 +22,7 @@ type OptionalAgentSessionMethodName = {
 
 const OPTIONAL_AGENT_SESSION_METHOD_NAMES = [
   "steerActiveTurn",
+  "abandonForegroundTurn",
   "listCommands",
   "setModel",
   "setThinkingOption",
@@ -168,6 +169,10 @@ class FakeSession implements AgentSession {
     this.recordedCalls.push("revertBoth");
   }
 
+  async abandonForegroundTurn() {
+    this.recordedCalls.push("abandonForegroundTurn");
+  }
+
   tryHandleOutOfBand(_prompt: AgentPromptInput) {
     this.recordedCalls.push("tryHandleOutOfBand");
     return {
@@ -190,6 +195,7 @@ describe("wrapSessionProvider", () => {
     const wrapped = wrapSessionProvider("custom-claude", session);
 
     await wrapped.steerActiveTurn?.("follow-up", { expectedTurnId: "turn-1" });
+    await wrapped.abandonForegroundTurn?.();
     await wrapped.listCommands?.();
     await wrapped.setModel?.("sonnet");
     await wrapped.setThinkingOption?.("high");
@@ -205,6 +211,7 @@ describe("wrapSessionProvider", () => {
     ]);
     expect(session.recordedCalls).toEqual([
       "steerActiveTurn",
+      "abandonForegroundTurn",
       "listCommands",
       "setModel",
       "setThinkingOption",
@@ -246,4 +253,10 @@ describe("wrapSessionProvider", () => {
     session.steerActiveTurn = undefined;
     expect(wrapSessionProvider("custom-claude", session).steerActiveTurn).toBeUndefined();
   });
+
+  test("leaves abandoned-turn release unavailable when the provider has no implementation", () => {
+    const session: AgentSession = new FakeSession();
+    session.abandonForegroundTurn = undefined;
+    expect(wrapSessionProvider("custom-claude", session).abandonForegroundTurn).toBeUndefined();
+  });
 });

packages/server/src/server/agent/provider-registry.ts

@@ -468,6 +468,7 @@ export function wrapSessionProvider(provider: AgentProvider, inner: AgentSession
     respondToPermission: (requestId, response) => inner.respondToPermission(requestId, response),
     describePersistence: () => mapPersistenceHandle(provider, inner.describePersistence()),
     interrupt: () => inner.interrupt(),
+    abandonForegroundTurn: inner.abandonForegroundTurn?.bind(inner),
     close: () => inner.close(),
     listCommands: inner.listCommands?.bind(inner),
     setModel: inner.setModel?.bind(inner),

packages/server/src/server/agent/providers/acp-abandon-kill-timeout.repro.test.ts

@@ -0,0 +1,84 @@
+import { mkdtemp, writeFile, readFile, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { createRequire } from "node:module";
+import { pathToFileURL } from "node:url";
+import { expect, test } from "vitest";
+import { ACPAgentSession } from "./acp-agent.js";
+import { createTestLogger } from "../../../test-utils/test-logger.js";
+import { terminateWithTreeKill } from "../../../utils/tree-kill.js";
+
+const source = `
+import { appendFileSync } from 'node:fs';
+import { Readable, Writable } from 'node:stream';
+const { AgentSideConnection, PROTOCOL_VERSION, ndJsonStream } = await import(process.env.ACP_SDK_URL);
+appendFileSync(process.env.PROBE_PIDS, String(process.pid) + '\\n');
+new AgentSideConnection(() => ({
+  async initialize() { return { protocolVersion: PROTOCOL_VERSION, agentCapabilities: { loadSession: true } }; },
+  async newSession() { return { sessionId: 'held-session' }; },
+  async loadSession() { return {}; },
+  async authenticate() {},
+  prompt() { return new Promise(() => {}); },
+  async cancel() {},
+}), ndJsonStream(Writable.toWeb(process.stdout), Readable.toWeb(process.stdin)));
+`;
+
+test("a failed abandoned-runtime kill must block a second provider process", async () => {
+  const dir = await mkdtemp(path.join(tmpdir(), "acp-abandon-timeout-"));
+  const script = path.join(dir, "agent.mjs");
+  const pidFile = path.join(dir, "pids");
+  await writeFile(script, source);
+  let allowTermination = false;
+  const session = new ACPAgentSession(
+    { provider: "abandon-probe", cwd: dir },
+    {
+      provider: "abandon-probe",
+      logger: createTestLogger(),
+      defaultCommand: [process.execPath, script],
+      defaultModes: [],
+      capabilities: {
+        supportsStreaming: true,
+        supportsSessionPersistence: true,
+        supportsDynamicModes: true,
+        supportsMcpServers: true,
+        supportsReasoningStream: true,
+        supportsToolInvocations: true,
+      },
+      launchEnv: {
+        ACP_SDK_URL: pathToFileURL(
+          createRequire(import.meta.url).resolve("@agentclientprotocol/sdk"),
+        ).href,
+        PROBE_PIDS: pidFile,
+      },
+      // Inject the documented termination result; the ACP subprocess itself is real.
+      terminateProcess: (child, options) =>
+        allowTermination ? terminateWithTreeKill(child, options) : Promise.resolve("kill-timeout"),
+    },
+  );
+  try {
+    await session.initializeNewSession();
+    await session.startTurn("first");
+    await expect(session.abandonForegroundTurn()).rejects.toThrow("provider process did not exit");
+    const firstPid = Number((await readFile(pidFile, "utf8")).trim());
+    expect(() => process.kill(firstPid, 0)).not.toThrow();
+    const next = await session.startTurn("second").then(
+      () => "started",
+      () => "refused",
+    );
+    const pids = (await readFile(pidFile, "utf8")).trim().split("\n").map(Number);
+    for (const pid of pids) expect(() => process.kill(pid, 0)).not.toThrow();
+    console.log({ next, liveProviderProcesses: pids.length });
+    expect(next).toBe("refused");
+    expect(pids).toHaveLength(1);
+  } finally {
+    allowTermination = true;
+    await session.close();
+    const pids = await readFile(pidFile, "utf8").catch(() => "");
+    for (const pid of pids.trim().split("\n").filter(Boolean).map(Number)) {
+      try {
+        process.kill(pid, "SIGKILL");
+      } catch {}
+    }
+    await rm(dir, { recursive: true, force: true });
+  }
+});

packages/server/src/server/agent/providers/acp-agent.test.ts

@@ -4269,6 +4269,125 @@ describe("ACP session/load invariant — cwd and mcpServers always passed", () =
   });
 });
 
+describe("ACP abandoned foreground turn", () => {
+  /**
+   * A session whose spawn is stubbed, so tests can hold a foreground turn that never settles
+   * while asserting what the release path does to the runtime that owned it.
+   */
+  function makeDownturnSession(options: { terminator?: ProcessTerminator } = {}) {
+    const loadSession = vi.fn().mockResolvedValue({
+      sessionId: "session-1",
+      modes: null,
+      models: null,
+      configOptions: [],
+    });
+    const prompt = vi.fn(() => new Promise<PromptResponse>(() => {}));
+
+    class TestSession extends ACPAgentSession {
+      spawnedChildren = 0;
+
+      protected override async spawnProcess(): Promise<SpawnedACPProcess> {
+        this.spawnedChildren += 1;
+        return {
+          child: createProbeChildStub(),
+          connection: {
+            prompt,
+            loadSession,
+            unstable_resumeSession: loadSession,
+          } as unknown as ClientSideConnection,
+          initialize: { agentCapabilities: { loadSession: true } },
+        } as SpawnedACPProcess;
+      }
+    }
+
+    const session = new TestSession(
+      { provider: "claude-acp", cwd: "/tmp/paseo-acp-test" },
+      {
+        provider: "claude-acp",
+        logger: createTestLogger(),
+        defaultCommand: ["claude", "--acp"],
+        defaultModes: [],
+        capabilities: {
+          supportsStreaming: true,
+          supportsSessionPersistence: true,
+          supportsDynamicModes: true,
+          supportsMcpServers: true,
+          supportsReasoningStream: true,
+          supportsToolInvocations: true,
+        },
+        handle: { sessionId: "session-1", provider: "claude-acp" },
+        ...(options.terminator ? { terminateProcess: options.terminator } : {}),
+      },
+    );
+    return { session, prompt, loadSession };
+  }
+
+  test("releases the turn slot and stops the runtime that never answered cancel", async () => {
+    const terminator = new FakeTerminator();
+    const { session } = makeDownturnSession({ terminator: terminator.terminate });
+    await session.initializeResumedSession();
+    await session.startTurn("first");
+    expect(asInternals<ACPSessionInternals>(session).activeForegroundTurnId).toBeTruthy();
+
+    await session.abandonForegroundTurn();
+
+    expect(asInternals<ACPSessionInternals>(session).activeForegroundTurnId).toBeNull();
+    expect(terminator.terminated).toHaveLength(1);
+  });
+
+  test("the next prompt reconnects in a new process instead of failing on the held slot", async () => {
+    const terminator = new FakeTerminator();
+    const { session } = makeDownturnSession({ terminator: terminator.terminate });
+    await session.initializeResumedSession();
+    await session.startTurn("first");
+    await session.abandonForegroundTurn();
+
+    await expect(session.startTurn("second")).resolves.toEqual({ turnId: expect.any(String) });
+
+    expect(terminator.terminated).toHaveLength(1);
+    // One spawn for the first turn, one for the resume the release forced.
+    expect((session as unknown as { spawnedChildren: number }).spawnedChildren).toBe(2);
+    expect(asInternals<ACPSessionInternals>(session).activeForegroundTurnId).toBeTruthy();
+  });
+
+  test("does not replay history a second time when it reconnects", async () => {
+    const terminator = new FakeTerminator();
+    const { session } = makeDownturnSession({ terminator: terminator.terminate });
+    await session.initializeResumedSession();
+    const historyBefore = [];
+    for await (const event of session.streamHistory()) {
+      historyBefore.push(event);
+    }
+    await session.startTurn("first");
+    await session.abandonForegroundTurn();
+    await session.startTurn("second");
+
+    const replay = [];
+    for await (const event of session.streamHistory()) {
+      replay.push(event);
+    }
+    expect(replay).toEqual([]);
+  });
+
+  test("rejects when the provider process cannot be stopped", async () => {
+    const session = makeDownturnSession({
+      terminator: async () => "kill-timeout",
+    }).session;
+    await session.initializeResumedSession();
+    await session.startTurn("first");
+
+    await expect(session.abandonForegroundTurn()).rejects.toThrow(
+      "the provider process did not exit",
+    );
+    expect(asInternals<ACPSessionInternals>(session).activeForegroundTurnId).toBeNull();
+  });
+
+  test("is a no-op on a session that has no id yet", async () => {
+    const session = makeDownturnSession().session;
+    await expect(session.abandonForegroundTurn()).resolves.toBeUndefined();
+  });
+});
+
 const SILENT_CLOSE_ACP_AGENT = `
 import { readFileSync, writeFileSync } from "node:fs";
 import { Readable, Writable } from "node:stream";

packages/server/src/server/agent/providers/acp-agent.ts

@@ -1734,7 +1734,7 @@ export class ACPAgentSession implements AgentSession, ACPClient {
   private readonly toolCalls = new Map<string, ACPToolSnapshot>();
   private readonly terminalEntries = new Map<string, TerminalEntry>();
   private readonly persistedHistory: AgentTimelineItem[] = [];
-  private readonly initialHandle?: AgentPersistenceHandle;
+  private initialHandle?: AgentPersistenceHandle;
   private readonly resumePurpose: AgentResumePurpose;
 
   private readonly config: AgentSessionConfig;
@@ -1927,6 +1927,33 @@ export class ACPAgentSession implements AgentSession, ACPClient {
     this.settleCommandsReady();
   }
 
+  /**
+   * Resume the durable session in a fresh process after `abandonForegroundTurn()` killed the
+   * runtime that owned the abandoned turn.
+   *
+   * The replay that `initializeResumedSession()` drains is duplicate history: the daemon already
+   * has every row up to the abandoned turn. Drop the replayed rows and keep the prior
+   * `historyPending` flag so the timeline is not appended twice.
+   */
+  private async reconnectAbandonedRuntime(): Promise<void> {
+    if (this.connection || !this.sessionId || this.closed) {
+      return;
+    }
+    // A session created in this process never received a resume handle; its durable id is the
+    // handle, the same object `describePersistence()` publishes.
+    this.initialHandle ??= {
+      provider: this.provider,
+      sessionId: this.sessionId,
+      nativeHandle: this.sessionId,
+    };
+    const historyLength = this.persistedHistory.length;
+    const historyPending = this.historyPending;
+    await this.initializeResumedSession();
+    this.persistedHistory.length = historyLength;
+    this.historyPending = historyPending;
+    this.bootstrapThreadEventPending = false;
+  }
+
   private async closeAfterInitializationFailure(error: unknown): Promise<never> {
     try {
       await this.close();
@@ -1963,6 +1990,7 @@ export class ACPAgentSession implements AgentSession, ACPClient {
     if (this.closed) {
       throw new Error(`${this.provider} session is closed`);
     }
+    await this.reconnectAbandonedRuntime();
     if (!this.connection || !this.sessionId) {
       throw new Error(`${this.provider} session is not initialized`);
     }
@@ -2554,6 +2582,46 @@ export class ACPAgentSession implements AgentSession, ACPClient {
     }
   }
 
+  /**
+   * ACP turns end when `session/prompt` returns. The manager force-cancels after two seconds
+   * without that answer, so an agent that keeps working after `session/cancel` leaves this
+   * session's turn slot held. The next `startTurn` would then throw "A foreground turn is
+   * already active" on every later prompt, because the daemon believes the agent is idle.
+   *
+   * The runtime that still owns that turn cannot be trusted, so this kills it and drops the
+   * slot. The next prompt respawns through `initializeResumedSession()`; `startTurn` also
+   * reconnects when it finds no connection, which covers the very next message.
+   */
+  async abandonForegroundTurn(): Promise<void> {
+    if (!this.sessionId) {
+      return;
+    }
+
+    this.synthesizeCanceledToolCalls();
+    this.activeForegroundTurnId = null;
+    this.fallbackAssistantMessageId = null;
+    this.submittedUserMessageTurnId = null;
+
+    const child = this.child;
+    this.connection = null;
+    this.child = null;
+    // A command from createTerminal is a peer of the agent process. The dead connection can no
+    // longer release it, so stop those terminals here as well.
+    const terminations = Array.from(this.terminalEntries.values(), (terminal) =>
+      this.terminateProcess(terminal.child, { gracefulTimeoutMs: 200, forceTimeoutMs: 500 }),
+    );
+    this.terminalEntries.clear();
+    if (child) {
+      terminations.push(
+        this.terminateProcess(child, { gracefulTimeoutMs: 200, forceTimeoutMs: 500 }),
+      );
+    }
+    const results = await Promise.all(terminations);
+    if (results.includes("kill-timeout")) {
+      throw new Error("ACP abandoned turn: the provider process did not exit");
+    }
+  }
+
   async close(): Promise<void> {
     if (this.closed) {
       return;