Keep older daemons usable and let clients choose timeline recovery (#4737)

* fix(subscriptions): preserve older-host workflows and explicit timeline recovery

Keep established app and SDK workflows available on older daemons without emulating independent directory filters. Restore live timeline membership without choosing the consumer's history recovery policy. Preserve daemon filesystem paths separately from legacy workspace IDs.

* fix(client): keep legacy workspace replies scoped to the requested page

docs/plugins.md

@@ -356,8 +356,10 @@ query, and toast providers inside the surface content, including overflow pages
 plugins. Providers only around the trigger do not reach those bodies.
 
 Request observation with `client.paseo.agents.list({ subscribe: {} })` and consume the returned
-`subscription` handle. Plain `list()` and local `.subscribe(handler)` listeners create no daemon
-demand. Each list-and-subscribe call has its own server ID, even for the same query. Handle snapshots
+`subscription` handle. Plain `list()` and agent/workspace directory `.subscribe(handler)` listeners create no daemon
+demand. Provider and project `subscribe()` calls establish their own demand. On capable daemons, each
+list-and-subscribe call has its own server ID, even for the same query. Older daemons retain
+[shared delivery behavior](protocol-compatibility.md#owned-observations). Handle snapshots
 also run after reconnect; replace your view before applying its subsequent updates. The installation
 owns every observation created through its API and releases them on unload, including setup failure.
 Mounted surfaces and command invocations have shorter API lifetimes.

docs/protocol-compatibility.md

@@ -46,15 +46,26 @@ The app, CLI, and plugins inherit those defaults; they supply only host resource
 automation, or explicit overrides. A schema accepting a message does not establish support for
 its delivery semantics.
 
-Connecting creates no timeline or event demand. Client subscriptions own their network membership,
+On capable daemons, connecting creates no timeline or event demand. Client subscriptions own their network membership,
 release it on unsubscribe, and restore it after reconnect. Raw message observers inspect traffic
 without requesting streams. Application caches and which agents are visible remain caller-owned.
 
 ## Owned observations
 
-`owned_subscriptions` and `server_info.features.ownedSubscriptions` negotiate the entire source-owned
-contract described in [architecture](architecture.md#websocket-protocol). The client gates observation
-once and tells users to update an older host. It does not emulate independent handles over legacy slots.
+`owned_subscriptions` and `server_info.features.ownedSubscriptions` negotiate the source-owned
+contract described in [architecture](architecture.md#websocket-protocol). The client selects delivery
+behavior once at its connection boundary. App workflows use the same observation interface on both.
+
+With an older daemon, the client uses the existing connection and legacy RPCs. Directory subscriptions
+remain shared and last-query-wins. Local handle IDs identify listeners; they do not promise independent
+server filters. Timeline and event membership retain their existing shared behavior. Releasing a handle
+detaches its listener and uses the old unsubscribe operation where one exists. Broadcast-only hosts keep
+broadcasting; readiness there means local attachment, not a daemon acknowledgement. No extra sockets
+or multiplexing emulation are introduced.
+
+Preserve established workflows when changing delivery internals. Independent filters and quiet
+connections require a capable daemon; opening the app, reading history and using terminals do not.
+Pre-registry workspace grouping and legacy event normalization belong inside the client boundary.
 
 Old clients keep their existing wire shapes and slot behavior at the daemon's source boundary.
 The adapter keys legacy slots by physical socket, so an old connection cannot replace a modern

packages/app/e2e/browser/owned-subscriptions-old-daemon.spec.ts

@@ -1,14 +1,42 @@
 import { randomUUID } from "node:crypto";
+import { submitMessage, expectComposerEditable } from "../support/helpers/composer";
+import {
+  focusTerminalSurface,
+  typeInTerminal,
+  getTerminalBufferText,
+} from "../support/helpers/terminal-perf";
+import { buildHostWorkspaceRoute } from "../../src/utils/host-routes";
 import { metroTest as test, expect } from "../support/fixtures";
 import { buildCreateAgentPreferences, buildSeededHost } from "../support/helpers/daemon-registry";
 import { startIsolatedHostDaemon } from "../support/helpers/isolated-host-daemon";
-import { buildAgentRoute } from "../support/helpers/mock-agent";
+import type { MockAgentWorkspace } from "../support/helpers/mock-agent";
 import { seedWorkspace } from "../support/helpers/seed-client";
+import {
+  expectTimelinePromptVisible,
+  holdOlderHistoryPages,
+  openAgentTimeline,
+  scrollThroughOlderHistoryPages,
+} from "../support/helpers/timeline-pagination";
 
-test("requires a host update before observing a published 0.2.5 daemon", async ({ page }) => {
+test("does not repeat an assistant block when the current app paginates a published 0.2.5 daemon", async ({
+  page,
+}) => {
   test.setTimeout(120_000);
   const serverId = `srv_old_pagination_${randomUUID().replaceAll("-", "").slice(0, 12)}`;
   const daemon = await startIsolatedHostDaemon(serverId, { publishedVersion: "0.2.5" });
+  const observationSockets = new Set<unknown>();
+  page.on("websocket", (socket) => {
+    socket.on("framesent", ({ payload }) => {
+      if (typeof payload !== "string") return;
+      const frame = JSON.parse(payload);
+      if (
+        frame.type === "session" &&
+        frame.message.type === "fetch_agents_request" &&
+        frame.message.subscribe
+      )
+        observationSockets.add(socket);
+    });
+  });
   const workspace = await seedWorkspace({
     repoPrefix: "timeline-old-daemon-pagination-",
     port: daemon.port,
@@ -21,16 +49,32 @@ test("requires a host update before observing a published 0.2.5 daemon", async (
     modeId: "load-test",
     model: "ten-second-stream",
   });
-  const requests: unknown[] = [];
-  page.on("websocket", (socket) =>
-    socket.on("framesent", ({ payload }) => {
-      if (typeof payload !== "string") return;
-      const frame = JSON.parse(payload);
-      if (frame.type === "session" && frame.message.type !== "ping") requests.push(frame.message);
-    }),
-  );
+  const agent: MockAgentWorkspace = {
+    agentId: createdAgent.id,
+    workspaceId: workspace.workspaceId,
+    cwd: workspace.repoPath,
+    client: workspace.client,
+    cleanup: workspace.cleanup,
+  };
 
   try {
+    for (let index = 0; index < 40; index += 1) {
+      await agent.client.sendAgentMessage(
+        agent.agentId,
+        `timeline-pagination-older-turn-${index}: emit 1 coalesced agent stream updates`,
+      );
+      await agent.client.waitForFinish(agent.agentId, 15_000);
+    }
+    await agent.client.sendAgentMessage(agent.agentId, "build a realistic long mock timeline");
+    await agent.client.waitForFinish(agent.agentId, 20_000);
+    for (let index = 0; index < 20; index += 1) {
+      await agent.client.sendAgentMessage(
+        agent.agentId,
+        `timeline-pagination-turn-${index}: emit 1 coalesced agent stream updates`,
+      );
+      await agent.client.waitForFinish(agent.agentId, 15_000);
+    }
+
     const host = buildSeededHost({
       serverId,
       endpoint: `127.0.0.1:${daemon.port}`,
@@ -45,20 +89,50 @@ test("requires a host update before observing a published 0.2.5 daemon", async (
       { seededHost: host, preferences: buildCreateAgentPreferences() },
     );
 
-    await page.goto(buildAgentRoute(workspace.workspaceId, createdAgent.id, serverId));
-    const updateHost = page.getByText("Update the host to use this version of Paseo.", {
-      exact: true,
+    const history = await holdOlderHistoryPages(page, agent, daemon.port);
+    await openAgentTimeline(page, agent, serverId);
+    await expectTimelinePromptVisible(
+      page,
+      "timeline-pagination-turn-19: emit 1 coalesced agent stream updates",
+    );
+    await scrollThroughOlderHistoryPages(page, 3, history);
+
+    history.expectRepeatedEntries();
+    await history.expectOwnedTextRendered("Now I have a clearer picture.");
+    expect(observationSockets.size).toBe(1);
+
+    await test.step("send a live turn after reconnecting the old daemon", async () => {
+      await daemon.restart();
+      await expect.poll(() => observationSockets.size, { timeout: 30_000 }).toBe(2);
+      await expectComposerEditable(page);
+      await submitMessage(
+        page,
+        "compatibility-after-reconnect: emit 1 coalesced agent stream updates",
+      );
+      await expectTimelinePromptVisible(
+        page,
+        "compatibility-after-reconnect: emit 1 coalesced agent stream updates",
+      );
+      await agent.client.waitForFinish(agent.agentId, 20_000);
+      await page.screenshot({ path: test.info().outputPath("old-daemon-live-reconnect.png") });
+    });
+
+    await test.step("open a terminal and receive its output", async () => {
+      const terminal = await agent.client.createTerminal(agent.cwd, "Compatibility terminal");
+      expect(terminal.error).toBeNull();
+      const terminalId = terminal.terminal!.id;
+      const route = buildHostWorkspaceRoute(serverId, workspace.workspaceId);
+      await page.goto(`${route}?open=${encodeURIComponent(`terminal:${terminalId}`)}`);
+      await focusTerminalSurface(page);
+      await typeInTerminal(page, "printf 'compatibility-%s\\n' terminal-output\n");
+      await expect
+        .poll(() => getTerminalBufferText(page), { timeout: 15_000 })
+        .toContain("compatibility-terminal-output");
+      await page.screenshot({ path: test.info().outputPath("old-daemon-terminal.png") });
+      await agent.client.killTerminal(terminalId);
     });
-    await expect(updateHost).toHaveCount(1);
-    await expect(updateHost).toBeVisible();
-    await expect(page.getByRole("button", { name: "Manage host", exact: true })).toBeVisible();
-    expect(requests).toEqual([]);
-    await page.reload();
-    await expect(updateHost).toHaveCount(1);
-    await expect(updateHost).toBeVisible();
-    expect(requests).toEqual([]);
   } finally {
-    await workspace.cleanup();
+    await agent.cleanup();
     await daemon.close();
   }
 });

packages/app/src/runtime/directory-sync/index.ts

@@ -487,13 +487,8 @@ export class DirectorySync {
     try {
       await this.waitForSessionMetadata(client, source);
       const serverInfo = useSessionStore.getState().sessions[this.serverId]?.serverInfo;
-      if (serverInfo?.features?.workspaceMultiplicity !== true) {
-        const deltas = this.workspaceTransactions.fail(transaction);
-        if (deltas) for (const delta of deltas) this.applyWorkspaceDelta(delta);
-        return;
-      }
-      const supportsProjectList = serverInfo.features?.projectList === true;
-      const supportsDirectorySync = serverInfo.features?.directorySync === true;
+      const supportsProjectList = serverInfo?.features?.projectList === true;
+      const supportsDirectorySync = serverInfo?.features?.directorySync === true;
       if (supportsProjectList) {
         await this.fetchProjectSnapshot(client, source, transaction, supportsDirectorySync);
       }

packages/app/src/runtime/host-runtime.test.ts

@@ -570,7 +570,7 @@ class BrowserClientLifecycle {
 }
 
 describe("HostRuntimeController", () => {
-  it("gates an old host before publishing the app client or mounting observations", async () => {
+  it("publishes an old host and mounts observations through the client interface", async () => {
     const host = makeHost();
     const client = new FakeDaemonClient();
     client.ownedSubscriptions = false;
@@ -595,12 +595,12 @@ describe("HostRuntimeController", () => {
     const stop = controller.subscribe(() => publishedClients.push(controller.getSnapshot().client));
     await controller.activateConnection({ connectionId: host.connections[0]!.id });
     expect(controller.getSnapshot()).toMatchObject({
-      connectionStatus: "error",
-      client: null,
-      lastError: "Update the host to use this version of Paseo.",
+      connectionStatus: "online",
+      client,
+      lastError: null,
     });
-    expect(mounts).toBe(0);
-    expect(publishedClients.every((value) => value === null)).toBe(true);
+    expect(mounts).toBe(1);
+    expect(publishedClients).toContain(client);
     expect(client.fetchAgentsCalls).toEqual([]);
     stop();
     await controller.stop();

packages/app/src/runtime/host-runtime.ts

@@ -1273,24 +1273,13 @@ export class HostRuntimeController {
         /* Preserve the compatibility/connection error. */
       }
     };
-    const requireAppHost = () => {
-      if (client.getLastServerInfoMessage()?.features?.ownedSubscriptions !== true)
-        throw new Error("Update the host to use this version of Paseo.");
-    };
     try {
       if (!existingClient) await client.connect();
       if (!this.isCurrentSwitchRequest(requestVersion) || this.activeClient !== client) return;
-      requireAppHost();
       this.unsubscribeClientHandlers =
         this.deps.mountClientHandlers?.({ client, host: this.host, connection }) ?? null;
       this.unsubscribeClientStatus = client.subscribeConnectionStatus((state) => {
         if (!this.isCurrentSwitchRequest(requestVersion) || this.activeClient !== client) return;
-        try {
-          if (state.status === "connected") requireAppHost();
-        } catch (error) {
-          void failConnection(error);
-          return;
-        }
         this.applyConnectionEvent({ type: "client_state", state, lastError: client.lastError });
         this.updateSnapshot({
           ...toSnapshotConnectionPatch(this.connectionMachineState, this.connectionEpoch),

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

@@ -145,9 +145,8 @@ export async function runAttachCommand(
     const unsubscribe = client.subscribeAgentTimeline(resolvedId, (message) => {
       if (message.type === "agent.timeline.replacement") {
         console.log("\n[Timeline replaced; earlier output is no longer current]");
-      } else if (message.type === "agent.timeline.snapshot") {
-        console.log("\n[Reconnected; current recent history follows]");
-        for (const entry of message.payload.page.entries) printTimelineItem(entry.item);
+      } else if (message.type === "agent.timeline.subscription_restored") {
+        console.log("\n[Reconnected; live output resumed. Events may have been missed.]");
       } else if (message.type === "agent.timeline.error") {
         console.error(`Timeline observation stopped: ${message.payload.error}`);
       } else {

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

@@ -203,12 +203,8 @@ async function runFollowMode(
       console.error(`Timeline observation stopped: ${message.payload.error}`);
       return;
     }
-    if (message.type === "agent.timeline.snapshot") {
-      console.log("\n[Reconnected; current recent history follows]");
-      const items = message.payload.page.entries
-        .map((entry) => entry.item)
-        .filter((item) => !options.filter || matchesFilter(item, options.filter));
-      console.log(formatAgentActivityTranscript(items, tailCount));
+    if (message.type === "agent.timeline.subscription_restored") {
+      console.log("\n[Reconnected; live output resumed. Events may have been missed.]");
       return;
     }
     if (message.payload.event.type === "timeline") {

packages/client/examples/events-and-timeline.ts

@@ -30,8 +30,8 @@ export async function subscribeToEvents(
   });
 
   const unsubscribeTimeline = client.agents.ref(agentId).timeline.subscribe((event) => {
-    if (event.event.type === "snapshot") {
-      console.log("Replace recent history", event.event.page.epoch, event.event.page.entries);
+    if (event.event.type === "subscription_restored") {
+      console.log("Live subscription restored; request any missed history explicitly.");
     } else if (event.event.type === "error") {
       console.error(event.event.error);
     }

packages/client/src/connection.test.ts

@@ -6,6 +6,10 @@ function connection(
   options: {
     acknowledgeSubscriptions?: boolean;
     ownedSubscriptions?: boolean;
+    ownedCapability?: boolean;
+    workspaceMultiplicity?: boolean;
+    broadcasts?: boolean;
+    browserHost?: { hostKind: string; supportedCommands: string[] };
     acknowledgeTimelineReads?: boolean;
   } = {},
 ) {
@@ -40,8 +44,9 @@ function connection(
                 version: null,
                 features: {
                   ...(options.ownedSubscriptions === false ? {} : { ownedSubscriptions: true }),
-                  selectiveAgentTimeline: true,
-                  explicitEventSubscriptions: true,
+                  workspaceMultiplicity: options.workspaceMultiplicity ?? true,
+                  selectiveAgentTimeline: !options.broadcasts,
+                  explicitEventSubscriptions: !options.broadcasts,
                 },
               },
             },
@@ -112,6 +117,12 @@ function connection(
   const client = new DaemonClient({
     url: "ws://test",
     clientId: "test",
+    capabilities: {
+      [CLIENT_CAPS.browserHost]: options.browserHost,
+      ...(options.ownedCapability === undefined
+        ? {}
+        : { [CLIENT_CAPS.ownedSubscriptions]: options.ownedCapability }),
+    },
     transportFactory: () => transport,
     reconnect: { enabled: false },
   });
@@ -382,19 +393,30 @@ test("failed terminal bootstrap reports the domain error without reconnecting or
   }
 });
 
-test("an old host produces an update-host error without sending a legacy subscription", async () => {
+test("older hosts share event demand on one connection and release only their own listeners", async () => {
   const h = connection({ ownedSubscriptions: false });
   try {
     const connecting = h.client.connect();
     h.open();
     await connecting;
-    const observation = h.client.observeAgents({ filter: { labels: { role: "orchestrator" } } });
-    await expect(observation.ready).rejects.toThrow(
-      "Update the host to use independent subscriptions.",
-    );
-    await observation.release();
-    expect(h.sent.filter((frame) => frame.type === "session")).toEqual([]);
-    expect(h.client.isConnected).toBe(true);
+    const first = h.client.observeEvents(["project.update"]);
+    const second = h.client.observeEvents(["project.update"]);
+    await Promise.all([first.ready, second.ready]);
+    const received: string[] = [];
+    first.subscribe({ snapshot: () => {}, update: () => received.push("first") });
+    second.subscribe({ snapshot: () => {}, update: () => received.push("second") });
+    h.receive({ type: "project.update", payload: { kind: "remove", projectId: "project" } });
+    expect(received).toEqual(["first", "second"]);
+    await first.release();
+    h.receive({ type: "project.update", payload: { kind: "remove", projectId: "project" } });
+    expect(received).toEqual(["first", "second", "second"]);
+    expect(h.sent.at(-1)?.message?.events).toEqual(["project.update"]);
+    await second.release();
+    expect(h.sent.at(-1)?.message?.events).toEqual([]);
+    expect(h.sent.filter((frame) => frame.type === "hello")).toHaveLength(1);
+    expect(
+      h.sent.filter((frame) => frame.message?.type === "subscription.release.request"),
+    ).toEqual([]);
   } finally {
     await h.client.close();
   }
@@ -474,9 +496,13 @@ function timelinePage(requestId: string | undefined, epoch: string, seq: number)
   };
 }
 
-test("timeline recovery snapshots precede live updates for independent owners and survive epoch changes", async () => {
+test.each([
+  { name: "modern", ownedSubscriptions: true, broadcasts: false },
+  { name: "legacy selective", ownedSubscriptions: false, broadcasts: false },
+  { name: "legacy broadcast", ownedSubscriptions: false, broadcasts: true },
+])("$name timelines restore live delivery without requesting history", async (mode) => {
   const { createPaseoApi } = await import("./index");
-  const h = connection({ acknowledgeTimelineReads: false });
+  const h = connection(mode);
   const api = createPaseoApi(h.client);
   const received: import("./index").PaseoAgentTimelineEvent[][] = [[], [], []];
   try {
@@ -487,118 +513,154 @@ test("timeline recovery snapshots precede live updates for independent owners an
       api.agents.ref("agent").timeline.subscribe((event) => events.push(event)),
     );
     await Promise.all(owners.map((owner) => owner.ready));
-    const reads = () =>
-      h.sent.filter((frame) => frame.message?.type === "fetch_agent_timeline_request");
-    expect(reads()).toHaveLength(0); // Initial subscription remains live-only.
-    const oldIds = new Set(owners.map((owner) => owner.subscriptionId));
-    h.disconnect();
-    const reconnected = h.client.connect();
-    h.open();
-    await reconnected;
-    await expect.poll(() => reads().length).toBe(3);
-    expect(new Set(owners.map((owner) => owner.subscriptionId)).size).toBe(3);
-    expect(owners.every((owner) => !oldIds.has(owner.subscriptionId))).toBe(true);
-    await owners[2].release(); // The history reply can arrive after the owner is gone.
-    const live = (index: number, seq: number, epoch: string) =>
-      h.receive({
-        type: "agent_stream",
-        payload: {
+    expect(received).toEqual([[], [], []]);
+    for (let cycle = 0; cycle < 2; cycle++) {
+      const oldIds = owners.map((owner) => owner.subscriptionId);
+      h.disconnect();
+      await owners[2].release();
+      const reconnected = h.client.connect();
+      h.open();
+      await reconnected;
+      await expect.poll(() => owners[0].subscriptionId).not.toBe(oldIds[0]);
+      expect(
+        h.sent.filter((frame) => frame.message?.type === "fetch_agent_timeline_request"),
+      ).toEqual([]);
+      for (const index of [0, 1]) {
+        expect(owners[index].subscriptionId).not.toBeNull();
+        expect(received[index].at(-1)).toEqual({
           agentId: "agent",
           subscriptionId: owners[index].subscriptionId,
-          seq,
-          epoch,
-          timestamp: new Date(0).toISOString(),
-          event: {
-            type: "timeline",
-            provider: "codex",
-            item: { type: "user_message", text: `live-${seq}` },
-          },
-        },
-      });
-    live(0, 4, "old");
-    live(0, 5, "old");
-    h.receive(timelinePage(reads()[1].message?.requestId, "new", 4));
-    h.receive(timelinePage(reads()[2].message?.requestId, "new", 4));
-    await expect.poll(() => received[1].length).toBe(1);
-    expect(received[0]).toEqual([]);
-    live(1, 5, "new");
-    h.receive(timelinePage(reads()[0].message?.requestId, "old", 4));
-    await expect.poll(() => received[0].length).toBe(2);
-    for (const [index, epoch] of [
-      [0, "old"],
-      [1, "new"],
-    ] as const) {
-      expect(received[index][0]).toMatchObject({
-        event: {
-          type: "snapshot",
-          reason: "reconnect",
-          page: {
-            epoch,
-            hasOlder: true,
-            startCursor: { epoch, seq: 4 },
-            entries: [{ item: { text: `persisted-${epoch}-4` } }],
-          },
-        },
-      });
-      expect(received[index][1]).toMatchObject({ seq: 5 });
+          event: { type: "subscription_restored" },
+        });
+      }
+      const live = {
+        agentId: "agent",
+        timestamp: new Date(0).toISOString(),
+        event: { type: "turn_completed", provider: "codex" },
+      };
+      if (mode.ownedSubscriptions) {
+        for (const owner of owners.slice(0, 2))
+          h.receive({
+            type: "agent_stream",
+            payload: { ...live, subscriptionId: owner.subscriptionId },
+          });
+      } else h.receive({ type: "agent_stream", payload: live });
+      for (const events of received.slice(0, 2))
+        expect(events.map((update) => update.event.type)).toEqual(
+          Array.from({ length: cycle + 1 }, () => [
+            "subscription_restored",
+            "turn_completed",
+          ]).flat(),
+        );
+      expect(received[2]).toEqual([]);
     }
-    expect(received[2]).toEqual([]);
-    await owners[0].release();
-    live(1, 6, "new");
-    expect(received[1]).toHaveLength(3);
   } finally {
     await api.dispose();
     await h.client.close();
   }
 });
 
-test("timeline recovery replaces an obsolete read after bounded overflow and epoch invalidation", async () => {
+test("a consumer chooses its recovery cursor and a failed read leaves live delivery active", async () => {
+  const { createPaseoApi } = await import("./index");
   const h = connection({ acknowledgeTimelineReads: false });
-  const messages: unknown[] = [];
+  const api = createPaseoApi(h.client);
+  const received: import("./index").PaseoAgentTimelineEvent[] = [];
+  let read: Promise<unknown> | undefined;
   try {
     const connected = h.client.connect();
     h.open();
     await connected;
-    const owner = h.client.subscribeAgentTimeline("agent", (message) => messages.push(message));
+    const timeline = api.agents.ref("agent").timeline;
+    const owner = timeline.subscribe((message) => {
+      received.push(message);
+      if (message.event.type === "subscription_restored") {
+        read = timeline.refetch({
+          direction: "after",
+          cursor: { epoch: "cached", seq: 42 },
+          limit: 7,
+        });
+        void read.catch(() => {});
+      }
+    });
     await owner.ready;
     h.disconnect();
     const reconnected = h.client.connect();
     h.open();
     await reconnected;
-    const reads = () =>
-      h.sent.filter((frame) => frame.message?.type === "fetch_agent_timeline_request");
-    await expect.poll(() => reads().length).toBe(1);
-    for (let seq = 1; seq <= 130; seq++)
+    await expect.poll(() => read !== undefined).toBe(true);
+    const requests = h.sent.filter(
+      (frame) => frame.message?.type === "fetch_agent_timeline_request",
+    );
+    expect(requests).toHaveLength(1);
+    expect(requests[0].message).toMatchObject({
+      direction: "after",
+      cursor: { epoch: "cached", seq: 42 },
+      limit: 7,
+    });
+    // A slow consumer-owned history read must not buffer or filter live events.
+    for (let seq = 43; seq < 173; seq++)
       h.receive({
         type: "agent_stream",
         payload: {
           agentId: "agent",
           subscriptionId: owner.subscriptionId,
           seq,
-          epoch: "old",
+          epoch: "cached",
           timestamp: new Date(0).toISOString(),
-          event: {
-            type: "timeline",
-            provider: "codex",
-            item: { type: "assistant_message", text: "piece" },
-          },
+          event: { type: "turn_completed", provider: "codex" },
         },
       });
-    h.receive(timelinePage(reads()[0].message?.requestId, "old", 1));
-    await expect.poll(() => reads().length).toBe(2);
-    expect(messages).toEqual([]);
+    expect(received).toHaveLength(131);
     h.receive({
       type: "agent.timeline.replacement",
       payload: { agentId: "agent", subscriptionId: owner.subscriptionId, epoch: "new" },
     });
-    h.receive(timelinePage(reads()[1].message?.requestId, "old", 130));
-    await expect.poll(() => reads().length).toBe(3);
-    expect(messages).toEqual([]);
-    h.receive(timelinePage(reads()[2].message?.requestId, "new", 2));
-    await expect.poll(() => messages.length).toBe(2);
-    expect(messages).toMatchObject([
-      { type: "agent.timeline.replacement", payload: { epoch: "new" } },
-      { type: "agent.timeline.snapshot", payload: { page: { epoch: "new" } } },
+    expect(received.at(-1)?.event).toEqual({ type: "replacement", epoch: "new" });
+    const page = timelinePage(requests[0].message?.requestId, "cached", 42);
+    h.receive({ ...page, payload: { ...page.payload, error: "History unavailable" } });
+    await expect(read).rejects.toThrow("History unavailable");
+    expect(owner.subscriptionId).not.toBeNull();
+    expect(received.some((message) => message.event.type === "error")).toBe(false);
+    expect(
+      h.sent.filter((frame) => frame.message?.type === "fetch_agent_timeline_request"),
+    ).toHaveLength(1);
+    await owner.release();
+  } finally {
+    await api.dispose();
+    await h.client.close();
+  }
+});
+
+test("subscription-restored notification waits for the new membership acknowledgement", async () => {
+  const h = connection({ acknowledgeSubscriptions: false });
+  const received: unknown[] = [];
+  const acknowledge = (id: string) =>
+    h.receive({
+      type: "agent.timeline.set_subscription.response",
+      payload: {
+        agentIds: ["agent"],
+        requestId: h.sent.at(-1)?.message?.requestId,
+        subscriptionId: id,
+      },
+    });
+  try {
+    const connected = h.client.connect();
+    h.open();
+    await connected;
+    const owner = h.client.subscribeAgentTimeline("agent", (message) => received.push(message));
+    acknowledge("first");
+    await owner.ready;
+    h.disconnect();
+    const reconnected = h.client.connect();
+    h.open();
+    await reconnected;
+    expect(received).toEqual([]);
+    acknowledge("second");
+    expect(received).toEqual([
+      {
+        type: "agent.timeline.subscription_restored",
+        payload: { agentId: "agent", subscriptionId: "second" },
+      },
     ]);
     await owner.release();
   } finally {
@@ -606,31 +668,361 @@ test("timeline recovery replaces an obsolete read after bounded overflow and epo
   }
 });
 
-test("timeline recovery failure is observable and releases its owner", async () => {
-  const h = connection({ acknowledgeTimelineReads: false });
-  const messages: unknown[] = [];
-  try {
-    const connected = h.client.connect();
-    h.open();
-    await connected;
-    const owner = h.client.subscribeAgentTimeline("agent", (message) => messages.push(message));
-    await owner.ready;
-    h.disconnect();
-    const reconnected = h.client.connect();
-    h.open();
-    await reconnected;
-    const reads = () =>
-      h.sent.filter((frame) => frame.message?.type === "fetch_agent_timeline_request");
-    await expect.poll(() => reads().length).toBe(1);
-    const page = timelinePage(reads()[0].message?.requestId, "epoch", 0);
-    h.receive({ ...page, payload: { ...page.payload, error: "Agent no longer available" } });
-    await expect.poll(() => owner.subscriptionId).toBe(null);
-    expect(messages).toEqual([
-      {
-        type: "agent.timeline.error",
-        payload: { agentId: "agent", error: "Agent no longer available" },
+function legacyAgent(input: {
+  id: string;
+  cwd: string;
+  status?: "idle" | "running";
+  updatedAt?: string;
+  projectRoot?: string;
+}) {
+  const updatedAt = input.updatedAt ?? "2026-06-18T10:00:00.000Z";
+  return {
+    agent: {
+      id: input.id,
+      provider: "mock",
+      cwd: input.cwd,
+      model: null,
+      createdAt: updatedAt,
+      updatedAt,
+      lastUserMessageAt: null,
+      status: input.status ?? "idle",
+      capabilities: {
+        supportsStreaming: true,
+        supportsSessionPersistence: true,
+        supportsDynamicModes: true,
+        supportsMcpServers: true,
+        supportsReasoningStream: true,
+        supportsToolInvocations: true,
       },
+      currentModeId: null,
+      availableModes: [],
+      pendingPermissions: [],
+      persistence: null,
+      title: null,
+      labels: {},
+    },
+    project: {
+      projectKey: "/repo",
+      projectName: "repo",
+      workspaceName: "app",
+      checkout: {
+        cwd: input.cwd,
+        isGit: true,
+        currentBranch: "main",
+        remoteUrl: "git@example.com:repo/app.git",
+        worktreeRoot: input.cwd,
+        isPaseoOwnedWorktree: false,
+        mainRepoRoot: input.projectRoot ?? "/repo",
+      },
+    },
+  };
+}
+
+test.each([
+  { cwd: "/repo/app", id: "/repo/app", root: "/repo" },
+  { cwd: "C:\\repo\\app", id: "C:/repo/app", root: "C:\\repo" },
+  { cwd: "C:\\", id: "C:", root: "C:\\" },
+  { cwd: "C:/", id: "C:", root: "C:/" },
+  { cwd: "\\\\server\\share\\", id: "//server/share", root: "\\\\server\\share\\" },
+  { cwd: "\\\\?\\C:\\repo\\app", id: "//?/C:/repo/app", root: "\\\\?\\C:\\repo" },
+  { cwd: "/repo/trailing space ", id: "/repo/trailing space", root: "/repo/ " },
+  { cwd: "/repo/back\\slash", id: "/repo/back/slash", root: "/repo" },
+])(
+  "old workspaces preserve daemon filesystem paths ($cwd) and stable IDs",
+  async ({ cwd, id, root }) => {
+    const h = connection({ ownedSubscriptions: false, workspaceMultiplicity: false });
+    try {
+      const connecting = h.client.connect();
+      h.open();
+      await connecting;
+      const workspaces = h.client.observeWorkspaces();
+      const request = h.sent.at(-1)!.message!;
+      expect(request.type).toBe("fetch_agents_request");
+      h.receive({
+        type: "fetch_agents_response",
+        payload: {
+          requestId: request.requestId,
+          entries: [legacyAgent({ id: "agent", cwd, projectRoot: root, status: "idle" })],
+          pageInfo: { hasMore: false, nextCursor: null, prevCursor: null },
+        },
+      });
+      expect((await workspaces.ready).entries).toEqual([
+        expect.objectContaining({
+          id,
+          workspaceDirectory: cwd,
+          projectRootPath: root,
+          projectId: "/repo",
+          status: "done",
+        }),
+      ]);
+      const updates: unknown[] = [];
+      workspaces.subscribe({ snapshot: () => {}, update: (message) => updates.push(message) });
+      const changed = legacyAgent({ id: "agent", cwd, projectRoot: root, status: "running" });
+      h.receive({ type: "agent_update", payload: { kind: "upsert", ...changed } });
+      expect(updates).toEqual([
+        expect.objectContaining({
+          type: "workspace_update",
+          payload: expect.objectContaining({
+            kind: "upsert",
+            workspace: expect.objectContaining({
+              id,
+              workspaceDirectory: cwd,
+              projectRootPath: root,
+              status: "running",
+            }),
+          }),
+        }),
+      ]);
+      const fetching = h.client.fetchAgent("agent");
+      const detail = h.sent.at(-1)!.message!;
+      h.receive({
+        type: "fetch_agent_response",
+        payload: { requestId: detail.requestId, ...changed, error: null },
+      });
+      expect((await fetching).agent?.workspaceId).toBe(id);
+      expect(h.sent.filter((frame) => frame.type === "hello")).toHaveLength(1);
+      await workspaces.release();
+    } finally {
+      await h.client.close();
+    }
+  },
+);
+
+test("legacy workspace pages contain only that page's groups and retain earlier live state", async () => {
+  const h = connection({ ownedSubscriptions: false, workspaceMultiplicity: false });
+  try {
+    const connecting = h.client.connect();
+    h.open();
+    await connecting;
+    const workspaces = h.client.observeWorkspaces();
+    h.receive({
+      type: "fetch_agents_response",
+      payload: {
+        requestId: h.sent.at(-1)!.message!.requestId,
+        entries: [legacyAgent({ id: "first", cwd: "/first", status: "running" })],
+        pageInfo: { hasMore: true, nextCursor: "next-page", prevCursor: null },
+      },
+    });
+    const first = await workspaces.ready;
+    const next = h.client.fetchWorkspaces({
+      page: { cursor: first.pageInfo.nextCursor!, limit: 1 },
+    });
+    h.receive({
+      type: "fetch_agents_response",
+      payload: {
+        requestId: h.sent.at(-1)!.message!.requestId,
+        entries: [legacyAgent({ id: "second", cwd: "/second" })],
+        pageInfo: { hasMore: false, nextCursor: null, prevCursor: "previous-page" },
+      },
+    });
+    const second = await next;
+    expect([...first.entries, ...second.entries].map((workspace) => workspace.id)).toEqual([
+      "/first",
+      "/second",
     ]);
+    const updates: unknown[] = [];
+    workspaces.subscribe({ snapshot: () => {}, update: (message) => updates.push(message) });
+    // Earlier pages must remain in the aggregate: this idle sibling cannot
+    // replace the running status of the first workspace.
+    h.receive({
+      type: "agent_update",
+      payload: { kind: "upsert", ...legacyAgent({ id: "sibling", cwd: "/first" }) },
+    });
+    expect(updates).toEqual([]);
+    h.receive({ type: "agent_update", payload: { kind: "remove", agentId: "first" } });
+    expect(updates).toEqual([
+      expect.objectContaining({
+        type: "workspace_update",
+        payload: expect.objectContaining({
+          kind: "upsert",
+          workspace: expect.objectContaining({ id: "/first", status: "done" }),
+        }),
+      }),
+    ]);
+    await workspaces.release();
+  } finally {
+    await h.client.close();
+  }
+});
+
+test("old attention stream events reach the current notification interface", async () => {
+  const h = connection({ ownedSubscriptions: false });
+  try {
+    const connecting = h.client.connect();
+    h.open();
+    await connecting;
+    const observation = h.client.observeEvents(["agent_attention_required"]);
+    await observation.ready;
+    const received: unknown[] = [];
+    observation.subscribe({ snapshot: () => {}, update: (message) => received.push(message) });
+    h.receive({
+      type: "agent_stream",
+      payload: {
+        agentId: "agent",
+        timestamp: "2026-09-11T00:00:00Z",
+        event: {
+          type: "attention_required",
+          provider: "mock",
+          reason: "finished",
+          timestamp: "2026-09-11T00:00:00Z",
+          shouldNotify: true,
+        },
+      },
+    });
+    expect(received).toEqual([
+      {
+        type: "agent_attention_required",
+        payload: {
+          agentId: "agent",
+          reason: "finished",
+          timestamp: "2026-09-11T00:00:00Z",
+          shouldNotify: true,
+          subscriptionId: observation.subscriptionId,
+        },
+      },
+    ]);
+    await observation.release();
+  } finally {
+    await h.client.close();
+  }
+});
+
+test("broadcast-only hosts expose local timeline readiness without sending an unsupported RPC", async () => {
+  const h = connection({ ownedSubscriptions: false, broadcasts: true });
+  try {
+    const connected = h.client.connect();
+    h.open();
+    await connected;
+    const observation = h.client.observeTimeline(["agent"]);
+    expect(await observation.ready).toEqual({
+      agentIds: ["agent"],
+      requestId: expect.any(String),
+      subscriptionId: observation.subscriptionId,
+    });
+    expect(h.sent).toHaveLength(1);
+    await observation.release();
+    expect(h.sent).toHaveLength(1);
+  } finally {
+    await h.client.close();
+  }
+});
+
+test("old browser hosting requires the registration sent in hello", async () => {
+  const h = connection({ ownedSubscriptions: false });
+  try {
+    const connected = h.client.connect();
+    h.open();
+    await connected;
+    const observation = h.client.registerBrowserHost({
+      hostKind: "test",
+      supportedCommands: ["list_tabs"],
+    });
+    await expect(observation.ready).rejects.toThrow("browser_host");
+    expect(h.sent).toHaveLength(1);
+  } finally {
+    await h.client.close();
+  }
+});
+
+test("old timeline observers retain shared membership and restore surviving interests", async () => {
+  const h = connection({ ownedSubscriptions: false });
+  try {
+    const connected = h.client.connect();
+    h.open();
+    await connected;
+    const app = h.client.observeTimeline(["app"]);
+    const plugin = h.client.observeTimeline(["plugin"]);
+    await Promise.all([app.ready, plugin.ready]);
+    expect(h.sent.at(-1)?.message?.agentIds).toEqual(["app", "plugin"]);
+    await plugin.release();
+    expect(h.sent.at(-1)?.message?.agentIds).toEqual(["app"]);
+    h.disconnect();
+    const reconnecting = h.client.connect();
+    h.open();
+    await reconnecting;
+    await expect.poll(() => h.sent.at(-1)?.message?.agentIds).toEqual(["app"]);
+    await app.release();
+    expect(h.sent.at(-1)?.message?.agentIds).toEqual([]);
+    expect(h.sent.some((frame) => frame.message?.type === "subscription.release.request")).toBe(
+      false,
+    );
+  } finally {
+    await h.client.close();
+  }
+});
+
+test("old browser hosts attach to their hello registration without another connection", async () => {
+  const registration = {
+    hostKind: "desktop app",
+    supportedCommands: ["list_tabs"] as ["list_tabs"],
+  };
+  const h = connection({ ownedSubscriptions: false, browserHost: registration });
+  try {
+    const connected = h.client.connect();
+    h.open();
+    await connected;
+    const observation = h.client.registerBrowserHost(registration);
+    expect(await observation.ready).toEqual({
+      requestId: expect.any(String),
+      subscriptionId: observation.subscriptionId,
+    });
+    await observation.release();
+    expect(h.sent).toHaveLength(1);
+  } finally {
+    await h.client.close();
+  }
+});
+
+test("delivery follows the negotiated client capability as well as the server feature", async () => {
+  const h = connection({ ownedCapability: false });
+  try {
+    const connected = h.client.connect();
+    h.open();
+    await connected;
+    const observation = h.client.observeEvents(["project.update"]);
+    expect((await observation.ready).subscriptionId).toMatch(/^legacy:/);
+    await observation.release();
+    expect(h.sent.at(-1)?.message).toMatchObject({
+      type: "session.events.set_subscription.request",
+      events: [],
+    });
+  } finally {
+    await h.client.close();
+  }
+});
+
+test("subscribeFile returns its initial version without calling the change callback", async () => {
+  const h = connection();
+  try {
+    const connected = h.client.connect();
+    h.open();
+    await connected;
+    const changes: unknown[] = [];
+    const initial = { status: "missing" as const, cwd: "/repo", path: "file.txt" };
+    const subscribing = h.client.subscribeFile({ cwd: "/repo", path: "file.txt" }, (value) =>
+      changes.push(value),
+    );
+    h.receive({
+      type: "fs.file.subscribe.response",
+      payload: {
+        requestId: h.sent.at(-1)!.message!.requestId,
+        subscriptionId: "file-owner",
+        initial,
+      },
+    });
+    const file = await subscribing;
+    expect(file.initial).toEqual(initial);
+    expect(changes).toEqual([]);
+    const version = {
+      status: "ready",
+      cwd: "/repo",
+      path: "file.txt",
+      size: 1,
+      modifiedAt: "2026-09-11T00:00:00Z",
+    };
+    h.receive({ type: "fs.file.update", payload: { subscriptionId: "file-owner", version } });
+    expect(changes).toEqual([version]);
+    await file.unsubscribe();
   } finally {
     await h.client.close();
   }

packages/client/src/connection/index.ts

@@ -1,6 +1,127 @@
 export { OwnedSubscriptions, type OwnedSubscription, type SubscriptionObserver } from "./owned.js";
+import {
+  OwnedSubscriptions,
+  type OwnedSubscription,
+  type OwnedSubscriptionsHost,
+} from "./owned.js";
+import { LegacySubscriptions, type ObservationRequest } from "./legacy.js";
+import {
+  type SessionInboundMessage,
+  type SessionOutboundMessage,
+  type ServerInfoStatusPayload,
+} from "@getpaseo/protocol/messages";
 import { CLIENT_CAPS, type ClientCapability } from "@getpaseo/protocol/client-capabilities";
 
+export class ConnectionSubscriptions extends OwnedSubscriptions {
+  private legacy: LegacySubscriptions | null = null;
+
+  constructor(
+    private readonly connection: OwnedSubscriptionsHost & {
+      send(message: ObservationRequest): Promise<void>;
+    },
+  ) {
+    super({
+      ...connection,
+      release: async (id) => {
+        if (!this.legacy) return connection.release(id);
+        const message = this.legacy.release(id);
+        if (message) await connection.send(message);
+      },
+    });
+  }
+
+  override restore(
+    info?: ServerInfoStatusPayload,
+    capabilities?: Partial<Record<ClientCapability, unknown>>,
+  ): void {
+    if (info) {
+      const clientCapabilities = capabilities ?? DEFAULT_CLIENT_CAPABILITIES;
+      const owned =
+        info.features?.ownedSubscriptions === true &&
+        clientCapabilities[CLIENT_CAPS.ownedSubscriptions] === true;
+      this.legacy = null;
+      if (!owned) {
+        this.legacy = new LegacySubscriptions(
+          info,
+          this.connection.send,
+          capabilities?.[CLIENT_CAPS.browserHost],
+        );
+      }
+    }
+    super.restore();
+  }
+
+  override disconnected(): void {
+    super.disconnected();
+    this.legacy = null;
+  }
+
+  observeRequest<T>(
+    query: ObservationRequest,
+    request: (query: ObservationRequest, accept: (snapshot: T) => void) => Promise<T>,
+    signal?: AbortSignal,
+    failed?: () => void,
+  ): OwnedSubscription<T> {
+    return super.observe(
+      async (accept) => {
+        const legacy = this.legacy;
+        if (!legacy) return request(query, accept);
+        const interest = legacy.start(query);
+        const snapshot = (payload: T): T & { subscriptionId: string } => ({
+          ...payload,
+          subscriptionId: interest.id,
+        });
+        try {
+          const message = legacy.request(interest);
+          if (!message) {
+            // Broadcast-era hosts have no acknowledgement. Readiness means the
+            // local listener is attached; no server membership is being claimed.
+            const value = snapshot({
+              requestId: interest.id,
+              ...(query.type === "agent.timeline.set_subscription.request"
+                ? { agentIds: query.agentIds }
+                : {}),
+            } as T);
+            accept(value);
+            return value;
+          }
+          return snapshot(await request(message, (payload) => accept(snapshot(payload))));
+        } catch (error) {
+          const release = legacy.release(interest.id);
+          if (release && this.legacy === legacy)
+            await this.connection.send(release).catch(this.connection.failed);
+          throw error;
+        }
+      },
+      signal,
+      failed,
+    );
+  }
+
+  override owns(message: SessionOutboundMessage): boolean {
+    return this.legacy ? this.legacy.owns(message) : super.owns(message);
+  }
+
+  override receive(message: SessionOutboundMessage): void {
+    if (this.legacy) this.legacy.receive(message, (update) => super.receive(update));
+    else super.receive(message);
+  }
+
+  normalize(message: SessionOutboundMessage): SessionOutboundMessage {
+    return this.legacy?.normalize(message) ?? message;
+  }
+
+  prepareRequest(message: SessionInboundMessage) {
+    return (
+      this.legacy?.prepareRequest(message) ?? {
+        message,
+        receive: (value: SessionOutboundMessage) => value,
+        finish: async () => {},
+      }
+    );
+  }
+}
+
 // Protocol support belongs to the installed client. Only browser hosting needs
 // a resource supplied by the caller. Keep this exhaustive as the protocol evolves.
 export const DEFAULT_CLIENT_CAPABILITIES = {
@@ -21,7 +142,7 @@ export const DEFAULT_CLIENT_CAPABILITIES = {
   [CLIENT_CAPS.explicitEventSubscriptions]: true,
 } satisfies Record<Exclude<ClientCapability, typeof CLIENT_CAPS.browserHost>, true>;
 
-/** Calling releases demand; ready acknowledges the initial daemon membership. */
+/** Calling releases demand; ready waits for membership, or local attachment on broadcast hosts. */
 export type TimelineSubscription = (() => void) & {
   readonly ready: Promise<void>;
   readonly subscriptionId: string | null;

packages/client/src/connection/legacy-workspaces.ts

@@ -0,0 +1,176 @@
+import { AgentSnapshotPayloadSchema } from "@getpaseo/protocol/messages";
+import type { AgentSnapshotPayload, SessionOutboundMessage } from "@getpaseo/protocol/messages";
+import {
+  deriveAgentStateBucket,
+  getWorkspaceStateBucketPriority,
+} from "@getpaseo/protocol/agent-state-bucket";
+
+type AgentEntry = Extract<
+  SessionOutboundMessage,
+  { type: "fetch_agents_response" }
+>["payload"]["entries"][number];
+type Workspace = Extract<
+  SessionOutboundMessage,
+  { type: "fetch_workspaces_response" }
+>["payload"]["entries"][number];
+type WorkspaceUpdate = Extract<SessionOutboundMessage, { type: "workspace_update" }>;
+
+// Preserve the pre-registry app's identity format. This is an opaque ID, never
+// a filesystem path: converting C:\ to C: changes a drive root into a relative path.
+function legacyWorkspaceId(value: string): string {
+  return value.trim().replace(/\\/g, "/").replace(/\/+$/, "") || "/";
+}
+
+// COMPAT(legacyWorkspaceDaemon): restored in v0.8.0; remove after 2027-03-11 once daemon floor >= v0.1.97.
+// The pre-registry app grouped agents by checkout path. Keep that representation
+// at the client edge, including live updates, so app workflows need no version branch.
+export class LegacyWorkspaces {
+  private readonly agents = new Map<string, AgentEntry>();
+
+  normalize(message: SessionOutboundMessage): SessionOutboundMessage {
+    const stamp = (agent: AgentSnapshotPayload): AgentSnapshotPayload => ({
+      ...agent,
+      workspaceId: agent.workspaceId ?? legacyWorkspaceId(agent.cwd),
+    });
+    if (message.type === "fetch_agents_response")
+      return {
+        ...message,
+        payload: {
+          ...message.payload,
+          entries: message.payload.entries.map((entry) => ({
+            ...entry,
+            agent: stamp(entry.agent),
+          })),
+        },
+      };
+    if (message.type === "agent_update" && message.payload.kind === "upsert")
+      return { ...message, payload: { ...message.payload, agent: stamp(message.payload.agent) } };
+    if (
+      (message.type === "fetch_agent_response" ||
+        message.type === "fetch_agent_timeline_response" ||
+        message.type === "cancel_agent_response") &&
+      message.payload.agent
+    ) {
+      return {
+        ...message,
+        payload: { ...message.payload, agent: stamp(message.payload.agent) },
+      } as SessionOutboundMessage;
+    }
+    if (
+      message.type === "status" &&
+      (message.payload.status === "agent_created" || message.payload.status === "agent_resumed")
+    ) {
+      return {
+        ...message,
+        payload: {
+          ...message.payload,
+          agent: stamp(AgentSnapshotPayloadSchema.parse(message.payload.agent)),
+        },
+      };
+    }
+    return message;
+  }
+
+  read(entries: AgentEntry[], reset: boolean): Workspace[] {
+    if (reset) this.agents.clear();
+    for (const entry of entries) this.agents.set(entry.agent.id, entry);
+    const pageIds = new Set(entries.map(workspaceId));
+    return [...this.workspaces().values()].filter((workspace) => pageIds.has(workspace.id));
+  }
+
+  update(message: SessionOutboundMessage): WorkspaceUpdate[] {
+    if (message.type !== "agent_update") return [];
+    const before = this.workspaces();
+    const update = message.payload;
+    if (update.kind === "remove") this.agents.delete(update.agentId);
+    else {
+      const project = update.project ?? this.agents.get(update.agent.id)?.project;
+      if (update.agent.archivedAt) this.agents.delete(update.agent.id);
+      else if (project) this.agents.set(update.agent.id, { agent: update.agent, project });
+    }
+    const after = this.workspaces();
+    const changes: WorkspaceUpdate[] = [];
+    for (const [id, workspace] of after) {
+      if (JSON.stringify(before.get(id)) !== JSON.stringify(workspace))
+        changes.push({ type: "workspace_update", payload: { kind: "upsert", workspace } });
+    }
+    for (const id of before.keys())
+      if (!after.has(id))
+        changes.push({ type: "workspace_update", payload: { kind: "remove", id } });
+    return changes;
+  }
+
+  private workspaces(): Map<string, Workspace> {
+    const workspaces = new Map<string, Workspace>();
+    for (const entry of this.agents.values()) {
+      const { agent, project } = entry;
+      const { checkout } = project;
+      const id = workspaceId(entry);
+      const status = deriveAgentStateBucket({
+        status: agent.status,
+        pendingPermissionCount: agent.pendingPermissions.length,
+        requiresAttention: agent.requiresAttention,
+        attentionReason: agent.attentionReason,
+      });
+      const existing = workspaces.get(id);
+      if (
+        existing &&
+        getWorkspaceStateBucketPriority(existing.status) <= getWorkspaceStateBucketPriority(status)
+      )
+        continue;
+      workspaces.set(id, {
+        id,
+        projectId: project.projectKey,
+        projectDisplayName: project.projectName,
+        projectCustomName: null,
+        projectRootPath: checkout.mainRepoRoot ?? checkout.worktreeRoot ?? checkout.cwd,
+        workspaceDirectory: checkout.cwd,
+        projectKind: checkout.isGit ? "git" : "non_git",
+        workspaceKind: workspaceKind(checkout),
+        name: workspaceName(entry, id),
+        title: null,
+        status,
+        statusEnteredAt: agent.attentionTimestamp ?? agent.updatedAt,
+        activityAt: agent.updatedAt,
+        archivingAt: null,
+        diffStat: null,
+        scripts: [],
+        gitRuntime: gitRuntime(checkout),
+        githubRuntime: null,
+        project,
+      });
+    }
+    return workspaces;
+  }
+}
+
+function workspaceId(entry: AgentEntry): string {
+  return entry.agent.workspaceId ?? legacyWorkspaceId(entry.project.checkout.cwd);
+}
+
+function workspaceKind(checkout: AgentEntry["project"]["checkout"]): Workspace["workspaceKind"] {
+  if (!checkout.isGit) return "directory";
+  if (checkout.isPaseoOwnedWorktree) return "worktree";
+  return "checkout";
+}
+
+function workspaceName(entry: AgentEntry, id: string): string {
+  const name = entry.project.workspaceName?.trim();
+  if (name) return name;
+  const branch = entry.project.checkout.currentBranch?.trim();
+  if (branch && branch !== "HEAD") return branch;
+  return id.slice(id.lastIndexOf("/") + 1);
+}
+
+function gitRuntime(checkout: AgentEntry["project"]["checkout"]): Workspace["gitRuntime"] {
+  if (!checkout.isGit) return null;
+  return {
+    currentBranch: checkout.currentBranch,
+    remoteUrl: checkout.remoteUrl,
+    isPaseoOwnedWorktree: checkout.isPaseoOwnedWorktree,
+    isDirty: null,
+    aheadBehind: null,
+    aheadOfOrigin: null,
+    behindOfOrigin: null,
+  };
+}

packages/client/src/connection/legacy.ts

@@ -0,0 +1,282 @@
+import { SessionInboundMessageSchema } from "@getpaseo/protocol/messages";
+import { LegacyWorkspaces } from "./legacy-workspaces.js";
+import { BrowserAutomationHostCapabilitySchema } from "@getpaseo/protocol/browser-automation/capabilities";
+import type {
+  ServerInfoStatusPayload,
+  SessionInboundMessage,
+  SessionOutboundMessage,
+} from "@getpaseo/protocol/messages";
+
+export type ObservationRequest = { type: SessionInboundMessage["type"] } & Record<string, unknown>;
+interface Interest {
+  id: string;
+  query: ObservationRequest;
+}
+
+// COMPAT(ownedSubscriptions): added in v0.8.0; remove after 2027-03-11 once daemon floor >= v0.8.0.
+// Legacy directories retain their connection-wide, last-query-wins semantics.
+// IDs below identify local listeners, not independent server subscriptions.
+export class LegacySubscriptions {
+  private readonly interests = new Map<string, Interest>();
+
+  private readonly workspaces: LegacyWorkspaces | null;
+  constructor(
+    private readonly info: ServerInfoStatusPayload,
+    private readonly send: (message: ObservationRequest) => Promise<void>,
+    private readonly browserHost: unknown,
+  ) {
+    this.workspaces = info.features?.workspaceMultiplicity === true ? null : new LegacyWorkspaces();
+  }
+
+  normalize(message: SessionOutboundMessage): SessionOutboundMessage {
+    return this.workspaces?.normalize(message) ?? message;
+  }
+
+  prepareRequest(message: SessionInboundMessage): {
+    message: SessionInboundMessage;
+    receive(message: SessionOutboundMessage): SessionOutboundMessage;
+    finish(): Promise<void>;
+  } {
+    const identity = {
+      message,
+      receive: (value: SessionOutboundMessage) => value,
+      finish: async () => {},
+    };
+    const workspaces = this.workspaces;
+    if (workspaces && message.type === "fetch_workspaces_request") {
+      return {
+        ...identity,
+        message: SessionInboundMessageSchema.parse({
+          type: "fetch_agents_request",
+          requestId: message.requestId,
+          scope: "active",
+          sort: [{ key: "updated_at", direction: "desc" }],
+          page: message.page,
+          subscribe: message.subscribe,
+        }),
+        receive: (value) => {
+          if (value.type !== "fetch_agents_response") return value;
+          if (value.payload.requestId !== message.requestId) return value;
+          return {
+            type: "fetch_workspaces_response",
+            payload: {
+              ...value.payload,
+              entries: workspaces.read(value.payload.entries, !message.page?.cursor),
+              emptyProjects: [],
+            },
+          };
+        },
+      };
+    }
+    if (message.type === "subscribe_terminals_request") {
+      return {
+        ...identity,
+        receive: (value) => {
+          if (value.type !== "terminals_changed" || value.payload.cwd !== message.cwd) return value;
+          return { ...value, payload: { ...value.payload, requestId: message.requestId } };
+        },
+      };
+    }
+    if (message.type === "workspace.label.list.request" && !message.subscribe) {
+      return {
+        ...identity,
+        message: { ...message, subscribe: { subscriptionId: `legacy:${crypto.randomUUID()}` } },
+      };
+    }
+    if (message.type === "checkout.diff.get.request") {
+      const subscriptionId = `legacy:${crypto.randomUUID()}`;
+      return {
+        message: SessionInboundMessageSchema.parse({
+          ...message,
+          type: "subscribe_checkout_diff_request",
+          subscriptionId,
+        }),
+        receive: (value) => {
+          if (value.type !== "subscribe_checkout_diff_response") return value;
+          if (value.payload.requestId !== message.requestId) return value;
+          return { ...value, type: "checkout.diff.get.response" };
+        },
+        finish: () => this.send({ type: "unsubscribe_checkout_diff_request", subscriptionId }),
+      };
+    }
+    return identity;
+  }
+
+  start(query: ObservationRequest): Interest {
+    const interest = { id: `legacy:${crypto.randomUUID()}`, query };
+    this.interests.set(interest.id, interest);
+    return interest;
+  }
+
+  request({ id, query }: Interest): ObservationRequest | null {
+    switch (query.type) {
+      case "fetch_agents_request":
+      case "fetch_workspaces_request":
+      case "workspace.label.list.request":
+        return { ...query, subscribe: { subscriptionId: id } };
+      case "fs.file.subscribe.request":
+      case "subscribe_checkout_diff_request":
+        return { ...query, subscriptionId: id };
+      case "agent.timeline.set_subscription.request":
+        if (!this.info.features?.selectiveAgentTimeline) return null;
+        return { ...query, agentIds: this.members(query.type, "agentIds") };
+      case "session.events.set_subscription.request":
+        if (!this.info.features?.explicitEventSubscriptions) return null;
+        return {
+          ...query,
+          events: this.members(query.type, "events").filter(isLegacyEvent),
+        };
+      case "browser.host.register.request":
+        if (!matchesBrowserRegistration(this.browserHost, query)) {
+          throw new Error(
+            "This daemon requires browser_host registration in the client's hello capabilities",
+          );
+        }
+        return null; // Older hosts register in hello.
+      default:
+        return query;
+    }
+  }
+
+  private members(type: ObservationRequest["type"], field: string): string[] {
+    const members = new Set<string>();
+    for (const interest of this.interests.values()) {
+      if (interest.query.type !== type) continue;
+      for (const value of interest.query[field] as string[]) members.add(value);
+    }
+    return [...members].sort();
+  }
+
+  release(id: string): ObservationRequest | null {
+    const interest = this.interests.get(id);
+    if (!interest) return null;
+    this.interests.delete(id);
+    const { query } = interest;
+    switch (query.type) {
+      case "agent.timeline.set_subscription.request":
+      case "session.events.set_subscription.request":
+        return this.request(interest);
+      case "fs.file.subscribe.request":
+        return { type: "fs.file.unsubscribe.request", subscriptionId: id };
+      case "subscribe_checkout_diff_request":
+        return { type: "unsubscribe_checkout_diff_request", subscriptionId: id };
+      case "subscribe_terminal_request":
+        if (
+          [...this.interests.values()].some(
+            (item) => item.query.type === query.type && item.query.terminalId === query.terminalId,
+          )
+        )
+          return null;
+        return { type: "unsubscribe_terminal_request", terminalId: query.terminalId };
+      case "subscribe_terminals_request":
+        if (
+          [...this.interests.values()].some(
+            (item) =>
+              item.query.type === query.type &&
+              item.query.cwd === query.cwd &&
+              item.query.workspaceId === query.workspaceId,
+          )
+        )
+          return null;
+        return {
+          type: "unsubscribe_terminals_request",
+          cwd: query.cwd,
+          workspaceId: query.workspaceId,
+        };
+      default:
+        // Directory slots have no release RPC. Detach the local listener.
+        return null;
+    }
+  }
+
+  owns(message: SessionOutboundMessage): boolean {
+    return [...this.interests.values()].some((interest) => this.matches(interest, message));
+  }
+
+  receive(
+    message: SessionOutboundMessage,
+    deliver: (message: SessionOutboundMessage) => void,
+  ): void {
+    const updates: SessionOutboundMessage[] = [
+      message,
+      ...(this.workspaces?.update(message) ?? []),
+    ];
+    if (message.type === "agent_stream" && message.payload.event.type === "attention_required") {
+      const { agentId, event } = message.payload;
+      updates.push({
+        type: "agent_attention_required",
+        payload: {
+          agentId,
+          reason: event.reason,
+          timestamp: event.timestamp,
+          shouldNotify: event.shouldNotify,
+          ...(event.notification ? { notification: event.notification } : {}),
+        },
+      });
+    }
+    for (const update of updates)
+      for (const interest of this.interests.values()) {
+        if (!this.matches(interest, update)) continue;
+        // Normalize at the connection edge; all consumers receive the same handle shape.
+        deliver(
+          ("payload" in update
+            ? { ...update, payload: { ...update.payload, subscriptionId: interest.id } }
+            : { ...update, subscriptionId: interest.id }) as SessionOutboundMessage,
+        );
+      }
+  }
+
+  private matches({ id, query }: Interest, message: SessionOutboundMessage): boolean {
+    switch (query.type) {
+      case "fetch_agents_request":
+        return message.type === "agent_update";
+      case "fetch_workspaces_request":
+        return message.type === "workspace_update";
+      case "workspace.label.list.request":
+        return message.type === "workspace.label.update";
+      case "fs.file.subscribe.request":
+        return message.type === "fs.file.update" && message.payload.subscriptionId === id;
+      case "subscribe_checkout_diff_request":
+        return message.type === "checkout_diff_update" && message.payload.subscriptionId === id;
+      case "subscribe_terminals_request":
+        return message.type === "terminals_changed" && message.payload.cwd === query.cwd;
+      case "subscribe_terminal_request":
+        return (
+          message.type === "terminal_stream_exit" && message.payload.terminalId === query.terminalId
+        );
+      case "agent.timeline.set_subscription.request":
+        return (
+          (message.type === "agent_stream" || message.type === "agent.timeline.replacement") &&
+          (query.agentIds as string[]).includes(message.payload.agentId)
+        );
+      case "session.events.set_subscription.request":
+        return (query.events as string[]).includes(
+          message.type === "status" ? `status.${message.payload.status}` : message.type,
+        );
+      case "browser.host.register.request":
+        return message.type === "browser.automation.execute.request";
+      default:
+        return false;
+    }
+  }
+}
+
+function isLegacyEvent(event: string): boolean {
+  return [
+    "project.update",
+    "providers_snapshot_update",
+    "agent_attention_required",
+    "agent_permission_request",
+    "agent_permission_resolved",
+  ].includes(event);
+}
+
+function matchesBrowserRegistration(advertised: unknown, request: ObservationRequest): boolean {
+  const host = BrowserAutomationHostCapabilitySchema.safeParse(advertised);
+  if (!host.success) return false;
+  if (host.data.hostKind !== request.hostKind) return false;
+  for (const command of request.supportedCommands as string[]) {
+    if (!host.data.supportedCommands.some((supported) => supported === command)) return false;
+  }
+  return true;
+}

packages/client/src/connection/owned.ts

@@ -8,7 +8,7 @@ export interface SubscriptionObserver<T> {
   error?(error: unknown): void;
 }
 
-/** A stable local lifetime. The host ID changes when the transport reconnects. */
+/** A stable local lifetime. IDs change on reconnect; legacy IDs identify local listeners only. */
 export interface OwnedSubscription<T> {
   readonly subscriptionId: string | null;
   readonly ready: Promise<SubscriptionSnapshot<T>>;

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

@@ -328,7 +328,7 @@ test("Hub management requires daemon support before dispatching requests", async
   expect(mock.sent).toEqual([]);
 });
 
-test("owned timeline observation requires the host feature", async () => {
+test("timeline observation consumes broadcasts from a host without selective delivery", async () => {
   const mock = createMockTransport();
   const client = new DaemonClient({
     url: "ws://test",
@@ -341,7 +341,21 @@ test("owned timeline observation requires the host feature", async () => {
   mock.triggerOpen({ features: {} });
   await connecting;
   const observation = client.observeTimeline(["agent"]);
-  await expect(observation.ready).rejects.toThrow("Update the host");
+  await expect(observation.ready).resolves.toMatchObject({ agentIds: ["agent"] });
+  const updates: unknown[] = [];
+  observation.subscribe({ snapshot: () => {}, update: (message) => updates.push(message) });
+  mock.triggerMessage(
+    wrapSessionMessage({
+      type: "agent_stream",
+      payload: {
+        agentId: "agent",
+        timestamp: "2026-09-11T00:00:00Z",
+        event: { type: "turn_completed", provider: "codex" },
+      },
+    }),
+  );
+  expect(updates).toEqual([expect.objectContaining({ type: "agent_stream" })]);
+  await observation.release();
   expect(mock.sent).toEqual([]);
 });
 

packages/client/src/daemon-client.ts

@@ -1,7 +1,7 @@
 import { subscribeTimeline, type TimelineMessage } from "./timeline-subscription/index.js";
 import { ProviderSnapshotUpdates } from "./provider-snapshots/index.js";
 import {
-  OwnedSubscriptions,
+  ConnectionSubscriptions,
   type OwnedSubscription,
   DEFAULT_CLIENT_CAPABILITIES,
   type TimelineSubscription,
@@ -1094,7 +1094,11 @@ export class DaemonClient {
     emit: (message) => this.deliverSessionMessage(message),
     failed: (error) => this.logger.error({ err: error }, "Failed to resolve provider snapshot"),
   });
-  private readonly owned = new OwnedSubscriptions({
+  private readonly owned = new ConnectionSubscriptions({
+    send: (message) =>
+      this.sendSessionMessageOrThrow(
+        SessionInboundMessageSchema.parse({ ...message, requestId: this.createRequestId() }),
+      ),
     release: async (subscriptionId) => {
       if (!this.isConnected) return;
       try {
@@ -1731,6 +1735,7 @@ export class DaemonClient {
     select: (msg: SessionOutboundMessage) => T | null;
     options?: { skipQueue?: boolean };
   }): Promise<T> {
+    const wire = this.owned.prepareRequest(params.message);
     const timeout = params.timeout ?? DEFAULT_SESSION_RPC_TIMEOUT_MS;
     const { promise, cancel } = this.waitForWithCancel<RpcWaitResult<T>>(
       (msg) => {
@@ -1745,7 +1750,7 @@ export class DaemonClient {
             }),
           };
         }
-        const value = params.select(msg);
+        const value = params.select(wire.receive(msg));
         if (value === null) {
           return null;
         }
@@ -1756,7 +1761,7 @@ export class DaemonClient {
     );
 
     try {
-      await this.sendSessionMessageOrThrow(params.message);
+      await this.sendSessionMessageOrThrow(wire.message);
     } catch (error) {
       const err = error instanceof Error ? error : new Error(String(error));
       cancel(err);
@@ -1764,11 +1769,13 @@ export class DaemonClient {
       throw err;
     }
 
-    const result = await promise;
-    if (result.kind === "error") {
-      throw result.error;
+    try {
+      const result = await promise;
+      if (result.kind === "error") throw result.error;
+      return result.value;
+    } finally {
+      await wire.finish();
     }
-    return result.value;
   }
 
   private async sendCorrelatedRequest<
@@ -2098,23 +2105,18 @@ export class DaemonClient {
   // Agent RPCs (requestId-correlated)
   // ============================================================================
 
-  private requireOwnedSubscriptions(): void {
-    if (this.lastServerInfoMessage?.features?.ownedSubscriptions !== true)
-      throw new Error("Update the host to use independent subscriptions.");
-  }
-
   private observe<T extends CorrelatedResponseType>(
     responseType: T,
     message: { type: SessionInboundMessage["type"] } & Record<string, unknown>,
     options?: { requestId?: string; timeout?: number; signal?: AbortSignal },
   ): OwnedSubscription<CorrelatedResponsePayload<T>> {
     let resetSource = false;
-    return this.owned.observe<CorrelatedResponsePayload<T>>(
-      async (accept) => {
+    return this.owned.observeRequest<CorrelatedResponsePayload<T>>(
+      message,
+      async (query, accept) => {
         resetSource = false;
-        this.requireOwnedSubscriptions();
         const requestId = this.createRequestId(options?.requestId);
-        const request = SessionInboundMessageSchema.parse({ ...message, requestId });
+        const request = SessionInboundMessageSchema.parse({ ...query, requestId });
         try {
           return await this.sendCorrelatedRequest<T, CorrelatedResponsePayload<T>>({
             message: request,
@@ -3177,17 +3179,8 @@ export class DaemonClient {
     agentId: string,
     handler: (message: TimelineMessage) => void,
   ): TimelineSubscription {
-    return subscribeTimeline(
-      agentId,
-      this.observeTimeline([agentId]),
-      () =>
-        this.fetchAgentTimeline(agentId, {
-          direction: "before",
-          limit: 100,
-          projection: "projected",
-        }),
-      handler,
-      (error) => this.logger.error({ err: error }, "Timeline observation failed"),
+    return subscribeTimeline(agentId, this.observeTimeline([agentId]), handler, (error) =>
+      this.logger.error({ err: error }, "Timeline observation failed"),
     );
   }
 
@@ -3900,7 +3893,6 @@ export class DaemonClient {
     compare: { mode: "uncommitted" | "base"; baseRef?: string; ignoreWhitespace?: boolean },
     requestId?: string,
   ): Promise<CheckoutDiffPayload> {
-    this.requireOwnedSubscriptions();
     return this.sendCorrelatedSessionRequest({
       message: {
         type: "checkout.diff.get.request",
@@ -4550,8 +4542,13 @@ export class DaemonClient {
     unsubscribe: () => Promise<void>;
   }> {
     const subscription = this.observeFile(input);
+    let initial = true;
     subscription.subscribe({
-      snapshot: (snapshot) => onUpdate(snapshot.initial),
+      snapshot: (snapshot) => {
+        // The initial version is returned below. Subsequent snapshots repair reconnects.
+        if (!initial) onUpdate(snapshot.initial);
+        initial = false;
+      },
       update: (message) => {
         if (message.type === "fs.file.update") onUpdate(message.payload.version);
       },
@@ -6142,6 +6139,7 @@ export class DaemonClient {
       this.lastErrorValue = reason.trim();
     }
 
+    this.owned.disconnected();
     this.providerSnapshotUpdates.clear();
 
     // Clear all pending waiters and queued sends since the connection was lost
@@ -6242,6 +6240,7 @@ export class DaemonClient {
   }
 
   private handleSessionMessage(msg: SessionOutboundMessage): void {
+    msg = this.owned.normalize(msg);
     if (
       msg.type === "providers_snapshot_update" &&
       this.config.providerSnapshots !== "wire" &&
@@ -6269,7 +6268,10 @@ export class DaemonClient {
           this.reconnectAttempt = 0;
           this.updateConnectionState({ status: "connected" }, { event: "HELLO_SERVER_INFO" });
           this.startLivenessHeartbeat();
-          this.owned.restore();
+          this.owned.restore(serverInfo, {
+            ...DEFAULT_CLIENT_CAPABILITIES,
+            ...this.config.capabilities,
+          });
           this.flushPendingSendQueue();
           this.resolveConnect();
         }

packages/client/src/index.test.ts

@@ -102,9 +102,9 @@ async function connectClient(
     ownedSubscriptions: true,
   },
 ): Promise<{ client: PaseoClient; ws: FakeWebSocket }> {
-  vi.stubGlobal("WebSocket", FakeWebSocket);
   const client = createPaseoClient({
     url: "ws://daemon.test",
+    webSocketFactory: (url) => new FakeWebSocket(url),
     reconnect: { enabled: false },
   });
 
@@ -377,13 +377,15 @@ test("project actions list registered projects through the existing RPC", async
 
 test("project actions subscribe to existing project updates", async () => {
   const { client, ws } = await connectClient();
-  const observation = client.observeEvents(["project.update"]);
-  acknowledgeObservation(ws, "projects-sdk");
-  await observation.ready;
   const updates: string[] = [];
   const unsubscribe = client.projects.subscribe((update) => {
     updates.push(update.kind === "upsert" ? update.project.projectDisplayName : update.projectId);
   });
+  expect(parseSentSessionMessage(ws.sent.at(-1))).toMatchObject({
+    type: "session.events.set_subscription.request",
+    events: ["project.update"],
+  });
+  acknowledgeObservation(ws, "projects-sdk");
 
   ws.message(
     sessionMessage({
@@ -1367,15 +1369,17 @@ test("provider actions delegate to existing provider RPCs and local snapshot upd
     ],
   });
 
-  const observation = client.observeEvents(["providers_snapshot_update"]);
-  acknowledgeObservation(ws, "provider-updates");
-  await observation.ready;
   const snapshotUpdates: string[] = [];
   const snapshotModelDefaults: Array<string | undefined> = [];
   const unsubscribe = client.providers.subscribe((update) => {
     snapshotUpdates.push(update.generatedAt);
     snapshotModelDefaults.push(update.entries[0]?.models?.[0]?.defaultThinkingOptionId);
   });
+  expect(parseSentSessionMessage(ws.sent.at(-1))).toMatchObject({
+    type: "session.events.set_subscription.request",
+    events: ["providers_snapshot_update"],
+  });
+  acknowledgeObservation(ws, "provider-updates");
   ws.message(
     sessionMessage({
       type: "providers_snapshot_update",
@@ -1408,15 +1412,25 @@ test("provider actions delegate to existing provider RPCs and local snapshot upd
   await client.close();
 });
 
-test("waitForReady requires owned subscriptions from the host", async () => {
+test("waitForReady reads an older host on the existing connection", async () => {
   const { client, ws } = await connectClient({});
-  const sentBeforeWait = ws.sent.length;
-
-  await expect(client.providers.waitForReady({ cwd: "/repo/./sdk" })).rejects.toThrow(
-    "Update the host to use independent subscriptions.",
+  const waiting = client.providers.waitForReady({ cwd: "/repo/./sdk" });
+  await expect
+    .poll(() => parseSentSessionMessage(ws.sent.at(-1)).type)
+    .toBe("get_providers_snapshot_request");
+  const request = parseSentSessionMessage(ws.sent.at(-1));
+  ws.message(
+    sessionMessage({
+      type: "get_providers_snapshot_response",
+      payload: {
+        requestId: request.requestId,
+        entries: [],
+        generatedAt: "2026-09-11T00:00:00.000Z",
+      },
+    }),
   );
-  expect(ws.sent).toHaveLength(sentBeforeWait);
-
+  await expect(waiting).resolves.toMatchObject({ entries: [] });
+  expect(FakeWebSocket.instances).toHaveLength(1);
   await client.close();
 });
 

packages/client/src/index.ts

@@ -303,7 +303,7 @@ export type PaseoAgentTimelineEvent =
   | {
       agentId: string;
       subscriptionId: string;
-      event: { type: "snapshot"; reason: "reconnect"; page: FetchAgentTimelinePayload };
+      event: { type: "subscription_restored" };
     }
   | { agentId: string; event: { type: "error"; error: string } };
 
@@ -318,10 +318,10 @@ export interface PaseoAgentTimelineHandle {
    */
   refetch(options?: PaseoAgentTimelineRefetchOptions): Promise<FetchAgentTimelinePayload>;
   /**
-   * Initially delivers live events only. Reconnect delivers a projected snapshot
-   * of the latest 100 entries before subsequent updates. Replace your recent view
-   * with this page; use its cursors to load older history. A replacement event
-   * invalidates the previous epoch. Recovery errors release this observation.
+   * Delivers live events only. After reconnect, subscription_restored precedes
+   * subsequent updates. History may have been missed; use refetch() to request
+   * the range you need. No history is fetched automatically. A replacement event
+   * invalidates the previous epoch. Subscription errors release this observation.
    * Await the returned unsubscribe function's `ready` promise before starting
    * work that must be observed. It rejects if establishment fails.
    */
@@ -520,8 +520,6 @@ export function createPaseoApi(
   const handles = new Set<{ release(): Promise<void> }>();
   const agentListeners = new Set<PaseoAgentUpdateHandler>();
   const workspaceListeners = new Set<PaseoWorkspaceUpdateHandler>();
-  const projectListeners = new Set<PaseoProjectUpdateHandler>();
-  const providerListeners = new Set<(update: PaseoProviderSnapshotUpdate) => void>();
   const lifetime = new AbortController();
   const own = <T extends { release(): Promise<void> }>(create: () => T): T => {
     if (lifetime.signal.aborted) throw new Error("Paseo API is disposed");
@@ -598,8 +596,6 @@ export function createPaseoApi(
     scopeOptions?.signal?.removeEventListener("abort", abort);
     agentListeners.clear();
     workspaceListeners.clear();
-    projectListeners.clear();
-    providerListeners.clear();
     disposal = Promise.allSettled([...handles].map((handle) => handle.release())).then(
       (results) => {
         handles.clear();
@@ -619,18 +615,20 @@ export function createPaseoApi(
   if (scopeOptions?.signal?.aborted) abort();
   else scopeOptions?.signal?.addEventListener("abort", abort, { once: true });
 
-  const observeEvents: DaemonClient["observeEvents"] = (events, options) => {
-    const subscription = own(() => daemonClient.observeEvents(events, options));
-    subscription.subscribe({
-      snapshot: () => {},
-      update: (message) => {
-        if (message.type === "project.update")
-          for (const listener of projectListeners) listener(message.payload);
-        if (message.type === "providers_snapshot_update")
-          for (const listener of providerListeners) listener(message.payload);
-      },
-    });
-    return subscription;
+  const observeEvents: DaemonClient["observeEvents"] = (events, options) =>
+    own(() => daemonClient.observeEvents(events, options));
+
+  const subscribeEvent = (
+    event: "project.update" | "providers_snapshot_update",
+    update: (message: SessionOutboundMessage) => void,
+  ): (() => void) => {
+    const observation = observeEvents([event]);
+    observation.subscribe({ snapshot: () => {}, update });
+    return () => {
+      void observation
+        .release()
+        .catch((error) => console.error("Event subscription cleanup failed", error));
+    };
   };
 
   function listWorkspaces(options: PaseoWorkspaceListOptions & { subscribe: {} }): Promise<
@@ -686,11 +684,9 @@ export function createPaseoApi(
     projects: {
       list: (options) => daemonClient.listProjects(options),
       subscribe: (handler) => {
-        if (lifetime.signal.aborted) throw new Error("Paseo API is disposed");
-        projectListeners.add(handler);
-        return () => {
-          projectListeners.delete(handler);
-        };
+        return subscribeEvent("project.update", (message) => {
+          if (message.type === "project.update") handler(message.payload);
+        });
       },
     },
     workspaces: {
@@ -735,11 +731,9 @@ export function createPaseoApi(
       diagnostic: (provider, options) => daemonClient.getProviderDiagnostic(provider, options),
       listUsage: (options) => listProviderUsage(daemonClient, options),
       subscribe: (handler) => {
-        if (lifetime.signal.aborted) throw new Error("Paseo API is disposed");
-        providerListeners.add(handler);
-        return () => {
-          providerListeners.delete(handler);
-        };
+        return subscribeEvent("providers_snapshot_update", (message) => {
+          if (message.type === "providers_snapshot_update") handler(message.payload);
+        });
       },
     },
     config: {
@@ -865,11 +859,11 @@ function createAgentHandleFactory(
             switch (message.type) {
               case "agent_stream":
                 return handler(message.payload);
-              case "agent.timeline.snapshot":
+              case "agent.timeline.subscription_restored":
                 return handler({
                   agentId: id,
                   subscriptionId: message.payload.subscriptionId,
-                  event: { type: "snapshot", reason: "reconnect", page: message.payload.page },
+                  event: { type: "subscription_restored" },
                 });
               case "agent.timeline.error":
                 return handler({

packages/client/src/terminal-stream-router.test.ts

@@ -44,3 +44,20 @@ test("two registrations for one terminal retain independent slots and release",
   expect(a).toHaveLength(1);
   expect(b).toHaveLength(2);
 });
+
+test("legacy listeners share a terminal slot without releasing each other", () => {
+  const router = new TerminalStreamRouter();
+  const received: string[] = [];
+  const release = router.attach("first", "terminal", 1, () => received.push("first"));
+  router.attach("second", "terminal", 1, () => received.push("second"));
+  const frame = {
+    opcode: TerminalStreamOpcode.Output,
+    slot: 1,
+    payload: new TextEncoder().encode("bytes"),
+  };
+  router.handleFrame(frame);
+  expect(received).toEqual(["first", "second"]);
+  release();
+  router.handleFrame(frame);
+  expect(received).toEqual(["first", "second", "second"]);
+});

packages/client/src/terminal-stream-router.ts

@@ -14,11 +14,11 @@ export type TerminalStreamEvent = { terminalId: string; subscriptionId: string }
 export class TerminalStreamRouter {
   private readonly slots = new Map<
     number,
-    {
+    Set<{
       subscriptionId: string;
       terminalId: string;
       receive: (event: TerminalStreamEvent) => void;
-    }
+    }>
   >();
   private readonly listeners = new Set<(event: TerminalStreamEvent) => void>();
 
@@ -37,9 +37,12 @@ export class TerminalStreamRouter {
     receive: (event: TerminalStreamEvent) => void,
   ): () => void {
     const registration = { subscriptionId, terminalId, receive };
-    this.slots.set(slot, registration);
+    const registrations = this.slots.get(slot) ?? new Set();
+    registrations.add(registration);
+    this.slots.set(slot, registrations);
     return () => {
-      if (this.slots.get(slot) === registration) this.slots.delete(slot);
+      registrations.delete(registration);
+      if (registrations.size === 0) this.slots.delete(slot);
     };
   }
 
@@ -48,28 +51,30 @@ export class TerminalStreamRouter {
   }
 
   handleFrame(frame: TerminalStreamFrame): void {
-    const registration = this.slots.get(frame.slot);
-    if (!registration) return;
-    const identity = {
-      subscriptionId: registration.subscriptionId,
-      terminalId: registration.terminalId,
-    };
-    let event: TerminalStreamEvent;
-    if (frame.opcode === TerminalStreamOpcode.Output)
-      event = { ...identity, type: "output", data: frame.payload };
-    else if (frame.opcode === TerminalStreamOpcode.Restore)
-      event = { ...identity, type: "restore", data: frame.payload };
-    else if (frame.opcode === TerminalStreamOpcode.Snapshot) {
-      const state = decodeTerminalSnapshotPayload(frame.payload);
-      if (!state) return;
-      event = { ...identity, type: "snapshot", state };
-    } else return;
-    registration.receive(event);
-    for (const listener of this.listeners) {
-      try {
-        listener(event);
-      } catch {
-        /* Passive diagnostics cannot interrupt delivery. */
+    const registrations = this.slots.get(frame.slot);
+    if (!registrations) return;
+    for (const registration of registrations) {
+      const identity = {
+        subscriptionId: registration.subscriptionId,
+        terminalId: registration.terminalId,
+      };
+      let event: TerminalStreamEvent;
+      if (frame.opcode === TerminalStreamOpcode.Output)
+        event = { ...identity, type: "output", data: frame.payload };
+      else if (frame.opcode === TerminalStreamOpcode.Restore)
+        event = { ...identity, type: "restore", data: frame.payload };
+      else if (frame.opcode === TerminalStreamOpcode.Snapshot) {
+        const state = decodeTerminalSnapshotPayload(frame.payload);
+        if (!state) return;
+        event = { ...identity, type: "snapshot", state };
+      } else return;
+      registration.receive(event);
+      for (const listener of this.listeners) {
+        try {
+          listener(event);
+        } catch {
+          /* Passive diagnostics cannot interrupt delivery. */
+        }
       }
     }
   }

packages/client/src/timeline-subscription/index.ts

@@ -1,5 +1,4 @@
 import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
-import type { FetchAgentTimelinePayload } from "../daemon-client.js";
 import type { OwnedSubscription, TimelineSubscription } from "../connection/index.js";
 
 type TimelineUpdate = Extract<
@@ -7,36 +6,24 @@ type TimelineUpdate = Extract<
   { type: "agent_stream" | "agent.timeline.replacement" }
 >;
 
-/** Local SDK recovery messages; these are not additional wire RPCs. */
+/** Local SDK lifecycle messages; these are not additional wire RPCs. */
 export type TimelineMessage =
   | TimelineUpdate
   | {
-      type: "agent.timeline.snapshot";
-      payload: { agentId: string; subscriptionId: string; page: FetchAgentTimelinePayload };
+      type: "agent.timeline.subscription_restored";
+      payload: { agentId: string; subscriptionId: string };
     }
   | { type: "agent.timeline.error"; payload: { agentId: string; error: string } };
 
-const MAX_RECOVERY_UPDATES = 128;
-
-/** A live-only initial observation, with a fresh bounded history view after each outage. */
+/** Restores live delivery; the consumer owns history reads and gap recovery. */
 export function subscribeTimeline(
   agentId: string,
   observation: OwnedSubscription<{ agentIds: string[] }>,
-  readSnapshot: () => Promise<FetchAgentTimelinePayload>,
   handler: (message: TimelineMessage) => void,
   reportError: (error: unknown) => void,
 ): TimelineSubscription {
   let established = false;
   let released = false;
-  interface Recovery {
-    id: string;
-    revision: number;
-    updates: TimelineUpdate[];
-    replacement?: Extract<TimelineUpdate, { type: "agent.timeline.replacement" }>;
-  }
-  let recovery: Recovery | null = null;
-  const current = (state: Recovery) =>
-    !released && recovery === state && observation.subscriptionId === state.id;
   const notify = (message: TimelineMessage) => {
     if (released) return;
     try {
@@ -47,7 +34,6 @@ export function subscribeTimeline(
   };
   const release = () => {
     released = true;
-    recovery = null;
     return observation.release();
   };
   const stop = () => {
@@ -61,65 +47,20 @@ export function subscribeTimeline(
     });
     stop();
   };
-  const recover = async (state: Recovery) => {
-    while (current(state)) {
-      const revision = state.revision;
-      let page: FetchAgentTimelinePayload;
-      try {
-        page = await readSnapshot();
-      } catch (error) {
-        if (current(state)) fail(error);
-        return;
-      }
-      if (!current(state)) return;
-      // Overflow or replacement makes this in-flight read obsolete. Read a fresh
-      // bounded page instead of retaining an unbounded backlog of transient events.
-      if (revision !== state.revision) continue;
-      if (state.replacement) notify(state.replacement);
-      if (!current(state)) return;
-      notify({
-        type: "agent.timeline.snapshot",
-        payload: { agentId, subscriptionId: state.id, page },
-      });
-      if (!current(state)) return;
-      recovery = null;
-      for (const message of state.updates) {
-        if (released || observation.subscriptionId !== state.id) break;
-        if (message.type === "agent_stream" && typeof message.payload.seq === "number") {
-          if (
-            message.payload.epoch !== page.epoch ||
-            message.payload.seq <= (page.window.maxSeq ?? 0)
-          )
-            continue;
-        }
-        notify(message);
-      }
-    }
-  };
   observation.subscribe({
     snapshot: ({ subscriptionId }) => {
       if (!established) {
         established = true;
         return;
       }
-      const state: Recovery = { id: subscriptionId, revision: 0, updates: [] };
-      recovery = state;
-      void recover(state);
+      notify({
+        type: "agent.timeline.subscription_restored",
+        payload: { agentId, subscriptionId },
+      });
     },
     update: (message) => {
-      if (message.type !== "agent_stream" && message.type !== "agent.timeline.replacement") return;
-      if (!recovery) return notify(message);
-      if (message.type === "agent.timeline.replacement") {
-        recovery.replacement = message;
-        recovery.revision++;
-        recovery.updates.length = 0;
-      } else {
-        if (recovery.updates.length === MAX_RECOVERY_UPDATES) {
-          recovery.revision++;
-          recovery.updates.length = 0;
-        }
-        recovery.updates.push(message);
-      }
+      if (message.type === "agent_stream" || message.type === "agent.timeline.replacement")
+        notify(message);
     },
     error: fail,
   });

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

@@ -104,13 +104,13 @@ test("public SDK scope cancellation during bootstrap releases the returned ID",
   }
 });
 
-test("a surviving public SDK timeline recovers persistent rows written during its socket outage", async () => {
+test("a restored public SDK timeline leaves missed history to the consumer", async () => {
   const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
   const sockets: WebSocket[] = [];
   const makeClient = (observer: boolean) =>
     new PublicDaemonClient({
       url: `ws://127.0.0.1:${daemon.port}/ws`,
-      clientId: observer ? "r5-observer" : "r5-actor",
+      clientId: observer ? "timeline-observer" : "timeline-actor",
       appVersion: "0.8.0",
       reconnect: { enabled: false },
       webSocketFactory: (url) => {
@@ -136,28 +136,32 @@ test("a surviving public SDK timeline recovers persistent rows written during it
     const handle = api.agents.ref(agent.id);
     const subscription = handle.timeline.subscribe((event) => received.push(event));
     await subscription.ready;
-    await actor.sendMessage(agent.id, "before-outage-r5");
+    await actor.sendMessage(agent.id, "before-outage");
     await actor.waitForFinish(agent.id);
-    await expect.poll(() => JSON.stringify(received)).toContain("before-outage-r5");
+    await expect.poll(() => JSON.stringify(received)).toContain("before-outage");
+    const cached = await handle.timeline.refetch();
+    expect(cached.endCursor).not.toBeNull();
     const oldId = subscription.subscriptionId;
     sockets[0]!.terminate();
     await expect.poll(() => observer.getConnectionState().status).not.toBe("connected");
-    await actor.sendMessage(agent.id, "during-outage-r5");
+    await actor.sendMessage(agent.id, "during-outage");
     await actor.waitForFinish(agent.id);
     await observer.connect();
     await expect
       .poll(() => subscription.subscriptionId)
       .toSatisfy((id: unknown) => typeof id === "string" && id !== oldId);
-    await actor.sendMessage(agent.id, "after-outage-r5");
+    await expect.poll(() => JSON.stringify(received)).toContain("subscription_restored");
+    expect(JSON.stringify(received)).not.toContain("during-outage");
+    await actor.sendMessage(agent.id, "after-outage");
     await actor.waitForFinish(agent.id);
-    await expect.poll(() => JSON.stringify(received)).toContain("after-outage-r5");
+    await expect.poll(() => JSON.stringify(received)).toContain("after-outage");
     const history = await handle.timeline.refetch({
-      direction: "before",
-      limit: 100,
+      direction: "after",
+      cursor: cached.endCursor!,
       projection: "projected",
     });
-    expect(JSON.stringify(history)).toContain("during-outage-r5");
-    expect(JSON.stringify(received)).toContain("during-outage-r5");
+    expect(JSON.stringify(history)).toContain("during-outage");
+    expect(JSON.stringify(received)).not.toContain("during-outage");
   } finally {
     await api.dispose();
     await observer.close();

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

@@ -1,4 +1,5 @@
 import { MockLoadTestAgentClient } from "./agent/providers/mock-load-test-agent.js";
+import { createPaseoApi } from "@getpaseo/client";
 import { execFileSync } from "node:child_process";
 import { mkdir, writeFile } from "node:fs/promises";
 import path from "node:path";
@@ -952,7 +953,7 @@ test("browser capability advertisement is passive and explicit hosts own command
   }
 });
 
-test("adding public SDK listeners does not create server observation demand", async () => {
+test("adding raw client listeners does not create server observation demand", async () => {
   const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
   const client = new DaemonClient({
     url: `ws://127.0.0.1:${daemon.port}/ws`,
@@ -1672,3 +1673,83 @@ test("mark unread replies remain source-owned while directory observers receive
     await daemon.close();
   }
 });
+
+test("legacy event subscribers retain notifications without the new notifications flag", async () => {
+  const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
+  const admin = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, appVersion: "0.8.0" });
+  let legacy: SubscriptionPeer | undefined;
+  try {
+    await admin.connect();
+    const workspace = (
+      await admin.createWorkspace({ source: { kind: "directory", path: daemon.staticDir } })
+    ).workspace!;
+    const agent = await admin.createAgent({
+      provider: "codex",
+      cwd: daemon.staticDir,
+      workspaceId: workspace.id,
+    });
+    legacy = await SubscriptionPeer.connect(daemon.port, "legacy-notifications", {
+      owned_subscriptions: false,
+      selective_agent_timeline: true,
+      explicit_event_subscriptions: true,
+    });
+    await legacy.request({
+      type: "fetch_agents_request",
+      requestId: "directory",
+      subscribe: { subscriptionId: "directory" },
+    });
+    await legacy.request({
+      type: "session.events.set_subscription.request",
+      requestId: "events",
+      events: ["agent_attention_required"],
+    });
+    legacy.send({
+      type: "client_heartbeat",
+      deviceType: "web",
+      focusedAgentId: null,
+      lastActivityAt: new Date().toISOString(),
+      appVisible: true,
+    });
+    await legacy.request({ type: "ping", requestId: "barrier", clientSentAt: 1 });
+    const peer = legacy;
+    await admin.sendMessage(agent.id, "Finish a legacy notification turn");
+    await expect
+      .poll(
+        () =>
+          peer.frames.flatMap((frame) =>
+            frame.type === "session" && frame.message.type === "agent_attention_required"
+              ? [frame.message.payload.shouldNotify]
+              : [],
+          ),
+        { timeout: 10_000 },
+      )
+      .toEqual([true]);
+  } finally {
+    legacy?.close();
+    await admin.close();
+    await daemon.close();
+  }
+});
+
+test("public project subscriptions request updates and release their producer demand", async () => {
+  const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
+  const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  const api = createPaseoApi(client);
+  try {
+    await client.connect();
+    const updates: unknown[] = [];
+    const unsubscribe = api.projects.subscribe((update) => updates.push(update));
+    await client.createWorkspace({ source: { kind: "directory", path: daemon.staticDir } });
+    await expect
+      .poll(() => updates)
+      .toEqual(expect.arrayContaining([expect.objectContaining({ kind: "upsert" })]));
+    unsubscribe();
+    await expect
+      .poll(async () => (await client.collectDiagnostics()).diagnostic)
+      .toContain("Registrations: 0");
+  } finally {
+    await api.dispose();
+    await client.close();
+    await daemon.close();
+  }
+});

packages/server/src/terminal/terminal-session-controller.test.ts

@@ -1143,7 +1143,7 @@ describe("terminal output before natural exit", () => {
 });
 
 describe("terminal snapshot failure", () => {
-  test.each(["initial", "legacy initial", "backpressure"])(
+  test.each(["initial", "backpressure"])(
     "releases only the failed observer after %s rejection",
     async (phase) => {
       vi.useFakeTimers();
@@ -1155,17 +1155,13 @@ describe("terminal snapshot failure", () => {
       }, phase === "backpressure");
       try {
         const source = {};
-        const failed = await f.subscribe(source, phase !== "legacy initial", "failed");
+        const failed = await f.subscribe(source, true, "failed");
         await vi.advanceTimersByTimeAsync(0);
         if (phase === "backpressure") {
           f.output("x".repeat(300 * 1024), 1);
           await vi.advanceTimersByTimeAsync(5);
         }
-        const healthy = await f.subscribe(
-          phase === "legacy initial" ? {} : source,
-          phase !== "legacy initial",
-          "healthy",
-        );
+        const healthy = await f.subscribe(source, true, "healthy");
         await vi.advanceTimersByTimeAsync(0);
         f.output("SIBLING-ALIVE", 2);
         await vi.advanceTimersByTimeAsync(5);
@@ -1182,7 +1178,7 @@ describe("terminal snapshot failure", () => {
             type: "terminal_stream_exit",
             payload: {
               terminalId: "exit-terminal",
-              ...(phase === "legacy initial" ? {} : { subscriptionId: failed.id }),
+              subscriptionId: failed.id,
               error: "Terminal worker request timed out: getTerminalState",
             },
           },
@@ -1201,7 +1197,7 @@ describe("terminal snapshot failure", () => {
           type: "terminal_stream_exit",
           payload: {
             terminalId: "exit-terminal",
-            ...(phase === "legacy initial" ? {} : { subscriptionId: healthy.id }),
+            subscriptionId: healthy.id,
           },
         });
         expect(f.delivery.registrationCount).toBe(0);
@@ -1238,3 +1234,45 @@ describe("terminal snapshot failure", () => {
     }
   });
 });
+
+test.each([false, true])(
+  "legacy snapshot failure retries without reporting a live PTY exited (backpressure=%s)",
+  async (backpressure) => {
+    vi.useFakeTimers();
+    let reads = 0;
+    const f = exitFixture(async () => {
+      if (++reads === (backpressure ? 2 : 1)) throw new Error("Snapshot unavailable");
+      return { state: terminalState("recovered"), revision: 0 };
+    }, backpressure);
+    try {
+      await f.subscribe({}, false, "legacy-recovery");
+      await vi.advanceTimersByTimeAsync(0);
+      if (backpressure) {
+        f.output("x".repeat(300 * 1024), 1);
+        await vi.advanceTimersByTimeAsync(5);
+      }
+      expect(f.frames.filter((row) => row.message?.type === "terminal_stream_exit")).toEqual([]);
+      f.output("RECOVERED-OUTPUT", 2);
+      for (const listener of f.outputs) listener({ type: "snapshotReady", revision: 2 });
+      await vi.advanceTimersByTimeAsync(5);
+      expect(reads).toBe(backpressure ? 3 : 2);
+      expect(f.frames.some((row) => row.binary?.opcode === TerminalStreamOpcode.Snapshot)).toBe(
+        true,
+      );
+      f.exit();
+      await vi.advanceTimersByTimeAsync(0);
+      expect(f.frames.filter((row) => row.message?.type === "terminal_stream_exit")).toHaveLength(
+        1,
+      );
+      expect(f.frames.at(-1)?.message).toEqual({
+        type: "terminal_stream_exit",
+        payload: { terminalId: "exit-terminal" },
+      });
+      expect(f.delivery.registrationCount).toBe(0);
+      expect(f.outputs.size + f.exits.size).toBe(0);
+    } finally {
+      await f.delivery.close();
+      vi.useRealTimers();
+    }
+  },
+);

packages/server/src/terminal/terminal-session-controller.ts

@@ -66,6 +66,7 @@ interface ActiveTerminalStream {
   unsubscribe: () => void;
   needsSnapshot: boolean;
   snapshotInFlight: boolean;
+  retrySnapshotErrors: boolean;
   readyRevision?: number;
   restore?: TerminalRestoreOptions;
   bufferedOutputs: BufferedTerminalOutput[];
@@ -706,7 +707,10 @@ export class TerminalSessionController {
       });
     }
 
-    const slot = this.bindActiveStream(session, owner, { restore: msg.restore });
+    const slot = this.bindActiveStream(session, owner, {
+      restore: msg.restore,
+      retrySnapshotErrors: !ownership.isModern(owner.source),
+    });
     if (slot === null) {
       await owner.release();
       this.sessionLogger.warn(
@@ -852,7 +856,7 @@ export class TerminalSessionController {
   private bindActiveStream(
     terminal: TerminalSession,
     owner: OwnedSubscription,
-    options?: { restore?: TerminalRestoreOptions },
+    options: { restore?: TerminalRestoreOptions; retrySnapshotErrors: boolean },
   ): number | null {
     if (!this.hasBinaryChannel()) {
       return null;
@@ -870,6 +874,7 @@ export class TerminalSessionController {
       unsubscribe: () => {},
       needsSnapshot: true,
       snapshotInFlight: false,
+      retrySnapshotErrors: options.retrySnapshotErrors,
       exiting: false,
       readyRevision: undefined,
       restore: options?.restore,
@@ -1010,6 +1015,14 @@ export class TerminalSessionController {
       );
       // Natural completion owns the buffered final bytes and waits for this read.
       if (activeStream.exiting) return;
+      // COMPAT(terminalSnapshotErrors): restored in v0.8.0; remove after 2027-03-11 once client floor >= v0.8.0.
+      // Old clients interpret every stream exit as PTY exit. Preserve their
+      // attached stream after failure. Another snapshot notification can retry;
+      // this does not schedule recovery. Source teardown still releases the stream.
+      if (activeStream.retrySnapshotErrors) {
+        activeStream.needsSnapshot = true;
+        return;
+      }
       activeStream.owner.emit({
         type: "terminal_stream_exit",
         payload: {

public-docs/sdk/events.md

@@ -8,7 +8,7 @@ category: TypeScript SDK
 
 # SDK events
 
-Use an owned subscription to fetch a snapshot and follow its changes. Connecting, plain reads, and local listeners do not start observation. Each observation gets a new server-issued ID, even when its filter matches another observation.
+Use an owned subscription to fetch a snapshot and follow its changes. On capable daemons, connecting, plain reads, and local directory listeners do not start observation. Each observation gets a new server-issued ID, even when its filter matches another observation.
 
 ## Follow one agent's status
 
@@ -47,10 +47,9 @@ await directory.subscription.release();
 ```ts
 const unsubscribe = agent.timeline.subscribe((update) => {
   const { event } = update;
-  if (event.type === "snapshot") {
-    // Replace your recent history view; these are complete projected entries,
-    // not additional live message fragments.
-    console.log("Reconnected", event.page.epoch, event.page.entries);
+  if (event.type === "subscription_restored") {
+    // Live delivery has resumed. Choose whether to fetch missed history.
+    console.log("Reconnected; events may have been missed");
     return;
   }
   if (event.type === "error") {
@@ -72,11 +71,13 @@ const unsubscribe = agent.timeline.subscribe((update) => {
 });
 ```
 
-Await `unsubscribe.ready` before starting work whose events you need to observe. It acknowledges the initial live subscription; initial history is a separate [read](#fetch-timeline-history). Call `unsubscribe()` to release demand, or await `unsubscribe.release()` for teardown.
+Await `unsubscribe.ready` before starting work whose events you need to observe. It waits for the initial live subscription (local listener attachment on broadcast-only hosts); initial history is a separate [read](#fetch-timeline-history). Call `unsubscribe()` to release demand, or await `unsubscribe.release()` for teardown.
 
-After reconnect, the same handle receives `{ agentId, subscriptionId, event: { type: "snapshot", reason: "reconnect", page } }` before subsequent updates. The host assigns a fresh subscription ID. `page` is the latest 100 projected history entries, with the same epoch, cursors and paging flags as `timeline.refetch()`. Replace your recent view with this page; do not concatenate it with earlier message fragments. If `hasOlder` is true, fetch older pages with `page.startCursor`. Compare epochs before retaining any previously loaded history: a reconnect may preserve or replace the epoch. A live `replacement` event still means the previous epoch is invalid.
+After reconnect, the same handle receives `{ agentId, subscriptionId, event: { type: "subscription_restored" } }` before subsequent live updates. This is a local SDK notification after membership acknowledgement (local attachment on broadcast-only hosts). The subscription gets a fresh ID. No history is fetched automatically, and missed events are not replayed.
 
-Persisted history is recovered; transient events such as an offline turn-completion notification are not replayed. Recovery buffers at most 128 current-connection updates. Overflow or a concurrent replacement discards the obsolete buffer and reads a fresh page. A recovery read failure delivers `{ agentId, event: { type: "error", error } }` and releases that observation; establish a new subscription when ready to retry. Release suppresses pending recovery callbacks. Each same-agent subscription recovers independently.
+Choose recovery for your consumer: continue live, request a recent page, or call `timeline.refetch({ direction: "after", cursor: { epoch, seq } })` using your saved position. Live delivery continues while your read is pending; buffer or reconcile those events with the returned page by epoch and sequence. Follow `hasNewer` and `endCursor` to read further pages when needed. A live `replacement` invalidates the previous epoch. A failed explicit history read rejects that read and leaves the live subscription active.
+
+Subscription establishment failures deliver `{ agentId, event: { type: "error", error } }` and release the observation. Establish a new subscription when ready to retry. Releasing a subscription stops its callbacks and prevents restoration after reconnect.
 
 Assistant messages can arrive in pieces. Concatenate their text when you need a complete message, or use `run()` and read `lastMessage` when you only need the final reply.
 
@@ -118,7 +119,6 @@ await directory.subscription.release();
 ## Follow provider catalog changes
 
 ```ts
-const observation = client.observeEvents(["providers_snapshot_update"]);
 const unsubscribe = client.providers.subscribe((update) => {
   const ready = update.entries.filter((entry) => entry.status === "ready");
   console.log(
@@ -126,14 +126,19 @@ const unsubscribe = client.providers.subscribe((update) => {
     ready.map((entry) => entry.provider),
   );
 });
-await observation.ready;
 console.log(await client.providers.snapshot());
 
 // When this view closes:
 unsubscribe();
-await observation.release();
 ```
 
-Project listeners work the same way: request `client.observeEvents(["project.update"])` before relying on `client.projects.subscribe()`. For an initial project cache, buffer updates while awaiting `client.projects.list()`, then apply them after the snapshot.
+Provider and project `subscribe()` calls request their own updates and release that demand on unsubscribe. For an initial project cache, buffer updates while awaiting `client.projects.list()`, then apply them after the snapshot.
 
-Call `client.close()` when the application no longer needs the connection. Observation requires a host that advertises independent subscriptions; an older host returns an update-host error. Plain reads remain available.
+Call `client.close()` when the application no longer needs the connection.
+
+## Older daemons
+
+The same methods use the existing connection and legacy delivery behavior. Old directory subscriptions
+share a server slot: a later filtered observation replaces that slot's filter. Handles have local IDs
+and separate callbacks, but independent server filters require an updated daemon. Releasing a handle
+detaches its callbacks; old daemons may continue sending broadcasts.

public-docs/sdk/reference.md

@@ -55,7 +55,7 @@ Create a new client after `close()`.
 | `ref(agentOrId)`     | `PaseoAgentHandle`     | Creates a local handle without fetching.                                                                     |
 | `subscribe(handler)` | Unsubscribe function   | Local listener for this API instance. Requires an owned `list({ subscribe: {} })` observation.               |
 
-`list({ subscribe: {} })` also returns a server-issued `subscriptionId` and an owned `subscription`. Its `subscribe({ snapshot, update, error? })` callbacks receive the snapshot before scoped wire updates; `release()` ends that observation. Plain lists create no observation. The same contract applies to workspace lists. See [events](./events.md).
+`list({ subscribe: {} })` also returns a `subscriptionId` and an owned `subscription`. Its `subscribe({ snapshot, update, error? })` callbacks receive the snapshot before scoped wire updates; `release()` ends that observation. Capable daemons assign the ID and keep observations independent. Older daemons use local IDs and their established shared delivery behavior. Plain lists create no observation. The same contract applies to workspace lists. See [events](./events.md).
 
 Creation options include `config`, `cwd`, `parent`, `title`, `prompt`, `env`, `outputSchema`, `images`, `attachments`, `git`, `worktree`, `autoArchive`, and `labels`.
 
@@ -110,14 +110,14 @@ Creation options include `config`, `cwd`, `parent`, `title`, `prompt`, `env`, `o
 
 `agent.timeline.refetch(options?)` fetches a page. Options are `direction`, `cursor`, `limit`, `projection`, and `requestId`.
 
-`agent.timeline.subscribe(handler)` establishes network demand for this agent and restores it after reconnect. Its unsubscribe function releases that demand; await `unsubscribe.ready` for initial daemon acknowledgement before starting work. Initial delivery is live-only. Reconnect delivers a bounded projected history snapshot before later updates; a live replacement invalidates the previous epoch. See the callback shapes, paging and failure behavior in [timeline events](./events.md#follow-timeline-events).
+`agent.timeline.subscribe(handler)` establishes network demand for this agent and restores it after reconnect. Its unsubscribe function releases that demand; await `unsubscribe.ready` for initial daemon acknowledgement before starting work. Delivery is live-only. Reconnect emits a local `subscription_restored` event; request missed history explicitly with `refetch()`. A live replacement invalidates the previous epoch. See the callback shapes, paging and failure behavior in [timeline events](./events.md#follow-timeline-events).
 
 ## `client.projects`
 
 | Method               | Result                   | Behavior                                                                                       |
 | -------------------- | ------------------------ | ---------------------------------------------------------------------------------------------- |
 | `list(options?)`     | `PaseoProjectListResult` | Lists every registered project, including projects with no active workspaces.                  |
-| `subscribe(handler)` | Unsubscribe function     | Local listener. Requires `observeEvents(["project.update"])`; `list()` supplies initial state. |
+| `subscribe(handler)` | Unsubscribe function     | Requests future project updates; unsubscribe releases demand. `list()` supplies initial state. |
 
 See [events](./events.md#follow-provider-catalog-changes) for explicit event observation and cleanup.
 
@@ -187,7 +187,7 @@ Use `workspace.terminals.create(options?)` and `workspace.terminals.list(options
 | `listFeatures(draftConfig)`      | Features result               | Discovers features for the current draft provider configuration.                                                                                         |
 | `diagnostic(provider)`           | Diagnostic result             | Returns human-readable setup diagnostics.                                                                                                                |
 | `listUsage(options?)`            | `PaseoProviderUsageResult`    | Returns normalized subscription windows, balances, and provider details. Rejects with an update-host error when unsupported. Options: `requestId`.       |
-| `subscribe(handler)`             | Unsubscribe function          | Local listener. Requires `observeEvents(["providers_snapshot_update"])`.                                                                                 |
+| `subscribe(handler)`             | Unsubscribe function          | Requests future catalog updates; unsubscribe releases demand.                                                                                            |
 
 ## `client.config`