Reduce daemon churn from terminals and archived agents

Decouple terminal list snapshots from workspace status invalidation, and filter archived stored agents before projection so seeded homes do less work.
2026-06-15 20:07:33 +07:00
parent 754f6336c0
commit 1c9e78616f
16 changed files with 837 additions and 105 deletions
@@ -1,6 +1,7 @@
 import { it, expect, afterEach } from "vitest";
 import { isPlatform } from "../test-utils/platform.js";
 import { createTerminalManager, type TerminalManager } from "./terminal-manager.js";
+import type { TerminalWorkspaceContributionChangedEvent } from "./terminal-manager.js";
 import { existsSync, mkdtempSync, mkdirSync, readFileSync, realpathSync, rmSync } from "node:fs";
 import { join } from "node:path";
 import { tmpdir } from "node:os";
@@ -528,3 +529,81 @@ it("clearTerminalAttention leaves non-attention terminals unchanged", async () =

  expect(session.getActivity()?.state).toBe("working");
 });
+
+it("emits workspace contribution event when activity bucket changes", async () => {
+  manager = createTerminalManager();
+  const cwd = realpathSync(tmpdir());
+  const events: TerminalWorkspaceContributionChangedEvent[] = [];
+  const unsubscribe = manager.subscribeTerminalWorkspaceContributionChanged((event) => {
+    events.push(event);
+  });
+
+  const session = await manager.createTerminal({ cwd, workspaceId: "ws-test" });
+  await manager.setTerminalActivity(session.id, "working");
+
+  expect(events).toEqual([
+    {
+      terminalId: session.id,
+      cwd,
+      workspaceId: "ws-test",
+    },
+  ]);
+
+  unsubscribe();
+});
+
+it("does not emit workspace contribution event for title-only changes", async () => {
+  manager = createTerminalManager();
+  const cwd = realpathSync(tmpdir());
+  const events: TerminalWorkspaceContributionChangedEvent[] = [];
+  const unsubscribe = manager.subscribeTerminalWorkspaceContributionChanged((event) => {
+    events.push(event);
+  });
+
+  const session = await manager.createTerminal({ cwd });
+  manager.setTerminalTitle(session.id, "New title");
+
+  expect(events).toEqual([]);
+
+  unsubscribe();
+});
+
+it("emits workspace contribution event when a contributing terminal is removed", async () => {
+  manager = createTerminalManager();
+  const cwd = realpathSync(tmpdir());
+  const events: TerminalWorkspaceContributionChangedEvent[] = [];
+  const unsubscribe = manager.subscribeTerminalWorkspaceContributionChanged((event) => {
+    events.push(event);
+  });
+
+  const session = await manager.createTerminal({ cwd, workspaceId: "ws-test" });
+  await manager.setTerminalActivity(session.id, "working");
+  events.length = 0;
+  manager.killTerminal(session.id);
+
+  expect(events).toEqual([
+    {
+      terminalId: session.id,
+      cwd,
+      workspaceId: "ws-test",
+    },
+  ]);
+
+  unsubscribe();
+});
+
+it("does not emit workspace contribution event when an idle terminal is removed", async () => {
+  manager = createTerminalManager();
+  const cwd = realpathSync(tmpdir());
+  const events: TerminalWorkspaceContributionChangedEvent[] = [];
+  const unsubscribe = manager.subscribeTerminalWorkspaceContributionChanged((event) => {
+    events.push(event);
+  });
+
+  const session = await manager.createTerminal({ cwd });
+  manager.killTerminal(session.id);
+
+  expect(events).toEqual([]);
+
+  unsubscribe();
+});
@@ -10,6 +10,7 @@ import { randomBytes, randomUUID } from "node:crypto";
 import { resolve, sep } from "node:path";
 import { assertAbsolutePath, isSameOrDescendantPath } from "../server/path-utils.js";
 import type { TerminalActivity, TerminalActivityState } from "@getpaseo/protocol/terminal-activity";
+import { deriveTerminalActivityStatusBucket } from "@getpaseo/protocol/terminal-activity";

 export interface TerminalListItem {
  id: string;
@@ -37,6 +38,16 @@ export interface TerminalActivityTransitionEvent {

 export type TerminalActivityListener = (event: TerminalActivityTransitionEvent) => void;

+export interface TerminalWorkspaceContributionChangedEvent {
+  terminalId: string;
+  cwd: string;
+  workspaceId?: string;
+}
+
+export type TerminalWorkspaceContributionChangedListener = (
+  event: TerminalWorkspaceContributionChangedEvent,
+) => void;
+
 export interface TerminalManager {
  getTerminals(cwd: string, options?: { workspaceId?: string }): Promise<TerminalSession[]>;
  createTerminal(options: {
@@ -74,6 +85,9 @@ export interface TerminalManager {
  killAll(): void;
  subscribeTerminalsChanged(listener: TerminalsChangedListener): () => void;
  subscribeTerminalActivity(listener: TerminalActivityListener): () => void;
+  subscribeTerminalWorkspaceContributionChanged(
+    listener: TerminalWorkspaceContributionChangedListener,
+  ): () => void;
 }

 export interface TerminalManagerOptions {
@@ -95,6 +109,8 @@ export function createTerminalManager(
  const terminalActivityTokenById = new Map<string, string>();
  const terminalsChangedListeners = new Set<TerminalsChangedListener>();
  const terminalActivityListeners = new Set<TerminalActivityListener>();
+  const terminalWorkspaceContributionChangedListeners =
+    new Set<TerminalWorkspaceContributionChangedListener>();
  const defaultEnvByRootCwd = new Map<string, Record<string, string>>();

  function removeSessionById(id: string, options: { kill: boolean }): void {
@@ -137,6 +153,16 @@ export function createTerminalManager(
      session.kill();
    }

+    const previousActivity = session.getActivity();
+    const previousBucket = deriveTerminalActivityStatusBucket(previousActivity);
+    if (previousBucket !== null) {
+      emitTerminalWorkspaceContributionChanged({
+        terminalId: session.id,
+        cwd: session.cwd,
+        ...(session.workspaceId ? { workspaceId: session.workspaceId } : {}),
+      });
+    }
+
    emitTerminalsChanged({ cwd: session.cwd });
  }

@@ -168,6 +194,15 @@ export function createTerminalManager(
    const unsubscribeActivity = session.onActivityChange((transition) => {
      emitTerminalActivityTransition({ session, transition });
      emitTerminalsChanged({ cwd: session.cwd });
+      const previousBucket = deriveTerminalActivityStatusBucket(transition.previous);
+      const nextBucket = deriveTerminalActivityStatusBucket(transition.activity);
+      if (previousBucket !== nextBucket) {
+        emitTerminalWorkspaceContributionChanged({
+          terminalId: session.id,
+          cwd: session.cwd,
+          ...(session.workspaceId ? { workspaceId: session.workspaceId } : {}),
+        });
+      }
    });
    terminalExitUnsubscribeById.set(session.id, unsubscribeExit);
    terminalTitleUnsubscribeById.set(session.id, unsubscribeTitle);
@@ -231,6 +266,18 @@ export function createTerminalManager(
    }
  }

+  function emitTerminalWorkspaceContributionChanged(
+    event: TerminalWorkspaceContributionChangedEvent,
+  ): void {
+    for (const listener of terminalWorkspaceContributionChangedListeners) {
+      try {
+        listener(event);
+      } catch {
+        // no-op
+      }
+    }
+  }
+
  return {
    async getTerminals(
      cwd: string,
@@ -431,5 +478,14 @@ export function createTerminalManager(
        terminalActivityListeners.delete(listener);
      };
    },
+
+    subscribeTerminalWorkspaceContributionChanged(
+      listener: TerminalWorkspaceContributionChangedListener,
+    ): () => void {
+      terminalWorkspaceContributionChangedListeners.add(listener);
+      return () => {
+        terminalWorkspaceContributionChangedListeners.delete(listener);
+      };
+    },
  };
 }
@@ -98,6 +98,7 @@ describe("terminal-session-controller restore", () => {
      killAll: vi.fn(),
      subscribeTerminalsChanged: vi.fn(() => vi.fn()),
      subscribeTerminalActivity: vi.fn(() => vi.fn()),
+      subscribeTerminalWorkspaceContributionChanged: vi.fn(() => vi.fn()),
    };
    const controller = new TerminalSessionController({
      terminalManager,
@@ -232,6 +233,7 @@ describe("terminal-session-controller wrap-flag gating", () => {
      killAll: vi.fn(),
      subscribeTerminalsChanged: vi.fn(() => vi.fn()),
      subscribeTerminalActivity: vi.fn(() => vi.fn()),
+      subscribeTerminalWorkspaceContributionChanged: vi.fn(() => vi.fn()),
    } as unknown as TerminalManager;
    const controller = new TerminalSessionController({
      terminalManager,
@@ -307,6 +309,7 @@ describe("terminal-session-controller subdirectory aggregation", () => {
        return vi.fn();
      }),
      subscribeTerminalActivity: vi.fn(() => vi.fn()),
+      subscribeTerminalWorkspaceContributionChanged: vi.fn(() => vi.fn()),
    };

    const outboundMessages: SessionOutboundMessage[] = [];
@@ -371,6 +374,7 @@ describe("terminal-session-controller subdirectory aggregation", () => {
      killAll: vi.fn(),
      subscribeTerminalsChanged: vi.fn(() => vi.fn()),
      subscribeTerminalActivity: vi.fn(() => vi.fn()),
+      subscribeTerminalWorkspaceContributionChanged: vi.fn(() => vi.fn()),
    };
    const outboundMessages: SessionOutboundMessage[] = [];
    const controller = new TerminalSessionController({
@@ -449,6 +453,7 @@ describe("terminal-session-controller workspace-scoped subscriptions", () => {
        return vi.fn();
      }),
      subscribeTerminalActivity: vi.fn(() => vi.fn()),
+      subscribeTerminalWorkspaceContributionChanged: vi.fn(() => vi.fn()),
    };

    const outboundMessages: SessionOutboundMessage[] = [];
@@ -188,14 +188,6 @@ function watchTerminal(session: TerminalSession): void {
  ]);
 }

-manager.subscribeTerminalsChanged((event) => {
-  sendToParent({
-    type: "terminalsChanged",
-    cwd: event.cwd,
-    terminals: event.terminals,
-  });
-});
-
 function enqueueCreateTerminalRequest(message: TerminalCreateRequest): Promise<void> {
  const nextRequest = createTerminalQueue.then(() => handleCreateTerminalRequest(message));
  createTerminalQueue = nextRequest.catch(() => {});
@@ -265,33 +257,17 @@ async function handleRequest(message: TerminalWorkerRequest): Promise<void> {
    }

    case "killTerminal": {
-      const session = manager.getTerminal(message.terminalId);
-      const cwd = session?.cwd;
      manager.killTerminal(message.terminalId);
+      // Removal is owned by session.onExit -> terminalExit; the parent mirror
+      // clears contribution and emits terminalsChanged from that single path.
      clearTerminalSubscriptions(message.terminalId);
-      if (cwd) {
-        sendToParent({
-          type: "terminalRemoved",
-          terminalId: message.terminalId,
-          cwd,
-        });
-      }
      sendToParent({ type: "response", requestId: message.requestId, ok: true });
      return;
    }

    case "killTerminalAndWait": {
-      const session = manager.getTerminal(message.terminalId);
-      const cwd = session?.cwd;
      await manager.killTerminalAndWait(message.terminalId, message.options);
      clearTerminalSubscriptions(message.terminalId);
-      if (cwd) {
-        sendToParent({
-          type: "terminalRemoved",
-          terminalId: message.terminalId,
-          cwd,
-        });
-      }
      sendToParent({ type: "response", requestId: message.requestId, ok: true });
      return;
    }
@@ -115,11 +115,6 @@ export type TerminalWorkerEvent =
      terminal: WorkerTerminalInfo;
      state: TerminalState;
    }
-  | {
-      type: "terminalRemoved";
-      terminalId: string;
-      cwd: string;
-    }
  | {
      type: "terminalMessage";
      terminalId: string;
@@ -142,11 +137,6 @@ export type TerminalWorkerEvent =
        exitCode: number | null;
      };
    }
-  | {
-      type: "terminalsChanged";
-      cwd: string;
-      terminals: WorkerTerminalInfo[];
-    }
  | {
      type: "terminalActivityChange";
      terminalId: string;
@@ -5,7 +5,11 @@ import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync
 import { delimiter, join } from "node:path";
 import { tmpdir } from "node:os";
 import { createWorkerTerminalManager } from "./worker-terminal-manager.js";
-import type { TerminalActivityTransitionEvent, TerminalManager } from "./terminal-manager.js";
+import type {
+  TerminalActivityTransitionEvent,
+  TerminalManager,
+  TerminalWorkspaceContributionChangedEvent,
+} from "./terminal-manager.js";
 import {
  resolvePaseoCliBinDir,
  resolvePaseoCliExecutablePath,
@@ -753,3 +757,148 @@ it("removes worker terminals after killAndWait", async () => {
  expect(manager.getTerminal(session.id)).toBeUndefined();
  expect(manager.listDirectories()).not.toContain(cwd);
 });
+
+it("produces one terminals-changed snapshot per title change", async () => {
+  const worker = new FakeTerminalWorker();
+  manager = createWorkerTerminalManager({
+    requestTimeoutMs: 50,
+    forkWorker: () => worker,
+  });
+
+  worker.emitWorkerMessage({
+    type: "terminalCreated",
+    terminal: {
+      id: "terminal-a",
+      name: "Shell",
+      cwd: "/workspace",
+      activity: null,
+    },
+    state: createTerminalState(),
+  });
+
+  const snapshots: Array<{ cwd: string; terminalIds: string[] }> = [];
+  manager.subscribeTerminalsChanged((event) => {
+    snapshots.push({ cwd: event.cwd, terminalIds: event.terminals.map((t) => t.id) });
+  });
+
+  worker.emitWorkerMessage({
+    type: "terminalTitleChange",
+    terminalId: "terminal-a",
+    title: "Updated",
+  });
+
+  expect(snapshots).toHaveLength(1);
+  expect(snapshots[0]).toEqual({
+    cwd: "/workspace",
+    terminalIds: ["terminal-a"],
+  });
+});
+
+it("produces one terminals-changed snapshot and one contribution event per activity change", async () => {
+  const worker = new FakeTerminalWorker();
+  manager = createWorkerTerminalManager({
+    requestTimeoutMs: 50,
+    forkWorker: () => worker,
+  });
+
+  worker.emitWorkerMessage({
+    type: "terminalCreated",
+    terminal: {
+      id: "terminal-a",
+      name: "Shell",
+      cwd: "/workspace",
+      workspaceId: "ws-test",
+      activity: null,
+    },
+    state: createTerminalState(),
+  });
+
+  const snapshots: Array<{ cwd: string; terminalIds: string[] }> = [];
+  manager.subscribeTerminalsChanged((event) => {
+    snapshots.push({ cwd: event.cwd, terminalIds: event.terminals.map((t) => t.id) });
+  });
+  const contributions: TerminalWorkspaceContributionChangedEvent[] = [];
+  manager.subscribeTerminalWorkspaceContributionChanged((event) => {
+    contributions.push(event);
+  });
+
+  const workingActivity = { state: "working" as const, changedAt: 1000 };
+  worker.emitWorkerMessage({
+    type: "terminalActivityChange",
+    terminalId: "terminal-a",
+    activity: workingActivity,
+    previous: null,
+  });
+
+  expect(snapshots).toHaveLength(1);
+  expect(snapshots[0]).toEqual({
+    cwd: "/workspace",
+    terminalIds: ["terminal-a"],
+  });
+  expect(contributions).toEqual([
+    {
+      terminalId: "terminal-a",
+      cwd: "/workspace",
+      workspaceId: "ws-test",
+    },
+  ]);
+});
+
+it("removes a killed worker terminal from terminalExit without duplicate snapshots", async () => {
+  const worker = new FakeTerminalWorker();
+  manager = createWorkerTerminalManager({
+    requestTimeoutMs: 50,
+    forkWorker: () => worker,
+  });
+  const workingActivity = { state: "working" as const, changedAt: 1000 };
+
+  worker.emitWorkerMessage({
+    type: "terminalCreated",
+    terminal: {
+      id: "terminal-a",
+      name: "Shell",
+      cwd: "/workspace",
+      workspaceId: "ws-test",
+      activity: workingActivity,
+    },
+    state: createTerminalState(),
+  });
+
+  const snapshots: Array<{ cwd: string; terminalIds: string[] }> = [];
+  manager.subscribeTerminalsChanged((event) => {
+    snapshots.push({ cwd: event.cwd, terminalIds: event.terminals.map((terminal) => terminal.id) });
+  });
+  const contributions: TerminalWorkspaceContributionChangedEvent[] = [];
+  manager.subscribeTerminalWorkspaceContributionChanged((event) => {
+    contributions.push(event);
+  });
+
+  manager.killTerminal("terminal-a");
+  const request = worker.sentMessages.find(
+    (message) => message.type === "killTerminal" && message.terminalId === "terminal-a",
+  );
+  if (!request) {
+    throw new Error("killTerminal request not sent");
+  }
+  worker.emitWorkerMessage({ type: "response", requestId: request.requestId, ok: true });
+
+  worker.emitWorkerMessage({
+    type: "terminalExit",
+    terminalId: "terminal-a",
+    info: {
+      exitCode: null,
+      signal: null,
+      lastOutputLines: [],
+    },
+  });
+
+  expect(manager.getTerminal("terminal-a")).toBeUndefined();
+  expect(snapshots).toEqual([{ cwd: "/workspace", terminalIds: [] }]);
+  expect(contributions).toEqual([
+    {
+      terminalId: "terminal-a",
+      cwd: "/workspace",
+      workspaceId: "ws-test",
+    },
+  ]);
+});
@@ -4,6 +4,7 @@ import { randomBytes, randomUUID } from "node:crypto";
 import { assertAbsolutePath, isSameOrDescendantPath } from "../server/path-utils.js";
 import type { TerminalState } from "@getpaseo/protocol/messages";
 import type { TerminalActivity, TerminalActivityState } from "@getpaseo/protocol/terminal-activity";
+import { deriveTerminalActivityStatusBucket } from "@getpaseo/protocol/terminal-activity";
 import type {
  ClientMessage,
  ServerMessage,
@@ -19,6 +20,8 @@ import type {
  TerminalActivityTransitionEvent,
  TerminalListItem,
  TerminalManager,
+  TerminalWorkspaceContributionChangedEvent,
+  TerminalWorkspaceContributionChangedListener,
  TerminalsChangedEvent,
  TerminalsChangedListener,
 } from "./terminal-manager.js";
@@ -141,6 +144,8 @@ export function createWorkerTerminalManager(
  const terminalActivityTokenById = new Map<string, string>();
  const terminalsChangedListeners = new Set<TerminalsChangedListener>();
  const terminalActivityListeners = new Set<TerminalActivityListener>();
+  const terminalWorkspaceContributionChangedListeners =
+    new Set<TerminalWorkspaceContributionChangedListener>();
  let workerExited = false;
  let workerShutdownTimer: ReturnType<typeof setTimeout> | null = null;

@@ -154,6 +159,18 @@ export function createWorkerTerminalManager(
    }
  }

+  function emitTerminalWorkspaceContributionChanged(
+    event: TerminalWorkspaceContributionChangedEvent,
+  ): void {
+    for (const listener of Array.from(terminalWorkspaceContributionChangedListeners)) {
+      try {
+        listener(event);
+      } catch {
+        // no-op
+      }
+    }
+  }
+
  function emitTerminalActivityTransition(event: TerminalActivityTransitionEvent): void {
    for (const listener of Array.from(terminalActivityListeners)) {
      try {
@@ -415,7 +432,15 @@ export function createWorkerTerminalManager(
      listener(message.info);
    }
    record.exitListeners.clear();
-    removeRecord(message.terminalId);
+    const previousBucket = deriveTerminalActivityStatusBucket(record.activity);
+    const removedRecord = removeRecord(message.terminalId);
+    if (previousBucket !== null && removedRecord) {
+      emitTerminalWorkspaceContributionChanged({
+        terminalId: removedRecord.info.id,
+        cwd: removedRecord.info.cwd,
+        ...(removedRecord.info.workspaceId ? { workspaceId: removedRecord.info.workspaceId } : {}),
+      });
+    }
    emitTerminalsChanged({
      cwd: record.info.cwd,
      terminals: listTerminalItemsForCwd(record.info.cwd),
@@ -468,6 +493,7 @@ export function createWorkerTerminalManager(
    if (!record) {
      return;
    }
+    const previousActivity = record.activity;
    record.activity = message.activity;
    const transition: TerminalActivityTransition = {
      activity: message.activity,
@@ -483,46 +509,28 @@ export function createWorkerTerminalManager(
      activity: message.activity,
      previous: message.previous,
    });
+    const previousBucket = deriveTerminalActivityStatusBucket(previousActivity);
+    const nextBucket = deriveTerminalActivityStatusBucket(message.activity);
+    if (previousBucket !== nextBucket) {
+      emitTerminalWorkspaceContributionChanged({
+        terminalId: record.info.id,
+        cwd: record.info.cwd,
+        ...(record.info.workspaceId ? { workspaceId: record.info.workspaceId } : {}),
+      });
+    }
    emitTerminalsChanged({
      cwd: record.info.cwd,
      terminals: listTerminalItemsForCwd(record.info.cwd),
    });
  }

-  function handleTerminalsChangedEvent(
-    message: Extract<TerminalWorkerToParentMessage, { type: "terminalsChanged" }>,
-  ): void {
-    for (const terminal of message.terminals) {
-      const record = recordsById.get(terminal.id);
-      if (record) {
-        record.activity = terminal.activity;
-      }
-    }
-    emitTerminalsChanged({
-      cwd: message.cwd,
-      terminals: message.terminals.map((terminal) => ({
-        id: terminal.id,
-        name: terminal.name,
-        cwd: terminal.cwd,
-        ...(terminal.workspaceId ? { workspaceId: terminal.workspaceId } : {}),
-        ...(terminal.title ? { title: terminal.title } : {}),
-        activity: terminal.activity,
-      })),
-    });
-  }
-
  function handleWorkerEvent(message: TerminalWorkerToParentMessage): void {
    switch (message.type) {
      case "terminalCreated": {
        registerRecord({ info: message.terminal, state: message.state });
-        return;
-      }
-
-      case "terminalRemoved": {
-        removeRecord(message.terminalId);
        emitTerminalsChanged({
-          cwd: message.cwd,
-          terminals: listTerminalItemsForCwd(message.cwd),
+          cwd: message.terminal.cwd,
+          terminals: listTerminalItemsForCwd(message.terminal.cwd),
        });
        return;
      }
@@ -547,11 +555,6 @@ export function createWorkerTerminalManager(
        return;
      }

-      case "terminalsChanged": {
-        handleTerminalsChangedEvent(message);
-        return;
-      }
-
      case "terminalActivityChange": {
        handleTerminalActivityChangeEvent(message);
        return;
@@ -827,6 +830,15 @@ export function createWorkerTerminalManager(
        terminalActivityListeners.delete(listener);
      };
    },
+
+    subscribeTerminalWorkspaceContributionChanged(
+      listener: TerminalWorkspaceContributionChangedListener,
+    ): () => void {
+      terminalWorkspaceContributionChangedListeners.add(listener);
+      return () => {
+        terminalWorkspaceContributionChangedListeners.delete(listener);
+      };
+    },
  };
 }