修复 Terminal 历史重放触发控制序列乱码

docs/terminal-replay-input-gate/change.md

@@ -0,0 +1,24 @@
+# Terminal replay input gate
+
+## Intent
+
+Terminal views recreate xterm when tabs, projects, or views remount. Replayed history can contain terminal queries whose xterm-generated replies must not be sent into the still-running PTY and corrupt the current prompt. Preserve bounded history, live interaction, UTF-8 streaming, and the existing PTY lifecycle.
+
+## Spec
+
+- Treat cached history replay and live delivery as distinct phases.
+- Subscribe to live messages atomically with taking the history snapshot so output arriving during replay is buffered without loss or reordering.
+- Do not register xterm `onData` until xterm's write completion callbacks confirm all replayed output has been processed.
+- Register `onData` before flushing buffered live output so live terminal queries retain their normal responses.
+- If disposed during initialization or replay, unsubscribe and never register input or write to the PTY; dispose xterm and related UI resources.
+- Keep the 1 MiB history budget, close cleanup, lifecycle messages, and streaming UTF-8 decoding behavior.
+
+## Plan
+
+1. Refactor `terminalStream.subscribe` to return an atomic replay snapshot and live unsubscribe handle.
+2. Add a small replay/live coordinator around xterm's write callback and `onData` registration.
+3. Integrate the coordinator into `TerminalInstance` and strengthen disposal cleanup.
+4. Add regression tests for replay-generated input suppression, post-replay keyboard and live-query input, live output ordering during replay, and disposal during replay.
+5. Run focused Terminal tests and the full `pnpm check` gate.
+
+Risk: standard. This changes renderer-side Terminal timing and an internal subscription interface, but does not change IPC, persistent data, security boundaries, or PTY lifecycle contracts.

src/features/terminal/TerminalInstance.tsx

@@ -1,12 +1,12 @@
 import { useEffect, useRef } from "react";
 import "@xterm/xterm/css/xterm.css";
-import { commands, type TerminalMessage } from "../../bindings";
+import { commands } from "../../bindings";
 import { useEditorStore } from "../editor/editorStore";
 import { terminalTheme } from "../theme/themeColors";
 import { useThemeStore } from "../theme/themeStore";
 import { useSettingsStore } from "../settings/settingsStore";
 import { terminalRegistry } from "./terminalRegistry";
-import { decodeTerminalOutput, terminalStream } from "./terminalStream";
+import { attachTerminalSession } from "./terminalReplay";
 
 export function TerminalInstance({
   worktreeId,
@@ -51,24 +51,22 @@ export function TerminalInstance({
         fit.fit();
         terminal.focus();
         terminalRegistry.set(terminalId, terminal);
-        const unsubscribe = terminalStream.subscribe(
+        const session = attachTerminalSession(
           terminalId,
-          (message: TerminalMessage) => {
-            if (message.type === "output")
-              terminal.write(decodeTerminalOutput(terminalId, message.data));
-            else
-              useEditorStore
-                .getState()
-                .setTerminalStatus(
-                  worktreeId,
-                  terminalId,
-                  message.type === "exit" ? "exited" : "error",
-                );
+          terminal,
+          (data) => {
+            void commands.terminalWrite(terminalId, data);
+          },
+          (message) => {
+            useEditorStore
+              .getState()
+              .setTerminalStatus(
+                worktreeId,
+                terminalId,
+                message.type === "exit" ? "exited" : "error",
+              );
           },
         );
-        const input = terminal.onData((data) => {
-          void commands.terminalWrite(terminalId, data);
-        });
         let resizeTimer = 0;
         observer = new ResizeObserver(() => {
           window.clearTimeout(resizeTimer);
@@ -83,8 +81,7 @@ export function TerminalInstance({
         });
         observer.observe(host.current);
         cleanup = () => {
-          input.dispose();
-          unsubscribe();
+          session.dispose();
           window.clearTimeout(resizeTimer);
           terminalRegistry.delete(terminalId);
         };

src/features/terminal/terminalReplay.test.ts

@@ -0,0 +1,140 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { terminalStream } from "./terminalStream";
+import { attachTerminalSession, type ReplayTerminal } from "./terminalReplay";
+
+class FakeTerminal implements ReplayTerminal {
+  readonly writes: Array<{ data: string; done?: () => void }> = [];
+  onDataCalls = 0;
+  private dataListener?: (data: string) => void;
+
+  write(data: string, done?: () => void): void {
+    this.writes.push({ data, done });
+  }
+
+  onData(listener: (data: string) => void): { dispose(): void } {
+    this.onDataCalls += 1;
+    this.dataListener = listener;
+    return {
+      dispose: () => {
+        if (this.dataListener === listener) this.dataListener = undefined;
+      },
+    };
+  }
+
+  completeWrite(index: number, generatedResponse?: string): void {
+    if (generatedResponse) this.dataListener?.(generatedResponse);
+    this.writes[index]?.done?.();
+  }
+
+  emitInput(data: string): void {
+    this.dataListener?.(data);
+  }
+}
+
+afterEach(() => terminalStream.close("t1"));
+
+describe("terminal replay input gate", () => {
+  it("does not forward xterm responses generated by replayed history", () => {
+    terminalStream.push({
+      type: "output",
+      terminalId: "t1",
+      data: "history-query",
+    });
+    const terminal = new FakeTerminal();
+    const writePty = vi.fn();
+
+    attachTerminalSession("t1", terminal, writePty, vi.fn());
+
+    expect(terminal.writes.map(({ data }) => data)).toEqual(["history-query"]);
+    terminal.completeWrite(0, "history-response");
+    expect(writePty).not.toHaveBeenCalled();
+
+    terminal.emitInput("keyboard-input");
+    expect(writePty).toHaveBeenCalledWith("keyboard-input");
+  });
+
+  it("buffers live output in order and restores responses for live queries", () => {
+    terminalStream.push({ type: "output", terminalId: "t1", data: "history" });
+    const terminal = new FakeTerminal();
+    const writePty = vi.fn();
+
+    attachTerminalSession("t1", terminal, writePty, vi.fn());
+    terminalStream.push({ type: "output", terminalId: "t1", data: "live-1" });
+    terminalStream.push({ type: "output", terminalId: "t1", data: "live-2" });
+
+    expect(terminal.writes.map(({ data }) => data)).toEqual(["history"]);
+    terminal.completeWrite(0);
+    expect(terminal.writes.map(({ data }) => data)).toEqual([
+      "history",
+      "live-1",
+      "live-2",
+    ]);
+
+    terminal.completeWrite(1, "live-query-response");
+    expect(writePty).toHaveBeenCalledWith("live-query-response");
+  });
+
+  it("preserves streaming UTF-8 decoding for output buffered during replay", () => {
+    terminalStream.push({ type: "output", terminalId: "t1", data: "history" });
+    const terminal = new FakeTerminal();
+
+    attachTerminalSession("t1", terminal, vi.fn(), vi.fn());
+    terminalStream.push({
+      type: "output",
+      terminalId: "t1",
+      data: new Uint8Array([226]),
+    });
+    terminalStream.push({
+      type: "output",
+      terminalId: "t1",
+      data: new Uint8Array([156, 147]),
+    });
+    terminal.completeWrite(0);
+
+    expect(terminal.writes.map(({ data }) => data).join("")).toBe("history✓");
+  });
+
+  it("starts a fresh UTF-8 decoder when replaying after a remount", () => {
+    terminalStream.push({
+      type: "output",
+      terminalId: "t1",
+      data: new Uint8Array([226]),
+    });
+    const firstTerminal = new FakeTerminal();
+    const firstSession = attachTerminalSession(
+      "t1",
+      firstTerminal,
+      vi.fn(),
+      vi.fn(),
+    );
+    firstTerminal.completeWrite(0);
+    firstSession.dispose();
+
+    const remountedTerminal = new FakeTerminal();
+    attachTerminalSession("t1", remountedTerminal, vi.fn(), vi.fn());
+    remountedTerminal.completeWrite(0);
+    terminalStream.push({
+      type: "output",
+      terminalId: "t1",
+      data: new Uint8Array([156, 147]),
+    });
+
+    expect(remountedTerminal.writes.map(({ data }) => data).join("")).toBe("✓");
+  });
+
+  it("does not register input or process buffered output after disposal", () => {
+    terminalStream.push({ type: "output", terminalId: "t1", data: "history" });
+    const terminal = new FakeTerminal();
+    const writePty = vi.fn();
+    const session = attachTerminalSession("t1", terminal, writePty, vi.fn());
+    terminalStream.push({ type: "output", terminalId: "t1", data: "live" });
+
+    session.dispose();
+    terminal.completeWrite(0);
+    terminal.emitInput("late-input");
+
+    expect(terminal.writes.map(({ data }) => data)).toEqual(["history"]);
+    expect(terminal.onDataCalls).toBe(0);
+    expect(writePty).not.toHaveBeenCalled();
+  });
+});

src/features/terminal/terminalReplay.ts

@@ -0,0 +1,78 @@
+import type { TerminalMessage } from "../../bindings";
+import { createTerminalOutputDecoder, terminalStream } from "./terminalStream";
+
+interface Disposable {
+  dispose(): void;
+}
+
+export interface ReplayTerminal {
+  write(data: string, callback?: () => void): void;
+  onData(listener: (data: string) => void): Disposable;
+}
+
+export function attachTerminalSession(
+  terminalId: string,
+  terminal: ReplayTerminal,
+  writePty: (data: string) => void,
+  handleLifecycle: (
+    message: Exclude<TerminalMessage, { type: "output" }>,
+  ) => void,
+): Disposable {
+  let disposed = false;
+  let replaying = true;
+  let input: Disposable | undefined;
+  let pending: TerminalMessage[] = [];
+  const decodeOutput = createTerminalOutputDecoder();
+
+  const deliverLive = (message: TerminalMessage): void => {
+    if (message.type === "output") {
+      terminal.write(decodeOutput(message.data));
+    } else {
+      handleLifecycle(message);
+    }
+  };
+
+  const subscription = terminalStream.subscribe(terminalId, (message) => {
+    if (disposed) return;
+    if (replaying) pending.push(message);
+    else deliverLive(message);
+  });
+
+  const finishReplay = (): void => {
+    if (disposed) return;
+    input = terminal.onData(writePty);
+    replaying = false;
+    const buffered = pending;
+    pending = [];
+    for (const message of buffered) deliverLive(message);
+  };
+
+  const replayNext = (index: number): void => {
+    if (disposed) return;
+    const message = subscription.replay[index];
+    if (!message) {
+      finishReplay();
+      return;
+    }
+    if (message.type === "output") {
+      terminal.write(decodeOutput(message.data), () => {
+        replayNext(index + 1);
+      });
+    } else {
+      handleLifecycle(message);
+      replayNext(index + 1);
+    }
+  };
+
+  replayNext(0);
+
+  return {
+    dispose(): void {
+      if (disposed) return;
+      disposed = true;
+      pending = [];
+      input?.dispose();
+      subscription.unsubscribe();
+    },
+  };
+}

src/features/terminal/terminalStream.test.ts

@@ -12,26 +12,33 @@ describe("terminal stream", () => {
       terminalId: "t1",
       data: "shell ready\r\n",
     });
-    terminalStream.subscribe("t1", (message) => received.push(message));
+    const subscription = terminalStream.subscribe("t1", (message) =>
+      received.push(message),
+    );
+    received.push(...subscription.replay);
     expect(received).toEqual([
       { type: "output", terminalId: "t1", data: "shell ready\r\n" },
     ]);
   });
 
   it("replays output after the xterm view remounts", () => {
     const first: TerminalMessage[] = [];
-    const unsubscribe = terminalStream.subscribe("t1", (message) =>
+    const firstSubscription = terminalStream.subscribe("t1", (message) =>
       first.push(message),
     );
+    first.push(...firstSubscription.replay);
     terminalStream.push({
       type: "output",
       terminalId: "t1",
       data: "before project switch\r\n",
     });
-    unsubscribe();
+    firstSubscription.unsubscribe();
 
     const remounted: TerminalMessage[] = [];
-    terminalStream.subscribe("t1", (message) => remounted.push(message));
+    const remountedSubscription = terminalStream.subscribe("t1", (message) =>
+      remounted.push(message),
+    );
+    remounted.push(...remountedSubscription.replay);
 
     expect(first).toHaveLength(1);
     expect(remounted).toEqual(first);
@@ -49,20 +56,28 @@ describe("terminal stream", () => {
       data: "b".repeat(700_000),
     });
 
-    const replayed: TerminalMessage[] = [];
-    terminalStream.subscribe("t1", (message) => replayed.push(message));
+    const replayed = terminalStream.subscribe("t1", () => {}).replay;
 
     expect(replayed).toHaveLength(1);
     expect(replayed[0]).toMatchObject({ type: "output" });
     if (replayed[0]?.type === "output")
       expect(String(replayed[0].data).startsWith("b")).toBe(true);
   });
 
+  it("replays the latest lifecycle status after output", () => {
+    terminalStream.push({ type: "output", terminalId: "t1", data: "done" });
+    terminalStream.push({ type: "exit", terminalId: "t1", exitCode: 0 });
+
+    expect(terminalStream.subscribe("t1", () => {}).replay).toEqual([
+      { type: "output", terminalId: "t1", data: "done" },
+      { type: "exit", terminalId: "t1", exitCode: 0 },
+    ]);
+  });
+
   it("clears replay history when the terminal closes", () => {
     terminalStream.push({ type: "output", terminalId: "t1", data: "gone" });
     terminalStream.close("t1");
-    const replayed: TerminalMessage[] = [];
-    terminalStream.subscribe("t1", (message) => replayed.push(message));
+    const replayed = terminalStream.subscribe("t1", () => {}).replay;
     expect(replayed).toEqual([]);
   });