Merge pull request #56 from spencerkit/develop

fix(web): stabilize terminal recovery replay flow

Author: pallyoung

.changeset/stale-timers-repair.md

@@ -0,0 +1,7 @@
+---
+"@spencer-kit/coder-studio": patch
+---
+
+Fix terminal recovery so session output no longer stalls after noop reconcile
+decisions or gets cleared when queued live chunks flush after snapshot
+hydration.

packages/cli/src/pm2-control.test.ts

@@ -148,14 +148,15 @@ describe("pm2-control", () => {
   });
 
   it("waits for the previous PM2 app to disappear before starting a replacement", async () => {
+    const previousPid = 999_990;
     describeProcess
       .mockImplementationOnce(
         (_name: string, callback: (error: Error | null, result: unknown[]) => void) =>
-          callback(null, [{ pid: 111, pm2_env: { status: "online", restart_time: 0 } }])
+          callback(null, [{ pid: previousPid, pm2_env: { status: "online", restart_time: 0 } }])
       )
       .mockImplementationOnce(
         (_name: string, callback: (error: Error | null, result: unknown[]) => void) =>
-          callback(null, [{ pid: 111, pm2_env: { status: "stopping", restart_time: 0 } }])
+          callback(null, [{ pid: previousPid, pm2_env: { status: "stopping", restart_time: 0 } }])
       )
       .mockImplementationOnce(
         (_name: string, callback: (error: Error | null, result: unknown[]) => void) =>
@@ -180,14 +181,15 @@ describe("pm2-control", () => {
   });
 
   it("reuses one pm2 session while polling deletion during startup", async () => {
+    const previousPid = 999_992;
     describeProcess
       .mockImplementationOnce(
         (_name: string, callback: (error: Error | null, result: unknown[]) => void) =>
-          callback(null, [{ pid: 111, pm2_env: { status: "online", restart_time: 0 } }])
+          callback(null, [{ pid: previousPid, pm2_env: { status: "online", restart_time: 0 } }])
       )
       .mockImplementationOnce(
         (_name: string, callback: (error: Error | null, result: unknown[]) => void) =>
-          callback(null, [{ pid: 111, pm2_env: { status: "stopping", restart_time: 0 } }])
+          callback(null, [{ pid: previousPid, pm2_env: { status: "stopping", restart_time: 0 } }])
       )
       .mockImplementationOnce(
         (_name: string, callback: (error: Error | null, result: unknown[]) => void) =>
@@ -204,15 +206,55 @@ describe("pm2-control", () => {
     expect(disconnect).toHaveBeenCalledTimes(1);
   });
 
+  it("waits for the previous process pid to exit before starting a replacement", async () => {
+    const previousPid = 999_991;
+    const killSpy = vi.spyOn(process, "kill");
+    killSpy
+      .mockImplementationOnce(() => true)
+      .mockImplementationOnce(() => true)
+      .mockImplementationOnce(() => {
+        const error = new Error("process not found") as NodeJS.ErrnoException;
+        error.code = "ESRCH";
+        throw error;
+      });
+    describeProcess
+      .mockImplementationOnce(
+        (_name: string, callback: (error: Error | null, result: unknown[]) => void) =>
+          callback(null, [{ pid: previousPid, pm2_env: { status: "online", restart_time: 0 } }])
+      )
+      .mockImplementationOnce(
+        (_name: string, callback: (error: Error | null, result: unknown[]) => void) =>
+          callback(null, [])
+      );
+
+    const pendingStart = startManagedServer({
+      script: "/cli/dist/esm/server-runner.js",
+      cwd: "/repo",
+      waitMs: 10,
+    });
+
+    await expect(
+      Promise.race([
+        pendingStart.then(() => "started"),
+        new Promise((resolve) => setTimeout(() => resolve("waiting"), 20)),
+      ])
+    ).resolves.toBe("waiting");
+
+    expect(start).not.toHaveBeenCalled();
+    await pendingStart;
+    expect(killSpy).toHaveBeenCalledWith(previousPid, 0);
+  });
+
   it("keeps waiting during startup when delete reports missing but the old app still lingers", async () => {
+    const previousPid = 999_993;
     describeProcess
       .mockImplementationOnce(
         (_name: string, callback: (error: Error | null, result: unknown[]) => void) =>
-          callback(null, [{ pid: 111, pm2_env: { status: "online", restart_time: 0 } }])
+          callback(null, [{ pid: previousPid, pm2_env: { status: "online", restart_time: 0 } }])
       )
       .mockImplementationOnce(
         (_name: string, callback: (error: Error | null, result: unknown[]) => void) =>
-          callback(null, [{ pid: 111, pm2_env: { status: "stopping", restart_time: 0 } }])
+          callback(null, [{ pid: previousPid, pm2_env: { status: "stopping", restart_time: 0 } }])
       )
       .mockImplementationOnce(
         (_name: string, callback: (error: Error | null, result: unknown[]) => void) =>

packages/cli/src/pm2-control.ts

@@ -265,6 +265,39 @@ const waitForManagedServerDeletion = async (pm2: Pm2Module, waitMs: number): Pro
   throw new Error(`Timed out waiting for the managed server to stop after ${waitMs}ms.`);
 };
 
+const isMissingProcessError = (error: unknown): boolean =>
+  Boolean(
+    error &&
+      typeof error === "object" &&
+      "code" in error &&
+      (error as NodeJS.ErrnoException).code === "ESRCH"
+  );
+
+const waitForProcessExit = async (pid: number, waitMs: number): Promise<void> => {
+  const deadline = Date.now() + waitMs;
+
+  while (Date.now() <= deadline) {
+    try {
+      process.kill(pid, 0);
+    } catch (error) {
+      if (isMissingProcessError(error)) {
+        return;
+      }
+
+      throw error;
+    }
+
+    const remainingMs = deadline - Date.now();
+    if (remainingMs <= 0) {
+      break;
+    }
+
+    await sleep(Math.min(STARTUP_POLL_INTERVAL_MS, remainingMs));
+  }
+
+  throw new Error(`Timed out waiting for the managed server pid ${pid} to exit after ${waitMs}ms.`);
+};
+
 const deleteManagedServerInSession = async (
   pm2: Pm2Module,
   {
@@ -277,19 +310,27 @@ const deleteManagedServerInSession = async (
   if (processes.length === 0) {
     return false;
   }
+  const previousPid = processes[0]?.pid;
+  const deadline = Date.now() + PM2_DELETE_WAIT_MS;
 
   try {
     await removeManagedServer(pm2);
   } catch (error) {
     if (ignoreMissing && isMissingManagedServerError(error)) {
-      await waitForManagedServerDeletion(pm2, PM2_DELETE_WAIT_MS);
+      await waitForManagedServerDeletion(pm2, Math.max(0, deadline - Date.now()));
+      if (typeof previousPid === "number" && previousPid > 0) {
+        await waitForProcessExit(previousPid, Math.max(0, deadline - Date.now()));
+      }
       return false;
     }
 
     throw error;
   }
 
-  await waitForManagedServerDeletion(pm2, PM2_DELETE_WAIT_MS);
+  await waitForManagedServerDeletion(pm2, Math.max(0, deadline - Date.now()));
+  if (typeof previousPid === "number" && previousPid > 0) {
+    await waitForProcessExit(previousPid, Math.max(0, deadline - Date.now()));
+  }
   return true;
 };
 

packages/cli/src/update-worker.test.ts

@@ -96,4 +96,64 @@ describe("update-worker", () => {
     expect(state.manualCommand).toBe("coder-studio serve --restart");
     expect(state.errorSummary).toContain("restart failed");
   });
+
+  it("sanitizes pm2 and runtime override env before invoking install and restart commands", async () => {
+    const env = createEnv();
+    const runCommand = vi.fn(async () => {});
+    const originalEnv = {
+      PM2_HOME: process.env.PM2_HOME,
+      PM2_PROGRAMMATIC: process.env.PM2_PROGRAMMATIC,
+      PM2_JSON_PROCESSING: process.env.PM2_JSON_PROCESSING,
+      PM2_INTERACTOR_PROCESSING: process.env.PM2_INTERACTOR_PROCESSING,
+      NODE_APP_INSTANCE: process.env.NODE_APP_INSTANCE,
+      NODE_CHANNEL_FD: process.env.NODE_CHANNEL_FD,
+      NODE_CHANNEL_SERIALIZATION_MODE: process.env.NODE_CHANNEL_SERIALIZATION_MODE,
+      CODER_STUDIO_RUNTIME_JSON_PATH: process.env.CODER_STUDIO_RUNTIME_JSON_PATH,
+      CODER_STUDIO_SESSION_ID: process.env.CODER_STUDIO_SESSION_ID,
+      CODER_STUDIO_UPDATE_STATE_PATH: process.env.CODER_STUDIO_UPDATE_STATE_PATH,
+      pm_id: process.env.pm_id,
+    };
+
+    process.env.PM2_HOME = "/tmp/custom-pm2-home";
+    process.env.PM2_PROGRAMMATIC = "true";
+    process.env.PM2_JSON_PROCESSING = "true";
+    process.env.PM2_INTERACTOR_PROCESSING = "true";
+    process.env.NODE_APP_INSTANCE = "0";
+    process.env.NODE_CHANNEL_FD = "3";
+    process.env.NODE_CHANNEL_SERIALIZATION_MODE = "json";
+    process.env.CODER_STUDIO_RUNTIME_JSON_PATH = "/tmp/runtime.json";
+    process.env.CODER_STUDIO_SESSION_ID = "sess_test";
+    process.env.CODER_STUDIO_UPDATE_STATE_PATH = "/tmp/update-state.json";
+    process.env.pm_id = "0";
+
+    try {
+      await runUpdateWorker(env, {
+        runCommand,
+        now: () => 1000,
+      });
+    } finally {
+      for (const [key, value] of Object.entries(originalEnv)) {
+        if (value === undefined) {
+          delete process.env[key];
+        } else {
+          process.env[key] = value;
+        }
+      }
+    }
+
+    for (const call of runCommand.mock.calls) {
+      const options = call[2] as { env?: NodeJS.ProcessEnv };
+      expect(options.env?.PM2_HOME).toBe("/tmp/custom-pm2-home");
+      expect(options.env?.PM2_PROGRAMMATIC).toBeUndefined();
+      expect(options.env?.PM2_JSON_PROCESSING).toBeUndefined();
+      expect(options.env?.PM2_INTERACTOR_PROCESSING).toBeUndefined();
+      expect(options.env?.NODE_APP_INSTANCE).toBeUndefined();
+      expect(options.env?.NODE_CHANNEL_FD).toBeUndefined();
+      expect(options.env?.NODE_CHANNEL_SERIALIZATION_MODE).toBeUndefined();
+      expect(options.env?.CODER_STUDIO_RUNTIME_JSON_PATH).toBeUndefined();
+      expect(options.env?.CODER_STUDIO_SESSION_ID).toBeUndefined();
+      expect(options.env?.CODER_STUDIO_UPDATE_STATE_PATH).toBeUndefined();
+      expect(options.env?.pm_id).toBeUndefined();
+    }
+  });
 });

packages/cli/src/update-worker.ts

@@ -97,15 +97,47 @@ function buildManualCommand(input: WorkerEnv): string {
   ].join("\n");
 }
 
+const INTERNAL_ENV_KEYS = new Set([
+  "CODER_STUDIO_RUNTIME_JSON_PATH",
+  "CODER_STUDIO_SESSION_ID",
+  "NODE_APP_INSTANCE",
+  "NODE_CHANNEL_FD",
+  "NODE_CHANNEL_SERIALIZATION_MODE",
+  "PM2_INTERACTOR_PROCESSING",
+  "PM2_JSON_PROCESSING",
+  "PM2_PROGRAMMATIC",
+]);
+
+function buildChildProcessEnv(env = process.env): NodeJS.ProcessEnv {
+  const nextEnv: NodeJS.ProcessEnv = { ...env };
+
+  for (const key of Object.keys(nextEnv)) {
+    if (INTERNAL_ENV_KEYS.has(key)) {
+      delete nextEnv[key];
+      continue;
+    }
+
+    if (key.startsWith("CODER_STUDIO_UPDATE_") || key.startsWith("pm_")) {
+      delete nextEnv[key];
+    }
+  }
+
+  return nextEnv;
+}
+
 function runCommand(
   command: string,
   args: string[],
-  options?: { stdio?: "ignore" | "pipe"; logStream?: NodeJS.WritableStream }
+  options?: {
+    stdio?: "ignore" | "pipe";
+    logStream?: NodeJS.WritableStream;
+    env?: NodeJS.ProcessEnv;
+  }
 ): Promise<void> {
   return new Promise((resolve, reject) => {
     const child = spawn(command, args, {
       stdio: options?.stdio === "ignore" ? "ignore" : "pipe",
-      env: process.env,
+      env: options?.env ?? process.env,
     });
 
     if (options?.logStream && child.stdout) {
@@ -137,12 +169,13 @@ export async function runUpdateWorker(
   await mkdir(dirname(input.logFilePath), { recursive: true });
   const logStream = createWriteStream(input.logFilePath, { flags: "a" });
   const execute = deps?.runCommand ?? runCommand;
+  const childEnv = buildChildProcessEnv(process.env);
 
   try {
     await execute(
       input.npmCommand,
       [...input.installArgsPrefix, `${input.packageName}@${input.targetVersion}`],
-      { logStream }
+      { logStream, env: childEnv }
     );
   } catch (error) {
     const message = error instanceof Error ? error.message : String(error);
@@ -183,7 +216,7 @@ export async function runUpdateWorker(
   });
 
   try {
-    await execute(input.cliCommand, input.restartArgs, { logStream });
+    await execute(input.cliCommand, input.restartArgs, { logStream, env: childEnv });
   } catch (error) {
     const message = error instanceof Error ? error.message : String(error);
     await writeState(input.stateFilePath, {