feat(plugin): observe terminals across connected hosts

Share host-wide terminal observations across plugin surfaces and gate them on terminalCollection. Keep directory observations compatible, and carry creation time in worker v6 state so v5 terminals can drain with an unknown timestamp.

packages/app/e2e/browser/sidebar-views.spec.ts

@@ -9,6 +9,7 @@ test("sidebar views replace the content, preserve chrome, and recover to the def
     await views.expectDefault();
     await views.chooseWorkspaces();
     await views.exerciseExamples();
+    await views.exerciseTerminals();
     await views.restoreDefault();
     await views.recoverFromFailure();
     await views.chooseWorkspaces();

packages/app/e2e/support/helpers/new-workspace.ts

@@ -17,6 +17,8 @@ type NewWorkspaceDaemonClient = Pick<
   | "connect"
   | "createPaseoWorktree"
   | "createWorkspace"
+  | "createTerminal"
+  | "killTerminal"
   | "fetchAgents"
   | "fetchWorkspaces"
   | "getPaseoWorktreeList"

packages/app/e2e/support/helpers/sidebar-views.ts

@@ -134,6 +134,38 @@ function sidebarViews(
         )
         .toEqual(["Inbox"]);
     },
+    async exerciseTerminals() {
+      await choose("Example terminals");
+      const slot = visibleTestId(page, "plugin-sidebar-view");
+      await slot
+        .getByRole("button", { name: `Terminals on ${getServerId()}`, exact: true })
+        .click();
+      await expect(slot.getByText("No terminals", { exact: true })).toBeVisible();
+      const workspace = (await client.fetchWorkspaces()).entries.find(
+        (item) => item.id === workspaceId,
+      )!;
+      const created = await client.createTerminal(
+        workspace.workspaceDirectory,
+        "Sidebar terminal",
+        undefined,
+        { workspaceId },
+      );
+      if (!created.terminal) throw new Error(created.error ?? "Terminal creation failed");
+      try {
+        await expect(
+          slot.getByTestId(`sidebar-terminal-open-${created.terminal.id}`),
+        ).toBeVisible();
+        await slot.getByTestId(`sidebar-terminal-open-${created.terminal.id}`).click();
+        await expect(page.getByTestId("terminal-surface").first()).toBeVisible();
+        await slot.getByTestId(`sidebar-terminal-kill-${created.terminal.id}`).click();
+        await expect
+          .poll(async () => (await client.listTerminals()).terminals.map((item) => item.id))
+          .not.toContain(created.terminal!.id);
+        await expect(slot.getByText("No terminals", { exact: true })).toBeVisible();
+      } finally {
+        await client.killTerminal(created.terminal.id);
+      }
+    },
     async restoreDefault() {
       await choose("Default view");
       await expectDefault();

packages/app/src/data/push-router.test.ts

@@ -55,8 +55,8 @@ function createFakeClient(config: { rejectCheckoutDiffSubscribe?: boolean } = {}
     subscriptionId: string;
   }>;
   unsubscribeCheckoutDiffCalls: string[];
-  subscribeTerminalCalls: Array<{ cwd: string; workspaceId?: string }>;
-  unsubscribeTerminalCalls: Array<{ cwd: string; workspaceId?: string }>;
+  subscribeTerminalCalls: Array<{ cwd?: string; workspaceId?: string }>;
+  unsubscribeTerminalCalls: Array<{ cwd?: string; workspaceId?: string }>;
 } {
   const handlers: Record<RouterMessageType, RouterHandler[]> = {
     providers_snapshot_update: [],
@@ -71,8 +71,8 @@ function createFakeClient(config: { rejectCheckoutDiffSubscribe?: boolean } = {}
     subscriptionId: string;
   }> = [];
   const unsubscribeCheckoutDiffCalls: string[] = [];
-  const subscribeTerminalCalls: Array<{ cwd: string; workspaceId?: string }> = [];
-  const unsubscribeTerminalCalls: Array<{ cwd: string; workspaceId?: string }> = [];
+  const subscribeTerminalCalls: Array<{ cwd?: string; workspaceId?: string }> = [];
+  const unsubscribeTerminalCalls: Array<{ cwd?: string; workspaceId?: string }> = [];
 
   function on<K extends RouterMessageType>(
     type: K,

packages/app/src/plugins/client-state/api.ts

@@ -15,11 +15,11 @@ interface InstallationApi {
 }
 // Registry presentation clones share the installation lifetime and its client pool.
 const apis = new WeakMap<AbortSignal, InstallationApi>();
-const state = createPluginClientStateSource();
 
 export function getPluginClientApi(plugin: InstalledPlugin): InstallationApi {
   const cached = apis.get(plugin.lifetime.signal);
   if (cached) return cached;
+  const state = createPluginClientStateSource();
   const hosts = createPluginHosts(getHostRuntimeStore(), plugin.lifetime.signal);
   const paseo = createPluginReferenceApi({
     paseo: plugin.paseo,

packages/app/src/plugins/client-state/source.browser.test.tsx

@@ -7,7 +7,7 @@ import { fr } from "@/i18n/resources/fr";
 import React, { act } from "react";
 import { createRoot, type Root } from "react-dom/client";
 import { afterEach, describe, expect, it } from "vitest";
-import { PluginClientStateProvider } from "@getpaseo/plugin/client/host";
+import { createEntityRefs, PluginClientStateProvider } from "@getpaseo/plugin/client/host";
 import {
   useAgents,
   useWorkspaces,
@@ -377,3 +377,44 @@ describe("sidebar data hooks through the app store and plugin provider", () => {
     expect(new Set(source.getSnapshot().workspaces.map((item) => item.ref.key)).size).toBe(4);
   });
 });
+
+it("observes terminals only while a terminal reader is subscribed", () => {
+  let active = 0;
+  const source = createPluginClientStateSource(undefined, {
+    hosts: { getSnapshot: () => [], subscribe: () => () => {} },
+    terminals: {
+      getSnapshot: () => [],
+      subscribe: () => {
+        active++;
+        return () => {
+          active--;
+        };
+      },
+    },
+  });
+  const releaseWorkspace = source.subscribe(() => {});
+  expect(active).toBe(0);
+  const releaseTerminal = source.subscribe(() => {}, { terminals: true });
+  const releaseSecond = source.subscribe(() => {}, { terminals: true });
+  expect(active).toBe(1);
+  releaseTerminal();
+  expect(active).toBe(1);
+  releaseSecond();
+  expect(active).toBe(0);
+  releaseWorkspace();
+});
+
+it("terminal row reads propagate only the referenced host's observation error", () => {
+  const error = new Error("Update the host");
+  const source = createPluginClientStateSource(undefined, {
+    hosts: { getSnapshot: () => [], subscribe: () => () => {} },
+    terminals: {
+      getSnapshot: () => [],
+      subscribe: () => () => {},
+      getErrors: () => new Map([["unsupported", error]]),
+    },
+  });
+  const refs = createEntityRefs();
+  expect(() => source.getTerminal(refs.terminal("unsupported", "retained"))).toThrow(error);
+  expect(source.getTerminal(refs.terminal("supported", "missing"))).toBeNull();
+});

packages/app/src/plugins/client-state/source.ts

@@ -5,6 +5,7 @@ import { useCreateFlowStore } from "@/stores/create-flow-store";
 import { useSessionStore, type SessionState } from "@/stores/session-store";
 import { normalizeWorkspaceOpaqueId } from "@/utils/workspace-identity";
 import { connectedPluginHosts } from "./connected-hosts";
+import { pluginTerminals } from "./terminals";
 import { entityRefs } from "./refs";
 import {
   createPluginAgentSnapshot,
@@ -20,13 +21,10 @@ export interface PluginConnectedHostsSource {
 export interface PluginTerminalsSource {
   subscribe(listener: () => void): () => void;
   getSnapshot(): readonly PluginTerminalSnapshot[];
+  getErrors?(): ReadonlyMap<string, Error>;
 }
 
 const emptyTerminals: readonly PluginTerminalSnapshot[] = Object.freeze([]);
-const noTerminals: PluginTerminalsSource = {
-  subscribe: () => () => {},
-  getSnapshot: () => emptyTerminals,
-};
 const listeners = new Set<() => void>();
 let releaseStores: (() => void) | null = null;
 
@@ -139,12 +137,15 @@ export function createPluginClientStateSource(
   serverId?: string,
   options: { hosts?: PluginConnectedHostsSource; terminals?: PluginTerminalsSource } = {},
 ): PluginClientStateSource {
-  const terminals = options.terminals ?? noTerminals;
+  const terminals = options.terminals ?? pluginTerminals;
   const hosts = options.hosts ?? connectedPluginHosts;
   const subscribers = new Set<() => void>();
+  const terminalSubscribers = new Set<() => void>();
+  let releaseTerminals: (() => void) | null = null;
   let release: (() => void) | null = null;
   let previousProjections: readonly HostProjection[] = [];
   let previousTerminals = emptyTerminals;
+  let previousTerminalErrors: ReadonlyMap<string, Error> | undefined;
   let snapshot: PluginClientSnapshot = {
     workspaces: [],
     agents: [],
@@ -159,41 +160,58 @@ export function createPluginClientStateSource(
       if (projection) projections.push(projection);
     }
     const terminalSnapshots = terminals.getSnapshot();
-    if (sameSequence(previousProjections, projections) && previousTerminals === terminalSnapshots)
+    const terminalErrors = terminals.getErrors?.();
+    if (
+      sameSequence(previousProjections, projections) &&
+      previousTerminals === terminalSnapshots &&
+      previousTerminalErrors === terminalErrors
+    )
       return snapshot;
     previousProjections = projections;
     previousTerminals = terminalSnapshots;
+    previousTerminalErrors = terminalErrors;
     snapshot = Object.freeze({
       workspaces: Object.freeze(projections.flatMap((host) => host.snapshot.workspaces)),
       agents: Object.freeze(projections.flatMap((host) => host.snapshot.agents)),
       projects: Object.freeze(projections.flatMap((host) => host.snapshot.projects)),
       terminals: terminalSnapshots,
+      terminalErrors,
     });
     return snapshot;
   }
   return {
-    subscribe(listener) {
+    subscribe(listener, subscriptionOptions) {
       subscribers.add(listener);
+      if (subscriptionOptions?.terminals) {
+        terminalSubscribers.add(listener);
+        releaseTerminals ??= terminals.subscribe(() => {
+          for (const subscriber of terminalSubscribers) subscriber();
+        });
+      }
       if (!release) {
         const notify = () => {
           for (const subscriber of subscribers) subscriber();
         };
         const releaseStore = subscribeStores(notify);
         const releaseHosts = hosts.subscribe(notify);
-        const releaseTerminals = terminals.subscribe(notify);
         release = () => {
           releaseStore();
           releaseHosts();
-          releaseTerminals();
         };
       }
       return () => {
         subscribers.delete(listener);
+        terminalSubscribers.delete(listener);
+        if (terminalSubscribers.size === 0 && releaseTerminals) {
+          releaseTerminals();
+          releaseTerminals = null;
+        }
         if (subscribers.size === 0 && release) {
           release();
           release = null;
           previousProjections = [];
           previousTerminals = emptyTerminals;
+          previousTerminalErrors = undefined;
           snapshot = { workspaces: [], agents: [], projects: [], terminals: emptyTerminals };
         }
       };
@@ -221,6 +239,10 @@ export function createPluginClientStateSource(
       return detail && workspaceId ? createPluginAgentSnapshot(detail, workspaceId) : null;
     },
     getProject: (ref) => projectHost(ref.hostId)?.projectById.get(ref.id) ?? null,
-    getTerminal: (ref) => terminals.getSnapshot().find((terminal) => terminal.ref === ref) ?? null,
+    getTerminal(ref) {
+      const error = terminals.getErrors?.().get(ref.hostId);
+      if (error) throw error;
+      return terminals.getSnapshot().find((terminal) => terminal.ref === ref) ?? null;
+    },
   };
 }

packages/app/src/plugins/client-state/terminals.ts

@@ -0,0 +1,154 @@
+import type { PluginTerminalSnapshot } from "@getpaseo/plugin";
+import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
+import type { TerminalsChanged } from "@getpaseo/protocol/messages";
+import { getHostRuntimeStore } from "@/runtime/host-runtime";
+import { entityRefs } from "./refs";
+import type { PluginTerminalsSource } from "./source";
+
+interface TerminalHost {
+  serverId: string;
+  client: DaemonClient;
+  epoch: number;
+}
+export interface TerminalHostsSource {
+  subscribe(listener: () => void): () => void;
+  getSnapshot(): readonly TerminalHost[];
+}
+interface Observation {
+  client: DaemonClient;
+  epoch: number;
+  items: readonly PluginTerminalSnapshot[];
+  release(): void;
+}
+
+/** One owned observation per connected host, shared by every plugin surface. */
+export function createPluginTerminalsSource(hosts: TerminalHostsSource): PluginTerminalsSource {
+  const observations = new Map<string, Observation>();
+  const listeners = new Set<() => void>();
+  let snapshot: readonly PluginTerminalSnapshot[] = Object.freeze([]);
+  let errors: ReadonlyMap<string, Error> = new Map();
+  let releaseHosts: (() => void) | null = null;
+  function publish() {
+    const next = [...observations.values()].flatMap((item) => item.items);
+    if (next.length !== snapshot.length || next.some((item, index) => item !== snapshot[index]))
+      snapshot = Object.freeze(next);
+    for (const listener of listeners) listener();
+  }
+  function receive(hostId: string, observation: Observation, payload: TerminalsChanged["payload"]) {
+    if (observations.get(hostId) !== observation) return;
+    const previous = new Map(observation.items.map((item) => [item.ref.id, item]));
+    observation.items = Object.freeze(
+      payload.terminals.flatMap((terminal) => {
+        // Host-wide observation includes unowned legacy terminals. Sidebar navigation needs a workspace.
+        if (!terminal.workspaceId) return [];
+        const ref = entityRefs.terminal(hostId, terminal.id);
+        const workspace = entityRefs.workspace(hostId, terminal.workspaceId);
+        const title = terminal.title ?? terminal.name;
+        const createdAt = terminal.createdAt ?? null;
+        const isScript = terminal.isScript === true;
+        const cached = previous.get(terminal.id);
+        if (
+          cached &&
+          cached.workspace === workspace &&
+          cached.title === title &&
+          cached.createdAt === createdAt &&
+          cached.isScript === isScript
+        )
+          return [cached];
+        return [Object.freeze({ ref, hostId, workspace, title, createdAt, isScript })];
+      }),
+    );
+    errors = new Map([...errors].filter(([id]) => id !== hostId));
+    publish();
+  }
+  function reconcile() {
+    const active = new Map(hosts.getSnapshot().map((host) => [host.serverId, host]));
+    for (const [id, observation] of observations) {
+      if (
+        active.get(id)?.client !== observation.client ||
+        active.get(id)?.epoch !== observation.epoch
+      ) {
+        observations.delete(id);
+        observation.release();
+      }
+    }
+    errors = new Map([...errors].filter(([id]) => active.has(id)));
+    for (const [hostId, { client, epoch }] of active) {
+      if (observations.has(hostId)) continue;
+      const observation: Observation = { client, epoch, items: [], release() {} };
+      observations.set(hostId, observation);
+      try {
+        // The SDK owns the capability gate; a failed host is reported only to terminal readers.
+        const subscription = client.observeTerminals({});
+        const unsubscribe = subscription.subscribe({
+          snapshot: (payload) => receive(hostId, observation, payload),
+          update: (message) => {
+            if (message.type === "terminals_changed") receive(hostId, observation, message.payload);
+          },
+          error: (error) => {
+            if (observations.get(hostId) !== observation) return;
+            errors = new Map(errors).set(
+              hostId,
+              error instanceof Error ? error : new Error(String(error)),
+            );
+            publish();
+          },
+        });
+        void subscription.ready.catch(() => {}); // Observer owns error delivery.
+        observation.release = () => {
+          unsubscribe();
+          void subscription.release().catch(() => {});
+        };
+      } catch (error) {
+        errors = new Map(errors).set(
+          hostId,
+          error instanceof Error ? error : new Error(String(error)),
+        );
+      }
+    }
+    publish();
+  }
+  return {
+    getSnapshot: () => snapshot,
+    getErrors: () => errors,
+    subscribe(listener) {
+      listeners.add(listener);
+      if (!releaseHosts) {
+        releaseHosts = hosts.subscribe(reconcile);
+        reconcile();
+      }
+      return () => {
+        listeners.delete(listener);
+        if (!listeners.size && releaseHosts) {
+          releaseHosts();
+          releaseHosts = null;
+          for (const observation of observations.values()) observation.release();
+          observations.clear();
+          snapshot = Object.freeze([]);
+          errors = new Map();
+        }
+      };
+    },
+  };
+}
+
+export const pluginTerminals = createPluginTerminalsSource({
+  getSnapshot() {
+    const runtime = getHostRuntimeStore();
+    return runtime.getHosts().flatMap((host) => {
+      const state = runtime.getSnapshot(host.serverId);
+      return state?.connectionStatus === "online" && state.client
+        ? [{ serverId: host.serverId, client: state.client, epoch: state.connectionEpoch }]
+        : [];
+    });
+  },
+  subscribe(listener) {
+    const runtime = getHostRuntimeStore();
+    const releaseRuntime = runtime.subscribeAll(listener);
+    const releaseHosts = runtime.subscribeHostList(listener);
+    return () => {
+      releaseRuntime();
+      releaseHosts();
+    };
+  },
+});