fix(app): reuse file tabs opened from nested-project Changes

Changes paths are relative to the current worktree. Resolve file actions
from that root, then give file tabs and chat attachments the existing
workspace-relative target for files inside the project. Keep absolute
targets for changed files outside it so both paths remain addressable.

The contributor's root-path fix opened inside-project files with absolute
targets, which differed from Files and created duplicate retained tabs.
A failing-first browser journey covers navigation between both views.

Co-authored-by: 3ae3ae <ogyshjsm@naver.com>

docs/plugins.md

@@ -357,27 +357,9 @@ keys are stored under a `param.` prefix; plugins never see it and may use any ke
 `openScreen` validates its input once (`parsePluginOpenScreenInput`); code past it trusts the
 params.
 
-RPC and subscription contracts validate inputs, outputs, and events in both the app and plugin
-subprocess. `useRpc` returns a typed async function. Use the host-provided `@tanstack/react-query`
-for request state and caching; Paseo gives each plugin installation its own query client.
-
-A subscription is a session-owned observation (`plugin-rpc` delivery family): the session that
-subscribed owns its lifetime, so release, disconnect, plugin stop, and reload need no plugin code.
-Each subscriber gets its own handler call in the subprocess; there is no fan-out or replay. The
-runtime buffers pushes until the session has sent `plugin.rpc.subscribe.response`, so a client
-never sees an event for an unknown subscription. The app installation's lifetime aborts every
-subscription its surfaces left open, and a reload replaces the installation, which resubscribes
-mounted hooks.
-
-The byte rule (a top-level `Uint8Array` is binary, a nested one is a validation error) is checked
-once per hop by `assertPluginWireValue`. Bytes ride plugin binary frames, opcodes `0x20`-`0x2f`
-(`packages/protocol/src/binary-frames/plugin.ts`); nothing on the path copies or re-encodes them,
-and `plugin-subscriptions.perf.e2e.test.ts` locks that. Keep it that way: never stringify a plugin
-payload for a log, and the client's bootstrap buffer holds bytes only until the first observer
-attaches. Binary outputs reach only clients that send `pluginBinaryPayloads`. A binary input
-frame (`0x22`) becomes a synthetic `plugin.rpc.invoke.request` and runs through the normal message
-path, so it is authorized like any other invoke. The author contract, including the native upload
-limit, is in [stream events from the daemon](../public-docs/plugins/reference.md#subscriptions).
+RPC contracts validate inputs and outputs in both the app and plugin subprocess. `useRpc` returns a
+typed async function. Use the host-provided `@tanstack/react-query` for request state and caching;
+Paseo gives each plugin installation its own query client.
 
 `usePaseo()` and the handler's `{ paseo }` context expose the same `PaseoApi`: projects,
 workspaces, agents, terminals, providers, and daemon config. They do not expose connection lifecycle.

packages/app/e2e/browser/changes-pane.spec.ts

@@ -8,6 +8,7 @@ import { daemonWsRoutePattern } from "../support/helpers/daemon-port";
 import { getServerId } from "../support/helpers/server-id";
 import { connectSeedClient } from "../support/helpers/seed-client";
 import { createTempGitRepo } from "../support/helpers/workspace";
+import { openFileExplorer, openFileFromExplorer } from "../support/helpers/file-explorer";
 import { openChangesPanel, waitForWorkspaceTabsVisible } from "../support/helpers/workspace-tabs";
 
 interface DirtyWorkspace {
@@ -16,6 +17,7 @@ interface DirtyWorkspace {
 }
 
 interface WorkspaceFixtureOptions {
+  workspaceSubdirectory?: string;
   includeDeletedFile?: boolean;
   includeNestedFolders?: boolean;
   includeRenamedFile?: boolean;
@@ -435,6 +437,114 @@ test("changes file actions open below the right-click without a reserved kebab",
   await expect(page.getByTestId("workspace-tab-file_src/use-mounted-tab-set.ts")).toBeVisible();
 });
 
+for (const { workspaceSubdirectory, filePath, content, visibleContent } of [
+  {
+    workspaceSubdirectory: ".",
+    filePath: "src/use-mounted-tab-set.ts",
+    content: AFTER,
+    visibleContent: "interface DeriveRenderMountedTabIdsInput",
+  },
+  {
+    workspaceSubdirectory: "src",
+    filePath: "src/use-mounted-tab-set.ts",
+    content: AFTER,
+    visibleContent: "interface DeriveRenderMountedTabIdsInput",
+  },
+  {
+    workspaceSubdirectory: "src",
+    filePath: "a-root-note.txt",
+    content: "root note\n",
+    visibleContent: "root note",
+  },
+]) {
+  test(`changes file actions resolve ${filePath} from project ${workspaceSubdirectory}`, async ({
+    page,
+    context,
+  }, testInfo) => {
+    const workspace = await createWorkspaceWithMountedTabDiff({
+      workspaceSubdirectory,
+      includeRootFileSortingFirst: filePath === "a-root-note.txt",
+    });
+    await context.grantPermissions(["clipboard-read", "clipboard-write"]);
+    await openWorkspaceChanges(page, workspace);
+    const panel = page.getByTestId("working-diff-panel").filter({ visible: true });
+    const header = diffHeaderForPath(panel, filePath).getByTestId(/^diff-file-\d+-toggle$/);
+    await header.click({ button: "right" });
+    await testInfo.attach("changes-file-menu", {
+      body: await page.screenshot(),
+      contentType: "image/png",
+    });
+    await test.step("Duplicate creates a copy next to the source", async () => {
+      const file = path.parse(filePath);
+      const copyPath = path.join(workspace.repoPath, file.dir, `${file.name} copy${file.ext}`);
+      await page.getByRole("menuitem", { name: "Duplicate", exact: true }).click();
+      await expect
+        .poll(() => readFileIfPresent(copyPath), { message: `Expected duplicate at ${copyPath}` })
+        .toBe(content);
+    });
+    await header.click({ button: "right" });
+    await page.getByRole("menuitem", { name: "Copy relative path", exact: true }).click();
+    await expectClipboardText(page, filePath);
+    await header.click({ button: "right" });
+    await page.getByRole("menuitem", { name: "Copy path", exact: true }).click();
+    await expectClipboardText(page, path.join(workspace.repoPath, filePath));
+
+    await header.click({ button: "right" });
+    const downloaded = page.waitForEvent("download");
+    await page.getByRole("menuitem", { name: "Download", exact: true }).click();
+    const download = await downloaded;
+    expect(download.suggestedFilename()).toBe(path.basename(filePath));
+    expect(await readFile((await download.path())!, "utf8")).toBe(content);
+
+    await header.click({ button: "right" });
+    await page.getByRole("menuitem", { name: "Open file", exact: true }).click();
+    const editor = page.getByTestId("workspace-file-pane").filter({ visible: true });
+    await expect(editor.locator(".cm-content")).toContainText(visibleContent);
+    await testInfo.attach("opened-changed-file", {
+      body: await page.screenshot(),
+      contentType: "image/png",
+    });
+  });
+}
+
+test("Changes and Files reuse a retained file tab in a subfolder project", async ({ page }) => {
+  const workspace = await createWorkspaceWithMountedTabDiff({ workspaceSubdirectory: "src" });
+  await openWorkspaceChanges(page, workspace);
+  await openChangedFile(page, "src/use-mounted-tab-set.ts");
+  await editAndSaveChangedFile(page, workspace);
+  await openChangesPanel(page);
+  await openFileExplorer(page);
+  await openFileFromExplorer(page, "use-mounted-tab-set.ts");
+
+  await expect(page.getByTestId(/^workspace-tab-file_/).filter({ visible: true })).toHaveCount(1);
+  await expect(page.getByTestId("workspace-tab-file_use-mounted-tab-set.ts")).toHaveAttribute(
+    "aria-selected",
+    "true",
+  );
+  await expect(visibleFileEditor(page)).toContainText("// retained file tab");
+});
+
+async function openChangedFile(page: Page, filePath: string): Promise<void> {
+  const panel = page.getByTestId("working-diff-panel").filter({ visible: true });
+  await diffHeaderForPath(panel, filePath)
+    .getByTestId(/^diff-file-\d+-toggle$/)
+    .click({ button: "right" });
+  await page.getByRole("menuitem", { name: "Open file", exact: true }).click();
+  await expect(visibleFileEditor(page)).toContainText("interface DeriveRenderMountedTabIdsInput");
+}
+
+function visibleFileEditor(page: Page): Locator {
+  return page.getByTestId("workspace-file-pane").filter({ visible: true }).locator(".cm-content");
+}
+
+async function editAndSaveChangedFile(page: Page, workspace: DirtyWorkspace): Promise<void> {
+  const content = "// retained file tab\n" + AFTER;
+  await visibleFileEditor(page).fill(content);
+  await expect
+    .poll(() => readFile(path.join(workspace.repoPath, "src/use-mounted-tab-set.ts"), "utf8"))
+    .toBe(content);
+}
+
 test("canvas file headers select without toggling for context menu and long press", async ({
   page,
 }) => {
@@ -1591,7 +1701,7 @@ async function createWorkspaceWithMountedTabDiff(
     );
   }
   const createdWorkspace = await client.createWorkspace({
-    source: { kind: "directory", path: repo.path },
+    source: { kind: "directory", path: path.join(repo.path, options.workspaceSubdirectory ?? ".") },
   });
   if (!createdWorkspace.workspace) {
     throw new Error(createdWorkspace.error ?? `Failed to create workspace ${repo.path}`);

packages/app/e2e/browser/file-editing.spec.ts

@@ -63,24 +63,6 @@ function htmlPreview(page: Page) {
   };
 }
 
-async function followHtmlPreviewLink(page: Page, name: string): Promise<void> {
-  await htmlPreview(page).document.getByRole("link", { name, exact: true }).click();
-}
-
-async function expectPageHandlesFragment(
-  page: Page,
-  handler: "Document" | "Window",
-): Promise<void> {
-  const preview = htmlPreview(page);
-  await followHtmlPreviewLink(page, `${handler} handled jump`);
-  await expect(preview.document.getByRole("status")).toHaveText(`${handler} handled click`);
-  await expect(
-    preview.document.getByRole("heading", { name: "Target section" }),
-  ).not.toBeInViewport();
-  expect(await preview.document.locator("body").evaluate(() => location.hash)).toBe("");
-  expect(await preview.document.locator("body").evaluate(() => window.scrollY)).toBe(0);
-}
-
 async function selectFileView(page: Page, view: "Preview" | "Source"): Promise<void> {
   const option = page.getByTestId("file-panel-bar").getByRole("button", {
     name: view,
@@ -844,192 +826,6 @@ test.describe("CodeMirror workspace file editing", () => {
     await expect(preview.host).toBeVisible();
   });
 
-  test("keeps HTML fragment links in the preview and scrolls to the target", async ({
-    page,
-    withWorkspace,
-  }) => {
-    const workspace = await withWorkspace({ prefix: "html-fragments-" });
-    await writeFile(
-      path.join(workspace.repoPath, "fragments.html"),
-      `<!doctype html><html><body>
-<a href="#section">Jump to section</a>
-<div style="height: 1200px"></div>
-<h2 id="section">Target section</h2>
-</body></html>`,
-      "utf8",
-    );
-    await workspace.navigateTo();
-    await openWorkspaceFile(page, "fragments.html");
-
-    const preview = htmlPreview(page);
-    await expect(preview.document.getByRole("link", { name: "Jump to section" })).toBeVisible();
-    await page.screenshot({ path: test.info().outputPath("fragment-before-click.png") });
-    await preview.document.getByRole("link", { name: "Jump to section" }).click();
-
-    await expect(
-      preview.document.getByRole("heading", { name: "Target section" }),
-    ).toBeInViewport();
-    expect(await preview.document.locator("body").evaluate(() => location.hash)).toBe("#section");
-    expect(await preview.document.locator("body").evaluate(() => window.scrollY)).toBeGreaterThan(
-      0,
-    );
-    await page.screenshot({ path: test.info().outputPath("fragment-after-click.png") });
-  });
-
-  test("lets an HTML page prevent fragment navigation", async ({ page, withWorkspace }) => {
-    const workspace = await withWorkspace({ prefix: "html-fragment-handler-" });
-    await writeFile(
-      path.join(workspace.repoPath, "handled.html"),
-      `<!doctype html><html><body>
-<a href="#section" onclick="event.preventDefault(); document.querySelector('output').textContent = 'Page handled click'">Handled jump</a>
-<output>Waiting for click</output>
-<div style="height: 1200px"></div><h2 id="section">Target section</h2>
-</body></html>`,
-      "utf8",
-    );
-    await workspace.navigateTo();
-    await openWorkspaceFile(page, "handled.html");
-
-    const preview = htmlPreview(page);
-    await preview.document.getByRole("link", { name: "Handled jump" }).click();
-    await expect(preview.document.locator("output")).toHaveText("Page handled click");
-    await expect(
-      preview.document.getByRole("heading", { name: "Target section" }),
-    ).not.toBeInViewport();
-    expect(await preview.document.locator("body").evaluate(() => location.hash)).toBe("");
-    expect(await preview.document.locator("body").evaluate(() => window.scrollY)).toBe(0);
-  });
-
-  test("respects delegated HTML fragment handlers and normal URL parsing", async ({
-    page,
-    withWorkspace,
-  }) => {
-    const workspace = await withWorkspace({ prefix: "html-fragment-defaults-" });
-    await writeFile(
-      path.join(workspace.repoPath, "defaults.html"),
-      `<!doctype html><html><body>
-<nav style="position: fixed; top: 0; background: white">
-<a href="#section" id="document-handler">Document handled jump</a>
-<a href="#section" id="window-handler">Window handled jump</a>
-<a href=" \t#section\n">Whitespace jump</a>
-<a href="#100%">Percent jump</a>
-<output>Waiting for click</output>
-</nav>
-<div style="height: 1200px"></div><h2 id="section">Target section</h2>
-<div style="height: 1200px"></div><h2 id="100%">Percent target</h2>
-<div style="height: 1200px"></div>
-<script>
-document.addEventListener('click', function(event) {
-  if (event.target.id === 'document-handler') {
-    event.preventDefault(); document.querySelector('output').textContent = 'Document handled click';
-  }
-});
-window.addEventListener('click', function(event) {
-  if (event.target.id === 'window-handler') {
-    event.preventDefault(); document.querySelector('output').textContent = 'Window handled click';
-  }
-});
-</script></body></html>`,
-      "utf8",
-    );
-    await workspace.navigateTo();
-    await openWorkspaceFile(page, "defaults.html");
-
-    const preview = htmlPreview(page);
-    await expectPageHandlesFragment(page, "Document");
-    await expectPageHandlesFragment(page, "Window");
-
-    await followHtmlPreviewLink(page, "Whitespace jump");
-    await expect(
-      preview.document.getByRole("heading", { name: "Target section" }),
-    ).toBeInViewport();
-    await followHtmlPreviewLink(page, "Percent jump");
-    await expect(
-      preview.document.getByRole("heading", { name: "Percent target" }),
-    ).toBeInViewport();
-  });
-
-  test("handles HTML fragment targets, empty and missing fragments, and modified clicks", async ({
-    page,
-    withWorkspace,
-  }) => {
-    const workspace = await withWorkspace({ prefix: "html-fragment-targets-" });
-    await writeFile(
-      path.join(workspace.repoPath, "targets.html"),
-      `<!doctype html><html><head><style>:target { color: rgb(255, 0, 0); }</style></head><body>
-<nav style="position: fixed; top: 0; background: white">
-<a href="#section"><span>Nested jump</span></a>
-<a href="#legacy">Named jump</a>
-<a href="#encoded%20section">Encoded jump</a>
-<a href="#missing">Missing jump</a>
-<a href="#">Empty jump</a>
-<output>Waiting for hashchange</output><output id="modified">Waiting for modified click</output>
-</nav>
-<div style="height: 1200px"></div><h2 id="section">Target section</h2>
-<div style="height: 1200px"></div><a name="legacy">Named target</a>
-<div style="height: 1200px"></div><h2 id="encoded section">Encoded target</h2>
-<div style="height: 1200px"></div>
-<script>
-window.addEventListener('hashchange', function () { document.querySelector('output').textContent = location.hash || '(empty)'; });
-function modifiedClick(event) {
-  if (event.button === 1 || event.ctrlKey || event.metaKey || event.shiftKey || event.altKey) {
-    document.getElementById('modified').textContent = String(event.defaultPrevented);
-    event.preventDefault();
-  }
-}
-window.addEventListener('click', modifiedClick);
-window.addEventListener('auxclick', modifiedClick);
-window.addEventListener('contextmenu', modifiedClick);
-</script></body></html>`,
-      "utf8",
-    );
-    await workspace.navigateTo();
-    await openWorkspaceFile(page, "targets.html");
-
-    const preview = htmlPreview(page);
-    const body = preview.document.locator("body");
-    const nested = preview.document.getByText("Nested jump", { exact: true });
-    await nested.click();
-    await expect(preview.document.locator("output").first()).toHaveText("#section");
-    await expect(
-      preview.document.getByRole("heading", { name: "Target section" }),
-    ).toBeInViewport();
-    await expect(preview.document.locator("#section")).toHaveCSS("color", "rgb(255, 0, 0)");
-
-    await body.evaluate(() => window.scrollTo(0, 0));
-    await nested.click();
-    await expect(
-      preview.document.getByRole("heading", { name: "Target section" }),
-    ).toBeInViewport();
-
-    await preview.document.getByRole("link", { name: "Named jump" }).click();
-    await expect(preview.document.getByText("Named target", { exact: true })).toBeInViewport();
-    await expect(preview.document.locator("output").first()).toHaveText("#legacy");
-
-    await preview.document.getByRole("link", { name: "Encoded jump" }).click();
-    await expect(
-      preview.document.getByRole("heading", { name: "Encoded target" }),
-    ).toBeInViewport();
-    await expect(preview.document.locator("output").first()).toHaveText("#encoded%20section");
-    const scrollY = await body.evaluate(() => window.scrollY);
-    await preview.document.getByRole("link", { name: "Missing jump" }).click();
-    expect(await body.evaluate(() => location.hash)).toBe("#missing");
-    expect(await body.evaluate(() => window.scrollY)).toBe(scrollY);
-
-    await preview.document.getByRole("link", { name: "Empty jump" }).click();
-    await expect(preview.document.locator("output").first()).toHaveText("(empty)");
-    expect(await body.evaluate(() => window.scrollY)).toBe(0);
-
-    for (const modifier of ["Control", "Meta", "Shift", "Alt"] as const) {
-      await nested.click({ modifiers: [modifier] });
-      await expect(preview.document.locator("#modified")).toHaveText("false");
-    }
-    await nested.click({ button: "middle" });
-    await expect(preview.document.locator("#modified")).toHaveText("false");
-    expect(await body.evaluate(() => location.hash)).toBe("");
-    expect(await body.evaluate(() => window.scrollY)).toBe(0);
-  });
-
   test("runs inline scripts without allowing fetch egress", async ({ page, withWorkspace }) => {
     test.setTimeout(90_000);
     const workspace = await withWorkspace({ prefix: "file-editing-html-csp-" });

packages/app/e2e/browser/file-explorer-collapse.spec.ts

@@ -1,6 +1,4 @@
-import { writeFile } from "node:fs/promises";
-import path from "node:path";
-import { expect, test } from "../support/fixtures";
+import { test } from "../support/fixtures";
 import {
   collapseFolder,
   expandFolder,
@@ -10,7 +8,6 @@ import {
   openFileExplorer,
   openFileFromExplorer,
 } from "../support/helpers/file-explorer";
-import { delayDirectoryListingResponse } from "../support/helpers/directory-listing-gate";
 import { gotoWorkspace } from "../support/helpers/launcher";
 import { seedWorkspace, type SeededWorkspace } from "../support/helpers/seed-client";
 
@@ -52,66 +49,3 @@ test.describe("File explorer collapse", () => {
     await expectExplorerEntryVisible(page, "guide.md");
   });
 });
-
-test("Files Refresh reveals new files when a cached collapsed folder is reopened", async ({
-  page,
-}) => {
-  await gotoWorkspace(page, workspace.workspaceId);
-  await openFileExplorer(page);
-  await expandFolder(page, "docs");
-  await expectExplorerEntryVisible(page, "guide.md");
-  await collapseFolder(page, "docs");
-  await expectExplorerEntryHidden(page, "guide.md");
-  await createFileOutsidePaseo("docs/NEW.md");
-  await refreshFiles(page);
-  await expandFolder(page, "docs");
-  await expectExplorerEntryVisible(page, "guide.md");
-  await expectExplorerEntryVisible(page, "NEW.md");
-
-  await createFileOutsidePaseo("docs/NEXT.md");
-  await refreshFiles(page);
-  await expectExplorerEntryVisible(page, "NEXT.md");
-});
-
-async function refreshFiles(page: import("@playwright/test").Page): Promise<void> {
-  const refresh = page.getByRole("button", { name: "Refresh files", exact: true });
-  await refresh.click();
-  await expect(refresh).toBeEnabled();
-}
-
-async function createFileOutsidePaseo(filePath: string): Promise<void> {
-  await writeFile(path.join(workspace.repoPath, filePath), "# New output\n");
-}
-
-test("Files Refresh ignores an old reply for a folder collapsed while loading", async ({
-  page,
-}) => {
-  const reply = await delayDirectoryListingResponse(page, "docs");
-  await gotoWorkspace(page, workspace.workspaceId);
-  await openFileExplorer(page);
-  await expandFolder(page, "docs");
-  await reply.waitUntilHeld();
-  await collapseFolder(page, "docs");
-  await createFileOutsidePaseo("docs/DELAYED.md");
-  await refreshFiles(page);
-  await reply.release();
-  await expandFolder(page, "docs");
-  await expectExplorerEntryVisible(page, "DELAYED.md");
-});
-
-test("Files Refresh preserves expanded folders while their saved view is restoring", async ({
-  page,
-}) => {
-  await gotoWorkspace(page, workspace.workspaceId);
-  await openFileExplorer(page);
-  await expandFolder(page, "docs");
-  await expectExplorerEntryVisible(page, "guide.md");
-  const reply = await delayDirectoryListingResponse(page, "docs");
-  await page.reload();
-  await openFileExplorer(page);
-  await reply.waitUntilHeld();
-  await refreshFiles(page);
-  await expectExplorerEntryVisible(page, "guide.md");
-  await reply.release();
-  await expectExplorerEntryVisible(page, "guide.md");
-});

packages/app/e2e/browser/plugin-subscription-examples.spec.ts

@@ -1,79 +0,0 @@
-import { writeFile } from "node:fs/promises";
-import path from "node:path";
-import { expect, test, type Page } from "../support/fixtures";
-import { gotoAppShell } from "../support/helpers/app";
-import { openCommandCenter } from "../support/helpers/command-center";
-import { connectNewWorkspaceDaemonClient } from "../support/helpers/new-workspace";
-import { copyPluginExample } from "../support/helpers/plugin-fixture";
-import { seedWorkspace, type SeededWorkspace } from "../support/helpers/seed-client";
-import { getServerId } from "../support/helpers/server-id";
-import {
-  switchWorkspaceViaSidebar,
-  waitForWorkspaceInSidebar,
-} from "../support/helpers/workspace-ui";
-
-async function runCommand(page: Page, title: string): Promise<void> {
-  const panel = await openCommandCenter(page);
-  await panel.getByTestId("command-center-input").fill(title);
-  await panel.getByRole("button", { name: title, exact: true }).click();
-  await expect(panel).not.toBeVisible();
-}
-
-/** Installs a plugin example, opens a seeded workspace, and runs `exercise` against them. */
-async function withExample(
-  page: Page,
-  example: { name: string; id: string },
-  exercise: (workspace: SeededWorkspace) => Promise<void>,
-): Promise<void> {
-  const client = await connectNewWorkspaceDaemonClient({ ownProjects: false });
-  const previous = await client.getDaemonConfig();
-  const copy = await copyPluginExample(example.name);
-  const workspace = await seedWorkspace({ repoPrefix: `${example.name}-` });
-  try {
-    await client.patchDaemonConfig({ pluginsEnabled: true });
-    await client.installDirectoryPlugin(copy.directory);
-    await page.setViewportSize({ width: 1280, height: 800 });
-    await gotoAppShell(page);
-    const target = { serverId: getServerId(), workspaceId: workspace.workspaceId };
-    await waitForWorkspaceInSidebar(page, target);
-    await switchWorkspaceViaSidebar({ page, ...target });
-    await exercise(workspace);
-  } finally {
-    try {
-      await client.removePlugin(example.id);
-    } finally {
-      await client.patchDaemonConfig({ pluginsEnabled: previous.config.pluginsEnabled });
-      await workspace.cleanup();
-      await client.close();
-      await copy.cleanup();
-    }
-  }
-}
-
-test.describe("plugin subscriptions", () => {
-  test.describe.configure({ timeout: 180_000 });
-
-  test("a panel follows events the daemon pushes as files change", async ({ page }) => {
-    await withExample(page, { name: "diff-watch", id: "diff-watch" }, async (workspace) => {
-      await runCommand(page, "Open live changes");
-      await expect(page.getByText("No changes", { exact: true })).toBeVisible();
-
-      await writeFile(path.join(workspace.repoPath, "pushed.txt"), "from the test\n");
-      await expect(page.getByText("1 file changed", { exact: true })).toBeVisible();
-      await expect(page.getByText("pushed.txt", { exact: true })).toBeVisible();
-    });
-  });
-
-  test("a panel plays streamed bytes and sends bytes back", async ({ page }) => {
-    await withExample(page, { name: "media-stream", id: "media-stream" }, async () => {
-      await runCommand(page, "Open media stream");
-      await expect(page.getByText("live", { exact: true })).toBeVisible();
-      await expect(page.getByText(/^[1-9]\d* video frames/)).toBeVisible();
-
-      await page.getByRole("button", { name: "Send this frame back", exact: true }).click();
-      await expect(
-        page.getByText(/^The daemon received [1-9]\d* bytes, sha256 [0-9a-f]{12}$/),
-      ).toBeVisible();
-    });
-  });
-});

packages/app/e2e/browser/question-prompt-pagination.spec.ts

@@ -108,7 +108,10 @@ test.describe("Question option text", () => {
     }
   });
 
-  test("selecting description text keeps the question open", async ({ page, context }) => {
+  test("selecting description text never answers, and the label still does", async ({
+    page,
+    context,
+  }) => {
     test.setTimeout(180_000);
     const { session, card } = await openLinkedQuestions(page, context);
     try {
@@ -117,13 +120,16 @@ test.describe("Question option text", () => {
 
       await dragAcross(page, description);
       await expect.poll(() => selectedText(page)).toBe(DONE_DESCRIPTION);
+      await expectChecked(done, false);
+
       await description.getByText("Logs are at", { exact: false }).dblclick({
         position: { x: 4, y: 4 },
       });
       await expect.poll(() => selectedText(page)).toBe("Logs");
       await expectCurrentQuestion(page, { index: 1, total: 2, question: LINKED_QUESTION });
-      await expectChecked(done, true);
-      await continueToNextQuestion(page);
+      await expectChecked(done, false);
+
+      await done.getByText("Done", { exact: true }).click();
       await expectCurrentQuestion(page, { index: 2, total: 2, question: CHECKS_QUESTION });
       await openQuestion(page, { index: 1, total: 2 });
       await expectChecked(done, true);
@@ -132,17 +138,11 @@ test.describe("Question option text", () => {
     }
   });
 
-  test("double-clicking a single-select label selects once without advancing", async ({
-    page,
-    context,
-  }) => {
+  test("double-clicking a single-select label answers once", async ({ page, context }) => {
     test.setTimeout(180_000);
     const { session, card } = await openLinkedQuestions(page, context);
     try {
       await card.getByRole("radio", { name: "Done" }).getByText("Done", { exact: true }).dblclick();
-      await expectCurrentQuestion(page, { index: 1, total: 2, question: LINKED_QUESTION });
-      await expectChecked(card.getByRole("radio", { name: "Done" }), true);
-      await continueToNextQuestion(page);
 
       await expectCurrentQuestion(page, { index: 2, total: 2, question: CHECKS_QUESTION });
       await expectChecked(card.getByRole("checkbox", { name: "Lint" }), false);
@@ -154,12 +154,14 @@ test.describe("Question option text", () => {
     }
   });
 
-  test("multi-select descriptions support links and text selection", async ({ page, context }) => {
+  test("multi-select description links and selection leave the option unchanged", async ({
+    page,
+    context,
+  }) => {
     test.setTimeout(180_000);
     const { session, card } = await openLinkedQuestions(page, context);
     try {
       await card.getByRole("radio", { name: "Done" }).getByText("Done", { exact: true }).click();
-      await continueToNextQuestion(page);
       await expectCurrentQuestion(page, { index: 2, total: 2, question: CHECKS_QUESTION });
 
       const lint = card.getByRole("checkbox", { name: "Lint" });
@@ -175,16 +177,18 @@ test.describe("Question option text", () => {
 
       await dragAcross(page, description);
       await expect.poll(() => selectedText(page)).toBe(LINT_DESCRIPTION);
+      await expectChecked(lint, false);
+
       await description.getByText("Rules:", { exact: false }).dblclick({
         position: { x: 4, y: 4 },
       });
       await expect.poll(() => selectedText(page)).toBe("Rules");
-      await expectChecked(lint, true);
+      await expectChecked(lint, false);
 
       await typecheck.getByText("Typecheck", { exact: true }).dblclick();
       await expectChecked(typecheck, true);
       await lint.getByText("Lint", { exact: true }).click();
-      await expectChecked(lint, false);
+      await expectChecked(lint, true);
     } finally {
       await session.cleanup();
     }
@@ -264,7 +268,6 @@ test.describe("Question prompt pagination", () => {
       await expectQuestionHidden(page, SUCCESS_QUESTION);
 
       await chooseQuestionOption(page, "App");
-      await continueToNextQuestion(page);
       await expectCurrentQuestion(page, {
         index: 2,
         total: TOTAL_QUESTIONS,
@@ -285,7 +288,6 @@ test.describe("Question prompt pagination", () => {
         preview: 'flags.enabled("new-surface")',
       });
       await chooseQuestionOption(page, "Behind feature flag");
-      await continueToNextQuestion(page);
       await expectCurrentQuestion(page, {
         index: 3,
         total: TOTAL_QUESTIONS,
@@ -464,9 +466,7 @@ retryTest(
     retryTest.setTimeout(180_000);
     const card = page.getByTestId("question-form-card");
     await chooseQuestionOption(page, "App");
-    await continueToNextQuestion(page);
     await chooseQuestionOption(page, "Behind feature flag");
-    await continueToNextQuestion(page);
     await fillQuestionAnswer(page, {
       question: SUCCESS_QUESTION,
       answer: "Preserve my answers after a failed submission.",
@@ -520,9 +520,7 @@ retryTest(
     const card = page.getByTestId("question-form-card");
     questionClient.releaseFirstResponse();
     await chooseQuestionOption(page, "App");
-    await continueToNextQuestion(page);
     await chooseQuestionOption(page, "Behind feature flag");
-    await continueToNextQuestion(page);
     await fillQuestionAnswer(page, {
       question: SUCCESS_QUESTION,
       answer: "Start with the internal team.",

packages/app/e2e/mobile/timeline-status/placement.mjs

@@ -1,54 +0,0 @@
-// Open an existing compact chat, with the keyboard closed, then interrupt its host connection.
-// Hold that interruption externally (for example, stop a disposable SSH TCP forward).
-// Leave the daemon running. This test observes the public reconnecting state.
-// node placement.mjs <session> <platform> <coordinate-scale> <composer-label> <artifacts>
-// coordinate-scale is screenshot pixels per layout point (Android density; iOS snapshots use 1).
-import assert from "node:assert/strict";
-import { execFileSync } from "node:child_process";
-import { mkdirSync, writeFileSync } from "node:fs";
-import path from "node:path";
-
-const [, , session, platform, coordinateScale, composerLabel, artifacts] = process.argv;
-assert(
-  session && ["android", "ios"].includes(platform) && composerLabel && artifacts,
-  "Pass all five arguments",
-);
-const scale = Number(coordinateScale);
-assert(scale > 0, "Pass a positive coordinate scale");
-mkdirSync(artifacts, { recursive: true });
-function ad(...args) {
-  return execFileSync("agent-device", ["--session", session, "--platform", platform, ...args], {
-    encoding: "utf8",
-    timeout: 90_000,
-  });
-}
-function capture(name) {
-  const snapshot = ad("snapshot", "--raw", "--json");
-  writeFileSync(path.join(artifacts, `${name}.json`), snapshot);
-  ad("screenshot", path.join(artifacts, `${name}.png`));
-  const nodes = JSON.parse(snapshot).data.nodes;
-  const header = nodes.find((node) => node.identifier === "floating-header");
-  const toast = nodes.find((node) => node.identifier === "agent-reconnecting-toast");
-  assert(header?.rect && toast?.rect, "Measured header and reconnecting toast must be present");
-  assert(toast.rect.height > 0 && toast.rect.width > 0, "Toast must have visible bounds");
-  const gap = (toast.rect.y - header.rect.y - header.rect.height) / scale;
-  assert(Math.abs(gap - 12) <= 1, `${name}: actual header-to-toast gap ${gap}, expected12 points`);
-  return toast.rect.y / scale;
-}
-
-ad("wait", 'id="agent-reconnecting-toast"', "20000");
-const closedTop = capture("keyboard-closed");
-ad("press", `label=${JSON.stringify(composerLabel)}`);
-const opened = ad("snapshot", "--raw", "--json");
-const nodes = JSON.parse(opened).data.nodes;
-assert(
-  nodes.some(
-    (node) => /keyboard/i.test(node.type ?? "") || /inputmethod/i.test(node.bundleId ?? ""),
-  ),
-  "Use a real visible software keyboard; open Android sessions with --no-test-ime",
-);
-const openTop = capture("keyboard-open");
-assert(Math.abs(openTop - closedTop) <= 1, "Keyboard must not translate the toast");
-console.log(
-  "PASS: rendered reconnecting toast stays12 points below measured header, keyboard closed/open",
-);

packages/app/e2e/support/helpers/directory-listing-gate.ts

@@ -1,44 +0,0 @@
-import type { Page, WebSocketRoute } from "@playwright/test";
-import { daemonWsRoutePattern } from "./daemon-port";
-
-function listingResponse(message: string | Buffer) {
-  try {
-    const envelope = JSON.parse(message.toString());
-    return envelope.message?.type === "file_explorer_response" ? envelope.message.payload : null;
-  } catch {
-    return null;
-  }
-}
-
-/** Delay one real directory response after the daemon has read its contents. */
-export async function delayDirectoryListingResponse(page: Page, directoryPath: string) {
-  let held: { socket: WebSocketRoute; message: string | Buffer; requestId: string } | null = null;
-  let resolveHeld!: () => void;
-  const heldResponse = new Promise<void>((resolve) => {
-    resolveHeld = resolve;
-  });
-  await page.routeWebSocket(daemonWsRoutePattern(), (socket) => {
-    const server = socket.connectToServer();
-    socket.onMessage((message) => server.send(message));
-    server.onMessage((message) => {
-      const response = listingResponse(message);
-      if (!held && response?.mode === "list" && response.path === directoryPath) {
-        held = { socket, message, requestId: response.requestId };
-        resolveHeld();
-        return;
-      }
-      socket.send(message);
-    });
-  });
-  return {
-    waitUntilHeld: () => heldResponse,
-    release: async () => {
-      if (!held) throw new Error("Directory response must be held before release");
-      held.socket.send(held.message);
-      // Let the delivered reply update the UI before the user reopens the folder.
-      await page.evaluate(
-        () => new Promise<void>((resolve) => requestAnimationFrame(() => resolve())),
-      );
-    },
-  };
-}

packages/app/src/agent-stream/presentation.test.ts

@@ -69,9 +69,6 @@ function installProbe(
     queryClient: new QueryClient(),
     paseo: {} as InstalledPlugin["paseo"],
     invoke: async () => undefined,
-    subscribe: () => {
-      throw new Error("Subscriptions are not used here");
-    },
   };
 }
 

packages/app/src/components/file-explorer-pane.tsx

@@ -64,15 +64,10 @@ import { useIsLocalDaemon } from "@/hooks/use-is-local-daemon";
 import { buildWorkspaceExplorerStateKey } from "@/hooks/use-file-explorer-actions";
 import { usePanelStore, type ExpandedPathsUpdate, type SortOption } from "@/stores/panel-store";
 import { buildAbsoluteExplorerPath } from "@/utils/explorer-paths";
-import {
-  DIRECTORY_LISTING_INVALIDATED,
-  type DirectoryListingResult,
-} from "@/file-explorer/directory-listings";
 import { isHiddenExplorerPath } from "@/file-explorer/visibility";
 import {
   flattenExplorerTree,
   reconcileRestoredExpandedPaths,
-  refreshExplorerDirectories,
   restoreExpandedDirectories,
   setExpandedDirectoryPath,
   showHiddenFilesAndRestoreExpandedDirectories,
@@ -440,7 +435,6 @@ export function FileExplorerPane({
 
   const {
     requestDirectoryListing,
-    invalidateDirectoryListings,
     createEntry,
     renameEntry,
     duplicateEntry,
@@ -892,11 +886,9 @@ export function FileExplorerPane({
       refreshExplorerDirectories({
         hasWorkspaceScope,
         expandedPaths,
-        showHiddenFiles: usePanelStore.getState().explorerShowHiddenFiles,
         requestDirectoryListing,
-        invalidateDirectoryListings,
       }),
-    [invalidateDirectoryListings, expandedPaths, hasWorkspaceScope, requestDirectoryListing],
+    [expandedPaths, hasWorkspaceScope, requestDirectoryListing],
   );
   const { refetch: refetchExplorer, isFetching: isRefreshFetching } = useQuery({
     queryKey: ["fileExplorerRefresh", serverId, workspaceStateKey],
@@ -1432,7 +1424,7 @@ function toggleDirectory({
   requestDirectoryListing: (
     path: string,
     opts?: { recordHistory?: boolean; setCurrentPath?: boolean },
-  ) => Promise<DirectoryListingResult>;
+  ) => Promise<ExplorerDirectory | null>;
   setExpandedPathsForWorkspace: (workspaceStateKey: string, paths: ExpandedPathsUpdate) => void;
 }): void {
   if (!workspaceStateKey) {
@@ -1569,7 +1561,7 @@ async function initializeExplorer({
   requestDirectoryListing: (
     path: string,
     opts?: { recordHistory?: boolean; setCurrentPath?: boolean },
-  ) => Promise<DirectoryListingResult>;
+  ) => Promise<ExplorerDirectory | null>;
   setExpandedPathsForWorkspace: (workspaceStateKey: string, paths: ExpandedPathsUpdate) => void;
 }): Promise<void> {
   if (!hasWorkspaceScope || hasInitializedRef.current) {
@@ -1580,7 +1572,6 @@ async function initializeExplorer({
     recordHistory: false,
     setCurrentPath: false,
   });
-  if (rootDirectory === DIRECTORY_LISTING_INVALIDATED) return;
   if (!rootDirectory) {
     hasInitializedRef.current = false;
     return;
@@ -1602,10 +1593,7 @@ async function initializeExplorer({
   const hiddenPersistedPaths = showHiddenFiles
     ? []
     : Array.from(persistedExpandedPaths).filter(isHiddenExplorerPath);
-  const restoredPathsWithHidden =
-    restoredPaths === DIRECTORY_LISTING_INVALIDATED
-      ? DIRECTORY_LISTING_INVALIDATED
-      : [...restoredPaths, ...hiddenPersistedPaths];
+  const restoredPathsWithHidden = [...restoredPaths, ...hiddenPersistedPaths];
   setExpandedPathsForWorkspace(workspaceStateKey, (currentPaths) =>
     reconcileRestoredExpandedPaths({
       persistedExpandedPaths,
@@ -1615,6 +1603,39 @@ async function initializeExplorer({
   );
 }
 
+async function refreshExplorerDirectories({
+  hasWorkspaceScope,
+  expandedPaths,
+  requestDirectoryListing,
+}: {
+  hasWorkspaceScope: boolean;
+  expandedPaths: Set<string>;
+  requestDirectoryListing: (
+    path: string,
+    opts?: { recordHistory?: boolean; setCurrentPath?: boolean },
+  ) => Promise<ExplorerDirectory | null>;
+}): Promise<null> {
+  if (!hasWorkspaceScope) {
+    return null;
+  }
+  const showHiddenFiles = usePanelStore.getState().explorerShowHiddenFiles;
+  const directoryPaths = Array.from(expandedPaths).filter(
+    (path) => showHiddenFiles || !isHiddenExplorerPath(path),
+  );
+  if (!directoryPaths.includes(".")) {
+    directoryPaths.unshift(".");
+  }
+  await Promise.all(
+    directoryPaths.map((path) =>
+      requestDirectoryListing(path, {
+        recordHistory: false,
+        setCurrentPath: false,
+      }),
+    ),
+  );
+  return null;
+}
+
 function getErrorRecoveryPath(state: AgentFileExplorerState | undefined): string {
   if (!state) {
     return ".";

packages/app/src/components/floating-header/index.tsx

@@ -1,26 +1,8 @@
-import { createContext, useContext, useCallback, useState, useId, type ReactNode } from "react";
+import { useCallback, useState, type ReactNode } from "react";
 import { View, type LayoutChangeEvent } from "react-native";
 import { StyleSheet } from "react-native-unistyles";
-import { Portal, PortalHost } from "@gorhom/portal";
-import { useRetainedPanelActive } from "@/components/retained-panel";
 import { EdgeFade } from "@/components/ui/edge-fade";
 
-const OverlayClearanceContext = createContext(0);
-const OverlayHostContext = createContext<string | null>(null);
-
-/** Keep status overlays above the fade without lifting the scrolling content. */
-export function FloatingHeaderOverlay({ children }: { children: ReactNode }) {
-  const hostName = useContext(OverlayHostContext);
-  const active = useRetainedPanelActive();
-  if (!hostName) return children;
-  return active ? <Portal hostName={hostName}>{children}</Portal> : null;
-}
-
-/** Clearance for stationary overlays inside this layout, measured to the header bottom. */
-export function useFloatingHeaderOverlayClearance(): number {
-  return useContext(OverlayClearanceContext);
-}
-
 /** How far the fade runs past the header's bottom edge, so content dissolves instead of clipping. */
 const FADE_OVERHANG = 48;
 /** Share of the fade that stays opaque, so the header's title and icons keep contrast. */
@@ -38,24 +20,15 @@ interface FloatingHeaderLayoutProps {
  * same tree position in both modes, so toggling `floating` never remounts them.
  */
 export function FloatingHeaderLayout({ header, floating, children }: FloatingHeaderLayoutProps) {
-  const overlayHostName = useId();
   const [headerHeight, setHeaderHeight] = useState(0);
   const handleHeaderLayout = useCallback((event: LayoutChangeEvent) => {
     setHeaderHeight(Math.round(event.nativeEvent.layout.height));
   }, []);
 
-  const overlayClearance = floating ? headerHeight : 0;
-
   return (
     <View style={styles.root}>
       {floating ? null : header}
-      <View style={styles.content}>
-        <OverlayClearanceContext.Provider value={overlayClearance}>
-          <OverlayHostContext.Provider value={floating ? overlayHostName : null}>
-            {children(floating ? headerHeight : 0)}
-          </OverlayHostContext.Provider>
-        </OverlayClearanceContext.Provider>
-      </View>
+      <View style={styles.content}>{children(floating ? headerHeight : 0)}</View>
       {floating ? (
         <View style={styles.overlay} pointerEvents="box-none">
           <EdgeFade
@@ -64,19 +37,11 @@ export function FloatingHeaderLayout({ header, floating, children }: FloatingHea
             backdrop="surfaceWorkspace"
             style={styles.fade}
           />
-          <View
-            testID="floating-header"
-            collapsable={false}
-            pointerEvents="box-none"
-            onLayout={handleHeaderLayout}
-          >
+          <View pointerEvents="box-none" onLayout={handleHeaderLayout}>
             {header}
           </View>
         </View>
       ) : null}
-      <View style={styles.statusOverlay} pointerEvents="box-none">
-        <PortalHost name={overlayHostName} />
-      </View>
     </View>
   );
 }
@@ -90,13 +55,6 @@ const styles = StyleSheet.create({
     flex: 1,
     minHeight: 0,
   },
-  statusOverlay: {
-    position: "absolute",
-    top: 0,
-    bottom: 0,
-    left: 0,
-    right: 0,
-  },
   overlay: {
     position: "absolute",
     top: 0,

packages/app/src/components/question-form-card.browser.test.tsx

@@ -39,10 +39,7 @@ afterEach(() => {
   }
 });
 
-function buildPermission(
-  question: Record<string, unknown>,
-  following: Record<string, unknown>[] = [],
-): PendingPermission {
+function buildPermission(question: Record<string, unknown>): PendingPermission {
   return {
     key: "perm-1",
     agentId: "agent-1",
@@ -51,12 +48,12 @@ function buildPermission(
       provider: "claude",
       name: "AskUserQuestion",
       kind: "question",
-      input: { questions: [question, ...following] },
+      input: { questions: [question] },
     },
   };
 }
 
-function mountCard(question: Record<string, unknown>, following: Record<string, unknown>[] = []) {
+function mountCard(question: Record<string, unknown>) {
   const onRespond = vi.fn<(response: AgentPermissionResponse) => void>();
   const container = document.createElement("div");
   document.body.appendChild(container);
@@ -64,7 +61,7 @@ function mountCard(question: Record<string, unknown>, following: Record<string,
   act(() =>
     root.render(
       <QuestionFormCard
-        permission={buildPermission(question, following)}
+        permission={buildPermission(question)}
         onRespond={onRespond}
         isResponding={false}
         responseError={null}
@@ -74,14 +71,10 @@ function mountCard(question: Record<string, unknown>, following: Record<string,
   mounted.push({ root, container });
 
   const view = within(container);
+  const optionRole = question.multiSelect ? "checkbox" : "radio";
   const otherInput = () =>
     view.getByRole<HTMLInputElement>("textbox", { name: String(question.question) });
-  const check = (label: string) =>
-    act(() =>
-      (
-        view.queryByRole("radio", { name: label }) ?? view.getByRole("checkbox", { name: label })
-      ).click(),
-    );
+  const check = (label: string) => act(() => view.getByRole(optionRole, { name: label }).click());
   const type = (text: string) => {
     const input = otherInput();
     const valueSetter = Object.getOwnPropertyDescriptor(HTMLInputElement.prototype, "value")?.set;
@@ -97,7 +90,7 @@ function mountCard(question: Record<string, unknown>, following: Record<string,
     if (!response || response.behavior !== "allow") throw new Error("card did not submit");
     return (response.updatedInput as { answers: Record<string, string> }).answers;
   };
-  return { check, type, otherInput, submit, submittedAnswers, view, onRespond };
+  return { check, type, otherInput, submit, submittedAnswers };
 }
 
 const multiSelectQuestion = {
@@ -161,31 +154,3 @@ describe("QuestionFormCard other answers", () => {
     expect(card.submittedAnswers()).toEqual({ Provider: "Codex" });
   });
 });
-
-describe("QuestionFormCard explicit navigation", () => {
-  it("keeps a picked option visible until Next and submits only on Submit", () => {
-    const card = mountCard(singleSelectQuestion, [multiSelectQuestion]);
-    card.check("Codex");
-    expect(card.view.getByRole("radio", { name: "Codex" }).getAttribute("aria-checked")).toBe(
-      "true",
-    );
-    expect(card.onRespond).not.toHaveBeenCalled();
-    act(() => card.view.getByRole("button", { name: "Next" }).click());
-    card.check("Apple");
-    expect(card.onRespond).not.toHaveBeenCalled();
-    card.submit();
-    expect(card.submittedAnswers()).toEqual({ Provider: "Codex", Fruits: "Apple" });
-  });
-
-  it("selects an option when its description is clicked without submitting", () => {
-    const card = mountCard({
-      ...singleSelectQuestion,
-      options: [{ label: "Codex", description: "Use this provider" }],
-    });
-    act(() => card.view.getByText("Use this provider").click());
-    expect(card.view.getByRole("radio", { name: "Codex" }).getAttribute("aria-checked")).toBe(
-      "true",
-    );
-    expect(card.onRespond).not.toHaveBeenCalled();
-  });
-});

packages/app/src/components/question-form-card.tsx

@@ -43,16 +43,20 @@ const IS_WEB = isWeb;
 // drag to the preview's nested scroll, so lines past a cap would be unreachable.
 const OPTION_PREVIEW_MAX_HEIGHT = IS_WEB ? 240 : undefined;
 
-// Links are independent actions. Ignore repeat clicks so double-click text selection
-// toggles an option only once.
-function isNonAnsweringClick(event: GestureResponderEvent): boolean {
+// Web clicks that must not answer: a link in the label, anything in the description, and the
+// second click of a double-click, which would otherwise land on the next question's options.
+function isNonAnsweringClick(event: GestureResponderEvent, description: View | null): boolean {
   if (!IS_WEB) return false;
   const click = event.nativeEvent as unknown as MouseEvent;
   if (click.detail > 1) return true;
   const target = click.target as Element | null;
-  return Boolean(target?.closest("a"));
+  const descriptionElement = description as unknown as Element | null;
+  return Boolean(target?.closest("a") || descriptionElement?.contains(target));
 }
 
+// The description is selectable text, so a touch on it belongs to the text, not the option.
+const claimResponder = () => true;
+
 function getQuestionInputPlaceholder({
   question,
   answerPlaceholder,
@@ -88,9 +92,10 @@ function QuestionOptionRow({
 }: QuestionOptionRowProps) {
   const { theme } = useUnistyles();
 
+  const descriptionRef = useRef<View>(null);
   const handlePress = useCallback(
     (event: GestureResponderEvent) => {
-      if (isNonAnsweringClick(event)) return;
+      if (isNonAnsweringClick(event, descriptionRef.current)) return;
       onToggle(qIndex, optIndex, multiSelect);
     },
     [onToggle, qIndex, optIndex, multiSelect],
@@ -147,7 +152,11 @@ function QuestionOptionRow({
         <View style={styles.optionTextBlock}>
           <MarkdownRenderer text={option.label} compact muted={!isSelected} selectable={false} />
           {option.description ? (
-            <View testID="question-form-option-description">
+            <View
+              ref={descriptionRef}
+              testID="question-form-option-description"
+              onStartShouldSetResponder={claimResponder}
+            >
               <MarkdownRenderer text={option.description} compact muted />
             </View>
           ) : null}
@@ -404,8 +413,12 @@ export function QuestionFormCard({
         });
         otherInputRef.current?.replaceText("");
       }
+
+      if (!multiSelect && next.size > 0 && qIndex === activeQuestionIndex && questions) {
+        setActiveQuestionIndex(Math.min(qIndex + 1, questions.length - 1));
+      }
     },
-    [otherTexts, selections],
+    [activeQuestionIndex, otherTexts, questions, selections],
   );
 
   const setOtherText = useCallback(

packages/app/src/components/toast-host.tsx

@@ -1,7 +1,3 @@
-import {
-  FloatingHeaderOverlay,
-  useFloatingHeaderOverlayClearance,
-} from "@/components/floating-header";
 import { useCallback, useEffect, useMemo, useRef, useState, type ReactNode } from "react";
 import { createPortal } from "react-dom";
 import {
@@ -194,11 +190,9 @@ export function ToastViewport({
     return clearTimer;
   }, [clearTimer, durationMs, scheduleDismiss, toastId]);
 
-  const panelClearance = useFloatingHeaderOverlayClearance();
   const headerHeight = isMobile ? HEADER_INNER_HEIGHT_MOBILE : HEADER_INNER_HEIGHT;
   const headerTopPadding = isMobile ? HEADER_TOP_PADDING_MOBILE : 0;
-  const topOffset =
-    placement === "app-shell" ? insets.top + headerTopPadding + headerHeight : panelClearance;
+  const topOffset = placement === "app-shell" ? insets.top + headerTopPadding + headerHeight : 0;
 
   const content = (
     <View
@@ -226,7 +220,7 @@ export function ToastViewport({
     return createPortal(content, getOverlayRoot());
   }
 
-  return placement === "panel" ? <FloatingHeaderOverlay>{content}</FloatingHeaderOverlay> : content;
+  return content;
 }
 
 function ToastCard({

packages/app/src/composer/dock/index.native.tsx

@@ -179,8 +179,8 @@ export function ComposerDock({
               <View style={fullscreen ? fullscreenSurface : dockStyles.composer}>{composer}</View>
             </ComposerViewportContent>
           </ComposerViewportContent>
+          {overlay}
         </KeyboardTranslateView>
-        {overlay}
       </ComposerViewport>
     );
   }
@@ -214,8 +214,8 @@ export function ComposerDock({
             {composer}
           </View>
         </ComposerViewportContent>
+        {overlay}
       </KeyboardTranslateView>
-      {overlay}
     </ComposerViewport>
   );
 }

packages/app/src/composer/dock/overlay-placement.test.ts

@@ -1,31 +0,0 @@
-import { readFileSync } from "node:fs";
-import ts from "typescript";
-import { expect, it } from "vitest";
-
-it("keeps dock overlays outside keyboard translation in every native layout", () => {
-  const source = ts.createSourceFile(
-    "dock.tsx",
-    readFileSync(new URL("./index.native.tsx", import.meta.url), "utf8"),
-    ts.ScriptTarget.Latest,
-    true,
-    ts.ScriptKind.TSX,
-  );
-  let surfaces = 0;
-  function visit(node: ts.Node) {
-    if (
-      ts.isJsxElement(node) &&
-      node.openingElement.tagName.getText(source) === "KeyboardTranslateView"
-    ) {
-      surfaces++;
-      function assertStationary(child: ts.Node) {
-        if (ts.isJsxExpression(child))
-          expect(child.expression?.getText(source)).not.toBe("overlay");
-        ts.forEachChild(child, assertStationary);
-      }
-      ts.forEachChild(node, assertStationary);
-    }
-    ts.forEachChild(node, visit);
-  }
-  visit(source);
-  expect(surfaces).toBe(2);
-});

packages/app/src/file-explorer/directory-listings.test.ts

@@ -1,180 +0,0 @@
-import { describe, expect, it } from "vitest";
-import type { AgentFileExplorerState, ExplorerDirectory } from "@/stores/session-store";
-import {
-  refreshExplorerDirectories,
-  restoreExpandedDirectories,
-  showHiddenFilesAndRestoreExpandedDirectories,
-  reconcileRestoredExpandedPaths,
-} from "./tree";
-import {
-  DIRECTORY_LISTING_INVALIDATED,
-  invalidateExplorerDirectoryListings,
-  loadExplorerDirectory,
-} from "./directory-listings";
-
-function directory(path: string, names: string[]): ExplorerDirectory {
-  return {
-    path,
-    entries: names.map((name) => ({
-      path: `${path}/${name}`,
-      name,
-      kind: "file",
-      size: 0,
-      modifiedAt: "2026-01-01T00:00:00.000Z",
-    })),
-  };
-}
-
-function deferred<T>() {
-  let resolve!: (value: T) => void;
-  let reject!: (error: Error) => void;
-  const promise = new Promise<T>((yes, no) => {
-    resolve = yes;
-    reject = no;
-  });
-  return { promise, resolve, reject };
-}
-
-function explorer() {
-  let state: AgentFileExplorerState = {
-    listingGeneration: 0,
-    directories: new Map([[".", directory(".", ["docs"])]]),
-    isLoading: false,
-    lastError: null,
-    pendingRequest: null,
-    currentPath: ".",
-    history: ["."],
-    lastVisitedPath: ".",
-    selectedEntryPath: null,
-  };
-  const updateState = (update: (state: AgentFileExplorerState) => AgentFileExplorerState) => {
-    state = update(state);
-    return state;
-  };
-  return {
-    state: () => state,
-    load: (path: string, result: Promise<ExplorerDirectory>) =>
-      loadExplorerDirectory({
-        path,
-        updateState,
-        listDirectory: () => result,
-        fallbackError: "Directory listing failed",
-        options: { recordHistory: false, setCurrentPath: false },
-      }),
-    refresh: (retained: readonly string[]) =>
-      invalidateExplorerDirectoryListings(updateState, retained),
-  };
-}
-
-describe("directory listings across Refresh", () => {
-  it("ignores an earlier reply for a collapsed folder while the fresh root is loading", async () => {
-    const model = explorer();
-    const old = deferred<ExplorerDirectory>();
-    const oldLoad = model.load("docs", old.promise);
-    model.refresh(["."]);
-    const fresh = deferred<ExplorerDirectory>();
-    const freshLoad = model.load(".", fresh.promise);
-    old.resolve(directory("docs", ["OLD.md"]));
-
-    await expect(oldLoad).resolves.toBe(DIRECTORY_LISTING_INVALIDATED);
-    expect(model.state().directories.has("docs")).toBe(false);
-    expect(model.state().isLoading).toBe(true);
-    expect(model.state().pendingRequest?.path).toBe(".");
-
-    const root = directory(".", ["docs", "NEW.md"]);
-    fresh.resolve(root);
-    await expect(freshLoad).resolves.toEqual(root);
-    expect(model.state().directories.get(".")).toEqual(root);
-    expect(model.state().isLoading).toBe(false);
-  });
-
-  it("does not let a pre-refresh reply replace a newer listing of the same folder", async () => {
-    const model = explorer();
-    const old = deferred<ExplorerDirectory>();
-    const oldLoad = model.load("docs", old.promise);
-    model.refresh([".", "docs"]);
-    const fresh = directory("docs", ["NEW.md"]);
-    await model.load("docs", Promise.resolve(fresh));
-    old.resolve(directory("docs", ["OLD.md"]));
-
-    await expect(oldLoad).resolves.toBe(DIRECTORY_LISTING_INVALIDATED);
-    expect(model.state().directories.get("docs")).toEqual(fresh);
-  });
-
-  it("ignores errors from requests invalidated by Refresh", async () => {
-    const model = explorer();
-    const old = deferred<ExplorerDirectory>();
-    const oldLoad = model.load("docs", old.promise);
-    model.refresh(["."]);
-    const fresh = deferred<ExplorerDirectory>();
-    const freshLoad = model.load(".", fresh.promise);
-    old.reject(new Error("Old request failed"));
-
-    await expect(oldLoad).resolves.toBe(DIRECTORY_LISTING_INVALIDATED);
-    expect(model.state().lastError).toBeNull();
-    expect(model.state().isLoading).toBe(true);
-    fresh.resolve(directory(".", ["docs"]));
-    await freshLoad;
-  });
-
-  it("keeps expanded listings visible and discards collapsed listings on Refresh", async () => {
-    const model = explorer();
-    const docs = directory("docs", ["guide.md"]);
-    await model.load("docs", Promise.resolve(docs));
-    await model.load("assets", Promise.resolve(directory("assets", ["logo.png"])));
-    model.refresh([".", "docs"]);
-
-    expect(model.state().directories.get("docs")).toEqual(docs);
-    expect(model.state().directories.has("assets")).toBe(false);
-    expect(model.state().currentPath).toBe(".");
-    expect(model.state().history).toEqual(["."]);
-  });
-});
-
-it.each([false, true])(
-  "preserves saved expansion when Refresh invalidates restoration (hidden: %s)",
-  async (hidden) => {
-    const model = explorer();
-    const folder = hidden ? ".hidden" : "docs";
-    const root: ExplorerDirectory = {
-      path: ".",
-      entries: [
-        {
-          name: folder,
-          path: folder,
-          kind: "directory",
-          size: 0,
-          modifiedAt: "2026-01-01T00:00:00.000Z",
-        },
-      ],
-    };
-    const saved = new Set([".", folder]);
-    const pending = deferred<ExplorerDirectory>();
-    const input = {
-      rootDirectory: root,
-      persistedExpandedPaths: saved,
-      requestDirectoryListing: (path: string) => model.load(path, pending.promise),
-    };
-    const restoration = hidden
-      ? showHiddenFilesAndRestoreExpandedDirectories({ ...input, showHiddenFiles: () => {} })
-      : restoreExpandedDirectories({ ...input, showHiddenFiles: false });
-    const fresh = directory(folder, ["NEW.md"]);
-    await refreshExplorerDirectories({
-      hasWorkspaceScope: true,
-      expandedPaths: saved,
-      showHiddenFiles: hidden,
-      invalidateDirectoryListings: model.refresh,
-      requestDirectoryListing: (path) =>
-        model.load(path, Promise.resolve(path === "." ? root : fresh)),
-    });
-    pending.resolve(directory(folder, ["OLD.md"]));
-    const restored = await restoration;
-    const expanded = reconcileRestoredExpandedPaths({
-      persistedExpandedPaths: saved,
-      currentExpandedPaths: saved,
-      restoredExpandedPaths: restored,
-    });
-    expect(model.state().directories.get(folder)).toEqual(fresh);
-    expect(expanded).toEqual([".", folder]);
-  },
-);

packages/app/src/file-explorer/directory-listings.ts

@@ -1,91 +0,0 @@
-import type { AgentFileExplorerState, ExplorerDirectory } from "@/stores/session-store";
-
-export const DIRECTORY_LISTING_INVALIDATED = Symbol("directory-listing-invalidated");
-export type DirectoryListingResult =
-  | ExplorerDirectory
-  | null
-  | typeof DIRECTORY_LISTING_INVALIDATED;
-
-type UpdateExplorerState = (
-  update: (state: AgentFileExplorerState) => AgentFileExplorerState,
-) => AgentFileExplorerState | null;
-
-function pushHistory(history: string[], path: string): string[] {
-  const normalizedHistory = history.length === 0 ? ["."] : history;
-  return normalizedHistory.at(-1) === path ? normalizedHistory : [...normalizedHistory, path];
-}
-
-export async function loadExplorerDirectory({
-  path,
-  options,
-  updateState,
-  listDirectory,
-  fallbackError,
-}: {
-  path: string;
-  options?: { recordHistory?: boolean; setCurrentPath?: boolean };
-  updateState: UpdateExplorerState;
-  listDirectory: () => Promise<ExplorerDirectory>;
-  fallbackError: string;
-}): Promise<DirectoryListingResult> {
-  const shouldSetCurrentPath = options?.setCurrentPath ?? true;
-  const shouldRecordHistory = options?.recordHistory ?? shouldSetCurrentPath;
-  const initial = updateState((state) => ({
-    ...state,
-    isLoading: true,
-    lastError: null,
-    pendingRequest: { path, mode: "list" },
-    ...(shouldSetCurrentPath
-      ? {
-          currentPath: path,
-          history: shouldRecordHistory ? pushHistory(state.history, path) : state.history,
-          lastVisitedPath: path,
-        }
-      : {}),
-  }));
-  if (!initial) return null;
-  const generation = initial.listingGeneration;
-  try {
-    const directory = await listDirectory();
-    let accepted = false;
-    updateState((state) => {
-      if (state.listingGeneration !== generation) return state;
-      accepted = true;
-      const directories = new Map(state.directories);
-      directories.set(directory.path, directory);
-      return { ...state, directories, isLoading: false, lastError: null, pendingRequest: null };
-    });
-    return accepted ? directory : DIRECTORY_LISTING_INVALIDATED;
-  } catch (error) {
-    let invalidated = false;
-    updateState((state) => {
-      if (state.listingGeneration !== generation) {
-        invalidated = true;
-        return state;
-      }
-      return {
-        ...state,
-        isLoading: false,
-        lastError: error instanceof Error ? error.message : fallbackError,
-        pendingRequest: null,
-      };
-    });
-    return invalidated ? DIRECTORY_LISTING_INVALIDATED : null;
-  }
-}
-
-/** Refresh invalidates earlier replies as well as listings of collapsed folders. */
-export function invalidateExplorerDirectoryListings(
-  updateState: UpdateExplorerState,
-  retainedPaths: readonly string[],
-): void {
-  const retained = new Set(retainedPaths);
-  updateState((state) => ({
-    ...state,
-    listingGeneration: state.listingGeneration + 1,
-    directories: new Map([...state.directories].filter(([path]) => retained.has(path))),
-    isLoading: false,
-    pendingRequest: null,
-    lastError: null,
-  }));
-}

packages/app/src/file-explorer/tree.test.ts

@@ -1,10 +1,9 @@
 import { describe, expect, it } from "vitest";
-import type { ExplorerDirectory, ExplorerEntry } from "@/stores/session-store";
+import type { ExplorerEntry } from "@/stores/session-store";
 import {
   MAX_AUTO_EXPANDED_DIRECTORY_DEPTH,
   flattenExplorerTree,
   reconcileRestoredExpandedPaths,
-  refreshExplorerDirectories,
   restoreExpandedDirectories,
   setExpandedDirectoryPath,
   showHiddenFilesAndRestoreExpandedDirectories,
@@ -196,76 +195,3 @@ describe("file explorer tree", () => {
     await expect(restoration).resolves.toEqual([".", ".hidden"]);
   });
 });
-
-describe("refreshExplorerDirectories", () => {
-  function makeListing(path: string, names: string[]): ExplorerDirectory {
-    return {
-      path,
-      entries: names.map((name) => ({
-        name,
-        path: path === "." ? name : `${path}/${name}`,
-        kind: "file" as const,
-        size: 0,
-        modifiedAt: "2026-01-01T00:00:00.000Z",
-      })),
-    };
-  }
-
-  function createExplorer(disk: Record<string, string[]>, cached: Record<string, string[]>) {
-    const directories = new Map(
-      Object.entries(cached).map(([path, names]) => [path, makeListing(path, names)]),
-    );
-    return {
-      directories,
-      requestDirectoryListing: async (path: string) => {
-        const listing = makeListing(path, disk[path] ?? []);
-        directories.set(path, listing);
-        return listing;
-      },
-      invalidateDirectoryListings: (retainedPaths: readonly string[]) => {
-        const retained = new Set(retainedPaths);
-        for (const path of directories.keys()) if (!retained.has(path)) directories.delete(path);
-      },
-    };
-  }
-
-  it("does not keep a stale listing for a collapsed folder", async () => {
-    // docs/ was expanded and collapsed earlier, then NEW.md was created in it.
-    const explorer = createExplorer(
-      { ".": ["docs"], docs: ["README.md", "NEW.md"] },
-      { ".": ["docs"], docs: ["README.md"] },
-    );
-
-    await refreshExplorerDirectories({
-      hasWorkspaceScope: true,
-      expandedPaths: new Set(["."]),
-      showHiddenFiles: false,
-      requestDirectoryListing: explorer.requestDirectoryListing,
-      invalidateDirectoryListings: explorer.invalidateDirectoryListings,
-    });
-
-    // Expanding docs/ only fetches when it has no cached listing.
-    expect(explorer.directories.has("docs")).toBe(false);
-    expect(explorer.directories.get(".")?.entries.map((entry) => entry.name)).toEqual(["docs"]);
-  });
-
-  it("refreshes expanded folders and keeps their listings", async () => {
-    const explorer = createExplorer(
-      { ".": ["src"], src: ["a.ts", "b.ts"] },
-      { ".": ["src"], src: ["a.ts"] },
-    );
-
-    await refreshExplorerDirectories({
-      hasWorkspaceScope: true,
-      expandedPaths: new Set([".", "src"]),
-      showHiddenFiles: false,
-      requestDirectoryListing: explorer.requestDirectoryListing,
-      invalidateDirectoryListings: explorer.invalidateDirectoryListings,
-    });
-
-    expect(explorer.directories.get("src")?.entries.map((entry) => entry.name)).toEqual([
-      "a.ts",
-      "b.ts",
-    ]);
-  });
-});

packages/app/src/file-explorer/tree.ts

@@ -1,6 +1,5 @@
 import type { ExplorerDirectory, ExplorerEntry } from "@/stores/session-store";
 import type { SortOption } from "@/stores/panel-store/state";
-import { DIRECTORY_LISTING_INVALIDATED, type DirectoryListingResult } from "./directory-listings";
 import { filterVisibleExplorerEntries, isHiddenExplorerPath } from "./visibility";
 
 export const MAX_AUTO_EXPANDED_DIRECTORY_DEPTH = 5;
@@ -17,13 +16,11 @@ interface FlattenExplorerTreeInput {
   showHiddenFiles: boolean;
 }
 
-type RestoredExpandedPaths = string[] | typeof DIRECTORY_LISTING_INVALIDATED;
-
 interface RestoreExpandedDirectoriesInput {
   rootDirectory: ExplorerDirectory;
   persistedExpandedPaths: ReadonlySet<string>;
   showHiddenFiles: boolean;
-  requestDirectoryListing: (path: string) => Promise<DirectoryListingResult>;
+  requestDirectoryListing: (path: string) => Promise<ExplorerDirectory | null>;
 }
 
 interface ShowHiddenFilesAndRestoreExpandedDirectoriesInput extends Omit<
@@ -36,7 +33,7 @@ interface ShowHiddenFilesAndRestoreExpandedDirectoriesInput extends Omit<
 interface ReconcileRestoredExpandedPathsInput {
   persistedExpandedPaths: ReadonlySet<string>;
   currentExpandedPaths: ReadonlySet<string>;
-  restoredExpandedPaths: RestoredExpandedPaths;
+  restoredExpandedPaths: string[];
 }
 
 interface SetExpandedDirectoryPathInput {
@@ -88,7 +85,7 @@ export async function restoreExpandedDirectories({
   persistedExpandedPaths,
   showHiddenFiles,
   requestDirectoryListing,
-}: RestoreExpandedDirectoriesInput): Promise<RestoredExpandedPaths> {
+}: RestoreExpandedDirectoriesInput): Promise<string[]> {
   const restoredPaths = ["."];
   const restoredPathSet = new Set(restoredPaths);
   let parentDirectories = [rootDirectory];
@@ -113,11 +110,9 @@ export async function restoreExpandedDirectories({
     const requestedDirectories = await Promise.all(
       pathsToRequest.map((path) => requestDirectoryListing(path)),
     );
-    if (requestedDirectories.includes(DIRECTORY_LISTING_INVALIDATED))
-      return DIRECTORY_LISTING_INVALIDATED;
     parentDirectories = [];
     for (const directory of requestedDirectories) {
-      if (!directory || directory === DIRECTORY_LISTING_INVALIDATED) {
+      if (!directory) {
         continue;
       }
       restoredPaths.push(directory.path);
@@ -133,7 +128,7 @@ export function showHiddenFilesAndRestoreExpandedDirectories({
   persistedExpandedPaths,
   showHiddenFiles,
   requestDirectoryListing,
-}: ShowHiddenFilesAndRestoreExpandedDirectoriesInput): Promise<RestoredExpandedPaths> {
+}: ShowHiddenFilesAndRestoreExpandedDirectoriesInput): Promise<string[]> {
   showHiddenFiles();
   return restoreExpandedDirectories({
     rootDirectory,
@@ -148,8 +143,6 @@ export function reconcileRestoredExpandedPaths({
   currentExpandedPaths,
   restoredExpandedPaths,
 }: ReconcileRestoredExpandedPathsInput): string[] {
-  if (restoredExpandedPaths === DIRECTORY_LISTING_INVALIDATED)
-    return Array.from(currentExpandedPaths);
   const reconciledPaths = new Set(restoredExpandedPaths);
 
   for (const path of persistedExpandedPaths) {
@@ -212,7 +205,7 @@ function sortExplorerEntries(entries: ExplorerEntry[], sortOption: SortOption):
 /** Load and expand the path to a directory selected outside the tree. */
 export async function revealExplorerDirectory(input: {
   directoryPath: string;
-  requestDirectoryListing: (path: string) => Promise<DirectoryListingResult>;
+  requestDirectoryListing: (path: string) => Promise<ExplorerDirectory | null>;
   expandPaths: (paths: string[]) => void;
   showHiddenFiles: () => void;
 }): Promise<void> {
@@ -225,51 +218,7 @@ export async function revealExplorerDirectory(input: {
     }
   }
   for (const path of paths) {
-    const directory = await input.requestDirectoryListing(path);
-    if (!directory || directory === DIRECTORY_LISTING_INVALIDATED) return;
+    if (!(await input.requestDirectoryListing(path))) return;
   }
   input.expandPaths(paths);
 }
-
-/**
- * Files Refresh: re-lists the expanded directories (and the root), and drops the cached
- * listing of every other directory. Expanding a directory only fetches when it has no
- * cached listing, so keeping a collapsed directory's old listing would show it as it was
- * before the refresh.
- */
-export async function refreshExplorerDirectories({
-  hasWorkspaceScope,
-  expandedPaths,
-  showHiddenFiles,
-  requestDirectoryListing,
-  invalidateDirectoryListings,
-}: {
-  hasWorkspaceScope: boolean;
-  expandedPaths: Set<string>;
-  showHiddenFiles: boolean;
-  requestDirectoryListing: (
-    path: string,
-    opts?: { recordHistory?: boolean; setCurrentPath?: boolean },
-  ) => Promise<DirectoryListingResult>;
-  invalidateDirectoryListings: (retainedPaths: readonly string[]) => void;
-}): Promise<null> {
-  if (!hasWorkspaceScope) {
-    return null;
-  }
-  const directoryPaths = Array.from(expandedPaths).filter(
-    (path) => showHiddenFiles || !isHiddenExplorerPath(path),
-  );
-  if (!directoryPaths.includes(".")) {
-    directoryPaths.unshift(".");
-  }
-  invalidateDirectoryListings(directoryPaths);
-  await Promise.all(
-    directoryPaths.map((path) =>
-      requestDirectoryListing(path, {
-        recordHistory: false,
-        setCurrentPath: false,
-      }),
-    ),
-  );
-  return null;
-}

packages/app/src/file-panels/html-preview.web.tsx

@@ -25,17 +25,12 @@ const iframeStyle = {
 
 export function FileHtmlPreview({ html, testID }: { html: string; testID?: string }) {
   const { t } = useTranslation();
-  // A document URL lets the browser resolve fragment links within the preview.
-  // srcDoc would inherit the app URL and navigate the frame back into Paseo.
-  const source = useMemo(
-    () => `data:text/html;charset=utf-8,${encodeURIComponent(withPreviewCsp(html))}`,
-    [html],
-  );
+  const document = useMemo(() => withPreviewCsp(html), [html]);
   return (
     <iframe
       data-testid={testID}
       title={t("panels.file.filePanels.builtin.html")}
-      src={source}
+      srcDoc={document}
       sandbox={SANDBOX}
       referrerPolicy="no-referrer"
       style={iframeStyle}

packages/app/src/git/diff-pane.tsx

@@ -56,6 +56,7 @@ import * as Clipboard from "expo-clipboard";
 import { useFileDownload } from "@/hooks/use-file-download";
 import { useIsLocalDaemon } from "@/hooks/use-is-local-daemon";
 import { buildAbsoluteExplorerPath } from "@/utils/explorer-paths";
+import { resolveWorkspaceFilePaths } from "@/workspace/file-open";
 import { Tooltip, TooltipTrigger, TooltipContent } from "@/components/ui/tooltip";
 import { GitActionsSplitButton } from "@/git/actions-split-button";
 import type { GitActions } from "@/git/policy";
@@ -1452,6 +1453,10 @@ function useDiffTabNavigation({
   };
 }
 
+function resolveChangesFileRoot(status: CheckoutStatusPayload | null, cwd: string): string {
+  return status?.repoRoot ?? cwd;
+}
+
 export function ChangesSurface({
   serverId,
   workspaceId,
@@ -1631,14 +1636,33 @@ export function ChangesSurface({
     }),
     [appSettings.monoFontFamily, codeFontSize, effectiveLayout, wrapLines],
   );
-  const downloadFile = useFileDownload({ serverId, workspaceId, workspaceRoot: cwd });
+  const fileRoot = resolveChangesFileRoot(status, cwd);
+  const resolveFilePath = useCallback(
+    (path: string) => buildAbsoluteExplorerPath({ workspaceRoot: fileRoot, entryPath: path }),
+    [fileRoot],
+  );
+  const resolveFileTargetPath = useCallback(
+    (path: string) => {
+      const absolutePath = resolveFilePath(path);
+      const resolved = resolveWorkspaceFilePaths({ path: absolutePath, workspaceRoot: cwd });
+      return resolved?.relativePath ?? absolutePath;
+    },
+    [cwd, resolveFilePath],
+  );
+  const handleOpenFile = useCallback(
+    (path: string) => onOpenFile?.(resolveFileTargetPath(path)),
+    [onOpenFile, resolveFileTargetPath],
+  );
+  const handleAddToChat = useCallback(
+    (path: string) => onAddToChat?.(resolveFileTargetPath(path)),
+    [onAddToChat, resolveFileTargetPath],
+  );
+  const downloadFile = useFileDownload({ serverId, workspaceId, workspaceRoot: fileRoot });
   const handleCopyPath = useCallback(
     (path: string) => {
-      void Clipboard.setStringAsync(
-        buildAbsoluteExplorerPath({ workspaceRoot: cwd, entryPath: path }),
-      );
+      void Clipboard.setStringAsync(resolveFilePath(path));
     },
-    [cwd],
+    [resolveFilePath],
   );
   const handleCopyRelativePath = useCallback((path: string) => {
     void Clipboard.setStringAsync(path);
@@ -1652,7 +1676,7 @@ export function ChangesSurface({
         await openDesktopTarget({
           editorId: fileManagerTarget.id,
           workspacePath: cwd,
-          filePath: buildAbsoluteExplorerPath({ workspaceRoot: cwd, entryPath: path }),
+          filePath: resolveFilePath(path),
         });
       } catch (cause) {
         toast.error(
@@ -1660,7 +1684,7 @@ export function ChangesSurface({
         );
       }
     },
-    [cwd, fileManagerTarget, t, toast],
+    [cwd, fileManagerTarget, resolveFilePath, t, toast],
   );
   const handleDownloadPath = useCallback(
     (path: string) => {
@@ -1674,7 +1698,7 @@ export function ChangesSurface({
         return;
       }
       try {
-        const payload = await client.duplicateFileEntry({ cwd, path });
+        const payload = await client.duplicateFileEntry({ cwd: fileRoot, path });
         if (!payload.success) {
           toast.error(payload.error ?? t("workspace.fileExplorer.errors.duplicateFailed"));
         }
@@ -1682,7 +1706,7 @@ export function ChangesSurface({
         toast.error(cause instanceof Error ? cause.message : String(cause));
       }
     },
-    [client, cwd, t, toast],
+    [client, fileRoot, t, toast],
   );
   const onRevertPath = useDiscardChangesAction({ serverId, cwd, diffMode });
   const [localFocusRequest, setLocalFocusRequest] = useState<{
@@ -1718,9 +1742,9 @@ export function ChangesSurface({
       focusPath: documentFocusRequest?.path,
       focusRequestId: documentFocusRequest?.revision,
       workspaceFileDragScope: workspaceId ? { serverId, workspaceId } : undefined,
-      onOpenFile,
+      onOpenFile: onOpenFile ? handleOpenFile : undefined,
       onOpenToSide,
-      onAddToChat,
+      onAddToChat: onAddToChat ? handleAddToChat : undefined,
       onCopyPath: handleCopyPath,
       onCopyRelativePath: handleCopyRelativePath,
       onReveal: fileManagerTarget ? handleRevealPath : undefined,
@@ -1736,8 +1760,10 @@ export function ChangesSurface({
       serverId,
       workspaceId,
       onOpenFile,
+      handleOpenFile,
       onOpenToSide,
       onAddToChat,
+      handleAddToChat,
       handleCopyPath,
       handleCopyRelativePath,
       handleDownloadPath,

packages/app/src/hooks/use-file-explorer-actions.ts

@@ -1,18 +1,16 @@
 import { useCallback, useMemo } from "react";
 import { useTranslation } from "react-i18next";
-import { useSessionStore, type AgentFileExplorerState } from "@/stores/session-store";
+import {
+  useSessionStore,
+  type AgentFileExplorerState,
+  type ExplorerDirectory,
+} from "@/stores/session-store";
 import { usePanelStore } from "@/stores/panel-store";
 import { revealExplorerDirectory } from "@/file-explorer/tree";
-import {
-  loadExplorerDirectory,
-  type DirectoryListingResult,
-  invalidateExplorerDirectoryListings,
-} from "@/file-explorer/directory-listings";
 import { parentExplorerPath } from "@/utils/explorer-paths";
 
 function createExplorerState(): AgentFileExplorerState {
   return {
-    listingGeneration: 0,
     directories: new Map(),
     isLoading: false,
     lastError: null,
@@ -24,6 +22,15 @@ function createExplorerState(): AgentFileExplorerState {
   };
 }
 
+function pushHistory(history: string[], path: string): string[] {
+  const normalizedHistory = history.length === 0 ? ["."] : history;
+  const last = normalizedHistory[normalizedHistory.length - 1];
+  if (last === path) {
+    return normalizedHistory;
+  }
+  return [...normalizedHistory, path];
+}
+
 export interface FileExplorerWorkspaceScope {
   workspaceId?: string | null;
   workspaceRoot?: string | null;
@@ -68,21 +75,16 @@ export function useFileExplorerActions(params: { serverId: string } & FileExplor
   );
 
   const updateExplorerState = useCallback(
-    (
-      updater: (prev: AgentFileExplorerState) => AgentFileExplorerState,
-    ): AgentFileExplorerState | null => {
+    (updater: (prev: AgentFileExplorerState) => AgentFileExplorerState) => {
       if (!workspaceStateKey) {
-        return null;
+        return;
       }
-      let updated: AgentFileExplorerState | null = null;
       setFileExplorer(serverId, (prev) => {
         const next = new Map(prev);
         const current = next.get(workspaceStateKey) ?? createExplorerState();
-        updated = updater(current);
-        next.set(workspaceStateKey, updated);
+        next.set(workspaceStateKey, updater(current));
         return next;
       });
-      return updated;
     },
     [serverId, setFileExplorer, workspaceStateKey],
   );
@@ -91,23 +93,80 @@ export function useFileExplorerActions(params: { serverId: string } & FileExplor
     async (
       path: string,
       options?: { recordHistory?: boolean; setCurrentPath?: boolean },
-    ): Promise<DirectoryListingResult> => {
+    ): Promise<ExplorerDirectory | null> => {
       if (!workspaceStateKey) {
         return null;
       }
       const normalizedPath = path && path.length > 0 ? path : ".";
-      return loadExplorerDirectory({
-        path: normalizedPath,
-        options,
-        updateState: updateExplorerState,
-        fallbackError: t("workspace.fileExplorer.errors.failedToListDirectory"),
-        listDirectory: async () => {
-          if (!normalizedWorkspaceRoot)
-            throw new Error(t("workspace.fileExplorer.states.unavailable"));
-          if (!client) throw new Error(t("workspace.terminal.hostDisconnected"));
-          return client.listDirectory(normalizedWorkspaceRoot, normalizedPath);
-        },
-      });
+      const shouldSetCurrentPath = options?.setCurrentPath ?? true;
+      const shouldRecordHistory = options?.recordHistory ?? shouldSetCurrentPath;
+
+      updateExplorerState((state) => ({
+        ...state,
+        isLoading: true,
+        lastError: null,
+        pendingRequest: { path: normalizedPath, mode: "list" },
+        ...(shouldSetCurrentPath
+          ? {
+              currentPath: normalizedPath,
+              history: shouldRecordHistory
+                ? pushHistory(state.history, normalizedPath)
+                : state.history,
+              lastVisitedPath: normalizedPath,
+            }
+          : {}),
+      }));
+
+      if (!normalizedWorkspaceRoot) {
+        updateExplorerState((state) => ({
+          ...state,
+          isLoading: false,
+          lastError: t("workspace.fileExplorer.states.unavailable"),
+          pendingRequest: null,
+        }));
+        return null;
+      }
+
+      if (!client) {
+        updateExplorerState((state) => ({
+          ...state,
+          isLoading: false,
+          lastError: t("workspace.terminal.hostDisconnected"),
+          pendingRequest: null,
+        }));
+        return null;
+      }
+
+      try {
+        const directory = await client.listDirectory(normalizedWorkspaceRoot, normalizedPath);
+        updateExplorerState((state) => {
+          const nextState: AgentFileExplorerState = {
+            ...state,
+            isLoading: false,
+            lastError: null,
+            pendingRequest: null,
+            directories: state.directories,
+          };
+
+          const directories = new Map(state.directories);
+          directories.set(directory.path, directory);
+          nextState.directories = directories;
+
+          return nextState;
+        });
+        return directory;
+      } catch (error) {
+        updateExplorerState((state) => ({
+          ...state,
+          isLoading: false,
+          lastError:
+            error instanceof Error
+              ? error.message
+              : t("workspace.fileExplorer.errors.failedToListDirectory"),
+          pendingRequest: null,
+        }));
+        return null;
+      }
     },
     [client, normalizedWorkspaceRoot, t, updateExplorerState, workspaceStateKey],
   );
@@ -249,17 +308,10 @@ export function useFileExplorerActions(params: { serverId: string } & FileExplor
     [updateExplorerState],
   );
 
-  const invalidateDirectoryListings = useCallback(
-    (retainedPaths: readonly string[]) =>
-      invalidateExplorerDirectoryListings(updateExplorerState, retainedPaths),
-    [updateExplorerState],
-  );
-
   return {
     workspaceStateKey,
     revealDirectory,
     requestDirectoryListing,
-    invalidateDirectoryListings,
     requestFileDownloadToken,
     requestFileBytes,
     createEntry,

packages/app/src/plugins/buttons/model.test.ts

@@ -17,9 +17,6 @@ function installation(): InstalledPlugin {
     queryClient: new QueryClient(),
     paseo: {} as InstalledPlugin["paseo"],
     invoke: async () => undefined,
-    subscribe: () => {
-      throw new Error("Subscriptions are not used here");
-    },
     cleanup: () => undefined,
     settingsScreens: [],
     surfaces: [],

packages/app/src/plugins/catalog-sync-model.test.ts

@@ -153,9 +153,6 @@ function harness() {
         openPanel() {},
         addComposerPill: () => ({ update() {}, remove() {} }),
         addHeaderButton: () => ({ update() {}, remove() {} }),
-        subscribe: () => {
-          throw new Error("Subscriptions are not used here");
-        },
       };
     },
   });

packages/app/src/plugins/client-runtime.ts

@@ -2,7 +2,6 @@ import { createPlayAudio, type AudioEngine } from "@/audio";
 import { getHostRuntimeStore } from "@/runtime/host-runtime";
 import { createPluginHosts } from "./hosts";
 import type { PluginClientOpenPanelOptions } from "@getpaseo/plugin/client";
-import { subscribePluginRpc } from "@getpaseo/plugin/client/host";
 import {
   createPluginAgentActionContext,
   createPluginCapabilities,
@@ -37,8 +36,6 @@ export function createPluginClientRuntime(
     openPanel(panelId, options) {
       openClientPanel({ installation, state, panelId, options });
     },
-    subscribe: (contract, input, onEvent, options) =>
-      subscribePluginRpc(contract, installation.subscribe, input, onEvent, options),
   };
 }
 

packages/app/src/plugins/command-center/contributions.test.ts

@@ -71,9 +71,6 @@ function plugin(onAgentSelect: AgentCommandItem["onSelect"]): InstalledPlugin {
         clientType: "cli",
       }),
     ),
-    subscribe: () => {
-      throw new Error("Subscriptions are not used here");
-    },
     invoke: async (method: string, input: unknown) => {
       expect(method).toBe("review.inspect");
       return { value: inspect.input.parse(input).value + 1 };

packages/app/src/plugins/evaluate.ts

@@ -111,7 +111,6 @@ export type PluginClientRuntime = Pick<
   | "openScreen"
   | "openSurface"
   | "openPanel"
-  | "subscribe"
   | "addComposerPill"
   | "addHeaderButton"
 > & { hosts: ReturnType<typeof createPluginHosts> };

packages/app/src/plugins/installation-provider.test.tsx

@@ -29,9 +29,6 @@ function installation(invoked: string[]): InstalledPlugin {
       invoked.push(method);
       return { state: "green" };
     },
-    subscribe: () => {
-      throw new Error("Subscriptions are not used here");
-    },
     queryClient: new QueryClient(),
     cleanup: () => undefined,
     settingsScreens: [],

packages/app/src/plugins/installation-provider.tsx

@@ -18,7 +18,7 @@ export function PluginInstallationProvider({
   return (
     <QueryClientProvider client={plugin.queryClient}>
       <PaseoApiProvider paseo={plugin.paseo}>
-        <PluginRpcProvider invoke={plugin.invoke} subscribe={plugin.subscribe}>
+        <PluginRpcProvider invoke={plugin.invoke}>
           <PluginTranslationProvider runtime={plugin.i18n}>{children}</PluginTranslationProvider>
         </PluginRpcProvider>
       </PaseoApiProvider>

packages/app/src/plugins/registry-lifecycle.test.ts

@@ -62,9 +62,6 @@ function registry(
         openPanel: () => {},
         addComposerPill: () => ({ update() {}, remove() {} }),
         addHeaderButton: () => ({ update() {}, remove() {} }),
-        subscribe: () => {
-          throw new Error("Subscriptions are not used here");
-        },
       };
     },
   });

packages/app/src/plugins/registry-requirements.test.ts

@@ -39,9 +39,6 @@ function registry(version: string) {
         openPanel() {},
         addComposerPill: () => ({ update() {}, remove() {} }),
         addHeaderButton: () => ({ update() {}, remove() {} }),
-        subscribe: () => {
-          throw new Error("Subscriptions are not used here");
-        },
       };
     },
   });

packages/app/src/plugins/registry.ts

@@ -8,7 +8,6 @@ import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
 import { assertPluginCompatibility } from "@getpaseo/protocol/plugin-requirements";
 import { resolveAppVersion } from "@/utils/app-version";
 import { createPluginClientRuntime } from "./client-runtime";
-import { createPluginSubscribe } from "./subscriptions";
 import { runPluginClientBundle } from "./evaluate";
 import type { InstalledPlugin } from "./types";
 
@@ -90,18 +89,12 @@ export class PluginRegistry {
           return [existing];
         }
         const client = options.client;
-        const lifetime = new AbortController();
         installation = {
-          lifetime,
+          lifetime: new AbortController(),
           locale: this.dependencies.localeSource.getLocale(),
           i18n: new PluginTranslationRuntime(this.dependencies.localeSource),
           paseo: createPaseoApi(client),
           invoke: (method, input) => client.invokePluginRpc(entry.id, method, input),
-          subscribe: createPluginSubscribe({
-            client,
-            pluginId: entry.id,
-            lifetime: lifetime.signal,
-          }),
           id: entry.id,
           serverId,
           clientBundle: entry.clientBundle,

packages/app/src/plugins/sidebar-groups.test.ts

@@ -26,9 +26,6 @@ function installed(
     queryClient: new QueryClient(),
     paseo: {} as InstalledPlugin["paseo"],
     invoke: async () => undefined,
-    subscribe: () => {
-      throw new Error("Subscriptions are not used here");
-    },
     settingsScreens: [],
     surfaces: [{ id: "surface", title: "Surface", Component: () => null }],
     sidebarItems: {

packages/app/src/plugins/subscriptions.test.ts

@@ -1,189 +0,0 @@
-import { defineSubscription } from "@getpaseo/plugin";
-import type { PluginSubscriptionClosed } from "@getpaseo/plugin/client";
-import { subscribePluginRpc } from "@getpaseo/plugin/client/host";
-import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
-import type { SubscriptionObserver } from "@getpaseo/client";
-import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
-import { expect, it } from "vitest";
-import { z } from "zod";
-import { createPluginSubscribe } from "./subscriptions";
-
-const changes = defineSubscription({
-  name: "diff.watch",
-  input: z.object({ ref: z.string() }),
-  event: z.union([z.instanceof(Uint8Array), z.object({ path: z.string() })]),
-});
-
-type Observed = Parameters<DaemonClient["observePluginRpc"]>;
-
-/** One host-side subscription as the daemon client exposes it, driven by the test. */
-class HostSubscription {
-  readonly released: Promise<void>;
-  isReleased = false;
-  private observer: SubscriptionObserver<never> | null = null;
-  private accept!: () => void;
-  private refuse!: (error: Error) => void;
-  private markReleased!: () => void;
-  readonly ready = new Promise<never>((resolve, reject) => {
-    this.accept = () => resolve({ requestId: "r", subscriptionId: "s" } as never);
-    this.refuse = reject;
-  });
-
-  constructor(
-    readonly method: string,
-    readonly input: unknown,
-    readonly signal: AbortSignal | undefined,
-  ) {
-    void this.ready.catch(() => undefined);
-    this.released = new Promise((resolve) => {
-      this.markReleased = resolve;
-    });
-    signal?.addEventListener("abort", () => void this.release(), { once: true });
-  }
-
-  handle() {
-    return {
-      subscriptionId: "s",
-      ready: this.ready,
-      subscribe: (observer: SubscriptionObserver<never>) => {
-        this.observer = observer;
-        return () => undefined;
-      },
-      release: () => this.release(),
-    };
-  }
-
-  acknowledge() {
-    this.accept();
-  }
-
-  refuseWith(error: Error) {
-    this.refuse(error);
-    this.observer?.error?.(error);
-  }
-
-  event(event: unknown) {
-    this.observer?.update({
-      type: "plugin.rpc.event",
-      payload: { subscriptionId: "s", event },
-    } as SessionOutboundMessage);
-  }
-
-  bytes(payload: Uint8Array) {
-    this.observer?.binary?.(payload);
-  }
-
-  closed(error?: string) {
-    this.observer?.update({
-      type: "plugin.rpc.closed",
-      payload: { subscriptionId: "s", ...(error ? { error } : {}) },
-    } as SessionOutboundMessage);
-  }
-
-  async release() {
-    this.isReleased = true;
-    this.markReleased();
-  }
-}
-
-/** An installation's subscribe on a host the test drives, and what one subscriber observed. */
-function installation() {
-  const subscriptions: HostSubscription[] = [];
-  const client = {
-    observePluginRpc: (...[pluginId, method, input, options]: Observed) => {
-      expect(pluginId).toBe("diffs");
-      const subscription = new HostSubscription(method, input, options?.signal);
-      subscriptions.push(subscription);
-      return subscription.handle() as unknown as ReturnType<DaemonClient["observePluginRpc"]>;
-    },
-  };
-  const lifetime = new AbortController();
-  const subscribe = createPluginSubscribe({ client, pluginId: "diffs", lifetime: lifetime.signal });
-  function open(ref: string, signal?: AbortSignal) {
-    const received: string[] = [];
-    const closed: PluginSubscriptionClosed[] = [];
-    const handle = subscribePluginRpc(
-      changes,
-      subscribe,
-      { ref },
-      (event) =>
-        received.push(event instanceof Uint8Array ? `bytes:${[...event]}` : `path:${event.path}`),
-      { signal, onClosed: (result) => closed.push(result) },
-    );
-    return { handle, received, closed, host: subscriptions[subscriptions.length - 1]! };
-  }
-  return { open, lifetime };
-}
-
-it("delivers JSON and byte events in order and releases the host subscription", async () => {
-  const { open } = installation();
-  const { handle, received, closed, host } = open("main");
-  expect(host.method).toBe("diff.watch");
-  expect(host.input).toEqual({ ref: "main" });
-  host.acknowledge();
-  await handle.ready;
-
-  host.event({ path: "a.ts" });
-  host.bytes(new Uint8Array([1, 2]));
-  expect(received).toEqual(["path:a.ts", "bytes:1,2"]);
-
-  handle();
-  await host.released;
-  expect(closed).toEqual([]);
-});
-
-it("reports a refusal, a failure after going live, and a clean close", async () => {
-  const { open } = installation();
-  const refused = open("a");
-  refused.host.refuseWith(new Error("no such ref"));
-  await expect(refused.handle.ready).rejects.toThrow("no such ref");
-  expect(refused.closed).toEqual([{ error: new Error("no such ref") }]);
-
-  const failed = open("b");
-  failed.host.acknowledge();
-  await failed.handle.ready;
-  failed.host.closed("Plugin stopped: diffs");
-  expect(failed.closed).toEqual([{ error: new Error("Plugin stopped: diffs") }]);
-  await failed.host.released;
-
-  // A handler that returns at once closes in the same tick the host accepts it.
-  const finished = open("c");
-  finished.host.acknowledge();
-  finished.host.closed();
-  await expect(finished.handle.ready).resolves.toBeUndefined();
-  expect(finished.closed).toEqual([{}]);
-});
-
-it("closes a subscription whose event fails the contract and drops what follows", async () => {
-  const { open } = installation();
-  const { handle, received, closed, host } = open("a");
-  host.acknowledge();
-  await handle.ready;
-  host.event({ path: 3 });
-  host.event({ path: "late.ts" });
-  expect(received).toEqual([]);
-  expect(closed).toEqual([
-    { error: new Error("Plugin subscription diff.watch sent an invalid event") },
-  ]);
-  await host.released;
-});
-
-it("releases on its signal, on its remover, and when the installation stops", async () => {
-  const { open, lifetime } = installation();
-  const controller = new AbortController();
-  const bySignal = open("a", controller.signal);
-  bySignal.host.acknowledge();
-  await bySignal.handle.ready;
-  controller.abort();
-  await bySignal.host.released;
-
-  const byRemover = open("b");
-  byRemover.handle();
-  await byRemover.host.released;
-
-  const byLifetime = open("c");
-  byLifetime.host.event({ path: "a.ts" });
-  expect(byLifetime.received).toEqual(["path:a.ts"]);
-  lifetime.abort();
-  await byLifetime.host.released;
-});

packages/app/src/plugins/subscriptions.ts

@@ -1,52 +0,0 @@
-import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
-import type { PluginSubscribe } from "@getpaseo/plugin/client/host";
-
-/**
- * Plugin subscriptions on the installation's client. The client re-subscribes after a reconnect;
- * the installation's lifetime releases whatever its surfaces left open.
- */
-export function createPluginSubscribe(input: {
-  client: Pick<DaemonClient, "observePluginRpc">;
-  pluginId: string;
-  lifetime: AbortSignal;
-}): PluginSubscribe {
-  const { client, pluginId, lifetime } = input;
-  return (method, subscriptionInput, observer) => {
-    const controller = new AbortController();
-    const abort = () => controller.abort();
-    lifetime.addEventListener("abort", abort, { once: true });
-    controller.signal.addEventListener(
-      "abort",
-      () => lifetime.removeEventListener("abort", abort),
-      { once: true },
-    );
-    if (lifetime.aborted) controller.abort();
-    const subscription = client.observePluginRpc(pluginId, method, subscriptionInput, {
-      signal: controller.signal,
-    });
-    subscription.subscribe({
-      snapshot: () => undefined,
-      update: (message) => {
-        if (message.type === "plugin.rpc.event") observer.event(message.payload.event);
-        else if (message.type === "plugin.rpc.closed") {
-          controller.abort();
-          const error = message.payload.error;
-          observer.closed(error === undefined ? undefined : new Error(error));
-        }
-      },
-      binary: (payload) => observer.event(payload),
-      error: (error) => {
-        controller.abort();
-        observer.closed(error instanceof Error ? error : new Error(String(error)));
-      },
-    });
-    const release = () => {
-      controller.abort();
-      return subscription.release();
-    };
-    return Object.assign(() => void release().catch(() => undefined), {
-      ready: subscription.ready.then(() => undefined),
-      release,
-    });
-  };
-}

packages/app/src/plugins/surface-contribution.test.ts

@@ -26,9 +26,6 @@ function installation(
     queryClient: new QueryClient(),
     paseo: {} as InstalledPlugin["paseo"],
     invoke: async () => undefined,
-    subscribe: () => {
-      throw new Error("Subscriptions are not used here");
-    },
     cleanup: () => undefined,
     settingsScreens: [],
     surfaces: surfaces.map((id) => ({ id, title: id, Component: () => null })),

packages/app/src/plugins/theme.test.ts

@@ -56,9 +56,6 @@ function installed(serverId: string, themes: PluginThemeContribution[]): Install
     queryClient: new QueryClient(),
     paseo: {} as InstalledPlugin["paseo"],
     invoke: async () => undefined,
-    subscribe: () => {
-      throw new Error("Subscriptions are not used here");
-    },
     settingsScreens: [],
     surfaces: [],
     sidebarItems: { header: [], footer: [] },

packages/app/src/plugins/timeline/model.test.ts

@@ -18,9 +18,6 @@ function plugin(input: {
     queryClient: new QueryClient(),
     paseo: {} as InstalledPlugin["paseo"],
     invoke: async () => undefined,
-    subscribe: () => {
-      throw new Error("Subscriptions are not used here");
-    },
     cleanup: () => {},
     settingsScreens: [],
     surfaces: [],

packages/app/src/plugins/types.ts

@@ -71,8 +71,6 @@ export interface InstalledPlugin extends EvaluatedPlugin {
   paseo: PaseoApi;
   /** Calls one of the plugin's server RPC methods on its host. */
   invoke(method: string, input: unknown): Promise<unknown>;
-  /** Subscribes to one of the plugin's server subscriptions; released with the lifetime. */
-  subscribe: import("@getpaseo/plugin/client/host").PluginSubscribe;
   serverId: string;
   requirements?: PluginRequirements;
   clientBundle: string;

packages/app/src/plugins/workspace-panels/resolution.test.ts

@@ -15,9 +15,6 @@ function installed(): InstalledPlugin {
     queryClient: new QueryClient(),
     paseo: {} as InstalledPlugin["paseo"],
     invoke: async () => undefined,
-    subscribe: () => {
-      throw new Error("Subscriptions are not used here");
-    },
     cleanup: () => {},
     settingsScreens: [],
     surfaces: [],

packages/app/src/stores/session-store.ts

@@ -260,7 +260,6 @@ interface ExplorerRequestState {
 }
 
 export interface AgentFileExplorerState {
-  listingGeneration: number;
   directories: Map<string, ExplorerDirectory>;
   isLoading: boolean;
   lastError: string | null;

packages/cli/src/commands/agent/logs-json.test.ts

@@ -1,50 +0,0 @@
-import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import { mkdir, mkdtemp } from "node:fs/promises";
-import { resolve } from "node:path";
-import { connectToDaemon } from "../../utils/client.js";
-import { runPaseoCli, startTestDaemon } from "../../../tests/helpers/test-daemon.js";
-
-describe("paseo logs JSON output", () => {
-  let daemon: Awaited<ReturnType<typeof startTestDaemon>>;
-  let agentId: string;
-  let host: string;
-
-  beforeAll(async () => {
-    await mkdir(".artifacts", { recursive: true });
-    daemon = await startTestDaemon({
-      paseoHome: await mkdtemp(resolve(".artifacts/logs-json-home-")),
-      workDir: await mkdtemp(resolve(".artifacts/logs-json-work-")),
-    });
-    host = `127.0.0.1:${daemon.port}`;
-    const client = await connectToDaemon({ target: { kind: "endpoint", host } });
-    try {
-      // Creating an idle agent needs no inference or external provider connection.
-      const agent = await client.createAgent({
-        provider: "claude",
-        cwd: daemon.workDir,
-        title: "JSON logs",
-      });
-      agentId = agent.id;
-    } finally {
-      await client.close();
-    }
-  }, 60_000);
-
-  afterAll(async () => {
-    await daemon?.stop();
-  });
-
-  it.each([
-    { label: "top-level --json", args: ["logs", "--json"] },
-    { label: "agent logs --json", args: ["agent", "logs", "--json"] },
-    { label: "--format json alias", args: ["logs", "--format", "json"] },
-  ])(
-    "prints JSON through $label",
-    async ({ args }) => {
-      const result = await runPaseoCli(daemon, [...args, agentId, "--host", host]);
-      expect(result.exitCode, result.stderr).toBe(0);
-      expect(JSON.parse(result.stdout)).toEqual({ agentId, items: [] });
-    },
-    30_000,
-  );
-});

packages/cli/src/commands/agent/logs.test.ts

@@ -1,10 +1,6 @@
 import { describe, expect, it } from "vitest";
 import type { AgentTimelineItem } from "@getpaseo/protocol/agent-types";
-import {
-  createFollowJsonWriter,
-  createFollowTranscriptWriter,
-  formatAgentActivityTranscript,
-} from "./logs.js";
+import { createFollowTranscriptWriter, formatAgentActivityTranscript } from "./logs.js";
 
 function collect(items: readonly AgentTimelineItem[], turnId?: string): string {
   let output = "";
@@ -153,31 +149,3 @@ describe("createFollowTranscriptWriter", () => {
     expect(output).toBe("# Title\n\nBody\n");
   });
 });
-
-describe("createFollowJsonWriter", () => {
-  it("writes one JSON item per line and honours the filter", () => {
-    let output = "";
-    const writer = createFollowJsonWriter("tools", (chunk) => {
-      output += chunk;
-    });
-    const toolCall = {
-      type: "tool_call",
-      callId: "call_1",
-      name: "shell",
-      status: "completed",
-      detail: { type: "unknown", input: null, output: null },
-      error: null,
-    } as AgentTimelineItem;
-
-    writer.push({ type: "assistant_message", text: "Running it.", messageId: "msg_1" });
-    writer.push(toolCall);
-    writer.end();
-
-    expect(
-      output
-        .trimEnd()
-        .split("\n")
-        .map((line) => JSON.parse(line)),
-    ).toEqual([toolCall]);
-  });
-});

packages/cli/src/commands/agent/logs.ts

@@ -70,35 +70,6 @@ export function formatAgentActivityTranscript(
   );
 }
 
-/**
- * Picks the items `--tail` keeps, the same way the transcript does: the last n
- * timeline items, before any of them are dropped from the curated view.
- */
-function selectTailItems(
-  timelineItems: AgentTimelineItem[],
-  tailCount?: number,
-): AgentTimelineItem[] {
-  if (tailCount === undefined) {
-    return timelineItems;
-  }
-  if (tailCount === 0) {
-    return [];
-  }
-  return timelineItems.slice(-tailCount);
-}
-
-function formatAgentLogsJson(
-  agentId: string,
-  timelineItems: AgentTimelineItem[],
-  tailCount?: number,
-): string {
-  return JSON.stringify({ agentId, items: selectTailItems(timelineItems, tailCount) }, null, 2);
-}
-
-function usesJsonOutput(options: AgentLogsOptions): boolean {
-  return options.json === true || options.format?.trim().toLowerCase() === "json";
-}
-
 type StreamingTextType = "assistant_message" | "reasoning";
 
 const THOUGHT_PREFIX = "[Thought] ";
@@ -227,21 +198,6 @@ export function createFollowTranscriptWriter(
   };
 }
 
-/** Follow mode with `--json`: one JSON timeline item per line (NDJSON). */
-export function createFollowJsonWriter(
-  filter?: string,
-  write: (chunk: string) => void = (chunk) => void process.stdout.write(chunk),
-): FollowTranscriptWriter {
-  return {
-    push(item) {
-      if (matchesFilter(item, filter)) {
-        write(`${JSON.stringify(item)}\n`);
-      }
-    },
-    end() {},
-  };
-}
-
 function parseTailCount(raw: string | undefined): number | undefined {
   if (raw === undefined) return undefined;
   const parsed = Number.parseInt(raw, 10);
@@ -350,11 +306,6 @@ export async function runLogsCommand(
 
     await client.close();
 
-    if (usesJsonOutput(options)) {
-      console.log(formatAgentLogsJson(resolvedId, timelineItems, tailCount));
-      return;
-    }
-
     // Use curateAgentActivity to format the transcript
     if (tailCount === 0) {
       return;
@@ -380,104 +331,78 @@ async function runFollowMode({
   options,
   sinceTimestampMs,
 }: FollowModeInput): Promise<void> {
-  // Observe shutdown before history writes can encounter a closed reader.
-  // Keep an early stop resolved while subscription setup is in flight.
-  const stopController = new AbortController();
-  const stopped = waitForStop(stopController.signal);
-  let unsubscribe: (() => void) | undefined;
-  const json = usesJsonOutput(options);
-  const writer = json
-    ? createFollowJsonWriter(options.filter)
-    : createFollowTranscriptWriter(undefined, options.filter);
+  const DEFAULT_FOLLOW_TAIL = 10;
+  const tailCount = parseTailCount(options.tail) ?? DEFAULT_FOLLOW_TAIL;
+
+  // First, get existing timeline.
+  let existingItems: AgentTimelineItem[] = [];
   try {
-    const DEFAULT_FOLLOW_TAIL = 10;
-    const tailCount = parseTailCount(options.tail) ?? DEFAULT_FOLLOW_TAIL;
-
-    // First, get existing timeline.
-    let existingItems: AgentTimelineItem[] = [];
-    try {
-      const history = await Promise.race([
-        fetchAgentTimelineItems(client, agentId, {
-          timeoutMs: LIVE_HISTORY_FETCH_TIMEOUT_MS,
-          sinceTimestampMs,
-        }),
-        stopped,
-      ]);
-      if (!history) return;
-      existingItems = history;
-    } catch (error) {
-      console.warn("Warning: failed to fetch existing timeline", error);
-    }
-
-    // Apply filter to existing items
-    if (options.filter) {
-      existingItems = existingItems.filter((item) => matchesFilter(item, options.filter));
-    }
-
-    // Print existing transcript (tail-like behavior)
-    if (json) {
-      for (const item of selectTailItems(existingItems, tailCount)) {
-        console.log(JSON.stringify(item));
-      }
-    } else if (tailCount > 0) {
-      const existingTranscript = formatAgentActivityTranscript(existingItems, tailCount);
-      if (existingTranscript !== NO_ACTIVITY_MESSAGE) {
-        console.log(existingTranscript);
-      }
-    }
-
-    // Subscribe to new events
-    const tailLabel =
-      tailCount === 0 ? "no history" : `last ${tailCount} entr${tailCount === 1 ? "y" : "ies"}`;
-
-    // Keep stdout parseable in JSON mode: one timeline item per line, notices on stderr.
-    const notice = json ? console.error : console.log;
-
-    const subscription = client.subscribeAgentTimeline(agentId, (message) => {
-      if (message.type === "agent.timeline.replacement") {
-        writer.end();
-        notice("\n[Timeline replaced; earlier output is no longer current]");
-        return;
-      }
-
-      if (message.type === "agent.timeline.error") {
-        writer.end();
-        console.error(`Timeline observation stopped: ${message.payload.error}`);
-        return;
-      }
-      if (message.type === "agent.timeline.subscription_restored") {
-        writer.end();
-        notice("\n[Reconnected; live output resumed. Events may have been missed.]");
-        return;
-      }
-      if (message.payload.event.type === "timeline") {
-        const matchesSince =
-          sinceTimestampMs === undefined ||
-          Date.parse(message.payload.timestamp) >= sinceTimestampMs;
-        if (!matchesSince) {
-          return;
-        }
-        const item = message.payload.event.item;
-        writer.push(item, message.payload.event.turnId);
-      } else if (
-        message.payload.event.type === "turn_completed" ||
-        message.payload.event.type === "turn_failed" ||
-        message.payload.event.type === "turn_canceled"
-      ) {
-        writer.end();
-      }
+    existingItems = await fetchAgentTimelineItems(client, agentId, {
+      timeoutMs: LIVE_HISTORY_FETCH_TIMEOUT_MS,
+      sinceTimestampMs,
     });
-
-    unsubscribe = subscription;
-    const ready = await Promise.race([subscription.ready.then(() => true), stopped]);
-    if (!ready) return;
-    notice(`\n--- Following logs (${tailLabel}; Ctrl+C to stop) ---\n`);
-
-    await stopped;
-  } finally {
-    stopController.abort();
-    writer.end();
-    unsubscribe?.();
-    await client.close();
+  } catch (error) {
+    console.warn("Warning: failed to fetch existing timeline", error);
   }
+
+  // Apply filter to existing items
+  if (options.filter) {
+    existingItems = existingItems.filter((item) => matchesFilter(item, options.filter));
+  }
+
+  // Print existing transcript (tail-like behavior)
+  if (tailCount > 0) {
+    const existingTranscript = formatAgentActivityTranscript(existingItems, tailCount);
+    if (existingTranscript !== NO_ACTIVITY_MESSAGE) {
+      console.log(existingTranscript);
+    }
+  }
+
+  // Subscribe to new events
+  const tailLabel =
+    tailCount === 0 ? "no history" : `last ${tailCount} entr${tailCount === 1 ? "y" : "ies"}`;
+
+  const writer = createFollowTranscriptWriter(undefined, options.filter);
+
+  const unsubscribe = client.subscribeAgentTimeline(agentId, (message) => {
+    if (message.type === "agent.timeline.replacement") {
+      writer.end();
+      console.log("\n[Timeline replaced; earlier output is no longer current]");
+      return;
+    }
+
+    if (message.type === "agent.timeline.error") {
+      writer.end();
+      console.error(`Timeline observation stopped: ${message.payload.error}`);
+      return;
+    }
+    if (message.type === "agent.timeline.subscription_restored") {
+      writer.end();
+      console.log("\n[Reconnected; live output resumed. Events may have been missed.]");
+      return;
+    }
+    if (message.payload.event.type === "timeline") {
+      const matchesSince =
+        sinceTimestampMs === undefined || Date.parse(message.payload.timestamp) >= sinceTimestampMs;
+      if (!matchesSince) {
+        return;
+      }
+      const item = message.payload.event.item;
+      writer.push(item, message.payload.event.turnId);
+    } else if (
+      message.payload.event.type === "turn_completed" ||
+      message.payload.event.type === "turn_failed" ||
+      message.payload.event.type === "turn_canceled"
+    ) {
+      writer.end();
+    }
+  });
+
+  await unsubscribe.ready;
+  console.log(`\n--- Following logs (${tailLabel}; Ctrl+C to stop) ---\n`);
+
+  await waitForStop();
+  writer.end();
+  unsubscribe();
+  await client.close();
 }

packages/cli/src/commands/agent/ls.test.ts

@@ -4,13 +4,13 @@ import { buildAgentLsFetchOptions, listAgents } from "./ls.js";
 
 const daemonTarget = { kind: "endpoint" as const, host: "example.test:12345" };
 
-function createPagingClient(archivedAgentNumber?: number) {
+function createPagingClient() {
   const daemonAgents: AgentSnapshotPayload[] = [3, 2, 1].map((n) => ({
     id: `${n}${n}${n}${n}${n}${n}${n}${n}-0000-4000-8000-000000000000`,
     provider: "codex",
     title: `agent ${n}`,
-    status: n === archivedAgentNumber ? "idle" : "closed",
-    archivedAt: n === archivedAgentNumber ? "2026-10-10T00:00:00.000Z" : null,
+    status: "closed",
+    archivedAt: null,
     cwd: "/tmp/project",
     createdAt: `2026-10-0${n}T00:00:00.000Z`,
     updatedAt: `2026-10-0${n}T00:00:00.000Z`,
@@ -130,28 +130,4 @@ describe("runLsCommand", () => {
 
     expect(result.data.map((item) => item.name)).toEqual(["agent 3", "agent 2", "agent 1"]);
   });
-
-  it("marks archived agents in table and JSON list data", async () => {
-    const result = await listAgents({ daemonTarget, all: true, global: true }, async () =>
-      createPagingClient(2),
-    );
-
-    expect(
-      result.data.map(({ name, archived, archivedAt }) => ({ name, archived, archivedAt })),
-    ).toEqual([
-      { name: "agent 2", archived: true, archivedAt: "2026-10-10T00:00:00.000Z" },
-      { name: "agent 3", archived: false, archivedAt: null },
-      { name: "agent 1", archived: false, archivedAt: null },
-    ]);
-
-    const archivedColumn = result.schema.columns.find((column) => column.header === "ARCHIVED");
-    expect(archivedColumn).toBeDefined();
-    expect(typeof archivedColumn?.field).toBe("function");
-    if (!archivedColumn || typeof archivedColumn.field !== "function") {
-      throw new Error("Expected the ARCHIVED column to have a display accessor");
-    }
-    expect(result.data[0]!.status).toBe("idle");
-    expect(archivedColumn.field(result.data[0]!)).toBe("yes");
-    expect(archivedColumn.field(result.data[1]!)).toBe("-");
-  });
 });

packages/cli/src/commands/agent/ls.ts

@@ -33,8 +33,6 @@ export interface AgentListItem {
   provider: string;
   thinking: string;
   status: string;
-  archived: boolean;
-  archivedAt: string | null;
   cwd: string;
   created: string;
 }
@@ -86,11 +84,6 @@ export const agentLsSchema: OutputSchema<AgentListItem> = {
         return undefined;
       },
     },
-    {
-      header: "ARCHIVED",
-      field: (item) => (item.archived ? "yes" : "-"),
-      width: 8,
-    },
     { header: "CWD", field: "cwd", width: 30 },
     { header: "CREATED", field: "created", width: 15 },
   ],
@@ -106,8 +99,6 @@ function toListItem(agent: AgentSnapshotPayload): AgentListItem {
     provider: model ? `${agent.provider}/${model}` : agent.provider,
     thinking: agent.effectiveThinkingOptionId ?? "auto",
     status: agent.status,
-    archived: agent.archivedAt != null,
-    archivedAt: agent.archivedAt ?? null,
     cwd: shortenPath(agent.cwd),
     created: relativeTime(agent.createdAt),
   };

packages/cli/src/utils/wait-for-stop.ts

@@ -3,10 +3,9 @@
 // their output closes its end, as `| head` does. A closed reader shows up as
 // EPIPE on the next write to stdout, so the command stops at its next line of
 // output.
-export function waitForStop(signal?: AbortSignal): Promise<void> {
+export function waitForStop(): Promise<void> {
   return new Promise((resolve) => {
     const stop = () => {
-      signal?.removeEventListener("abort", stop);
       process.off("SIGINT", stop);
       process.off("SIGTERM", stop);
       process.stdout.off("error", stopOnClosedStdout);
@@ -21,7 +20,5 @@ export function waitForStop(signal?: AbortSignal): Promise<void> {
     process.on("SIGINT", stop);
     process.on("SIGTERM", stop);
     process.stdout.on("error", stopOnClosedStdout);
-    signal?.addEventListener("abort", stop, { once: true });
-    if (signal?.aborted) stop();
   });
 }

packages/client/src/connection/index.ts

@@ -144,7 +144,6 @@ export const DEFAULT_CLIENT_CAPABILITIES = {
   [CLIENT_CAPS.pluginTimelineItems]: true,
   [CLIENT_CAPS.workspaceSetupBlocked]: true,
   [CLIENT_CAPS.workspaceSetupObservation]: true,
-  [CLIENT_CAPS.pluginBinaryPayloads]: true,
   [CLIENT_CAPS.explicitEventSubscriptions]: true,
 } satisfies Record<Exclude<ClientCapability, typeof CLIENT_CAPS.browserHost>, true>;
 

packages/client/src/connection/owned.test.ts

@@ -83,26 +83,6 @@ test("cancellation before the ID waits for bootstrap and releases exactly that I
   expect(c.released).toEqual(["cancelled"]);
 });
 
-test("a release right after the host accepts still reports the subscription as accepted", async () => {
-  const c = connection();
-  const handle = c.observe();
-  c.requests[0]!("closing");
-  const released = handle.release();
-  await expect(handle.ready).resolves.toMatchObject({ subscriptionId: "closing" });
-  await released;
-  expect(c.released).toEqual(["closing"]);
-});
-
-test("a release before the host accepts rejects ready", async () => {
-  const c = connection();
-  const handle = c.observe();
-  const released = handle.release();
-  await expect(handle.ready).rejects.toThrow("Subscription released");
-  c.requests[0]!("late");
-  await released;
-  expect(c.released).toEqual(["late"]);
-});
-
 test("surviving handles use new remote IDs after reconnect and released handles stay gone", async () => {
   const c = connection();
   const a = c.observe();

packages/client/src/connection/owned.ts

@@ -5,8 +5,6 @@ export type SubscriptionSnapshot<T> = T & { subscriptionId: string };
 export interface SubscriptionObserver<T> {
   snapshot(snapshot: T): void;
   update(message: SessionOutboundMessage): void;
-  /** A binary frame the host addressed to this subscription; the payload is a view. */
-  binary?(payload: Uint8Array): void;
   error?(error: unknown): void;
 }
 
@@ -22,12 +20,9 @@ interface Registration {
   start(): void;
   disconnected(): void;
   receive(message: SessionOutboundMessage): void;
-  receiveBinary(payload: Uint8Array): void;
   release(): Promise<void>;
 }
 
-type Update = { message: SessionOutboundMessage } | { binary: Uint8Array };
-
 export interface OwnedSubscriptionsHost {
   release(subscriptionId: string): Promise<void>;
   failed(error: unknown): void;
@@ -55,7 +50,7 @@ export class OwnedSubscriptions {
     let teardown: Promise<void> | null = null;
     let disposed = false;
     let overflow = false;
-    const buffered: Update[] = [];
+    const buffered: SessionOutboundMessage[] = [];
     const observers = new Set<SubscriptionObserver<SubscriptionSnapshot<T>>>();
     let resolve!: (snapshot: SubscriptionSnapshot<T>) => void;
     let reject!: (error: unknown) => void;
@@ -65,10 +60,12 @@ export class OwnedSubscriptions {
     });
     void ready.catch(() => {});
 
-    const deliver = (observer: SubscriptionObserver<SubscriptionSnapshot<T>>, update?: Update) => {
+    const deliver = (
+      observer: SubscriptionObserver<SubscriptionSnapshot<T>>,
+      message?: SessionOutboundMessage,
+    ) => {
       try {
-        if (update && "binary" in update) observer.binary?.(update.binary);
-        else if (update) observer.update(update.message);
+        if (message) observer.update(message);
         else if (snapshot) observer.snapshot(snapshot);
       } catch (error) {
         this.host.failed(error);
@@ -81,13 +78,6 @@ export class OwnedSubscriptions {
       this.routes.delete(released);
       await this.host.release(released);
     };
-    const buffer = (update: Update) => {
-      if (buffered.length < MAX_BOOTSTRAP_UPDATES && !overflow) buffered.push(update);
-      else {
-        buffered.length = 0;
-        overflow = true;
-      }
-    };
     const registration: Registration = {
       start: () => {
         if (disposed || pending || id || !this.connected) return;
@@ -138,14 +128,12 @@ export class OwnedSubscriptions {
       },
       receive: (message) => {
         if (disposed || ("payload" in message && Object.is(message.payload, snapshot))) return;
-        buffer({ message });
-        for (const observer of observers) deliver(observer, { message });
-      },
-      receiveBinary: (payload) => {
-        if (disposed) return;
-        // Bytes wait only for the first observer; retaining them after would hold every frame.
-        if (observers.size === 0) buffer({ binary: payload });
-        for (const observer of observers) deliver(observer, { binary: payload });
+        if (buffered.length < MAX_BOOTSTRAP_UPDATES && !overflow) buffered.push(message);
+        else {
+          buffered.length = 0;
+          overflow = true;
+        }
+        for (const observer of observers) deliver(observer, message);
       },
       release: () => {
         if (teardown) return teardown;
@@ -154,9 +142,7 @@ export class OwnedSubscriptions {
         observers.clear();
         buffered.length = 0;
         if (id) this.routes.delete(id);
-        // A host that already accepted keeps `ready` resolved, even if released in the same tick.
-        if (snapshot) resolve(snapshot);
-        else reject(new Error("Subscription released"));
+        reject(new Error("Subscription released"));
         teardown = (async () => {
           await pending;
           await releaseId();
@@ -188,7 +174,7 @@ export class OwnedSubscriptions {
             .catch(this.host.failed);
         } else {
           deliver(observer);
-          for (const update of buffered) deliver(observer, update);
+          for (const message of buffered) deliver(observer, message);
         }
         return () => {
           observers.delete(observer);
@@ -219,10 +205,6 @@ export class OwnedSubscriptions {
       this.routes.get(payload.subscriptionId)?.receive(message);
   }
 
-  receiveBinary(subscriptionId: string, payload: Uint8Array): void {
-    this.routes.get(subscriptionId)?.receiveBinary(payload);
-  }
-
   restore(): void {
     this.connected = true;
     for (const registration of this.registrations) registration.start();

packages/client/src/daemon-client.test.ts

@@ -1106,7 +1106,6 @@ test("advertises client capabilities in hello", async () => {
       terminal_chunked_replay: true,
       timeline_notifications: true,
       plugin_timeline_items: true,
-      plugin_binary_payloads: true,
       workspace_setup_blocked: true,
       workspace_setup_observation: true,
       hello_rejection: true,

packages/client/src/daemon-client.ts

@@ -164,11 +164,6 @@ import {
   FileTransferOpcode,
   TerminalStreamOpcode,
   type FileTransferFrame,
-  assertPluginWireValue,
-  decodePluginFrame,
-  encodePluginInputFrame,
-  isPluginFrameOpcode,
-  PluginFrameOpcode,
 } from "@getpaseo/protocol/binary-frames/index";
 import {
   createRelayE2eeTransportFactory,
@@ -2265,8 +2260,6 @@ export class DaemonClient {
     timeout?: number;
     select: (msg: SessionOutboundMessage) => T | null;
     options?: { skipQueue?: boolean };
-    /** Sends the request in another form, such as a binary frame; defaults to `message`. */
-    transmit?: () => Promise<void>;
   }): Promise<T> {
     const wire = this.owned.prepareRequest(params.message);
     const timeout = params.timeout ?? DEFAULT_SESSION_RPC_TIMEOUT_MS;
@@ -2297,9 +2290,7 @@ export class DaemonClient {
     // The response deadline still applies while the transport waits for drain.
     void promise.catch(() => undefined);
     try {
-      const sent = params.transmit
-        ? params.transmit()
-        : this.sendSessionMessageOrThrow(wire.message);
+      const sent = this.sendSessionMessageOrThrow(wire.message);
       const result = await Promise.race([promise, sent.then(() => promise)]);
       if (result.kind === "error") throw result.error;
       return result.value;
@@ -2322,14 +2313,12 @@ export class DaemonClient {
     responseType: TResponseType;
     options?: { skipQueue?: boolean };
     selectPayload?: (payload: CorrelatedResponsePayload<TResponseType>) => TResult | null;
-    transmit?: () => Promise<void>;
   }): Promise<TResult> {
     return this.sendRequest({
       requestId: params.requestId,
       message: params.message,
       timeout: params.timeout,
       options: params.options,
-      ...(params.transmit ? { transmit: params.transmit } : {}),
       select: (msg) => {
         const correlated = msg as CorrelatedResponseMessage;
         if (correlated.type !== params.responseType) {
@@ -2661,7 +2650,6 @@ export class DaemonClient {
         resetSource = false;
         if (query.type === "workspace.setup.subscribe.request")
           this.requireWorkspaceSetupObservation();
-        if (query.type === "plugin.rpc.subscribe.request") this.requirePluginSubscriptions();
         const requestId = this.createRequestId(options?.requestId);
         const request = SessionInboundMessageSchema.parse({ ...query, requestId });
         try {
@@ -6209,59 +6197,22 @@ export class DaemonClient {
     return payload.plugin;
   }
 
-  /**
-   * A `Uint8Array` input or output crosses the wire as a binary frame. Any other value is JSON and
-   * must not contain a `Uint8Array`.
-   */
   async invokePluginRpc(pluginId: string, method: string, input: unknown): Promise<unknown> {
-    assertPluginWireValue(input, "input");
     const requestId = this.createRequestId();
-    if (!(input instanceof Uint8Array)) {
-      const payload = await this.sendCorrelatedSessionRequest({
-        requestId,
-        message: {
-          type: "plugin.rpc.invoke.request",
-          requestId,
-          pluginId,
-          method,
-          input,
-        },
-        responseType: "plugin.rpc.invoke.response",
-      });
-      return payload.output;
-    }
-    this.requirePluginSubscriptions();
-    const frame = encodePluginInputFrame({ requestId, pluginId, method, payload: input });
-    const payload = await this.sendCorrelatedRequest({
+    const payload = await this.sendCorrelatedSessionRequest({
       requestId,
-      message: { type: "plugin.rpc.invoke.request", requestId, pluginId, method, input: null },
+      message: {
+        type: "plugin.rpc.invoke.request",
+        requestId,
+        pluginId,
+        method,
+        input,
+      },
       responseType: "plugin.rpc.invoke.response",
-      options: { skipQueue: true },
-      transmit: () => this.sendBinaryFrame(frame),
     });
     return payload.output;
   }
 
-  /** One subscription to a plugin subscription handler; events arrive as updates and binary. */
-  observePluginRpc(
-    pluginId: string,
-    method: string,
-    input: unknown,
-    options?: { signal?: AbortSignal },
-  ): OwnedSubscription<CorrelatedResponsePayload<"plugin.rpc.subscribe.response">> {
-    assertPluginWireValue(input, "input");
-    return this.observe(
-      "plugin.rpc.subscribe.response",
-      { type: "plugin.rpc.subscribe.request", pluginId, method, input },
-      options,
-    );
-  }
-
-  private requirePluginSubscriptions(): void {
-    if (this.lastServerInfoMessage?.features?.pluginSubscriptions !== true)
-      throw new Error("Update the host to use plugin subscriptions and binary plugin RPC");
-  }
-
   async respondToPermissionAndWait(
     agentId: string,
     requestId: string,
@@ -7024,23 +6975,6 @@ export class DaemonClient {
     }
   }
 
-  private handlePluginFrame(rawBytes: Uint8Array): void {
-    const frame = decodePluginFrame(rawBytes);
-    this.consecutiveLivenessFailures = 0;
-    this.runtimeMetrics?.recordBinaryFrame("plugin", rawBytes.byteLength, 0);
-    if (frame?.opcode === PluginFrameOpcode.Event) {
-      this.owned.receiveBinary(frame.subscriptionId, frame.payload);
-      return;
-    }
-    if (frame?.opcode === PluginFrameOpcode.Output) {
-      // The binary form of plugin.rpc.invoke.response.
-      this.resolveWaiters({
-        type: "plugin.rpc.invoke.response",
-        payload: { requestId: frame.requestId, output: frame.payload },
-      });
-    }
-  }
-
   private tryHandleBinaryFrame(rawBytes: Uint8Array): boolean {
     const hostFrame = decodeHostRouteFrame(rawBytes);
     if (rawBytes[0] === HOST_ROUTE_OPCODE && !hostFrame) {
@@ -7052,10 +6986,6 @@ export class DaemonClient {
       for (const handler of this.hostFrameListeners) handler(hostFrame);
       return true;
     }
-    if (isPluginFrameOpcode(rawBytes[0])) {
-      this.handlePluginFrame(rawBytes);
-      return true;
-    }
     const fileFrame = decodeFileTransferFrame(rawBytes);
     if (fileFrame) {
       this.traceInstant("paseo.ws.message.inbound", {

packages/plugin/src/boundaries.test.ts

@@ -323,68 +323,3 @@ describe("agent session reload SDK boundary", () => {
     ).toEqual([]);
   });
 });
-
-describe("plugin subscription SDK boundary", () => {
-  it("types contracts, overloaded handlers, the client hook, and the client entry handle", () => {
-    expect(
-      typeErrors(`
-      import { defineRpc, defineSubscription, type SubscriptionEvent, type SubscriptionInput } from "@getpaseo/plugin";
-      import type { PluginClientContext } from "@getpaseo/plugin/client";
-      import { useRpc, useSubscription } from "@getpaseo/plugin/client";
-      import type { PluginServerContext, PluginSubscriptionContext } from "@getpaseo/plugin/server";
-      import { z } from "zod";
-      const diffWatch = defineSubscription({
-        name: "diff.watch",
-        input: z.object({ a: z.string(), b: z.string() }),
-        event: z.object({ path: z.string(), status: z.enum(["added", "modified", "deleted"]) }),
-      });
-      const frames = defineSubscription({
-        name: "screen.capture",
-        input: z.object({ display: z.number() }),
-        event: z.union([z.instanceof(Uint8Array), z.object({ type: z.literal("resized"), width: z.number() })]),
-      });
-      const thumbnail = defineRpc({ name: "thumbnail.render", input: z.object({ path: z.string() }), output: z.instanceof(Uint8Array) });
-      const input: SubscriptionInput<typeof diffWatch> = { a: "main", b: "HEAD" };
-      const event: SubscriptionEvent<typeof diffWatch> = { path: "a.ts", status: "added" };
-      function server(server: PluginServerContext) {
-        server.handle(diffWatch, ({ a, b }, { push, signal, paseo }: PluginSubscriptionContext<SubscriptionEvent<typeof diffWatch>>) => {
-          const pushed: Promise<void> = push({ path: a + b, status: "modified" });
-          // @ts-expect-error events follow the contract
-          void push({ path: 1, status: "modified" });
-          signal.addEventListener("abort", () => undefined, { once: true });
-        });
-        server.handle(frames, async (_input, { push }) => {
-          await push(new Uint8Array(4));
-          await push({ type: "resized", width: 2 });
-        });
-        server.handle(thumbnail, ({ path }) => new Uint8Array(path.length));
-        // @ts-expect-error subscription handlers receive push, not an output
-        server.handle(diffWatch, () => ({ path: "a", status: "added" as const }) as unknown as number);
-      }
-      function Panel() {
-        const state = useSubscription(diffWatch, { a: "main", b: "HEAD" }, (change) => {
-          const status: "added" | "modified" | "deleted" = change.status;
-        });
-        if (state.status === "error") state.error.message;
-        // @ts-expect-error the hook has no ready state, only live
-        if (state.status === "ready") return null;
-        useSubscription(frames, { display: 1 }, (event) => {
-          if (event instanceof Uint8Array) event.byteLength;
-          else event.width;
-        });
-        const render = useRpc(thumbnail);
-        void render({ path: "x" }).then((png: Uint8Array) => png.byteLength);
-        return null;
-      }
-      function client(client: PluginClientContext) {
-        const controller = new AbortController();
-        const remover = client.subscribe(diffWatch, { a: "main", b: "HEAD" }, (change) => change.path, { signal: controller.signal });
-        const ready: Promise<void> = remover.ready;
-        const released: Promise<void> = remover.release();
-        remover();
-        return () => {};
-      }
-    `),
-    ).toEqual([]);
-  });
-});

packages/plugin/src/client/contracts.ts

@@ -4,13 +4,7 @@ import type { PaseoApi } from "@getpaseo/client";
 import type { AgentTimelineItem, JsonValue } from "@getpaseo/protocol/agent-types";
 import type { ZodType, input as ZodInput, output as ZodOutput } from "zod";
 import type { PluginFile } from "./file.js";
-import type {
-  PluginRpcContract,
-  PluginSubscribeOptions,
-  PluginSubscriptionContract,
-  PluginSubscriptionHandle,
-  PluginWireInput,
-} from "../rpc.js";
+import type { PluginRpcContract } from "../rpc.js";
 import type {
   PluginButtonRegistration,
   PluginHeaderButtonContribution,
@@ -209,16 +203,6 @@ export interface PluginClientContext extends PluginCommandCapabilities, PluginCl
     contribution: PluginTimelineRendererContribution<Schema>,
   ): PluginCleanup;
   openPanel(id: string, options: PluginClientOpenPanelOptions): void;
-  /**
-   * Subscribes to a plugin subscription on this client's host. Paseo re-subscribes after a
-   * reconnect and releases what is left when the plugin stops.
-   */
-  subscribe<InputSchema extends ZodType, EventSchema extends ZodType>(
-    contract: PluginSubscriptionContract<InputSchema, EventSchema>,
-    input: PluginWireInput<ZodInput<InputSchema>>,
-    onEvent: (event: ZodOutput<EventSchema>) => void,
-    options?: PluginSubscribeOptions,
-  ): PluginSubscriptionHandle;
 }
 
 export type PluginClientContribution = (client: PluginClientContext) => PluginCleanup;
@@ -311,7 +295,7 @@ export interface PluginCommandCapabilities {
   paseo: PaseoApi;
   rpc<InputSchema extends ZodType, OutputSchema extends ZodType>(
     contract: PluginRpcContract<InputSchema, OutputSchema>,
-    input: PluginWireInput<ZodInput<InputSchema>>,
+    input: ZodInput<InputSchema>,
   ): Promise<ZodOutput<OutputSchema>>;
   openScreen(input: PluginOpenScreenInput): void;
   /** @deprecated Use `openScreen`. */

packages/plugin/src/client/host.test.ts

@@ -24,23 +24,17 @@ describe("plugin RPC contracts", () => {
     expectTypeOf<RpcOutput<typeof contract>>().toEqualTypeOf<{ value: number }>();
   });
 
-  it("sends input unchanged for the plugin to parse", async () => {
-    let parses = 0;
-    const transformed = defineRpc({
-      name: "transformed",
-      input: z.string().transform((text) => {
-        parses += 1;
-        return Number(text);
-      }),
-      output: z.number(),
-    });
-    const invoke = async (_method: string, input: unknown) => {
-      expect(input).toBe("2");
-      expect(parses).toBe(0);
-      return transformed.input.parseAsync(input);
+  it("validates input before invoking the daemon", async () => {
+    let invocations = 0;
+    const invoke = async () => {
+      invocations += 1;
+      return { value: 2 };
     };
-    await expect(callPluginRpc(transformed, invoke, "2")).resolves.toBe(2);
-    expect(parses).toBe(1);
+
+    await expect(callPluginRpc(contract, invoke, { value: "wrong" })).rejects.toBeInstanceOf(
+      z.ZodError,
+    );
+    expect(invocations).toBe(0);
   });
 
   it("validates output before resolving", async () => {

packages/plugin/src/client/host.ts

@@ -8,7 +8,7 @@ export {
 import type { PluginAttachmentSourceContribution } from "../contracts.js";
 import { PluginRpcProvider } from "./rpc-context.js";
 import { PaseoApiProvider } from "./paseo-context.js";
-import { callPluginRpc, subscribePluginRpc, type PluginSubscriptionObserver } from "../rpc.js";
+import { callPluginRpc } from "../rpc.js";
 
 export async function searchPluginAttachments(
   source: PluginAttachmentSourceContribution<PluginText>,
@@ -19,9 +19,7 @@ export async function searchPluginAttachments(
   return PluginAttachmentSearchPayloadSchema.parseAsync(output);
 }
 
-export { callPluginRpc, subscribePluginRpc, PaseoApiProvider, PluginRpcProvider };
-export type { PluginSubscriptionObserver };
-export type { PluginSubscribe } from "./rpc-context.js";
+export { callPluginRpc, PaseoApiProvider, PluginRpcProvider };
 
 export {
   PluginTranslationRuntime,

packages/plugin/src/client/index.ts

@@ -71,12 +71,7 @@ export {
 } from "./file.js";
 export { usePaseo } from "./paseo-context.js";
 export { useAgent, useWorkspace } from "./client-state.js";
-export { useRpc, useSubscription, type PluginSubscriptionState } from "./rpc-context.js";
-export type {
-  PluginSubscribeOptions,
-  PluginSubscriptionClosed,
-  PluginSubscriptionHandle,
-} from "../rpc.js";
+export { useRpc } from "./rpc-context.js";
 import type { SettingsDefinition } from "../settings.js";
 import type { SettingsState } from "./contracts.js";
 import type { ZodType } from "zod";

packages/plugin/src/client/rpc-context.tsx

@@ -1,34 +1,9 @@
-import {
-  createContext,
-  useCallback,
-  useContext,
-  useEffect,
-  useMemo,
-  useRef,
-  useState,
-  type ReactNode,
-} from "react";
+import { createContext, useCallback, useContext, useMemo, type ReactNode } from "react";
 import type { ZodType, input as ZodInput, output as ZodOutput } from "zod";
-import {
-  callPluginRpc,
-  subscribePluginRpc,
-  type PluginRpcContract,
-  type PluginSubscriptionContract,
-  type PluginSubscriptionHandle,
-  type PluginSubscriptionObserver,
-  type PluginWireInput,
-} from "../rpc.js";
-import { shallowEqual } from "./shallow.js";
-
-export type PluginSubscribe = (
-  method: string,
-  input: unknown,
-  observer: PluginSubscriptionObserver,
-) => PluginSubscriptionHandle;
+import { callPluginRpc, type PluginRpcContract } from "../rpc.js";
 
 export interface PluginRpcContextValue {
   invoke(method: string, input: unknown): Promise<unknown>;
-  subscribe: PluginSubscribe;
 }
 
 const PluginRpcContext = createContext<PluginRpcContextValue | null>(null);
@@ -40,100 +15,22 @@ export function usePluginRpcContextValue(): PluginRpcContextValue | null {
 export function PluginRpcProvider({
   children,
   invoke,
-  subscribe,
 }: {
   children: ReactNode;
   invoke(method: string, input: unknown): Promise<unknown>;
-  subscribe: PluginSubscribe;
 }) {
-  const value = useMemo(() => ({ invoke, subscribe }), [invoke, subscribe]);
+  const value = useMemo(() => ({ invoke }), [invoke]);
   return <PluginRpcContext.Provider value={value}>{children}</PluginRpcContext.Provider>;
 }
 
 export function useRpc<InputSchema extends ZodType, OutputSchema extends ZodType>(
   contract: PluginRpcContract<InputSchema, OutputSchema>,
-): (input: PluginWireInput<ZodInput<InputSchema>>) => Promise<ZodOutput<OutputSchema>> {
+): (input: ZodInput<InputSchema>) => Promise<ZodOutput<OutputSchema>> {
   const context = usePluginRpcContextValue();
   if (!context) throw new Error("useRpc must run inside a contributed plugin surface");
 
   return useCallback(
-    (input: PluginWireInput<ZodInput<InputSchema>>) =>
-      callPluginRpc(contract, context.invoke, input),
+    (input: ZodInput<InputSchema>) => callPluginRpc(contract, context.invoke, input),
     [context, contract],
   );
 }
-
-export type PluginSubscriptionState =
-  | { status: "connecting" }
-  | { status: "live" }
-  /** The plugin's handler finished. The hook subscribes again when `input` changes. */
-  | { status: "closed" }
-  | { status: "error"; error: Error };
-
-/**
- * Subscribes while the component is mounted. Re-subscribes when `input` changes by shallow
- * comparison; `onEvent` always runs the latest callback.
- */
-export function useSubscription<InputSchema extends ZodType, EventSchema extends ZodType>(
-  contract: PluginSubscriptionContract<InputSchema, EventSchema>,
-  input: PluginWireInput<ZodInput<InputSchema>>,
-  onEvent: (event: ZodOutput<EventSchema>) => void,
-): PluginSubscriptionState {
-  const context = usePluginRpcContextValue();
-  if (!context) throw new Error("useSubscription must run inside a contributed plugin surface");
-  const subscribe = context.subscribe;
-  const onEventRef = useRef(onEvent);
-  onEventRef.current = onEvent;
-  const stableInput = useShallowStable(input);
-  const [state, setState] = useState<PluginSubscriptionState>({ status: "connecting" });
-
-  useEffect(() => {
-    let active = true;
-    setState({ status: "connecting" });
-    let handle: PluginSubscriptionHandle;
-    try {
-      handle = subscribePluginRpc(
-        contract,
-        subscribe,
-        stableInput,
-        (event) => onEventRef.current(event),
-        {
-          onClosed: ({ error }) => {
-            if (active) setState(error ? { status: "error", error } : CLOSED);
-          },
-        },
-      );
-    } catch (error) {
-      setState({ status: "error", error: toError(error) });
-      return;
-    }
-    handle.ready.then(
-      () => {
-        if (active) setState((current) => (current.status === "connecting" ? LIVE : current));
-        return undefined;
-      },
-      (error: unknown) => {
-        if (active) setState({ status: "error", error: toError(error) });
-      },
-    );
-    return () => {
-      active = false;
-      handle();
-    };
-  }, [contract, subscribe, stableInput]);
-
-  return state;
-}
-
-const LIVE: PluginSubscriptionState = { status: "live" };
-const CLOSED: PluginSubscriptionState = { status: "closed" };
-
-function useShallowStable<Value>(value: Value): Value {
-  const ref = useRef(value);
-  if (!shallowEqual(ref.current, value)) ref.current = value;
-  return ref.current;
-}
-
-function toError(error: unknown): Error {
-  return error instanceof Error ? error : new Error(String(error));
-}

packages/plugin/src/client/runtime-context-bridge.tsx

@@ -26,7 +26,7 @@ export function usePluginRuntimeContextBridge(): PluginRuntimeContextBridge {
       );
       return (
         <PaseoApiProvider paseo={paseo}>
-          <PluginRpcProvider invoke={rpc.invoke} subscribe={rpc.subscribe}>
+          <PluginRpcProvider invoke={rpc.invoke}>
             <PluginTranslationProvider runtime={translations}>{content}</PluginTranslationProvider>
           </PluginRpcProvider>
         </PaseoApiProvider>

packages/plugin/src/index.ts

@@ -19,16 +19,7 @@ export {
   type PluginAttachmentItem,
   type PluginAttachmentSearchPayload,
 } from "./attachments.js";
-export {
-  defineRpc,
-  defineSubscription,
-  type PluginRpcContract,
-  type PluginSubscriptionContract,
-  type RpcInput,
-  type RpcOutput,
-  type SubscriptionEvent,
-  type SubscriptionInput,
-} from "./rpc.js";
+export { defineRpc, type PluginRpcContract, type RpcInput, type RpcOutput } from "./rpc.js";
 
 export type { PluginText, PluginMessage } from "./text.js";
 export {

packages/plugin/src/rpc.ts

@@ -9,12 +9,6 @@ export interface PluginRpcContract<
   output: OutputSchema;
 }
 
-/**
- * `z.instanceof(Uint8Array)` infers `Uint8Array<ArrayBuffer>`; values crossing the wire accept any
- * `Uint8Array`, including a `Buffer` or a view of a shared buffer.
- */
-export type PluginWireInput<Value> = Value extends Uint8Array ? Uint8Array : Value;
-
 export type RpcInput<Contract extends PluginRpcContract> = ZodOutput<Contract["input"]>;
 export type RpcOutput<Contract extends PluginRpcContract> = ZodInput<Contract["output"]>;
 
@@ -37,98 +31,9 @@ export function defineRpc<InputSchema extends ZodType, OutputSchema extends ZodT
 export async function callPluginRpc<InputSchema extends ZodType, OutputSchema extends ZodType>(
   contract: PluginRpcContract<InputSchema, OutputSchema>,
   invoke: (method: string, input: unknown) => Promise<unknown>,
-  input: PluginWireInput<ZodInput<InputSchema>>,
+  input: ZodInput<InputSchema>,
 ): Promise<ZodOutput<OutputSchema>> {
-  const output = await invoke(contract.name, input);
+  const parsedInput = await contract.input.parseAsync(input);
+  const output = await invoke(contract.name, parsedInput);
   return contract.output.parseAsync(output);
 }
-
-export interface PluginSubscriptionContract<
-  InputSchema extends ZodType = ZodType,
-  EventSchema extends ZodType = ZodType,
-> {
-  name: string;
-  input: InputSchema;
-  event: EventSchema;
-}
-
-export type SubscriptionInput<Contract extends PluginSubscriptionContract> = ZodOutput<
-  Contract["input"]
->;
-export type SubscriptionEvent<Contract extends PluginSubscriptionContract> = ZodOutput<
-  Contract["event"]
->;
-
-interface SubscriptionDefinition<InputSchema extends ZodType, EventSchema extends ZodType> {
-  name: string;
-  input: InputSchema;
-  event: EventSchema;
-}
-
-export function defineSubscription<InputSchema extends ZodType, EventSchema extends ZodType>(
-  definition: SubscriptionDefinition<InputSchema, EventSchema>,
-): PluginSubscriptionContract<InputSchema, EventSchema> {
-  const name = definition.name.trim();
-  if (!RPC_NAME.test(name)) throw new Error(`Invalid plugin subscription: ${definition.name}`);
-  return { ...definition, name };
-}
-
-/** One subscription from a client surface. Calling it or `release()` ends it. */
-export type PluginSubscriptionHandle = (() => void) & {
-  /** Resolves once the plugin acknowledged the subscription; rejects if it refused. */
-  readonly ready: Promise<void>;
-  release(): Promise<void>;
-};
-
-export interface PluginSubscriptionObserver {
-  event(event: unknown): void;
-  closed(error?: Error): void;
-}
-
-export interface PluginSubscriptionClosed {
-  /** Set when the plugin's handler failed, the plugin stopped, or an event failed validation. */
-  error?: Error;
-}
-
-export interface PluginSubscribeOptions {
-  /** Aborting releases the subscription. */
-  signal?: AbortSignal;
-  /** Runs once when the daemon closes the subscription. Not called after a release. */
-  onClosed?: (closed: PluginSubscriptionClosed) => void;
-}
-
-export function subscribePluginRpc<InputSchema extends ZodType, EventSchema extends ZodType>(
-  contract: PluginSubscriptionContract<InputSchema, EventSchema>,
-  subscribe: (
-    method: string,
-    input: unknown,
-    observer: PluginSubscriptionObserver,
-  ) => PluginSubscriptionHandle,
-  input: PluginWireInput<ZodInput<InputSchema>>,
-  onEvent: (event: ZodOutput<EventSchema>) => void,
-  options: PluginSubscribeOptions = {},
-): PluginSubscriptionHandle {
-  // The plugin parses the input into the schema's output, so it receives the input as given.
-  contract.input.parse(input);
-  let closed = false;
-  function close(error?: Error) {
-    if (closed) return;
-    closed = true;
-    options.onClosed?.(error ? { error } : {});
-  }
-  const handle = subscribe(contract.name, input, {
-    event(event) {
-      if (closed) return;
-      const parsed = contract.event.safeParse(event);
-      if (parsed.success) return onEvent(parsed.data);
-      handle();
-      close(new Error(`Plugin subscription ${contract.name} sent an invalid event`));
-    },
-    closed: close,
-  });
-  if (options.signal) {
-    if (options.signal.aborted) handle();
-    else options.signal.addEventListener("abort", () => handle(), { once: true });
-  }
-  return handle;
-}

packages/plugin/src/server/contracts.ts

@@ -1,6 +1,6 @@
 import type { PaseoApi } from "@getpaseo/client";
 import type { ZodType, input as ZodInput, output as ZodOutput } from "zod";
-import type { PluginRpcContract, PluginSubscriptionContract, PluginWireInput } from "../rpc.js";
+import type { PluginRpcContract } from "../rpc.js";
 import type { PluginCleanup } from "../contracts.js";
 import type { ProviderRegistration } from "./provider.js";
 import type { UsageSourceRegistration } from "./usage.js";
@@ -10,18 +10,6 @@ export interface PluginHandlerContext {
   paseo: PaseoApi;
 }
 
-/** What a subscription handler receives beside its input. */
-export interface PluginSubscriptionContext<Event = unknown> extends PluginHandlerContext {
-  /**
-   * Sends one event to this subscriber. A `Uint8Array` event travels as a binary frame; any other
-   * event is JSON and must not contain a `Uint8Array`. Resolves once the event left the plugin
-   * process. Events pushed after `signal` aborts are dropped.
-   */
-  push(event: Event): Promise<void>;
-  /** Aborts when the subscriber releases, disconnects, or the plugin stops. */
-  signal: AbortSignal;
-}
-
 export type PluginSettingsState<Schema extends ZodType> =
   | {
       status: "ready";
@@ -43,24 +31,12 @@ export interface PluginServerContext extends PluginLifecycleRegistration {
   registerSettings<Schema extends ZodType>(
     definition: import("../settings.js").SettingsDefinition<Schema>,
   ): PluginSettings<Schema>;
-  /**
-   * Serves a subscription. The handler starts the work and returns; the subscription stays open
-   * until `signal` aborts. A handler that throws refuses the subscription. When it returns a
-   * promise, settling the promise closes the subscription, with the rejection as its error.
-   */
-  handle<InputSchema extends ZodType, EventSchema extends ZodType>(
-    contract: PluginSubscriptionContract<InputSchema, EventSchema>,
-    handler: (
-      input: ZodOutput<InputSchema>,
-      context: PluginSubscriptionContext<PluginWireInput<ZodInput<EventSchema>>>,
-    ) => void | Promise<void>,
-  ): void;
   handle<InputSchema extends ZodType, OutputSchema extends ZodType>(
     contract: PluginRpcContract<InputSchema, OutputSchema>,
     handler: (
       input: ZodOutput<InputSchema>,
       context: PluginHandlerContext,
-    ) => PluginWireInput<ZodInput<OutputSchema>> | Promise<PluginWireInput<ZodInput<OutputSchema>>>,
+    ) => ZodInput<OutputSchema> | Promise<ZodInput<OutputSchema>>,
   ): void;
   registerProvider(provider: ProviderRegistration): void;
   registerUsageSource(source: UsageSourceRegistration): void;

packages/plugin/src/server/index.ts

@@ -11,7 +11,6 @@ export type {
   PluginServerContribution,
   PluginSettings,
   PluginSettingsState,
-  PluginSubscriptionContext,
 } from "./contracts.js";
 export type {
   PluginHooks,

packages/protocol/src/binary-frames/demux.ts

@@ -4,7 +4,6 @@ import {
   FileTransferOpcode,
   type FileTransferFrame,
 } from "./file-transfer.js";
-import { decodePluginFrame, PluginFrameOpcode, type PluginFrame } from "./plugin.js";
 import {
   decodeTerminalStreamFrame,
   TerminalStreamOpcode,
@@ -14,8 +13,7 @@ import {
 export type BinaryFrame =
   | { kind: "host_route"; frame: HostRouteFrame }
   | { kind: "terminal"; frame: TerminalStreamFrame }
-  | { kind: "file_transfer"; frame: FileTransferFrame }
-  | { kind: "plugin"; frame: PluginFrame };
+  | { kind: "file_transfer"; frame: FileTransferFrame };
 
 export function decodeBinaryFrame(bytes: Uint8Array): BinaryFrame | null {
   switch (bytes[0]) {
@@ -40,12 +38,6 @@ export function decodeBinaryFrame(bytes: Uint8Array): BinaryFrame | null {
       const frame = decodeFileTransferFrame(bytes);
       return frame ? { kind: "file_transfer", frame } : null;
     }
-    case PluginFrameOpcode.Event:
-    case PluginFrameOpcode.Output:
-    case PluginFrameOpcode.Input: {
-      const frame = decodePluginFrame(bytes);
-      return frame ? { kind: "plugin", frame } : null;
-    }
     default:
       return null;
   }

packages/protocol/src/binary-frames/index.ts

@@ -2,4 +2,3 @@ export * from "./demux.js";
 export * from "./file-transfer.js";
 export * from "./terminal.js";
 export * from "./host-route.js";
-export * from "./plugin.js";

packages/protocol/src/binary-frames/plugin.test.ts

@@ -1,95 +0,0 @@
-import { describe, expect, it } from "vitest";
-
-import {
-  assertPluginWireValue,
-  decodeBinaryFrame,
-  decodePluginFrame,
-  encodePluginEventFrame,
-  encodePluginInputFrame,
-  encodePluginOutputFrame,
-  FileTransferOpcode,
-  MAX_PLUGIN_PAYLOAD_BYTES,
-  MAX_ROUTED_PAYLOAD_BYTES,
-  PluginFrameOpcode,
-  TerminalStreamOpcode,
-} from "./index.js";
-
-describe("plugin binary frames", () => {
-  it("uses the reserved plugin opcode range", () => {
-    for (const opcode of Object.values(PluginFrameOpcode)) {
-      expect(opcode).toBeGreaterThanOrEqual(0x20);
-      expect(opcode).toBeLessThanOrEqual(0x2f);
-      expect(Object.values(TerminalStreamOpcode)).not.toContain(opcode);
-      expect(Object.values(FileTransferOpcode)).not.toContain(opcode);
-    }
-  });
-
-  it("frames a subscription event as opcode, id length, id, and the untouched payload", () => {
-    const payload = new Uint8Array([9, 8, 7]);
-    const frame = encodePluginEventFrame("sub-1", payload);
-    expect([...frame]).toEqual([0x20, 5, ...new TextEncoder().encode("sub-1"), 9, 8, 7]);
-    expect(decodeBinaryFrame(frame)).toEqual({
-      kind: "plugin",
-      frame: { opcode: PluginFrameOpcode.Event, subscriptionId: "sub-1", payload },
-    });
-  });
-
-  it("decodes payloads as views into the received buffer", () => {
-    const frame = encodePluginOutputFrame("req-1", new Uint8Array([1, 2, 3, 4]));
-    const decoded = decodePluginFrame(frame);
-    if (decoded?.opcode !== PluginFrameOpcode.Output) throw new Error("Expected an output frame");
-    expect(decoded.requestId).toBe("req-1");
-    expect(decoded.payload.buffer).toBe(frame.buffer);
-    expect([...decoded.payload]).toEqual([1, 2, 3, 4]);
-  });
-
-  it("frames an RPC input with the plugin and method in its header", () => {
-    const frame = encodePluginInputFrame({
-      requestId: "req-2",
-      pluginId: "uploads",
-      method: "upload.store",
-      payload: new Uint8Array([5, 6]),
-    });
-    expect(decodePluginFrame(frame)).toEqual({
-      opcode: PluginFrameOpcode.Input,
-      requestId: "req-2",
-      pluginId: "uploads",
-      method: "upload.store",
-      payload: new Uint8Array([5, 6]),
-    });
-  });
-
-  it("rejects truncated frames", () => {
-    expect(decodePluginFrame(new Uint8Array([0x20]))).toBeNull();
-    expect(decodePluginFrame(new Uint8Array([0x20, 9, 1]))).toBeNull();
-    expect(decodePluginFrame(new Uint8Array([0x22, 1, 65, 0, 40]))).toBeNull();
-  });
-});
-
-describe("the plugin byte rule", () => {
-  it("accepts JSON values and a top-level Uint8Array", () => {
-    expect(() => assertPluginWireValue({ a: [1, "b", null] }, "event")).not.toThrow();
-    expect(() => assertPluginWireValue(new Uint8Array(4), "event")).not.toThrow();
-  });
-
-  it("rejects bytes that would not fit in one frame with the largest header", () => {
-    const bytes = new Uint8Array(MAX_PLUGIN_PAYLOAD_BYTES + 1);
-    expect(() => assertPluginWireValue(bytes.subarray(1), "event")).not.toThrow();
-    expect(() => assertPluginWireValue(bytes, "event")).toThrow(
-      `event is ${MAX_PLUGIN_PAYLOAD_BYTES + 1} bytes; one value carries at most ${MAX_PLUGIN_PAYLOAD_BYTES} bytes`,
-    );
-    const frame = encodePluginInputFrame({
-      requestId: "r".repeat(255),
-      pluginId: "p",
-      method: "m".repeat(0xffff - 30),
-      payload: bytes.subarray(1),
-    });
-    expect(frame.byteLength).toBeLessThanOrEqual(MAX_ROUTED_PAYLOAD_BYTES);
-  });
-
-  it("rejects bytes nested in a value and names the rule and the path", () => {
-    expect(() => assertPluginWireValue({ frames: [{ data: new Uint8Array(1) }] }, "event")).toThrow(
-      "event.frames[0].data is a Uint8Array. Bytes cross the wire only as the whole value: send the Uint8Array itself, or encode it into the value.",
-    );
-  });
-});

packages/protocol/src/binary-frames/plugin.ts

@@ -1,152 +0,0 @@
-import { MAX_ROUTED_PAYLOAD_BYTES } from "./host-route.js";
-
-/**
- * Plugin values that are a `Uint8Array` travel as binary frames. Opcodes 0x20-0x2f are
- * reserved for plugin traffic. Every frame is `[opcode][id length u8][id]...` and the payload
- * is the rest of the frame, decoded as a view so no hop copies it.
- */
-export const PluginFrameOpcode = {
-  /** `[0x20][id length][subscriptionId][payload]`: a subscription event. */
-  Event: 0x20,
-  /** `[0x21][id length][requestId][payload]`: an RPC output; no JSON response follows. */
-  Output: 0x21,
-  /** `[0x22][id length][requestId][header length u16][header JSON][payload]`: an RPC input. */
-  Input: 0x22,
-} as const;
-
-export type PluginFrameOpcode = (typeof PluginFrameOpcode)[keyof typeof PluginFrameOpcode];
-
-/** The largest frame header: an input's opcode, ID and header JSON, each at its maximum length. */
-const MAX_PLUGIN_FRAME_HEADER_BYTES = 2 + 0xff + 2 + 0xffff;
-
-/** The largest `Uint8Array` a plugin value carries, so its frame fits in one routed message. */
-export const MAX_PLUGIN_PAYLOAD_BYTES = MAX_ROUTED_PAYLOAD_BYTES - MAX_PLUGIN_FRAME_HEADER_BYTES;
-
-export type PluginFrame =
-  | { opcode: typeof PluginFrameOpcode.Event; subscriptionId: string; payload: Uint8Array }
-  | { opcode: typeof PluginFrameOpcode.Output; requestId: string; payload: Uint8Array }
-  | {
-      opcode: typeof PluginFrameOpcode.Input;
-      requestId: string;
-      pluginId: string;
-      method: string;
-      payload: Uint8Array;
-    };
-
-export function isPluginFrameOpcode(value: number | undefined): value is PluginFrameOpcode {
-  return (
-    value === PluginFrameOpcode.Event ||
-    value === PluginFrameOpcode.Output ||
-    value === PluginFrameOpcode.Input
-  );
-}
-
-export function encodePluginEventFrame(subscriptionId: string, payload: Uint8Array): Uint8Array {
-  return encodeFrame(PluginFrameOpcode.Event, subscriptionId, new Uint8Array(0), payload);
-}
-
-export function encodePluginOutputFrame(requestId: string, payload: Uint8Array): Uint8Array {
-  return encodeFrame(PluginFrameOpcode.Output, requestId, new Uint8Array(0), payload);
-}
-
-export function encodePluginInputFrame(input: {
-  requestId: string;
-  pluginId: string;
-  method: string;
-  payload: Uint8Array;
-}): Uint8Array {
-  const header = new TextEncoder().encode(
-    JSON.stringify({ pluginId: input.pluginId, method: input.method }),
-  );
-  if (header.byteLength > 0xffff) throw new RangeError("Plugin input header is too long");
-  const prefix = new Uint8Array(2 + header.byteLength);
-  new DataView(prefix.buffer).setUint16(0, header.byteLength);
-  prefix.set(header, 2);
-  return encodeFrame(PluginFrameOpcode.Input, input.requestId, prefix, input.payload);
-}
-
-export function decodePluginFrame(bytes: Uint8Array): PluginFrame | null {
-  const opcode = bytes[0];
-  if (bytes.byteLength < 2 || !isPluginFrameOpcode(opcode)) return null;
-  const idLength = bytes[1];
-  if (idLength === 0 || idLength > bytes.byteLength - 2) return null;
-  const id = new TextDecoder().decode(bytes.subarray(2, 2 + idLength));
-  const body = bytes.subarray(2 + idLength);
-  if (opcode === PluginFrameOpcode.Event) return { opcode, subscriptionId: id, payload: body };
-  if (opcode === PluginFrameOpcode.Output) return { opcode, requestId: id, payload: body };
-  if (body.byteLength < 2) return null;
-  const headerLength = new DataView(body.buffer, body.byteOffset, body.byteLength).getUint16(0);
-  if (headerLength > body.byteLength - 2) return null;
-  const header = decodeInputHeader(body.subarray(2, 2 + headerLength));
-  if (!header) return null;
-  return { opcode, requestId: id, ...header, payload: body.subarray(2 + headerLength) };
-}
-
-function encodeFrame(
-  opcode: PluginFrameOpcode,
-  id: string,
-  prefix: Uint8Array,
-  payload: Uint8Array,
-): Uint8Array {
-  const idBytes = new TextEncoder().encode(id);
-  if (idBytes.byteLength === 0) throw new RangeError("Plugin frame ID is required");
-  if (idBytes.byteLength > 0xff) throw new RangeError("Plugin frame ID is too long");
-  const headerLength = 2 + idBytes.byteLength + prefix.byteLength;
-  const frame = new Uint8Array(headerLength + payload.byteLength);
-  frame[0] = opcode;
-  frame[1] = idBytes.byteLength;
-  frame.set(idBytes, 2);
-  frame.set(prefix, 2 + idBytes.byteLength);
-  frame.set(payload, headerLength);
-  return frame;
-}
-
-function decodeInputHeader(bytes: Uint8Array): { pluginId: string; method: string } | null {
-  try {
-    const header: unknown = JSON.parse(new TextDecoder().decode(bytes));
-    if (typeof header !== "object" || header === null) return null;
-    const pluginId = Reflect.get(header, "pluginId");
-    const method = Reflect.get(header, "method");
-    if (typeof pluginId !== "string" || typeof method !== "string" || !pluginId || !method)
-      return null;
-    return { pluginId, method };
-  } catch {
-    return null;
-  }
-}
-
-/**
- * The byte rule: a plugin value is either a `Uint8Array`, sent as a binary frame, or a JSON value
- * with no `Uint8Array` inside it. Checks object structure only; never reads the bytes.
- */
-export function assertPluginWireValue(value: unknown, label: string): void {
-  if (value instanceof Uint8Array) {
-    if (value.byteLength > MAX_PLUGIN_PAYLOAD_BYTES)
-      throw new RangeError(
-        `${label} is ${value.byteLength} bytes; one value carries at most ${MAX_PLUGIN_PAYLOAD_BYTES} bytes`,
-      );
-    return;
-  }
-  const nested = findNestedBytes(value, label);
-  if (nested)
-    throw new TypeError(
-      `${nested} is a Uint8Array. Bytes cross the wire only as the whole value: send the Uint8Array itself, or encode it into the value.`,
-    );
-}
-
-function findNestedBytes(value: unknown, path: string): string | null {
-  if (value instanceof Uint8Array) return path;
-  if (typeof value !== "object" || value === null) return null;
-  if (Array.isArray(value)) {
-    for (let index = 0; index < value.length; index += 1) {
-      const found = findNestedBytes(value[index], `${path}[${index}]`);
-      if (found) return found;
-    }
-    return null;
-  }
-  for (const [key, entry] of Object.entries(value)) {
-    const found = findNestedBytes(entry, `${path}.${key}`);
-    if (found) return found;
-  }
-  return null;
-}

packages/protocol/src/client-capabilities.ts

@@ -46,9 +46,6 @@ export const CLIENT_CAPS = {
   // COMPAT(pluginTimelineItems): added in v0.8.0, remove after 2027-03-07 once client floor >= v0.8.0.
   pluginTimelineItems: "plugin_timeline_items",
   workspaceSetupObservation: "workspace_setup_observation",
-  // COMPAT(pluginBinaryPayloads): added in v0.15.0, remove after 2027-05-02 once client floor >= v0.15.0.
-  // The daemon sends plugin RPC outputs as binary opcode 0x21 only to clients that decode it.
-  pluginBinaryPayloads: "plugin_binary_payloads",
   // COMPAT(workspaceSetupBlocked): added in v0.8.0, remove after 2027-03-07 once client floor >= v0.8.0.
   workspaceSetupBlocked: "workspace_setup_blocked",
   browserHost: "browser_host",

packages/protocol/src/messages.plugins.test.ts

@@ -372,43 +372,3 @@ it("npm metadata leaves legacy plugin rows parseable without extending their clo
     payload: { requestId: "npm", plugins: [plugin] },
   });
 });
-
-describe("plugin subscription protocol", () => {
-  it("parses a subscribe request and its response", () => {
-    const request = {
-      type: "plugin.rpc.subscribe.request",
-      requestId: "r1",
-      pluginId: "diff-watch",
-      method: "diff.watch",
-      input: { a: "main", b: "HEAD" },
-    };
-    expect(SessionInboundMessageSchema.parse(request)).toEqual(request);
-    const response = {
-      type: "plugin.rpc.subscribe.response",
-      payload: { requestId: "r1", subscriptionId: "s1" },
-    };
-    expect(SessionOutboundMessageSchema.parse(response)).toEqual(response);
-  });
-
-  it("parses JSON events and closes with and without an error", () => {
-    for (const message of [
-      { type: "plugin.rpc.event", payload: { subscriptionId: "s1", event: { path: "a.ts" } } },
-      { type: "plugin.rpc.closed", payload: { subscriptionId: "s1" } },
-      { type: "plugin.rpc.closed", payload: { subscriptionId: "s1", error: "boom" } },
-    ]) {
-      expect(SessionOutboundMessageSchema.parse(message)).toEqual(message);
-    }
-  });
-
-  it("advertises the feature as an optional flag", () => {
-    const status = {
-      type: "status",
-      payload: {
-        status: "server_info",
-        serverId: "srv",
-        features: { pluginSubscriptions: true },
-      },
-    };
-    expect(StatusMessageSchema.parse(status)).toMatchObject(status);
-  });
-});

packages/protocol/src/messages.ts

@@ -1603,14 +1603,6 @@ export const PluginRpcInvokeRequestSchema = z.object({
   requestId: z.string(),
   pluginId: PluginIdSchema,
   method: z.string().min(1),
-  input: z.unknown().optional(),
-});
-
-export const PluginRpcSubscribeRequestSchema = z.object({
-  type: z.literal("plugin.rpc.subscribe.request"),
-  requestId: z.string(),
-  pluginId: PluginIdSchema,
-  method: z.string().min(1),
   input: z.unknown(),
 });
 
@@ -3347,7 +3339,6 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
   PluginDisableRequestSchema,
   PluginRemoveRequestSchema,
   PluginRpcInvokeRequestSchema,
-  PluginRpcSubscribeRequestSchema,
   AgentTimelineAppendRequestSchema,
   AgentSkillsGetStatusRequestSchema,
   AgentSkillsReconcileRequestSchema,
@@ -3748,8 +3739,6 @@ export const ServerInfoStatusPayloadSchema = z
         pluginThemes: z.boolean().optional(),
         pluginSettings: z.boolean().optional(),
         pluginTimelineItems: z.boolean().optional(),
-        // COMPAT(pluginSubscriptions): added in v0.15.0, remove gate after 2027-05-02.
-        pluginSubscriptions: z.boolean().optional(),
         // COMPAT(skillManagement): added in v0.4.0, remove gate after 2027-08-16.
         skillManagement: z.boolean().optional(),
         // COMPAT(idleRuntimeCleanupSetting): added in v0.13.0, remove gate after 2027-04-26.
@@ -6969,26 +6958,10 @@ export const PluginRpcInvokeResponseSchema = z.object({
   type: z.literal("plugin.rpc.invoke.response"),
   payload: z.object({
     requestId: z.string(),
-    output: z.unknown().optional(),
+    output: z.unknown(),
   }),
 });
 
-export const PluginRpcSubscribeResponseSchema = z.object({
-  type: z.literal("plugin.rpc.subscribe.response"),
-  payload: z.object({ requestId: z.string(), subscriptionId: z.string() }),
-});
-
-/** A JSON subscription event. A `Uint8Array` event rides as binary opcode 0x20 instead. */
-export const PluginRpcEventSchema = z.object({
-  type: z.literal("plugin.rpc.event"),
-  payload: z.object({ subscriptionId: z.string(), event: z.unknown() }),
-});
-
-export const PluginRpcClosedSchema = z.object({
-  type: z.literal("plugin.rpc.closed"),
-  payload: z.object({ subscriptionId: z.string(), error: z.string().optional() }),
-});
-
 export const AgentTimelineAppendResponseSchema = z.object({
   type: z.literal("agent.timeline.append.response"),
   payload: z.object({
@@ -7062,9 +7035,6 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
   PluginDisableResponseSchema,
   PluginRemoveResponseSchema,
   PluginRpcInvokeResponseSchema,
-  PluginRpcSubscribeResponseSchema,
-  PluginRpcEventSchema,
-  PluginRpcClosedSchema,
   AgentTimelineAppendResponseSchema,
   AgentSkillsGetStatusResponseSchema,
   AgentSkillsReconcileResponseSchema,

packages/protocol/src/schedule/cron-expression.test.ts

@@ -13,37 +13,6 @@ describe("cron expression parser", () => {
     expect(cron.dayOfWeek.matches(0)).toBe(false);
   });
 
-  it("rejects a step after a standalone number", () => {
-    expect(validateCronExpression("0/15 * * * *")).toBe(
-      "Invalid cron minute step: use * or a range before /",
-    );
-  });
-
-  it.each([
-    ["5/1 * * * *", "minute"],
-    ["0,5/15 * * * *", "minute"],
-    ["* 1/2 * * *", "hour"],
-    ["* * 1/2 * *", "day-of-month"],
-    ["* * * 1/2 *", "month"],
-    ["* * * * 1/2", "day-of-week"],
-  ])("rejects ignored numeric steps in %s", (expression, field) => {
-    expect(validateCronExpression(expression)).toBe(
-      `Invalid cron ${field} step: use * or a range before /`,
-    );
-  });
-
-  it.each([
-    { expression: "*/15 * * * *", expected: [0, 15, 30, 45] },
-    { expression: "0-59/15 * * * *", expected: [0, 15, 30, 45] },
-    { expression: "5-59/15 * * * *", expected: [5, 20, 35, 50] },
-  ])("preserves supported steps in $expression", ({ expression, expected }) => {
-    const cron = parseCronExpression(expression);
-    const minutes = Array.from({ length: 60 }, (_, minute) => minute);
-    const matches = minutes.filter((minute) => cron.minute.matches(minute));
-
-    expect(matches).toEqual(expected);
-  });
-
   it("reports invalid expressions with server-facing copy", () => {
     expect(validateCronExpression("* * *")).toBe("Cron expressions must have 5 fields");
     expect(validateCronExpression("*/5/2 * * * *")).toBe("Invalid cron minute step");

packages/protocol/src/schedule/cron-expression.ts

@@ -90,9 +90,6 @@ function parseField(source: string, bounds: CronFieldBounds): CronFieldMatcher {
     if (!/^\d+$/.test(base)) {
       throw new Error(`Invalid cron ${bounds.name} value`);
     }
-    // COMPAT(cronNumericSteps): added in v0.12.0, remove after 2027-04-17 once
-    // persisted numeric steps are migrated. Validation rejects new numeric steps;
-    // existing schedules retain their historical singleton match.
     const value = Number.parseInt(base, 10);
     if (!Number.isInteger(value) || value < bounds.min || value > bounds.max) {
       throw new Error(`Invalid cron ${bounds.name} value`);
@@ -125,12 +122,6 @@ export function parseCronExpression(expression: string): ParsedCronExpression {
 export function validateCronExpression(expression: string): string | null {
   try {
     parseCronExpression(expression);
-    const fields = expression.trim().split(/\s+/);
-    for (const [index, field] of fields.entries()) {
-      if (field.split(",").some((part) => /^\d+\//.test(part))) {
-        return `Invalid cron ${CRON_FIELD_BOUNDS[index].name} step: use * or a range before /`;
-      }
-    }
     return null;
   } catch (error) {
     return error instanceof Error ? error.message : "Invalid cron expression";

packages/protocol/tests/validation/ws-outbound.test.ts

@@ -5,7 +5,7 @@ import { fileURLToPath, pathToFileURL } from "node:url";
 import { createJiti } from "jiti";
 import type { z } from "zod";
 import { describe, expect, it } from "vitest";
-import { PluginRpcInvokeResponseSchema, WSOutboundMessageSchema } from "../../src/messages.js";
+import { WSOutboundMessageSchema } from "../../src/messages.js";
 import { validateWSOutboundMessage } from "../../src/validation/ws-outbound.js";
 
 interface GeneratedSchema {
@@ -401,8 +401,3 @@ const SourceSchema = z.object({
     });
   });
 });
-
-it("accepts omitted RPC output so the receiving client can apply schema defaults", () => {
-  const message = { type: "plugin.rpc.invoke.response", payload: { requestId: "default-output" } };
-  expect(PluginRpcInvokeResponseSchema.safeParse(message).success).toBe(true);
-});

packages/relay/src/e2e.test.ts

@@ -56,9 +56,6 @@ function spawnRelayDevServer(port: number): ChildProcess {
       wranglerCliPath,
       "dev",
       "--local",
-      // Local execution still inherits production vars; exercise the local Durable Object.
-      "--var",
-      "PASEO_RELAY_UPSTREAM:",
       "--ip",
       "127.0.0.1",
       "--port",

packages/server/src/server/agent/agent-manager.test.ts

@@ -7317,7 +7317,6 @@ test("ignores stale autonomous terminals without lowering the active turn lifecy
     const running = manager.getAgent(snapshot.id);
     expect(running?.lifecycle).toBe("running");
     expect(running ? toAgentPayload(running).activeTurn?.turnId : null).toBe("turn-b");
-    expect(running?.lastError).toBeUndefined();
   });
 
   const active = manager.getAgent(snapshot.id)!;
@@ -7360,7 +7359,7 @@ test("ignores stale autonomous terminals without lowering the active turn lifecy
     const stillRunning = manager.getAgent(snapshot.id);
     expect(stillRunning?.lifecycle).toBe("running");
     expect(stillRunning ? toAgentPayload(stillRunning).activeTurn?.turnId : null).toBe("turn-b");
-    expect(stillRunning?.lastError).toBeUndefined();
+    expect(stillRunning?.lastError).toBe("turn-b marker");
     expect(stillRunning?.lastUsage).toEqual({ inputTokens: 7 });
     expect(stillRunning?.pendingPermissions.has("turn-b-permission")).toBe(true);
     expect(manager.getTimeline(snapshot.id)).toEqual(timelineBeforeStaleTerminals);
@@ -8684,118 +8683,6 @@ test("clearAgentAttention on errored agent stays cleared until a new error trans
   expect(persistedAfterSecondFailure?.attentionReason).toBe("error");
 });
 
-test.each([
-  ["autonomous", "turn_completed", "idle", "finished", undefined],
-  ["autonomous", "turn_failed", "error", "error", "Retry failed."],
-  ["foreground", "turn_completed", "idle", "finished", undefined],
-  ["foreground", "turn_failed", "error", "error", "Retry failed."],
-] as const)(
-  "%s recovery clears error attention before %s",
-  async (mode, terminal, status, expectedReason, lastError) => {
-    const workdir = mkdtempSync(join(tmpdir(), "agent-manager-attention-recovery-"));
-    const storage = new AgentStorage(join(workdir, "agents"), logger);
-    class RecoverySession extends TestAgentSession {
-      override async startTurn(): Promise<{ turnId: string }> {
-        return { turnId: "recovered-turn" };
-      }
-    }
-
-    class RecoveryClient extends TestAgentClient {
-      readonly session = new RecoverySession({ provider: "codex", cwd: workdir });
-
-      override async createSession(): Promise<AgentSession> {
-        return this.session;
-      }
-    }
-
-    const client = new RecoveryClient();
-    const attentionReasons: Array<"finished" | "error" | "permission"> = [];
-    const manager = new AgentManager({
-      clients: { codex: client },
-      registry: storage,
-      logger,
-      onAgentAttention: ({ reason }) => attentionReasons.push(reason),
-    });
-    onTestFinished(async () => {
-      await manager.closeAgent(agent.id);
-      await manager.flush();
-      rmSync(workdir, { recursive: true, force: true });
-    });
-    const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
-      workspaceId: undefined,
-    });
-
-    const running = waitForAgentLifecycle(manager, agent.id, "running");
-    client.session.pushEvent({ type: "turn_started", provider: "codex", turnId: "failed-turn" });
-    await running;
-    const failed = waitForAgentLifecycle(manager, agent.id, "error");
-    client.session.pushEvent({
-      type: "turn_failed",
-      provider: "codex",
-      turnId: "failed-turn",
-      error: "Selected model is at capacity.",
-    });
-    await failed;
-    await vi.waitFor(() => expect(manager.hasInFlightRun(agent.id)).toBe(false));
-    expect(toAgentPayload(manager.getAgent(agent.id)!)).toMatchObject({
-      status: "error",
-      attentionReason: "error",
-      lastError: "Selected model is at capacity.",
-    });
-
-    const recovered = waitForAgentLifecycle(manager, agent.id, "running");
-    const stream =
-      mode === "foreground"
-        ? drainAsyncGenerator(manager.streamAgent(agent.id, "Continue after capacity recovers."))
-        : null;
-    if (mode === "autonomous") {
-      client.session.pushEvent({
-        type: "turn_started",
-        provider: "codex",
-        turnId: "recovered-turn",
-      });
-    }
-    await recovered;
-    await manager.flush();
-    expect(toAgentPayload(manager.getAgent(agent.id)!)).toMatchObject({
-      status: "running",
-      requiresAttention: false,
-      attentionReason: null,
-    });
-    expect(manager.getAgent(agent.id)?.lastError).toBeUndefined();
-    expect(await storage.get(agent.id)).toMatchObject({
-      lastStatus: "running",
-      requiresAttention: false,
-      attentionReason: null,
-    });
-    expect((await storage.get(agent.id))?.lastError).toBeUndefined();
-
-    const settled = waitForAgentLifecycle(manager, agent.id, status);
-    client.session.pushEvent({
-      type: terminal,
-      provider: "codex",
-      turnId: "recovered-turn",
-      ...(terminal === "turn_failed" ? { error: "Retry failed." } : {}),
-    });
-    await settled;
-    await stream;
-    await manager.flush();
-    expect(toAgentPayload(manager.getAgent(agent.id)!)).toMatchObject({
-      status,
-      requiresAttention: true,
-      attentionReason: expectedReason,
-    });
-    expect(manager.getAgent(agent.id)?.lastError).toBe(lastError);
-    expect(await storage.get(agent.id)).toMatchObject({
-      lastStatus: status,
-      requiresAttention: true,
-      attentionReason: expectedReason,
-    });
-    expect((await storage.get(agent.id))?.lastError).toBe(lastError);
-    expect(attentionReasons).toEqual(["error", expectedReason]);
-  },
-);
-
 test("streamAgent clears pending run when startTurn fails before a turn id exists", async () => {
   const workdir = mkdtempSync(join(tmpdir(), "agent-manager-start-turn-failure-"));
   const storagePath = join(workdir, "agents");

packages/server/src/server/agent/agent-manager.ts

@@ -2840,7 +2840,6 @@ export class AgentManager {
       this.runs.settleForegroundStart(pendingRun, { status: "started", turnId });
       agent.activeForegroundTurnId = turnId;
       this.openActiveTurn(agent, turnId, turnStartedAt);
-      agent.attention = { requiresAttention: false };
       agent.lifecycle = "running";
       this.touchUpdatedAt(agent);
       // AgentManager owns the accepted-turn boundary. Publish liveness before the canonical
@@ -4953,8 +4952,6 @@ export class AgentManager {
     if (eventTurnId) {
       this.openActiveTurn(agent, eventTurnId, new Date());
     }
-    agent.lastError = undefined;
-    agent.attention = { requiresAttention: false };
     agent.lifecycle = "running";
     this.emitState(agent);
   }

packages/server/src/server/agent/host-operations/index.ts

@@ -399,16 +399,14 @@ function remoteObservation(client: DaemonClient): HostObservation {
           // A new turn may start between the snapshot and the current-response read.
           // Bound that wait and refresh state instead of attributing the old event to it.
           const current = await client.waitForFinish(agentId, 100);
-          if (current.status === "timeout") {
-            // Queued delivery can leave a closed/error snapshot visible before the new turn.
-            // A timed-out wait has not established an outcome for that queued work.
-            if (event?.type === "permission_resolved") handler(snapshot, event);
-            return;
+          if (current.status === "timeout")
+            snapshot = (await client.fetchAgent(agentId))?.agent ?? null;
+          else {
+            if (!current.final && current.error && current.error !== `Agent not found: ${agentId}`)
+              throw new Error(current.error);
+            snapshot = current.final;
+            lastMessage = current.lastMessage;
           }
-          if (!current.final && current.error && current.error !== `Agent not found: ${agentId}`)
-            throw new Error(current.error);
-          snapshot = current.final;
-          lastMessage = current.lastMessage;
         }
         handler(snapshot, event, lastMessage);
       };

packages/server/src/server/agent/mcp-parity.e2e.test.ts

@@ -404,7 +404,6 @@ describe("Suite A: Core Fixes", () => {
           expect(parseAgentMessage(callerPrompts[0]).source).toEqual({
             kind: "agent-notification",
             agentId: childId,
-            serverId: daemon.daemon.getServerId(),
             title: "Notification child",
             event: "finished",
           });
@@ -518,7 +517,6 @@ describe("Suite A: Core Fixes", () => {
         expect(notification.source).toEqual({
           kind: "agent-notification",
           agentId: childId,
-          serverId: daemon.daemon.getServerId(),
           title: "Watched child",
           event: "errored",
         });
@@ -581,11 +579,7 @@ describe("Suite A: Core Fixes", () => {
             type: "tool_call",
             agentMessage: {
               event: outcome,
-              sender: {
-                id: childId,
-                serverId: daemonHandle.daemon.getServerId(),
-                title: "Follow-up child",
-              },
+              sender: { id: childId, title: "Follow-up child" },
               text: expect.stringContaining(`(Follow-up child) ${outcome}.`),
             },
           }),

packages/server/src/server/agent/providers/acp-agent.ts

@@ -643,7 +643,6 @@ interface PendingPermission {
   resolve: (response: RequestPermissionResponse) => void;
   reject: (error: Error) => void;
   turnId: string | null;
-  toolCallId: string;
 }
 
 interface PendingUserMessage {
@@ -2645,7 +2644,6 @@ export class ACPAgentSession implements AgentSession, ACPClient {
         resolve,
         reject,
         turnId: this.activeForegroundTurnId,
-        toolCallId: params.toolCall.toolCallId,
       });
     });
 
@@ -3209,9 +3207,6 @@ export class ACPAgentSession implements AgentSession, ACPClient {
       snapshot = this.toolSnapshotTransformer(snapshot);
     }
     this.toolCalls.set(toolCallId, snapshot);
-    if (update.status === "completed" || update.status === "failed") {
-      this.cancelPendingPermissions((pending) => pending.toolCallId === toolCallId);
-    }
     return [this.wrapTimeline(mapToolSnapshotToTimeline(snapshot, this.terminalEntries))];
   }
 
@@ -3419,11 +3414,11 @@ export class ACPAgentSession implements AgentSession, ACPClient {
     event: Extract<AgentStreamEvent, { type: "turn_completed" | "turn_failed" | "turn_canceled" }>,
   ): void {
     this.deliverTranslatedEvents(this.flushPendingUserMessage());
-    // Failed and canceled turns clear every request in the manager. Answer the
-    // same set here so an outside-turn request cannot keep the provider blocked.
-    this.cancelPendingPermissions(
-      (pending) => event.type !== "turn_completed" || pending.turnId === event.turnId,
-    );
+    // A turn that ends without completing leaves no way to answer its open
+    // permission requests, so tell the agent they were cancelled.
+    if (event.type !== "turn_completed") {
+      this.cancelPendingPermissions();
+    }
     this.activeForegroundTurnId = null;
     this.fallbackAssistantMessageId = null;
     if (this.submittedUserMessageTurnId === event.turnId) {
@@ -3432,21 +3427,11 @@ export class ACPAgentSession implements AgentSession, ACPClient {
     this.pushEvent(event);
   }
 
-  private cancelPendingPermissions(
-    matches: (pending: PendingPermission) => boolean = () => true,
-  ): void {
-    for (const [requestId, pending] of this.pendingPermissions) {
-      if (!matches(pending)) continue;
-      this.pendingPermissions.delete(requestId);
+  private cancelPendingPermissions(): void {
+    for (const pending of this.pendingPermissions.values()) {
       pending.resolve({ outcome: { outcome: "cancelled" } });
-      this.pushEvent({
-        type: "permission_resolved",
-        provider: this.provider,
-        requestId,
-        resolution: { behavior: "deny", message: "Permission request canceled" },
-        turnId: pending.turnId ?? undefined,
-      });
     }
+    this.pendingPermissions.clear();
   }
 
   private emitBootstrapThreadEvent(): void {

packages/server/src/server/agent/providers/claude/agent.error-result.test.ts

@@ -13,11 +13,6 @@ interface ScriptedTurn {
   before?: string;
   after?: string;
   isError: boolean;
-  subtype?:
-    | "error_max_turns"
-    | "error_during_execution"
-    | "error_max_budget_usd"
-    | "error_max_structured_output_retries";
   usage?: {
     inputTokens: number;
     cachedInputTokens: number;
@@ -62,12 +57,11 @@ function scriptedQuery(turns: ScriptedTurn[]): ClaudeQueryFactory {
         if (turn.after) yield assistant(`after-${index}`, turn.after);
         yield {
           type: "result",
-          subtype: turn.subtype ?? "success",
+          subtype: "success",
           session_id: "error-session",
           duration_ms: 1,
           duration_api_ms: 0,
           is_error: turn.isError,
-          ...(turn.subtype ? { errors: ["Reached maximum number of turns (1)"] } : {}),
           num_turns: 1,
           result: turn.isError ? (turn.errorReply ?? "") : (turn.after ?? turn.errorReply ?? ""),
           stop_reason: "stop_sequence",
@@ -194,33 +188,3 @@ test("an error reply the turn moves past is still shown", async () => {
   expect(terminal(completed)).toMatchObject({ type: "turn_completed" });
   expect(assistantTexts(completed)).toEqual([LIMIT]);
 });
-
-test.each([
-  "error_max_turns",
-  "error_during_execution",
-  "error_max_budget_usd",
-  "error_max_structured_output_retries",
-] as const)("a %s result publishes final usage before failure", async (subtype) => {
-  const usage = {
-    inputTokens: 2,
-    cachedInputTokens: 12577,
-    outputTokens: 110,
-    totalCostUsd: 0.00365537,
-  };
-  const [failed, next] = await runTurns([
-    { before: "Running the command.", isError: true, subtype, usage },
-    { after: "Done.", isError: false },
-  ]);
-
-  // Collection ends at the terminal event, so this also checks ordering.
-  expect(failed).toContainEqual(
-    expect.objectContaining({ type: "usage_updated", usage: expect.objectContaining(usage) }),
-  );
-  expect(terminal(failed)).toMatchObject({
-    type: "turn_failed",
-    error: "Reached maximum number of turns (1)",
-  });
-  expect(assistantTexts(failed)).toEqual(["Running the command."]);
-  expect(terminal(next)).toMatchObject({ type: "turn_completed" });
-  expect(assistantTexts(next)).toEqual(["Done."]);
-});

packages/server/src/server/agent/providers/claude/agent.ts

@@ -5037,7 +5037,6 @@ class ClaudeAgentSession implements AgentSession {
       "errors" in message && Array.isArray(message.errors) && message.errors.length > 0
         ? message.errors.join("\n")
         : "Claude run failed";
-    if (usage) events.push({ type: "usage_updated", provider: "claude", usage });
     events.push(...this.sidechainTracker.finishAll("failed"));
     events.push(this.buildTurnFailedEvent(errorMessage));
   }

packages/server/src/server/agent/providers/codex-app-server-agent.test.ts

@@ -227,57 +227,6 @@ function deferred<T>() {
   return { promise, resolve };
 }
 
-describe("Codex file-change approvals", () => {
-  test("file-change approval includes the matching patch when notifications are still queued", async () => {
-    const appServer = createFakeCodexAppServer();
-    const { session } = await startPublicSteeringSession(appServer);
-    try {
-      for (const [id, filePath, diff] of [
-        ["file-alpha", "/workspace/project/alpha.txt", "@@ -1 +1 @@\n-blue\n+green\n"],
-        ["file-beta", "/workspace/project/beta.txt", "@@ -1 +1 @@\n-small\n+large\n"],
-      ]) {
-        appServer.startsFileChange({
-          threadId: "thread-1",
-          turnId: "native-A",
-          callId: id,
-          path: filePath,
-          diff,
-        });
-      }
-      const permission = waitForNextPermission(session);
-      appServer.requestFileChangeApproval({
-        itemId: "file-alpha",
-        threadId: "thread-1",
-        turnId: "native-A",
-        reason: "Apply the patch",
-      });
-
-      expect((await permission).request).toEqual({
-        id: "permission-file-alpha",
-        provider: "codex",
-        name: "CodexFileChange",
-        kind: "tool",
-        title: "Apply file changes: alpha.txt",
-        description: "Apply the patch",
-        detail: {
-          type: "edit",
-          filePath: "alpha.txt",
-          unifiedDiff: "@@ -1 +1 @@\n-blue\n+green\n",
-        },
-        metadata: { itemId: "file-alpha", threadId: "thread-1", turnId: "native-A" },
-      });
-
-      await session.respondToPermission("permission-file-alpha", { behavior: "allow" });
-      await expect(appServer.waitForApprovalDecision("file-alpha")).resolves.toEqual({
-        decision: "accept",
-      });
-      appServer.assertNoErrors();
-    } finally {
-      await session.close();
-    }
-  });
-});
-
 describe("Codex active-turn steering admission", () => {
   test("steers an accepted turn whose native id arrives after startTurn returns", async () => {
     const appServer = createFakeCodexAppServer({

packages/server/src/server/agent/providers/codex-app-server-agent.ts

@@ -3483,7 +3483,6 @@ export class CodexAppServerAgentSession implements AgentSession {
   private pendingReasoning = new Map<string, string[]>();
   private pendingCommandOutputDeltas = new Map<string, string[]>();
   private pendingFileChangeOutputDeltas = new Map<string, string[]>();
-  private pendingToolCallDetails = new Map<string, ToolCallTimelineItem["detail"]>();
   private pendingAssistantMessageBoundary = false;
   private terminalCommandByProcessId = new Map<string, string>();
   // The root thread's still-running command process ids, keyed by the native turn that
@@ -5252,7 +5251,6 @@ export class CodexAppServerAgentSession implements AgentSession {
   async close(): Promise<void> {
     this.closed = true;
     this.clearPendingPermissions();
-    this.pendingToolCallDetails.clear();
     this.pendingSubAgentNotificationsByThreadId.clear();
     this.subscribers.clear();
     this.startupEvents = null;
@@ -6418,7 +6416,6 @@ export class CodexAppServerAgentSession implements AgentSession {
     this.pendingReasoning.clear();
     this.pendingCommandOutputDeltas.clear();
     this.pendingFileChangeOutputDeltas.clear();
-    this.pendingToolCallDetails.clear();
     this.pendingAssistantMessageBoundary = false;
     this.warnedIncompleteEditToolCallIds.clear();
     this.pendingRootCompactionItemIds.clear();
@@ -6792,9 +6789,6 @@ export class CodexAppServerAgentSession implements AgentSession {
     if (shouldIgnoreMirroredLifecycleItem(parsed.source, parsed.item)) {
       return;
     }
-    if (parsed.item.id) {
-      this.pendingToolCallDetails.delete(parsed.item.id);
-    }
     this.receiveAsyncQuestion(parsed.threadId, parsed.item);
     if (this.isUserMessageItem(parsed.item)) {
       await this.handleUserMessageItem(parsed);
@@ -6842,8 +6836,10 @@ export class CodexAppServerAgentSession implements AgentSession {
       if (timelineItem.type === "assistant_message") {
         this.pendingAssistantMessageBoundary = true;
       }
-      this.emittedItemCompletedIds.add(itemId);
-      this.emittedItemStartedIds.delete(itemId);
+      if (itemId) {
+        this.emittedItemCompletedIds.add(itemId);
+        this.emittedItemStartedIds.delete(itemId);
+      }
       await this.replayPendingSubAgentNotifications(registeredChildThreadIds);
       return;
     }
@@ -6879,7 +6875,7 @@ export class CodexAppServerAgentSession implements AgentSession {
   private consumeStreamedTextCompletion(
     timelineItem: AgentTimelineItem,
     itemId: string | null | undefined,
-  ): itemId is string {
+  ): boolean {
     if (!itemId) {
       return false;
     }
@@ -6988,7 +6984,6 @@ export class CodexAppServerAgentSession implements AgentSession {
     if (!timelineItem || timelineItem.type !== "tool_call") {
       return;
     }
-    this.pendingToolCallDetails.set(timelineItem.callId, timelineItem.detail);
     const registeredChildThreadIds = this.registerSubAgentToolCall({
       timelineItem,
       rawItem: parsed.item,
@@ -7258,8 +7253,7 @@ export class CodexAppServerAgentSession implements AgentSession {
     });
   }
 
-  private async handleFileChangeApprovalRequest(params: unknown): Promise<unknown> {
-    await this.pendingNotifications;
+  private handleFileChangeApprovalRequest(params: unknown): Promise<unknown> {
     const parsed = z
       .object({
         itemId: z.string(),
@@ -7269,25 +7263,20 @@ export class CodexAppServerAgentSession implements AgentSession {
       })
       .parse(params);
     const requestId = `permission-${parsed.itemId}`;
-    const detail: ToolCallTimelineItem["detail"] = this.pendingToolCallDetails.get(
-      parsed.itemId,
-    ) ?? {
-      type: "unknown",
-      input: { reason: parsed.reason ?? null },
-      output: null,
-    };
-    const title =
-      detail.type === "edit" && detail.filePath
-        ? `Apply file changes: ${detail.filePath}`
-        : "Apply file changes";
     const request: AgentPermissionRequest = {
       id: requestId,
       provider: CODEX_PROVIDER,
       name: "CodexFileChange",
       kind: "tool",
-      title,
+      title: "Apply file changes",
       description: parsed.reason ?? undefined,
-      detail,
+      detail: {
+        type: "unknown",
+        input: {
+          reason: parsed.reason ?? null,
+        },
+        output: null,
+      },
       metadata: {
         itemId: parsed.itemId,
         threadId: parsed.threadId,

packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts

@@ -43,13 +43,6 @@ interface FakeTerminalInput {
   processId: string;
   text: string;
 }
-interface FakeFileChange {
-  threadId: string;
-  turnId: string;
-  callId: string;
-  path: string;
-  diff: string;
-}
 interface FakeLegacyPatch {
   threadId: string;
   callId: string;
@@ -95,7 +88,6 @@ export interface FakeCodexAppServer {
   completesSubAgentActivity(params: FakeSubAgentActivity): void;
   completesCompaction(params: { threadId: string; itemId: string }): void;
   runsLegacyCommand(params: FakeLegacyCommand): void;
-  startsFileChange(params: FakeFileChange): void;
   appliesLegacyPatch(params: FakeLegacyPatch): void;
   completesCommand(params: FakeLegacyCommand): void;
   completesSilentCommand(params: FakeSilentCommand): void;
@@ -475,20 +467,6 @@ export function createFakeCodexAppServer(
         success: true,
       });
     },
-    startsFileChange(params) {
-      writeNotification("item/started", {
-        threadId: params.threadId,
-        turnId: params.turnId,
-        item: {
-          type: "fileChange",
-          id: params.callId,
-          status: "inProgress",
-          changes: [
-            { path: params.path, kind: { type: "update", move_path: null }, diff: params.diff },
-          ],
-        },
-      });
-    },
     appliesLegacyPatch(params) {
       const changes = [
         {

packages/server/src/server/agent/providers/generic-acp-agent.permission-turn-failure.test.ts

@@ -5,7 +5,6 @@ import path from "node:path";
 import { describe, expect, test } from "vitest";
 
 import { createTestLogger } from "../../../test-utils/test-logger.js";
-import { AgentManager } from "../agent-manager.js";
 import type { AgentSession, AgentStreamEvent } from "../agent-sdk-types.js";
 import { GenericACPAgentClient } from "./generic-acp-agent.js";
 
@@ -13,7 +12,6 @@ describe("GenericACPAgentClient permission requests when a turn ends early", ()
   test.each([
     { ending: "fail", terminalEvent: "turn_failed" },
     { ending: "cancel", terminalEvent: "turn_canceled" },
-    { ending: "end_turn", terminalEvent: "turn_completed" },
   ] as const)(
     "answers a permission request left open by a $terminalEvent turn with cancelled",
     async ({ ending, terminalEvent }) => {
@@ -27,16 +25,6 @@ describe("GenericACPAgentClient permission requests when a turn ends early", ()
           await waitForEvent(events, terminalEvent);
           expect(hasEvent(events, "permission_requested")).toBe(true);
           expect(session.getPendingPermissions()).toEqual([]);
-          const requested = permissionRequests(events)[0]!;
-          expect(resolvedEvents(events)).toEqual([
-            {
-              type: "permission_resolved",
-              provider: "acp",
-              requestId: requested.request.id,
-              resolution: { behavior: "deny", message: "Permission request canceled" },
-              turnId: requested.turnId,
-            },
-          ]);
 
           const replyStart = events.length;
           await session.startTurn("turn two");
@@ -51,84 +39,6 @@ describe("GenericACPAgentClient permission requests when a turn ends early", ()
       });
     },
   );
-
-  test.each(["completed", "failed"] as const)(
-    "cancels only the matching permission when its tool becomes %s",
-    async (status) => {
-      await withFakeACPAgent(status, async (command, cwd) => {
-        const client = new GenericACPAgentClient({ logger: createTestLogger(), command });
-        const session = await client.createSession({ provider: "acp", cwd });
-        try {
-          const events = collectEvents(session);
-          await session.startTurn("turn one");
-          await waitForToolUpdate(events);
-          expect(pendingToolIds(session)).toEqual(["tool-2"]);
-          expect(resolvedEvents(events)).toEqual([
-            expect.objectContaining({ requestId: permissionRequests(events)[0]!.request.id }),
-          ]);
-        } finally {
-          await session.close();
-        }
-      });
-    },
-  );
-
-  test("keeps a permission raised outside a turn answerable after another turn completes", async () => {
-    await withFakeACPAgent("outside_turn", async (command, cwd) => {
-      const client = new GenericACPAgentClient({ logger: createTestLogger(), command });
-      const session = await client.createSession({ provider: "acp", cwd });
-      try {
-        const outside = session.getPendingPermissions();
-        expect(pendingToolIds(session)).toEqual(["tool-outside"]);
-        const events = collectEvents(session);
-        await session.startTurn("turn one");
-        await waitForEvent(events, "turn_completed");
-        expect(session.getPendingPermissions()).toEqual(outside);
-        await session.respondToPermission(outside[0]!.id, { behavior: "allow" });
-        expect(session.getPendingPermissions()).toEqual([]);
-      } finally {
-        await session.close();
-      }
-    });
-  });
-
-  test.each(["fail", "cancel"] as const)(
-    "settles an outside-turn permission that the manager clears when a turn ends with %s",
-    async (ending) => {
-      await withFakeACPAgent(`outside_${ending}`, async (command, cwd) => {
-        const logger = createTestLogger();
-        const client = new GenericACPAgentClient({
-          logger,
-          command,
-          waitForInitialCommands: false,
-        });
-        const manager = new AgentManager({ clients: { acp: client }, logger });
-        const agent = await manager.createAgent({ provider: "acp", cwd }, undefined, {
-          workspaceId: undefined,
-        });
-        try {
-          await waitForManagedPermission(manager, agent.id);
-          if (ending === "fail") {
-            await expect(manager.runAgent(agent.id, "turn one")).rejects.toThrow(
-              "turn failed with a permission open",
-            );
-          } else {
-            await expect(manager.runAgent(agent.id, "turn one")).resolves.toMatchObject({
-              canceled: true,
-            });
-          }
-          expect(manager.getPendingPermissions(agent.id)).toEqual([]);
-          const result = await manager.runAgent(agent.id, "turn two");
-          expect(result.finalText).toBe(
-            'perm-outside answered: {"outcome":{"outcome":"cancelled"}}',
-          );
-        } finally {
-          await manager.closeAgent(agent.id);
-          await manager.flush();
-        }
-      });
-    },
-  );
 });
 
 function collectEvents(session: AgentSession): AgentStreamEvent[] {
@@ -137,10 +47,6 @@ function collectEvents(session: AgentSession): AgentStreamEvent[] {
   return events;
 }
 
-async function waitForManagedPermission(manager: AgentManager, agentId: string): Promise<void> {
-  await expect.poll(() => manager.getPendingPermissions(agentId).length).toBe(1);
-}
-
 function hasEvent(
   events: AgentStreamEvent[],
   type: AgentStreamEvent["type"],
@@ -171,32 +77,8 @@ function assistantText(events: AgentStreamEvent[]): string {
     .join("");
 }
 
-function permissionRequests(events: AgentStreamEvent[]) {
-  return events.filter((event) => event.type === "permission_requested");
-}
-
-function resolvedEvents(events: AgentStreamEvent[]) {
-  return events.filter((event) => event.type === "permission_resolved");
-}
-
-function pendingToolIds(session: AgentSession) {
-  return session.getPendingPermissions().map((request) => request.metadata?.toolCallId);
-}
-
-async function waitForToolUpdate(events: AgentStreamEvent[]): Promise<void> {
-  await expect.poll(() => assistantText(events)).toBe("tool terminal");
-}
-
 async function withFakeACPAgent(
-  ending:
-    | "fail"
-    | "cancel"
-    | "end_turn"
-    | "completed"
-    | "failed"
-    | "outside_turn"
-    | "outside_fail"
-    | "outside_cancel",
+  ending: "fail" | "cancel",
   run: (command: [string, ...string[]], cwd: string) => Promise<void>,
 ): Promise<void> {
   const testDir = await mkdtemp(path.join(tmpdir(), "paseo-acp-permission-turn-failure-"));
@@ -209,7 +91,8 @@ async function withFakeACPAgent(
   }
 }
 
-// Turn one ends while its permission request is still open.
+// Turn one asks for permission, then fails or is cancelled while the request is
+// still open.
 // Later turns report whether that request was ever answered.
 const fakeACPAgentScript = `
 const readline = require("node:readline");
@@ -218,34 +101,15 @@ const ending = process.argv[2];
 const rl = readline.createInterface({ input: process.stdin });
 let promptCount = 0;
 let permissionAnswer = null;
-let outsidePermissionAnswer = null;
 
 function write(message) {
   process.stdout.write(JSON.stringify({ jsonrpc: "2.0", ...message }) + "\\n");
 }
 
-function ask(sessionId, requestId, toolCallId) {
-  const toolCall = { toolCallId, title: "echo hello", kind: "execute", status: "pending" };
-  write({ method: "session/update", params: { sessionId, update: { sessionUpdate: "tool_call", ...toolCall } } });
-  write({
-    id: requestId,
-    method: "session/request_permission",
-    params: {
-      sessionId,
-      toolCall,
-      options: [
-        { optionId: "allow", name: "Allow once", kind: "allow_once" },
-        { optionId: "reject", name: "Reject", kind: "reject_once" },
-      ],
-    },
-  });
-}
-
 rl.on("line", (line) => {
   const message = JSON.parse(line);
-  if (message.method === undefined) {
-    if (message.id === "perm-1") permissionAnswer = message.result ?? { error: message.error };
-    if (message.id === "perm-outside") outsidePermissionAnswer = message.result ?? { error: message.error };
+  if (message.method === undefined && message.id === "perm-1") {
+    permissionAnswer = message.result ?? { error: message.error };
     return;
   }
   if (message.method === "initialize") {
@@ -253,7 +117,6 @@ rl.on("line", (line) => {
     return;
   }
   if (message.method === "session/new") {
-    if (ending.startsWith("outside_")) ask("session-1", "perm-outside", "tool-outside");
     write({ id: message.id, result: { sessionId: "session-1" } });
     return;
   }
@@ -261,32 +124,35 @@ rl.on("line", (line) => {
     const sessionId = message.params.sessionId;
     promptCount += 1;
     if (promptCount === 1) {
-      ask(sessionId, "perm-1", "tool-1");
-      if (ending === "completed" || ending === "failed") {
-        ask(sessionId, "perm-2", "tool-2");
-        write({ method: "session/update", params: { sessionId, update: { sessionUpdate: "tool_call_update", toolCallId: "tool-1", status: ending } } });
-        write({ method: "session/update", params: { sessionId, update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: "tool terminal" } } } });
-        return;
-      }
+      const toolCall = { toolCallId: "tool-1", title: "echo hello", kind: "execute", status: "pending" };
+      write({ method: "session/update", params: { sessionId, update: { sessionUpdate: "tool_call", ...toolCall } } });
+      write({
+        id: "perm-1",
+        method: "session/request_permission",
+        params: {
+          sessionId,
+          toolCall,
+          options: [
+            { optionId: "allow", name: "Allow once", kind: "allow_once" },
+            { optionId: "reject", name: "Reject", kind: "reject_once" },
+          ],
+        },
+      });
       setTimeout(() => {
-        if (ending === "cancel" || ending === "outside_cancel") {
+        if (ending === "cancel") {
           write({ id: message.id, result: { stopReason: "cancelled" } });
-        } else if (ending === "end_turn" || ending === "outside_turn") {
-          write({ id: message.id, result: { stopReason: "end_turn" } });
         } else {
           write({ id: message.id, error: { code: -32603, message: "turn failed with a permission open" } });
         }
       }, 50);
       return;
     }
-    const answer = ending.startsWith("outside_") ? outsidePermissionAnswer : permissionAnswer;
-    const requestId = ending.startsWith("outside_") ? "perm-outside" : "perm-1";
-    const state = answer === null ? "unanswered" : "answered: " + JSON.stringify(answer);
+    const state = permissionAnswer === null ? "unanswered" : "answered: " + JSON.stringify(permissionAnswer);
     write({
       method: "session/update",
       params: {
         sessionId,
-        update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: requestId + " " + state } },
+        update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: "perm-1 " + state } },
       },
     });
     write({ id: message.id, result: { stopReason: "end_turn" } });

packages/server/src/server/agent/providers/pi/agent.test.ts

@@ -1600,111 +1600,6 @@ describe("PiRpcAgentSession", () => {
     ]);
   });
 
-  test("closes a history-only Pi runtime after replaying its timeline", async () => {
-    const pi = new FakePi();
-    pi.queueSessionSetup((runtimeSession) => {
-      runtimeSession.messages = [
-        { role: "user", content: "Old question" },
-        { role: "assistant", content: [{ type: "text", text: "Old answer" }] },
-      ];
-    });
-
-    const client: import("../../agent-sdk-types.js").AgentClient = createClient(pi);
-    const session = await client.resumeSession(
-      {
-        provider: "pi",
-        sessionId: "pi-session-1",
-        nativeHandle: "/tmp/native-pi-session",
-        metadata: { cwd: tmpdir() },
-      },
-      {},
-      undefined,
-      { purpose: "history" },
-    );
-
-    onTestFinished(() => session.close());
-    const history: AgentStreamEvent[] = [];
-    for await (const event of session.streamHistory()) history.push(event);
-
-    expect(history).toEqual(
-      expect.arrayContaining([
-        expect.objectContaining({
-          type: "timeline",
-          provider: "pi",
-          item: expect.objectContaining({ type: "assistant_message", text: "Old answer" }),
-        }),
-      ]),
-    );
-    expect(pi.latestSession().closeCalls).toBe(1);
-
-    await session.close();
-    expect(pi.latestSession().closeCalls).toBe(1);
-  });
-
-  test("closes a history-only Pi runtime when replay is stopped", async () => {
-    const pi = new FakePi();
-    pi.queueSessionSetup((runtimeSession) => {
-      runtimeSession.messages = [
-        { role: "user", content: "Old question" },
-        { role: "assistant", content: [{ type: "text", text: "Old answer" }] },
-      ];
-    });
-    const session = await createClient(pi).resumeSession(
-      {
-        provider: "pi",
-        sessionId: "pi-session-1",
-        nativeHandle: "/tmp/native-pi-session",
-        metadata: { cwd: tmpdir() },
-      },
-      {},
-      undefined,
-      { purpose: "history" },
-    );
-    onTestFinished(() => session.close());
-    const history = session.streamHistory();
-    expect((await history.next()).done).toBe(false);
-    await history.return();
-    expect(pi.latestSession().closeCalls).toBe(1);
-  });
-
-  test("closes a history-only Pi runtime when replay fails", async () => {
-    const pi = new FakePi();
-    pi.queueSessionSetup((runtimeSession) => {
-      runtimeSession.getMessagesError = new Error("History unavailable");
-    });
-    const session = await createClient(pi).resumeSession(
-      {
-        provider: "pi",
-        sessionId: "pi-session-1",
-        nativeHandle: "/tmp/native-pi-session",
-        metadata: { cwd: tmpdir() },
-      },
-      {},
-      undefined,
-      { purpose: "history" },
-    );
-    onTestFinished(() => session.close());
-    const history = session.streamHistory();
-    await expect(history.next()).rejects.toThrow("History unavailable");
-    expect(pi.latestSession().closeCalls).toBe(1);
-  });
-
-  test("keeps an interactive Pi runtime open after replaying history", async () => {
-    const pi = new FakePi();
-    const session = await createClient(pi).resumeSession({
-      provider: "pi",
-      sessionId: "pi-session-1",
-      nativeHandle: "/tmp/native-pi-session",
-      metadata: { cwd: tmpdir() },
-    });
-    onTestFinished(() => session.close());
-    for await (const _event of session.streamHistory()) {
-      // Hydrate the session before its next interactive prompt.
-    }
-    expect(pi.latestSession().closeCalls).toBe(0);
-    await expect(session.getRuntimeInfo()).resolves.toMatchObject({ provider: "pi" });
-  });
-
   test("resumes a session whose model was removed on the model Pi falls back to", async () => {
     const pi = new FakePi();
     pi.removeModel("9router/deepseek-v4-flash");
@@ -3845,17 +3740,14 @@ readline.createInterface({ input: process.stdin }).on("line", (line) => {
     },
   );
 
-  test("a history resume of an agent whose folder is gone removes its session copy after replay and leaves the original unchanged", async () => {
+  test("a history resume of an agent whose folder is gone removes its session copy on close and leaves the original unchanged", async () => {
     const archived = writeArchivedSession(createDeletedFolder());
     const pi = new FakePi();
 
     const session = await resumeArchived(createClient(pi), archived, "history");
     const sessionCopy = pi.recordedLaunches[0]!.session!;
     expect(existsSync(sessionCopy)).toBe(true);
-    for await (const _event of session.streamHistory()) {
-      // Completion releases both the runtime and its temporary session copy.
-    }
-    expect(pi.latestSession().closeCalls).toBe(1);
+    await session.close();
 
     expect(existsSync(path.dirname(sessionCopy))).toBe(false);
     expect(readFileSync(archived.sessionFile)).toEqual(archived.originalBytes);

packages/server/src/server/agent/providers/pi/agent.ts

@@ -253,7 +253,6 @@ interface PiRpcAgentSessionOptions {
   currentModeId?: string | null;
   persistedSessionFile?: string;
   cleanup?: () => void;
-  closeAfterHistory?: boolean;
   extensionTimeoutMs?: number;
   logger: Logger;
   usagePollScheduler?: PiUsagePollScheduler;
@@ -1256,7 +1255,6 @@ export class PiRpcAgentSession implements AgentSession {
     this.currentModeId = options.currentModeId ?? null;
     this.persistedSessionFile = options.persistedSessionFile;
     this.cleanup = options.cleanup;
-    this.closeAfterHistory = options.closeAfterHistory ?? false;
     this.extensionTimeoutMs = options.extensionTimeoutMs ?? DEFAULT_PI_EXTENSION_RESULT_TIMEOUT_MS;
     this.logger = options.logger;
     this.extensionHost = createPiExtensionHost(this.logger);
@@ -1288,7 +1286,6 @@ export class PiRpcAgentSession implements AgentSession {
   private readonly config: AgentSessionConfig;
   private readonly persistedSessionFile?: string;
   private readonly cleanup?: () => void;
-  private readonly closeAfterHistory: boolean;
   private readonly extensionTimeoutMs: number;
 
   get id(): string | null {
@@ -1457,20 +1454,16 @@ export class PiRpcAgentSession implements AgentSession {
   }
 
   async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
-    try {
-      await this.requestEntryCapture("history");
-      yield* streamPiHistory(
-        this.provider,
-        await this.runtimeSession.getMessages(),
-        this.contextUserEntries,
-        {},
-        // At most eight 2 MiB child files per replay; later cards retain their summaries.
-        createPiExtensionHost(this.logger, undefined, 16 * 1024 * 1024),
-        this.closeController.signal,
-      );
-    } finally {
-      if (this.closeAfterHistory) await this.close();
-    }
+    await this.requestEntryCapture("history");
+    yield* streamPiHistory(
+      this.provider,
+      await this.runtimeSession.getMessages(),
+      this.contextUserEntries,
+      {},
+      // At most eight 2 MiB child files per replay; later cards retain their summaries.
+      createPiExtensionHost(this.logger, undefined, 16 * 1024 * 1024),
+      this.closeController.signal,
+    );
   }
 
   async getRuntimeInfo(): Promise<AgentRuntimeInfo> {
@@ -2745,7 +2738,6 @@ export class PiRpcAgentClient implements AgentClient {
         capabilities: capabilitiesForSession(mcp !== null),
         persistedSessionFile: sessionCopy ? sessionFile : undefined,
         cleanup: cleanupTempFiles,
-        closeAfterHistory: options?.purpose === "history",
         extensionTimeoutMs: providerOptions.extensionTimeoutMs,
         logger: this.logger,
         usagePollScheduler: this.usagePollScheduler,

packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts

@@ -153,7 +153,6 @@ export class FakePiSession implements PiRuntimeSession {
   // The user entries on the current branch that getMessages() replays.
   contextUserEntries: FakePiUserEntry[] = [];
   abortRequested = false;
-  closeCalls = 0;
   readonly canceledExtensionUiRequests: string[] = [];
   readonly extensionUiResponses: Array<{
     id: string;
@@ -162,7 +161,6 @@ export class FakePiSession implements PiRuntimeSession {
   setModelResult: PiModel | null = null;
   models: PiModel[] = [];
   messages: PiAgentMessage[] = [];
-  getMessagesError: Error | null = null;
   stats: PiSessionStats = {
     tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
     cost: 0,
@@ -305,7 +303,6 @@ export class FakePiSession implements PiRuntimeSession {
   }
 
   async getMessages(): Promise<PiAgentMessage[]> {
-    if (this.getMessagesError) throw this.getMessagesError;
     return this.messages;
   }
 
@@ -425,9 +422,7 @@ export class FakePiSession implements PiRuntimeSession {
     this.respondToExtensionUiRequest(id, { cancelled: true });
   }
 
-  async close(): Promise<void> {
-    this.closeCalls += 1;
-  }
+  async close(): Promise<void> {}
 
   emit(event: PiRuntimeEvent): void {
     for (const subscriber of this.subscribers) {

packages/server/src/server/authorization/operation-permissions.ts

@@ -142,7 +142,6 @@ const INBOUND_PERMISSION = {
   "plugin.reload.request": "daemon.manage",
   "plugin.remove.request": "daemon.manage",
   "plugin.rpc.invoke.request": "daemon.manage",
-  "plugin.rpc.subscribe.request": "daemon.manage",
   "plugin.source.install.request": "daemon.manage",
   "plugin.source.status.request": "daemon.manage",
   "plugin.source.update.request": "daemon.manage",
@@ -379,9 +378,6 @@ const OUTBOUND_PERMISSION = {
   "plugin.reload.response": "daemon.manage",
   "plugin.remove.response": "daemon.manage",
   "plugin.rpc.invoke.response": "daemon.manage",
-  "plugin.rpc.subscribe.response": "daemon.manage",
-  "plugin.rpc.event": "daemon.manage",
-  "plugin.rpc.closed": "daemon.manage",
   "plugin.source.install.response": "daemon.manage",
   "plugin.source.status.response": "daemon.manage",
   "plugin.source.update.response": "daemon.manage",

packages/server/src/server/connected-hosts/hosts.e2e.test.ts

@@ -2357,74 +2357,3 @@ test("a pre-write host budget rejection leaves sibling routes connected", async
     true,
   ]);
 }, 30000);
-
-test.each([false, true])(
-  "remote prompt to a closed agent (archived=%s) reports its new turn instead of stale closure",
-  { timeout: 30000 },
-  async (archived) => {
-    const parentPrompts: string[] = [];
-    const started = Promise.withResolvers<void>();
-    const finish = Promise.withResolvers<void>();
-    const a = await daemon({
-      agentClients: createTestAgentClients({
-        onStartTurn: (prompt) => parentPrompts.push(String(prompt)),
-      }),
-    });
-    const b = await daemon({
-      agentClients: createTestAgentClients({
-        onStartTurn: () => started.resolve(),
-        beforeTurnComplete: () => finish.promise,
-      }),
-    });
-    cleanup.push(async () => finish.resolve());
-    const origin = await api(a);
-    const destination = await api(b);
-    await origin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` });
-    const remoteId = (await online(origin.paseo)).hosts[0]!.serverId!;
-    const workspace = await origin.paseo.workspaces.create({
-      source: { kind: "directory", path: a.staticDir },
-    });
-    const remoteWorkspace = await destination.paseo.workspaces.create({
-      source: { kind: "directory", path: b.staticDir },
-    });
-    const parent = await workspace.agents.create({
-      config: { provider: "claude/claude-test", modeId: "full-access" },
-      title: "Closed target caller",
-    });
-    const child = await remoteWorkspace.agents.create({
-      config: { provider: "claude/claude-test", modeId: "full-access" },
-      title: "Closed target",
-    });
-    if (archived) await child.archive();
-    else await b.daemon.agentManager.closeAgent(child.id);
-    expect((await destination.client.fetchAgent(child.id))?.agent.status).toBe("closed");
-    const mcp = new Client({ name: "closed-target-caller", version: "1" });
-    await mcp.connect(
-      new StreamableHTTPClientTransport(
-        new URL(`http://127.0.0.1:${a.port}/mcp/agents?callerAgentId=${parent.id}`),
-      ),
-    );
-    cleanup.push(() => mcp.close());
-    const result = await mcp.callTool({
-      name: "send_agent_prompt",
-      arguments: { serverId: remoteId, agentId: child.id, prompt: "Finish the new task" },
-    });
-    expect(result.isError).not.toBe(true);
-    expect(result.structuredContent).toMatchObject({ deliveryStatus: "queued", status: "closed" });
-    await started.promise;
-    expect(parentPrompts).toEqual([]);
-    finish.resolve();
-    await expect.poll(() => parentPrompts, { timeout: 10000 }).toHaveLength(1);
-    expect(parentPrompts[0]).toContain('event="finished"');
-    expect(parentPrompts[0]).not.toContain('event="closed"');
-    expect((await destination.client.fetchAgent(child.id))?.agent).toMatchObject({
-      status: "idle",
-      archivedAt: null,
-    });
-    await expect
-      .poll(async () =>
-        JSON.parse(await readFile(join(a.paseoHome, "finish-subscriptions.json"), "utf8")),
-      )
-      .toEqual([]);
-  },
-);

packages/server/src/server/daemon-e2e/cli-logs-follow.e2e.test.ts

@@ -1,10 +1,7 @@
 import { spawn } from "node:child_process";
 import { once } from "node:events";
-import { mkdir, mkdtemp } from "node:fs/promises";
-import { join, resolve } from "node:path";
 import { fileURLToPath } from "node:url";
 import { expect, test } from "vitest";
-import { WebSocket, WebSocketServer } from "ws";
 import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
 import { DaemonClient } from "../test-utils/daemon-client.js";
 
@@ -140,225 +137,3 @@ test("omitted after limit pages forward while explicit zero preserves complete C
     await daemon.close();
   }
 }, 60_000);
-
-test.each([
-  { label: "--json", flags: ["--json"] },
-  { label: "--format json", flags: ["--format", "json"] },
-])(
-  "logs --follow $label keeps history, live items and notices on their channels",
-  async ({ flags }) => {
-    await mkdir(".artifacts", { recursive: true });
-    const paseoHomeRoot = await mkdtemp(resolve(".artifacts/logs-follow-json-"));
-    const staticDir = join(paseoHomeRoot, "static");
-    await mkdir(staticDir);
-    const daemon = await createTestPaseoDaemon({ paseoHomeRoot, staticDir });
-    const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
-    let follow: ReturnType<typeof spawn> | undefined;
-    let stdout = "";
-    let stderr = "";
-    try {
-      await client.connect();
-      const agent = await client.createAgent({ provider: "claude", cwd: daemon.paseoHome });
-      const history = [
-        { type: "user_message" as const, text: "Saved JSON prompt" },
-        { type: "assistant_message" as const, text: "Saved JSON reply", messageId: "saved" },
-      ];
-      for (const item of history)
-        await daemon.daemon.agentManager.appendTimelineItem(agent.id, item);
-      follow = spawn(
-        process.execPath,
-        [
-          "--conditions=source",
-          "--import",
-          "tsx",
-          fileURLToPath(new URL("../../../../cli/src/index.ts", import.meta.url)),
-          "--host",
-          `127.0.0.1:${daemon.port}`,
-          "logs",
-          agent.id,
-          "--follow",
-          ...flags,
-        ],
-        {
-          env: { ...process.env, PASEO_HOME: daemon.paseoHome },
-          stdio: ["ignore", "pipe", "pipe"],
-        },
-      );
-      const exited = once(follow, "exit");
-      follow.stdout?.on("data", (chunk: Buffer) => {
-        stdout += chunk.toString();
-      });
-      follow.stderr?.on("data", (chunk: Buffer) => {
-        stderr += chunk.toString();
-      });
-      await expect
-        .poll(
-          () => {
-            if (follow?.exitCode !== null)
-              throw new Error(`CLI exited before following logs: ${stderr}`);
-            return stdout + stderr;
-          },
-          { timeout: 20_000 },
-        )
-        .toContain("Following logs");
-      const live = {
-        type: "assistant_message" as const,
-        text: "Live JSON reply",
-        messageId: "live",
-      };
-      await daemon.daemon.agentManager.appendTimelineItem(agent.id, live);
-      await expect.poll(() => stdout, { timeout: 10_000 }).toContain(live.text);
-      follow.kill("SIGINT");
-      await exited;
-      expect(
-        stdout
-          .trim()
-          .split("\n")
-          .map((line) => JSON.parse(line)),
-      ).toEqual([...history, live]);
-      expect(stderr).toContain("Following logs");
-      expect(stdout).not.toContain("Following logs");
-    } finally {
-      follow?.kill("SIGTERM");
-      await client.close();
-      await daemon.close();
-    }
-  },
-  60_000,
-);
-
-test.each([
-  { label: "--json", command: ["logs"], flags: ["--json"] },
-  { label: "--format json", command: ["agent", "logs"], flags: ["--format", "json"] },
-])(
-  "logs --follow $label exits when stdout closes before saved history is written",
-  async ({ command, flags }) => {
-    await mkdir(".artifacts", { recursive: true });
-    const paseoHomeRoot = await mkdtemp(resolve(".artifacts/logs-follow-closed-"));
-    const staticDir = join(paseoHomeRoot, "static");
-    await mkdir(staticDir);
-    const daemon = await createTestPaseoDaemon({ paseoHomeRoot, staticDir });
-    const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
-    let follow: ReturnType<typeof spawn> | undefined;
-    let stderr = "";
-    try {
-      await client.connect();
-      // No turn is started: the saved history must trigger shutdown on its own.
-      const agent = await client.createAgent({ provider: "claude", cwd: daemon.paseoHome });
-      await daemon.daemon.agentManager.appendTimelineItem(agent.id, {
-        type: "assistant_message",
-        text: "Saved reply before the reader closes",
-        messageId: "saved",
-      });
-      follow = spawn(
-        process.execPath,
-        [
-          "--conditions=source",
-          "--import",
-          "tsx",
-          fileURLToPath(new URL("../../../../cli/src/index.ts", import.meta.url)),
-          "--host",
-          `127.0.0.1:${daemon.port}`,
-          ...command,
-          agent.id,
-          "--follow",
-          ...flags,
-        ],
-        {
-          env: { ...process.env, PASEO_HOME: daemon.paseoHome },
-          stdio: ["ignore", "pipe", "pipe"],
-        },
-      );
-      const exited = once(follow, "exit");
-      follow.stderr?.on("data", (chunk: Buffer) => {
-        stderr += chunk.toString();
-      });
-      follow.stdout?.destroy();
-      await expect.poll(() => follow?.exitCode, { timeout: 10_000 }).toBe(0);
-      expect(await exited, stderr).toEqual([0, null]);
-    } finally {
-      follow?.kill("SIGKILL");
-      await client.close();
-      await daemon.close();
-    }
-  },
-  30_000,
-);
-
-// Windows kill("SIGINT") force-terminates rather than delivering a console interrupt.
-test.skipIf(process.platform === "win32")(
-  "logs --follow stops on SIGINT while subscription startup is waiting",
-  async () => {
-    await mkdir(".artifacts", { recursive: true });
-    const paseoHomeRoot = await mkdtemp(resolve(".artifacts/logs-follow-startup-"));
-    const staticDir = join(paseoHomeRoot, "static");
-    await mkdir(staticDir);
-    const daemon = await createTestPaseoDaemon({ paseoHomeRoot, staticDir });
-    const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
-    const proxy = new WebSocketServer({ port: 0, host: "127.0.0.1" });
-    await once(proxy, "listening");
-    const address = proxy.address();
-    if (typeof address === "string" || address === null) throw new Error("Missing proxy port");
-    const upstreams: WebSocket[] = [];
-    let waitingForSubscription = false;
-    proxy.on("connection", (downstream) => {
-      const upstream = new WebSocket(`ws://127.0.0.1:${daemon.port}/ws`);
-      upstreams.push(upstream);
-      const connected = once(upstream, "open");
-      downstream.on("message", async (data, binary) => {
-        await connected;
-        const frame = binary ? undefined : JSON.parse(data.toString());
-        // Hold only the subscription request. All other traffic uses the real daemon.
-        if (frame?.message?.type === "agent.timeline.set_subscription.request") {
-          waitingForSubscription = true;
-          return;
-        }
-        upstream.send(data, { binary });
-      });
-      upstream.on("message", (data, binary) => downstream.send(data, { binary }));
-      downstream.on("close", () => upstream.close());
-    });
-    let follow: ReturnType<typeof spawn> | undefined;
-    try {
-      await client.connect();
-      const agent = await client.createAgent({ provider: "claude", cwd: daemon.paseoHome });
-      follow = spawn(
-        process.execPath,
-        [
-          "--conditions=source",
-          "--import",
-          "tsx",
-          fileURLToPath(new URL("../../../../cli/src/index.ts", import.meta.url)),
-          "--host",
-          `127.0.0.1:${address.port}`,
-          "logs",
-          agent.id,
-          "--follow",
-          "--json",
-        ],
-        {
-          env: { ...process.env, PASEO_HOME: daemon.paseoHome },
-          stdio: ["ignore", "pipe", "pipe"],
-        },
-      );
-      follow.stdout?.resume();
-      follow.stderr?.resume();
-      const exited = once(follow, "exit");
-      await expect.poll(() => waitingForSubscription, { timeout: 10_000 }).toBe(true);
-      follow.kill("SIGINT");
-      await expect.poll(() => follow?.exitCode, { timeout: 5_000 }).toBe(0);
-      expect(await exited).toEqual([0, null]);
-    } finally {
-      if (follow && follow.exitCode === null && follow.signalCode === null) {
-        follow.kill("SIGKILL");
-        await once(follow, "exit");
-      }
-      for (const upstream of upstreams) upstream.terminate();
-      for (const downstream of proxy.clients) downstream.terminate();
-      await new Promise<void>((done) => proxy.close(() => done()));
-      await client.close();
-      await daemon.close();
-    }
-  },
-  30_000,
-);

packages/server/src/server/plugins/index.posix.test.ts

@@ -15,7 +15,7 @@ import { tmpdir } from "node:os";
 import path from "node:path";
 import { pathToFileURL } from "node:url";
 import pino from "pino";
-import { afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest";
+import { afterEach, beforeAll, describe, expect, it } from "vitest";
 import { DaemonConfigStore } from "../daemon-config-store.js";
 import { PluginService } from "./index.js";
 import { BuiltinPluginLoader } from "./builtin/index.js";
@@ -34,13 +34,7 @@ const emptyUsageRuntime = {
   getUsageSourceRegistrations: () => [],
   fetchUsage: async () => undefined,
   discoverUsage: async () => [],
-  startSubscription: () => {
-    throw new Error("No subscriptions in this test");
-  },
-} satisfies Pick<
-  TestPluginRuntime,
-  "getUsageSourceRegistrations" | "fetchUsage" | "discoverUsage" | "startSubscription"
->;
+} satisfies Pick<TestPluginRuntime, "getUsageSourceRegistrations" | "fetchUsage" | "discoverUsage">;
 
 async function createPlugin(id: string, source: string): Promise<string> {
   const directory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-service-"));
@@ -638,6 +632,91 @@ describe("PluginService", () => {
     await service.stopAllPlugins();
   });
 
+  it("resolves a bare registry id even when a matching directory exists in the daemon cwd", async () => {
+    const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
+    const owner = path.join(process.cwd(), `registry-shadow-${randomUUID()}`);
+    await mkdir(owner);
+    roots.push(home, owner);
+    const source = `${path.basename(owner)}/example`;
+    const repository = await createPlugin("registry-example", "export default () => () => {};\n");
+    await mkdir(path.join(owner, "example"));
+    await cp(repository, path.join(owner, "example"), { recursive: true });
+    await runGitCommand(["init", "-b", "main"], { cwd: repository });
+    await runGitCommand(["add", "-A"], { cwd: repository });
+    await runGitCommand(
+      [
+        "-c",
+        "user.name=Paseo Tests",
+        "-c",
+        "user.email=paseo@example.test",
+        "commit",
+        "-m",
+        "fixture",
+      ],
+      { cwd: repository },
+    );
+    const { stdout } = await runGitCommand(["rev-parse", "HEAD"], { cwd: repository });
+    const requests: Array<string | undefined> = [];
+    const server = createServer((request, response) => {
+      requests.push(request.url);
+      response.setHeader("Content-Type", "application/json");
+      response.end(
+        JSON.stringify({
+          id: source,
+          name: "Example",
+          description: "Test",
+          categories: [],
+          author: { github: path.basename(owner) },
+          repository: { url: "https://github.com/fixture/example" },
+          artifact: {
+            kind: "git",
+            remote: pathToFileURL(repository).href,
+            commit: stdout.trim(),
+            pluginPath: ".",
+          },
+          media: [],
+          submittedAt: "2026-10-03",
+          reviewedAt: "2026-10-03",
+          updatedAt: "2026-10-03",
+          publishedAt: "2026-10-03",
+          readme: "# Test",
+        }),
+      );
+    });
+    await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+    const address = server.address();
+    if (!address || typeof address === "string") throw new Error("Missing registry address");
+    const url = `http://127.0.0.1:${address.port}`;
+    const store = createStore(home);
+    await store.patch({ pluginsEnabled: false });
+    const service = bindTestSessionHost(
+      new PluginService(pino({ level: "silent" }), store, "0.4.0", {
+        managedSources: new ManagedPluginSources(home, { defaultUrl: url }),
+      }),
+    );
+    try {
+      await service.start();
+      await expect(service.installSource({ source })).resolves.toMatchObject({
+        id: "registry-example",
+        status: "disabled",
+        installation: {
+          identity: { kind: "git", registry: { url, id: source } },
+          currentRevision: stdout.trim(),
+        },
+      });
+      expect(requests).toEqual([`/plugins/${source}.json`]);
+      await expect(
+        service.installSource({ source: `./${source}`, id: "local-example" }),
+      ).resolves.toMatchObject({
+        installation: { identity: { kind: "directory", path: path.join(owner, "example") } },
+      });
+      expect(requests).toEqual([`/plugins/${source}.json`]);
+    } finally {
+      await service.stopAllPlugins();
+      await new Promise<void>((resolve) => server.close(() => resolve()));
+    }
+  }, 30_000);
+
   it("names a missing explicit directory instead of trying a managed source", async () => {
     const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
     roots.push(home);
@@ -1441,105 +1520,6 @@ export default function contribute(plugin: unknown) {
   });
 });
 
-describe("PluginService registry source resolution", () => {
-  let home: string;
-  let owner: string;
-  let source: string;
-  let repository: string;
-  let revision: string;
-
-  // Prepare the real Git fixture before the install deadline starts. On Windows,
-  // these four Git launches can consume much of the test’s 30-second budget.
-  beforeEach(async () => {
-    home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
-    owner = path.join(process.cwd(), `registry-shadow-${randomUUID()}`);
-    await mkdir(owner);
-    roots.push(home, owner);
-    source = `${path.basename(owner)}/example`;
-    repository = await createPlugin("registry-example", "export default () => () => {};\n");
-    await mkdir(path.join(owner, "example"));
-    await cp(repository, path.join(owner, "example"), { recursive: true });
-    await runGitCommand(["init", "-b", "main"], { cwd: repository });
-    await runGitCommand(["add", "-A"], { cwd: repository });
-    await runGitCommand(
-      [
-        "-c",
-        "user.name=Paseo Tests",
-        "-c",
-        "user.email=paseo@example.test",
-        "commit",
-        "-m",
-        "fixture",
-      ],
-      { cwd: repository },
-    );
-    const { stdout } = await runGitCommand(["rev-parse", "HEAD"], { cwd: repository });
-    revision = stdout.trim();
-  });
-
-  it("resolves a bare registry id even when a matching directory exists in the daemon cwd", async () => {
-    const requests: Array<string | undefined> = [];
-    const server = createServer((request, response) => {
-      requests.push(request.url);
-      response.setHeader("Content-Type", "application/json");
-      response.end(
-        JSON.stringify({
-          id: source,
-          name: "Example",
-          description: "Test",
-          categories: [],
-          author: { github: path.basename(owner) },
-          repository: { url: "https://github.com/fixture/example" },
-          artifact: {
-            kind: "git",
-            remote: pathToFileURL(repository).href,
-            commit: revision,
-            pluginPath: ".",
-          },
-          media: [],
-          submittedAt: "2026-10-03",
-          reviewedAt: "2026-10-03",
-          updatedAt: "2026-10-03",
-          publishedAt: "2026-10-03",
-          readme: "# Test",
-        }),
-      );
-    });
-    await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
-    const address = server.address();
-    if (!address || typeof address === "string") throw new Error("Missing registry address");
-    const url = `http://127.0.0.1:${address.port}`;
-    const store = createStore(home);
-    await store.patch({ pluginsEnabled: false });
-    const service = bindTestSessionHost(
-      new PluginService(pino({ level: "silent" }), store, "0.4.0", {
-        managedSources: new ManagedPluginSources(home, { defaultUrl: url }),
-      }),
-    );
-    try {
-      await service.start();
-      await expect(service.installSource({ source })).resolves.toMatchObject({
-        id: "registry-example",
-        status: "disabled",
-        installation: {
-          identity: { kind: "git", registry: { url, id: source } },
-          currentRevision: revision,
-        },
-      });
-      expect(requests).toEqual([`/plugins/${source}.json`]);
-      await expect(
-        service.installSource({ source: `./${source}`, id: "local-example" }),
-      ).resolves.toMatchObject({
-        installation: { identity: { kind: "directory", path: path.join(owner, "example") } },
-      });
-      expect(requests).toEqual([`/plugins/${source}.json`]);
-    } finally {
-      await service.stopAllPlugins();
-      await new Promise<void>((resolve) => server.close(() => resolve()));
-    }
-  }, 30_000);
-});
-
 describe("npm plugin installation", () => {
   let registry: Awaited<ReturnType<typeof startNpmRegistry>>;
 

packages/server/src/server/plugins/index.ts

@@ -25,7 +25,6 @@ import { readPluginManifest } from "./manifest.js";
 import { expandTilde } from "../../utils/path.js";
 import { runPluginBuild } from "./preparation.js";
 import { PluginRuntime } from "./runtime.js";
-import type { PluginRuntimeSubscription } from "./runtime-subscription.js";
 import { BuiltinPluginLoader, type BuiltinPlugin } from "./builtin/index.js";
 import type { PluginProviderMetadata } from "./plugin-process-protocol.js";
 import { readPluginProviderIcon } from "./provider-icon.js";
@@ -49,7 +48,6 @@ interface PluginRuntimePort {
   wait?: PluginLifecycle["wait"];
   catalog: PluginRuntime["catalog"];
   invoke(pluginId: string, method: string, input: unknown): Promise<unknown>;
-  startSubscription: PluginRuntime["startSubscription"];
   getLogs(pluginId: string): PluginLogEntry[];
   clearLogs(pluginId: string): void;
   getProviderRegistrations?(pluginId: string): readonly PluginProviderMetadata[];
@@ -541,10 +539,6 @@ export class PluginService {
     return this.runtime.invoke(pluginId, method, input);
   }
 
-  subscribePluginRpc(pluginId: string, method: string, input: unknown): PluginRuntimeSubscription {
-    return this.runtime.startSubscription(pluginId, method, input);
-  }
-
   async stopAllPlugins(): Promise<void> {
     this.globalStartsBlocked = true;
     const stopping = this.stopAll();

packages/server/src/server/plugins/plugin-process-protocol.ts

@@ -60,14 +60,6 @@ export type PluginProcessRequest =
   | { type: "usage.fetch"; requestId: string; sourceId: string; input: unknown }
   | { type: "usage.discover"; requestId: string; sourceId: string; scope: UsageScope }
   | { type: "invoke"; requestId: string; method: string; input: unknown }
-  | {
-      type: "subscribe";
-      requestId: string;
-      subscriptionId: string;
-      method: string;
-      input: unknown;
-    }
-  | { type: "unsubscribe"; subscriptionId: string }
   | {
       type: "provider.connect";
       providerId: string;
@@ -91,7 +83,6 @@ export type PluginProcessMessage =
   | {
       type: "ready";
       methods: string[];
-      subscriptions?: string[];
       providers: PluginProviderMetadata[];
       usageSources?: PluginUsageSourceMetadata[];
       hooks?: string[];
@@ -99,8 +90,6 @@ export type PluginProcessMessage =
   | { type: "result"; requestId: string; output: unknown }
   | { type: "error"; requestId: string; error: string }
   | { type: "fatal"; error: string }
-  | { type: "subscription.event"; subscriptionId: string; event: unknown }
-  | { type: "subscription.closed"; subscriptionId: string; error?: string }
   | {
       type: "provider.connected";
       connectionId: string;
@@ -251,16 +240,6 @@ export const PluginProcessRequestSchema: z.ZodType<PluginProcessRequest> = z.dis
         input: z.unknown(),
       })
       .strict(),
-    z
-      .object({
-        type: z.literal("subscribe"),
-        requestId: z.string().min(1),
-        subscriptionId: z.string().min(1),
-        method: z.string().min(1),
-        input: z.unknown(),
-      })
-      .strict(),
-    z.object({ type: z.literal("unsubscribe"), subscriptionId: z.string().min(1) }).strict(),
     z
       .object({
         type: z.literal("provider.connect"),
@@ -293,7 +272,6 @@ export const PluginProcessMessageSchema: z.ZodType<PluginProcessMessage> = z.dis
       .object({
         type: z.literal("ready"),
         methods: z.array(z.string()),
-        subscriptions: z.array(z.string()).optional(),
         providers: z.array(providerMetadataSchema),
         usageSources: z.array(usageSourceMetadataSchema).optional(),
         hooks: hooksSchema.optional(),
@@ -306,20 +284,6 @@ export const PluginProcessMessageSchema: z.ZodType<PluginProcessMessage> = z.dis
       .object({ type: z.literal("error"), requestId: z.string().min(1), error: z.string() })
       .strict(),
     z.object({ type: z.literal("fatal"), error: z.string() }).strict(),
-    z
-      .object({
-        type: z.literal("subscription.event"),
-        subscriptionId: z.string().min(1),
-        event: z.unknown(),
-      })
-      .strict(),
-    z
-      .object({
-        type: z.literal("subscription.closed"),
-        subscriptionId: z.string().min(1),
-        error: z.string().optional(),
-      })
-      .strict(),
     z
       .object({
         type: z.literal("provider.connected"),

packages/server/src/server/plugins/plugin-process.ts

@@ -7,21 +7,8 @@ import {
   type PluginProcessRequest,
 } from "./plugin-process-protocol.js";
 import type { UsageSourceRegistration } from "@getpaseo/plugin/server/usage";
-import type {
-  SettingsDefinition,
-  PluginRpcContract,
-  PluginSubscriptionContract,
-} from "@getpaseo/plugin";
-import type {
-  PluginHandlerContext,
-  PluginServerContext,
-  PluginServerContribution,
-} from "@getpaseo/plugin/server";
-import { assertPluginWireValue } from "@getpaseo/protocol/binary-frames/index";
-import {
-  PluginSubscriptionHandlers,
-  type PluginSubscriptionHandler,
-} from "./subscription-handlers.js";
+import type { SettingsDefinition, PluginRpcContract } from "@getpaseo/plugin";
+import type { PluginHandlerContext, PluginServerContribution } from "@getpaseo/plugin/server";
 import { fileURLToPath } from "node:url";
 import type { ZodType } from "zod";
 import {
@@ -79,7 +66,6 @@ export function createPluginWorker(options: {
     send({ type: "hooks.changed", hooks: hooks.catalog() });
   }, agentTools);
   const handlers = new Map<string, RegisteredRpc>();
-  const subscriptions = new PluginSubscriptionHandlers({ send, sendAndWait });
   const providers = new Map<string, ProviderRegistration>();
   const usageSources = new Map<string, UsageSourceRegistration>();
   const providerConnections = new Map<
@@ -112,36 +98,12 @@ export function createPluginWorker(options: {
     if (!/^[a-z][a-z0-9._-]*$/.test(normalized)) {
       throw new Error(`Invalid plugin RPC method: ${method}`);
     }
-    if (handlers.has(normalized) || subscriptions.has(normalized)) {
+    if (handlers.has(normalized)) {
       throw new Error(`Duplicate plugin RPC method: ${normalized}`);
     }
     return normalized;
   }
 
-  function registerSubscription(
-    contract: PluginSubscriptionContract,
-    handler: PluginSubscriptionHandler,
-  ): void {
-    if (typeof handler !== "function") {
-      throw new Error(`Plugin subscription ${contract.name} must provide a handler`);
-    }
-    const method = validateMethod(contract.name);
-    subscriptions.register({ ...contract, name: method }, handler);
-  }
-
-  // One name serves either an RPC or a subscription; the contract says which.
-  const handle = ((
-    contract: PluginRpcContract | PluginSubscriptionContract,
-    handler: (input: unknown, context: never) => unknown,
-  ) => {
-    if ("event" in contract) {
-      // The subscription handlers parse the input once and pass the result here.
-      registerSubscription(contract, handler as PluginSubscriptionHandler);
-      return;
-    }
-    register(contract, (input, context) => handler(input, context as never));
-  }) as PluginServerContext["handle"];
-
   function register(contract: PluginRpcContract, handler: RpcHandler): void {
     if (typeof handler !== "function") {
       throw new Error(`Plugin RPC ${contract.name} must provide a handler`);
@@ -322,7 +284,8 @@ export function createPluginWorker(options: {
         )
       : null;
     const contributedCleanup = contribute({
-      handle,
+      handle: (contract, handler) =>
+        register(contract, (input, context) => handler(contract.input.parse(input), context)),
       registerProvider,
       registerUsageSource,
       registerSettings,
@@ -348,7 +311,6 @@ export function createPluginWorker(options: {
     send({
       type: "ready",
       methods: [...handlers.keys()].sort(),
-      subscriptions: subscriptions.catalog(),
       hooks: hooks.catalog(),
       providers: [...providers.values()]
         .sort((left, right) => left.id.localeCompare(right.id))
@@ -365,7 +327,6 @@ export function createPluginWorker(options: {
       .catch((error) => console.error("Plugin API cleanup failed", error));
     hooks.close();
     agentTools.close();
-    subscriptions.releaseAll();
     for (const pending of pendingProviderConnections.values()) pending.tombstoned = true;
     const currentCleanup = cleanup;
     cleanup = null;
@@ -404,8 +365,7 @@ export function createPluginWorker(options: {
       message.type === "provider.status" ||
       message.type === "provider.catalog_key" ||
       message.type === "usage.fetch" ||
-      message.type === "usage.discover" ||
-      message.type === "subscribe"
+      message.type === "usage.discover"
     ) {
       send({ type: "error", requestId: message.requestId, error: "Plugin is stopping" });
     } else if (message.type === "provider.connect") {
@@ -462,14 +422,6 @@ export function createPluginWorker(options: {
     return false;
   }
 
-  function handleSubscriptionMessage(
-    message: Extract<PluginProcessRequest, { type: "subscribe" | "unsubscribe" }>,
-  ): void {
-    if (message.type === "unsubscribe") subscriptions.release(message.subscriptionId);
-    else if (paseo) subscriptions.start(message, paseo);
-    else send({ type: "error", requestId: message.requestId, error: "Plugin API unavailable" });
-  }
-
   function handleMessage(rawMessage: unknown): void {
     const parsed = PluginProcessRequestSchema.safeParse(rawMessage);
     if (!parsed.success) {
@@ -573,11 +525,6 @@ export function createPluginWorker(options: {
       handleHookMessage(message);
       return;
     }
-    if (message.type === "invoke") invoke(message);
-    else handleSubscriptionMessage(message);
-  }
-
-  function invoke(message: Extract<PluginProcessRequest, { type: "invoke" }>): void {
     const registered = handlers.get(message.method);
     if (!registered) {
       send({
@@ -593,10 +540,7 @@ export function createPluginWorker(options: {
         if (!paseo) throw new Error("Plugin Paseo API is unavailable");
         return registered.handler(input, { paseo });
       })
-      .then((output: unknown) => {
-        assertPluginWireValue(output, "output");
-        return output;
-      })
+      .then((output) => registered.contract.output.parseAsync(output))
       .then(
         (output) => send({ type: "result", requestId: message.requestId, output }),
         (error) =>

packages/server/src/server/plugins/plugin-subscriptions.e2e.test.ts

@@ -1,428 +0,0 @@
-import { mkdtemp, rm, writeFile } from "node:fs/promises";
-import { tmpdir } from "node:os";
-import path from "node:path";
-import { WebSocket } from "ws";
-import {
-  DaemonClient,
-  type OwnedSubscription,
-  type WebSocketLike,
-} from "@getpaseo/client/internal/daemon-client";
-import { afterEach, expect, test } from "vitest";
-import { defineRpc } from "@getpaseo/plugin";
-import { callPluginRpc } from "@getpaseo/plugin/client/host";
-import { z } from "zod";
-import { resolveDaemonVersion } from "../daemon-version.js";
-import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js";
-
-const PLUGIN_ID = "subscriptions";
-
-// Written as an author writes a plugin: one contract per name, handlers that read like English.
-const SERVER_ENTRY = `
-import { defineRpc, defineSubscription } from "@getpaseo/plugin";
-import type { PluginServerContext } from "@getpaseo/plugin/server";
-import { z } from "zod";
-
-const ticks = defineSubscription({
-  name: "ticks",
-  input: z.object({ count: z.number().int(), bytes: z.boolean() }),
-  event: z.union([z.instanceof(Uint8Array), z.object({ n: z.number() })]),
-});
-const failing = defineSubscription({
-  name: "failing",
-  input: z.object({ when: z.enum(["before-ack", "after-ack"]) }),
-  event: z.object({}),
-});
-const nestedPush = defineSubscription({
-  name: "nested-push",
-  input: z.object({}),
-  event: z.object({ data: z.instanceof(Uint8Array) }),
-});
-const repeated = defineSubscription({
-  name: "repeated",
-  input: z.object({ times: z.string().transform(Number) }),
-  event: z.string().transform((text) => text.length),
-});
-const transformed = defineRpc({
-  name: "transformed",
-  input: z.object({ when: z.string().transform(async (text) => new Date(text)) }),
-  output: z.object({ n: z.string().transform(async (text) => Number(text)) }),
-});
-const defaulted = defineRpc({ name: "defaulted", input: z.string().default("x"), output: z.string() });
-const defaultedOutput = defineRpc({ name: "defaulted-output", input: z.object({}), output: z.string().default("x") });
-const watchers = defineRpc({
-  name: "watchers",
-  input: z.object({}),
-  output: z.object({ starts: z.number(), active: z.number() }),
-});
-const thumbnail = defineRpc({
-  name: "thumbnail",
-  input: z.object({ size: z.number().int() }),
-  output: z.instanceof(Uint8Array),
-});
-const nestedOutput = defineRpc({
-  name: "nested-output",
-  input: z.object({}),
-  output: z.object({ data: z.instanceof(Uint8Array) }),
-});
-const reverse = defineRpc({
-  name: "reverse",
-  input: z.instanceof(Uint8Array),
-  output: z.instanceof(Uint8Array),
-});
-
-export default function contribute(server: PluginServerContext) {
-  let starts = 0;
-  let active = 0;
-  server.handle(ticks, ({ count, bytes }, { push, signal }) => {
-    starts += 1;
-    active += 1;
-    signal.addEventListener("abort", () => { active -= 1; }, { once: true });
-    for (let n = 0; n < count; n += 1) void push(bytes ? new Uint8Array([n, n + 1]) : { n });
-  });
-  server.handle(failing, ({ when }) => {
-    if (when === "before-ack") throw new Error("refused before acknowledging");
-    return Promise.reject(new Error("failed after acknowledging"));
-  });
-  server.handle(nestedPush, (_input, { push }) => {
-    void push({ data: new Uint8Array(1) }).catch((error) => {
-      console.error("nested push", error.message);
-    });
-  });
-  server.handle(repeated, async ({ times }, { push }) => {
-    await push(typeof times === "number" ? "x".repeat(times) : "input arrived untransformed");
-  });
-  server.handle(transformed, ({ when }) => ({ n: String(when.getUTCFullYear()) }));
-  server.handle(defaulted, (input) => input);
-  server.handle(defaultedOutput, () => undefined);
-  server.handle(watchers, () => ({ starts, active }));
-  server.handle(thumbnail, ({ size }) => Buffer.alloc(size, 7));
-  server.handle(nestedOutput, () => ({ data: new Uint8Array(1) }));
-  server.handle(reverse, (bytes) => Uint8Array.from(bytes).reverse());
-  return () => undefined;
-}
-`;
-
-type Received =
-  | { type: "event"; event: unknown }
-  | { type: "bytes"; bytes: number[] }
-  | { type: "closed"; error?: string };
-
-const roots: string[] = [];
-
-afterEach(async () => {
-  await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
-});
-
-interface Harness {
-  daemon: TestPaseoDaemon;
-  client: DaemonClient;
-  sockets: WebSocket[];
-  close(): Promise<void>;
-}
-
-async function writePlugin(id: string, serverEntry: string): Promise<string> {
-  const pluginDirectory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-subscriptions-"));
-  roots.push(pluginDirectory);
-  await writeFile(
-    path.join(pluginDirectory, "paseo-plugin.json"),
-    JSON.stringify({ id, requirements: { paseo: `>=${resolveDaemonVersion(import.meta.url)}` } }),
-  );
-  await writeFile(path.join(pluginDirectory, "index.server.ts"), serverEntry);
-  return pluginDirectory;
-}
-
-async function startHarness(): Promise<Harness> {
-  const pluginDirectory = await writePlugin(PLUGIN_ID, SERVER_ENTRY);
-  const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
-  const sockets: WebSocket[] = [];
-  const client = new DaemonClient({
-    url: `ws://127.0.0.1:${daemon.port}/ws`,
-    clientId: `clid_plugin_subscriptions_${Date.now()}`,
-    appVersion: resolveDaemonVersion(import.meta.url),
-    reconnect: { enabled: true, baseDelayMs: 10, maxDelayMs: 50 },
-    webSocketFactory: (url) => {
-      const socket = new WebSocket(url);
-      sockets.push(socket);
-      return socket as unknown as WebSocketLike;
-    },
-  });
-  await client.connect();
-  await client.patchDaemonConfig({ pluginsEnabled: true });
-  await expect(client.installDirectoryPlugin(pluginDirectory)).resolves.toMatchObject({
-    id: PLUGIN_ID,
-    status: "running",
-  });
-  return {
-    daemon,
-    client,
-    sockets,
-    async close() {
-      await client.close().catch(() => undefined);
-      await daemon.close();
-    },
-  };
-}
-
-function collect(subscription: OwnedSubscription<unknown>): Received[] {
-  const received: Received[] = [];
-  subscription.subscribe({
-    snapshot: () => undefined,
-    update: (message) => {
-      if (message.type === "plugin.rpc.event")
-        received.push({ type: "event", event: message.payload.event });
-      if (message.type === "plugin.rpc.closed")
-        received.push({
-          type: "closed",
-          ...(message.payload.error ? { error: message.payload.error } : {}),
-        });
-    },
-    binary: (payload) => received.push({ type: "bytes", bytes: [...payload] }),
-  });
-  return received;
-}
-
-function watcherCount(client: DaemonClient) {
-  return client.invokePluginRpc(PLUGIN_ID, "watchers", {});
-}
-
-test("a subscription delivers JSON and binary events in order and release aborts its handler", async () => {
-  const harness = await startHarness();
-  const { client } = harness;
-  try {
-    const json = client.observePluginRpc(PLUGIN_ID, "ticks", { count: 3, bytes: false });
-    const jsonEvents = collect(json);
-    await json.ready;
-    const binary = client.observePluginRpc(PLUGIN_ID, "ticks", { count: 2, bytes: true });
-    const binaryEvents = collect(binary);
-    await binary.ready;
-
-    await expect
-      .poll(() => jsonEvents)
-      .toEqual([
-        { type: "event", event: { n: 0 } },
-        { type: "event", event: { n: 1 } },
-        { type: "event", event: { n: 2 } },
-      ]);
-    await expect
-      .poll(() => binaryEvents)
-      .toEqual([
-        { type: "bytes", bytes: [0, 1] },
-        { type: "bytes", bytes: [1, 2] },
-      ]);
-    expect(json.subscriptionId).not.toBe(binary.subscriptionId);
-    await expect(watcherCount(client)).resolves.toEqual({ starts: 2, active: 2 });
-
-    await json.release();
-    await expect(watcherCount(client)).resolves.toEqual({ starts: 2, active: 1 });
-    await binary.release();
-    await expect(watcherCount(client)).resolves.toEqual({ starts: 2, active: 0 });
-  } finally {
-    await harness.close();
-  }
-}, 60_000);
-
-test("a schema transform runs once: the handler gets the parsed input and the wire carries the pushed event", async () => {
-  const harness = await startHarness();
-  try {
-    const subscription = harness.client.observePluginRpc(PLUGIN_ID, "repeated", { times: "2" });
-    const received = collect(subscription);
-    await subscription.ready;
-    // The client parses "xx" into 2; parsing it on the plugin side too would reject it.
-    await expect.poll(() => received).toEqual([{ type: "event", event: "xx" }, { type: "closed" }]);
-  } finally {
-    await harness.close();
-  }
-}, 60_000);
-
-test("dropping the connection releases subscriptions and reconnecting runs the handler again", async () => {
-  const harness = await startHarness();
-  const { client, sockets } = harness;
-  try {
-    const subscription = client.observePluginRpc(PLUGIN_ID, "ticks", { count: 1, bytes: false });
-    const events = collect(subscription);
-    await subscription.ready;
-    await expect.poll(() => events).toEqual([{ type: "event", event: { n: 0 } }]);
-    const firstId = subscription.subscriptionId;
-
-    sockets.at(-1)!.terminate();
-    await expect
-      .poll(() => subscription.subscriptionId, { timeout: 10_000 })
-      .toSatisfy((id: unknown) => typeof id === "string" && id !== firstId);
-    await expect
-      .poll(() => events)
-      .toEqual([
-        { type: "event", event: { n: 0 } },
-        { type: "event", event: { n: 0 } },
-      ]);
-    // The first handler's signal aborted when its connection dropped; one handler runs now.
-    await expect(watcherCount(client)).resolves.toEqual({ starts: 2, active: 1 });
-    await subscription.release();
-  } finally {
-    await harness.close();
-  }
-}, 60_000);
-
-test("a handler that throws before acknowledging rejects; one that fails later closes", async () => {
-  const harness = await startHarness();
-  const { client } = harness;
-  try {
-    const refused = client.observePluginRpc(PLUGIN_ID, "failing", { when: "before-ack" });
-    await expect(refused.ready).rejects.toThrow("refused before acknowledging");
-
-    const later = client.observePluginRpc(PLUGIN_ID, "failing", { when: "after-ack" });
-    const events = collect(later);
-    await later.ready;
-    await expect
-      .poll(() => events)
-      .toEqual([{ type: "closed", error: "failed after acknowledging" }]);
-    await later.release();
-
-    const invalid = client.observePluginRpc(PLUGIN_ID, "ticks", { count: "many" });
-    await expect(invalid.ready).rejects.toThrow();
-    const unknown = client.observePluginRpc(PLUGIN_ID, "watchers", {});
-    await expect(unknown.ready).rejects.toThrow("does not contribute subscription watchers");
-  } finally {
-    await harness.close();
-  }
-}, 60_000);
-
-test("reloading the plugin closes its subscriptions with an error", async () => {
-  const harness = await startHarness();
-  const { client } = harness;
-  try {
-    const subscription = client.observePluginRpc(PLUGIN_ID, "ticks", { count: 0, bytes: false });
-    const events = collect(subscription);
-    await subscription.ready;
-    await client.reloadPlugin(PLUGIN_ID);
-    await expect
-      .poll(() => events)
-      .toEqual([{ type: "closed", error: `Plugin stopped: ${PLUGIN_ID}` }]);
-    await subscription.release();
-    // The reloaded plugin starts fresh and serves new subscriptions.
-    const again = client.observePluginRpc(PLUGIN_ID, "ticks", { count: 1, bytes: false });
-    const againEvents = collect(again);
-    await again.ready;
-    await expect.poll(() => againEvents).toEqual([{ type: "event", event: { n: 0 } }]);
-    await again.release();
-  } finally {
-    await harness.close();
-  }
-}, 60_000);
-
-test("RPC bytes ride as binary frames in both directions and nested bytes fail validation", async () => {
-  const harness = await startHarness();
-  const { client } = harness;
-  try {
-    const output = await client.invokePluginRpc(PLUGIN_ID, "thumbnail", { size: 4 });
-    expect(output).toBeInstanceOf(Uint8Array);
-    expect([...(output as Uint8Array)]).toEqual([7, 7, 7, 7]);
-
-    const reversed = await client.invokePluginRpc(PLUGIN_ID, "reverse", new Uint8Array([1, 2, 3]));
-    expect([...(reversed as Uint8Array)]).toEqual([3, 2, 1]);
-
-    await expect(client.invokePluginRpc(PLUGIN_ID, "nested-output", {})).rejects.toThrow(
-      "output.data is a Uint8Array. Bytes cross the wire only as the whole value",
-    );
-    await expect(
-      client.invokePluginRpc(PLUGIN_ID, "reverse", { data: new Uint8Array(1) }),
-    ).rejects.toThrow("input.data is a Uint8Array");
-
-    const nested = client.observePluginRpc(PLUGIN_ID, "nested-push", {});
-    await nested.ready;
-    await expect
-      .poll(async () => JSON.stringify(await client.getPluginLogs(PLUGIN_ID)))
-      .toContain("event.data is a Uint8Array");
-    await nested.release();
-  } finally {
-    await harness.close();
-  }
-}, 60_000);
-
-test("one name cannot serve both an RPC and a subscription", async () => {
-  const harness = await startHarness();
-  try {
-    const directory = await writePlugin(
-      "collision",
-      `
-import { defineRpc, defineSubscription } from "@getpaseo/plugin";
-import type { PluginServerContext } from "@getpaseo/plugin/server";
-import { z } from "zod";
-
-export default function contribute(server: PluginServerContext) {
-  server.handle(defineRpc({ name: "watch", input: z.object({}), output: z.null() }), () => null);
-  server.handle(
-    defineSubscription({ name: "watch", input: z.object({}), event: z.object({}) }),
-    () => undefined,
-  );
-  return () => undefined;
-}
-`,
-    );
-    await expect(harness.client.installDirectoryPlugin(directory)).rejects.toThrow(
-      "Duplicate plugin RPC method: watch",
-    );
-  } finally {
-    await harness.close();
-  }
-}, 60_000);
-
-test("plain RPC parses async input and output transforms once at the receiver", async () => {
-  const transformed = defineRpc({
-    name: "transformed",
-    input: z.object({ when: z.string().transform(async (text) => new Date(text)) }),
-    output: z.object({ n: z.string().transform(async (text) => Number(text)) }),
-  });
-
-  const harness = await startHarness();
-  try {
-    await expect(
-      callPluginRpc(
-        transformed,
-        (method, input) => harness.client.invokePluginRpc(PLUGIN_ID, method, input),
-        { when: "2026-01-01T00:00:00.000Z" },
-      ),
-    ).resolves.toEqual({ n: 2026 });
-  } finally {
-    await harness.close();
-  }
-});
-
-test("plain RPC applies a top-level input default in the plugin process", async () => {
-  const contract = defineRpc({
-    name: "defaulted",
-    input: z.string().default("x"),
-    output: z.string(),
-  });
-  const harness = await startHarness();
-  try {
-    await expect(
-      callPluginRpc(
-        contract,
-        (method, input) => harness.client.invokePluginRpc(PLUGIN_ID, method, input),
-        undefined,
-      ),
-    ).resolves.toBe("x");
-  } finally {
-    await harness.close();
-  }
-});
-
-test("plain RPC applies a top-level output default on the client", async () => {
-  const contract = defineRpc({
-    name: "defaulted-output",
-    input: z.object({}),
-    output: z.string().default("x"),
-  });
-  const harness = await startHarness();
-  try {
-    await expect(
-      callPluginRpc(
-        contract,
-        (method, input) => harness.client.invokePluginRpc(PLUGIN_ID, method, input),
-        {},
-      ),
-    ).resolves.toBe("x");
-  } finally {
-    await harness.close();
-  }
-}, 30_000);

packages/server/src/server/plugins/plugin-subscriptions.perf.e2e.test.ts

@@ -1,108 +0,0 @@
-import { mkdtemp, rm, writeFile } from "node:fs/promises";
-import { tmpdir } from "node:os";
-import path from "node:path";
-import { DaemonClient } from "@getpaseo/client/internal/daemon-client";
-import { afterEach, expect, test } from "vitest";
-import { resolveDaemonVersion } from "../daemon-version.js";
-import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
-
-const PUSH_BYTES = 256 * 1024;
-const TOTAL_BYTES = 100 * 1024 * 1024;
-const PUSHES = TOTAL_BYTES / PUSH_BYTES;
-const MIN_MEGABYTES_PER_SECOND = 200;
-
-// One buffer, pushed until the total is out: what a capture loop looks like.
-const SERVER_ENTRY = `
-import { defineSubscription } from "@getpaseo/plugin";
-import type { PluginServerContext } from "@getpaseo/plugin/server";
-import { z } from "zod";
-
-const stream = defineSubscription({
-  name: "stream",
-  input: z.object({ pushes: z.number().int(), bytes: z.number().int() }),
-  event: z.instanceof(Uint8Array),
-});
-
-export default function contribute(server: PluginServerContext) {
-  server.handle(stream, async ({ pushes, bytes }, { push, signal }) => {
-    const chunk = new Uint8Array(bytes).fill(7);
-    for (let sent = 0; sent < pushes && !signal.aborted; sent += 1) await push(chunk);
-    await new Promise((resolve) => signal.addEventListener("abort", resolve, { once: true }));
-  });
-  return () => undefined;
-}
-`;
-
-const roots: string[] = [];
-
-afterEach(async () => {
-  await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
-});
-
-test(`a plugin streams ${TOTAL_BYTES / 1024 / 1024} MiB in ${PUSH_BYTES / 1024} KiB pushes at ${MIN_MEGABYTES_PER_SECOND} MB/s without copying per frame`, async () => {
-  const pluginDirectory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-stream-"));
-  roots.push(pluginDirectory);
-  await writeFile(
-    path.join(pluginDirectory, "paseo-plugin.json"),
-    JSON.stringify({
-      id: "stream",
-      requirements: { paseo: `>=${resolveDaemonVersion(import.meta.url)}` },
-    }),
-  );
-  await writeFile(path.join(pluginDirectory, "index.server.ts"), SERVER_ENTRY);
-  const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
-  const client = new DaemonClient({
-    url: `ws://127.0.0.1:${daemon.port}/ws`,
-    clientId: `clid_plugin_stream_${Date.now()}`,
-    appVersion: resolveDaemonVersion(import.meta.url),
-  });
-  try {
-    await client.connect();
-    await client.patchDaemonConfig({ pluginsEnabled: true });
-    await client.installDirectoryPlugin(pluginDirectory);
-
-    let received = 0;
-    let frames = 0;
-    let copied = 0;
-    let finished!: () => void;
-    const done = new Promise<void>((resolve) => {
-      finished = resolve;
-    });
-    const baseline = process.memoryUsage().arrayBuffers;
-    let peak = baseline;
-    const started = performance.now();
-    const subscription = client.observePluginRpc("stream", "stream", {
-      pushes: PUSHES,
-      bytes: PUSH_BYTES,
-    });
-    subscription.subscribe({
-      snapshot: () => undefined,
-      update: () => undefined,
-      binary: (payload) => {
-        received += payload.byteLength;
-        frames += 1;
-        // The payload is a view into the frame the socket received, not a copy of it.
-        if (payload.buffer.byteLength - payload.byteLength > 1024) copied += 1;
-        if (frames % 40 === 0) peak = Math.max(peak, process.memoryUsage().arrayBuffers);
-        if (received >= TOTAL_BYTES) finished();
-      },
-    });
-    await done;
-    const seconds = (performance.now() - started) / 1000;
-    const megabytesPerSecond = TOTAL_BYTES / 1e6 / seconds;
-    process.stderr.write(
-      `[plugin stream] ${frames} frames, ${(TOTAL_BYTES / 1e6).toFixed(1)} MB in ${(seconds * 1000).toFixed(0)} ms: ${megabytesPerSecond.toFixed(0)} MB/s, peak external buffers +${((peak - baseline) / 1024 / 1024).toFixed(1)} MiB\n`,
-    );
-
-    expect(frames).toBe(PUSHES);
-    expect(received).toBe(TOTAL_BYTES);
-    expect(copied).toBe(0);
-    expect(megabytesPerSecond).toBeGreaterThanOrEqual(MIN_MEGABYTES_PER_SECOND);
-    // Memory held while streaming stays bounded by in-flight frames, not by the payload total.
-    expect(peak - baseline).toBeLessThan(TOTAL_BYTES / 2);
-    await subscription.release();
-  } finally {
-    await client.close().catch(() => undefined);
-    await daemon.close();
-  }
-}, 120_000);

packages/server/src/server/plugins/runtime-subscription.ts

@@ -1,56 +0,0 @@
-/** Receives a plugin subscription's events once its owner attached. */
-export interface PluginSubscriptionSink {
-  event(event: unknown): void;
-  /** The plugin ended the subscription, or stopped. Not called after `release()`. */
-  closed(error?: string): void;
-}
-
-/** A started plugin subscription, from the daemon's side of the subprocess. */
-export interface PluginRuntimeSubscription {
-  /** Resolves when the plugin acknowledged; rejects when it refused. */
-  readonly ready: Promise<void>;
-  /** Delivers what arrived so far, then everything that follows. */
-  attach(sink: PluginSubscriptionSink): void;
-  release(): void;
-}
-
-type Pending = { type: "event"; event: unknown } | { type: "closed"; error?: string };
-
-/**
- * Holds events until the owner attaches, so the owner can answer the subscribe request before
- * the first event goes out.
- */
-export class BufferedPluginSubscription {
-  private sink: PluginSubscriptionSink | null = null;
-  private pending: Pending[] = [];
-  private ended = false;
-
-  attach(sink: PluginSubscriptionSink): void {
-    if (this.sink) throw new Error("Plugin subscription is already attached");
-    this.sink = sink;
-    const pending = this.pending;
-    this.pending = [];
-    for (const entry of pending) {
-      if (entry.type === "event") sink.event(entry.event);
-      else sink.closed(entry.error);
-    }
-  }
-
-  event(event: unknown): void {
-    if (this.ended) return;
-    if (this.sink) this.sink.event(event);
-    else this.pending.push({ type: "event", event });
-  }
-
-  closed(error?: string): void {
-    if (this.ended) return;
-    this.ended = true;
-    if (this.sink) this.sink.closed(error);
-    else this.pending.push({ type: "closed", ...(error === undefined ? {} : { error }) });
-  }
-
-  released(): void {
-    this.ended = true;
-    this.pending = [];
-  }
-}

packages/server/src/server/plugins/runtime.posix.test.ts

@@ -1399,44 +1399,6 @@ export default function contribute(server: { registerProvider(provider: Provider
     }
   });
 
-  it("releases a subscription in the plugin when its subscribe times out", async () => {
-    vi.useFakeTimers();
-    try {
-      const directory = await createPlugin(
-        "slow-subscribe",
-        `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`,
-      );
-      const child = createReloadChild("slow-subscribe", []);
-      const sent: Array<{ type: string; subscriptionId?: string }> = [];
-      const originalSend = child.send.bind(child);
-      child.send = (message, callback) => {
-        sent.push(message);
-        if (message.type !== "initialize") return originalSend(message, callback);
-        callback?.(null);
-        queueMicrotask(() =>
-          child.emitMessage({ type: "ready", methods: [], subscriptions: ["feed"], providers: [] }),
-        );
-        return true;
-      };
-      const runtime = createTestRuntime({ spawnChild: () => child });
-      await runtime.startPlugin("slow-subscribe", directory);
-
-      const subscription = runtime.startSubscription("slow-subscribe", "feed", {});
-      const rejection = subscription.ready.catch((error: Error) => error);
-      await vi.advanceTimersByTimeAsync(30_000);
-      await expect(rejection).resolves.toMatchObject({
-        message: expect.stringContaining("timed out"),
-      });
-      const subscribe = sent.find((message) => message.type === "subscribe");
-      expect(sent).toContainEqual({
-        type: "unsubscribe",
-        subscriptionId: subscribe?.subscriptionId,
-      });
-    } finally {
-      vi.useRealTimers();
-    }
-  });
-
   it("escalates a plugin that ignores graceful shutdown from TERM to KILL", async () => {
     vi.useFakeTimers();
     try {
@@ -1853,7 +1815,7 @@ export function inspect() { void Surface; return {}; }`,
     await runtime.stopAll();
   });
 
-  it("forwards a handler result for the client to validate against its RPC output schema", async () => {
+  it("rejects a handler result that does not match its RPC output schema", async () => {
     const directory = await createPlugin(
       "invalid-output",
       `import { z } from "zod";
@@ -1872,9 +1834,7 @@ export default function contribute(plugin: any) {
 
     await runtime.startPlugin("invalid-output", directory);
 
-    await expect(runtime.invoke("invalid-output", "broken", {})).resolves.toEqual({
-      value: "wrong",
-    });
+    await expect(runtime.invoke("invalid-output", "broken", {})).rejects.toThrow();
     await runtime.stopAll();
   });