fix(process): release terminals after observation failures

Keep queued observation failures available through frames while allowing the resource owner to release xterm and its owned addons. Catch write-callback failures so cleanup cannot hang, and retain actual disposal failures for retry.

src/process/capture/TerminalRenderer.test.ts

@@ -1,6 +1,17 @@
-import { expect, it } from "vitest";
+import { createRequire } from "node:module";
+import { mkdtemp, rm, stat } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { expect, it, vi } from "vitest";
 
 import { TerminalRenderer } from "./TerminalRenderer.js";
+import { ProcessCaptureResourceScope } from "./ProcessCaptureLifecycle.js";
+
+const require = createRequire(import.meta.url);
+const { Terminal } =
+  require("@xterm/headless") as typeof import("@xterm/headless");
+const { SerializeAddon } =
+  require("@xterm/addon-serialize") as typeof import("@xterm/addon-serialize");
 
 const renderFrames = async (
   data: string,
@@ -45,3 +56,93 @@ it("trims only after normalization sees the full-width row", async () => {
   );
   expect(frames[0]?.lines[0]).toBe("pid <pid>");
 });
+
+it.each(["write", "resize"] as const)(
+  "releases terminal resources after a queued %s observation fails",
+  async (operation) => {
+    const failure = new Error("normalization failed");
+    const terminalDispose = vi.spyOn(Terminal.prototype, "dispose");
+    const addonDispose = vi.spyOn(SerializeAddon.prototype, "dispose");
+    const renderer = new TerminalRenderer({
+      columns: 20,
+      rows: 4,
+      scrollback: 10,
+      maxBytes: 100_000,
+      normalize: () => {
+        throw failure;
+      },
+    });
+    const scope = new ProcessCaptureResourceScope();
+    const temporaryRoot = await mkdtemp(
+      join(tmpdir(), "rea-terminal-cleanup-"),
+    );
+    try {
+      if (operation === "write") renderer.write("output", 0);
+      else renderer.resize(30, 4, 0);
+      await expect(renderer.frames()).rejects.toBe(failure);
+      const cleanup = await scope.release({
+        timers: new Set(),
+        terminal: undefined,
+        renderer,
+        runId: "failed-observation",
+        temporaryRoot,
+      });
+      expect(cleanup.terminal_renderer).toEqual({
+        state: "cleaned",
+        reason: null,
+      });
+      expect(terminalDispose).toHaveBeenCalledTimes(1);
+      expect(addonDispose).toHaveBeenCalledTimes(1);
+      await expect(stat(temporaryRoot)).rejects.toMatchObject({
+        code: "ENOENT",
+      });
+      await expect(scope.run(async () => "next capture")).resolves.toBe(
+        "next capture",
+      );
+      await expect(renderer.frames()).rejects.toBe(failure);
+    } finally {
+      await scope.close();
+      terminalDispose.mockRestore();
+      addonDispose.mockRestore();
+      await rm(temporaryRoot, { recursive: true, force: true });
+    }
+  },
+);
+
+it("retains disposal failures for a later cleanup retry", async () => {
+  const failure = new Error("terminal disposal failed");
+  const terminalDispose = vi.spyOn(Terminal.prototype, "dispose");
+  terminalDispose.mockImplementationOnce(() => {
+    throw failure;
+  });
+  const renderer = new TerminalRenderer({
+    columns: 20,
+    rows: 4,
+    scrollback: 10,
+    maxBytes: 100_000,
+    normalize: (value) => value,
+  });
+  const scope = new ProcessCaptureResourceScope();
+  const temporaryRoot = await mkdtemp(join(tmpdir(), "rea-terminal-retry-"));
+  try {
+    const cleanup = await scope.release({
+      timers: new Set(),
+      terminal: undefined,
+      renderer,
+      runId: "failed-disposal",
+      temporaryRoot,
+    });
+    expect(cleanup.terminal_renderer).toEqual({
+      state: "failed",
+      reason: failure.message,
+    });
+    await expect(scope.run(async () => "next capture")).resolves.toBe(
+      "next capture",
+    );
+    expect(terminalDispose).toHaveBeenCalledTimes(2);
+  } finally {
+    await scope.close();
+    terminalDispose.mockRestore();
+    await rm(temporaryRoot, { recursive: true, force: true });
+  }
+});

src/process/capture/TerminalRenderer.ts

@@ -55,12 +55,16 @@ export class TerminalRenderer {
 
   /** Queue one PTY chunk and capture state only after xterm has parsed it. */
   write(data: string, atMs: number): void {
-    this.#pending = this.#pending.then(
+    this.#queue(
       () =>
-        new Promise<void>((resolveWrite) => {
+        new Promise<void>((resolveWrite, rejectWrite) => {
           this.#terminal.write(data, () => {
-            this.#capture(atMs);
-            resolveWrite();
+            try {
+              this.#capture(atMs);
+              resolveWrite();
+            } catch (error) {
+              rejectWrite(error);
+            }
           });
         }),
     );
@@ -68,7 +72,7 @@ export class TerminalRenderer {
 
   /** Queue a terminal resize after every preceding write. */
   resize(columns: number, rows: number, atMs: number): void {
-    this.#pending = this.#pending.then(() => {
+    this.#queue(() => {
       this.#terminal.resize(columns, rows);
       this.#capture(atMs);
     });
@@ -98,11 +102,19 @@ export class TerminalRenderer {
 
   /** Release addon and terminal resources after all writes settle. */
   async dispose(): Promise<void> {
-    await this.#pending;
-    this.#serializeAddon.dispose();
+    // frames() owns observation failures; cleanup reports only disposal failures.
+    await this.#pending.catch(() => undefined);
+    // xterm owns and disposes every addon loaded through loadAddon().
     this.#terminal.dispose();
   }
 
+  #queue(operation: () => void | Promise<void>): void {
+    this.#pending = this.#pending.then(operation);
+    // PTY callbacks can fail before capture completion awaits frames(). Keep
+    // the original rejection observable there without an unhandled rejection.
+    void this.#pending.catch(() => undefined);
+  }
+
   #capture(atMs: number): void {
     const buffer = this.#terminal.buffer.active;
     const lines: string[] = [];