fix(process): never start cancelled queued replacements (#114951)

src/process/supervisor/supervisor.queued-cancellation.test.ts

@@ -0,0 +1,182 @@
+// Queued supervisor replacements must not launch after their caller cancels.
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { createDeferred } from "../../test-utils/deferred.js";
+import { createProcessSupervisor } from "./supervisor.js";
+import type { SpawnInput, SpawnProcessAdapter } from "./types.js";
+
+const { createChildAdapterMock, createPtyAdapterMock } = vi.hoisted(() => ({
+  createChildAdapterMock: vi.fn(),
+  createPtyAdapterMock: vi.fn(),
+}));
+
+vi.mock("./adapters/child.js", () => ({
+  createChildAdapter: createChildAdapterMock,
+}));
+
+vi.mock("./adapters/pty.js", () => ({
+  createPtyAdapter: createPtyAdapterMock,
+}));
+
+type StubProcessAdapter = SpawnProcessAdapter<NodeJS.Signals | null> & {
+  killMock: ReturnType<typeof vi.fn>;
+  settle: (code: number | null, signal?: NodeJS.Signals | null) => void;
+};
+
+function createStubProcessAdapter(pid = 1234): StubProcessAdapter {
+  const completion = createDeferred<{ code: number | null; signal: NodeJS.Signals | null }>();
+  const killMock = vi.fn();
+  return {
+    pid,
+    onStdout: () => undefined,
+    onStderr: () => undefined,
+    wait: async () => completion.promise,
+    kill: (signal) => killMock(signal),
+    dispose: () => undefined,
+    killMock,
+    settle: (code, signal = null) => completion.resolve({ code, signal }),
+  };
+}
+
+function createSpawnInput(params: {
+  runId: string;
+  scopeKey: string;
+  mode?: "child" | "pty";
+  replaceExistingScope?: boolean;
+}): SpawnInput {
+  const common = {
+    runId: params.runId,
+    sessionId: "queued-cancellation",
+    backendId: "test",
+    scopeKey: params.scopeKey,
+    replaceExistingScope: params.replaceExistingScope,
+  };
+  return params.mode === "pty"
+    ? { ...common, mode: "pty", ptyCommand: "printf should-not-run" }
+    : {
+        ...common,
+        mode: "child",
+        argv: [process.execPath, "-e", "process.stdout.write('should-not-run')"],
+      };
+}
+
+describe("process supervisor queued cancellation", () => {
+  beforeEach(() => {
+    createChildAdapterMock.mockReset();
+    createPtyAdapterMock.mockReset();
+  });
+
+  it.each(["child", "pty"] as const)(
+    "does not start an already-cancelled queued %s replacement",
+    async (mode) => {
+      const first = createStubProcessAdapter();
+      const replacement = createStubProcessAdapter();
+      const firstStartup = createDeferred<StubProcessAdapter>();
+      createChildAdapterMock.mockReturnValueOnce(firstStartup.promise);
+      if (mode === "pty") {
+        createPtyAdapterMock.mockResolvedValueOnce(replacement);
+      } else {
+        createChildAdapterMock.mockResolvedValueOnce(replacement);
+      }
+
+      const supervisor = createProcessSupervisor();
+      const scopeKey = "scope:cancel-queued";
+      const firstRunPromise = supervisor.spawn(
+        createSpawnInput({ runId: `cancel-queued-${mode}-first`, scopeKey }),
+      );
+      const replacementRunId = `cancel-queued-${mode}-replacement`;
+      const replacementPromise = supervisor.spawn(
+        createSpawnInput({
+          runId: replacementRunId,
+          scopeKey,
+          mode,
+          replaceExistingScope: true,
+        }),
+      );
+
+      expect(createChildAdapterMock).toHaveBeenCalledTimes(1);
+      expect(createPtyAdapterMock).not.toHaveBeenCalled();
+
+      supervisor.cancel(replacementRunId, "manual-cancel");
+      firstStartup.resolve(first);
+      const [firstRun, replacementRun] = await Promise.all([firstRunPromise, replacementPromise]);
+
+      expect(createChildAdapterMock).toHaveBeenCalledTimes(1);
+      expect(createPtyAdapterMock).not.toHaveBeenCalled();
+      expect(first.killMock).not.toHaveBeenCalled();
+      expect(replacement.killMock).not.toHaveBeenCalled();
+      expect(replacementRun.pid).toBeUndefined();
+      await expect(replacementRun.wait()).resolves.toMatchObject({
+        reason: "manual-cancel",
+        exitCode: null,
+        exitSignal: null,
+      });
+      expect(supervisor.getRecord(replacementRunId)).toMatchObject({
+        state: "exited",
+        terminationReason: "manual-cancel",
+      });
+
+      first.settle(0);
+      await expect(firstRun.wait()).resolves.toMatchObject({ reason: "exit" });
+    },
+  );
+
+  it("never starts cancelled queued replacements or cancels their surviving scope", async () => {
+    const replacementCount = 32;
+    const first = createStubProcessAdapter(1234);
+    const later = createStubProcessAdapter(1235);
+    const firstStartup = createDeferred<StubProcessAdapter>();
+    createChildAdapterMock.mockReturnValueOnce(firstStartup.promise).mockResolvedValueOnce(later);
+
+    const supervisor = createProcessSupervisor();
+    const scopeKey = "scope:cancel-many-queued";
+    const firstRunPromise = supervisor.spawn(
+      createSpawnInput({ runId: "cancel-many-queued-first", scopeKey }),
+    );
+    const replacements = Array.from({ length: replacementCount }, (_unused, index) => {
+      const runId = `cancel-many-queued-replacement-${index}`;
+      const replacement = supervisor.spawn(
+        createSpawnInput({
+          runId,
+          scopeKey,
+          mode: index % 2 === 0 ? "child" : "pty",
+          replaceExistingScope: true,
+        }),
+      );
+      supervisor.cancel(runId, "manual-cancel");
+      return replacement;
+    });
+    const laterRunPromise = supervisor.spawn(
+      createSpawnInput({ runId: "cancel-many-queued-later", scopeKey }),
+    );
+
+    expect(createChildAdapterMock).toHaveBeenCalledTimes(1);
+    expect(createPtyAdapterMock).not.toHaveBeenCalled();
+
+    firstStartup.resolve(first);
+    const [firstRun, replacementRuns, laterRun] = await Promise.all([
+      firstRunPromise,
+      Promise.all(replacements),
+      laterRunPromise,
+    ]);
+
+    expect(createChildAdapterMock).toHaveBeenCalledTimes(2);
+    expect(createPtyAdapterMock).not.toHaveBeenCalled();
+    expect(first.killMock).not.toHaveBeenCalled();
+    expect(later.killMock).not.toHaveBeenCalled();
+    for (const replacement of replacementRuns) {
+      expect(replacement.pid).toBeUndefined();
+    }
+    await expect(Promise.all(replacementRuns.map((run) => run.wait()))).resolves.toEqual(
+      Array.from({ length: replacementCount }, () =>
+        expect.objectContaining({ reason: "manual-cancel" }),
+      ),
+    );
+
+    first.settle(0);
+    later.settle(0);
+    await expect(Promise.all([firstRun.wait(), laterRun.wait()])).resolves.toEqual([
+      expect.objectContaining({ reason: "exit" }),
+      expect.objectContaining({ reason: "exit" }),
+    ]);
+  });
+});

src/process/supervisor/supervisor.test.ts

@@ -389,17 +389,9 @@ describe("process supervisor", () => {
         current.settle(null, signal ?? "SIGTERM");
       },
     });
-    const replacement = createStubChildAdapter({
-      onKill: (signal, current) => {
-        current.settle(null, signal ?? "SIGTERM");
-      },
-    });
     const later = createStubChildAdapter();
     const firstStartup = createDeferred<StubChildAdapter>();
-    createChildAdapterMock
-      .mockReturnValueOnce(firstStartup.promise)
-      .mockResolvedValueOnce(replacement)
-      .mockResolvedValueOnce(later);
+    createChildAdapterMock.mockReturnValueOnce(firstStartup.promise).mockResolvedValueOnce(later);
 
     const supervisor = createProcessSupervisor();
     const firstRunPromise = spawnChild(supervisor, {
@@ -432,8 +424,9 @@ describe("process supervisor", () => {
       replacementPromise,
       laterPromise,
     ]);
+    expect(createChildAdapterMock).toHaveBeenCalledTimes(2);
     expect(first.killMock).toHaveBeenCalledWith("SIGTERM");
-    expect(replacement.killMock).toHaveBeenCalledWith("SIGTERM");
+    expect(replacementRun.pid).toBeUndefined();
     expect(later.killMock).not.toHaveBeenCalled();
 
     later.settle(0);

src/process/supervisor/supervisor.ts

@@ -156,21 +156,15 @@ export function createProcessSupervisor(): ProcessSupervisor {
     runId: string,
     startingRun: StartingRun,
   ): Promise<ManagedRun> => {
-    if (input.replaceExistingScope && scopeKey) {
-      // Scope admission already waited for predecessor startups. Do not
-      // cancel this replacement or later runs reserved behind its fence.
-      cancelActiveScope(scopeKey, "manual-cancel");
-    }
     const startedAtMs = Date.now();
+    const startingTerminationReason = startingRun.terminationReason;
     const record: RunRecord = {
       runId,
       sessionId: input.sessionId,
       backendId: input.backendId,
       scopeKey,
-      state: startingRun.terminationReason ? "exiting" : "starting",
-      ...(startingRun.terminationReason
-        ? { terminationReason: startingRun.terminationReason }
-        : {}),
+      state: startingTerminationReason ? "exiting" : "starting",
+      ...(startingTerminationReason ? { terminationReason: startingTerminationReason } : {}),
       startedAtMs,
       lastOutputAtMs: startedAtMs,
       createdAtMs: startedAtMs,
@@ -178,6 +172,38 @@ export function createProcessSupervisor(): ProcessSupervisor {
     };
     registry.add(record);
 
+    if (startingTerminationReason) {
+      // A replacement can be cancelled behind its scope fence. Never launch
+      // its command or terminate the surviving scope after that cancellation.
+      const exit: RunExit = {
+        reason: startingTerminationReason,
+        exitCode: null,
+        exitSignal: null,
+        durationMs: Date.now() - startedAtMs,
+        stdout: "",
+        stderr: "",
+        timedOut: isTimeoutReason(startingTerminationReason),
+        noOutputTimedOut: startingTerminationReason === "no-output-timeout",
+      };
+      registry.finalize(runId, {
+        reason: exit.reason,
+        exitCode: exit.exitCode,
+        exitSignal: exit.exitSignal,
+      });
+      return {
+        runId,
+        startedAtMs,
+        wait: async () => exit,
+        cancel: () => undefined,
+      };
+    }
+
+    if (input.replaceExistingScope && scopeKey) {
+      // Scope admission already waited for predecessor startups. Do not
+      // cancel this replacement or later runs reserved behind its fence.
+      cancelActiveScope(scopeKey, "manual-cancel");
+    }
+
     let forcedReason: TerminationReason | null = startingRun.terminationReason ?? null;
     let settled = false;
     let stdout = "";