fix(plugins): bind button observations to mounted lifetimes

Compose shared header and composer buttons with owned subscription scopes. Renderer errors and closed content must release their mounted API without disrupting sibling buttons. Migrate incoming fixtures to explicit demand and verify mark-unread reply provenance.

packages/app/e2e/browser/plugin-buttons.spec.ts

@@ -12,3 +12,10 @@ test("plugin header and composer buttons share actions, menus, content, and upda
     await buttons.hideAndUnloadOpenButtons();
   });
 });
+
+test("plugin button observations follow mounted content and error lifetimes", async ({
+  page,
+}, testInfo) => {
+  test.setTimeout(120_000);
+  await withButtonShowcase(page, testInfo, (buttons) => buttons.verifyObservationLifetime());
+});

packages/app/e2e/browser/plugin-workspace-panels.spec.ts

@@ -67,7 +67,7 @@ function contributeClient(client) {
     pills.get(agentId)?.();
     pills.delete(agentId);
   };
-  const unsubscribe = client.paseo.agents.subscribe((update) => {
+  const apply = (update) => {
     if (update.kind === "remove") {
       remove(update.agentId);
       return;
@@ -94,9 +94,20 @@ function contributeClient(client) {
       },
     });
     pills.set(agent.id, () => pill.remove());
-  });
+  };
+  const lifetime = new AbortController();
+  void client.paseo.agents.list({ subscribe: {}, signal: lifetime.signal }).then(({ subscription }) => {
+    subscription.subscribe({
+      snapshot({ entries }) {
+        for (const removePill of pills.values()) removePill();
+        pills.clear();
+        for (const { agent } of entries) apply({ kind: "upsert", agent });
+      },
+      update(message) { if (message.type === "agent_update") apply(message.payload); },
+    });
+  }).catch((error) => { if (!lifetime.signal.aborted) console.error(error); });
   return () => {
-    unsubscribe();
+    lifetime.abort();
     for (const removePill of pills.values()) removePill();
     pills.clear();
   };

packages/app/e2e/support/helpers/plugin-buttons.ts

@@ -18,7 +18,7 @@ const COMPACT = { width: 390, height: 844 };
 function clientSource(workspaceId: string, agentId: string): string {
   return `import React from "react";
 import { Text, View, Pressable } from "react-native";
-import { useWorkspace, useRpc } from "@getpaseo/plugin/client";
+import { useWorkspace, useRpc, usePaseo } from "@getpaseo/plugin/client";
 import { useQuery } from "@tanstack/react-query";
 import { summary } from "./shared/rpc";
 
@@ -28,9 +28,22 @@ function StatusIcon({ size, theme }) {
 
 function Details({ workspaceId, theme, layout, close }) {
   const workspace = useWorkspace(workspaceId, (workspace) => workspace.name);
+  const paseo = usePaseo();
+  const [ownerId, setOwnerId] = React.useState("");
+  const [updates, setUpdates] = React.useState(0);
+  const [failed, setFailed] = React.useState(false);
+  React.useEffect(() => {
+    const owner = paseo.observeEvents(["project.update"]);
+    owner.subscribe({ snapshot(value) { setOwnerId(value.subscriptionId); }, update() { setUpdates((n) => n + 1); } });
+    // Deliberately leave this observation to the mounted host scope.
+  }, [paseo]);
   const rpc = useRpc(summary);
   const query = useQuery({ queryKey: ["summary"], queryFn: () => rpc({}) });
+  if (failed) throw new Error("Button content failed");
   return <View style={{ gap: 12 }}>
+    <Text accessibilityLabel="Content observation" style={{ color: theme.colors.foreground }}>{ownerId}</Text>
+    <Text style={{ color: theme.colors.foreground }}>Observed updates: {updates}</Text>
+    <Pressable accessibilityRole="button" accessibilityLabel="Fail button content" onPress={() => setFailed(true)}><Text style={{ color: theme.colors.foreground }}>Fail content</Text></Pressable>
     <Text style={{ color: theme.colors.foreground, fontSize: 18 }}>Deployment status</Text>
     <Text style={{ color: theme.colors.foregroundMuted }}>{workspace}</Text>
     <Text style={{ color: theme.colors.statusSuccess }}>{query.data?.message ?? "Loading deployment..."}</Text>
@@ -83,12 +96,19 @@ export default function contribute(client) {
   command("icon-only", "Use icon-only header", () => deploy.update({ label: undefined }));
   command("hide-menu", "Hide composer menu", () => pillMenu.update({ visible: false }));
   command("show-menu", "Show composer menu", () => pillMenu.update({ visible: true }));
-  const unsubscribe = client.paseo.workspaces.ref(workspaceId).subscribe((update) => {
-    if (update.kind !== "upsert") return;
-    deploy.update({ visible: update.workspace.name !== "Compact checks" });
-    pillMenu.update({ visible: update.workspace.name !== "Hide composer menu" });
-  });
-  return () => { unsubscribe(); finish(); finishReview(); for (const registration of registrations) registration.remove(); };
+  const lifetime = new AbortController();
+  const applyWorkspace = (workspace) => {
+    if (workspace.id !== workspaceId) return;
+    deploy.update({ visible: workspace.name !== "Compact checks" });
+    pillMenu.update({ visible: workspace.name !== "Hide composer menu" });
+  };
+  void client.paseo.workspaces.list({ subscribe: {}, signal: lifetime.signal }).then(({ subscription }) => {
+    subscription.subscribe({
+      snapshot({ entries }) { for (const workspace of entries) applyWorkspace(workspace); },
+      update(message) { if (message.type === "workspace_update" && message.payload.kind === "upsert") applyWorkspace(message.payload.workspace); },
+    });
+  }).catch((error) => { if (!lifetime.signal.aborted) console.error(error); });
+  return () => { lifetime.abort(); finish(); finishReview(); for (const registration of registrations) registration.remove(); };
 }`;
 }
 
@@ -222,8 +242,18 @@ export async function withButtonShowcase(
     runAndUpdateActions(): Promise<void>;
     openCompactSheets(): Promise<void>;
     hideAndUnloadOpenButtons(): Promise<void>;
+    verifyObservationLifetime(): Promise<void>;
   }) => Promise<void>,
 ) {
+  const released = new Set<string>();
+  page.on("websocket", (socket) =>
+    socket.on("framereceived", ({ payload }) => {
+      if (typeof payload !== "string") return;
+      const frame = JSON.parse(payload);
+      if (frame.message?.type === "subscription.release.response")
+        released.add(frame.message.payload.subscriptionId);
+    }),
+  );
   const workspace = await seedWorkspace({ repoPrefix: "plugin-buttons-" });
   const agent = await workspace.client.createAgent({
     provider: "mock",
@@ -241,6 +271,45 @@ export async function withButtonShowcase(
     await expect(page.getByRole("button", { name: "Run review", exact: true })).toBeVisible();
 
     await run({
+      verifyObservationLifetime: () =>
+        test.step("closing and crashing content release its owner while sibling buttons and app remain live", async () => {
+          await press(page, "Header status");
+          await expectDetails(page);
+          const observation = page.getByLabel("Content observation", { exact: true });
+          await expect(observation).not.toHaveText("");
+          const first = (await observation.textContent())!;
+          await press(page, "Close deployment details");
+          await expect.poll(() => released.has(first)).toBe(true);
+          await press(page, "Composer status");
+          await expectDetails(page);
+          const second = (await observation.textContent())!;
+          expect(second).not.toBe(first);
+          await workspace.client.renameProject(workspace.projectId, "Observer control");
+          await expect(page.getByText("Observed updates: 1", { exact: true })).toBeVisible();
+          await press(page, "Fail button content");
+          await expect(
+            page.getByText("Plugin failed: Button content failed", { exact: true }),
+          ).toBeVisible();
+          await expect.poll(() => released.has(second)).toBe(true);
+          await page.keyboard.press("Escape");
+          await press(page, "Header status");
+          await expectDetails(page);
+          const recovered = (await observation.textContent())!;
+          expect([first, second]).not.toContain(recovered);
+          await showcase.setTitle("Healthy app after button failure");
+          await expect(page.getByTestId("workspace-header-title")).toHaveText(
+            "Healthy app after button failure",
+          );
+          await showcase.disable();
+          await expect.poll(() => released.has(recovered)).toBe(true);
+          await expect(page.getByRole("button", { name: "Run review", exact: true })).toHaveCount(
+            0,
+          );
+          await showcase.setTitle("Healthy app after plugin unload");
+          await expect(page.getByTestId("workspace-header-title")).toHaveText(
+            "Healthy app after plugin unload",
+          );
+        }),
       openWideMenusAndPopovers: () =>
         test.step("wide buttons have placement-owned labels and chevrons", async () => {
           await expect(

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

@@ -206,17 +206,18 @@ describe("server data push router", () => {
     const cwd = "/repo";
     const queryKey = checkoutDiffQueryKey(serverId, cwd, "uncommitted", undefined, false);
     const subscriptionId = "diff-sharing";
-    queryClient.getQueryCache().build(queryClient, {
+    const observer = new QueryObserver(queryClient, {
       queryKey,
       queryFn: skipToken,
       meta: checkoutDiffPushRoute({
-        enabled: false,
+        enabled: true,
         serverId,
         cwd,
         subscriptionId,
         compare: { mode: "uncommitted" },
       }),
     });
+    const unsubscribe = observer.subscribe(() => {});
     const unmount = mountServerDataPushRouter({ queryClient, client: fake.client, serverId });
     type Payload = SubscribeCheckoutDiffResponseMessage["payload"];
     const files: Payload["files"] = ["a.ts", "b.ts"].map((path) => ({
@@ -248,7 +249,13 @@ describe("server data push router", () => {
     const publish = (incomingFiles: Payload["files"], requestId: string) => {
       fake.emit({
         type: "subscribe_checkout_diff_response",
-        payload: { subscriptionId, cwd, files: incomingFiles, requestId, error: null },
+        payload: {
+          subscriptionId: fake.subscribeCheckoutDiffCalls[0]!.subscriptionId,
+          cwd,
+          files: incomingFiles,
+          requestId,
+          error: null,
+        },
       });
       return queryClient.getQueryData<Payload>(queryKey)!;
     };
@@ -265,6 +272,7 @@ describe("server data push router", () => {
       expect(next.files[0]!.hunks[0]!.lines[0]!.tokens![0]!.style).toBe("variable");
       expect(publish([], "deleted").files).toEqual([]);
     } finally {
+      unsubscribe();
       unmount();
       queryClient.clear();
     }

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

@@ -6,6 +6,7 @@ import { PluginButtonStore, buttonMatches } from "./model";
 
 function installation(): InstalledPlugin {
   return {
+    lifetime: new AbortController(),
     id: "review",
     serverId: "host-a",
     clientBundle: "bundle",

packages/app/src/plugins/buttons/view.tsx

@@ -36,7 +36,6 @@ import type { Theme } from "@/styles/theme";
 import { createPluginClientStateSource } from "../client-state/source";
 import { Icon } from "../icons";
 import { PluginRuntimeBoundary } from "../runtime-boundary";
-import { createPluginSurfaceRuntime, type PluginSurfaceRuntime } from "../surface-runtime";
 import { SurfaceErrorBoundary } from "../surface-error-boundary";
 import { toPluginTheme } from "../theme";
 import { buttonMatches, type RegisteredPluginButton } from "./model";
@@ -45,7 +44,7 @@ import { pluginButtonStore } from "./store";
 interface ButtonView {
   entry: RegisteredPluginButton;
   props: PluginHostProps & RegisteredPluginButton["context"];
-  runtime: PluginSurfaceRuntime;
+  client: NonNullable<ReturnType<typeof useHostRuntimeClient>>;
   state: ReturnType<typeof createPluginClientStateSource>;
   toast: ReturnType<typeof useToast>;
 }
@@ -79,7 +78,7 @@ function resolvePlatform(): PluginHostProps["layout"]["platform"] {
 function ButtonEnvironment({ view, children }: { view: ButtonView; children: ReactNode }) {
   return (
     <ToastApiProvider api={view.toast}>
-      <PluginRuntimeBoundary plugin={view.entry.installation} runtime={view.runtime}>
+      <PluginRuntimeBoundary plugin={view.entry.installation} client={view.client}>
         <PluginClientStateProvider source={view.state}>{children}</PluginClientStateProvider>
       </PluginRuntimeBoundary>
     </ToastApiProvider>
@@ -219,11 +218,11 @@ function ButtonSurfaceBody({
   path: readonly string[];
 }) {
   return (
-    <ButtonEnvironment view={view}>
-      <SurfaceErrorBoundary installation={view.entry.installation} Surface={behavior}>
+    <SurfaceErrorBoundary installation={view.entry.installation} Surface={behavior}>
+      <ButtonEnvironment view={view}>
         <ButtonBody view={view} behavior={behavior} path={path} />
-      </SurfaceErrorBoundary>
-    </ButtonEnvironment>
+      </ButtonEnvironment>
+    </SurfaceErrorBoundary>
   );
 }
 
@@ -396,11 +395,10 @@ function createButtonView({
   hostLabel: string;
   theme: PluginTheme;
 }): ButtonView | null {
-  const runtime = createPluginSurfaceRuntime(client, entry.installation.id);
-  if (!runtime) return null;
+  if (!client) return null;
   return {
     entry,
-    runtime,
+    client,
     toast,
     state: createPluginClientStateSource(entry.installation.serverId),
     props: {
@@ -486,11 +484,15 @@ function OverflowPages({
       const view = createButtonView({ entry, client, toast, compact, hostLabel, theme });
       if (!view) continue;
       rows.push(
-        <ButtonEnvironment key={entry.key} view={view}>
-          <SurfaceErrorBoundary installation={entry.installation} Surface={entry.button.icon}>
+        <SurfaceErrorBoundary
+          key={entry.key}
+          installation={entry.installation}
+          Surface={entry.button.icon}
+        >
+          <ButtonEnvironment view={view}>
             <OverflowButton view={view} />
-          </SurfaceErrorBoundary>
-        </ButtonEnvironment>,
+          </ButtonEnvironment>
+        </SurfaceErrorBoundary>,
       );
       if (entry.button.behavior.kind === "action") continue;
       pages.push(

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

@@ -24,7 +24,8 @@ function registry() {
         openSurface: () => {},
         openSettings: () => {},
         openPanel: () => {},
-        addComposerPill: () => () => {},
+        addComposerPill: () => ({ update() {}, remove() {} }),
+        addHeaderButton: () => ({ update() {}, remove() {} }),
       };
     },
   });

packages/server/src/server/owned-subscriptions.e2e.test.ts

@@ -1585,3 +1585,90 @@ test("permission outcomes and domain observations keep independent source owners
     await daemon.close();
   }
 });
+
+test("mark unread replies remain source-owned while directory observers receive attention", async () => {
+  const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
+  const admin = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  const peers: SubscriptionPeer[] = [];
+  try {
+    await admin.connect();
+    const agent = await admin.createAgent({
+      provider: "codex",
+      cwd: daemon.staticDir,
+      title: "Finished",
+    });
+    const actor = await SubscriptionPeer.connect(daemon.port, "mark-unread-shared");
+    peers.push(actor);
+    const sibling = await SubscriptionPeer.connect(daemon.port, "mark-unread-shared");
+    peers.push(sibling);
+    const quiet = await SubscriptionPeer.connect(daemon.port, "mark-unread-shared");
+    peers.push(quiet);
+    await sibling.request({ type: "fetch_agents_request", requestId: "directory", subscribe: {} });
+    const snapshot = sibling.frames.flatMap((frame) =>
+      frame.type === "session" && frame.message.type === "fetch_agents_response"
+        ? [frame.message.payload]
+        : [],
+    )[0];
+    const workspaceId = snapshot.entries.find((entry) => entry.agent.id === agent.id)!.agent
+      .workspaceId!;
+    await actor.request({
+      type: "workspace.clear_attention.request",
+      workspaceId,
+      requestId: "clear",
+    });
+    actor.frames.length = sibling.frames.length = quiet.frames.length = 0;
+    await actor.request({ type: "workspace.mark_unread.request", workspaceId, requestId: "mark" });
+    expect(actor.frames).toContainEqual(
+      expect.objectContaining({
+        message: {
+          type: "workspace.mark_unread.response",
+          payload: {
+            workspaceId,
+            requestId: "mark",
+            markedAgentId: agent.id,
+            success: true,
+            error: null,
+          },
+        },
+      }),
+    );
+    await expect
+      .poll(() =>
+        sibling.frames.some(
+          (frame) =>
+            frame.type === "session" &&
+            frame.message.type === "agent_update" &&
+            frame.message.payload.kind === "upsert" &&
+            frame.message.payload.agent.id === agent.id &&
+            frame.message.payload.agent.requiresAttention &&
+            frame.message.payload.subscriptionId === snapshot.subscriptionId,
+        ),
+      )
+      .toBe(true);
+    expect(
+      sibling.frames.some(
+        (frame) =>
+          frame.type === "session" && frame.message.type === "workspace.mark_unread.response",
+      ),
+    ).toBe(false);
+    expect(quiet.frames).toEqual([]);
+    expect(quiet.binaryFrames).toEqual([]);
+    await actor.request({
+      type: "workspace.mark_unread.request",
+      workspaceId,
+      requestId: "already-unread",
+    });
+    expect(actor.frames).toContainEqual(
+      expect.objectContaining({
+        message: expect.objectContaining({
+          type: "workspace.mark_unread.response",
+          payload: expect.objectContaining({ requestId: "already-unread", success: false }),
+        }),
+      }),
+    );
+  } finally {
+    for (const peer of peers) peer.close();
+    await admin.close();
+    await daemon.close();
+  }
+});

public-docs/plugins/v0.8/reference.md

@@ -1357,7 +1357,7 @@ const pill = client.addComposerPill({
 });
 ```
 
-For pills that follow the agent directory, use an explicit [owned list subscription](/docs/sdk/events#agent-directory).
+For pills that follow the agent directory, use an explicit [owned list subscription](/docs/sdk/events#follow-one-agents-status).
 The [local plugin example](https://github.com/getpaseo/paseo/blob/main/plugin-examples/local-plugin/client/main.tsx)
 replaces registrations on each snapshot and aborts the observation during entry cleanup, including pending bootstrap.