fix(plugins): parse a subscription's input once in the plugin process

server.handle wrapped subscription handlers in a second input parse, so a
transforming input schema rejected its own parsed output before the
handler ran.

docs/android.md

@@ -174,7 +174,7 @@ REACT_NATIVE_PACKAGER_HOSTNAME=localhost \
 
 This is the Android counterpart of the iOS local-simulator flow in [development.md](development.md): on iOS the simulator shares the Mac's loopback so `localhost:<port>` works directly; on Android you need `10.0.2.2` or `adb reverse`.
 
-## Inverted timeline scrolling
+## Inverted timeline selection
 
 Android focus and selection visibility requests must not reposition inverted timelines. The
 `modules/paseo-scroll` package keeps React Native's scroll manager interface and returns zero for
@@ -183,16 +183,6 @@ Native's Android list uses `scale: -1`, which decomposes into a 180-degree rotat
 `scaleY`; checking the scale property alone misses it. Dragging and explicit scroll
 commands still work; non-inverted scroll views keep Android's default behavior.
 
-Nested panels also keep their edge overshoot and unconsumed flings inside the panel instead of
-moving the inverted timeline in the opposite direction. Nested-scroll registration and scrolling
-inside the panel are preserved. Run the native selection and nested-scroll regressions after prebuild:
-
-```bash
-cd packages/app/android
-./gradlew :paseo-scroll:connectedDebugAndroidTest \
-  --no-daemon --max-workers=2 -Dorg.gradle.parallel=false
-```
-
 Register this package before React Native's core package through its Expo config plugin. Normal
 Android builds use the prebuilt `react-android` library, so patching Java under `node_modules` does
 not change the shipped scroll view. Keep this behavior in the app's compiled native module.

docs/plugins.md

@@ -357,9 +357,27 @@ 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 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.
+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).
 
 `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/composer-recall.spec.ts

@@ -3,7 +3,7 @@ import {
   composerLocator,
   expectComposerDraft,
   expectComposerVisible,
-  submitMessage,
+  fillComposerDraft,
   typeIntoFocusedComposer,
 } from "../support/helpers/composer";
 import { openAgentRoute, seedMockAgentWorkspace } from "../support/helpers/mock-agent";
@@ -12,8 +12,14 @@ function userMessages(page: Page) {
   return page.getByTestId("user-message");
 }
 
+async function sendDraftWhenReady(page: Page) {
+  await expect(page.getByRole("button", { name: "Send message", exact: true })).toBeEnabled();
+  await pressInComposer(page, "Enter");
+}
+
 async function sendAndWaitForMessage(page: Page, text: string, expectedCount: number) {
-  await submitMessage(page, text);
+  await fillComposerDraft(page, text);
+  await sendDraftWhenReady(page);
   await expect(userMessages(page)).toHaveCount(expectedCount, { timeout: 30_000 });
   await expectComposerDraft(page, "");
 }
@@ -49,7 +55,7 @@ test.describe("Composer recall", () => {
       await pressInComposer(page, "ArrowUp");
       await expectComposerDraft(page, "add tests for login");
 
-      await pressInComposer(page, "Enter");
+      await sendDraftWhenReady(page);
       await expect(userMessages(page)).toHaveCount(3, { timeout: 30_000 });
       await expect(userMessages(page).last()).toContainText("add tests for login");
     } finally {

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

@@ -0,0 +1,79 @@
+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/modules/paseo-scroll/android/src/androidTest/AndroidManifest.xml

@@ -1,5 +0,0 @@
-<manifest xmlns:android="http://schemas.android.com/apk/res/android">
-  <application>
-    <activity android:name="sh.paseo.scroll.NestedScrollActivity" android:theme="@android:style/Theme.Material.Light.NoActionBar" android:exported="false" />
-  </application>
-</manifest>

packages/app/modules/paseo-scroll/android/src/androidTest/java/sh/paseo/scroll/NestedScrollTest.kt

@@ -1,134 +0,0 @@
-package sh.paseo.scroll
-
-import android.app.Activity
-import android.content.Intent
-import android.os.SystemClock
-import android.view.View
-import android.view.ViewGroup
-import android.widget.FrameLayout
-import androidx.test.ext.junit.runners.AndroidJUnit4
-import androidx.test.platform.app.InstrumentationRegistry
-import com.facebook.react.bridge.JavaOnlyArray
-import com.facebook.react.bridge.JavaOnlyMap
-import com.facebook.react.soloader.OpenSourceMergedSoMapping
-import com.facebook.react.uimanager.DisplayMetricsHolder
-import com.facebook.react.uimanager.ReactStylesDiffMap
-import com.facebook.react.uimanager.ThemedReactContext
-import com.facebook.react.views.scroll.ReactScrollView
-import com.facebook.react.views.scroll.ReactScrollViewManager
-import com.facebook.soloader.SoLoader
-import org.junit.Assert.*
-import org.junit.Test
-import org.junit.runner.RunWith
-
-class NestedScrollActivity : Activity()
-
-@RunWith(AndroidJUnit4::class)
-class NestedScrollTest {
-  @Test fun invertedTimelineDoesNotConsumePanelOvershoot() = withNestedScroll(inverted = true) { parent, child ->
-    val before = parent.scrollY
-    assertTrue(child.startNestedScroll(View.SCROLL_AXIS_VERTICAL))
-    child.scrollTo(0, 600)
-    assertEquals(600, child.scrollY)
-    assertTrue(child.dispatchNestedScroll(0, 0, 0, 100, null))
-    assertEquals("Panel overshoot must not move the inverted timeline", before, parent.scrollY)
-    child.scrollTo(0, 0)
-    assertTrue(child.dispatchNestedScroll(0, 0, 0, -100, null))
-    assertEquals("Overshoot at either edge must stay inside the panel", before, parent.scrollY)
-    child.stopNestedScroll()
-  }
-
-  @Test fun ordinaryParentStillConsumesPanelOvershoot() = withNestedScroll(inverted = false) { parent, child ->
-    val before = parent.scrollY
-    assertTrue(child.startNestedScroll(View.SCROLL_AXIS_VERTICAL))
-    child.scrollTo(0, 600)
-    assertEquals(600, child.scrollY)
-    assertTrue(child.dispatchNestedScroll(0, 0, 0, 100, null))
-    assertEquals(before + 100, parent.scrollY)
-    child.stopNestedScroll()
-  }
-
-  @Test fun panelCanStillScrollInsideInvertedTimeline() = withNestedScroll(inverted = true) { parent, child ->
-    val before = parent.scrollY
-    assertTrue(child.startNestedScroll(View.SCROLL_AXIS_VERTICAL))
-    child.scrollTo(0, 100)
-    assertTrue(child.dispatchNestedScroll(0, 100, 0, 0, null))
-    assertEquals(100, child.scrollY)
-    assertEquals(before, parent.scrollY)
-    child.stopNestedScroll()
-    parent.scrollTo(0, before + 100)
-    assertEquals(before + 100, parent.scrollY)
-  }
-
-  @Test fun invertedTimelineAbsorbsUnconsumedPanelFling() = checkFling(inverted = true)
-
-  @Test fun ordinaryParentStillAcceptsUnconsumedPanelFling() = checkFling(inverted = false)
-
-  private fun checkFling(inverted: Boolean) = withNestedScroll(
-    inverted = inverted,
-    afterFrames = { parent ->
-      if (inverted) assertEquals("Panel fling must not move the inverted timeline", 400, parent.scrollY)
-      else assertTrue("Control fling must move an ordinary parent", parent.scrollY > 400)
-    }
-  ) { parent, child ->
-    parent.scrollTo(0, 400)
-    assertTrue(child.startNestedScroll(View.SCROLL_AXIS_VERTICAL))
-    child.scrollTo(0, 600)
-    assertEquals(600, child.scrollY)
-    assertTrue(child.dispatchNestedFling(0f, 1500f, false))
-    child.stopNestedScroll()
-  }
-
-  private fun withNestedScroll(
-    inverted: Boolean,
-    afterFrames: ((ReactScrollView) -> Unit)? = null,
-    check: (ReactScrollView, ReactScrollView) -> Unit
-  ) {
-    val instrumentation = InstrumentationRegistry.getInstrumentation()
-    val activity = instrumentation.startActivitySync(
-      Intent(instrumentation.context, NestedScrollActivity::class.java).addFlags(Intent.FLAG_ACTIVITY_NEW_TASK)
-    )
-    lateinit var parent: ReactScrollView
-    try {
-      instrumentation.runOnMainSync {
-        val appContext = instrumentation.targetContext
-        DisplayMetricsHolder.initDisplayMetricsIfNotInitialized(appContext)
-        SoLoader.init(appContext, OpenSourceMergedSoMapping)
-        val reactContext = ScrollTestContext(appContext)
-        val context = ThemedReactContext(reactContext, appContext)
-        val manager = PaseoScrollPackage().createViewManager(reactContext, ReactScrollViewManager.REACT_CLASS) as ReactScrollViewManager
-        parent = manager.createViewInstance(context)
-        val child = manager.createViewInstance(context).apply { isNestedScrollingEnabled = true }
-        val panelContent = View(context)
-        child.addView(panelContent, FrameLayout.LayoutParams(300, 800))
-        val content = FrameLayout(context)
-        content.addView(child, FrameLayout.LayoutParams(300, 200).apply { topMargin = 500 })
-        parent.addView(content, FrameLayout.LayoutParams(300, 2000))
-        panelContent.measure(exact(300), exact(800))
-        panelContent.layout(0, 0, 300, 800)
-        child.measure(exact(300), exact(200))
-        child.layout(0, 500, 300, 700)
-        content.measure(exact(300), exact(2000))
-        content.layout(0, 0, 300, 2000)
-        parent.measure(exact(300), exact(400))
-        parent.layout(0, 0, 300, 400)
-        // The same transform React Native applies to an inverted FlatList on Android.
-        manager.updateProperties(parent, ReactStylesDiffMap(JavaOnlyMap.of(
-          "transform", JavaOnlyArray.of(JavaOnlyMap.of("scale", if (inverted) -1.0 else 1.0))
-        )))
-        activity.setContentView(parent, ViewGroup.LayoutParams(300, 400))
-        parent.scrollTo(0, 400)
-        check(parent, child)
-      }
-      if (afterFrames != null) {
-        // Let the framework's real fling animation advance, with an ordinary-parent control.
-        SystemClock.sleep(500)
-        instrumentation.runOnMainSync { afterFrames(parent) }
-      }
-    } finally {
-      instrumentation.runOnMainSync { activity.finish() }
-    }
-  }
-
-  private fun exact(size: Int) = View.MeasureSpec.makeMeasureSpec(size, View.MeasureSpec.EXACTLY)
-}

packages/app/modules/paseo-scroll/android/src/androidTest/java/sh/paseo/scroll/ScrollSelectionTest.kt

@@ -8,6 +8,7 @@ import android.widget.FrameLayout
 import android.widget.TextView
 import androidx.test.ext.junit.runners.AndroidJUnit4
 import androidx.test.platform.app.InstrumentationRegistry
+import com.facebook.react.bridge.BridgeReactContext
 import com.facebook.react.bridge.JavaOnlyArray
 import com.facebook.react.bridge.JavaOnlyMap
 import com.facebook.react.soloader.OpenSourceMergedSoMapping
@@ -69,7 +70,7 @@ class ScrollSelectionTest {
       val appContext = instrumentation.targetContext
       DisplayMetricsHolder.initDisplayMetricsIfNotInitialized(appContext)
       SoLoader.init(appContext, OpenSourceMergedSoMapping)
-      val reactContext = ScrollTestContext(appContext)
+      val reactContext = BridgeReactContext(appContext)
       val context = ThemedReactContext(reactContext, appContext)
       val manager = PaseoScrollPackage().createViewManager(reactContext, ReactScrollViewManager.REACT_CLASS) as ReactScrollViewManager
       val scroll = manager.createViewInstance(context)

packages/app/modules/paseo-scroll/android/src/androidTest/java/sh/paseo/scroll/ScrollTestContext.java

@@ -1,19 +0,0 @@
-package sh.paseo.scroll;
-
-import android.content.Context;
-import com.facebook.react.bridge.BridgeReactContext;
-import com.facebook.react.uimanager.events.BlackHoleEventDispatcher;
-import com.facebook.react.uimanager.events.EventDispatcher;
-import com.facebook.react.uimanager.events.EventDispatcherProvider;
-
-/** Native-only tests use RN's dispatcher for a context without a JS instance. */
-final class ScrollTestContext extends BridgeReactContext implements EventDispatcherProvider {
-  ScrollTestContext(Context context) {
-    super(context);
-  }
-
-  @Override
-  public EventDispatcher getEventDispatcher() {
-    return BlackHoleEventDispatcher.INSTANCE;
-  }
-}

packages/app/modules/paseo-scroll/android/src/main/java/sh/paseo/scroll/PaseoScrollPackage.kt

@@ -2,7 +2,6 @@ package sh.paseo.scroll
 
 import android.graphics.Matrix
 import android.graphics.Rect
-import android.view.View
 import com.facebook.react.ReactPackage
 import com.facebook.react.ViewManagerOnDemandReactPackage
 import com.facebook.react.bridge.NativeModule
@@ -31,30 +30,13 @@ private class PaseoScrollViewManager : ReactScrollViewManager() {
 private class PaseoScrollView(context: ThemedReactContext) : ReactScrollView(context) {
   private val transformValues = FloatArray(9)
 
-  private val isInverted: Boolean
-    get() {
-      // RN's Android scale: -1 becomes a 180-degree rotation with positive scaleY.
-      matrix.getValues(transformValues)
-      return transformValues[Matrix.MSCALE_Y] < 0
-    }
-
   // Both RN focus handling and Android selection visibility requests use this calculation.
   // Its coordinates ignore inversion. Keep focus/selection bookkeeping, but leave timeline
   // movement to user gestures and explicit scroll commands.
   override fun computeScrollDeltaToGetChildRectOnScreen(rect: Rect): Int {
-    if (isInverted) return 0
+    // RN's Android scale: -1 becomes a 180-degree rotation with positive scaleY.
+    matrix.getValues(transformValues)
+    if (transformValues[Matrix.MSCALE_Y] < 0) return 0
     return super.computeScrollDeltaToGetChildRectOnScreen(rect)
   }
-
-  // A nested panel's remaining delta uses its own visual direction. Consuming it in
-  // an inverted timeline moves the timeline the opposite way when the panel hits an edge.
-  override fun onNestedScroll(target: View, dxConsumed: Int, dyConsumed: Int, dxUnconsumed: Int, dyUnconsumed: Int) {
-    if (isInverted) return
-    super.onNestedScroll(target, dxConsumed, dyConsumed, dxUnconsumed, dyUnconsumed)
-  }
-
-  override fun onNestedFling(target: View, velocityX: Float, velocityY: Float, consumed: Boolean): Boolean {
-    if (isInverted && !consumed) return true
-    return super.onNestedFling(target, velocityX, velocityY, consumed)
-  }
 }

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

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

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

@@ -17,6 +17,9 @@ 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,6 +153,9 @@ 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,6 +2,7 @@ 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,
@@ -36,6 +37,8 @@ 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,6 +71,9 @@ 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,6 +111,7 @@ export type PluginClientRuntime = Pick<
   | "openScreen"
   | "openSurface"
   | "openPanel"
+  | "subscribe"
   | "addComposerPill"
   | "addHeaderButton"
 > & { hosts: ReturnType<typeof createPluginHosts> };

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

@@ -29,6 +29,9 @@ 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}>
+        <PluginRpcProvider invoke={plugin.invoke} subscribe={plugin.subscribe}>
           <PluginTranslationProvider runtime={plugin.i18n}>{children}</PluginTranslationProvider>
         </PluginRpcProvider>
       </PaseoApiProvider>

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

@@ -62,6 +62,9 @@ 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,6 +39,9 @@ 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,6 +8,7 @@ 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";
 
@@ -89,12 +90,18 @@ export class PluginRegistry {
           return [existing];
         }
         const client = options.client;
+        const lifetime = new AbortController();
         installation = {
-          lifetime: new AbortController(),
+          lifetime,
           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,6 +26,9 @@ 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

@@ -0,0 +1,189 @@
+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

@@ -0,0 +1,52 @@
+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,6 +26,9 @@ 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,6 +56,9 @@ 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,6 +18,9 @@ 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,6 +71,8 @@ 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,6 +15,9 @@ 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/client/src/connection/index.ts

@@ -144,6 +144,7 @@ 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,6 +83,26 @@ 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,6 +5,8 @@ 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;
 }
 
@@ -20,9 +22,12 @@ 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;
@@ -50,7 +55,7 @@ export class OwnedSubscriptions {
     let teardown: Promise<void> | null = null;
     let disposed = false;
     let overflow = false;
-    const buffered: SessionOutboundMessage[] = [];
+    const buffered: Update[] = [];
     const observers = new Set<SubscriptionObserver<SubscriptionSnapshot<T>>>();
     let resolve!: (snapshot: SubscriptionSnapshot<T>) => void;
     let reject!: (error: unknown) => void;
@@ -60,12 +65,10 @@ export class OwnedSubscriptions {
     });
     void ready.catch(() => {});
 
-    const deliver = (
-      observer: SubscriptionObserver<SubscriptionSnapshot<T>>,
-      message?: SessionOutboundMessage,
-    ) => {
+    const deliver = (observer: SubscriptionObserver<SubscriptionSnapshot<T>>, update?: Update) => {
       try {
-        if (message) observer.update(message);
+        if (update && "binary" in update) observer.binary?.(update.binary);
+        else if (update) observer.update(update.message);
         else if (snapshot) observer.snapshot(snapshot);
       } catch (error) {
         this.host.failed(error);
@@ -78,6 +81,13 @@ 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;
@@ -128,12 +138,14 @@ export class OwnedSubscriptions {
       },
       receive: (message) => {
         if (disposed || ("payload" in message && Object.is(message.payload, snapshot))) return;
-        if (buffered.length < MAX_BOOTSTRAP_UPDATES && !overflow) buffered.push(message);
-        else {
-          buffered.length = 0;
-          overflow = true;
-        }
-        for (const observer of observers) deliver(observer, message);
+        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 });
       },
       release: () => {
         if (teardown) return teardown;
@@ -142,7 +154,9 @@ export class OwnedSubscriptions {
         observers.clear();
         buffered.length = 0;
         if (id) this.routes.delete(id);
-        reject(new Error("Subscription released"));
+        // 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"));
         teardown = (async () => {
           await pending;
           await releaseId();
@@ -174,7 +188,7 @@ export class OwnedSubscriptions {
             .catch(this.host.failed);
         } else {
           deliver(observer);
-          for (const message of buffered) deliver(observer, message);
+          for (const update of buffered) deliver(observer, update);
         }
         return () => {
           observers.delete(observer);
@@ -205,6 +219,10 @@ 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,6 +1106,7 @@ 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,6 +164,11 @@ import {
   FileTransferOpcode,
   TerminalStreamOpcode,
   type FileTransferFrame,
+  assertPluginWireValue,
+  decodePluginFrame,
+  encodePluginInputFrame,
+  isPluginFrameOpcode,
+  PluginFrameOpcode,
 } from "@getpaseo/protocol/binary-frames/index";
 import {
   createRelayE2eeTransportFactory,
@@ -2260,6 +2265,8 @@ 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;
@@ -2290,7 +2297,9 @@ export class DaemonClient {
     // The response deadline still applies while the transport waits for drain.
     void promise.catch(() => undefined);
     try {
-      const sent = this.sendSessionMessageOrThrow(wire.message);
+      const sent = params.transmit
+        ? params.transmit()
+        : this.sendSessionMessageOrThrow(wire.message);
       const result = await Promise.race([promise, sent.then(() => promise)]);
       if (result.kind === "error") throw result.error;
       return result.value;
@@ -2313,12 +2322,14 @@ 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) {
@@ -2650,6 +2661,7 @@ 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 {
@@ -6197,22 +6209,59 @@ 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();
-    const payload = await this.sendCorrelatedSessionRequest({
-      requestId,
-      message: {
-        type: "plugin.rpc.invoke.request",
+    if (!(input instanceof Uint8Array)) {
+      const payload = await this.sendCorrelatedSessionRequest({
         requestId,
-        pluginId,
-        method,
-        input,
-      },
+        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({
+      requestId,
+      message: { type: "plugin.rpc.invoke.request", requestId, pluginId, method, input: null },
       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,
@@ -6975,6 +7024,23 @@ 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) {
@@ -6986,6 +7052,10 @@ 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,3 +323,68 @@ 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,7 +4,13 @@ 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 } from "../rpc.js";
+import type {
+  PluginRpcContract,
+  PluginSubscribeOptions,
+  PluginSubscriptionContract,
+  PluginSubscriptionHandle,
+  PluginWireInput,
+} from "../rpc.js";
 import type {
   PluginButtonRegistration,
   PluginHeaderButtonContribution,
@@ -203,6 +209,16 @@ 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;
@@ -295,7 +311,7 @@ export interface PluginCommandCapabilities {
   paseo: PaseoApi;
   rpc<InputSchema extends ZodType, OutputSchema extends ZodType>(
     contract: PluginRpcContract<InputSchema, OutputSchema>,
-    input: ZodInput<InputSchema>,
+    input: PluginWireInput<ZodInput<InputSchema>>,
   ): Promise<ZodOutput<OutputSchema>>;
   openScreen(input: PluginOpenScreenInput): void;
   /** @deprecated Use `openScreen`. */

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

packages/plugin/src/client/index.ts

@@ -71,7 +71,12 @@ export {
 } from "./file.js";
 export { usePaseo } from "./paseo-context.js";
 export { useAgent, useWorkspace } from "./client-state.js";
-export { useRpc } from "./rpc-context.js";
+export { useRpc, useSubscription, type PluginSubscriptionState } from "./rpc-context.js";
+export type {
+  PluginSubscribeOptions,
+  PluginSubscriptionClosed,
+  PluginSubscriptionHandle,
+} from "../rpc.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,9 +1,34 @@
-import { createContext, useCallback, useContext, useMemo, type ReactNode } from "react";
+import {
+  createContext,
+  useCallback,
+  useContext,
+  useEffect,
+  useMemo,
+  useRef,
+  useState,
+  type ReactNode,
+} from "react";
 import type { ZodType, input as ZodInput, output as ZodOutput } from "zod";
-import { callPluginRpc, type PluginRpcContract } from "../rpc.js";
+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;
 
 export interface PluginRpcContextValue {
   invoke(method: string, input: unknown): Promise<unknown>;
+  subscribe: PluginSubscribe;
 }
 
 const PluginRpcContext = createContext<PluginRpcContextValue | null>(null);
@@ -15,22 +40,100 @@ 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 }), [invoke]);
+  const value = useMemo(() => ({ invoke, subscribe }), [invoke, subscribe]);
   return <PluginRpcContext.Provider value={value}>{children}</PluginRpcContext.Provider>;
 }
 
 export function useRpc<InputSchema extends ZodType, OutputSchema extends ZodType>(
   contract: PluginRpcContract<InputSchema, OutputSchema>,
-): (input: ZodInput<InputSchema>) => Promise<ZodOutput<OutputSchema>> {
+): (input: PluginWireInput<ZodInput<InputSchema>>) => Promise<ZodOutput<OutputSchema>> {
   const context = usePluginRpcContextValue();
   if (!context) throw new Error("useRpc must run inside a contributed plugin surface");
 
   return useCallback(
-    (input: ZodInput<InputSchema>) => callPluginRpc(contract, context.invoke, input),
+    (input: PluginWireInput<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}>
+          <PluginRpcProvider invoke={rpc.invoke} subscribe={rpc.subscribe}>
             <PluginTranslationProvider runtime={translations}>{content}</PluginTranslationProvider>
           </PluginRpcProvider>
         </PaseoApiProvider>

packages/plugin/src/index.ts

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

packages/plugin/src/rpc.ts

@@ -9,6 +9,12 @@ 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"]>;
 
@@ -31,9 +37,99 @@ 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: ZodInput<InputSchema>,
+  input: PluginWireInput<ZodInput<InputSchema>>,
 ): Promise<ZodOutput<OutputSchema>> {
   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 } from "../rpc.js";
+import type { PluginRpcContract, PluginSubscriptionContract, PluginWireInput } from "../rpc.js";
 import type { PluginCleanup } from "../contracts.js";
 import type { ProviderRegistration } from "./provider.js";
 import type { UsageSourceRegistration } from "./usage.js";
@@ -10,6 +10,18 @@ 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";
@@ -31,12 +43,24 @@ 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,
-    ) => ZodInput<OutputSchema> | Promise<ZodInput<OutputSchema>>,
+    ) => PluginWireInput<ZodInput<OutputSchema>> | Promise<PluginWireInput<ZodInput<OutputSchema>>>,
   ): void;
   registerProvider(provider: ProviderRegistration): void;
   registerUsageSource(source: UsageSourceRegistration): void;

packages/plugin/src/server/index.ts

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

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

@@ -4,6 +4,7 @@ import {
   FileTransferOpcode,
   type FileTransferFrame,
 } from "./file-transfer.js";
+import { decodePluginFrame, PluginFrameOpcode, type PluginFrame } from "./plugin.js";
 import {
   decodeTerminalStreamFrame,
   TerminalStreamOpcode,
@@ -13,7 +14,8 @@ import {
 export type BinaryFrame =
   | { kind: "host_route"; frame: HostRouteFrame }
   | { kind: "terminal"; frame: TerminalStreamFrame }
-  | { kind: "file_transfer"; frame: FileTransferFrame };
+  | { kind: "file_transfer"; frame: FileTransferFrame }
+  | { kind: "plugin"; frame: PluginFrame };
 
 export function decodeBinaryFrame(bytes: Uint8Array): BinaryFrame | null {
   switch (bytes[0]) {
@@ -38,6 +40,12 @@ 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,3 +2,4 @@ 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

@@ -0,0 +1,95 @@
+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

@@ -0,0 +1,152 @@
+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,6 +46,9 @@ 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,3 +372,43 @@ 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

@@ -1606,6 +1606,14 @@ export const PluginRpcInvokeRequestSchema = z.object({
   input: z.unknown(),
 });
 
+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(),
+});
+
 export const AgentTimelineAppendRequestSchema = z.object({
   type: z.literal("agent.timeline.append.request"),
   requestId: z.string(),
@@ -3339,6 +3347,7 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
   PluginDisableRequestSchema,
   PluginRemoveRequestSchema,
   PluginRpcInvokeRequestSchema,
+  PluginRpcSubscribeRequestSchema,
   AgentTimelineAppendRequestSchema,
   AgentSkillsGetStatusRequestSchema,
   AgentSkillsReconcileRequestSchema,
@@ -3739,6 +3748,8 @@ 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.
@@ -6962,6 +6973,22 @@ export const PluginRpcInvokeResponseSchema = z.object({
   }),
 });
 
+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({
@@ -7035,6 +7062,9 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
   PluginDisableResponseSchema,
   PluginRemoveResponseSchema,
   PluginRpcInvokeResponseSchema,
+  PluginRpcSubscribeResponseSchema,
+  PluginRpcEventSchema,
+  PluginRpcClosedSchema,
   AgentTimelineAppendResponseSchema,
   AgentSkillsGetStatusResponseSchema,
   AgentSkillsReconcileResponseSchema,

packages/server/package.json

@@ -89,7 +89,7 @@
     "speech:transcribe:local": "tsx scripts/transcribe-local-wav.ts",
     "test": "npm run test:unit && npm run test:integration",
     "test:unit": "vitest run --fileParallelism --exclude \"**/*.e2e.test.ts\"",
-    "test:integration": "vitest run --maxWorkers=1 src/server/git-filename-diffs.e2e.test.ts src/server/daemon-e2e/checkout-diff-content.e2e.test.ts src/server/daemon-e2e/filesystem.e2e.test.ts src/server/daemon-e2e/file-download.e2e.test.ts src/server/daemon-e2e/cli-logs-follow.e2e.test.ts src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/creation-status.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/terminal-worker-lifecycle.e2e.test.ts src/server/workspace-create-worktree-source.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/daemon-e2e/oversized-message.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts src/server/plugins/usage-sources.e2e.test.ts src/server/plugins/lifecycle.e2e.test.ts src/server/connected-hosts/hosts.e2e.test.ts src/server/owned-subscriptions.e2e.test.ts && vitest run --maxWorkers=1 src/server/agent/providers/omp/history.e2e.test.ts src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t \"merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments|connected hosts|password-protected daemon refuses|replays selected OMP history\"",
+    "test:integration": "vitest run --maxWorkers=1 src/server/git-filename-diffs.e2e.test.ts src/server/daemon-e2e/checkout-diff-content.e2e.test.ts src/server/daemon-e2e/create-agent-mode.e2e.test.ts src/server/daemon-e2e/filesystem.e2e.test.ts src/server/daemon-e2e/file-download.e2e.test.ts src/server/daemon-e2e/cli-logs-follow.e2e.test.ts src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/creation-status.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/terminal-worker-lifecycle.e2e.test.ts src/server/workspace-create-worktree-source.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/daemon-e2e/oversized-message.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts src/server/plugins/usage-sources.e2e.test.ts src/server/plugins/lifecycle.e2e.test.ts src/server/connected-hosts/hosts.e2e.test.ts src/server/owned-subscriptions.e2e.test.ts && vitest run --maxWorkers=1 src/server/agent/providers/omp/history.e2e.test.ts src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t \"merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments|connected hosts|password-protected daemon refuses|replays selected OMP history\"",
     "test:hub-cli-contract": "vitest run --maxWorkers=1 src/server/hub/hub-cli-contract.test.ts",
     "test:integration:all": "npm run test:e2e",
     "test:integration:real": "vitest run real.e2e.test.ts",

packages/server/src/server/agent/create-agent-mode.test.ts

@@ -78,7 +78,7 @@ describe("resolveAndValidateCreateAgentMode", () => {
         availableModes: OPENCODE_MODES,
       }),
     ).toThrow(
-      "cannot inherit mode 'bypassPermissions' from caller (provider 'claude') for new agent (provider 'opencode'). Pass an explicit mode. Available modes for 'opencode': build, plan",
+      "cannot inherit mode 'bypassPermissions' from caller (provider 'claude') for new agent (provider 'opencode'). Pass an explicit mode in settings.modeId. Available modes for 'opencode': build, plan",
     );
   });
 
@@ -92,7 +92,7 @@ describe("resolveAndValidateCreateAgentMode", () => {
         availableModes: CODEX_MODES,
       }),
     ).toThrow(
-      "cannot inherit mode '<none>' from caller (provider 'opencode') for new agent (provider 'codex'). Pass an explicit mode. Available modes for 'codex': auto, full-access",
+      "cannot inherit mode '<none>' from caller (provider 'opencode') for new agent (provider 'codex'). Pass an explicit mode in settings.modeId. Available modes for 'codex': auto, full-access",
     );
   });
 
@@ -180,7 +180,7 @@ describe("resolveAndValidateCreateAgentMode", () => {
         targetUnattendedMode: "full-access",
       }),
     ).toThrow(
-      "cannot inherit mode 'default' from caller (provider 'claude') for new agent (provider 'codex'). Pass an explicit mode. Available modes for 'codex': auto, full-access",
+      "cannot inherit mode 'default' from caller (provider 'claude') for new agent (provider 'codex'). Pass an explicit mode in settings.modeId. Available modes for 'codex': auto, full-access",
     );
   });
 
@@ -195,7 +195,7 @@ describe("resolveAndValidateCreateAgentMode", () => {
         targetUnattendedMode: undefined,
       }),
     ).toThrow(
-      "cannot inherit mode 'bypassPermissions' from caller (provider 'claude') for new agent (provider 'zai-custom'). Pass an explicit mode. Available modes for 'zai-custom': unknown",
+      "cannot inherit mode 'bypassPermissions' from caller (provider 'claude') for new agent (provider 'zai-custom'). Pass an explicit mode in settings.modeId. Available modes for 'zai-custom': unknown",
     );
   });
 

packages/server/src/server/agent/create-agent-mode.ts

@@ -77,7 +77,7 @@ export function resolveAndValidateCreateAgentMode(
   }
 
   throw new Error(
-    `cannot inherit mode '${formatCreateConfigParentMode(parent)}' from ${formatCreateConfigParentSource(parent)} for new agent (provider '${targetProvider}'). Pass an explicit mode. Available modes for '${targetProvider}': ${listModes(availableModes)}`,
+    `cannot inherit mode '${formatCreateConfigParentMode(parent)}' from ${formatCreateConfigParentSource(parent)} for new agent (provider '${targetProvider}'). Pass an explicit mode in settings.modeId. Available modes for '${targetProvider}': ${listModes(availableModes)}`,
   );
 }
 

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

@@ -0,0 +1,190 @@
+import { expect, test } from "vitest";
+import { createTestLogger } from "../../../../test-utils/test-logger.js";
+import { collectSessionTurnEvents } from "../test-utils/session-stream-adapter.js";
+import { ClaudeAgentClient } from "./agent.js";
+import type { ClaudeQueryFactory } from "./query.js";
+
+const LIMIT = "You've hit your session limit · resets 5:20pm (Asia/Singapore)";
+
+interface ScriptedTurn {
+  // An API error reply as Claude Code writes it: a synthetic assistant message with `error`.
+  errorReply?: string;
+  // Assistant text the model wrote before or after it.
+  before?: string;
+  after?: string;
+  isError: boolean;
+  usage?: {
+    inputTokens: number;
+    cachedInputTokens: number;
+    outputTokens: number;
+    totalCostUsd: number;
+  };
+}
+
+// A query dependency that speaks Claude Code's stream interface, one scripted turn per prompt.
+function scriptedQuery(turns: ScriptedTurn[]): ClaudeQueryFactory {
+  return ({ prompt }) => {
+    let index = 0;
+    const assistant = (id: string, text: string, error?: string) => ({
+      type: "assistant",
+      session_id: "error-session",
+      parent_tool_use_id: null,
+      uuid: id,
+      message: {
+        id,
+        role: "assistant",
+        model: error ? "<synthetic>" : "claude-opus-5-5",
+        content: [{ type: "text", text }],
+        stop_reason: "stop_sequence",
+        usage: { input_tokens: 0, output_tokens: 0 },
+      },
+      ...(error ? { error } : {}),
+    });
+    const messages = (async function* () {
+      if (typeof prompt === "string") throw new Error("Expected streaming input");
+      for await (const _message of prompt) {
+        const turn = turns[index];
+        index += 1;
+        yield {
+          type: "system",
+          subtype: "init",
+          session_id: "error-session",
+          permissionMode: "default",
+          model: "claude-opus-5-5",
+        };
+        if (turn.before) yield assistant(`before-${index}`, turn.before);
+        if (turn.errorReply) yield assistant(`error-${index}`, turn.errorReply, "rate_limit");
+        if (turn.after) yield assistant(`after-${index}`, turn.after);
+        yield {
+          type: "result",
+          subtype: "success",
+          session_id: "error-session",
+          duration_ms: 1,
+          duration_api_ms: 0,
+          is_error: turn.isError,
+          num_turns: 1,
+          result: turn.isError ? (turn.errorReply ?? "") : (turn.after ?? turn.errorReply ?? ""),
+          stop_reason: "stop_sequence",
+          total_cost_usd: turn.usage?.totalCostUsd ?? 0,
+          usage: {
+            input_tokens: turn.usage?.inputTokens ?? 0,
+            output_tokens: turn.usage?.outputTokens ?? 0,
+            cache_read_input_tokens: turn.usage?.cachedInputTokens ?? 0,
+          },
+          modelUsage: {},
+          permission_denials: [],
+          uuid: `result-${index}`,
+        };
+      }
+    })();
+    return Object.assign(messages, {
+      close: () => {
+        void messages.return();
+      },
+      interrupt: async () => undefined,
+      setModel: async () => undefined,
+      setPermissionMode: async () => undefined,
+      supportedModels: async () => [],
+      supportedCommands: async () => [],
+    }) as ReturnType<ClaudeQueryFactory>;
+  };
+}
+
+async function runTurns(turns: ScriptedTurn[]) {
+  const client = new ClaudeAgentClient({
+    logger: createTestLogger(),
+    queryFactory: scriptedQuery(turns),
+    resolveBinary: async () => process.execPath,
+  });
+  const session = await client.createSession({
+    provider: "claude",
+    cwd: process.cwd(),
+  });
+  try {
+    const results = [];
+    for (const [index] of turns.entries()) {
+      results.push(await collectSessionTurnEvents(session, `turn ${index + 1}`));
+    }
+    return results;
+  } finally {
+    await session.close();
+  }
+}
+
+function assistantTexts(events: Awaited<ReturnType<typeof runTurns>>[number]): string[] {
+  return events.flatMap((event) =>
+    event.type === "timeline" && event.item.type === "assistant_message" ? [event.item.text] : [],
+  );
+}
+
+function terminal(events: Awaited<ReturnType<typeof runTurns>>[number]) {
+  return events.find(
+    (event) =>
+      event.type === "turn_completed" ||
+      event.type === "turn_failed" ||
+      event.type === "turn_canceled",
+  );
+}
+
+test("a turn Claude Code ends on an API error fails with the error, shown once", async () => {
+  const [limited, overloaded, next] = await runTurns([
+    { errorReply: LIMIT, isError: true },
+    {
+      before: "Checking the deploy doc.",
+      errorReply: "API Error: 529 Overloaded",
+      isError: true,
+    },
+    { after: "Done.", isError: false },
+  ]);
+
+  expect(terminal(limited)).toMatchObject({
+    type: "turn_failed",
+    error: LIMIT,
+  });
+  expect(assistantTexts(limited)).toEqual([]);
+
+  expect(terminal(overloaded)).toMatchObject({
+    type: "turn_failed",
+    error: "API Error: 529 Overloaded",
+  });
+  expect(assistantTexts(overloaded)).toEqual(["Checking the deploy doc."]);
+
+  // The session keeps working after a failed turn.
+  expect(terminal(next)).toMatchObject({ type: "turn_completed" });
+  expect(assistantTexts(next)).toEqual(["Done."]);
+});
+
+test("a failed turn retains its final token counts and cost", async () => {
+  const usage = {
+    inputTokens: 120,
+    cachedInputTokens: 80,
+    outputTokens: 30,
+    totalCostUsd: 0.012,
+  };
+  const [failed] = await runTurns([
+    { before: "Checking the deploy doc.", errorReply: LIMIT, isError: true, usage },
+  ]);
+
+  expect(failed).toContainEqual(
+    expect.objectContaining({ type: "usage_updated", usage: expect.objectContaining(usage) }),
+  );
+  expect(terminal(failed)).toMatchObject({ type: "turn_failed", error: LIMIT });
+  expect(assistantTexts(failed)).toEqual(["Checking the deploy doc."]);
+});
+
+test("an error reply the turn moves past is still shown", async () => {
+  const [recovered, completed] = await runTurns([
+    {
+      errorReply: "API Error: 529 Overloaded",
+      after: "Recovered.",
+      isError: false,
+    },
+    { errorReply: LIMIT, isError: false },
+  ]);
+
+  expect(terminal(recovered)).toMatchObject({ type: "turn_completed" });
+  expect(assistantTexts(recovered)).toEqual(["API Error: 529 Overloaded", "Recovered."]);
+
+  expect(terminal(completed)).toMatchObject({ type: "turn_completed" });
+  expect(assistantTexts(completed)).toEqual([LIMIT]);
+});

packages/server/src/server/agent/providers/claude/agent.import-titles.test.ts

@@ -1,3 +1,4 @@
+import { constants as bufferConstants } from "node:buffer";
 import * as fs from "node:fs/promises";
 import * as os from "node:os";
 import * as path from "node:path";
@@ -236,4 +237,64 @@ describe("importing a renamed Claude session", () => {
 
     await expect(importSession(sessionId)).rejects.toThrow();
   });
+
+  test("reads the rename from a transcript too large to hold as one string", async () => {
+    const sessionId = "oversized-transcript";
+    const projectDir = claudeProjectDirSync(SESSION_CWD, { configDir });
+    await fs.mkdir(projectDir, { recursive: true });
+    const transcript = path.join(projectDir, `${sessionId}.jsonl`);
+    // A hole stands in for the long session this used to fail on: the file only
+    // has to be longer than a string can be, and the rename sits at its end.
+    await fs.writeFile(transcript, "");
+    await fs.truncate(transcript, bufferConstants.MAX_STRING_LENGTH + 1);
+    await fs.appendFile(
+      transcript,
+      `\n${[
+        userRecord({ sessionId, content: "Trace the slow query" }),
+        { type: "custom-title", customTitle: "Slow query hunt", sessionId },
+      ]
+        .map((record) => JSON.stringify(record))
+        .join("\n")}\n`,
+    );
+
+    const imported = await importSession(sessionId);
+
+    expect(imported.config.title).toBe("Slow query hunt");
+  });
+
+  test("reads the rename from behind more records than one read covers", async () => {
+    const sessionId = "late-records-rename";
+    await writeSession({
+      sessionId,
+      records: [
+        userRecord({ sessionId, content: "Review this project" }),
+        { type: "custom-title", customTitle: "Scrollback review", sessionId },
+        // One record longer than a single read, then enough records to leave the
+        // rename several reads back from the end of the file.
+        userRecord({ sessionId, content: "x".repeat(128 * 1024) }),
+        ...Array.from({ length: 64 }, (_, index) =>
+          userRecord({ sessionId, content: `${"y".repeat(4 * 1024)} ${index}` }),
+        ),
+      ],
+    });
+
+    const imported = await importSession(sessionId);
+
+    expect(imported.config.title).toBe("Scrollback review");
+  });
+
+  test("imports a large record after the last rename without repeatedly copying it", async () => {
+    const sessionId = "large-record-after-rename";
+    await writeSession({
+      sessionId,
+      records: [
+        { type: "custom-title", customTitle: "Large attachment review", sessionId },
+        userRecord({ sessionId, content: "x".repeat(96 * 1024 * 1024) }),
+      ],
+    });
+
+    const imported = await importSession(sessionId);
+
+    expect(imported.config.title).toBe("Large attachment review");
+  });
 });

packages/server/src/server/agent/providers/claude/agent.runtime-info.test.ts

@@ -0,0 +1,48 @@
+import pino from "pino";
+import { describe, expect, test } from "vitest";
+
+import { ClaudeAgentClient } from "./agent.js";
+
+describe("Claude runtime info before a turn", () => {
+  const client = new ClaudeAgentClient({ logger: pino({ level: "silent" }) });
+
+  test("reports the explicitly configured model before the first turn", async () => {
+    const session = await client.createSession({
+      provider: "claude",
+      cwd: process.cwd(),
+      model: "claude-haiku-5-5",
+    });
+    try {
+      expect(await session.getRuntimeInfo()).toMatchObject({ model: "claude-haiku-5-5" });
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("reports the persisted configured model after rebuilding an idle session", async () => {
+    const session = await client.resumeSession({
+      provider: "claude",
+      sessionId: "runtime-model-before-turn",
+      metadata: { cwd: process.cwd(), model: "claude-haiku-5-5" },
+    });
+    try {
+      expect(await session.getRuntimeInfo()).toMatchObject({
+        sessionId: "runtime-model-before-turn",
+        model: "claude-haiku-5-5",
+      });
+      expect((await session.getRuntimeInfo()).model).toBe("claude-haiku-5-5");
+      expect((await session.getRuntimeInfo()).extra?.runtimeModel).toBeUndefined();
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("leaves the model unknown when Claude will choose its default", async () => {
+    const session = await client.createSession({ provider: "claude", cwd: process.cwd() });
+    try {
+      expect((await session.getRuntimeInfo()).model).toBeNull();
+    } finally {
+      await session.close();
+    }
+  });
+});

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

@@ -2389,6 +2389,12 @@ class ClaudeAgentSession implements AgentSession {
   private mainTurnInFlight = false;
   private foregroundHasVisibleActivity = false;
   private activeTurnHasAssistantText = false;
+  /**
+   * The reply Claude Code writes when a turn stops on an API error ("You've hit your session
+   * limit", "API Error: 529 ..."), held back until the result says whether the turn failed: a
+   * failed turn shows the same text as its error, so showing both would repeat it.
+   */
+  private withheldErrorReply: AgentTimelineItem[] = [];
   private readonly contextUsage: ClaudeContextUsageState;
   private userMessageIds: string[] = [];
   private readonly emittedUserMessageIds = new Set<string>();
@@ -2460,7 +2466,7 @@ class ClaudeAgentSession implements AgentSession {
     const info: AgentRuntimeInfo = {
       provider: "claude",
       sessionId: this.claudeSessionId,
-      model: this.lastOptionsModel,
+      model: this.lastOptionsModel ?? this.config.model ?? null,
       modeId: this.currentMode ?? null,
       ...(this.lastRuntimeModel
         ? {
@@ -2531,6 +2537,7 @@ class ClaudeAgentSession implements AgentSession {
     this.activeForegroundTurnId = turnId;
     this.foregroundHasVisibleActivity = false;
     this.activeTurnHasAssistantText = false;
+    this.withheldErrorReply = [];
     this.contextUsage.beginTurn();
     this.transitionTurnState("foreground", "foreground turn started");
     this.clearRecentStderr();
@@ -4419,6 +4426,21 @@ class ClaudeAgentSession implements AgentSession {
               }) satisfies AgentStreamEvent,
           );
 
+    if (!readClaudeParentToolUseId(message)) {
+      if (message.type === "assistant" && message.error) {
+        this.withheldErrorReply.push(...assistantTimelineEvents.map((event) => event.item));
+        return messageEvents;
+      }
+      if (
+        message.type === "assistant" ||
+        message.type === "user" ||
+        message.type === "stream_event"
+      ) {
+        const released: AgentStreamEvent[] = [];
+        this.releaseWithheldErrorReply(released);
+        return [...released, ...messageEvents, ...assistantTimelineEvents];
+      }
+    }
     return [...messageEvents, ...assistantTimelineEvents];
   }
 
@@ -4966,6 +4988,23 @@ class ClaudeAgentSession implements AgentSession {
     events: AgentStreamEvent[],
   ): void {
     const usage = this.convertUsage(message);
+    // Claude Code reports a turn that stopped on an API error (a usage limit, an overloaded or
+    // rejected request) as `success` with `is_error`, carrying the error text in `result`.
+    if (message.subtype === "success" && message.is_error) {
+      const withheld = this.withheldErrorReply;
+      this.withheldErrorReply = [];
+      const resultText = typeof message.result === "string" ? message.result.trim() : "";
+      const withheldText = withheld
+        .map((item) => (item.type === "assistant_message" ? item.text : ""))
+        .join("\n")
+        .trim();
+      if (usage) events.push({ type: "usage_updated", provider: "claude", usage });
+      events.push(...this.sidechainTracker.finishAll("failed"));
+      events.push(this.buildTurnFailedEvent(resultText || withheldText || "Claude run failed"));
+      return;
+    }
+    const releasedErrorReply = this.withheldErrorReply.length > 0;
+    this.releaseWithheldErrorReply(events);
     if (message.subtype === "success") {
       events.push(...this.sidechainTracker.finishAll("completed"));
       // Built-in slash commands (e.g. /voice, /usage, "Unknown command: …")
@@ -4975,7 +5014,12 @@ class ClaudeAgentSession implements AgentSession {
       // accounting from provider gateways does not duplicate streamed output.
       const resultText = typeof message.result === "string" ? message.result.trim() : "";
       const outputTokens = message.usage?.output_tokens;
-      if (resultText.length > 0 && outputTokens === 0 && !this.activeTurnHasAssistantText) {
+      if (
+        resultText.length > 0 &&
+        outputTokens === 0 &&
+        !this.activeTurnHasAssistantText &&
+        !releasedErrorReply
+      ) {
         events.push({
           type: "timeline",
           provider: "claude",
@@ -4997,6 +5041,14 @@ class ClaudeAgentSession implements AgentSession {
     events.push(this.buildTurnFailedEvent(errorMessage));
   }
 
+  private releaseWithheldErrorReply(events: AgentStreamEvent[]): void {
+    if (this.withheldErrorReply.length === 0) return;
+    for (const item of this.withheldErrorReply) {
+      events.push({ type: "timeline", item, provider: "claude" });
+    }
+    this.withheldErrorReply = [];
+  }
+
   private createClaudeSessionChangedNotice(
     oldSessionId: string,
     newSessionId: string,
@@ -6853,40 +6905,79 @@ function normalizeClaudeSessionTitle(title: string | null): string | null {
 }
 
 const CLAUDE_CUSTOM_TITLE_TYPE = "custom-title";
+const CLAUDE_CUSTOM_TITLE_NEEDLE = Buffer.from(`"${CLAUDE_CUSTOM_TITLE_TYPE}"`);
+const CLAUDE_TITLE_SCAN_CHUNK_BYTES = 64 * 1024;
+
+function readClaudeCustomTitleFromRecord(record: Buffer): string | null {
+  // The needle also matches a prompt that quotes it, so a record still has to
+  // parse as a `custom-title` entry carrying a usable name. Records without the
+  // needle are never decoded, so a long one costs a byte search and nothing more.
+  if (!record.includes(CLAUDE_CUSTOM_TITLE_NEEDLE)) {
+    return null;
+  }
+  let parsed: unknown;
+  try {
+    parsed = JSON.parse(record.toString("utf8").trim());
+  } catch {
+    return null;
+  }
+  const entry = toObjectRecord(parsed);
+  if (!entry || entry.isSidechain || entry.type !== CLAUDE_CUSTOM_TITLE_TYPE) {
+    return null;
+  }
+  return normalizeClaudeSessionTitle(
+    typeof entry.customTitle === "string" ? entry.customTitle : null,
+  );
+}
 
 /**
  * The `custom-title` record is re-emitted every time the session is saved, so
- * the current name is the last usable one in the file. Searching backwards finds it without
- * walking the transcript, which matters because sessions reach tens of
- * megabytes.
+ * the current name is the last usable one in the file. Reading backwards a chunk
+ * at a time finds it without walking the transcript, which matters because
+ * sessions reach tens of megabytes, and never holds the file as one string, so a
+ * transcript past V8's maximum string length still imports.
  */
-function readLastClaudeCustomTitle(content: string): string | null {
-  const needle = `"${CLAUDE_CUSTOM_TITLE_TYPE}"`;
-  let at = content.lastIndexOf(needle);
-  while (at >= 0) {
-    const lineStart = content.lastIndexOf("\n", at) + 1;
-    const lineEnd = content.indexOf("\n", at);
-    const line = lineEnd === -1 ? content.slice(lineStart) : content.slice(lineStart, lineEnd);
-    let parsed: unknown;
-    try {
-      parsed = JSON.parse(line.trim());
-    } catch {
-      parsed = undefined;
-    }
-    const entry = toObjectRecord(parsed);
-    if (entry && !entry.isSidechain && entry.type === CLAUDE_CUSTOM_TITLE_TYPE) {
-      const title = normalizeClaudeSessionTitle(
-        typeof entry.customTitle === "string" ? entry.customTitle : null,
+async function readLastClaudeCustomTitle(filePath: string): Promise<string | null> {
+  const handle = await fsPromises.open(filePath, "r");
+  try {
+    const stats = await handle.stat();
+    if (!stats.isFile()) {
+      // Opening a directory succeeds on Windows even though reading it does not,
+      // so reject the way reading the transcript as a file would.
+      throw Object.assign(
+        new Error(`EISDIR: illegal operation on a directory, read '${filePath}'`),
+        { code: "EISDIR", path: filePath },
       );
-      if (title) {
-        return title;
-      }
     }
-    // The needle also matches a prompt that quotes it, and a record can carry no
-    // usable name, so keep looking backwards for one that does.
-    at = at <= 0 ? -1 : content.lastIndexOf(needle, at - 1);
+    let end = stats.size;
+    // Retain unfinished record pieces in reverse file order. Assemble each
+    // complete record once, so a large tool/image record costs linear work.
+    let pieces: Buffer[] = [];
+    while (end > 0) {
+      const length = Math.min(CLAUDE_TITLE_SCAN_CHUNK_BYTES, end);
+      const chunk = Buffer.alloc(length);
+      await handle.read(chunk, 0, length, end - length);
+      let recordEnd = chunk.length;
+      for (
+        let at = chunk.lastIndexOf(10);
+        at >= 0;
+        at = at === 0 ? -1 : chunk.lastIndexOf(10, at - 1)
+      ) {
+        pieces.push(chunk.subarray(at + 1, recordEnd));
+        const record = Buffer.concat(pieces.toReversed());
+        const title = readClaudeCustomTitleFromRecord(record);
+        if (title) return title;
+        pieces = [];
+        recordEnd = at;
+      }
+      if (recordEnd > 0) pieces.push(chunk.subarray(0, recordEnd));
+      end -= length;
+    }
+    // The first record in the file has no newline in front of it.
+    return readClaudeCustomTitleFromRecord(Buffer.concat(pieces.toReversed()));
+  } finally {
+    await handle.close();
   }
-  return null;
 }
 
 function isErrnoException(error: unknown): error is NodeJS.ErrnoException {
@@ -6898,9 +6989,8 @@ async function readClaudeRenamedSessionTitle(input: {
   sessionId: string;
   configDir: string;
 }): Promise<string | null> {
-  let content: string;
   try {
-    content = await fsPromises.readFile(await claudeTranscriptPath(input), "utf8");
+    return await readLastClaudeCustomTitle(await claudeTranscriptPath(input));
   } catch (error) {
     // A session with no transcript on disk has no rename to carry over. Every
     // other read failure is a real fault and stays explicit.
@@ -6909,7 +6999,6 @@ async function readClaudeRenamedSessionTitle(input: {
     }
     throw error;
   }
-  return readLastClaudeCustomTitle(content);
 }
 
 function normalizeImportablePromptPreview(text: string): string | null {

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

@@ -150,6 +150,15 @@ function isCodexAlreadyUnarchivedError(error: unknown, threadId: string): boolea
 
 const TURN_START_TIMEOUT_MS = 90 * 1000;
 const INTERRUPT_TIMEOUT_MS = 2_000;
+// `turn/interrupt` ends a turn but leaves the shell commands it started running.
+// Each terminate is bounded by the first constant. A turn start waits for the
+// interrupted turn's commands to end, at most the interrupt's own bound plus one
+// terminate's and a margin, so it gives up only when a request outlives its timeout.
+const INTERRUPTED_TURN_TERMINATE_TIMEOUT_MS = 2_000;
+const INTERRUPTED_TURN_CLEANUP_WAIT_MS =
+  INTERRUPT_TIMEOUT_MS + INTERRUPTED_TURN_TERMINATE_TIMEOUT_MS + 500;
+// Turns whose commands are remembered; older turns are forgotten first.
+const INTERRUPTED_TURN_TRACKED_TURNS = 16;
 const CODEX_PROVIDER = "codex" as const;
 // These native provider identities belong to Codex, not to an older custom
 // Paseo provider. Never include them in a custom provider's legacy import scan.
@@ -3476,6 +3485,14 @@ export class CodexAppServerAgentSession implements AgentSession {
   private pendingFileChangeOutputDeltas = new Map<string, string[]>();
   private pendingAssistantMessageBoundary = false;
   private terminalCommandByProcessId = new Map<string, string>();
+  // The root thread's still-running command process ids, keyed by the native turn that
+  // started them. Codex's `thread/backgroundTerminals/list` carries no turn, so
+  // `item/started` is the only record of which turn owns a process.
+  private terminalProcessIdsByTurnId = new Map<
+    string,
+    { processIds: Set<string>; terminatedProcessIds: Set<string>; interruptAccepted: boolean }
+  >();
+  private interruptedTurnCleanup: Promise<void> | null = null;
   private pendingUnlabeledTerminalInteractions = new Map<
     string,
     Array<{ callId: string; stdin: string | null }>
@@ -4431,6 +4448,9 @@ export class CodexAppServerAgentSession implements AgentSession {
         hasDeveloperInstructions: turnStart.hasDeveloperInstructions,
         hasCodexConfig: turnStart.hasCodexConfig,
       });
+      // The interrupted turn's commands end before this turn begins, so none of them
+      // outlives it into the new turn.
+      await this.waitForInterruptedTurnCleanup();
       if (pendingStart.cancelRequested) {
         throw new Error("Codex turn start was interrupted before reaching Codex");
       }
@@ -5036,15 +5056,27 @@ export class CodexAppServerAgentSession implements AgentSession {
     if (!turnId || (foregroundTurnId && this.activeForegroundTurnId !== foregroundTurnId)) {
       throw new Error("Cannot interrupt Codex before turn/started identifies the active turn");
     }
+    const client = this.client;
+    const threadId = this.currentThreadId;
+    const interruptRequest = client.request(
+      "turn/interrupt",
+      {
+        threadId,
+        turnId,
+      },
+      INTERRUPT_TIMEOUT_MS,
+    );
+    // Once Codex accepts the interrupt, end the commands this turn started. Registered
+    // before the await, so a turn start racing this interrupt waits for it. An already-idle
+    // turn was not interrupted, and what it left running stays.
+    this.trackInterruptedTurnCleanup(
+      interruptRequest.then(
+        () => this.endInterruptedTurnProcesses(client, threadId, turnId),
+        () => undefined,
+      ),
+    );
     try {
-      await this.client.request(
-        "turn/interrupt",
-        {
-          threadId: this.currentThreadId,
-          turnId,
-        },
-        INTERRUPT_TIMEOUT_MS,
-      );
+      await interruptRequest;
     } catch (error) {
       if (!isCodexAlreadyIdleInterrupt(error)) {
         throw error;
@@ -5085,6 +5117,137 @@ export class CodexAppServerAgentSession implements AgentSession {
     return !this.client.hasPendingRequests();
   }
 
+  // Remember which turn started a command, so an interrupt ends that turn's commands and
+  // no other turn's. Only the root thread's: a subagent's commands live on its own thread.
+  // A command that starts after its turn's interrupt was accepted is ended as soon as it
+  // is seen.
+  private recordTurnTerminalProcess(
+    parsed: Extract<ParsedCodexNotification, { kind: "item_started" | "item_completed" }>,
+  ): void {
+    const processId = parsed.item.processId;
+    if (
+      parsed.source !== "item" ||
+      !parsed.turnId ||
+      !parsed.threadId ||
+      parsed.threadId !== this.currentThreadId ||
+      typeof processId !== "string" ||
+      !processId ||
+      normalizeCodexThreadItemType(
+        typeof parsed.item.type === "string" ? parsed.item.type : undefined,
+      ) !== "commandExecution"
+    ) {
+      return;
+    }
+    const record = this.terminalProcessIdsByTurnId.get(parsed.turnId);
+    if (parsed.kind === "item_completed") {
+      // Only a completed command has exited. A failed one can still be running:
+      // Codex reports an interrupted turn's live command as failed.
+      if (parsed.item.status === "completed") record?.processIds.delete(processId);
+      return;
+    }
+    const current = record ?? this.getTurnTerminalRecord(parsed.turnId);
+    current.processIds.add(processId);
+    if (current.interruptAccepted && this.client) {
+      this.trackInterruptedTurnCleanup(
+        this.endInterruptedTurnProcesses(this.client, parsed.threadId, parsed.turnId),
+      );
+    }
+  }
+
+  // An accepted interrupt needs a record even before any of its turn's commands is seen:
+  // an `item/started` can still be queued behind an earlier notification's conversion.
+  private getTurnTerminalRecord(turnId: string) {
+    const existing = this.terminalProcessIdsByTurnId.get(turnId);
+    if (existing) return existing;
+    const record = {
+      processIds: new Set<string>(),
+      terminatedProcessIds: new Set<string>(),
+      interruptAccepted: false,
+    };
+    this.terminalProcessIdsByTurnId.set(turnId, record);
+    for (const oldestTurnId of this.terminalProcessIdsByTurnId.keys()) {
+      if (this.terminalProcessIdsByTurnId.size <= INTERRUPTED_TURN_TRACKED_TURNS) break;
+      this.terminalProcessIdsByTurnId.delete(oldestTurnId);
+    }
+    return record;
+  }
+
+  // A later sweep joins the pending one, so a turn start waits for both.
+  private trackInterruptedTurnCleanup(cleanup: Promise<void>): void {
+    const tracked = Promise.all([this.interruptedTurnCleanup, cleanup])
+      .then(() => undefined)
+      .finally(() => {
+        if (this.interruptedTurnCleanup === tracked) this.interruptedTurnCleanup = null;
+      });
+    this.interruptedTurnCleanup = tracked;
+  }
+
+  // Terminate each process the interrupted turn started, by its own id, once each.
+  // `thread/backgroundTerminals/clean` is never used: it ends every terminal on the thread,
+  // an earlier turn's included. A process already gone answers `terminated: false`. Never
+  // rejects; failures are logged.
+  private async endInterruptedTurnProcesses(
+    client: CodexAppServerClient,
+    threadId: string | null,
+    turnId: string,
+  ): Promise<void> {
+    if (!threadId) return;
+    const record = this.getTurnTerminalRecord(turnId);
+    record.interruptAccepted = true;
+    const processIds = [...record.processIds].filter(
+      (processId) => !record.terminatedProcessIds.has(processId),
+    );
+    for (const processId of processIds) record.terminatedProcessIds.add(processId);
+    const logContext = { agentId: this.agentId, provider: CODEX_PROVIDER, threadId, turnId };
+    await Promise.all(
+      processIds.map(async (processId) => {
+        try {
+          const response = await client.request(
+            "thread/backgroundTerminals/terminate",
+            { threadId, processId },
+            INTERRUPTED_TURN_TERMINATE_TIMEOUT_MS,
+          );
+          this.logger.info(
+            { ...logContext, processId, terminated: toObjectRecord(response)?.terminated ?? null },
+            "provider.codex.interrupt.terminal_terminated",
+          );
+        } catch (error) {
+          this.logger.warn(
+            { ...logContext, processId, err: error },
+            "provider.codex.interrupt.terminal_terminate_failed",
+          );
+        }
+      }),
+    );
+  }
+
+  // A new turn starts only after every pending sweep of interrupted turns' commands has
+  // ended, or after the bound, whichever is first.
+  private async waitForInterruptedTurnCleanup(): Promise<void> {
+    if (!this.interruptedTurnCleanup) return;
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    const timedOut = await Promise.race([
+      (async () => {
+        while (this.interruptedTurnCleanup) await this.interruptedTurnCleanup;
+        return false;
+      })(),
+      new Promise<boolean>((resolve) => {
+        timer = setTimeout(() => resolve(true), INTERRUPTED_TURN_CLEANUP_WAIT_MS);
+      }),
+    ]);
+    clearTimeout(timer);
+    if (timedOut) {
+      this.logger.warn(
+        {
+          agentId: this.agentId,
+          provider: CODEX_PROVIDER,
+          waitMs: INTERRUPTED_TURN_CLEANUP_WAIT_MS,
+        },
+        "provider.codex.interrupt.terminal_cleanup_timed_out",
+      );
+    }
+  }
+
   async close(): Promise<void> {
     this.closed = true;
     this.clearPendingPermissions();
@@ -6209,6 +6372,11 @@ export class CodexAppServerAgentSession implements AgentSession {
       return;
     }
     this.completePendingRootCompactions();
+    // What a turn that was not interrupted leaves running, it left on purpose; no later
+    // interrupt may end it.
+    if (parsed.status !== "interrupted" && this.currentTurnId) {
+      this.terminalProcessIdsByTurnId.delete(this.currentTurnId);
+    }
     if (parsed.status === "failed") {
       this.emitEvent({
         type: "turn_failed",
@@ -6613,6 +6781,7 @@ export class CodexAppServerAgentSession implements AgentSession {
   private async handleItemCompletedNotification(
     parsed: Extract<ParsedCodexNotification, { kind: "item_completed" }>,
   ): Promise<void> {
+    this.recordTurnTerminalProcess(parsed);
     // Codex emits mirrored lifecycle notifications via both `codex/event/item_*`
     // and canonical `item/*`. Render ordinary items only from the canonical
     // channel, but accept a legacy-only child announcement so it can establish
@@ -6775,6 +6944,7 @@ export class CodexAppServerAgentSession implements AgentSession {
   private async handleItemStartedNotification(
     parsed: Extract<ParsedCodexNotification, { kind: "item_started" }>,
   ): Promise<void> {
+    this.recordTurnTerminalProcess(parsed);
     if (shouldIgnoreMirroredLifecycleItem(parsed.source, parsed.item)) {
       return;
     }

packages/server/src/server/agent/providers/codex-interrupted-turn-terminals.test.ts

@@ -0,0 +1,374 @@
+import { afterEach, describe, expect, test, vi } from "vitest";
+
+import type { AgentSession } from "../agent-sdk-types.js";
+import { createTestLogger } from "../../../test-utils/test-logger.js";
+import { CodexAppServerAgentSession } from "./codex-app-server-agent.js";
+import {
+  createFakeCodexAppServer,
+  type FakeCodexAppServer,
+  waitForNextEvent,
+  waitForTimelineToolCall,
+} from "./codex/test-utils/fake-app-server.js";
+
+const THREAD = "thread-1";
+const TERMINATE = "thread/backgroundTerminals/terminate";
+
+type Handler = (params: unknown) => unknown;
+
+function deferred<T = void>() {
+  let resolve!: (value: T) => void;
+  const promise = new Promise<T>((resolvePromise) => {
+    resolve = resolvePromise;
+  });
+  return { promise, resolve };
+}
+
+function paramsOf(value: unknown): Record<string, unknown> {
+  return value && typeof value === "object" ? (value as Record<string, unknown>) : {};
+}
+
+let sessions: AgentSession[] = [];
+
+afterEach(async () => {
+  for (const session of sessions) await session.close().catch(() => undefined);
+  sessions = [];
+});
+
+/**
+ * A Codex app-server whose interrupt ends the turn the way Codex does: an `interrupted`
+ * turn completion, then the response. `terminate` answers per process.
+ */
+function createHarness(
+  options: {
+    terminate?: (processId: string) => unknown;
+    interruptResponse?: (completeTurn: () => void) => unknown;
+  } = {},
+) {
+  const handlers: Record<string, Handler> = {
+    "turn/interrupt": () => {
+      const completeTurn = () =>
+        appServer.completeTurn({ threadId: THREAD, status: "interrupted" });
+      if (options.interruptResponse) return options.interruptResponse(completeTurn);
+      completeTurn();
+      return {};
+    },
+    [TERMINATE]: (params) =>
+      options.terminate?.(String(paramsOf(params).processId)) ?? { terminated: true },
+  };
+  const appServer: FakeCodexAppServer = createFakeCodexAppServer(handlers);
+  const session = new CodexAppServerAgentSession(
+    {
+      provider: "codex",
+      cwd: "/workspace/project",
+      modeId: "auto",
+      model: "gpt-5.4",
+    },
+    null,
+    createTestLogger(),
+    async () => appServer.child,
+  );
+  sessions.push(session);
+
+  function requests(method: string): Record<string, unknown>[] {
+    return appServer
+      .requests()
+      .filter((message) => message.method === method && "id" in message)
+      .map((message) => paramsOf(message.params));
+  }
+
+  function terminated(): string[] {
+    return requests(TERMINATE)
+      .map((params) => String(params.processId))
+      .sort();
+  }
+
+  // Starts a turn and lets Codex identify it as `nativeTurnId`.
+  async function startTurn(prompt: string, nativeTurnId: string): Promise<void> {
+    const before = requests("turn/start").length;
+    const started = session.startTurn(prompt);
+    await vi.waitFor(() => expect(requests("turn/start")).toHaveLength(before + 1), {
+      timeout: 5_000,
+    });
+    const turnStarted = waitForNextEvent(session, "turn_started");
+    appServer.startsTurn({ threadId: THREAD, turnId: nativeTurnId });
+    await started;
+    await turnStarted;
+  }
+
+  async function runsCommand(params: { turnId: string; processId: string }): Promise<void> {
+    const itemId = `exec-${params.processId}`;
+    const shown = waitForTimelineToolCall(session, itemId);
+    appServer.startsTerminalCommand({
+      threadId: THREAD,
+      turnId: params.turnId,
+      itemId,
+      processId: params.processId,
+      command: `sleep ${params.processId}`,
+    });
+    await shown;
+  }
+
+  async function completesTurn(): Promise<void> {
+    const completed = waitForNextEvent(session, "turn_completed");
+    appServer.completeTurn({ threadId: THREAD });
+    await completed;
+  }
+
+  // Interrupts the running turn and waits for the session to report it canceled, as a
+  // caller does before it starts the next turn: Codex notifications are handled in order
+  // but asynchronously, so the interrupt's answer can arrive before the turn has ended.
+  async function interrupts(): Promise<void> {
+    const canceled = waitForNextEvent(session, "turn_canceled");
+    await session.interrupt();
+    await canceled;
+  }
+
+  return {
+    appServer,
+    session,
+    requests,
+    terminated,
+    startTurn,
+    runsCommand,
+    completesTurn,
+    interrupts,
+  };
+}
+
+// Turn A leaves a server running on purpose and completes. Turn B starts two
+// commands, and a subagent on its own thread starts a third, then B is interrupted.
+async function interruptTurnB(harness: ReturnType<typeof createHarness>): Promise<void> {
+  await harness.startTurn("start a server", "native-A");
+  await harness.runsCommand({ turnId: "native-A", processId: "101" });
+  await harness.completesTurn();
+
+  await harness.startTurn("run two commands", "native-B");
+  await harness.runsCommand({ turnId: "native-B", processId: "201" });
+  // A subagent thread's command shows no row here; the next command's row shows it
+  // was read, because notifications are read in order.
+  harness.appServer.startsTerminalCommand({
+    threadId: "child-thread",
+    turnId: "native-B",
+    itemId: "exec-209",
+    processId: "209",
+    command: "sleep 209",
+  });
+  await harness.runsCommand({ turnId: "native-B", processId: "202" });
+  await harness.interrupts();
+}
+
+describe("an interrupted Codex turn's commands end before the next turn", () => {
+  test("terminates exactly the processes the interrupted turn started, by id, and never cleans", async () => {
+    const harness = createHarness();
+    await interruptTurnB(harness);
+    await harness.startTurn("correction", "native-C");
+
+    expect(harness.terminated()).toEqual(["201", "202"]);
+    expect(harness.requests(TERMINATE).every((params) => params.threadId === THREAD)).toBe(true);
+    // `clean` would end every terminal on the thread, an earlier turn's included.
+    expect(
+      harness.appServer
+        .requests()
+        .filter((message) => String(message.method).startsWith("thread/backgroundTerminals/"))
+        .map((message) => message.method),
+    ).toEqual([TERMINATE, TERMINATE]);
+    harness.appServer.assertNoErrors();
+  });
+
+  test("leaves the next turn's commands running when Codex starts that turn before acknowledging the interrupt", async () => {
+    const acknowledge = deferred();
+    const harness = createHarness({
+      // Codex ends B, starts a turn of its own (a goal continuation), and only then
+      // answers the interrupt.
+      interruptResponse: (completeTurn) => {
+        completeTurn();
+        return acknowledge.promise.then(() => ({}));
+      },
+    });
+    await harness.startTurn("run two commands", "native-B");
+    await harness.runsCommand({ turnId: "native-B", processId: "201" });
+    await harness.runsCommand({ turnId: "native-B", processId: "202" });
+    const interrupting = harness.session.interrupt();
+    await vi.waitFor(() => expect(harness.requests("turn/interrupt")).toHaveLength(1));
+    const autonomousTurn = waitForNextEvent(harness.session, "turn_started");
+    harness.appServer.startsTurn({ threadId: THREAD, turnId: "native-C" });
+    await autonomousTurn;
+    await harness.runsCommand({ turnId: "native-C", processId: "301" });
+
+    acknowledge.resolve();
+    await interrupting;
+    await vi.waitFor(() => expect(harness.terminated()).toEqual(["201", "202"]));
+    await new Promise((resolve) => setTimeout(resolve, 50));
+    expect(harness.terminated()).toEqual(["201", "202"]);
+  });
+
+  test("sends the next turn/start only after the terminates are answered", async () => {
+    const answer = deferred();
+    const harness = createHarness({
+      terminate: () => answer.promise.then(() => ({ terminated: true })),
+    });
+    await interruptTurnB(harness);
+    const next = harness.startTurn("correction", "native-C");
+    await vi.waitFor(() => expect(harness.terminated()).toEqual(["201", "202"]));
+    await new Promise((resolve) => setTimeout(resolve, 200));
+    expect(harness.requests("turn/start")).toHaveLength(2);
+
+    answer.resolve();
+    await next;
+    expect(harness.requests("turn/start")).toHaveLength(3);
+  });
+
+  test.each(["initial", "late"] as const)(
+    "tracks both cleanup sweeps when the %s answer arrives first",
+    async (answeredFirst) => {
+      const first = deferred();
+      const second = deferred();
+      const harness = createHarness({
+        terminate: (processId) =>
+          (processId === "201" ? first : second).promise.then(() => ({ terminated: true })),
+      });
+      await harness.startTurn("run commands", "native-B");
+      await harness.runsCommand({ turnId: "native-B", processId: "201" });
+      await harness.interrupts();
+      await vi.waitFor(() => expect(harness.terminated()).toEqual(["201"]));
+      const next = harness.startTurn("correction", "native-C");
+      await new Promise((resolve) => setTimeout(resolve, 100));
+      expect(harness.requests("turn/start")).toHaveLength(1);
+      // The next turn is already waiting when another process is announced.
+      harness.appServer.startsTerminalCommand({
+        threadId: THREAD,
+        turnId: "native-B",
+        itemId: "late-202",
+        processId: "202",
+        command: "sleep 202",
+      });
+      harness.appServer.startsTerminalCommand({
+        threadId: THREAD,
+        turnId: "native-B",
+        itemId: "duplicate-202",
+        processId: "202",
+        command: "sleep 202",
+      });
+      await vi.waitFor(() => expect(harness.terminated()).toEqual(["201", "202"]));
+      (answeredFirst === "initial" ? first : second).resolve();
+      await new Promise((resolve) => setTimeout(resolve, 100));
+      expect(harness.requests("turn/start")).toHaveLength(1);
+      (answeredFirst === "initial" ? second : first).resolve();
+      await next;
+      expect(harness.terminated()).toEqual(["201", "202"]);
+      // Accepted state survives the completed sweep and a newer foreground turn.
+      harness.appServer.startsTerminalCommand({
+        threadId: THREAD,
+        turnId: "native-B",
+        itemId: "late-203",
+        processId: "203",
+        command: "sleep 203",
+      });
+      await vi.waitFor(() => expect(harness.terminated()).toEqual(["201", "202", "203"]));
+      harness.appServer.assertNoErrors();
+    },
+  );
+
+  test("keeps the first sweep pending when a late sweep answers before the next start", async () => {
+    const first = deferred();
+    const harness = createHarness({
+      terminate: (processId) =>
+        processId === "201"
+          ? first.promise.then(() => ({ terminated: true }))
+          : { terminated: true },
+    });
+    await harness.startTurn("run commands", "native-B");
+    await harness.runsCommand({ turnId: "native-B", processId: "201" });
+    await harness.interrupts();
+    harness.appServer.startsTerminalCommand({
+      threadId: THREAD,
+      turnId: "native-B",
+      itemId: "late-202",
+      processId: "202",
+      command: "sleep 202",
+    });
+    await vi.waitFor(() => expect(harness.terminated()).toEqual(["201", "202"]));
+    const next = harness.startTurn("correction", "native-C");
+    await new Promise((resolve) => setTimeout(resolve, 100));
+    expect(harness.requests("turn/start")).toHaveLength(1);
+    first.resolve();
+    await next;
+    harness.appServer.assertNoErrors();
+  });
+
+  test("forgets completed commands but retains failed commands at interrupt", async () => {
+    const harness = createHarness();
+    await harness.startTurn("run commands", "native-B");
+    for (const processId of ["201", "202"]) {
+      await harness.runsCommand({ turnId: "native-B", processId });
+      harness.appServer.child.stdout.write(
+        JSON.stringify({
+          method: "item/completed",
+          params: {
+            threadId: THREAD,
+            turnId: "native-B",
+            item: {
+              id: `exec-${processId}`,
+              type: "commandExecution",
+              processId,
+              command: `sleep ${processId}`,
+              status: processId === "201" ? "completed" : "failed",
+              exitCode: processId === "201" ? 0 : null,
+            },
+          },
+        }) + "\n",
+      );
+    }
+    // A following command's row is a barrier for both completion handlers.
+    await harness.runsCommand({ turnId: "native-B", processId: "203" });
+    await harness.interrupts();
+    await harness.startTurn("correction", "native-C");
+    expect(harness.terminated()).toEqual(["202", "203"]);
+    harness.appServer.assertNoErrors();
+  });
+
+  test("retains accepted interrupts for only the latest sixteen turns", async () => {
+    const harness = createHarness();
+    for (let index = 0; index < 17; index++) {
+      await harness.startTurn("run a command", `native-${index}`);
+      await harness.interrupts();
+    }
+    await harness.startTurn("correction", "native-current");
+    harness.appServer.startsTerminalCommand({
+      threadId: THREAD,
+      turnId: "native-0",
+      itemId: "evicted-201",
+      processId: "201",
+      command: "sleep 201",
+    });
+    harness.appServer.startsTerminalCommand({
+      threadId: THREAD,
+      turnId: "native-16",
+      itemId: "retained-202",
+      processId: "202",
+      command: "sleep 202",
+    });
+    await harness.runsCommand({ turnId: "native-current", processId: "301" });
+    expect(harness.terminated()).toEqual(["202"]);
+    harness.appServer.assertNoErrors();
+  });
+
+  test("a failing or unanswered terminate does not block or fail the next turn", async () => {
+    const harness = createHarness({
+      // 201 is refused while the next turn waits; 202 is never answered.
+      terminate: (processId) =>
+        processId === "201"
+          ? new Promise((resolve) => setTimeout(resolve, 300)).then(() => ({
+              __jsonRpcError: { code: -32600, message: "invalid background terminal process id" },
+            }))
+          : new Promise(() => {}),
+    });
+    await interruptTurnB(harness);
+    const startedAt = Date.now();
+    await harness.startTurn("correction", "native-C");
+
+    expect(Date.now() - startedAt).toBeLessThan(5_000);
+    expect(harness.requests("turn/start")).toHaveLength(3);
+    expect(harness.terminated()).toEqual(["201", "202"]);
+  }, 15_000);
+});

packages/server/src/server/agent/providers/opencode/v2/permissions.test.ts

@@ -4,6 +4,102 @@ import { createTestLogger } from "../../../../../test-utils/test-logger.js";
 import type { V2Api } from "./api.js";
 import { OpenCodeV2AgentClient } from "./agent.js";
 import { V2Harness } from "../test-utils/v2-harness.js";
+import type { AgentPermissionResponse } from "../../../agent-sdk-types.js";
+
+describe("OpenCode v2 denial reasons", () => {
+  test.each<{
+    response: AgentPermissionResponse;
+    decision: "reject" | "once" | "always";
+    message?: string;
+  }>([
+    {
+      response: { behavior: "deny", message: "Use the existing file instead." },
+      decision: "reject",
+      message: "Use the existing file instead.",
+    },
+    { response: { behavior: "deny" }, decision: "reject" },
+    { response: { behavior: "allow" }, decision: "once" },
+    { response: { behavior: "allow", selectedActionId: "always" }, decision: "always" },
+  ])("preserves a permission response: $response", async ({ response, decision, message }) => {
+    const harness = new V2Harness();
+    let pending = true;
+    harness.api.permission.list = async () =>
+      pending
+        ? [{ id: "approval", sessionID: "session", action: "shell", resources: ["pwd"] }]
+        : [];
+    const replies: Parameters<V2Api["permission"]["reply"]>[0][] = [];
+    harness.api.permission.reply = async (input) => {
+      replies.push(input);
+      pending = false;
+    };
+    const client = new OpenCodeV2AgentClient({
+      logger: createTestLogger(),
+      runtime: harness.runtime,
+    });
+    const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
+    try {
+      await session.respondToPermission("approval", response);
+      expect(replies).toEqual([
+        {
+          sessionID: "session",
+          requestID: "approval",
+          decision,
+          ...(message === undefined ? {} : { message }),
+        },
+      ]);
+      expect(session.getPendingPermissions()).toEqual([]);
+    } finally {
+      await session.close();
+    }
+  });
+
+  test.each([true, false])(
+    "explains cards superseded by a steer when clearing is %s",
+    async (clearPendingPermissions) => {
+      const harness = new V2Harness();
+      harness.autoComplete = false;
+      let pending = true;
+      harness.api.permission.list = async () =>
+        pending
+          ? [{ id: "approval", sessionID: "session", action: "shell", resources: ["pwd"] }]
+          : [];
+      const replies: Parameters<V2Api["permission"]["reply"]>[0][] = [];
+      harness.api.permission.reply = async (input) => {
+        replies.push(input);
+        pending = false;
+      };
+      const client = new OpenCodeV2AgentClient({
+        logger: createTestLogger(),
+        runtime: harness.runtime,
+      });
+      const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
+      try {
+        const { turnId } = await session.startTurn("Run pwd");
+        await expect.poll(() => harness.prompts).toEqual(["Run pwd"]);
+        await expect(
+          session.steerActiveTurn!("go ahead", { expectedTurnId: turnId, clearPendingPermissions }),
+        ).resolves.toEqual({ status: "accepted" });
+        expect(harness.prompts).toEqual(["Run pwd", "go ahead"]);
+        expect(replies).toEqual(
+          clearPendingPermissions
+            ? [
+                {
+                  sessionID: "session",
+                  requestID: "approval",
+                  decision: "reject",
+                  message:
+                    "The user answered with a message instead of approving. Their message follows.",
+                },
+              ]
+            : [],
+        );
+        expect(session.getPendingPermissions()).toHaveLength(clearPendingPermissions ? 0 : 1);
+      } finally {
+        await session.close();
+      }
+    },
+  );
+});
 
 describe("OpenCode v2 questions", () => {
   test("maps selected labels to native values and parses numeric and boolean answers", async () => {

packages/server/src/server/agent/providers/opencode/v2/permissions.ts

@@ -68,6 +68,9 @@ export class SessionPermissions {
         sessionID: this.permissionOwners.get(requestId) ?? this.id,
         requestID: requestId,
         decision: permissionReply(response),
+        ...(response.behavior === "deny" && response.message !== undefined
+          ? { message: response.message }
+          : {}),
       });
     }
     this.resolvePending(requestId, response);

packages/server/src/server/agent/providers/opencode/v2/session.ts

@@ -100,7 +100,11 @@ export class OpenCodeV2Session implements AgentSession {
         ),
       clearPermissions: async () => {
         for (const request of this.permissions.list())
-          await this.permissions.respondToPermission(request.id, { behavior: "deny" });
+          await this.permissions.respondToPermission(request.id, {
+            behavior: "deny",
+            message:
+              "The user answered with a message instead of approving. Their message follows.",
+          });
       },
     });
     this.usage = new SessionUsage({

packages/server/src/server/agent/tools/paseo-tools.ts

@@ -1590,7 +1590,7 @@ function createToolCatalog(
     {
       title: "Create agent",
       description:
-        "Create an agent. Agent-scoped creation defaults to your workspace and creates your subagent. Top-level creation without workspaceId creates a new local workspace. Requires provider/model (for example codex/gpt-5.4) and an initial prompt. Do not guess; call list_providers and list_models first if uncertain.",
+        "Create an agent. Agent-scoped creation defaults to your workspace and creates your subagent. Top-level creation without workspaceId creates a new local workspace. Requires provider/model (for example codex/gpt-5.4) and an initial prompt. For a different provider, set settings.modeId from inspect_provider; omission is accepted for targets with no modes and for unattended execution supported by the target. Do not guess; call list_providers and list_models first if uncertain.",
       inputSchema: createAgentInputSchema,
       outputSchema: {
         agentId: z.string(),

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

@@ -142,6 +142,7 @@ 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",
@@ -378,6 +379,9 @@ 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

@@ -183,7 +183,7 @@ test("ordinary routed APIs isolate scopes and enforce one-hop, disable and origi
   await paseo.dispose();
 });
 
-test("routed authority is bounded by the caller, with isolated simultaneous sessions", async () => {
+test("routed authority isolates simultaneous sessions and preserves ordinary permission RPC errors", async () => {
   const b = await daemon();
   const a = await daemon();
   const owner = await api(a);
@@ -210,33 +210,16 @@ test("routed authority is bounded by the caller, with isolated simultaneous sess
     code: "permission_denied",
   });
   expect((await owner.paseo.hosts.list()).hosts).toHaveLength(1);
-  await remote.dispose();
-});
-
-test("SDK host clients preserve ordinary permission RPC errors", async () => {
-  const b = await daemon();
-  const a = await daemon();
-  const owner = await api(a);
-  await owner.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` });
-  const serverId = (await online(owner.paseo)).hosts[0]!.serverId!;
-  const clients = [a, b].map((host) =>
-    createHostClient(`127.0.0.1:${host.port}`, {
-      clientId: crypto.randomUUID(),
-      sessionPermissions: ["daemon.read", "workspace.read"],
-      reconnect: { enabled: false },
-    }),
-  );
-  for (const client of clients) {
-    cleanup.push(() => client.close());
-    await client.connect();
-  }
-  const origin = createPaseoApi(clients[0]!);
-  const direct = createPaseoApi(clients[1]!);
-  const remote = await origin.hosts.client(serverId);
+  const directClient = createHostClient(`127.0.0.1:${b.port}`, {
+    clientId: crypto.randomUUID(),
+    sessionPermissions: ["daemon.read", "workspace.read"],
+    reconnect: { enabled: false },
+  });
+  await directClient.connect();
+  const direct = createPaseoApi(directClient);
   cleanup.push(async () => {
-    await remote.dispose();
     await direct.dispose();
-    await origin.dispose();
+    await directClient.close();
   });
   const input = { source: { kind: "directory" as const, path: b.staticDir } };
   const directError = (await direct.workspaces.create(input).catch((error: unknown) => error)) as {
@@ -247,6 +230,7 @@ test("SDK host clients preserve ordinary permission RPC errors", async () => {
   };
   expect(directError.code).toBe("access_denied");
   expect(remoteError.code).toBe(directError.code);
+  await remote.dispose();
 });
 
 test("distinct origin sessions cannot collide through a caller-selected route scope, including terminal binary output", async () => {

packages/server/src/server/daemon-e2e/create-agent-mode.e2e.test.ts

@@ -0,0 +1,67 @@
+import { mkdir, mkdtemp, rm } from "node:fs/promises";
+import path from "node:path";
+import { Client } from "@modelcontextprotocol/sdk/client/index.js";
+import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
+import { expect, test } from "vitest";
+import { createDaemonTestContext } from "../test-utils/index.js";
+
+test("an attended MCP caller can discover and supply a cross-provider child mode", async () => {
+  const artifacts = path.join(process.cwd(), ".artifacts");
+  await mkdir(artifacts, { recursive: true });
+  const root = await mkdtemp(path.join(artifacts, "create-agent-mode-"));
+  const ctx = await createDaemonTestContext({ paseoHomeRoot: root, mcpEnabled: true });
+  const mcp = new Client({ name: "mode-contract-test", version: "1" });
+  try {
+    const parent = await ctx.client.createAgent({
+      provider: "claude",
+      cwd: root,
+      modeId: "default",
+      title: "Attended parent",
+    });
+    await mcp.connect(
+      new StreamableHTTPClientTransport(
+        new URL(`http://127.0.0.1:${ctx.daemon.port}/mcp/agents?callerAgentId=${parent.id}`),
+      ),
+    );
+    const { tools } = await mcp.listTools();
+    const create = tools.find((tool) => tool.name === "create_agent");
+    expect.soft(create?.description).toContain("settings.modeId");
+
+    const args = {
+      provider: "codex/gpt-5.4-mini",
+      title: "Cross-provider child",
+      initialPrompt: "Reply with OK",
+    };
+    const refused = await mcp.callTool({ name: "create_agent", arguments: args });
+    expect(refused.isError).toBe(true);
+    expect.soft(refused.content).toEqual([
+      expect.objectContaining({
+        type: "text",
+        text: expect.stringContaining("Pass an explicit mode in settings.modeId"),
+      }),
+    ]);
+    expect((await ctx.client.fetchAgents({ subscribe: {} })).entries).toHaveLength(1);
+
+    const inspected = await mcp.callTool({
+      name: "inspect_provider",
+      arguments: { provider: "codex" },
+    });
+    expect(inspected.isError).not.toBe(true);
+    expect(inspected.structuredContent).toMatchObject({
+      modes: expect.arrayContaining([expect.objectContaining({ id: "auto" })]),
+    });
+    const created = await mcp.callTool({
+      name: "create_agent",
+      arguments: { ...args, settings: { modeId: "auto" } },
+    });
+    expect(created.isError).not.toBe(true);
+    expect(created.structuredContent).toMatchObject({ type: "codex", currentModeId: "auto" });
+    const childId = (created.structuredContent as { agentId: string }).agentId;
+    expect((await ctx.client.fetchAgent(childId))?.agent.currentModeId).toBe("auto");
+    expect((await ctx.client.waitForFinish(childId, 5000)).status).toBe("idle");
+  } finally {
+    await mcp.close();
+    await ctx.cleanup();
+    await rm(root, { recursive: true, force: true });
+  }
+}, 30_000);

packages/server/src/server/daemon-e2e/file-download.e2e.test.ts

@@ -102,53 +102,52 @@ describe("daemon E2E", () => {
       },
     );
 
-    test.each([
-      { fileName: "download.txt", disposition: 'attachment; filename="download.txt"' },
-      {
-        fileName: "café.txt",
-        disposition: "attachment; filename=\"caf?.txt\"; filename*=UTF-8''caf%C3%A9.txt",
-      },
-      {
-        fileName: "中文报告 (最终版).txt",
-        disposition:
-          "attachment; filename=\"???? (???).txt\"; filename*=UTF-8''%E4%B8%AD%E6%96%87%E6%8A%A5%E5%91%8A%20%28%E6%9C%80%E7%BB%88%E7%89%88%29.txt",
-      },
-      {
-        fileName: "報告書.md",
-        disposition:
-          "attachment; filename=\"???.md\"; filename*=UTF-8''%E5%A0%B1%E5%91%8A%E6%9B%B8.md",
-      },
-      {
-        fileName: "보고서.txt",
-        disposition:
-          "attachment; filename=\"???.txt\"; filename*=UTF-8''%EB%B3%B4%EA%B3%A0%EC%84%9C.txt",
-      },
-      {
-        fileName: "report-📄.txt",
-        disposition:
-          "attachment; filename=\"report-??.txt\"; filename*=UTF-8''report-%F0%9F%93%84.txt",
-      },
-    ])(
-      "issues token over WS and downloads $fileName via HTTP",
-      async ({ fileName, disposition }) => {
-        const cwd = tmpCwd();
-        try {
+    test("downloads ASCII and Unicode filenames through one workspace", async () => {
+      const cwd = tmpCwd();
+      try {
+        const agent = await ctx.client.createAgent({
+          provider: "codex",
+          model: CODEX_TEST_MODEL,
+          thinkingOptionId: CODEX_TEST_THINKING_OPTION_ID,
+          cwd,
+          title: "Download Token Test Agent",
+        });
+
+        expect(agent.id).toMatch(
+          /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/,
+        );
+
+        for (const { fileName, disposition } of [
+          { fileName: "download.txt", disposition: 'attachment; filename="download.txt"' },
+          {
+            fileName: "café.txt",
+            disposition: "attachment; filename=\"caf?.txt\"; filename*=UTF-8''caf%C3%A9.txt",
+          },
+          {
+            fileName: "中文报告 (最终版).txt",
+            disposition:
+              "attachment; filename=\"???? (???).txt\"; filename*=UTF-8''%E4%B8%AD%E6%96%87%E6%8A%A5%E5%91%8A%20%28%E6%9C%80%E7%BB%88%E7%89%88%29.txt",
+          },
+          {
+            fileName: "報告書.md",
+            disposition:
+              "attachment; filename=\"???.md\"; filename*=UTF-8''%E5%A0%B1%E5%91%8A%E6%9B%B8.md",
+          },
+          {
+            fileName: "보고서.txt",
+            disposition:
+              "attachment; filename=\"???.txt\"; filename*=UTF-8''%EB%B3%B4%EA%B3%A0%EC%84%9C.txt",
+          },
+          {
+            fileName: "report-📄.txt",
+            disposition:
+              "attachment; filename=\"report-??.txt\"; filename*=UTF-8''report-%F0%9F%93%84.txt",
+          },
+        ]) {
           const filePath = path.join(cwd, fileName);
-          const fileContents = "download test payload";
+          const fileContents = `download test payload: ${fileName}`;
           writeFileSync(filePath, fileContents, "utf-8");
 
-          const agent = await ctx.client.createAgent({
-            provider: "codex",
-            model: CODEX_TEST_MODEL,
-            thinkingOptionId: CODEX_TEST_THINKING_OPTION_ID,
-            cwd,
-            title: "Download Token Test Agent",
-          });
-
-          expect(agent.id).toMatch(
-            /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/,
-          );
-
           const tokenResponse = await ctx.client.requestDownloadToken(cwd, fileName);
 
           expect(tokenResponse.error).toBeNull();
@@ -167,12 +166,11 @@ describe("daemon E2E", () => {
 
           const body = await response.text();
           expect(body).toBe(fileContents);
-        } finally {
-          rmSync(cwd, { recursive: true, force: true });
         }
-      },
-      60000,
-    );
+      } finally {
+        rmSync(cwd, { recursive: true, force: true });
+      }
+    }, 60000);
 
     test("rejects invalid token", async () => {
       const response = await fetch(

packages/server/src/server/daemon-e2e/peer-message-permissions.test.ts

@@ -0,0 +1,129 @@
+import { mkdir, mkdtemp, rm } from "node:fs/promises";
+import path from "node:path";
+import type { FormInfo } from "@opencode/client";
+import { expect, test } from "vitest";
+
+import { createTestLogger } from "../../test-utils/test-logger.js";
+import { OpenCodeV2AgentClient } from "../agent/providers/opencode/v2/agent.js";
+import { V2Harness } from "../agent/providers/opencode/test-utils/v2-harness.js";
+import { createTestAgentClients } from "../test-utils/fake-agent-client.js";
+import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+import { DaemonClient } from "../test-utils/daemon-client.js";
+
+async function createQuestionHarness() {
+  const artifacts = path.resolve(".artifacts");
+  await mkdir(artifacts, { recursive: true });
+  const root = await mkdtemp(path.join(artifacts, "peer-permissions-"));
+  const runtime = new V2Harness();
+  runtime.autoComplete = false;
+  runtime.api.model.list = async () => ({ location: runtime.info.location, data: [] });
+  runtime.api.provider.list = async () => ({
+    location: runtime.info.location,
+    data: [{ id: "test", name: "Test", activation: "enabled", package: "test" }],
+  });
+  let form: FormInfo | null = null;
+  const answers: Array<Record<string, unknown>> = [];
+  runtime.api.session.form.list = async () => (form ? [form] : []);
+  runtime.api.session.form.cancel = async () => {
+    form = null;
+    runtime.finishExecution("interrupted");
+  };
+  runtime.api.session.form.reply = async ({ answer }) => {
+    answers.push(answer);
+    form = null;
+    runtime.finishExecution();
+  };
+  // A steer is admitted to the existing execution, rather than starting another one.
+  runtime.prompt = async (input) => {
+    runtime.prompts.push(input.text);
+    if (runtime.prompts.length === 1) runtime.startExecution();
+  };
+  const provider = new OpenCodeV2AgentClient({
+    logger: createTestLogger(),
+    runtime: runtime.runtime,
+  });
+  const daemon = await createTestPaseoDaemon({
+    agentClients: { ...createTestAgentClients(), opencode: provider },
+    paseoHomeRoot: path.join(root, "daemon"),
+    staticDir: path.join(root, "static"),
+    cleanup: false,
+  });
+  const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  try {
+    await client.connect();
+    await client.fetchAgents({ subscribe: {} });
+    const recipient = await client.createAgent({
+      provider: "opencode",
+      cwd: root,
+      title: "Waiting for a human answer",
+    });
+    const sender = await client.createAgent({ provider: "claude", cwd: root });
+    await client.sendMessage(recipient.id, "Ask which color to use");
+    await expect.poll(() => runtime.prompts).toHaveLength(1);
+    form = {
+      id: "color-question",
+      sessionID: "session",
+      title: "Color",
+      fields: [{ key: "color", title: "Color", type: "string" }],
+    };
+    runtime.push({ id: "question-created", created: 2, type: "form.created", data: { form } });
+    await client.waitForAgentUpsert(recipient.id, (agent) =>
+      (agent.pendingPermissions ?? []).some((request) => request.id === "color-question"),
+    );
+    return {
+      client,
+      recipient,
+      sender,
+      answers,
+      async close() {
+        await client.close();
+        await daemon.close();
+        await rm(root, { recursive: true, force: true });
+      },
+    };
+  } catch (error) {
+    await client.close();
+    await daemon.close();
+    await rm(root, { recursive: true, force: true });
+    throw error;
+  }
+}
+
+test("an agent-sourced steer leaves the question pending and answerable", async () => {
+  const { client, recipient, sender, answers, close } = await createQuestionHarness();
+  try {
+    await client.sendAgentMessage(recipient.id, "Peer status check", {
+      sourceAgentId: sender.id,
+    });
+    // The public wait includes admitted message delivery before reporting permission status.
+    const waiting = await client.waitForFinish(recipient.id);
+    expect(waiting.status).toBe("permission");
+    expect(waiting.final?.pendingPermissions).toMatchObject([
+      { id: "color-question", kind: "question" },
+    ]);
+    await client.respondToPermission(recipient.id, "color-question", {
+      behavior: "allow",
+      updatedInput: { answers: { Color: "Blue" } },
+    });
+    await client.waitForAgentUpsert(
+      recipient.id,
+      (agent) => (agent.pendingPermissions ?? []).length === 0 && agent.status === "idle",
+    );
+    expect(answers).toEqual([{ color: "Blue" }]);
+  } finally {
+    await close();
+  }
+});
+
+test("a human steer still dismisses the question", async () => {
+  const { client, recipient, answers, close } = await createQuestionHarness();
+  try {
+    await client.sendAgentMessage(recipient.id, "Use red instead", { activeTurnBehavior: "steer" });
+    const finished = await client.waitForFinish(recipient.id);
+    expect(finished.status).toBe("idle");
+    expect(finished.final?.pendingPermissions ?? []).toEqual([]);
+    expect(answers).toEqual([]);
+  } finally {
+    await close();
+  }
+});

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

@@ -34,7 +34,13 @@ const emptyUsageRuntime = {
   getUsageSourceRegistrations: () => [],
   fetchUsage: async () => undefined,
   discoverUsage: async () => [],
-} satisfies Pick<TestPluginRuntime, "getUsageSourceRegistrations" | "fetchUsage" | "discoverUsage">;
+  startSubscription: () => {
+    throw new Error("No subscriptions in this test");
+  },
+} satisfies Pick<
+  TestPluginRuntime,
+  "getUsageSourceRegistrations" | "fetchUsage" | "discoverUsage" | "startSubscription"
+>;
 
 async function createPlugin(id: string, source: string): Promise<string> {
   const directory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-service-"));

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

@@ -25,6 +25,7 @@ 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";
@@ -48,6 +49,7 @@ 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[];
@@ -539,6 +541,10 @@ 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,6 +60,14 @@ 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;
@@ -83,6 +91,7 @@ export type PluginProcessMessage =
   | {
       type: "ready";
       methods: string[];
+      subscriptions?: string[];
       providers: PluginProviderMetadata[];
       usageSources?: PluginUsageSourceMetadata[];
       hooks?: string[];
@@ -90,6 +99,8 @@ 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;
@@ -240,6 +251,16 @@ 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"),
@@ -272,6 +293,7 @@ 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(),
@@ -284,6 +306,20 @@ 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,8 +7,21 @@ import {
   type PluginProcessRequest,
 } from "./plugin-process-protocol.js";
 import type { UsageSourceRegistration } from "@getpaseo/plugin/server/usage";
-import type { SettingsDefinition, PluginRpcContract } from "@getpaseo/plugin";
-import type { PluginHandlerContext, PluginServerContribution } from "@getpaseo/plugin/server";
+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 { fileURLToPath } from "node:url";
 import type { ZodType } from "zod";
 import {
@@ -66,6 +79,7 @@ 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<
@@ -98,12 +112,36 @@ export function createPluginWorker(options: {
     if (!/^[a-z][a-z0-9._-]*$/.test(normalized)) {
       throw new Error(`Invalid plugin RPC method: ${method}`);
     }
-    if (handlers.has(normalized)) {
+    if (handlers.has(normalized) || subscriptions.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(contract.input.parse(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`);
@@ -284,8 +322,7 @@ export function createPluginWorker(options: {
         )
       : null;
     const contributedCleanup = contribute({
-      handle: (contract, handler) =>
-        register(contract, (input, context) => handler(contract.input.parse(input), context)),
+      handle,
       registerProvider,
       registerUsageSource,
       registerSettings,
@@ -311,6 +348,7 @@ 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))
@@ -327,6 +365,7 @@ 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;
@@ -365,7 +404,8 @@ export function createPluginWorker(options: {
       message.type === "provider.status" ||
       message.type === "provider.catalog_key" ||
       message.type === "usage.fetch" ||
-      message.type === "usage.discover"
+      message.type === "usage.discover" ||
+      message.type === "subscribe"
     ) {
       send({ type: "error", requestId: message.requestId, error: "Plugin is stopping" });
     } else if (message.type === "provider.connect") {
@@ -422,6 +462,14 @@ 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) {
@@ -525,6 +573,11 @@ 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({
@@ -541,6 +594,10 @@ export function createPluginWorker(options: {
         return registered.handler(input, { paseo });
       })
       .then((output) => registered.contract.output.parseAsync(output))
+      .then((output: unknown) => {
+        assertPluginWireValue(output, "output");
+        return output;
+      })
       .then(
         (output) => send({ type: "result", requestId: message.requestId, output }),
         (error) =>

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

@@ -0,0 +1,354 @@
+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 { 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 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(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);

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

@@ -0,0 +1,108 @@
+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

@@ -0,0 +1,56 @@
+/** 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,6 +1399,44 @@ 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 {

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

@@ -34,6 +34,10 @@ import type {
   PluginUsageSourceMetadata,
 } from "./plugin-process-protocol.js";
 import { PluginProcessMessageSchema } from "./plugin-process-protocol.js";
+import {
+  BufferedPluginSubscription,
+  type PluginRuntimeSubscription,
+} from "./runtime-subscription.js";
 import { PluginSessionSocket } from "./session-socket.js";
 import { InternalPluginChild } from "./internal-child.js";
 import { evaluateBundle } from "./bundle-evaluator.js";
@@ -75,6 +79,8 @@ interface LoadedPlugin {
   /** Lets a client that already holds this bundle skip downloading it again. */
   clientBundleHash: string;
   methods: ReadonlySet<string>;
+  subscriptionMethods: ReadonlySet<string>;
+  subscriptions: Map<string, BufferedPluginSubscription>;
   hooks: string[];
   providers: readonly PluginProviderMetadata[];
   usageSources: readonly PluginUsageSourceMetadata[];
@@ -514,6 +520,40 @@ export class PluginRuntime {
     return this.request(loaded, { type: "invoke", requestId: randomUUID(), method, input });
   }
 
+  /** Starts one subscriber's handler in the plugin process. */
+  startSubscription(pluginId: string, method: string, input: unknown): PluginRuntimeSubscription {
+    const loaded = this.plugins.get(pluginId);
+    if (!loaded) throw new Error(`Plugin is not available: ${pluginId}`);
+    if (!loaded.subscriptionMethods.has(method))
+      throw new Error(`Plugin ${pluginId} does not contribute subscription ${method}`);
+    const subscriptionId = randomUUID();
+    const subscription = new BufferedPluginSubscription();
+    loaded.subscriptions.set(subscriptionId, subscription);
+    const ready = this.request(loaded, {
+      type: "subscribe",
+      requestId: randomUUID(),
+      subscriptionId,
+      method,
+      input,
+    }).then(
+      () => undefined,
+      (error: unknown) => {
+        loaded.subscriptions.delete(subscriptionId);
+        throw error;
+      },
+    );
+    return {
+      ready,
+      attach: (sink) => subscription.attach(sink),
+      release: () => {
+        subscription.released();
+        if (!loaded.subscriptions.delete(subscriptionId)) return;
+        if (loaded.child?.connected)
+          void send(loaded.child, { type: "unsubscribe", subscriptionId }).catch(() => undefined);
+      },
+    };
+  }
+
   emit<Name extends keyof PluginHookEvents>(name: Name, event: PluginHookEvents[Name]): void {
     for (const loaded of this.plugins.values()) {
       if (!loaded.hooks.includes(name)) {
@@ -727,6 +767,11 @@ export class PluginRuntime {
           if (message.type === "hook") {
             void send(child, { type: "hook.cancel", requestId }).catch(() => {});
           }
+          // The handler may still start after the daemon gave up on it.
+          if (message.type === "subscribe") {
+            const { subscriptionId } = message;
+            void send(child, { type: "unsubscribe", subscriptionId }).catch(() => {});
+          }
           reject(new Error(`Plugin RPC timed out: ${pluginId}.${message.type}`));
         },
         message.type === "tool.call" ? TOOL_CALL_TIMEOUT_MS : REQUEST_TIMEOUT_MS,
@@ -781,6 +826,8 @@ export class PluginRuntime {
         clientBundleHash: hashClientBundle(clientBundle),
         requirements: manifest.requirements,
         methods: new Set(),
+        subscriptionMethods: new Set(),
+        subscriptions: new Map(),
         hooks: [],
         providers: [],
         usageSources: [],
@@ -911,6 +958,8 @@ export class PluginRuntime {
       clientBundleHash: hashClientBundle(clientBundle),
       requirements,
       methods: new Set(ready.methods),
+      subscriptionMethods: new Set(ready.subscriptions ?? []),
+      subscriptions: new Map(),
       hooks: ready.hooks ?? [],
       providers: ready.providers ?? [],
       usageSources: ready.usageSources ?? [],
@@ -1009,6 +1058,16 @@ export class PluginRuntime {
       this.handleProviderMessage(loaded, message);
       return;
     }
+    if (message.type === "subscription.event") {
+      loaded.subscriptions.get(message.subscriptionId)?.event(message.event);
+      return;
+    }
+    if (message.type === "subscription.closed") {
+      const subscription = loaded.subscriptions.get(message.subscriptionId);
+      loaded.subscriptions.delete(message.subscriptionId);
+      subscription?.closed(message.error);
+      return;
+    }
     if (message.type !== "result" && message.type !== "error") return;
     const pending = loaded.pending.get(message.requestId);
     if (!pending) return;
@@ -1367,12 +1426,16 @@ export class PluginRuntime {
     this.appendLog(loaded.id, "stdout", "[paseo] Plugin stopped");
   }
 
+  /** Rejects in-flight requests and closes open subscriptions with the reason. */
   private rejectPending(loaded: LoadedPlugin, message: string): void {
     for (const invocation of loaded.pending.values()) {
       clearTimeout(invocation.timeout);
       invocation.reject(new Error(message));
     }
     loaded.pending.clear();
+    const subscriptions = [...loaded.subscriptions.values()];
+    loaded.subscriptions.clear();
+    for (const subscription of subscriptions) subscription.closed(message);
   }
 
   private appendLog(pluginId: string, stream: PluginLogEntry["stream"], message: string): void {

packages/server/src/server/plugins/subscription-handlers.test.ts

@@ -0,0 +1,55 @@
+import { defineSubscription } from "@getpaseo/plugin";
+import { subscribePluginRpc } from "@getpaseo/plugin/client/host";
+import type { PluginSubscriptionContext } from "@getpaseo/plugin/server";
+import type { PaseoApi } from "@getpaseo/client";
+import { expect, test } from "vitest";
+import { z } from "zod";
+import type { PluginProcessMessage } from "./plugin-process-protocol.js";
+import { PluginSubscriptionHandlers } from "./subscription-handlers.js";
+
+const lengths = defineSubscription({
+  name: "lengths",
+  input: z.object({ times: z.string().transform(Number) }),
+  event: z.string().transform((text) => text.length),
+});
+
+/** The plugin process's handlers wired to the client helper through the channel between them. */
+function connect(handler: (input: unknown, context: PluginSubscriptionContext) => unknown) {
+  const handlers = new PluginSubscriptionHandlers({
+    send: (message) => deliver(message),
+    sendAndWait: async (message) => deliver(message),
+  });
+  handlers.register(lengths, handler);
+  let observer: Parameters<Parameters<typeof subscribePluginRpc>[1]>[2] | null = null;
+  function deliver(message: PluginProcessMessage) {
+    if (message.type === "subscription.event") observer?.event(message.event);
+    if (message.type === "subscription.closed") observer?.closed();
+  }
+  return (method: string, input: unknown, next: NonNullable<typeof observer>) => {
+    observer = next;
+    handlers.start(
+      { type: "subscribe", requestId: "r", subscriptionId: "s", method, input },
+      {} as PaseoApi,
+    );
+    return Object.assign(() => handlers.release("s"), {
+      ready: Promise.resolve(),
+      release: async () => handlers.release("s"),
+    });
+  };
+}
+
+test("an event schema that transforms is applied once, on the receiving side", async () => {
+  const handled: unknown[] = [];
+  const received: number[] = [];
+  const closed: unknown[] = [];
+  const subscribe = connect(async (input, { push }) => {
+    handled.push(input);
+    await push("abc");
+  });
+  subscribePluginRpc(lengths, subscribe, { times: "2" }, (length) => received.push(length), {
+    onClosed: (result) => closed.push(result),
+  });
+  await expect.poll(() => received).toEqual([3]);
+  expect(handled).toEqual([{ times: 2 }]);
+  expect(closed).toEqual([{}]);
+});

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

@@ -0,0 +1,132 @@
+import type { PluginSubscriptionContract } from "@getpaseo/plugin";
+import type { PluginSubscriptionContext } from "@getpaseo/plugin/server";
+import type { PaseoApi } from "@getpaseo/client";
+import { assertPluginWireValue } from "@getpaseo/protocol/binary-frames/index";
+import type { PluginProcessMessage, PluginProcessRequest } from "./plugin-process-protocol.js";
+
+export type PluginSubscriptionHandler = (
+  input: unknown,
+  context: PluginSubscriptionContext,
+) => unknown;
+
+interface ActiveSubscription {
+  controller: AbortController;
+}
+
+interface SubscriptionChannel {
+  send(message: PluginProcessMessage): void;
+  sendAndWait(message: PluginProcessMessage): Promise<void>;
+}
+
+type SubscribeRequest = Extract<PluginProcessRequest, { type: "subscribe" }>;
+
+/**
+ * The plugin subprocess side of subscriptions: one handler call per subscriber, an abort signal
+ * that fires on release, and `push` that validates each event and resolves when it left the
+ * process.
+ */
+export class PluginSubscriptionHandlers {
+  private readonly registered = new Map<
+    string,
+    { contract: PluginSubscriptionContract; handler: PluginSubscriptionHandler }
+  >();
+  private readonly active = new Map<string, ActiveSubscription>();
+  private readonly channel: SubscriptionChannel;
+
+  constructor(channel: SubscriptionChannel) {
+    this.channel = channel;
+  }
+
+  has(name: string): boolean {
+    return this.registered.has(name);
+  }
+
+  register(contract: PluginSubscriptionContract, handler: PluginSubscriptionHandler): void {
+    this.registered.set(contract.name, { contract, handler });
+  }
+
+  catalog(): string[] {
+    return [...this.registered.keys()].sort();
+  }
+
+  start(message: SubscribeRequest, paseo: PaseoApi): void {
+    const { requestId, subscriptionId } = message;
+    const registered = this.registered.get(message.method);
+    if (!registered) {
+      this.channel.send({
+        type: "error",
+        requestId,
+        error: `Unknown subscription: ${message.method}`,
+      });
+      return;
+    }
+    const parsed = registered.contract.input.safeParse(message.input);
+    if (!parsed.success) {
+      this.channel.send({
+        type: "error",
+        requestId,
+        error: `Invalid input for ${message.method}: ${parsed.error.message}`,
+      });
+      return;
+    }
+    const controller = new AbortController();
+    const { signal } = controller;
+    let acknowledge = () => {};
+    const acknowledged = new Promise<void>((resolve) => {
+      acknowledge = resolve;
+    });
+    const push = async (event: unknown): Promise<void> => {
+      if (signal.aborted) return;
+      // The receiver parses into the schema's output; sending that output would parse it twice.
+      registered.contract.event.parse(event);
+      assertPluginWireValue(event, "event");
+      // Events pushed while the handler runs follow the acknowledgement.
+      await acknowledged;
+      if (signal.aborted) return;
+      await this.channel.sendAndWait({ type: "subscription.event", subscriptionId, event });
+    };
+    this.active.set(subscriptionId, { controller });
+    let result: unknown;
+    try {
+      result = registered.handler(parsed.data, { paseo, push, signal });
+    } catch (error) {
+      this.active.delete(subscriptionId);
+      controller.abort();
+      this.channel.send({ type: "error", requestId, error: describeError(error) });
+      return;
+    }
+    this.channel.send({ type: "result", requestId, output: null });
+    acknowledge();
+    if (result instanceof Promise)
+      result.then(
+        () => this.close(subscriptionId),
+        (error: unknown) => this.close(subscriptionId, describeError(error)),
+      );
+  }
+
+  /** The daemon released the subscription; it needs no closed message back. */
+  release(subscriptionId: string): void {
+    const active = this.active.get(subscriptionId);
+    if (!active) return;
+    this.active.delete(subscriptionId);
+    active.controller.abort();
+  }
+
+  releaseAll(): void {
+    for (const subscriptionId of this.active.keys()) this.release(subscriptionId);
+  }
+
+  private close(subscriptionId: string, error?: string): void {
+    if (!this.active.has(subscriptionId)) return;
+    this.release(subscriptionId);
+    this.channel.send({
+      type: "subscription.closed",
+      subscriptionId,
+      ...(error === undefined ? {} : { error }),
+    });
+  }
+}
+
+function describeError(error: unknown): string {
+  return error instanceof Error ? error.message : String(error);
+}

packages/server/src/server/session.test.ts

@@ -348,6 +348,7 @@ interface SessionForTestOptions {
   messages?: unknown[];
   targetedMessages?: Array<{ source: object; message: SessionOutboundMessage }>;
   binaryMessages?: Uint8Array[];
+  clientCapabilities?: SessionOptions["clientCapabilities"];
   pluginRuntime?: SessionOptions["pluginRuntime"];
   orchestrationSkills?: SessionOptions["orchestrationSkills"];
   workspaceLabelService?: WorkspaceLabelService;
@@ -402,6 +403,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session {
             messages.push(message),
         }),
     onBinaryMessage: createBinaryMessageHandler(options.binaryMessages),
+    clientCapabilities: options.clientCapabilities,
     logger,
     downloadTokenStore: options.downloadTokenStore ?? asDownloadTokenStore(),
     pushNotifications: options.pushNotifications ?? asPushNotifications(),
@@ -601,6 +603,64 @@ test("routes plugin requests and releases its owned catalog subscription on clea
   expect(releasePluginSubscription).toHaveBeenCalledOnce();
 });
 
+test("losing daemon.manage releases the session's plugin subscriptions", async () => {
+  const messages: SessionOutboundMessage[] = [];
+  const release = vi.fn();
+  const pluginRuntime = {
+    subscribe: () => () => {},
+    subscribePluginRpc: () => ({ ready: Promise.resolve(), attach: () => {}, release }),
+  } as unknown as NonNullable<SessionOptions["pluginRuntime"]>;
+  const session = createSessionForTest({ messages, pluginRuntime });
+
+  await session.handleMessage({
+    type: "plugin.rpc.subscribe.request",
+    requestId: "watch",
+    pluginId: "example",
+    method: "screen.capture",
+    input: {},
+  });
+  expect(messages.map((message) => message.type)).toContain("plugin.rpc.subscribe.response");
+  session.setPermissions(["daemon.read"]);
+
+  // Bytes have no outbound permission check, so the stream itself has to end.
+  await vi.waitFor(() => expect(release).toHaveBeenCalledOnce());
+});
+
+test("losing daemon.manage during a plugin RPC withholds its JSON and binary results", async () => {
+  const messages: SessionOutboundMessage[] = [];
+  const binaryMessages: Uint8Array[] = [];
+  const outputs = new Map<string, (output: unknown) => void>();
+  const pluginRuntime = {
+    subscribe: () => () => {},
+    invokePluginRpc: (_pluginId: string, method: string) =>
+      new Promise((resolve) => outputs.set(method, resolve)),
+  } as unknown as NonNullable<SessionOptions["pluginRuntime"]>;
+  const session = createSessionForTest({
+    messages,
+    binaryMessages,
+    pluginRuntime,
+    clientCapabilities: { [CLIENT_CAPS.pluginBinaryPayloads]: true },
+  });
+  const invoke = (method: string) =>
+    session.handleMessage({
+      type: "plugin.rpc.invoke.request",
+      requestId: method,
+      pluginId: "example",
+      method,
+      input: {},
+    });
+
+  const pending = [invoke("capture.json"), invoke("capture.bytes")];
+  await vi.waitFor(() => expect(outputs.size).toBe(2));
+  session.setPermissions(["daemon.read"]);
+  outputs.get("capture.json")?.({ width: 1 });
+  outputs.get("capture.bytes")?.(new Uint8Array([1, 2, 3]));
+  await Promise.all(pending);
+
+  expect(messages.filter((message) => message.type === "plugin.rpc.invoke.response")).toEqual([]);
+  expect(binaryMessages).toEqual([]);
+});
+
 test("leaves out the client bundles a catalog request already holds", async () => {
   const messages: SessionOutboundMessage[] = [];
   const pluginRuntime: NonNullable<SessionOptions["pluginRuntime"]> = {

packages/server/src/server/session.ts

@@ -50,6 +50,11 @@ import type {
 import { TerminalSessionController } from "../terminal/terminal-session-controller.js";
 import type { TerminalActivity } from "@getpaseo/protocol/terminal-activity";
 import type { BinaryFrame } from "@getpaseo/protocol/binary-frames/index";
+import {
+  encodePluginEventFrame,
+  encodePluginOutputFrame,
+  PluginFrameOpcode,
+} from "@getpaseo/protocol/binary-frames/index";
 import { CursorError } from "./pagination/cursor.js";
 import { SortablePager, type SortSpec } from "./pagination/sortable-pager.js";
 import { matchesAgentHistoryQuery } from "./agent-history-search.js";
@@ -521,6 +526,11 @@ export interface SessionOptions {
     subscribeSettings?(listener: (pluginId: string, settingsId: string) => void): () => void;
     catalog(): Array<{ id: string; clientBundle: string; clientBundleHash?: string }>;
     invokePluginRpc(pluginId: string, method: string, input: unknown): Promise<unknown>;
+    subscribePluginRpc(
+      pluginId: string,
+      method: string,
+      input: unknown,
+    ): import("./plugins/runtime-subscription.js").PluginRuntimeSubscription;
     listUsageReports(options?: {
       forceRefresh?: boolean;
       reportIds?: string[];
@@ -926,7 +936,7 @@ export class Session {
     this.workspaceFilesSession = new WorkspaceFilesSession({
       host: {
         emit: (msg, source) => this.emitForSource(msg, source),
-        emitBinary: (frame, source) => this.emitBinaryForFileTransfer(frame, source),
+        emitBinary: (frame, source) => this.emitBinaryReply(frame, source),
         hasBinaryChannel: () => this.onBinaryMessage !== null,
       },
       downloadTokenStore,
@@ -2221,13 +2231,15 @@ export class Session {
       this.peakInflightRequests = this.inflightRequests;
     }
     try {
-      this.sessionLogger.trace(
-        {
-          messageType: msg.type,
-          payloadBytes: JSON.stringify(msg).length,
-        },
-        "agent.session.inbound",
-      );
+      // Stringifying a binary plugin input costs its size; test logger stubs lack isLevelEnabled.
+      if (this.sessionLogger.isLevelEnabled?.("trace"))
+        this.sessionLogger.trace(
+          {
+            messageType: msg.type,
+            payloadBytes: JSON.stringify(msg).length,
+          },
+          "agent.session.inbound",
+        );
       if (!this.authorization.allowsInbound(msg)) {
         const requestId = sessionRequestId(msg);
         if (requestId) {
@@ -2292,6 +2304,17 @@ export class Session {
           this.sessionLogger.error({ err: error }, "Failed to release revoked browser hosting"),
         );
     }
+    // Binary plugin events skip the outbound permission check, so revocation ends the streams.
+    if (!this.authorization.allowsPermission("daemon.manage")) {
+      void this.delivery
+        .releaseFamily("plugin-rpc")
+        .catch((error) =>
+          this.sessionLogger.error(
+            { err: error },
+            "Failed to release revoked plugin subscriptions",
+          ),
+        );
+    }
   }
 
   public getPermissions(): DaemonPermission[] {
@@ -2454,6 +2477,7 @@ export class Session {
       this.dispatchOrchestrationSkillsMessage(msg) ??
       this.dispatchPluginDirectoryMessage(msg) ??
       this.dispatchPluginMessage(msg) ??
+      this.dispatchPluginRpcMessage(msg) ??
       this.dispatchHostServiceMessage(msg) ??
       this.dispatchMiscMessage(msg);
     if (promise) await promise;
@@ -2621,21 +2645,98 @@ export class Session {
       });
       return undefined;
     }
+    return undefined;
+  }
+
+  private dispatchPluginRpcMessage(msg: SessionInboundMessage): Promise<void> | undefined {
     if (msg.type === "plugin.rpc.invoke.request") {
       if (!this.pluginRuntime) throw new Error("Plugin service is unavailable");
       return this.pluginRuntime
         .invokePluginRpc(msg.pluginId, msg.method, msg.input)
-        .then((output) => {
-          this.emit({
-            type: "plugin.rpc.invoke.response",
-            payload: { requestId: msg.requestId, output },
-          });
-          return undefined;
-        });
+        .then((output) => this.replyPluginRpc(msg, output));
     }
+    if (msg.type === "plugin.rpc.subscribe.request") return this.subscribePluginRpc(msg);
     return undefined;
   }
 
+  /** A `Uint8Array` output is the binary form of `plugin.rpc.invoke.response`. */
+  private async replyPluginRpc(
+    request: Extract<SessionInboundMessage, { type: "plugin.rpc.invoke.request" }>,
+    output: unknown,
+  ): Promise<void> {
+    const response: SessionOutboundMessage = {
+      type: "plugin.rpc.invoke.response",
+      payload: { requestId: request.requestId, output },
+    };
+    if (!(output instanceof Uint8Array)) {
+      this.emit(response);
+      return;
+    }
+    // The frame is the binary form of the response, so it needs the response's permission.
+    if (!this.authorization.allowsOutbound(response)) return;
+    if (!this.supports(CLIENT_CAPS.pluginBinaryPayloads) || this.onBinaryMessage === null)
+      throw new Error(
+        `Update Paseo to receive the binary output of plugin RPC ${request.pluginId}.${request.method}`,
+      );
+    await this.emitBinaryReply(
+      encodePluginOutputFrame(request.requestId, output),
+      this.delivery.currentSource,
+    );
+  }
+
+  private async subscribePluginRpc(
+    request: Extract<SessionInboundMessage, { type: "plugin.rpc.subscribe.request" }>,
+  ): Promise<void> {
+    if (!this.pluginRuntime) throw new Error("Plugin service is unavailable");
+    const subscription = this.pluginRuntime.subscribePluginRpc(
+      request.pluginId,
+      request.method,
+      request.input,
+    );
+    const owner = this.delivery.begin("plugin-rpc", undefined, () => subscription.release());
+    try {
+      await subscription.ready;
+    } catch (error) {
+      await owner.release();
+      throw error;
+    }
+    if (owner.signal.aborted) return;
+    this.emit({
+      type: "plugin.rpc.subscribe.response",
+      payload: { requestId: request.requestId, subscriptionId: owner.responseId },
+    });
+    const close = (reason?: string) => {
+      owner.emit({
+        type: "plugin.rpc.closed",
+        payload: {
+          subscriptionId: owner.responseId,
+          ...(reason === undefined ? {} : { error: reason }),
+        },
+      });
+      void owner.release().catch(() => undefined);
+    };
+    subscription.attach({
+      event: (event) => {
+        const message: SessionOutboundMessage = {
+          type: "plugin.rpc.event",
+          payload: { subscriptionId: owner.responseId, event },
+        };
+        if (!(event instanceof Uint8Array)) {
+          owner.emit(message);
+          return;
+        }
+        if (!this.authorization.allowsOutbound(message)) return;
+        if (this.onBinaryMessage === null) {
+          subscription.release();
+          close("This connection cannot carry binary plugin events");
+          return;
+        }
+        owner.emitBinary(encodePluginEventFrame(owner.responseId, event));
+      },
+      closed: close,
+    });
+  }
+
   private dispatchPluginDirectoryMessage(msg: SessionInboundMessage): Promise<void> | undefined {
     if (msg.type === "plugin.source.install.request") {
       if (!this.pluginRuntime) throw new Error("Plugin service is unavailable");
@@ -3272,6 +3373,22 @@ export class Session {
       this.connectedHostSession?.binary(binaryFrame.frame, source);
       return;
     }
+    if (binaryFrame.kind === "plugin") {
+      const frame = binaryFrame.frame;
+      // The binary form of plugin.rpc.invoke.request; it authorizes and replies like the JSON form.
+      if (frame.opcode !== PluginFrameOpcode.Input) return;
+      await this.handleMessage(
+        {
+          type: "plugin.rpc.invoke.request",
+          requestId: frame.requestId,
+          pluginId: frame.pluginId,
+          method: frame.method,
+          input: frame.payload,
+        },
+        source,
+      );
+      return;
+    }
     if (!this.authorization.allowsPermission("workspace.write")) {
       return;
     }
@@ -8318,7 +8435,7 @@ export class Session {
             },
             activeTurnBehavior: msg.activeTurnBehavior ?? "steer",
             sessionMode: msg.sessionMode,
-            clearPendingPermissions: true,
+            clearPendingPermissions: false,
           });
           deliveryStatus = "queued";
           queued();
@@ -8690,9 +8807,9 @@ export class Session {
     }
   }
 
-  private async emitBinaryForFileTransfer(frame: Uint8Array, source?: object): Promise<void> {
+  private async emitBinaryReply(frame: Uint8Array, source?: object): Promise<void> {
     if (source && this.onBinaryMessageToSource) {
-      this.delivery.authorizeFileReply(frame, source);
+      this.delivery.authorizeBinaryReply(frame, source);
       await this.onBinaryMessageToSource(source, frame);
       return;
     }

packages/server/src/server/session/owned-subscriptions/index.ts

@@ -56,6 +56,11 @@ export interface OwnedSubscription {
   release(): Promise<void>;
 }
 
+const BINARY_REPLY_REQUESTS: ReadonlySet<string> = new Set([
+  "file_explorer_request",
+  "plugin.rpc.invoke.request",
+]);
+
 // One registration per context kind; per-session registrations outlive closed sessions in Node.
 const requests = new AsyncLocalStorage<RequestOwner>();
 const projectionSource = new AsyncLocalStorage<ProjectionOwner>();
@@ -394,12 +399,13 @@ export class SessionDelivery {
     if (owner) await this.releaseOwner(owner);
   }
 
-  authorizeFileReply(frame: Uint8Array, socket: object): void {
+  /** Lets a binary frame answer the current request, for requests whose reply can be bytes. */
+  authorizeBinaryReply(frame: Uint8Array, socket: object): void {
     const owner = this.currentRequest;
     if (
       owner?.active &&
       owner.source.socket === socket &&
-      owner.request.type === "file_explorer_request"
+      BINARY_REPLY_REQUESTS.has(owner.request.type)
     )
       this.proofs.set(frame, owner);
   }

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

@@ -2066,6 +2066,8 @@ export class VoiceAssistantWebSocketServer {
         pluginThemes: true,
         pluginSettings: true,
         pluginTimelineItems: true,
+        // COMPAT(pluginSubscriptions): added in v0.15.0, remove gate after 2027-05-02.
+        pluginSubscriptions: true,
         // COMPAT(skillManagement): added in v0.4.0, remove gate after 2027-08-16.
         skillManagement: true,
         // COMPAT(idleRuntimeCleanupSetting): added in v0.13.0, remove gate after 2027-04-26.

packages/server/src/server/workspace-git-service.observation.integration.test.ts

@@ -1,3 +1,4 @@
+import { flushGitCommandTrace } from "../utils/git-command-trace.js";
 import { createWorkspaceRuntimeStore } from "./workspace-runtime-store/index.js";
 import { mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs";
 import { execFileSync } from "node:child_process";
@@ -730,3 +731,103 @@ test("Git event identity retains a symlink's literal target before canonicalizin
     rmSync(root, { recursive: true, force: true });
   }
 });
+
+test.skipIf(process.platform !== "linux")(
+  "missing folders stop degraded Git polling and explicit refreshes, then recover",
+  async () => {
+    const artifacts = path.resolve(".artifacts");
+    mkdirSync(artifacts, { recursive: true });
+    const root = mkdtempSync(path.join(artifacts, "workspace-git-missing-"));
+    const cwd = path.join(root, "repo");
+    const initializeRepo = (branch: string) => {
+      mkdirSync(cwd);
+      execFileSync("git", ["init", "-b", branch], { cwd, stdio: "pipe" });
+      writeFileSync(path.join(cwd, "README.md"), "fixture\n");
+      execFileSync("git", ["add", "."], { cwd, stdio: "pipe" });
+      execFileSync(
+        "git",
+        [
+          "-c",
+          "user.name=Test",
+          "-c",
+          "user.email=test@example.invalid",
+          "-c",
+          "commit.gpgsign=false",
+          "commit",
+          "-m",
+          "fixture",
+        ],
+        { cwd, stdio: "pipe" },
+      );
+    };
+    initializeRepo("main");
+    // Exceed the real Linux observer's directory limit to enter degraded polling.
+    // No watcher errors or Git responses are injected.
+    for (let index = 0; index < 5100; index++) {
+      mkdirSync(path.join(cwd, `generated-${index}`));
+    }
+    const trace = path.join(root, "git.jsonl");
+    writeFileSync(trace, "");
+    const previousTrace = process.env.PASEO_GIT_TRACE_FILE;
+    process.env.PASEO_GIT_TRACE_FILE = trace;
+    let degraded = false;
+    let suspended = false;
+    const logger = pino(
+      { level: "warn" },
+      {
+        write(line) {
+          const message = JSON.parse(line).msg;
+          if (message === "Failed to start working tree watcher; using degraded polling")
+            degraded = true;
+          if (
+            message ===
+            "Workspace folder no longer exists; suspended git observation until it is back"
+          )
+            suspended = true;
+        },
+      },
+    );
+    const service = new WorkspaceGitServiceImpl({ logger, paseoHome: path.join(root, "home") });
+    const subscription = service.registerWorkspace({ cwd }, () => {});
+    const commandCount = async () => {
+      await flushGitCommandTrace();
+      return readFileSync(trace, "utf8").trim().split("\n").filter(Boolean).length;
+    };
+    try {
+      await expect
+        .poll(() => degraded && service.getMetrics().workspaceRefreshInFlightCount === 0, {
+          timeout: 15_000,
+        })
+        .toBe(true);
+      rmSync(cwd, { recursive: true, force: true });
+      await expect.poll(() => suspended, { timeout: 15_000 }).toBe(true);
+      await expect.poll(() => service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
+      const commandsWhileMissing = await commandCount();
+      await service.getSnapshot(cwd, {
+        force: true,
+        includeForge: false,
+        reason: "refresh-button",
+      });
+      await service.refresh(cwd);
+      expect(await commandCount()).toBe(commandsWhileMissing);
+      await new Promise((resolve) => setTimeout(resolve, 16_000));
+      expect(await commandCount()).toBe(commandsWhileMissing);
+
+      initializeRepo("restored");
+      const snapshot = await service.getSnapshot(cwd, {
+        force: true,
+        includeForge: false,
+        reason: "refresh-button",
+      });
+      expect(snapshot.git).toMatchObject({ isGit: true, currentBranch: "restored" });
+      await expect.poll(() => service.getMetrics().repositoryWorkspaceLinkCount).toBe(1);
+    } finally {
+      subscription.unsubscribe();
+      await service.dispose();
+      if (previousTrace === undefined) delete process.env.PASEO_GIT_TRACE_FILE;
+      else process.env.PASEO_GIT_TRACE_FILE = previousTrace;
+      rmSync(root, { recursive: true, force: true });
+    }
+  },
+  60_000,
+);

packages/server/src/server/workspace-git-service.observation.test.ts

@@ -244,6 +244,7 @@ function createService(
         filterEvents: (events) => events,
         verify: vi.fn(async () => {}),
       })),
+      workspaceFolderExists: vi.fn(async () => true),
       ...overrides,
     } as never,
   });

packages/server/src/server/workspace-git-service.test.ts

@@ -1,5 +1,7 @@
 import { createWorkspaceRuntimeStore } from "./workspace-runtime-store/index.js";
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
+import { existsSync } from "node:fs";
+import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
 import os from "node:os";
 import path, { join } from "node:path";
 import type pino from "pino";
@@ -10,11 +12,14 @@ import type {
   PullRequestStatusResult,
 } from "../utils/checkout-git.js";
 import {
+  BACKGROUND_GIT_FETCH_INTERVAL_MS,
+  WORKSPACE_GIT_OBSERVATION_REENSURE_INTERVAL_MS,
   WORKSPACE_GIT_OBSERVATION_SETUP_CONCURRENCY,
   WORKSPACE_GIT_REFRESH_CONCURRENCY,
   WORKSPACE_GIT_WATCHER_SUBSCRIBE_TIMEOUT_MS,
   WorkspaceGitServiceImpl,
   type WorkspaceGitRuntimeSnapshot,
+  workspaceFolderExists,
 } from "./workspace-git-service.js";
 
 const REPO_CWD = path.resolve("/tmp/repo");
@@ -23,6 +28,7 @@ function createLogger() {
   const logger = {
     child: () => logger,
     debug: vi.fn(),
+    info: vi.fn(),
     warn: vi.fn(),
   };
   return logger;
@@ -251,7 +257,7 @@ function createGitHubServiceStub(): ForgeService {
 interface CreateServiceTestOptions {
   subscribe?: ReturnType<typeof vi.fn>;
   getCheckoutStatus?: ReturnType<typeof vi.fn>;
-  getCheckoutSnapshotFacts?: ReturnType<typeof vi.fn>;
+  getCheckoutSnapshotFacts?: (cwd: string) => Promise<CheckoutSnapshotFacts>;
   getCheckoutShortstat?: ReturnType<typeof vi.fn>;
   getCheckoutWorktreeState?: ReturnType<typeof vi.fn>;
   getPullRequestStatus?: ReturnType<typeof vi.fn>;
@@ -261,6 +267,7 @@ interface CreateServiceTestOptions {
   runGitFetch?: ReturnType<typeof vi.fn>;
   runGitCommand?: ReturnType<typeof vi.fn>;
   getWorkspaceGitSelfHealPhaseMs?: (cwd: string) => number;
+  workspaceFolderExists?: (cwd: string) => Promise<boolean>;
   now?: () => Date;
 }
 
@@ -294,6 +301,7 @@ function buildDefaultTestServiceDeps() {
       signal: null,
     })),
     getWorkspaceGitSelfHealPhaseMs: vi.fn(() => 30_000),
+    workspaceFolderExists: vi.fn(async () => true),
     createWatcherLivenessCanary: vi.fn(() => ({
       path: "",
       filterEvents: (events) => events,
@@ -324,6 +332,145 @@ function createService(options?: CreateServiceTestOptions) {
   });
 }
 
+type WatchCallback = (error: Error | null, events: Array<{ path: string; type: "update" }>) => void;
+
+interface FolderBackedGitOptions {
+  facts?: (cwd: string) => CheckoutSnapshotFacts;
+  unwatchableDirectories?: string[];
+  heldWatch?: { directory: string; release: Promise<void> };
+}
+
+function createFolderBackedGit(options?: FolderBackedGitOptions) {
+  const watchCallbacks = new Map<string, WatchCallback>();
+  const subscribe = vi.fn(async (directory: string, callback: WatchCallback) => {
+    if (!existsSync(directory) || options?.unwatchableDirectories?.includes(directory)) {
+      throw new Error(`Cannot watch ${directory}`);
+    }
+    if (options?.heldWatch?.directory === directory) {
+      await options.heldWatch.release;
+    }
+    watchCallbacks.set(directory, callback);
+    return createAsyncSubscription();
+  });
+  const getCheckoutSnapshotFacts = vi.fn(
+    async (cwd: string): Promise<CheckoutSnapshotFacts> =>
+      existsSync(cwd) ? (options?.facts ?? createCheckoutSnapshotFacts)(cwd) : { isGit: false },
+  );
+  const getCheckoutStatus = vi.fn(async (cwd: string) =>
+    existsSync(cwd) ? createCheckoutStatus(cwd) : ({ isGit: false } as const),
+  );
+  const getCheckoutShortstat = vi.fn(async (_cwd: string) => ({ additions: 1, deletions: 0 }));
+  const getCheckoutWorktreeState = vi.fn(async (_cwd: string) => ({
+    isDirty: false,
+    diffStat: { additions: 1, deletions: 0 },
+  }));
+  const runGitCommand = vi.fn(async (args: string[], runOptions: { cwd: string }) => ({
+    stdout: args[0] === "rev-parse" ? `${runOptions.cwd}\n` : "",
+    stderr: "",
+    truncated: false,
+    exitCode: 0,
+    signal: null,
+  }));
+  const runGitFetch = vi.fn(async (_cwd: string) => ({ changes: [], error: null }));
+  const callsIn = (fake: ReturnType<typeof vi.fn>, cwd: string) =>
+    fake.mock.calls.filter(([calledCwd]) => calledCwd === cwd).length;
+
+  return {
+    deps: {
+      subscribe,
+      getCheckoutSnapshotFacts,
+      getCheckoutStatus,
+      getCheckoutShortstat,
+      getCheckoutWorktreeState,
+      runGitCommand,
+      runGitFetch,
+      workspaceFolderExists: async (cwd: string) => existsSync(cwd),
+      now: () => new Date(),
+    },
+    watchCallbackFor(directory: string): WatchCallback {
+      const callback = watchCallbacks.get(directory);
+      if (!callback) {
+        throw new Error(`No watcher was started for ${directory}`);
+      }
+      return callback;
+    },
+    watchAttemptsFor: (directory: string) => callsIn(subscribe, directory),
+    pollsOf: (cwd: string) => callsIn(getCheckoutShortstat, cwd),
+    fetchesIn: (cwd: string) => callsIn(runGitFetch, cwd),
+    gitCommandsIn(cwd: string): number {
+      const cwdCommands = [
+        getCheckoutSnapshotFacts,
+        getCheckoutStatus,
+        getCheckoutShortstat,
+        getCheckoutWorktreeState,
+        runGitFetch,
+      ].reduce((count, fake) => count + callsIn(fake, cwd), 0);
+      const commandsRunInCwd = runGitCommand.mock.calls.filter(
+        ([, runOptions]) => runOptions.cwd === cwd,
+      ).length;
+      return cwdCommands + commandsRunInCwd;
+    },
+  };
+}
+
+function linkedWorktreeFacts(gitCommonDir: string) {
+  return (cwd: string): CheckoutSnapshotFacts => ({
+    ...createCheckoutSnapshotFacts(cwd),
+    absoluteGitDir: join(gitCommonDir, "worktrees", path.basename(cwd)),
+    gitCommonDir,
+  });
+}
+
+const SETTLE_ATTEMPTS = 1_000;
+
+async function settle(assertion: () => void): Promise<void> {
+  let lastFailure: unknown = null;
+  for (let attempt = 0; attempt < SETTLE_ATTEMPTS; attempt += 1) {
+    try {
+      assertion();
+      return;
+    } catch (failure) {
+      lastFailure = failure;
+    }
+    await vi.advanceTimersByTimeAsync(0);
+  }
+  throw lastFailure;
+}
+
+async function waitForRefreshesToSettle(service: WorkspaceGitServiceImpl): Promise<void> {
+  await settle(() => {
+    expect(service.getMetrics().workspaceRefreshInFlightCount).toBe(0);
+  });
+}
+
+async function waitForObservationToSettle(
+  service: WorkspaceGitServiceImpl,
+  repositoryWorkspaceLinkCount: number,
+): Promise<void> {
+  await settle(() => {
+    const metrics = service.getMetrics();
+    expect(metrics.repositoryWorkspaceLinkCount).toBe(repositoryWorkspaceLinkCount);
+    expect(metrics.workspaceObservationSetupInFlightCount).toBe(0);
+    expect(metrics.workspaceRefreshInFlightCount).toBe(0);
+  });
+}
+
+async function advanceThroughDegradedPolls(service: WorkspaceGitServiceImpl): Promise<void> {
+  for (const delayMs of [5_000, 10_000, 20_000, 40_000, 60_000, 60_000]) {
+    await vi.advanceTimersByTimeAsync(delayMs);
+    await waitForRefreshesToSettle(service);
+  }
+}
+
+async function advanceThroughWorkingTreeRecoveryAttempts(
+  service: WorkspaceGitServiceImpl,
+): Promise<void> {
+  for (const delayMs of [30_000, 60_000, 120_000]) {
+    await vi.advanceTimersByTimeAsync(delayMs);
+    await waitForRefreshesToSettle(service);
+  }
+}
+
 describe("WorkspaceGitServiceImpl", () => {
   beforeEach(() => {
     vi.useFakeTimers();
@@ -1553,3 +1700,460 @@ describe("WorkspaceGitServiceImpl", () => {
     service.dispose();
   });
 });
+
+describe("WorkspaceGitServiceImpl with workspace folders that no longer exist", () => {
+  const temporaryFolders: string[] = [];
+
+  async function createTemporaryFolder(): Promise<string> {
+    const folder = await mkdtemp(join(os.tmpdir(), "paseo-workspace-git-"));
+    temporaryFolders.push(folder);
+    return folder;
+  }
+
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+
+  afterEach(async () => {
+    vi.useRealTimers();
+    await Promise.all(
+      temporaryFolders.splice(0).map((folder) => rm(folder, { recursive: true, force: true })),
+    );
+  });
+
+  test("polling stops for a workspace whose folder no longer exists", async () => {
+    const workspace = await createTemporaryFolder();
+    const git = createFolderBackedGit();
+    const service = createService(git.deps);
+    const subscription = service.registerWorkspace({ cwd: workspace }, vi.fn());
+    await waitForObservationToSettle(service, 1);
+
+    await rm(workspace, { recursive: true, force: true });
+    const gitCommandsAtDeletion = git.gitCommandsIn(workspace);
+    await advanceThroughDegradedPolls(service);
+
+    expect(git.gitCommandsIn(workspace)).toBe(gitCommandsAtDeletion);
+    expect(service.getMetrics()).toMatchObject({
+      repositoryTargetCount: 0,
+      workingTreeWatchTargetCount: 0,
+    });
+    expect(vi.getTimerCount()).toBe(1);
+
+    subscription.unsubscribe();
+    await service.dispose();
+  });
+
+  test("a deleted worktree leaves its shared repository polling and fetch to the surviving worktree", async () => {
+    const gitCommonDir = join(await createTemporaryFolder(), ".git");
+    const deletedWorktree = await createTemporaryFolder();
+    const survivingWorktree = await createTemporaryFolder();
+    const git = createFolderBackedGit({ facts: linkedWorktreeFacts(gitCommonDir) });
+    let survivingWorktreeEdits = 0;
+    git.deps.getCheckoutShortstat.mockImplementation(async (cwd: string) => ({
+      additions: cwd === survivingWorktree ? ++survivingWorktreeEdits : 1,
+      deletions: 0,
+    }));
+    const service = createService(git.deps);
+    const deletedSubscription = service.registerWorkspace({ cwd: deletedWorktree }, vi.fn());
+    await waitForObservationToSettle(service, 1);
+    const survivingSubscription = service.registerWorkspace({ cwd: survivingWorktree }, vi.fn());
+    await waitForObservationToSettle(service, 2);
+
+    await rm(deletedWorktree, { recursive: true, force: true });
+    await vi.advanceTimersByTimeAsync(60_000);
+    await waitForRefreshesToSettle(service);
+    const deletedWorktreeCommands = git.gitCommandsIn(deletedWorktree);
+
+    for (let tick = 0; tick < 3; tick += 1) {
+      const survivingWorktreePolls = git.pollsOf(survivingWorktree);
+      await vi.advanceTimersByTimeAsync(5_000);
+      await settle(() => {
+        expect(git.pollsOf(survivingWorktree)).toBe(survivingWorktreePolls + 1);
+      });
+    }
+    expect(git.gitCommandsIn(deletedWorktree)).toBe(deletedWorktreeCommands);
+
+    const fetchesBefore = git.deps.runGitFetch.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(2 * BACKGROUND_GIT_FETCH_INTERVAL_MS);
+    expect(git.deps.runGitFetch.mock.calls.slice(fetchesBefore).map(([cwd]) => cwd)).toEqual([
+      survivingWorktree,
+      survivingWorktree,
+    ]);
+    expect(git.gitCommandsIn(deletedWorktree)).toBe(deletedWorktreeCommands);
+
+    deletedSubscription.unsubscribe();
+    survivingSubscription.unsubscribe();
+    await service.dispose();
+  });
+
+  test("deleting every worktree of a repository stops its polling and background fetch", async () => {
+    const gitCommonDir = join(await createTemporaryFolder(), ".git");
+    const firstWorktree = await createTemporaryFolder();
+    const secondWorktree = await createTemporaryFolder();
+    const git = createFolderBackedGit({ facts: linkedWorktreeFacts(gitCommonDir) });
+    const service = createService(git.deps);
+    const firstSubscription = service.registerWorkspace({ cwd: firstWorktree }, vi.fn());
+    await waitForObservationToSettle(service, 1);
+    const secondSubscription = service.registerWorkspace({ cwd: secondWorktree }, vi.fn());
+    await waitForObservationToSettle(service, 2);
+
+    await rm(firstWorktree, { recursive: true, force: true });
+    await rm(secondWorktree, { recursive: true, force: true });
+    const commandsAtDeletion = [
+      git.gitCommandsIn(firstWorktree),
+      git.gitCommandsIn(secondWorktree),
+    ];
+    await advanceThroughDegradedPolls(service);
+
+    expect([git.gitCommandsIn(firstWorktree), git.gitCommandsIn(secondWorktree)]).toEqual(
+      commandsAtDeletion,
+    );
+    expect(service.getMetrics()).toMatchObject({
+      repositoryTargetCount: 0,
+      workingTreeWatchTargetCount: 0,
+    });
+    expect(vi.getTimerCount()).toBe(2);
+
+    firstSubscription.unsubscribe();
+    secondSubscription.unsubscribe();
+    await service.dispose();
+  });
+
+  test("working tree fallback polling and watcher recovery stop for a workspace whose folder no longer exists", async () => {
+    const workspace = await createTemporaryFolder();
+    await mkdir(join(workspace, ".git"));
+    const git = createFolderBackedGit();
+    const service = createService(git.deps);
+    const subscription = service.registerWorkspace({ cwd: workspace }, vi.fn());
+    await waitForObservationToSettle(service, 1);
+
+    await rm(workspace, { recursive: true, force: true });
+    git.watchCallbackFor(workspace)(new Error("watched directory was deleted"), []);
+    await vi.advanceTimersByTimeAsync(1_000);
+    await waitForRefreshesToSettle(service);
+    const commandsAfterWatcherError = git.gitCommandsIn(workspace);
+    const watchAttemptsAfterWatcherError = git.watchAttemptsFor(workspace);
+    await advanceThroughDegradedPolls(service);
+
+    expect(git.gitCommandsIn(workspace)).toBe(commandsAfterWatcherError);
+    expect(git.watchAttemptsFor(workspace)).toBe(watchAttemptsAfterWatcherError);
+    expect(vi.getTimerCount()).toBe(1);
+
+    subscription.unsubscribe();
+    await service.dispose();
+  });
+
+  test("a workspace whose folder comes back resumes polling on its next subscription", async () => {
+    const workspace = await createTemporaryFolder();
+    const git = createFolderBackedGit();
+    const service = createService(git.deps);
+    const firstSubscription = service.registerWorkspace({ cwd: workspace }, vi.fn());
+    await waitForObservationToSettle(service, 1);
+    await rm(workspace, { recursive: true, force: true });
+    const commandsWhileMissing = git.gitCommandsIn(workspace);
+    await advanceThroughDegradedPolls(service);
+    expect(git.gitCommandsIn(workspace)).toBe(commandsWhileMissing);
+
+    await mkdir(workspace);
+    const secondSubscription = service.registerWorkspace({ cwd: workspace }, vi.fn());
+    await waitForObservationToSettle(service, 1);
+    const pollsAfterResume = git.pollsOf(workspace);
+    await vi.advanceTimersByTimeAsync(5_000);
+    await waitForRefreshesToSettle(service);
+
+    expect(git.pollsOf(workspace)).toBe(pollsAfterResume + 1);
+
+    firstSubscription.unsubscribe();
+    secondSubscription.unsubscribe();
+    await service.dispose();
+  });
+
+  test("a suspended workspace costs no git while its folder is missing, even when a refresh is requested", async () => {
+    const workspace = await createTemporaryFolder();
+    const git = createFolderBackedGit();
+    const service = createService(git.deps);
+    const subscription = service.registerWorkspace({ cwd: workspace }, vi.fn());
+    await waitForObservationToSettle(service, 1);
+    await rm(workspace, { recursive: true, force: true });
+    await advanceThroughDegradedPolls(service);
+    const commandsWhileMissing = git.gitCommandsIn(workspace);
+
+    for (let probe = 0; probe < 10; probe += 1) {
+      await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_OBSERVATION_REENSURE_INTERVAL_MS);
+      await waitForRefreshesToSettle(service);
+    }
+    service.scheduleRefreshForCwd(workspace);
+    service.onWorkspaceStateMayHaveChanged(workspace);
+    await vi.advanceTimersByTimeAsync(5_000);
+    await waitForRefreshesToSettle(service);
+
+    expect(git.gitCommandsIn(workspace)).toBe(commandsWhileMissing);
+    expect(vi.getTimerCount()).toBe(1);
+
+    subscription.unsubscribe();
+    await service.dispose();
+  });
+
+  test("a suspended worktree resumes polling and its repository fetch once its folder is back, without a new subscription", async () => {
+    const gitCommonDir = join(await createTemporaryFolder(), ".git");
+    const worktree = await createTemporaryFolder();
+    const git = createFolderBackedGit({ facts: linkedWorktreeFacts(gitCommonDir) });
+    const service = createService(git.deps);
+    const subscription = service.registerWorkspace({ cwd: worktree }, vi.fn());
+    await waitForObservationToSettle(service, 1);
+    await rm(worktree, { recursive: true, force: true });
+    await advanceThroughDegradedPolls(service);
+    await waitForObservationToSettle(service, 0);
+
+    await mkdir(worktree);
+    const fetchesBeforeRestore = git.fetchesIn(worktree);
+    await vi.advanceTimersByTimeAsync(WORKSPACE_GIT_OBSERVATION_REENSURE_INTERVAL_MS);
+    await waitForObservationToSettle(service, 1);
+    expect(git.fetchesIn(worktree)).toBe(fetchesBeforeRestore + 1);
+
+    const pollsAfterRestore = git.pollsOf(worktree);
+    await vi.advanceTimersByTimeAsync(5_000);
+    await waitForRefreshesToSettle(service);
+    expect(git.pollsOf(worktree)).toBe(pollsAfterRestore + 1);
+
+    subscription.unsubscribe();
+    await service.dispose();
+  });
+
+  test("a forced snapshot resumes a suspended workspace whose folder is back", async () => {
+    const workspace = await createTemporaryFolder();
+    const git = createFolderBackedGit();
+    const service = createService(git.deps);
+    const subscription = service.registerWorkspace({ cwd: workspace }, vi.fn());
+    await waitForObservationToSettle(service, 1);
+    await rm(workspace, { recursive: true, force: true });
+    await advanceThroughDegradedPolls(service);
+    await waitForObservationToSettle(service, 0);
+
+    await mkdir(workspace);
+    const snapshot = await service.getSnapshot(workspace, {
+      force: true,
+      reason: "refresh-button",
+    });
+    await waitForObservationToSettle(service, 1);
+
+    expect(snapshot.git.isGit).toBe(true);
+
+    subscription.unsubscribe();
+    await service.dispose();
+  });
+
+  test("a resumed workspace rebuilds observation from fresh facts, not the non-git facts cached while its folder was gone", async () => {
+    const workspace = await createTemporaryFolder();
+    const git = createFolderBackedGit();
+    const service = createService(git.deps);
+    const firstSubscription = service.registerWorkspace({ cwd: workspace }, vi.fn());
+    await waitForObservationToSettle(service, 1);
+    await rm(workspace, { recursive: true, force: true });
+    service.onWorkspaceStateMayHaveChanged(workspace);
+    await vi.advanceTimersByTimeAsync(1_000);
+    await waitForRefreshesToSettle(service);
+    expect((await service.getSnapshot(workspace)).git.isGit).toBe(false);
+    await advanceThroughDegradedPolls(service);
+    await waitForObservationToSettle(service, 0);
+
+    await mkdir(workspace);
+    const secondSubscription = service.registerWorkspace({ cwd: workspace }, vi.fn());
+    await waitForObservationToSettle(service, 1);
+
+    expect((await service.getSnapshot(workspace)).git.isGit).toBe(true);
+
+    firstSubscription.unsubscribe();
+    secondSubscription.unsubscribe();
+    await service.dispose();
+  });
+
+  test("resuming waits for an observation setup that was still running when the folder vanished", async () => {
+    const workspace = await createTemporaryFolder();
+    const gitDir = join(workspace, ".git");
+    await mkdir(gitDir);
+    const metadataWatchRelease = createDeferred<void>();
+    const git = createFolderBackedGit({
+      unwatchableDirectories: [workspace],
+      heldWatch: { directory: gitDir, release: metadataWatchRelease.promise },
+    });
+    const service = createService(git.deps);
+    const firstSubscription = service.registerWorkspace({ cwd: workspace }, vi.fn());
+    await settle(() => {
+      expect(git.watchAttemptsFor(gitDir)).toBe(1);
+    });
+
+    await rm(workspace, { recursive: true, force: true });
+    await vi.advanceTimersByTimeAsync(5_000);
+    await settle(() => {
+      expect(service.getMetrics()).toMatchObject({
+        repositoryTargetCount: 0,
+        workingTreeWatchTargetCount: 0,
+      });
+    });
+    await mkdir(workspace);
+    const secondSubscription = service.registerWorkspace({ cwd: workspace }, vi.fn());
+    metadataWatchRelease.resolve();
+    await waitForObservationToSettle(service, 1);
+
+    expect(service.getMetrics().workingTreeWatchTargetCount).toBe(1);
+
+    firstSubscription.unsubscribe();
+    secondSubscription.unsubscribe();
+    await service.dispose();
+  });
+
+  test("an open working tree watch neither notifies nor recovers while its folder is missing, and picks up once it is back", async () => {
+    const workspace = await createTemporaryFolder();
+    await mkdir(join(workspace, ".git"));
+    const git = createFolderBackedGit();
+    const service = createService(git.deps);
+    const workspaceSubscription = service.registerWorkspace({ cwd: workspace }, vi.fn());
+    await waitForObservationToSettle(service, 1);
+    const diffListener = vi.fn();
+    const diffWatch = await service.requestWorkingTreeWatch(workspace, diffListener);
+
+    await rm(workspace, { recursive: true, force: true });
+    git.watchCallbackFor(workspace)(new Error("watched directory was deleted"), []);
+    await vi.advanceTimersByTimeAsync(1_000);
+    await waitForRefreshesToSettle(service);
+    const commandsAfterWatcherError = git.gitCommandsIn(workspace);
+    const notificationsAfterWatcherError = diffListener.mock.calls.length;
+    const watchAttemptsAfterWatcherError = git.watchAttemptsFor(workspace);
+    await advanceThroughWorkingTreeRecoveryAttempts(service);
+
+    expect(git.gitCommandsIn(workspace)).toBe(commandsAfterWatcherError);
+    expect(diffListener.mock.calls.length).toBe(notificationsAfterWatcherError);
+    expect(git.watchAttemptsFor(workspace)).toBe(watchAttemptsAfterWatcherError);
+    expect(service.getMetrics().workingTreeWatchTargetCount).toBe(1);
+
+    await mkdir(join(workspace, ".git"), { recursive: true });
+    await vi.advanceTimersByTimeAsync(60_000);
+    await waitForRefreshesToSettle(service);
+    expect(diffListener.mock.calls.length).toBe(notificationsAfterWatcherError + 1);
+
+    await vi.advanceTimersByTimeAsync(240_000);
+    await waitForRefreshesToSettle(service);
+    expect(git.watchAttemptsFor(workspace)).toBe(watchAttemptsAfterWatcherError + 1);
+    const notificationsAfterResubscribe = diffListener.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(120_000);
+    await waitForRefreshesToSettle(service);
+    expect(diffListener.mock.calls.length).toBe(notificationsAfterResubscribe);
+
+    diffWatch.unsubscribe();
+    workspaceSubscription.unsubscribe();
+    await service.dispose();
+  });
+
+  test("a refresh crossing suspension preserves the cached snapshot and drops queued work", async () => {
+    const workspace = await createTemporaryFolder();
+    const git = createFolderBackedGit();
+    const started = createDeferred<void>();
+    const release = createDeferred<void>();
+    let holdNextRead = false;
+    let factsReads = 0;
+    const service = createService({
+      ...git.deps,
+      getCheckoutSnapshotFacts: async (cwd: string) => {
+        factsReads++;
+        if (holdNextRead) {
+          holdNextRead = false;
+          started.resolve();
+          await release.promise;
+        }
+        return git.deps.getCheckoutSnapshotFacts(cwd);
+      },
+    });
+    const updates: WorkspaceGitRuntimeSnapshot[] = [];
+    const subscription = service.registerWorkspace({ cwd: workspace }, (snapshot) =>
+      updates.push(snapshot),
+    );
+    await waitForObservationToSettle(service, 1);
+    const cached = await service.getSnapshot(workspace);
+    holdNextRead = true;
+    const active = service.getSnapshot(workspace, {
+      force: true,
+      includeForge: false,
+      reason: "overlapping-refresh",
+    });
+    await started.promise;
+    const queued = service.getSnapshot(workspace, {
+      force: true,
+      includeForge: true,
+      reason: "queued-refresh",
+    });
+    await rm(workspace, { recursive: true, force: true });
+    await vi.advanceTimersByTimeAsync(5_000);
+    await settle(() => expect(service.getMetrics().repositoryWorkspaceLinkCount).toBe(0));
+    const readsAtSuspension = factsReads;
+    const updatesAtSuspension = updates.length;
+    release.resolve();
+    expect(await active).toEqual(cached);
+    expect(await queued).toEqual(cached);
+    expect(await service.getSnapshot(workspace)).toEqual(cached);
+    expect(updates).toHaveLength(updatesAtSuspension);
+    expect(factsReads).toBe(readsAtSuspension);
+    subscription.unsubscribe();
+    await service.dispose();
+  });
+
+  test("restoration settles a suspended refresh before reloading facts", async () => {
+    const workspace = await createTemporaryFolder();
+    const git = createFolderBackedGit();
+    const staleFacts = createDeferred<CheckoutSnapshotFacts>();
+    const started = createDeferred<void>();
+    let holdNextRead = false;
+    const service = createService({
+      ...git.deps,
+      getCheckoutSnapshotFacts: async (cwd: string) => {
+        if (holdNextRead) {
+          holdNextRead = false;
+          started.resolve();
+          return staleFacts.promise;
+        }
+        return git.deps.getCheckoutSnapshotFacts(cwd);
+      },
+    });
+    const updates: WorkspaceGitRuntimeSnapshot[] = [];
+    const subscription = service.registerWorkspace({ cwd: workspace }, (snapshot) =>
+      updates.push(snapshot),
+    );
+    await waitForObservationToSettle(service, 1);
+    const cached = await service.getSnapshot(workspace);
+    holdNextRead = true;
+    const active = service.getSnapshot(workspace, {
+      force: true,
+      includeForge: false,
+      reason: "overlapping-refresh",
+    });
+    await started.promise;
+    await rm(workspace, { recursive: true, force: true });
+    await vi.advanceTimersByTimeAsync(5_000);
+    await settle(() => expect(service.getMetrics().repositoryWorkspaceLinkCount).toBe(0));
+    const updatesAtSuspension = updates.length;
+    await mkdir(workspace);
+    const restored = service.getSnapshot(workspace, {
+      force: true,
+      includeForge: false,
+      reason: "folder-returned",
+    });
+    await vi.advanceTimersByTimeAsync(0);
+    staleFacts.resolve({ isGit: false });
+    expect(await active).toEqual(cached);
+    expect((await restored).git.isGit).toBe(true);
+    expect(updates.slice(updatesAtSuspension).every((snapshot) => snapshot.git.isGit)).toBe(true);
+    await waitForObservationToSettle(service, 1);
+    subscription.unsubscribe();
+    await service.dispose();
+  });
+
+  test("workspaceFolderExists reports a missing folder and a path under a file as missing", async () => {
+    const folder = await createTemporaryFolder();
+    const file = join(folder, "file.txt");
+    await writeFile(file, "");
+
+    expect(await workspaceFolderExists(folder)).toBe(true);
+    expect(await workspaceFolderExists(join(folder, "missing"))).toBe(false);
+    expect(await workspaceFolderExists(join(file, "nested"))).toBe(false);
+  });
+});

packages/server/src/server/workspace-git-service.ts

@@ -1,5 +1,5 @@
 import { createHash } from "node:crypto";
-import { readFile } from "node:fs/promises";
+import { readFile, stat } from "node:fs/promises";
 import { basename, dirname, join, resolve } from "node:path";
 import { LRUCache } from "lru-cache";
 import { CheckoutDiffCache } from "./checkout-diff-cache.js";
@@ -74,7 +74,7 @@ import { checkoutLiteFromGitSnapshot } from "./workspace-registry-model.js";
 import { createWatcherLivenessCanary } from "./watcher-liveness-canary.js";
 
 const WORKSPACE_GIT_WATCH_DEBOUNCE_MS = 1_000;
-const BACKGROUND_GIT_FETCH_INTERVAL_MS = 180_000;
+export const BACKGROUND_GIT_FETCH_INTERVAL_MS = 180_000;
 const FETCH_METADATA_ECHO_TTL_MS = 5_000;
 export const WORKSPACE_GIT_OBSERVATION_REENSURE_INTERVAL_MS = 60_000;
 const FORGE_PR_STATUS_POLL_FAST_INTERVAL_MS = 20_000;
@@ -384,6 +384,7 @@ interface WorkspaceGitServiceDependencies {
   runGitCommand: typeof runGitCommand;
   getWorkspaceGitSelfHealPhaseMs: typeof getWorkspaceGitObservationReensurePhaseMs;
   createWatcherLivenessCanary: typeof createWatcherLivenessCanary;
+  workspaceFolderExists: (cwd: string) => Promise<boolean>;
   now: () => Date;
 }
 
@@ -435,6 +436,8 @@ interface WorkspaceGitTarget {
   repoGitRoot: string | null;
   observationSetupPromise: Promise<void> | null;
   observationSetupComplete: boolean;
+  folderMissing: boolean;
+  folderRestorePromise: Promise<void> | null;
   closed: boolean;
 }
 
@@ -571,10 +574,21 @@ function buildDefaultWorkspaceGitServiceDeps(
     runGitCommand,
     getWorkspaceGitSelfHealPhaseMs: getWorkspaceGitObservationReensurePhaseMs,
     createWatcherLivenessCanary,
+    workspaceFolderExists,
     now: () => new Date(),
   };
 }
 
+export async function workspaceFolderExists(cwd: string): Promise<boolean> {
+  try {
+    await stat(cwd);
+    return true;
+  } catch (error) {
+    const code = (error as NodeJS.ErrnoException).code;
+    return code !== "ENOENT" && code !== "ENOTDIR";
+  }
+}
+
 function resolveWorkspaceGitServiceDeps(
   subscribe: SubscribeToFileChanges,
   deps: Partial<WorkspaceGitServiceDependencies> | undefined,
@@ -660,6 +674,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     const cwd = resolve(params.cwd);
     const target = this.ensureWorkspaceTarget(cwd);
     target.listeners.add(listener);
+    if (target.folderMissing) {
+      void this.resumeWorkspaceIfFolderRestored(target);
+    }
     if (target.listeners.size === 1) {
       this.startWorkspaceSubscriptionTimers(target);
     }
@@ -751,6 +768,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     if (!request.force && target.latestSnapshot) {
       return target.latestSnapshot;
     }
+    if (target.folderMissing) {
+      await this.resumeWorkspaceIfFolderRestored(target);
+    }
 
     return this.requestWorkspaceSnapshot(target, request);
   }
@@ -972,6 +992,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     this.assertNotDisposed();
     cwd = resolve(cwd);
     const target = this.ensureWorkspaceTarget(cwd);
+    if (target.folderMissing) {
+      await this.resumeWorkspaceIfFolderRestored(target);
+    }
     await this.refreshWorkspaceTarget(target, {
       force: false,
       refreshStructure: true,
@@ -1005,9 +1028,14 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     this.assertNotDisposed();
     cwd = resolve(cwd);
     const target = this.workspaceTargets.get(cwd);
-    if (target) {
-      this.scheduleWorkspaceRefresh(target, { queueIfBusy: false });
+    if (!target) {
+      return;
     }
+    if (target.folderMissing) {
+      void this.resumeWorkspaceIfFolderRestored(target);
+      return;
+    }
+    this.scheduleWorkspaceRefresh(target, { queueIfBusy: false });
   }
 
   onWorkspaceStateMayHaveChanged(cwd: string): void {
@@ -1018,6 +1046,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       return;
     }
     this.invalidateForge(normalizedCwd);
+    if (target.folderMissing) {
+      void this.resumeWorkspaceIfFolderRestored(target);
+      return;
+    }
     this.scheduleWorkspaceRefresh(target, {
       force: true,
       includeForge: true,
@@ -1227,6 +1259,8 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       repoGitRoot: null,
       observationSetupPromise: null,
       observationSetupComplete: false,
+      folderMissing: false,
+      folderRestorePromise: null,
       closed: false,
     };
 
@@ -1358,6 +1392,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   private isActiveObservedWorkspaceTarget(target: WorkspaceGitTarget): boolean {
     return (
       !target.closed &&
+      !target.folderMissing &&
       target.listeners.size > 0 &&
       this.workspaceTargets.get(target.cwd) === target
     );
@@ -1612,11 +1647,110 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     workspaceTarget: WorkspaceGitTarget,
     request: WorkspaceGitRefreshRequest,
   ): Promise<boolean> {
+    if (await this.suspendWorkspaceIfFolderMissing(workspaceTarget)) {
+      return false;
+    }
     const fingerprintBefore = workspaceTarget.latestFingerprint;
     await this.refreshWorkspaceTarget(workspaceTarget, request);
     return workspaceTarget.latestFingerprint !== fingerprintBefore;
   }
 
+  private async suspendWorkspaceIfFolderMissing(target: WorkspaceGitTarget): Promise<boolean> {
+    if (target.folderMissing) {
+      return true;
+    }
+    if (await this.deps.workspaceFolderExists(target.cwd)) {
+      return false;
+    }
+    this.suspendWorkspaceWithMissingFolder(target);
+    return true;
+  }
+
+  private suspendWorkspaceWithMissingFolder(target: WorkspaceGitTarget): void {
+    if (target.folderMissing || target.closed || this.workspaceTargets.get(target.cwd) !== target) {
+      return;
+    }
+    target.folderMissing = true;
+    if (target.refreshState.status === "in-flight") {
+      target.refreshState.queued = null;
+    }
+    this.detachWorkspaceFromRepoTarget(target);
+    target.repoGitRoot = null;
+    this.stopWorkspaceObservation(target);
+    target.observationSetupComplete = false;
+    this.scheduleMissingFolderProbe(target);
+    this.logger.warn(
+      { cwd: target.cwd },
+      "Workspace folder no longer exists; suspended git observation until it is back",
+    );
+  }
+
+  private isSuspendedWorkspaceTarget(target: WorkspaceGitTarget): boolean {
+    return (
+      target.folderMissing && !target.closed && this.workspaceTargets.get(target.cwd) === target
+    );
+  }
+
+  private scheduleMissingFolderProbe(target: WorkspaceGitTarget): void {
+    if (
+      !this.isSuspendedWorkspaceTarget(target) ||
+      target.observationReensureTimer ||
+      target.folderRestorePromise
+    ) {
+      return;
+    }
+    target.observationReensureTimer = setTimeout(() => {
+      target.observationReensureTimer = null;
+      void this.resumeWorkspaceIfFolderRestored(target);
+    }, WORKSPACE_GIT_OBSERVATION_REENSURE_INTERVAL_MS);
+  }
+
+  private resumeWorkspaceIfFolderRestored(target: WorkspaceGitTarget): Promise<void> {
+    if (!target.folderMissing) {
+      return Promise.resolve();
+    }
+    target.folderRestorePromise ??= this.resumeWorkspaceWithRestoredFolder(target).finally(() => {
+      target.folderRestorePromise = null;
+      this.scheduleMissingFolderProbe(target);
+    });
+    return target.folderRestorePromise;
+  }
+
+  private async resumeWorkspaceWithRestoredFolder(target: WorkspaceGitTarget): Promise<void> {
+    if (!(await this.deps.workspaceFolderExists(target.cwd))) {
+      return;
+    }
+    await target.observationSetupPromise;
+    if (target.refreshState.status === "in-flight") {
+      await target.refreshState.promise.catch(() => undefined);
+    }
+    if (!this.isSuspendedWorkspaceTarget(target)) {
+      return;
+    }
+    target.latestFacts = null;
+    target.folderMissing = false;
+    await this.refreshWorkspaceTarget(target, {
+      force: false,
+      refreshStructure: true,
+      refreshWorktree: true,
+      includeForge: false,
+      reason: "workspace-folder-restored",
+      queueIfBusy: true,
+      movedRemoteRefs: new Set(),
+    });
+    if (target.closed || this.workspaceTargets.get(target.cwd) !== target) {
+      return;
+    }
+    target.observationSetupComplete = false;
+    if (target.observationReensureTimer) {
+      clearTimeout(target.observationReensureTimer);
+      target.observationReensureTimer = null;
+    }
+    this.logger.info({ cwd: target.cwd }, "Workspace folder is back; resumed git observation");
+    this.startWorkspaceSubscriptionTimers(target);
+    this.scheduleWorkspaceObservationSetup(target);
+  }
+
   private startWorkingTreeWatchFallback(
     target: WorkingTreeWatchTarget,
     reason: WorkingTreeWatchFallbackReason,
@@ -1654,7 +1788,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
           return changed;
         }),
       );
-      this.notifyWorkingTreeConsumers(target);
+      if (await this.deps.workspaceFolderExists(target.watchPath)) {
+        this.notifyWorkingTreeConsumers(target);
+      }
       if (!target.closed && (target.subscription === null || target.repoRoot === null)) {
         target.fallbackPollQuietTickCount = changes.some(Boolean)
           ? 0
@@ -1731,6 +1867,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     if (target.closed || target.subscription) {
       return;
     }
+    if (!(await this.deps.workspaceFolderExists(target.watchPath))) {
+      this.scheduleWorkingTreeWatchRecovery(target);
+      return;
+    }
     await this.refreshWorkingTreeIgnoredDirectories(target);
     if (target.closed || target.subscription) {
       return;
@@ -2786,7 +2926,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   }
 
   private updateForgePrStatusPollForTarget(target: WorkspaceGitTarget): void {
-    if (target.listeners.size === 0) {
+    if (!this.isActiveObservedWorkspaceTarget(target)) {
       this.stopForgePrStatusPollForTarget(target);
       return;
     }
@@ -3049,6 +3189,9 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     target: WorkspaceGitTarget,
     request: WorkspaceGitRefreshRequest,
   ): Promise<WorkspaceGitRuntimeSnapshot> {
+    if (target.folderMissing) {
+      return Promise.resolve(target.latestSnapshot ?? buildNotGitSnapshot(target.cwd));
+    }
     if (target.refreshState.status === "in-flight") {
       const active = target.refreshState.request;
       const addsWork =
@@ -3191,11 +3334,19 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       try {
         const runRefreshGitCommand = createRunGitCommand(`workspace-refresh:${request.reason}`);
         const admittedSnapshot = await this.workspaceRefreshLimit(() => {
-          if (target.closed || this.workspaceTargets.get(target.cwd) !== target) {
+          if (
+            target.closed ||
+            target.folderMissing ||
+            this.workspaceTargets.get(target.cwd) !== target
+          ) {
             return null;
           }
           return this.refreshSnapshot(target, request, runRefreshGitCommand);
         });
+        if (target.folderMissing) {
+          failure = null;
+          break;
+        }
         if (!admittedSnapshot) {
           break;
         }
@@ -3209,6 +3360,10 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
         failure = { error };
       }
 
+      if (target.folderMissing) {
+        failure = null;
+        break;
+      }
       if (this.disposed || target.closed || this.workspaceTargets.get(target.cwd) !== target) {
         break;
       }
@@ -3504,7 +3659,7 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
     snapshot: WorkspaceGitRuntimeSnapshot,
     options?: { forceEmit?: boolean },
   ): void {
-    if (target.closed || this.workspaceTargets.get(target.cwd) !== target) {
+    if (target.closed || target.folderMissing || this.workspaceTargets.get(target.cwd) !== target) {
       return;
     }
     target.latestSnapshot = snapshot;
@@ -3660,21 +3815,25 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
   }
 
   private removeWorkspaceTarget(target: WorkspaceGitTarget): void {
-    if (target.repoGitRoot) {
-      const repoTarget = this.repoTargets.get(target.repoGitRoot);
-      repoTarget?.workspaceKeys.delete(target.cwd);
-      if (repoTarget && repoTarget.workspaceKeys.size === 0) {
-        this.closeRepoTarget(repoTarget);
-        this.repoTargets.delete(target.repoGitRoot);
-      } else if (repoTarget?.cwd === target.cwd) {
-        repoTarget.cwd = repoTarget.workspaceKeys.values().next().value ?? repoTarget.cwd;
-      }
-    }
-
+    this.detachWorkspaceFromRepoTarget(target);
     this.closeWorkspaceTarget(target);
     this.workspaceTargets.delete(target.cwd);
   }
 
+  private detachWorkspaceFromRepoTarget(target: WorkspaceGitTarget): void {
+    if (!target.repoGitRoot) {
+      return;
+    }
+    const repoTarget = this.repoTargets.get(target.repoGitRoot);
+    repoTarget?.workspaceKeys.delete(target.cwd);
+    if (repoTarget && repoTarget.workspaceKeys.size === 0) {
+      this.closeRepoTarget(repoTarget);
+      this.repoTargets.delete(target.repoGitRoot);
+    } else if (repoTarget?.cwd === target.cwd) {
+      repoTarget.cwd = repoTarget.workspaceKeys.values().next().value ?? repoTarget.cwd;
+    }
+  }
+
   private removeWorkingTreeWatchListener(cwd: string, listener: () => void): void {
     const target = this.workingTreeWatchTargets.get(cwd);
     if (!target) {
@@ -3719,6 +3878,11 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
 
   private closeWorkspaceTarget(target: WorkspaceGitTarget): void {
     target.closed = true;
+    this.stopWorkspaceObservation(target);
+    target.listeners.clear();
+  }
+
+  private stopWorkspaceObservation(target: WorkspaceGitTarget): void {
     if (target.workingTreeWatchTarget) {
       this.removeWorkspaceWorkingTreeLink(target.workingTreeWatchTarget, target.cwd);
       target.workingTreeWatchTarget = null;
@@ -3727,12 +3891,12 @@ export class WorkspaceGitServiceImpl implements WorkspaceGitService {
       clearTimeout(target.debounceTimer);
       target.debounceTimer = null;
     }
+    target.pendingDebounceRequest = null;
     if (target.observationReensureTimer) {
       clearTimeout(target.observationReensureTimer);
       target.observationReensureTimer = null;
     }
     this.stopForgePrStatusPollForTarget(target);
-    target.listeners.clear();
   }
 
   private closeWorkingTreeWatchTarget(target: WorkingTreeWatchTarget): void {

packages/server/src/terminal/terminal-worker-runtime.test.ts

@@ -1,5 +1,6 @@
 import { afterEach, describe, expect, it } from "vitest";
 import {
+  chmodSync,
   existsSync,
   mkdirSync,
   mkdtempSync,
@@ -54,6 +55,7 @@ function makeSources(bundleSource = "export {};\n"): {
   mkdirSync(native, { recursive: true });
   writeFileSync(path.join(native, "pty.node"), "native");
   writeFileSync(path.join(native, "spawn-helper"), "#!/bin/sh\n", { mode: 0o755 });
+  chmodSync(path.join(native, "spawn-helper"), 0o755);
   // Another platform's build is not copied.
   mkdirSync(path.join(nodePtyDirectory, "prebuilds", "other-platform"));
   return { bundleDirectory, nodePtyDirectory };

plugin-examples/diff-watch/OVERVIEW.md

@@ -0,0 +1,7 @@
+# Live changes
+
+See which files differ from `HEAD` in a workspace panel that updates as you edit. The daemon watches the workspace and pushes the changed-file list to the panel; the panel never polls.
+
+Open **Open live changes** in the Command Center, or pick **Live changes** from the workspace tab menu. Each open panel runs its own watcher, which stops when you close the panel.
+
+The watcher reads the working tree recursively, so a checkout with very large ignored directories makes it slower to start.

plugin-examples/diff-watch/README.md

@@ -0,0 +1,14 @@
+# Diff watch plugin example
+
+A workspace panel subscribes to a daemon-side watcher and lists the files that differ from `HEAD`,
+updating as files change.
+
+```bash
+paseo plugin add /absolute/path/to/paseo/plugin-examples/diff-watch
+```
+
+- `shared/diff.ts` defines the `diff.watch` subscription contract.
+- `index.server.ts` handles it: an async handler that loops until the subscriber leaves, pushing
+  the file list whenever it changes. A git failure rejects the handler and closes the subscription
+  with that error.
+- `client/diff-panel.tsx` calls `useSubscription` and renders the latest list.

plugin-examples/diff-watch/client/diff-panel.tsx

@@ -0,0 +1,65 @@
+import {
+  useSubscription,
+  useWorkspace,
+  type PluginWorkspacePanelProps,
+} from "@getpaseo/plugin/client";
+import { useMemo, useState } from "react";
+import { ScrollView, StyleSheet, Text, View } from "react-native";
+import { diffWatch, type ChangedFile } from "../shared/diff";
+
+const BASE = "HEAD";
+
+export function DiffPanel({ workspaceId, theme, layout }: PluginWorkspacePanelProps) {
+  const cwd = useWorkspace(workspaceId, (workspace) => workspace.directory);
+  const styles = useStyles(theme, layout.compact);
+  if (!cwd) return <Text style={styles.muted}>Workspace unavailable</Text>;
+  return <ChangedFiles cwd={cwd} styles={styles} />;
+}
+
+function ChangedFiles({ cwd, styles }: { cwd: string; styles: ReturnType<typeof useStyles> }) {
+  const [files, setFiles] = useState<ChangedFile[] | null>(null);
+  const subscription = useSubscription(diffWatch, { cwd, from: BASE }, (event) => {
+    setFiles(event.files);
+  });
+  return (
+    <ScrollView style={styles.screen} contentContainerStyle={styles.content}>
+      <Text style={styles.title}>Changes against {BASE}</Text>
+      <Text style={styles.muted}>{describe(subscription, files)}</Text>
+      {files?.map((file) => (
+        <View key={file.path} style={styles.row}>
+          <Text style={styles.status}>{file.status}</Text>
+          <Text style={styles.path} numberOfLines={1}>
+            {file.path}
+          </Text>
+        </View>
+      ))}
+    </ScrollView>
+  );
+}
+
+function describe(
+  subscription: ReturnType<typeof useSubscription>,
+  files: ChangedFile[] | null,
+): string {
+  if (subscription.status === "error") return subscription.error.message;
+  if (subscription.status === "closed") return "Stopped watching";
+  if (!files) return "Reading changes…";
+  if (files.length === 0) return "No changes";
+  return files.length === 1 ? "1 file changed" : `${files.length} files changed`;
+}
+
+function useStyles(theme: PluginWorkspacePanelProps["theme"], compact: boolean) {
+  return useMemo(
+    () =>
+      StyleSheet.create({
+        screen: { flex: 1, backgroundColor: theme.colors.surface0 },
+        content: { padding: compact ? 16 : 24, gap: 8 },
+        title: { color: theme.colors.foreground, fontSize: 18, fontWeight: "600" },
+        muted: { color: theme.colors.foregroundMuted },
+        row: { flexDirection: "row", gap: 12, alignItems: "center" },
+        status: { color: theme.colors.foregroundMuted, width: 72 },
+        path: { color: theme.colors.foreground, flex: 1 },
+      }),
+    [theme, compact],
+  );
+}

plugin-examples/diff-watch/index.client.tsx

@@ -0,0 +1,22 @@
+import type { PluginClientContext } from "@getpaseo/plugin/client";
+import { DiffPanel } from "./client/diff-panel";
+
+export default function contribute(client: PluginClientContext) {
+  client.addWorkspacePanel({
+    id: "changes",
+    title: "Live changes",
+    icon: "GitCompare",
+    context: "workspace",
+    Component: DiffPanel,
+  });
+  client.addCommandCenterItem({
+    id: "open-changes",
+    title: "Open live changes",
+    icon: "GitCompare",
+    context: "workspace",
+    onSelect({ openPanel }) {
+      openPanel("changes");
+    },
+  });
+  return () => {};
+}

plugin-examples/diff-watch/index.server.ts

@@ -0,0 +1,19 @@
+import type { PluginServerContext } from "@getpaseo/plugin/server";
+import { readChangedFiles, watchDirectory } from "./server/diff";
+import { diffWatch } from "./shared/diff";
+
+export default function contribute(server: PluginServerContext) {
+  // The handler runs for as long as the panel is subscribed; a git failure closes it with the error.
+  server.handle(diffWatch, async (input, { push, signal }) => {
+    const directory = watchDirectory(input.cwd, signal);
+    let previous = "";
+    while (await directory.settled()) {
+      const files = await readChangedFiles(input);
+      const current = JSON.stringify(files);
+      if (current === previous) continue;
+      previous = current;
+      await push({ type: "changes", files });
+    }
+  });
+  return () => {};
+}

plugin-examples/diff-watch/paseo-plugin.json

@@ -0,0 +1,6 @@
+{
+  "id": "diff-watch",
+  "requirements": {
+    "paseo": ">=0.15.0-beta.2"
+  }
+}

plugin-examples/diff-watch/server/diff.ts

@@ -0,0 +1,79 @@
+import { watch } from "node:fs";
+import type { SubscriptionInput } from "@getpaseo/plugin";
+import { execCommand } from "@getpaseo/plugin/server";
+import type { ChangedFile, diffWatch } from "../shared/diff";
+
+const SETTLE_MS = 150;
+
+const STATUSES: Record<string, ChangedFile["status"]> = {
+  A: "added",
+  M: "modified",
+  D: "deleted",
+  R: "renamed",
+};
+
+/** Lists changed files the way `git diff --name-status` does, plus untracked files. */
+export async function readChangedFiles({
+  cwd,
+  from,
+  to,
+}: SubscriptionInput<typeof diffWatch>): Promise<ChangedFile[]> {
+  const range = to ? [from, to] : [from];
+  const { stdout } = await execCommand("git", ["diff", "--name-status", "-z", ...range], { cwd });
+  const files = parseNameStatus(stdout);
+  if (!to) {
+    const untracked = await execCommand(
+      "git",
+      ["ls-files", "--others", "--exclude-standard", "-z"],
+      {
+        cwd,
+      },
+    );
+    for (const path of untracked.stdout.split("\0").filter(Boolean))
+      files.push({ path, status: "added" });
+  }
+  return files.sort((left, right) => left.path.localeCompare(right.path));
+}
+
+function parseNameStatus(output: string): ChangedFile[] {
+  const fields = output.split("\0").filter(Boolean);
+  const files: ChangedFile[] = [];
+  for (let index = 0; index < fields.length; index += 1) {
+    const code = fields[index][0];
+    // Renames carry the old path before the new one.
+    if (code === "R") index += 1;
+    const path = fields[index + 1];
+    index += 1;
+    if (path) files.push({ path, status: STATUSES[code] ?? "modified" });
+  }
+  return files;
+}
+
+/** Resolves each time the directory settles after a change, and with false once `signal` aborts. */
+export function watchDirectory(cwd: string, signal: AbortSignal) {
+  let changed = true;
+  let wake: (() => void) | null = null;
+  const watcher = watch(cwd, { recursive: true }, () => {
+    changed = true;
+    wake?.();
+  });
+  signal.addEventListener(
+    "abort",
+    () => {
+      watcher.close();
+      wake?.();
+    },
+    { once: true },
+  );
+  return {
+    async settled(): Promise<boolean> {
+      // The watcher and the abort both wake a wait.
+      if (!changed && !signal.aborted) await new Promise<void>((resolve) => (wake = resolve));
+      wake = null;
+      if (signal.aborted) return false;
+      await new Promise((resolve) => setTimeout(resolve, SETTLE_MS));
+      changed = false;
+      return !signal.aborted;
+    },
+  };
+}

plugin-examples/diff-watch/shared/diff.ts

@@ -0,0 +1,16 @@
+import { defineSubscription } from "@getpaseo/plugin";
+import { z } from "zod";
+
+export const ChangedFileSchema = z.object({
+  path: z.string(),
+  status: z.enum(["added", "modified", "deleted", "renamed"]),
+});
+
+export type ChangedFile = z.infer<typeof ChangedFileSchema>;
+
+/** Files that differ between `from` and `to`; without `to`, the working tree. */
+export const diffWatch = defineSubscription({
+  name: "diff.watch",
+  input: z.object({ cwd: z.string(), from: z.string(), to: z.string().optional() }),
+  event: z.object({ type: z.literal("changes"), files: z.array(ChangedFileSchema) }),
+});

plugin-examples/media-stream/OVERVIEW.md

@@ -0,0 +1,7 @@
+# Media stream
+
+Watch a moving test pattern and an audio level meter, streamed live from your machine into a workspace panel. Video and audio arrive together on one connection, and the counters show how many frames of each have played.
+
+Open **Open media stream** in the Command Center, or pick **Media stream** from the workspace tab menu. Closing the panel stops the stream.
+
+Press **Send this frame back** to upload the picture on screen to your machine. The panel shows the size and checksum your machine received. A 16 by 9 picture is 432 bytes.

plugin-examples/media-stream/README.md

@@ -0,0 +1,17 @@
+# Media stream plugin example
+
+One subscription carries two binary streams. Each pushed `Uint8Array` is a binary frame; its first
+byte says whether it is video, a keyframe, or audio.
+
+```bash
+paseo plugin add /absolute/path/to/paseo/plugin-examples/media-stream
+```
+
+- `shared/media-frames.ts` owns the byte layout: `encodeMediaFrame`, `decodeMediaFrame`, and the
+  `media.stream` contract. `decodeMediaFrame` returns a view, so the panel never copies a payload.
+- `server/test-pattern.ts` is a synthetic source; replace it with a real capture.
+- `index.server.ts` pushes each source callback as a frame and stops the source when the
+  subscription's `signal` aborts.
+- `client/media-panel.tsx` decodes each frame and routes it by opcode.
+- `shared/frame-check.ts` is an RPC whose input is a `Uint8Array`: the panel uploads the picture
+  on screen and `server/frame-check.ts` reports its size and checksum.

plugin-examples/media-stream/client/media-panel.tsx

@@ -0,0 +1,196 @@
+import { useRpc, useSubscription, type PluginWorkspacePanelProps } from "@getpaseo/plugin/client";
+import type { RpcOutput } from "@getpaseo/plugin";
+import { type UseMutationResult, useMutation } from "@tanstack/react-query";
+import { useCallback, useMemo, useState } from "react";
+import { Pressable, StyleSheet, Text, View } from "react-native";
+import { checkFrame } from "../shared/frame-check";
+import {
+  decodeMediaFrame,
+  MediaOpcode,
+  mediaStream,
+  VIDEO,
+  type MediaFrame,
+} from "../shared/media-frames";
+
+interface Playback {
+  picture: Uint8Array | null;
+  level: number;
+  videoFrames: number;
+  keyframes: number;
+  audioChunks: number;
+  bytes: number;
+}
+
+const EMPTY: Playback = {
+  picture: null,
+  level: 0,
+  videoFrames: 0,
+  keyframes: 0,
+  audioChunks: 0,
+  bytes: 0,
+};
+
+export function MediaPanel({ theme, layout }: PluginWorkspacePanelProps) {
+  const styles = useStyles(theme, layout.compact);
+  const [playback, setPlayback] = useState(EMPTY);
+  const subscription = useSubscription(mediaStream, { source: "test-pattern" }, (bytes) => {
+    const frame = decodeMediaFrame(bytes);
+    if (frame) setPlayback((current) => play(current, frame, bytes.byteLength));
+  });
+  return (
+    <View style={styles.screen}>
+      <Text style={styles.title}>Test pattern</Text>
+      <Text style={styles.muted}>
+        {subscription.status === "error" ? subscription.error.message : subscription.status}
+      </Text>
+      <Picture pixels={playback.picture} cellSize={layout.compact ? 14 : 20} />
+      <LevelMeter level={playback.level} styles={styles} />
+      <Text style={styles.muted}>
+        {playback.videoFrames} video frames ({playback.keyframes} keyframes), {playback.audioChunks}{" "}
+        audio chunks, {(playback.bytes / 1024).toFixed(0)} KiB
+      </Text>
+      <SendFrameBack picture={playback.picture} styles={styles} />
+    </View>
+  );
+}
+
+/** Uploads the picture on screen and shows what the daemon received. */
+function SendFrameBack({
+  picture,
+  styles,
+}: {
+  picture: Uint8Array | null;
+  styles: ReturnType<typeof useStyles>;
+}) {
+  const sendFrame = useRpc(checkFrame);
+  const check = useMutation({ mutationFn: sendFrame });
+  const { mutate } = check;
+  const send = useCallback(() => {
+    if (picture) mutate(picture);
+  }, [mutate, picture]);
+  return (
+    <View style={styles.upload}>
+      <Pressable
+        accessibilityRole="button"
+        disabled={!picture || check.isPending}
+        onPress={send}
+        style={styles.button}
+      >
+        <Text style={styles.buttonText}>Send this frame back</Text>
+      </Pressable>
+      <Text style={styles.muted}>{describeCheck(check)}</Text>
+    </View>
+  );
+}
+
+function describeCheck(
+  check: UseMutationResult<RpcOutput<typeof checkFrame>, Error, Uint8Array>,
+): string {
+  if (check.status === "error") return check.error.message;
+  if (check.status === "success")
+    return `The daemon received ${check.data.bytes} bytes, sha256 ${check.data.sha256.slice(0, 12)}`;
+  return check.status === "pending" ? "Sending…" : "";
+}
+
+function play(current: Playback, frame: MediaFrame, bytes: number): Playback {
+  if (frame.opcode === MediaOpcode.Audio)
+    return {
+      ...current,
+      level: audioLevel(frame.payload),
+      audioChunks: current.audioChunks + 1,
+      bytes: current.bytes + bytes,
+    };
+  return {
+    ...current,
+    picture: frame.payload,
+    videoFrames: current.videoFrames + 1,
+    keyframes: current.keyframes + (frame.opcode === MediaOpcode.Keyframe ? 1 : 0),
+    bytes: current.bytes + bytes,
+  };
+}
+
+function audioLevel(samples: Uint8Array): number {
+  const view = new DataView(samples.buffer, samples.byteOffset, samples.byteLength);
+  let sum = 0;
+  const count = samples.byteLength / 2;
+  for (let index = 0; index < count; index += 1) sum += view.getInt16(index * 2, true) ** 2;
+  return Math.min(1, Math.sqrt(sum / Math.max(1, count)) / 16384);
+}
+
+function LevelMeter({ level, styles }: { level: number; styles: ReturnType<typeof useStyles> }) {
+  const fill = useMemo(
+    () => [styles.meter, { width: `${Math.round(level * 100)}%` as const }],
+    [level, styles],
+  );
+  return (
+    <View style={styles.meterTrack}>
+      <View style={fill} />
+    </View>
+  );
+}
+
+function Picture({ pixels, cellSize }: { pixels: Uint8Array | null; cellSize: number }) {
+  const rows = useMemo(() => pictureRows(pixels, cellSize), [pixels, cellSize]);
+  return (
+    <View accessibilityLabel="Test pattern video">
+      {rows.map((row) => (
+        <View key={row.key} style={pictureStyles.row}>
+          {row.cells.map((cell) => (
+            <View key={cell.key} style={cell.style} />
+          ))}
+        </View>
+      ))}
+    </View>
+  );
+}
+
+/** One style per pixel, row by row. */
+function pictureRows(pixels: Uint8Array | null, cellSize: number) {
+  const rows = [];
+  for (let y = 0; y < VIDEO.height; y += 1) {
+    const cells = [];
+    for (let x = 0; x < VIDEO.width; x += 1) {
+      const offset = (y * VIDEO.width + x) * 3;
+      const backgroundColor = pixels
+        ? `rgb(${pixels[offset]}, ${pixels[offset + 1]}, ${pixels[offset + 2]})`
+        : "transparent";
+      cells.push({ key: `${x}`, style: { width: cellSize, height: cellSize, backgroundColor } });
+    }
+    rows.push({ key: `${y}`, cells });
+  }
+  return rows;
+}
+
+const pictureStyles = StyleSheet.create({ row: { flexDirection: "row" } });
+
+function useStyles(theme: PluginWorkspacePanelProps["theme"], compact: boolean) {
+  return useMemo(
+    () =>
+      StyleSheet.create({
+        screen: {
+          flex: 1,
+          padding: compact ? 16 : 24,
+          gap: 12,
+          backgroundColor: theme.colors.surface0,
+        },
+        title: { color: theme.colors.foreground, fontSize: 18, fontWeight: "600" },
+        muted: { color: theme.colors.foregroundMuted },
+        meterTrack: {
+          height: 8,
+          borderRadius: 4,
+          overflow: "hidden",
+          backgroundColor: theme.colors.surface2,
+        },
+        meter: { height: 8, backgroundColor: theme.colors.accent },
+        upload: { gap: 8, alignItems: "flex-start" },
+        button: {
+          paddingVertical: 8,
+          paddingHorizontal: 14,
+          borderRadius: 8,
+          backgroundColor: theme.colors.accent,
+        },
+        buttonText: { color: theme.colors.accentForeground },
+      }),
+    [theme, compact],
+  );
+}

plugin-examples/media-stream/index.client.tsx

@@ -0,0 +1,22 @@
+import type { PluginClientContext } from "@getpaseo/plugin/client";
+import { MediaPanel } from "./client/media-panel";
+
+export default function contribute(client: PluginClientContext) {
+  client.addWorkspacePanel({
+    id: "media",
+    title: "Media stream",
+    icon: "MonitorPlay",
+    context: "workspace",
+    Component: MediaPanel,
+  });
+  client.addCommandCenterItem({
+    id: "open-media",
+    title: "Open media stream",
+    icon: "MonitorPlay",
+    context: "workspace",
+    onSelect({ openPanel }) {
+      openPanel("media");
+    },
+  });
+  return () => {};
+}