Address terminals by workspace ID in the SDK and CLI (#4358)

* feat(terminals): expose workspace ownership through the SDK and CLI

Directory-based terminal creation selected the oldest workspace sharing that path. Expose explicit workspace ownership in the shared public API and CLI, retain ownership and process directories in listings, and acknowledge terminal deletion after teardown.

* fix(file-observer): preserve nested shallow reconciliation scopes

A parent directory scan does not visit its descendants. Collapsing change scopes discarded nested scans and left coalesced native deletions unseen until a full audit.

* test(terminals): extract output waits from end-to-end scenarios

packages/server/src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts

@@ -0,0 +1,190 @@
+import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { afterEach, beforeEach, expect, test } from "vitest";
+import { createPaseoClient, type PaseoClient } from "@getpaseo/client";
+import { DaemonClient } from "../test-utils/daemon-client.js";
+import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+
+let daemon: TestPaseoDaemon;
+let client: DaemonClient;
+let cwd: string;
+let sdk: PaseoClient;
+
+beforeEach(async () => {
+  cwd = await mkdtemp(path.join(tmpdir(), "terminal-workspace-sdk-"));
+  daemon = await createTestPaseoDaemon();
+  client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  await client.connect();
+  sdk = createPaseoClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  await sdk.connect();
+});
+
+afterEach(async () => {
+  await client.close();
+  await sdk.close();
+  await daemon.close();
+  await rm(cwd, { recursive: true, force: true });
+});
+
+test("SDK and workspace handles preserve ownership and actual process directories", async () => {
+  const first = await createWorkspace("main");
+  const second = await createWorkspace("feature-work");
+  const nested = path.join(cwd, "nested");
+  await mkdir(nested);
+  const a = await sdk.terminals.create({ workspaceId: first, name: "Main" });
+  const b = await sdk.workspaces.ref(second).terminals.create({ name: "Feature", cwd: nested });
+  const c = await sdk.terminals.create({
+    workspaceId: second,
+    cwd: daemon.staticDir,
+    name: "Outside",
+  });
+  expect(a.current()).toEqual({ id: a.id, workspaceId: first, cwd, name: "Main" });
+  expect(b.current()).toEqual({ id: b.id, workspaceId: second, cwd: nested, name: "Feature" });
+  expect(c.current()).toEqual({
+    id: c.id,
+    workspaceId: second,
+    cwd: daemon.staticDir,
+    name: "Outside",
+  });
+  const result = await sdk.workspaces.ref(second).terminals.list({ requestId: "feature-list" });
+  expect(result.requestId).toBe("feature-list");
+  expect(result.entries).toEqual(expect.arrayContaining([b.current(), c.current()]));
+  expect(result.entries).toHaveLength(2);
+  expect((await sdk.terminals.list()).entries).toHaveLength(3);
+  expect((await sdk.terminals.list({ cwd })).entries).toEqual([a.current(), b.current()]);
+  const ref = sdk.terminals.ref(b.id);
+  expect(ref.current()).toBeNull();
+  expect(await ref.refresh()).toEqual(b.current());
+  await ref.kill();
+  expect(await ref.refresh()).toBeNull();
+  expect((await sdk.terminals.list({ workspaceId: second })).entries).toEqual([c.current()]);
+  expect((await sdk.terminals.list({ workspaceId: first })).entries).toEqual([a.current()]);
+});
+
+test("SDK creates a command terminal and sends literal input and key tokens", async () => {
+  const workspaceId = await createWorkspace("Input");
+  const terminal = await sdk.terminals.create({
+    workspaceId,
+    command: process.execPath,
+    args: [
+      "-e",
+      "process.stdin.setRawMode(true); process.stdin.resume(); console.log('READY'); let hex = ''; process.stdin.on('data', data => { hex += data.toString('hex'); console.log('HEX:' + hex); });",
+    ],
+    size: { rows: 35, cols: 120 },
+  });
+  const screen = async () => (await terminal.capture({ stripAnsi: true })).lines.join("\n");
+  await expect.poll(screen).toContain("READY");
+  expect(terminal.write("Enter")).toBe(5);
+  await expect.poll(screen).toContain("HEX:456e746572");
+  expect(terminal.sendKeys(["Enter", "Tab", "Escape", "C-c"])).toBe(4);
+  await expect.poll(screen).toContain("HEX:456e7465720d091b03");
+  const capture = await terminal.capture({
+    start: 0,
+    end: 0,
+    stripAnsi: true,
+    requestId: "capture-first",
+  });
+  expect(capture).toMatchObject({ terminalId: terminal.id, requestId: "capture-first" });
+  expect(capture.lines).toHaveLength(1);
+  await terminal.kill();
+  expect(terminal.current()).toBeNull();
+  expect((await sdk.terminals.list({ workspaceId })).entries).toEqual([]);
+});
+
+test("terminal creation rejects unknown and archived owners, including explicit cwd overrides", async () => {
+  const workspaceId = await createWorkspace("Archived");
+  await sdk.workspaces.ref(workspaceId).archive();
+  for (const id of [workspaceId, "wks_missing"]) {
+    await expect(sdk.terminals.create({ workspaceId: id })).rejects.toThrow(/not active/);
+    await expect(sdk.terminals.create({ workspaceId: id, cwd })).rejects.toThrow(/not active/);
+    const raw = await client.createTerminal(cwd, undefined, undefined, { workspaceId: id });
+    expect(raw).toMatchObject({
+      terminal: null,
+      error: `Workspace ${id} is not active or does not exist`,
+    });
+  }
+  expect((await sdk.terminals.list()).entries).toEqual([]);
+  expect((await client.fetchWorkspaces()).entries).toEqual([]);
+});
+
+test("plugin handlers operate terminals through their host-owned Paseo API", async () => {
+  const workspaceId = await createWorkspace("Plugin workspace");
+  const pluginDirectory = path.join(cwd, "plugin");
+  await mkdir(pluginDirectory);
+  await writeFile(
+    path.join(pluginDirectory, "paseo-plugin.json"),
+    JSON.stringify({ id: "terminal-sdk" }),
+  );
+  await writeFile(
+    path.join(pluginDirectory, "index.server.ts"),
+    `
+import { defineRpc } from "@getpaseo/plugin";
+import { z } from "zod";
+const operate = defineRpc({ name: "operate", input: z.object({ workspaceId: z.string(), command: z.string() }), output: z.object({ workspaceIds: z.array(z.string()), lines: z.array(z.string()), remaining: z.number() }) });
+async function waitForTerminalOutput(terminal, text) {
+  const deadline = Date.now() + 10000;
+  while (Date.now() < deadline) {
+    const capture = await terminal.capture({ stripAnsi: true });
+    if (capture.lines.some(line => line.includes(text))) return capture.lines;
+    await new Promise(resolve => setTimeout(resolve, 20));
+  }
+  throw new Error("Timed out waiting for plugin terminal output: " + text);
+}
+export default function contribute(server) {
+  server.handle(operate, async ({ workspaceId, command }, { paseo }) => {
+    const workspace = paseo.workspaces.ref(workspaceId);
+    const terminal = await workspace.terminals.create({ command, args: ["-e", "process.stdin.setRawMode(true); process.stdin.resume(); console.log('PLUGIN READY'); let hex = ''; process.stdin.on('data', data => { hex += data.toString('hex'); console.log('PLUGIN:' + hex); });"] });
+    try {
+      await waitForTerminalOutput(terminal, "PLUGIN READY");
+      terminal.write("Enter");
+      await waitForTerminalOutput(terminal, "PLUGIN:456e746572");
+      terminal.sendKeys(["Enter"]);
+      const lines = await waitForTerminalOutput(terminal, "PLUGIN:456e7465720d");
+      const listed = await workspace.terminals.list();
+      await paseo.terminals.ref(terminal.id).kill();
+      return { workspaceIds: listed.entries.map(entry => entry.workspaceId), lines, remaining: (await workspace.terminals.list()).entries.length };
+    } finally {
+      await terminal.kill();
+    }
+  });
+  return () => {};
+}`,
+  );
+  await client.patchDaemonConfig({ pluginsEnabled: true });
+  await client.installDirectoryPlugin(pluginDirectory);
+  try {
+    const result = await client.invokePluginRpc("terminal-sdk", "operate", {
+      workspaceId,
+      command: process.execPath,
+    });
+    expect(result).toMatchObject({
+      workspaceIds: [workspaceId],
+      remaining: 0,
+      lines: expect.arrayContaining([expect.stringContaining("PLUGIN:456e7465720d")]),
+    });
+  } finally {
+    await client.removePlugin("terminal-sdk");
+  }
+}, 30_000);
+
+async function createWorkspace(title: string): Promise<string> {
+  const result = await client.createWorkspace({ source: { kind: "directory", path: cwd }, title });
+  if (!result.workspace) throw new Error(result.error ?? "Workspace creation failed");
+  return result.workspace.id;
+}
+
+test("listing by workspace ID keeps terminals in a shared directory separate", async () => {
+  const first = await createWorkspace("main");
+  const second = await createWorkspace("feature-work");
+  await client.createTerminal(cwd, "main terminal", undefined, { workspaceId: first });
+  const created = await client.createTerminal(cwd, "feature terminal", undefined, {
+    workspaceId: second,
+  });
+  expect(created.error).toBeNull();
+
+  const result = await client.listTerminals(undefined, undefined, { workspaceId: second });
+  expect(result.terminals).toEqual([
+    expect.objectContaining({ id: created.terminal?.id, workspaceId: second, cwd }),
+  ]);
+});

packages/server/src/server/file-observer/internal/native-recursive.test.ts

@@ -0,0 +1,71 @@
+import { EventEmitter } from "node:events";
+import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { expect, test, vi } from "vitest";
+import { createFileObserver, type FileChange } from "../index.js";
+import { createNativeRecursiveBackend } from "./native-recursive.js";
+import { createObserverPaths } from "./paths.js";
+
+// Native watchers may coalesce file removals into change notifications. Keep the
+// filesystem real while controlling which notifications reach reconciliation.
+test("shallow parent scans retain nested change scopes", async () => {
+  const root = await mkdtemp(join(tmpdir(), "native-scopes-"));
+  const paths = createObserverPaths(process.platform);
+  const removed = [
+    join(root, "root.txt"),
+    join(root, "child", "child.txt"),
+    join(root, "child", "deep", "deep.txt"),
+  ];
+  await mkdir(join(root, "child", "deep"), { recursive: true });
+  await Promise.all(removed.map((path) => writeFile(path, "before")));
+  const events: FileChange[] = [];
+  const notifications = new EventEmitter();
+  let active = true;
+  const observer = createFileObserver();
+  const backend = createNativeRecursiveBackend(
+    {
+      root,
+      metrics: observer.getDiagnostics(),
+      isActive: () => active,
+      isIgnored: () => false,
+      isPathInside: paths.isInside,
+      queueEvent: (type, path) => events.push({ type, path }),
+      fail: (error) => {
+        throw error;
+      },
+    },
+    paths,
+    (_root, listener) => {
+      notifications.on("change", listener);
+      return {
+        close: () => {
+          notifications.removeAllListeners();
+        },
+        on: (event, onError) => notifications.on(event, onError),
+      };
+    },
+  );
+  try {
+    await backend.start();
+    await Promise.all(removed.map((path) => rm(path)));
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "performance"] });
+    for (const path of removed) notifications.emit("change", "change", path);
+    await vi.advanceTimersByTimeAsync(8_000);
+    vi.useRealTimers();
+    await expect
+      .poll(() =>
+        events
+          .filter((event) => event.type === "delete")
+          .map((event) => event.path)
+          .sort(),
+      )
+      .toEqual([...removed].sort());
+  } finally {
+    vi.useRealTimers();
+    active = false;
+    await backend.close();
+    await observer.close();
+    await rm(root, { recursive: true, force: true });
+  }
+});

packages/server/src/server/file-observer/internal/native-recursive.ts

@@ -1,4 +1,4 @@
-import { type Dirent, type FSWatcher, watch } from "node:fs";
+import { type Dirent, watch } from "node:fs";
 import { readdir, stat } from "node:fs/promises";
 import { dirname, join, resolve } from "node:path";
 import type { BackendDiagnostics, ObservationBackend, ObservationHost } from "./contracts.js";
@@ -25,15 +25,27 @@ interface Inventory {
   files: Set<string>;
 }
 
+type WatchDirectory = (
+  root: string,
+  listener: (eventType: string, filename: string | null) => void,
+) => {
+  close(): void;
+  on(event: "error", listener: (error: Error) => void): unknown;
+};
+
+const watchDirectory: WatchDirectory = (root, listener) =>
+  watch(root, { recursive: true }, listener);
+
 export function createNativeRecursiveBackend(
   host: ObservationHost,
   paths: ObserverPaths,
+  observe: WatchDirectory = watchDirectory,
 ): ObservationBackend {
-  return new NativeRecursiveBackend(host, paths);
+  return new NativeRecursiveBackend(host, paths, observe);
 }
 
 class NativeRecursiveBackend implements ObservationBackend {
-  private watcher: FSWatcher | null = null;
+  private watcher: ReturnType<WatchDirectory> | null = null;
   private files = new Set<string>();
   private directories = new Set<string>();
   private entries = new Map<string, DirectoryEntry>();
@@ -62,6 +74,7 @@ class NativeRecursiveBackend implements ObservationBackend {
   constructor(
     private readonly host: ObservationHost,
     private readonly paths: ObserverPaths,
+    private readonly observe: WatchDirectory,
   ) {}
 
   async start(): Promise<void> {
@@ -114,7 +127,7 @@ class NativeRecursiveBackend implements ObservationBackend {
   }
 
   private watchRoot(): void {
-    const watcher = watch(this.host.root, { recursive: true }, (eventType, filename) => {
+    const watcher = this.observe(this.host.root, (eventType, filename) => {
       if (!this.host.isActive()) return;
       this.host.metrics.nativeEventCount += 1;
       if (!filename) {
@@ -262,7 +275,8 @@ class NativeRecursiveBackend implements ObservationBackend {
       await this.reconcileSubtree(directory, generation);
       if (!this.canCommit(generation)) return;
     }
-    for (const directory of this.paths.collapse(changeScopes)) {
+    // Parent scans are shallow, so a queued parent cannot cover its children.
+    for (const directory of new Set(changeScopes)) {
       const alreadyCovered =
         forcedLocalScopes.has(directory) ||
         forcedRecursiveScopes.some((scope) => this.host.isPathInside(scope, directory));

packages/server/src/server/websocket-server.ts

@@ -1682,6 +1682,7 @@ export class VoiceAssistantWebSocketServer {
         "terminal-input-mode-replay": true,
         // COMPAT(terminalSizeOwnership): added in v0.2.6, remove gate after 2027-02-02.
         "terminal-size-ownership": true,
+        workspaceTerminals: true,
         // COMPAT(rewind): added in v0.1.X, drop the gate when floor >= v0.1.X.
         rewind: true,
         // COMPAT(agentTimelinePromptIndex): added in v0.2.X, drop the gate when floor >= v0.2.X.

packages/server/src/terminal/terminal-session-controller.test.ts

@@ -300,7 +300,7 @@ describe("terminal-session-controller legacy terminal creation", () => {
       hasBinaryChannel: () => true,
       isPathWithinRoot: isSameOrDescendantPath,
       sessionLogger: createLogger(),
-      listTerminalWorkspaceRefs: async () => [],
+      listTerminalWorkspaceRefs: async () => [{ workspaceId: "ws-1", cwd: "/work/repo" }],
     });
 
     await controller.dispatch({
@@ -548,7 +548,13 @@ describe("terminal-session-controller subdirectory aggregation", () => {
         payload: {
           cwd: rootCwd,
           terminals: [
-            { id: "root-term", name: "Terminal 1", workspaceId: "ws-test", activity: null },
+            {
+              id: "root-term",
+              name: "Terminal 1",
+              cwd: rootCwd,
+              workspaceId: "ws-test",
+              activity: null,
+            },
           ],
           requestId: "req-root",
         },
@@ -558,7 +564,13 @@ describe("terminal-session-controller subdirectory aggregation", () => {
         payload: {
           cwd: worktreeCwd,
           terminals: [
-            { id: "worktree-term", name: "Feature", workspaceId: "ws-test", activity: null },
+            {
+              id: "worktree-term",
+              name: "Feature",
+              cwd: worktreeCwd,
+              workspaceId: "ws-test",
+              activity: null,
+            },
           ],
           requestId: "req-worktree",
         },

packages/server/src/terminal/terminal-session-controller.ts

@@ -426,10 +426,16 @@ export class TerminalSessionController {
     }
 
     try {
-      const terminals =
-        typeof msg.cwd === "string"
-          ? await this.getTerminalsForWorkspaceRoot(msg.cwd, msg.workspaceId)
-          : await this.getAllTerminalSessions();
+      let terminals: TerminalSession[];
+      if (msg.workspaceId !== undefined) {
+        terminals = (await this.getAllTerminalSessions()).filter(
+          (terminal) => terminal.workspaceId === msg.workspaceId,
+        );
+      } else if (typeof msg.cwd === "string") {
+        terminals = await this.getTerminalsForWorkspaceRoot(msg.cwd);
+      } else {
+        terminals = await this.getAllTerminalSessions();
+      }
       for (const terminal of terminals) {
         this.ensureExitSubscription(terminal);
       }
@@ -437,7 +443,9 @@ export class TerminalSessionController {
         type: "list_terminals_response",
         payload: {
           ...(msg.cwd ? { cwd: msg.cwd } : {}),
-          terminals: terminals.map((terminal) => this.toTerminalInfo(terminal)),
+          terminals: terminals.map((terminal) =>
+            Object.assign(this.toTerminalInfo(terminal), { cwd: terminal.cwd }),
+          ),
           requestId: msg.requestId,
         },
       });
@@ -463,7 +471,9 @@ export class TerminalSessionController {
     const terminalsByDirectory = await Promise.all(
       directories.map((cwd) => manager.getTerminals(cwd)),
     );
-    return terminalsByDirectory.flat();
+    return [
+      ...new Map(terminalsByDirectory.flat().map((terminal) => [terminal.id, terminal])).values(),
+    ];
   }
 
   private async getTerminalsForWorkspaceRoot(
@@ -558,6 +568,11 @@ export class TerminalSessionController {
         return;
       }
 
+      const workspaces = await this.listTerminalWorkspaceRefs();
+      if (!workspaces.some((workspace) => workspace.workspaceId === workspaceId)) {
+        throw new Error(`Workspace ${workspaceId} is not active or does not exist`);
+      }
+
       const session = await this.terminalManager.createTerminal({
         cwd: msg.cwd,
         workspaceId,
@@ -746,12 +761,24 @@ export class TerminalSessionController {
   }
 
   private async handleKillTerminalRequest(msg: KillTerminalRequest): Promise<void> {
-    const result = this.killTerminalForClose(msg.terminalId);
+    let success = false;
+    if (this.terminalManager) {
+      try {
+        this.detachStream(msg.terminalId, { emitExit: true });
+        await this.terminalManager.killTerminalAndWait(msg.terminalId);
+        success = true;
+      } catch (error) {
+        this.sessionLogger.error(
+          { err: error, terminalId: msg.terminalId },
+          "Failed to kill terminal",
+        );
+      }
+    }
     this.emit({
       type: "kill_terminal_response",
       payload: {
-        terminalId: result.terminalId,
-        success: result.success,
+        terminalId: msg.terminalId,
+        success,
         requestId: msg.requestId,
       },
     });