fix(server): reap recorded Linux Claude survivors before recovery

apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts

@@ -82,6 +82,7 @@ import {
   mcpAppToolResourceUri,
 } from "@t3tools/shared/mcpApp";
 import { AgentScope } from "@t3tools/shared/AgentScope";
+import * as ProviderProcesses from "../../process/providerProcesses.ts";
 import * as HostProcess from "@t3tools/shared/HostProcess";
 import { applyClaudePromptEffortPrefix } from "@t3tools/shared/model";
 import {
@@ -358,21 +359,26 @@ const CLAUDE_STDERR_DRAIN_TIMEOUT = Duration.seconds(1);
 const CLAUDE_ROOT_REFUSAL =
   "--dangerously-skip-permissions cannot be used with root/sudo privileges for security reasons";
 
+// fd 4 gates the launch until the server has recorded the process it is about to become.
 const CLAUDE_GROUP_WATCH_SCRIPT = [
+  "read -r _ <&4 || exit 1",
+  "exec 4<&-",
   "{ trap '' TERM; read -r _ <&3; kill -TERM 0; sleep 5; kill -KILL 0; } </dev/null >/dev/null 2>&1 &",
   'exec "$@" 3<&-',
 ].join("\n");
 
 /**
  * Kill the CLI's group on exit to stop MCP grandchildren; `onExit` stops its agent scope for
  * setsid daemons. The CLI keeps working after its pipes close, so a watcher stops the group on
- * EOF from a pipe only this server holds, including when the server crashes.
+ * EOF from a pipe only this server holds, including when the server crashes. `register` records
+ * the CLI before it may run, so a restarted server can reap it if the watcher did not.
  */
 export function spawnClaudeProcessGroup(
   options: ClaudeSpawnOptions,
   onExit?: () => void,
   onStderr: (text: string) => void = () => {},
   onStderrClose: () => void = () => {},
+  register?: (pid: number) => () => void,
 ): ClaudeSpawnedProcess {
   const child = NodeChildProcess.spawn(
     "/bin/sh",
@@ -381,18 +387,34 @@ export function spawnClaudeProcessGroup(
       cwd: options.cwd,
       env: options.env,
       signal: options.signal,
-      stdio: ["pipe", "pipe", "pipe", "pipe"],
+      stdio: ["pipe", "pipe", "pipe", "pipe", "pipe"],
       detached: true,
     },
   );
+  let forget: (() => void) | undefined;
   // Always drained, so a chatty CLI never blocks on a full stderr pipe.
   child.stderr?.setEncoding("utf8").on("data", onStderr).once("close", onStderrClose);
   child.once("exit", () => {
     try {
       signalProcessGroup(child.pid ?? 0, "SIGKILL");
     } catch {}
+    try {
+      forget?.();
+    } catch {}
     onExit?.();
   });
+  const gate = child.stdio[4];
+  if (child.pid !== undefined && gate !== null && gate !== undefined && "end" in gate) {
+    try {
+      forget = register?.(child.pid);
+      gate.on("error", () => child.kill("SIGKILL"));
+      gate.end("ready\n");
+    } catch (cause) {
+      child.kill("SIGKILL");
+      gate.destroy();
+      throw cause;
+    }
+  }
   // Node's typings make stdin and stdout nullable once stdio has a fourth entry; both are pipes.
   return child as ClaudeSpawnedProcess;
 }
@@ -1163,6 +1185,7 @@ export const layerQueryRunner: Layer.Layer<
     const path = yield* Path.Path;
     const { native: nativeEventLogger } = yield* ProviderEventLoggers.ProviderEventLoggers;
     const agentScope = yield* AgentScope;
+    const providerProcesses = yield* ProviderProcesses.ProviderProcesses;
     const platform = yield* HostProcess.Platform;
     const runFork = Effect.runForkWith(yield* Effect.context<never>());
     const hostEnvironment = yield* HostProcess.Environment;
@@ -1238,7 +1261,13 @@ export const layerQueryRunner: Layer.Layer<
                 // Node can report "exit" before it has read the stderr the CLI wrote.
                 let stderrDone = () => {};
                 stderrClosed = new Promise((resolve) => (stderrDone = resolve));
-                return spawnClaudeProcessGroup(spawnOptions, stopScope, appendStderr, stderrDone);
+                return spawnClaudeProcessGroup(
+                  spawnOptions,
+                  stopScope,
+                  appendStderr,
+                  stderrDone,
+                  providerProcesses.register,
+                );
               }
               const child = NodeChildProcess.spawn(spawnOptions.command, spawnOptions.args, {
                 cwd: spawnOptions.cwd,

apps/server/src/orchestration-v2/Adapters/ClaudeAgentSdkQueryRunner.test.ts

@@ -655,6 +655,47 @@ describe.skipIf(HostProcess.Platform.defaultValue() === "win32")("Claude CLI pro
     }),
   );
 
+  // A restarted server reaps recorded CLIs, so a CLI must never run before its record exists.
+  it.effect("records the CLI before it runs and forgets it once it exits", () =>
+    Effect.gen(function* () {
+      const cli = fakeClaudeCli(
+        ['dir=$(dirname "$0")', 'printf \'%s %s\' "$$" "$(cat "$dir/record")" > "$dir/seen"'].join(
+          "\n",
+        ),
+      );
+      const dir = NodePath.dirname(cli);
+      NodeChildProcess.execFileSync("mkfifo", [NodePath.join(dir, "seen")]);
+      yield* Effect.addFinalizer(() =>
+        Effect.sync(() => NodeFS.rmSync(dir, { recursive: true, force: true })),
+      );
+      const forgotten = yield* Deferred.make<void>();
+      let registered = 0;
+      spawnClaudeProcessGroup(
+        {
+          command: cli,
+          args: [],
+          cwd: dir,
+          env: process.env,
+          // @effect-diagnostics-next-line abortControllerInEffect:off - The CLI exits on its own.
+          signal: new AbortController().signal,
+        },
+        undefined,
+        undefined,
+        undefined,
+        (pid) => {
+          registered = pid;
+          NodeFS.writeFileSync(NodePath.join(dir, "record"), String(pid));
+          return () => Deferred.doneUnsafe(forgotten, Effect.void);
+        },
+      );
+      const seen = yield* Effect.promise(() =>
+        NodeFS.promises.readFile(NodePath.join(dir, "seen"), "utf8"),
+      );
+      assert.equal(seen, `${registered} ${registered}`);
+      yield* Deferred.await(forgotten);
+    }),
+  );
+
   it.effect("marks the CLI's jobs with its thread and counts it as an agent until it exits", () => {
     let runningAgents = 0;
     let releaseAgent!: () => void;

apps/server/src/process/agentScope.test.ts

@@ -1,12 +1,16 @@
 // @effect-diagnostics nodeBuiltinImport:off -- fake systemd binaries live in a temp dir.
+import * as NodeChildProcess from "node:child_process";
+import * as NodeCrypto from "node:crypto";
 import * as NodeFS from "node:fs";
 import * as NodeOS from "node:os";
 import * as NodePath from "node:path";
 
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import { assert, describe, expect, it } from "@effect/vitest";
+import * as AcpSessionRuntime from "@t3tools/provider-acp/server/AcpSessionRuntime";
 import { AgentScope } from "@t3tools/shared/AgentScope";
 import * as HostProcess from "@t3tools/shared/HostProcess";
+import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
@@ -16,7 +20,8 @@ import { ChildProcess, ChildProcessSpawner } from "effect/process";
 import * as ServerConfig from "../config.ts";
 import { makeCodexAppServerSpawnCommand } from "../orchestration-v2/Adapters/CodexAdapterV2.ts";
 import * as ProcessRunner from "../processRunner.ts";
-import { classifyScope, make, ownedAgentSlice, type ScopeState } from "./agentScope.ts";
+import * as ProviderProcesses from "./providerProcesses.ts";
+import { classifyScope, layer, make, ownedAgentSlice, type ScopeState } from "./agentScope.ts";
 
 const running: ScopeState = {
   loadState: "loaded",
@@ -55,7 +60,14 @@ const withFakeSystemd = <A, E>(
   {
     oomPolicy = true,
     stopFails = false,
-  }: { readonly oomPolicy?: boolean; readonly stopFails?: boolean } = {},
+    leftovers = false,
+    liveOwner = false,
+  }: {
+    readonly oomPolicy?: boolean;
+    readonly stopFails?: boolean;
+    readonly leftovers?: boolean;
+    readonly liveOwner?: boolean;
+  } = {},
 ) =>
   Effect.gen(function* () {
     const bin = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-agent-scope-"));
@@ -102,7 +114,12 @@ const withFakeSystemd = <A, E>(
         }),
     });
     const fileSystem = yield* FileSystem.FileSystem;
+    const { stateDir } = yield* ServerConfig.ServerConfig;
+    if (leftovers) {
+      units.set(ownedAgentSlice(stateDir), { activeState: "active", result: "success" });
+    }
     const scope = yield* make.pipe(
+      Effect.provideService(ProviderProcesses.ProviderProcesses, { hasOtherLiveOwner: liveOwner }),
       Effect.provideService(ProcessRunner.ProcessRunner, runner),
       Effect.provideService(HostProcess.Platform, "linux"),
       Effect.provideService(HostProcess.Environment, { PATH: bin, XDG_RUNTIME_DIR: "/run/user/1" }),
@@ -114,7 +131,6 @@ const withFakeSystemd = <A, E>(
           fileSystem.writeFileString(path.replace("/sys/fs/cgroup", cgroupRoot), data, options),
       }),
     );
-    const { stateDir } = yield* ServerConfig.ServerConfig;
     return yield* body({ scope, units, calls, stateDir, cgroupRoot }).pipe(
       Effect.ensuring(Effect.sync(() => NodeFS.rmSync(bin, { recursive: true, force: true }))),
     );
@@ -308,27 +324,34 @@ describe("AgentScope service", () => {
     ),
   );
 
-  it.effect("stops agents a killed server left before the first launch", () =>
-    withFakeSystemd(({ scope, units, calls, stateDir }) =>
-      Effect.gen(function* () {
-        units.set(ownedAgentSlice(stateDir), { activeState: "active", result: "success" });
-        const agent = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
-        const stops = calls.filter(([, verb]) => verb === "stop");
-        expect(stops).toEqual([["--user", "stop", ownedAgentSlice(stateDir)]]);
-        expect(sliceOf(agent)).toBe(ownedAgentSlice(stateDir));
-        expect(agent.args).toContain("--property=TimeoutStopSec=10s");
-        const terminal = yield* scope.wrap({
-          command: "agent",
-          args: [],
-          name: "terminal",
-          survivesRestart: true,
-        });
-        expect(sliceOf(terminal)).toBe("app-t3code-agents.slice");
-        expect(calls.filter(([, verb]) => verb === "stop")).toHaveLength(1);
-      }),
+  it.effect("stops agents a killed server left before startup completes", () =>
+    withFakeSystemd(
+      ({ scope, calls, stateDir }) =>
+        Effect.gen(function* () {
+          expect(stopsOf(calls)).toEqual([["--user", "stop", ownedAgentSlice(stateDir)]]);
+          const agent = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
+          expect(sliceOf(agent)).toBe(ownedAgentSlice(stateDir));
+          expect(agent.args).toContain("--property=TimeoutStopSec=10s");
+          const terminal = yield* scope.wrap({
+            command: "agent",
+            args: [],
+            name: "terminal",
+            survivesRestart: true,
+          });
+          expect(sliceOf(terminal)).toBe("app-t3code-agents.slice");
+          expect(stopsOf(calls)).toHaveLength(1);
+        }),
+      { leftovers: true },
     ),
   );
 
+  it.effect("leaves the slice to another live server sharing this home", () =>
+    withFakeSystemd(({ calls }) => Effect.sync(() => expect(stopsOf(calls)).toEqual([])), {
+      leftovers: true,
+      liveOwner: true,
+    }),
+  );
+
   it.effect("stops leftover processes only in an active agent scope", () =>
     withFakeSystemd(({ scope, units, calls, cgroupRoot }) =>
       Effect.gen(function* () {
@@ -415,3 +438,116 @@ describe("AgentScope service", () => {
     ),
   );
 });
+
+// Real child processes stand in for servers and CLIs; only PIDs captured at spawn are signalled.
+const ownedProcess = Effect.gen(function* () {
+  const ready = yield* Deferred.make<void>();
+  const exited = yield* Deferred.make<void>();
+  const child = yield* Effect.acquireRelease(
+    Effect.sync(() => {
+      const child = NodeChildProcess.spawn(
+        process.execPath,
+        ["-e", 'process.stdin.resume(); process.send("ready");'],
+        { stdio: ["pipe", "ignore", "ignore", "ipc"] },
+      );
+      child.once("message", () => Deferred.doneUnsafe(ready, Effect.void));
+      child.once("error", (error) => Deferred.doneUnsafe(ready, Effect.die(error)));
+      child.once("exit", () => Deferred.doneUnsafe(exited, Effect.void));
+      return child;
+    }),
+    (child) =>
+      Effect.sync(() => child.kill("SIGKILL")).pipe(Effect.andThen(Deferred.await(exited))),
+  );
+  yield* Deferred.await(ready);
+  const pid = child.pid;
+  assert(pid !== undefined);
+  const controller = AcpSessionRuntime.makePosixProcessTreeController("linux");
+  const identity = controller.identity(pid);
+  assert(identity !== undefined);
+  const boot = NodeFS.readFileSync("/proc/sys/kernel/random/boot_id", "utf8").trim();
+  return {
+    pid,
+    startTime: `${boot}:${identity.startTime}`,
+    pgid: 0,
+    running: () => {
+      const observed = controller.identity(pid);
+      return observed !== undefined && !AcpSessionRuntime.posixProcessIsZombie(observed);
+    },
+    stop: Effect.sync(() => child.kill("SIGKILL")).pipe(Effect.andThen(Deferred.await(exited))),
+  };
+});
+
+describe.runIf(HostProcess.Platform.defaultValue() === "linux")(
+  "provider restart ownership",
+  () => {
+    it.live("reaps only recorded CLIs of dead servers before startup completes", () =>
+      Effect.gen(function* () {
+        const config = yield* ServerConfig.ServerConfig;
+        const directory = NodePath.join(config.stateDir, "provider-processes");
+        NodeFS.mkdirSync(directory, { recursive: true });
+        type Identity = { readonly pid: number; readonly startTime: string };
+        const record = (owner: Identity, processes: ReadonlyArray<Identity>) => {
+          const instanceId = NodeCrypto.randomUUID();
+          NodeFS.writeFileSync(
+            NodePath.join(directory, `${instanceId}.json`),
+            JSON.stringify({
+              version: 1,
+              instanceId,
+              hostname: NodeOS.hostname(),
+              stateDir: config.stateDir,
+              owner: { pid: owner.pid, startTime: owner.startTime },
+              processes: processes.map(({ pid, startTime }) => ({ pid, startTime, pgid: 0 })),
+            }),
+          );
+        };
+        const reused = (identity: Identity) => ({
+          ...identity,
+          startTime: identity.startTime.replace(/:\d+$/, ":1"),
+        });
+        const deadOwner = yield* ownedProcess;
+        const liveOwner = yield* ownedProcess;
+        const reusedOwner = yield* ownedProcess;
+        const liveProvider = yield* ownedProcess;
+        const reusedProvider = yield* ownedProcess;
+        const orphan = yield* ownedProcess;
+        const orphanOfReusedOwner = yield* ownedProcess;
+        const unrecorded = yield* ownedProcess;
+        record(liveOwner, [liveProvider]);
+        record(deadOwner, [reused(reusedProvider)]);
+        record(deadOwner, [orphan]);
+        record(reused(reusedOwner), [orphanOfReusedOwner]);
+        yield* deadOwner.stop;
+
+        yield* Layer.build(layer).pipe(
+          Effect.provideService(ServerConfig.ServerConfig, config),
+          Effect.provide(NodeServices.layer),
+        );
+
+        expect({
+          orphan: orphan.running(),
+          orphanOfReusedOwner: orphanOfReusedOwner.running(),
+          liveProvider: liveProvider.running(),
+          reusedProvider: reusedProvider.running(),
+          reusedOwner: reusedOwner.running(),
+          unrecorded: unrecorded.running(),
+        }).toEqual({
+          orphan: false,
+          orphanOfReusedOwner: false,
+          liveProvider: true,
+          reusedProvider: true,
+          reusedOwner: true,
+          unrecorded: true,
+        });
+      }).pipe(
+        Effect.scoped,
+        Effect.provideService(HostProcess.Platform, "linux"),
+        Effect.provideService(HostProcess.Environment, { PATH: process.env.PATH }),
+        Effect.provide(
+          ServerConfig.layerTest(process.cwd(), { prefix: "t3-restart-ownership-" }).pipe(
+            Layer.provideMerge(NodeServices.layer),
+          ),
+        ),
+      ),
+    );
+  },
+);