From 18aaec2772a5bd3d5ef1384022e5fa8d7650d5e4 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 8 Sep 2026 12:59:25 +0200 Subject: [PATCH] fix(client): keep daemon connections lean by default (#4470) Move supported protocol capabilities into client defaults and make session traffic subscription-driven. Preserve timeline readiness and reconnect recovery across SDK, CLI, and plugin consumers. --- docs/architecture.md | 3 +- docs/protocol-compatibility.md | 12 + packages/app/src/runtime/host-runtime.ts | 2 - packages/cli/src/commands/agent/attach.ts | 19 +- packages/cli/src/commands/agent/logs.ts | 14 +- packages/client/README.md | 5 + .../src/compat/normalize-provider-models.ts | 9 +- packages/client/src/connection.test.ts | 391 ++++++++++++++++++ packages/client/src/connection/index.ts | 119 ++++++ packages/client/src/daemon-client.test.ts | 7 +- packages/client/src/daemon-client.ts | 138 ++++++- packages/client/src/index.ts | 32 +- .../client/src/provider-snapshots/index.ts | 77 ++++ packages/protocol/src/client-capabilities.ts | 2 + packages/protocol/src/messages.ts | 23 ++ .../authorization/operation-permissions.ts | 2 + .../src/server/client-activity.e2e.test.ts | 11 +- .../src/server/daemon-client.e2e.test.ts | 2 +- .../user-message-contract.real.e2e.test.ts | 2 +- .../plugins/connection-demand.e2e.test.ts | 91 ++++ packages/server/src/server/session.ts | 65 ++- .../agent-updates-service.test.ts | 10 + .../agent-updates/agent-updates-service.ts | 12 + .../provider/provider-catalog-session.test.ts | 2 + .../provider/provider-catalog-session.ts | 2 + .../src/server/voice-local-agent.e2e.test.ts | 7 +- .../src/server/voice-roundtrip.e2e.test.ts | 2 +- .../server/src/server/websocket-server.ts | 1 + public-docs/plugins/v0.8/reference.md | 2 +- public-docs/sdk/events.md | 14 +- public-docs/sdk/reference.md | 2 +- 31 files changed, 1012 insertions(+), 68 deletions(-) create mode 100644 packages/client/src/connection.test.ts create mode 100644 packages/client/src/connection/index.ts create mode 100644 packages/client/src/provider-snapshots/index.ts create mode 100644 packages/server/src/server/plugins/connection-demand.e2e.test.ts diff --git a/docs/architecture.md b/docs/architecture.md index 134e8f3532..32fef93532 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -96,7 +96,8 @@ code imports from `@getpaseo/client`. `PaseoApi` is the capability-only boundary over workspaces, agents, terminals, providers, and config. `PaseoClient` adds connection lifecycle. App plugin surfaces borrow an API over their selected -host's client; plugin subprocesses use the same facade over a host-owned IPC transport. +host's client; plugin subprocesses use the same facade over a host-owned IPC transport. Protocol capability ownership and subscription lifetimes follow +[the client contract](protocol-compatibility.md#client-capability-ownership). ### `packages/app` — Mobile + web client (Expo) diff --git a/docs/protocol-compatibility.md b/docs/protocol-compatibility.md index dddff2c107..6a232b8061 100644 --- a/docs/protocol-compatibility.md +++ b/docs/protocol-compatibility.md @@ -38,6 +38,18 @@ The app checks for the capability and either runs the feature or tells the user Existing functionality keeps working across versions because of the protocol contract. Gating a new feature never substitutes for that. +## Client capability ownership + +The client package advertises the protocol behavior it implements. Add each new capability to +its exhaustive defaults and implement the associated subscription or decoding behavior there. +The app, CLI, and plugins inherit those defaults; they supply only host resources such as browser +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, +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. + ## Every shim is tagged and dated A shim that exists for old-app or old-daemon support carries a comment naming it, the version it arrived in, and when it can go: diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 9d551be1c7..0dd8bf776e 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -501,8 +501,6 @@ function createDefaultDeps(): HostRuntimeControllerDeps { } : undefined; const appCapabilities = { - [CLIENT_CAPS.selectiveAgentTimeline]: true, - [CLIENT_CAPS.timelineReplacementInvalidation]: true, ...browserAutomationCapabilities, }; diff --git a/packages/cli/src/commands/agent/attach.ts b/packages/cli/src/commands/agent/attach.ts index fb58f71fa0..e73bb0a2f3 100644 --- a/packages/cli/src/commands/agent/attach.ts +++ b/packages/cli/src/commands/agent/attach.ts @@ -12,7 +12,7 @@ import { } from "../../utils/timeline.js"; import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; import type { AgentTimelineItem } from "@getpaseo/protocol/agent-types"; -import type { AgentStreamEventPayload, AgentStreamMessage } from "@getpaseo/protocol/messages"; +import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages"; export interface AgentAttachOptions { host?: string; @@ -134,7 +134,7 @@ export async function runAttachCommand( const resolvedId = fetchResult.agent.id; // Print header - console.log(`Attaching to agent ${resolvedId.substring(0, 7)}...`); + console.log(`Fetching history for agent ${resolvedId.substring(0, 7)}...`); console.log(`(Press Ctrl+C to detach)\n`); // Print existing output from timeline fetch. @@ -152,14 +152,17 @@ export async function runAttachCommand( } // Subscribe to new events - const unsubscribe = client.on("agent_stream", (msg: unknown) => { - const message = msg as AgentStreamMessage; - if (message.type !== "agent_stream") return; - if (message.payload.agentId !== resolvedId) return; - - printStreamEvent(message.payload.event); + const unsubscribe = client.subscribeAgentTimeline(resolvedId, (message) => { + if (message.type === "agent.timeline.replacement") { + console.log("\n[Timeline replaced; earlier output is no longer current]"); + } else { + printStreamEvent(message.payload.event); + } }); + await unsubscribe.ready; + console.log(`Attached to agent ${resolvedId.substring(0, 7)}.`); + // Handle Ctrl+C to detach gracefully let detached = false; const detach = () => { diff --git a/packages/cli/src/commands/agent/logs.ts b/packages/cli/src/commands/agent/logs.ts index 14355aee81..e29c9f6c69 100644 --- a/packages/cli/src/commands/agent/logs.ts +++ b/packages/cli/src/commands/agent/logs.ts @@ -7,7 +7,6 @@ import { } from "../../utils/timeline.js"; import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; import type { AgentTimelineItem } from "@getpaseo/protocol/agent-types"; -import type { AgentStreamMessage } from "@getpaseo/protocol/messages"; import { curateAgentActivity } from "@getpaseo/server"; export function addLogsOptions(cmd: Command): Command { @@ -202,12 +201,12 @@ async function runFollowMode( // Subscribe to new events const tailLabel = tailCount === 0 ? "no history" : `last ${tailCount} entr${tailCount === 1 ? "y" : "ies"}`; - console.log(`\n--- Following logs (${tailLabel}; Ctrl+C to stop) ---\n`); - const unsubscribe = client.on("agent_stream", (msg: unknown) => { - const message = msg as AgentStreamMessage; - if (message.type !== "agent_stream") return; - if (message.payload.agentId !== agentId) return; + const unsubscribe = client.subscribeAgentTimeline(agentId, (message) => { + if (message.type === "agent.timeline.replacement") { + console.log("\n[Timeline replaced; earlier output is no longer current]"); + return; + } if (message.payload.event.type === "timeline") { const item = message.payload.event.item; @@ -223,6 +222,9 @@ async function runFollowMode( } }); + await unsubscribe.ready; + console.log(`\n--- Following logs (${tailLabel}; Ctrl+C to stop) ---\n`); + // Wait for interrupt await new Promise((resolve) => { const cleanup = () => { diff --git a/packages/client/README.md b/packages/client/README.md index 85d6757d42..29bed77a81 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -34,6 +34,11 @@ The client needs a WebSocket implementation. Modern browsers and Node.js 22 prov Use a WebSocket URL ending in `/ws`, such as `ws://127.0.0.1:6767/ws`. Pass `password` when the daemon requires authentication. +The client advertises its supported protocol capabilities by default. Optional `capabilities` +overrides extend or override that declaration; browser hosting must be supplied by the caller. +Connecting alone does not subscribe to agent timelines or catalog events. See the +[event guide](https://paseo.sh/docs/sdk/events) for subscription lifetimes and timeline replacements. + ## Stability The high-level API exported from `@getpaseo/client` is the supported SDK surface. The SDK and daemon remain protocol-compatible across versions, but newly added capabilities can require a newer daemon. diff --git a/packages/client/src/compat/normalize-provider-models.ts b/packages/client/src/compat/normalize-provider-models.ts index 023c8681e3..b2662c0d21 100644 --- a/packages/client/src/compat/normalize-provider-models.ts +++ b/packages/client/src/compat/normalize-provider-models.ts @@ -61,9 +61,16 @@ export function normalizeProvidersSnapshotPayload< T extends GetProvidersSnapshotPayload | ProvidersSnapshotUpdatePayload, >(payload: T, expand = true): T { if (payload.compactSnapshot && !expand) return payload; - const entries = payload.compactSnapshot + const decoded = payload.compactSnapshot ? expandProviderSnapshot(payload.compactSnapshot) : normalizeProviderSnapshotEntries(payload.entries); + const freshness = payload.fetchedAt; + const entries = + freshness && expand + ? decoded.map((entry) => + freshness[entry.provider] ? { ...entry, fetchedAt: freshness[entry.provider] } : entry, + ) + : decoded; return entries === payload.entries ? payload : { ...payload, entries }; } diff --git a/packages/client/src/connection.test.ts b/packages/client/src/connection.test.ts new file mode 100644 index 0000000000..6807c4bf63 --- /dev/null +++ b/packages/client/src/connection.test.ts @@ -0,0 +1,391 @@ +import { expect, test } from "vitest"; +import { CLIENT_CAPS } from "@getpaseo/protocol/client-capabilities"; +import { DaemonClient, type DaemonTransport } from "./daemon-client"; + +function connection(options: { acknowledgeSubscriptions?: boolean } = {}) { + const sent: Array<{ + type: string; + capabilities?: Record; + message?: { type: string; requestId?: string; agentIds?: string[]; events?: string[] }; + }> = []; + let receive = (_data: unknown) => {}; + let open = () => {}; + let closed = (_event?: unknown) => {}; + const transport: DaemonTransport = { + send(data) { + const frame = JSON.parse(String(data)); + sent.push(frame); + if (frame.type === "hello") { + receive( + JSON.stringify({ + type: "session", + message: { + type: "status", + payload: { + status: "server_info", + serverId: "test", + hostname: null, + version: null, + features: { selectiveAgentTimeline: true, explicitEventSubscriptions: true }, + }, + }, + }), + ); + } else if ( + options.acknowledgeSubscriptions !== false && + frame.type === "session" && + frame.message.type.endsWith("set_subscription.request") + ) { + receive( + JSON.stringify({ + type: "session", + message: { + type: frame.message.type.replace(".request", ".response"), + payload: { requestId: frame.message.requestId, agentIds: frame.message.agentIds }, + }, + }), + ); + } + }, + close() {}, + onMessage(handler) { + receive = handler; + return () => {}; + }, + onOpen(handler) { + open = handler; + return () => {}; + }, + onClose(handler) { + closed = handler; + return () => {}; + }, + onError() { + return () => {}; + }, + }; + const client = new DaemonClient({ + url: "ws://test", + clientId: "test", + transportFactory: () => transport, + reconnect: { enabled: false }, + }); + return { + client, + sent, + open: () => open(), + disconnect: () => closed(), + receive: (message: unknown) => receive(JSON.stringify({ type: "session", message })), + }; +} + +test("a plain client advertises every protocol capability and no browser host", async () => { + const h = connection(); + try { + const ready = h.client.connect(); + h.open(); + await ready; + const { browserHost: _browser, ...protocolCapabilities } = CLIENT_CAPS; + // Every new capability needs a deliberate default or a host-owned exception. + expect(h.sent).toHaveLength(1); + expect(h.sent[0].capabilities).toEqual( + Object.fromEntries(Object.values(protocolCapabilities).map((key) => [key, true])), + ); + } finally { + await h.client.close(); + } +}); + +test("SDK timeline listeners own their union across unsubscribe and reconnect", async () => { + const { createPaseoApi } = await import("./index"); + const h = connection(); + const api = createPaseoApi(h.client); + try { + const ready = h.client.connect(); + h.open(); + await ready; + const first = api.agents.ref("a").timeline.subscribe(() => {}); + const same = api.agents.ref("a").timeline.subscribe(() => {}); + const second = api.agents.ref("b").timeline.subscribe(() => {}); + await Promise.resolve(); + const memberships = () => + h.sent + .filter((f) => f.message?.type === "agent.timeline.set_subscription.request") + .map((f) => f.message?.agentIds); + expect(memberships()).toEqual([["a"], ["a"], ["a", "b"]]); + first(); + expect(memberships()).toEqual([["a"], ["a"], ["a", "b"]]); + same(); + expect(memberships().at(-1)).toEqual(["b"]); + h.disconnect(); + const reconnect = h.client.connect(); + h.open(); + await reconnect; + expect(memberships().at(-1)).toEqual(["b"]); + second(); + expect(memberships().at(-1)).toEqual([]); + } finally { + await h.client.close(); + } +}); + +test("SDK subscribers receive timeline replacement instead of silently losing history", async () => { + const { createPaseoApi } = await import("./index"); + const h = connection(); + const received: unknown[] = []; + try { + const ready = h.client.connect(); + h.open(); + await ready; + const off = createPaseoApi(h.client) + .agents.ref("agent") + .timeline.subscribe((event) => received.push(event)); + h.receive({ type: "agent.timeline.replacement", payload: { agentId: "agent", epoch: "next" } }); + expect(received).toEqual([{ agentId: "agent", event: { type: "replacement", epoch: "next" } }]); + off(); + } finally { + await h.client.close(); + } +}); + +test("event listeners release demand and restore only remaining interests", async () => { + const h = connection(); + try { + const ready = h.client.connect(); + h.open(); + await ready; + const projects = h.client.on("project.update", () => {}); + const providers = h.client.on("providers_snapshot_update", () => {}); + const subscriptions = () => + h.sent + .filter((f) => f.message?.type === "session.events.set_subscription.request") + .map((f) => f.message?.events); + expect(subscriptions()).toEqual([ + ["project.update"], + ["project.update", "providers_snapshot_update"], + ]); + projects(); + h.disconnect(); + const reconnect = h.client.connect(); + h.open(); + await reconnect; + expect(subscriptions().at(-1)).toEqual(["providers_snapshot_update"]); + providers(); + expect(subscriptions().at(-1)).toEqual([]); + } finally { + await h.client.close(); + } +}); + +test("provider references fetch a body only for interested SDK listeners", async () => { + const h = connection(); + const received: unknown[] = []; + const payload = { + entries: [], + snapshotHash: "catalog", + fetchedAt: { test: "2026-09-08T00:00:00.000Z" }, + generatedAt: "2026-09-08T00:00:00.000Z", + }; + try { + const ready = h.client.connect(); + h.open(); + await ready; + h.receive({ type: "providers_snapshot_update", payload }); + expect(h.sent).toHaveLength(1); + const off = h.client.on("providers_snapshot_update", (m) => received.push(m.payload.entries)); + h.receive({ type: "providers_snapshot_update", payload }); + h.receive({ type: "providers_snapshot_update", payload }); + const requests = h.sent.filter((f) => f.message?.type === "get_providers_snapshot_request"); + expect(requests).toHaveLength(1); + h.receive({ + type: "get_providers_snapshot_response", + payload: { + ...payload, + requestId: requests[0].message?.requestId, + compactSnapshot: { + entries: [ + { + provider: "test", + enabled: true, + status: "ready", + models: [{ id: "one", label: "One" }], + }, + ], + thinkingSets: [], + }, + }, + }); + await expect + .poll(() => received) + .toEqual([ + [ + { + provider: "test", + enabled: true, + status: "ready", + fetchedAt: "2026-09-08T00:00:00.000Z", + models: [{ provider: "test", id: "one", label: "One" }], + }, + ], + ]); + off(); + h.receive({ type: "providers_snapshot_update", payload: { ...payload, snapshotHash: "next" } }); + expect(h.sent.filter((f) => f.message?.type === "get_providers_snapshot_request")).toHaveLength( + 1, + ); + } finally { + await h.client.close(); + } +}); + +test("SDK listeners preserve the app's explicit timeline membership", async () => { + const h = connection(); + try { + const ready = h.client.connect(); + h.open(); + await ready; + await h.client.setAgentTimelineSubscription(["visible"]); + const off = h.client.subscribeAgentTimeline("plugin", () => {}); + await h.client.setAgentTimelineSubscription(["next"]); + expect(h.sent.at(-1)?.message?.agentIds).toEqual(["next", "plugin"]); + off(); + expect(h.sent.at(-1)?.message?.agentIds).toEqual(["next"]); + } finally { + await h.client.close(); + } +}); + +test("timeline readiness waits for daemon acknowledgement, including shared listeners", async () => { + const h = connection({ acknowledgeSubscriptions: false }); + try { + const connect = h.client.connect(); + h.open(); + await connect; + const first = h.client.subscribeAgentTimeline("agent", () => {}); + const second = h.client.subscribeAgentTimeline("agent", () => {}); + expect(first.ready).toBeInstanceOf(Promise); + let ready = false; + void second.ready.then(() => { + return (ready = true); + }); + await Promise.resolve(); + expect(ready).toBe(false); + const request = h.sent.at(-1)!.message!; + h.receive({ + type: "agent.timeline.set_subscription.response", + payload: { requestId: request.requestId, agentIds: ["agent"] }, + }); + await Promise.all([first.ready, second.ready]); + expect(ready).toBe(true); + h.disconnect(); + const third = h.client.subscribeAgentTimeline("agent", () => {}); + let reconnected = false; + void third.ready.then(() => { + return (reconnected = true); + }); + const reconnect = h.client.connect(); + h.open(); + await reconnect; + expect(reconnected).toBe(false); + h.receive({ + type: "agent.timeline.set_subscription.response", + payload: { requestId: h.sent.at(-1)!.message!.requestId, agentIds: ["agent"] }, + }); + await third.ready; + expect(reconnected).toBe(true); + first(); + second(); + third(); + } finally { + await h.client.close(); + } +}); + +test("an unresolved provider reference is fetched again after reconnect", async () => { + const h = connection(); + const received: unknown[] = []; + const payload = { entries: [], snapshotHash: "updated", generatedAt: "2026-09-08T00:00:00.000Z" }; + const requests = () => h.sent.filter((f) => f.message?.type === "get_providers_snapshot_request"); + try { + const connect = h.client.connect(); + h.open(); + await connect; + const off = h.client.on("providers_snapshot_update", (m) => + received.push(m.payload.snapshotHash), + ); + h.receive({ type: "providers_snapshot_update", payload }); + expect(requests()).toHaveLength(1); + h.disconnect(); + await Promise.resolve(); + await Promise.resolve(); + const reconnect = h.client.connect(); + h.open(); + await reconnect; + expect(requests()).toHaveLength(2); + h.receive({ + type: "get_providers_snapshot_response", + payload: { + ...payload, + requestId: requests()[1].message?.requestId, + compactSnapshot: { entries: [], thinkingSets: [] }, + }, + }); + await expect.poll(() => received).toEqual(["updated"]); + off(); + } finally { + await h.client.close(); + } +}); + +test("timeline readiness remains pending before connect and rejects when released", async () => { + const h = connection(); + try { + const release = h.client.subscribeAgentTimeline("agent", () => {}); + const canceled = h.client.subscribeAgentTimeline("canceled", () => {}); + canceled(); + await expect(canceled.ready).rejects.toThrow("released"); + let established = false; + void release.ready.then(() => { + return (established = true); + }); + await Promise.resolve(); + expect(established).toBe(false); + expect(h.sent).toEqual([]); + const connect = h.client.connect(); + h.open(); + await connect; + await release.ready; + expect(established).toBe(true); + release(); + } finally { + await h.client.close(); + } +}); + +test("unsubscribing while disconnected discards unresolved provider demand", async () => { + const h = connection(); + try { + const connect = h.client.connect(); + h.open(); + await connect; + const off = h.client.on("providers_snapshot_update", () => {}); + h.receive({ + type: "providers_snapshot_update", + payload: { + entries: [], + snapshotHash: "updated", + generatedAt: "2026-09-08T00:00:00.000Z", + }, + }); + h.disconnect(); + off(); + const reconnect = h.client.connect(); + h.open(); + await reconnect; + expect(h.sent.filter((f) => f.message?.type === "get_providers_snapshot_request")).toHaveLength( + 1, + ); + } finally { + await h.client.close(); + } +}); diff --git a/packages/client/src/connection/index.ts b/packages/client/src/connection/index.ts new file mode 100644 index 0000000000..fe8553588d --- /dev/null +++ b/packages/client/src/connection/index.ts @@ -0,0 +1,119 @@ +import type { SessionEventSubscription } from "@getpaseo/protocol/messages"; +import { CLIENT_CAPS, type ClientCapability } from "@getpaseo/protocol/client-capabilities"; + +// 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 = { + [CLIENT_CAPS.allProviders]: true, + [CLIENT_CAPS.selectiveAgentTimeline]: true, + [CLIENT_CAPS.reasoningMergeEnum]: true, + [CLIENT_CAPS.customModeIcons]: true, + [CLIENT_CAPS.terminalReflowableSnapshot]: true, + [CLIENT_CAPS.providerSubagents]: true, + [CLIENT_CAPS.projectUpdates]: true, + [CLIENT_CAPS.compactProviderSnapshots]: true, + [CLIENT_CAPS.providerSnapshotReferences]: true, + [CLIENT_CAPS.timelineReplacementInvalidation]: true, + [CLIENT_CAPS.timelineNotifications]: true, + [CLIENT_CAPS.pluginTimelineItems]: true, + [CLIENT_CAPS.workspaceSetupBlocked]: true, + [CLIENT_CAPS.explicitEventSubscriptions]: true, +} satisfies Record, true>; + +/** Calling releases demand; ready acknowledges the initial daemon membership. */ +export type TimelineSubscription = (() => void) & { readonly ready: Promise }; + +class TimelineInterest { + resolve!: () => void; + reject!: (error: unknown) => void; + readonly ready = new Promise((resolve, reject) => { + this.resolve = resolve; + this.reject = reject; + }); + + constructor() { + // Readiness is optional for fire-and-forget listeners. + void this.ready.catch(() => {}); + } +} + +/** Owns connection demand, independently of individual facades and React lifetimes. */ +export class ConnectionSubscriptions { + private viewed = new Set(); + private timelines = new Map>(); + private events: SessionEventSubscription[] = []; + + constructor( + private readonly send: { + timelines(agentIds: string[]): Promise | null; + events(events: SessionEventSubscription[]): Promise; + failed(error: unknown): void; + }, + ) {} + + private agentIds(): string[] { + return [...new Set([...this.viewed, ...this.timelines.keys()])].sort(); + } + + setViewed(agentIds: string[]): Promise { + this.viewed = new Set(agentIds); + return this.syncTimelines(); + } + + observeTimeline(agentId: string): TimelineSubscription { + const interests = this.timelines.get(agentId) ?? new Set(); + const interest = new TimelineInterest(); + interests.add(interest); + this.timelines.set(agentId, interests); + // Each caller gets acknowledgement for its own registration, even when an + // existing listener first subscribed on an earlier connection. + void this.syncTimelines().catch(this.send.failed); + let active = true; + return Object.assign( + () => { + if (!active) return; + active = false; + interest.reject(new Error("Timeline subscription released before it was ready")); + interests.delete(interest); + if (interests.size === 0) { + this.timelines.delete(agentId); + void this.syncTimelines().catch(this.send.failed); + } + }, + { ready: interest.ready }, + ); + } + + private async syncTimelines(): Promise { + const interests: TimelineInterest[] = []; + for (const listeners of this.timelines.values()) { + for (const interest of listeners) interests.push(interest); + } + try { + const sent = this.send.timelines(this.agentIds()); + if (!sent) return; + await sent; + for (const interest of interests) interest.resolve(); + } catch (error) { + for (const interest of interests) interest.reject(error); + throw error; + } + } + + setEvents(events: SessionEventSubscription[]): void { + if (JSON.stringify(this.events) === JSON.stringify(events)) return; + this.events = events; + void this.send.events(events).catch(this.send.failed); + } + + restore(): void { + if (this.agentIds().length) void this.syncTimelines().catch(this.send.failed); + if (this.events.length) void this.send.events(this.events).catch(this.send.failed); + } + + close(): void { + for (const interests of this.timelines.values()) { + for (const interest of interests) interest.reject(new Error("Daemon client closed")); + } + } +} diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index 44b9189425..bfb582e5ab 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -259,7 +259,7 @@ test("does not infer browser automation capabilities from Electron runtime", asy }) .parse(JSON.parse(assertStr(mock.sent[0]))); expect(hello.capabilities[CLIENT_CAPS.browserHost]).toBeUndefined(); - expect(hello.capabilities[CLIENT_CAPS.selectiveAgentTimeline]).toBeUndefined(); + expect(hello.capabilities[CLIENT_CAPS.selectiveAgentTimeline]).toBe(true); }); test("advertises consumer-provided browser automation capabilities", async () => { @@ -755,6 +755,11 @@ test("advertises client capabilities in hello", async () => { clientType: "cli", protocolVersion: 1, capabilities: { + all_providers: true, + selective_agent_timeline: true, + timeline_replacement_invalidation: true, + provider_snapshot_references: true, + explicit_event_subscriptions: true, compact_provider_snapshots: true, custom_mode_icons: true, project_updates: true, diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 576e111a71..aa2a410ab0 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -1,3 +1,10 @@ +import { ProviderSnapshotUpdates } from "./provider-snapshots/index.js"; +import type { SessionEventSubscription } from "@getpaseo/protocol/messages"; +import { + ConnectionSubscriptions, + DEFAULT_CLIENT_CAPABILITIES, + type TimelineSubscription, +} from "./connection/index.js"; import type { z } from "zod"; import { CLIENT_CAPS, type ClientCapability } from "@getpaseo/protocol/client-capabilities"; import type { AgentAttentionNotificationPayload } from "@getpaseo/protocol/agent-attention-notification"; @@ -1066,6 +1073,17 @@ interface PingProbe { } export class DaemonClient { + private readonly providerSnapshotUpdates = new ProviderSnapshotUpdates({ + fetch: (cwd) => this.requestProvidersSnapshot({ cwd }), + emit: (message) => this.deliverSessionMessage(message), + failed: (error) => this.logger.error({ err: error }, "Failed to resolve provider snapshot"), + }); + private readonly subscriptions = new ConnectionSubscriptions({ + timelines: (agentIds) => (this.isConnected ? this.sendTimelineSubscription(agentIds) : null), + events: (events) => this.sendEventSubscription(events), + failed: (error) => + this.logger.error({ err: error }, "Failed to update connection subscriptions"), + }); private transport: DaemonTransport | null = null; private transportCleanup: Array<() => void> = []; private rawMessageListeners: Set<(message: SessionOutboundMessage) => void> = new Set(); @@ -1366,6 +1384,7 @@ export class DaemonClient { return; } this.shouldReconnect = false; + this.subscriptions.close(); this.connectPromise = null; this.connectResolve = null; this.connectReject = null; @@ -1375,6 +1394,7 @@ export class DaemonClient { } this.resetConnectTimeout(); this.disposeTransport(1000, "Client closed"); + this.providerSnapshotUpdates.clear(); this.clearWaiters(new Error("Daemon client closed")); this.rejectPendingSendQueue(new Error("Daemon client closed")); this.rejectPingProbe(new Error("Daemon client closed")); @@ -1451,7 +1471,11 @@ export class DaemonClient { subscribe(handler: DaemonEventHandler): () => void { this.eventListeners.add(handler); - return () => this.eventListeners.delete(handler); + this.updateEventSubscriptions(); + return () => { + this.eventListeners.delete(handler); + this.updateEventSubscriptions(); + }; } subscribeRawMessages(handler: (message: SessionOutboundMessage) => void): () => void { @@ -1481,6 +1505,7 @@ export class DaemonClient { this.messageHandlers.set(type, new Set()); } this.messageHandlers.get(type)!.add(handler); + this.updateEventSubscriptions(); return () => { const handlers = this.messageHandlers.get(type); @@ -1491,6 +1516,7 @@ export class DaemonClient { if (handlers.size === 0) { this.messageHandlers.delete(type); } + this.updateEventSubscriptions(); }; } @@ -3014,7 +3040,70 @@ export class DaemonClient { return payload; } - async setAgentTimelineSubscription(agentIds: string[]): Promise { + setAgentTimelineSubscription(agentIds: string[]): Promise { + return this.subscriptions.setViewed(agentIds); + } + + subscribeAgentTimeline( + agentId: string, + handler: ( + message: Extract< + SessionOutboundMessage, + { type: "agent_stream" | "agent.timeline.replacement" } + >, + ) => void, + ): TimelineSubscription { + const stream = this.on("agent_stream", (message) => { + if (message.payload.agentId === agentId) handler(message); + }); + const replacement = this.on("agent.timeline.replacement", (message) => { + if (message.payload.agentId === agentId) handler(message); + }); + const release = this.subscriptions.observeTimeline(agentId); + return Object.assign( + () => { + stream(); + replacement(); + release(); + }, + { ready: release.ready }, + ); + } + + private updateEventSubscriptions(): void { + const events: SessionEventSubscription[] = [ + "project.update", + "providers_snapshot_update", + "agent_attention_required", + "agent_permission_request", + "agent_permission_resolved", + ]; + if (this.eventListeners.size === 0 && !this.messageHandlers.has("providers_snapshot_update")) { + this.providerSnapshotUpdates.clear(); + } + this.subscriptions.setEvents( + events.filter( + (type) => this.eventListeners.size > 0 || (this.messageHandlers.get(type)?.size ?? 0) > 0, + ), + ); + } + + private async sendEventSubscription(events: SessionEventSubscription[]): Promise { + if (this.connectionState.status !== "connected") return; + // COMPAT(explicitEventSubscriptions): added in v0.8.0, remove legacy broadcast handling after 2027-03-08. + if ( + !this.lastServerInfoMessage?.features?.explicitEventSubscriptions || + this.config.capabilities?.[CLIENT_CAPS.explicitEventSubscriptions] === false + ) + return; + await this.sendCorrelatedSessionRequest({ + message: { type: "session.events.set_subscription.request", events }, + responseType: "session.events.set_subscription.response", + }); + } + + private async sendTimelineSubscription(agentIds: string[]): Promise { + if (this.connectionState.status !== "connected") return; // COMPAT(selectiveAgentTimeline): added in v0.1.106. Old daemons keep their // legacy global stream and do not understand this RPC. Remove after // 2027-01-12 once the supported daemon floor is >= v0.1.106. @@ -4683,7 +4772,16 @@ export class DaemonClient { ifNoneMatch?: string; requestId?: string; }): Promise { - const payload = await this.sendCorrelatedSessionRequest({ + const payload = await this.requestProvidersSnapshot(options); + return normalizeProvidersSnapshotPayload(payload, this.config.providerSnapshots !== "wire"); + } + + private requestProvidersSnapshot(options?: { + cwd?: string; + ifNoneMatch?: string; + requestId?: string; + }): Promise { + return this.sendCorrelatedSessionRequest({ requestId: options?.requestId, message: { type: "get_providers_snapshot_request", @@ -4692,7 +4790,6 @@ export class DaemonClient { }, responseType: "get_providers_snapshot_response", }); - return normalizeProvidersSnapshotPayload(payload, this.config.providerSnapshots !== "wire"); } async getDaemonConfig( @@ -5671,18 +5768,7 @@ export class DaemonClient { clientType: this.config.clientType ?? "cli", protocolVersion: 1, capabilities: { - [CLIENT_CAPS.customModeIcons]: true, - [CLIENT_CAPS.reasoningMergeEnum]: true, - [CLIENT_CAPS.terminalReflowableSnapshot]: true, - [CLIENT_CAPS.providerSubagents]: true, - [CLIENT_CAPS.projectUpdates]: true, - [CLIENT_CAPS.compactProviderSnapshots]: true, - ...(this.config.providerSnapshots === "wire" - ? { [CLIENT_CAPS.providerSnapshotReferences]: true } - : {}), - [CLIENT_CAPS.timelineNotifications]: true, - [CLIENT_CAPS.pluginTimelineItems]: true, - [CLIENT_CAPS.workspaceSetupBlocked]: true, + ...DEFAULT_CLIENT_CAPABILITIES, ...this.config.capabilities, }, ...(this.config.appVersion ? { appVersion: this.config.appVersion } : {}), @@ -5699,6 +5785,7 @@ export class DaemonClient { } private disposeTransport(code = 1001, reason = "Reconnecting"): void { + this.providerSnapshotUpdates.pause(); this.stopLivenessHeartbeat(); this.cleanupTransport(); if (this.transport) { @@ -6010,6 +6097,8 @@ export class DaemonClient { this.lastErrorValue = reason.trim(); } + this.providerSnapshotUpdates.pause(); + // Clear all pending waiters and queued sends since the connection was lost // and responses from the previous connection will never arrive. this.clearWaiters(new Error(reason ?? "Connection lost")); @@ -6108,6 +6197,21 @@ export class DaemonClient { } private handleSessionMessage(msg: SessionOutboundMessage): void { + if ( + msg.type === "providers_snapshot_update" && + this.config.providerSnapshots !== "wire" && + msg.payload.snapshotHash && + !msg.payload.compactSnapshot + ) { + if (this.messageHandlers.has(msg.type) || this.eventListeners.size > 0) { + this.providerSnapshotUpdates.receive(msg); + } + return; + } + this.deliverSessionMessage(msg); + } + + private deliverSessionMessage(msg: SessionOutboundMessage): void { const consumerMessage = normalizeProviderSnapshotUpdateMessage( msg, this.config.providerSnapshots !== "wire", @@ -6122,6 +6226,8 @@ export class DaemonClient { this.reconnectAttempt = 0; this.updateConnectionState({ status: "connected" }, { event: "HELLO_SERVER_INFO" }); this.startLivenessHeartbeat(); + this.subscriptions.restore(); + this.providerSnapshotUpdates.resume(); this.resubscribeCheckoutDiffSubscriptions(); this.resubscribeTerminalDirectorySubscriptions(); this.resubscribeFileSubscriptions(); diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index 729ea4bc94..7068cdc0bf 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -1,3 +1,4 @@ +import type { DaemonClientConfig } from "./daemon-client.js"; import type { AgentPermissionResponse } from "@getpaseo/protocol/agent-types"; import type { AgentSnapshotPayload, @@ -75,6 +76,7 @@ export interface PaseoLogger { } export interface PaseoClientConfig { + capabilities?: DaemonClientConfig["capabilities"]; url: string; clientId?: string; appVersion?: string; @@ -282,6 +284,15 @@ export type PaseoAgentStream = Extract void; +export type PaseoAgentTimelineEvent = + | PaseoAgentStream + | { + agentId: string; + event: { type: "replacement"; epoch: string }; + }; + +export type PaseoAgentTimelineSubscription = ReturnType; + export interface PaseoAgentTimelineHandle { append(item: Omit): Promise<{ seq: number; epoch: string }>; /** @@ -291,10 +302,12 @@ export interface PaseoAgentTimelineHandle { */ refetch(options?: PaseoAgentTimelineRefetchOptions): Promise; /** - * Local listener for agent_stream events matching this handle id. It does not - * retain timeline entries or own application cache state. + * Subscribe to this agent and restore demand after reconnect. A replacement + * event invalidates previously fetched history; refetch the page you need. + * Await the returned unsubscribe function's `ready` promise before starting + * work that must be observed. It rejects if establishment fails. */ - subscribe(handler: (event: PaseoAgentStream) => void): () => void; + subscribe(handler: (event: PaseoAgentTimelineEvent) => void): PaseoAgentTimelineSubscription; } export interface PaseoAgentHandle { @@ -673,10 +686,15 @@ function createAgentHandleFactory(daemonClient: DaemonClient): AgentHandleFactor return result; }, subscribe: (handler) => - daemonClient.on("agent_stream", (message) => { - if (message.payload.agentId === id) { - handler(message.payload); - } + daemonClient.subscribeAgentTimeline(id, (message) => { + handler( + message.type === "agent_stream" + ? message.payload + : { + agentId: message.payload.agentId, + event: { type: "replacement", epoch: message.payload.epoch }, + }, + ); }), }, get workspaceId() { diff --git a/packages/client/src/provider-snapshots/index.ts b/packages/client/src/provider-snapshots/index.ts new file mode 100644 index 0000000000..7450a61639 --- /dev/null +++ b/packages/client/src/provider-snapshots/index.ts @@ -0,0 +1,77 @@ +import type { + GetProvidersSnapshotResponseMessage, + SessionOutboundMessage, +} from "@getpaseo/protocol/messages"; + +type Update = Extract; +type Snapshot = GetProvidersSnapshotResponseMessage["payload"]; + +interface PendingSnapshot { + latest: Update; + running: boolean; +} + +/** Resolve announcements only while observed, coalescing changes during a fetch. */ +export class ProviderSnapshotUpdates { + private pending = new Map(); + private paused = false; + + constructor( + private readonly host: { + fetch(cwd: string | undefined): Promise; + emit(update: Update): void; + failed(error: unknown): void; + }, + ) {} + + clear(): void { + this.pending.clear(); + } + + pause(): void { + this.paused = true; + // Invalidate old responses while retaining the announcements they would resolve. + for (const [cwd, pending] of this.pending) { + this.pending.set(cwd, { latest: pending.latest, running: false }); + } + } + + resume(): void { + this.paused = false; + for (const pending of this.pending.values()) this.receive(pending.latest); + } + + receive(update: Update): void { + const cwd = update.payload.cwd; + const pending = this.pending.get(cwd) ?? { latest: update, running: false }; + pending.latest = update; + this.pending.set(cwd, pending); + if (this.paused || pending.running) return; + pending.running = true; + void this.resolve(cwd, pending); + } + + private async resolve(cwd: string | undefined, pending: PendingSnapshot): Promise { + try { + while (this.pending.get(cwd) === pending) { + const announced = pending.latest; + const snapshot = await this.host.fetch(cwd); + if (this.pending.get(cwd) !== pending) return; + // A change arriving while the body was in flight needs the current body. + // If that response already covers it, no second fetch is necessary. + if ( + pending.latest !== announced && + pending.latest.payload.snapshotHash !== snapshot.snapshotHash + ) + continue; + this.pending.delete(cwd); + this.host.emit({ type: "providers_snapshot_update", payload: snapshot }); + return; + } + } catch (error) { + if (this.pending.get(cwd) === pending) this.host.failed(error); + } finally { + if (this.pending.get(cwd) === pending) this.pending.delete(cwd); + } + } +} diff --git a/packages/protocol/src/client-capabilities.ts b/packages/protocol/src/client-capabilities.ts index f15c3ea7d9..f60bf40a34 100644 --- a/packages/protocol/src/client-capabilities.ts +++ b/packages/protocol/src/client-capabilities.ts @@ -1,5 +1,7 @@ export const CLIENT_CAPS = { // Clients outside the Paseo app version series can explicitly accept provider IDs. + // COMPAT(explicitEventSubscriptions): added in v0.8.0, remove legacy broadcasts after 2027-03-08. + explicitEventSubscriptions: "explicit_event_subscriptions", allProviders: "all_providers", // COMPAT(selectiveAgentTimeline): added in v0.1.106. Capable clients receive // agent streams only for their explicit viewed set. Remove after 2027-01-12 diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 6be32a3525..6faceb4139 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -3040,7 +3040,27 @@ export const HubExecutionControlRequestSchema = z.object({ export type HubExecutionControlRequest = z.infer; +// These connection event streams have no directory bootstrap or timeline membership. +export const SessionEventSubscriptionSchema = z.enum([ + "project.update", + "providers_snapshot_update", + "agent_attention_required", + "agent_permission_request", + "agent_permission_resolved", +]); +export type SessionEventSubscription = z.infer; +export const SessionEventsSetSubscriptionRequestSchema = z.object({ + type: z.literal("session.events.set_subscription.request"), + requestId: z.string(), + events: z.array(SessionEventSubscriptionSchema), +}); +export const SessionEventsSetSubscriptionResponseSchema = z.object({ + type: z.literal("session.events.set_subscription.response"), + payload: z.object({ requestId: z.string() }), +}); + export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ + SessionEventsSetSubscriptionRequestSchema, HubExecutionAgentCreateRequestSchema, HubExecutionAgentValidateRequestSchema, HubExecutionControlRequestSchema, @@ -3536,6 +3556,7 @@ export const ServerInfoStatusPayloadSchema = z forgeProviders: z.boolean().optional(), // COMPAT(selectiveAgentTimeline): added in v0.1.106, remove after 2027-01-12. selectiveAgentTimeline: z.boolean().optional(), + explicitEventSubscriptions: z.boolean().optional(), // COMPAT(canonicalSubmittedPrompts): added in v0.2.6, remove gate after 2027-01-30. canonicalSubmittedPrompts: z.boolean().optional(), // COMPAT(agentTurnIdentity): accept peers that observed pre-release v0.2.6 through 2027-01-31. @@ -6418,6 +6439,7 @@ export const AgentSkillsImportLegacySelectionResponseSchema = z.object({ }); export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ + SessionEventsSetSubscriptionResponseSchema, HubExecutionAgentCreateResponseSchema, HubExecutionAgentValidateResponseSchema, HubExecutionControlResponseSchema, @@ -7098,6 +7120,7 @@ export const WSHelloMessageSchema = z.object({ .object({ voice: z.boolean().optional(), pushNotifications: z.boolean().optional(), + [CLIENT_CAPS.explicitEventSubscriptions]: z.boolean().optional(), [CLIENT_CAPS.allProviders]: z.boolean().optional(), [CLIENT_CAPS.reasoningMergeEnum]: z.boolean().optional(), [CLIENT_CAPS.selectiveAgentTimeline]: z.boolean().optional(), diff --git a/packages/server/src/server/authorization/operation-permissions.ts b/packages/server/src/server/authorization/operation-permissions.ts index dd65f47c9d..1f8b9e1d1f 100644 --- a/packages/server/src/server/authorization/operation-permissions.ts +++ b/packages/server/src/server/authorization/operation-permissions.ts @@ -20,6 +20,7 @@ const INBOUND_PERMISSION = { "agent.skills.uninstall.request": "daemon.manage", "agent.timeline.list_prompts.request": "workspace.read", "agent.timeline.append.request": "workspace.write", + "session.events.set_subscription.request": ["workspace.read", "daemon.read", "hub.execute"], "agent.timeline.set_subscription.request": ["workspace.read", "hub.execute"], agent_permission_response: "workspace.write", archive_agent_request: ["workspace.write", "hub.execute"], @@ -220,6 +221,7 @@ const OUTBOUND_PERMISSION = { "agent.timeline.list_prompts.response": "workspace.read", "agent.timeline.append.response": "workspace.write", "agent.timeline.replacement": ["workspace.read", "hub.execute"], + "session.events.set_subscription.response": ["workspace.read", "daemon.read", "hub.execute"], "agent.timeline.set_subscription.response": ["workspace.read", "hub.execute"], agent_archived: ["workspace.read", "hub.execute"], agent_attention_required: ["workspace.read", "hub.execute"], diff --git a/packages/server/src/server/client-activity.e2e.test.ts b/packages/server/src/server/client-activity.e2e.test.ts index 5348ea44d8..316fca8d85 100644 --- a/packages/server/src/server/client-activity.e2e.test.ts +++ b/packages/server/src/server/client-activity.e2e.test.ts @@ -84,14 +84,11 @@ describe("client activity tracking", () => { reject(new Error(`Timeout waiting for attention_required (${timeout}ms)`)); }, timeout); - const cleanup = client.on("agent_stream", (msg) => { - if (msg.type !== "agent_stream") return; - if (msg.payload.agentId !== agentId) return; - if (msg.payload.event.type !== "attention_required") return; - + const cleanup = client.onAgentAttentionRequired((notification) => { + if (notification.agentId !== agentId) return; clearTimeout(timer); cleanup(); - resolve(msg.payload.event); + resolve({ type: "attention_required", ...notification }); }); }); } @@ -107,7 +104,7 @@ describe("client activity tracking", () => { reject(new Error(`Timeout waiting for assistant timeline (${timeout}ms)`)); }, timeout); - const cleanup = client.on("agent_stream", (msg) => { + const cleanup = client.subscribeAgentTimeline(agentId, (msg) => { if (msg.type !== "agent_stream") return; if (msg.payload.agentId !== agentId) return; if (msg.payload.event.type !== "timeline") return; diff --git a/packages/server/src/server/daemon-client.e2e.test.ts b/packages/server/src/server/daemon-client.e2e.test.ts index 1c65316bdc..934125af11 100644 --- a/packages/server/src/server/daemon-client.e2e.test.ts +++ b/packages/server/src/server/daemon-client.e2e.test.ts @@ -1473,7 +1473,7 @@ test("creates agent and exercises lifecycle", async () => { sawAssistantMessage = true; } }); - const unsubscribeRawStream = ctx.client.on("agent_stream", (message) => { + const unsubscribeRawStream = ctx.client.subscribeAgentTimeline(agent.id, (message) => { if (message.type !== "agent_stream") { return; } diff --git a/packages/server/src/server/daemon-e2e/user-message-contract.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/user-message-contract.real.e2e.test.ts index b22327c6ba..41b40fed57 100644 --- a/packages/server/src/server/daemon-e2e/user-message-contract.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/user-message-contract.real.e2e.test.ts @@ -61,7 +61,7 @@ function tmpCwd(provider: ContractProvider): string { function collectUserMessageEvents(client: DaemonClient, agentId: string): AgentStreamEvent[] { const events: AgentStreamEvent[] = []; - client.on("agent_stream", (message) => { + client.subscribeAgentTimeline(agentId, (message) => { if (message.type !== "agent_stream" || message.payload.agentId !== agentId) { return; } diff --git a/packages/server/src/server/plugins/connection-demand.e2e.test.ts b/packages/server/src/server/plugins/connection-demand.e2e.test.ts new file mode 100644 index 0000000000..7c6cbfca28 --- /dev/null +++ b/packages/server/src/server/plugins/connection-demand.e2e.test.ts @@ -0,0 +1,91 @@ +import { mkdtemp, writeFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { expect, test } from "vitest"; +import { createPaseoApi } from "@getpaseo/client"; +import { DaemonClient } from "../test-utils/daemon-client.js"; +import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; + +test("an RPC-only plugin receives no agent, project or provider data", async () => { + const directory = await mkdtemp(path.join(tmpdir(), "paseo-quiet-plugin-")); + await writeFile(path.join(directory, "paseo-plugin.json"), JSON.stringify({ id: "quiet" })); + await writeFile( + path.join(directory, "index.server.ts"), + ` +import { defineRpc } from "@getpaseo/plugin"; +import { z } from "zod"; +export default function contribute(server) { + const counts = {}; + const observe = message => { + if (message.type !== "paseo_frame" || typeof message.data !== "string") return; + const frame = JSON.parse(message.data); + const type = frame.message?.type; + if (type) counts[type] = (counts[type] ?? 0) + 1; + }; + process.on("message", observe); + server.handle(defineRpc({name:"counts", input:z.object({}), output:z.record(z.string(),z.number())}), () => counts); + return () => process.off("message", observe); +}`, + ); + const daemon = await createTestPaseoDaemon(); + const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + const idle = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + const legacy = new DaemonClient({ + url: `ws://127.0.0.1:${daemon.port}/ws`, + capabilities: { + selective_agent_timeline: false, + explicit_event_subscriptions: false, + provider_snapshot_references: false, + }, + }); + const legacyTypes: string[] = []; + legacy.subscribeRawMessages((message) => legacyTypes.push(message.type)); + const idleTypes: string[] = []; + idle.subscribeRawMessages((message) => idleTypes.push(message.type)); + try { + await client.connect(); + await idle.connect(); + await legacy.connect(); + await client.patchDaemonConfig({ pluginsEnabled: true }); + await client.installDirectoryPlugin(directory); + const agent = await client.createAgent({ provider: "claude", cwd: directory }); + const api = createPaseoApi(client); + const received: string[] = []; + const release = api.agents + .ref(agent.id) + .timeline.subscribe((event) => received.push(event.event.type)); + await release.ready; + await daemon.daemon.agentManager.emitLiveTimelineItem(agent.id, { + type: "assistant_message", + id: "row", + text: "only the subscriber sees this", + }); + await expect.poll(() => received).toEqual(["timeline"]); + await client.sendMessage(agent.id, "Say hello"); + await client.waitForFinish(agent.id); + await client.refreshProvidersSnapshot({ cwd: directory }); + const counts = await client.invokePluginRpc("quiet", "counts", {}); + expect(counts).toEqual({ status: 1 }); + await legacy.getDaemonConfig(); + expect(legacyTypes).toEqual( + expect.arrayContaining(["project.update", "providers_snapshot_update", "agent_stream"]), + ); + expect(idleTypes.filter((type) => type !== "status")).toEqual([]); + const deliveredBeforeRelease = received.length; + release(); + await client.getDaemonConfig(); + await daemon.daemon.agentManager.emitLiveTimelineItem(agent.id, { + type: "assistant_message", + id: "row-2", + text: "unsubscribed", + }); + await client.getDaemonConfig(); + expect(received).toHaveLength(deliveredBeforeRelease); + } finally { + await legacy.close(); + await idle.close(); + await client.close(); + await daemon.close(); + await rm(directory, { recursive: true, force: true }); + } +}, 60_000); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 3bda1ec7cc..ae7a105aed 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1,3 +1,4 @@ +import type { SessionEventSubscription } from "@getpaseo/protocol/messages"; import type { AgentRequests } from "./agent/requests/index.js"; import equal from "fast-deep-equal"; import { v4 as uuidv4 } from "uuid"; @@ -715,6 +716,8 @@ export class Session { unsubscribe: () => void; } | null = null; private projectSyncEnabled = false; + private readonly defaultEventSubscriptionSource = {}; + private readonly eventSubscriptions = new Map>(); private readonly workspaceUpdateTails = new Map>(); private clientActivity: { deviceType: "web" | "mobile"; @@ -925,6 +928,7 @@ export class Session { isProviderVisibleToClient: (provider) => this.isProviderVisibleToClient(provider), supportsCustomModeIcons: () => this.supports(CLIENT_CAPS.customModeIcons), supportsCompactProviderSnapshots: () => this.supports(CLIENT_CAPS.compactProviderSnapshots), + wantsSnapshotChanges: () => this.wantsEvent("providers_snapshot_update"), supportsProviderSnapshotReferences: () => this.supports(CLIENT_CAPS.providerSnapshotReferences), listProviderAvailability: () => this.agentManager.listProviderAvailability(), @@ -1135,6 +1139,7 @@ export class Session { updateClientCapabilities(capabilities: Record | null, source?: object): void { this.clientCapabilities = parseClientCapabilities(capabilities); if (source) { + this.eventSubscriptions.delete(source); this.clientCapabilitiesBySource.set(source, this.clientCapabilities); } if (!source && !this.supports(CLIENT_CAPS.selectiveAgentTimeline)) { @@ -1145,6 +1150,7 @@ export class Session { clearAgentTimelineSubscription(source: object): void { this.clientCapabilitiesBySource.delete(source); + this.eventSubscriptions.delete(source); if (this.viewedTimelineAgentIdsBySource.delete(source)) { this.rebuildViewedTimelineAgentIds(); } @@ -1223,6 +1229,7 @@ export class Session { continue; const supportsSelectiveDelivery = capabilities.has(CLIENT_CAPS.selectiveAgentTimeline); if (supportsSelectiveDelivery && serializedEvent.type === "attention_required") { + if (!this.wantsEvent("agent_attention_required", source)) continue; this.onMessageToSource(source, { type: "agent_attention_required", payload: { @@ -1267,6 +1274,7 @@ export class Session { } private async publishProjectUpdate(update: ProjectUpdate): Promise { + if (!this.wantsEvent("project.update")) return; const projectedPayload = update.kind === "upsert" ? { kind: "upsert" as const, project: await this.buildProjectDescriptor(update.project) } @@ -1275,15 +1283,7 @@ export class Session { type: "project.update", payload: this.directorySync.sequenceProjectUpdate(projectedPayload, this.projectSyncEnabled), }; - if (this.clientCapabilitiesBySource.size === 0 || !this.onMessageToSource) { - if (this.supports(CLIENT_CAPS.projectUpdates)) this.emit(message); - return; - } - for (const [source, capabilities] of this.clientCapabilitiesBySource) { - if (capabilities.has(CLIENT_CAPS.projectUpdates)) { - this.onMessageToSource(source, message); - } - } + this.emit(message); } async syncWorkspaceGitObserverForWorkspace(workspace: PersistedWorkspaceRecord): Promise { @@ -2333,6 +2333,20 @@ export class Session { return this.handleProviderSubagentListRequest(msg); case "agent.provider_subagents.timeline.get.request": return this.handleProviderSubagentTimelineRequest(msg, source); + case "session.events.set_subscription.request": { + this.eventSubscriptions.set( + source ?? this.defaultEventSubscriptionSource, + new Set(msg.events), + ); + this.emitForSource( + { + type: "session.events.set_subscription.response", + payload: { requestId: msg.requestId }, + }, + source, + ); + return undefined; + } case "agent.timeline.set_subscription.request": { const agentIds = [...new Set(msg.agentIds)].sort(); if ( @@ -7703,10 +7717,43 @@ export class Session { /** * Emit a message to the client */ + // COMPAT(explicitEventSubscriptions): added in v0.8.0, remove legacy broadcasts after 2027-03-08. + private wantsEvent(event: SessionEventSubscription, source?: object): boolean { + if (!source && this.clientCapabilitiesBySource.size > 0) { + return [...this.clientCapabilitiesBySource.keys()].some((candidate) => + this.wantsEvent(event, candidate), + ); + } + const capabilities = source + ? this.clientCapabilitiesBySource.get(source)! + : this.clientCapabilities; + if (event === "project.update" && !capabilities.has(CLIENT_CAPS.projectUpdates)) return false; + return ( + !capabilities.has(CLIENT_CAPS.explicitEventSubscriptions) || + this.eventSubscriptions.get(source ?? this.defaultEventSubscriptionSource)?.has(event) === + true + ); + } + private emit(msg: SessionOutboundMessage): void { if (!this.authorization.allowsOutbound(msg)) { return; } + if ( + msg.type === "project.update" || + msg.type === "providers_snapshot_update" || + msg.type === "agent_attention_required" || + msg.type === "agent_permission_request" || + msg.type === "agent_permission_resolved" + ) { + if (this.clientCapabilitiesBySource.size > 0 && this.onMessageToSource) { + for (const source of this.clientCapabilitiesBySource.keys()) { + if (this.wantsEvent(msg.type, source)) this.onMessageToSource(source, msg); + } + return; + } + if (!this.wantsEvent(msg.type)) return; + } // JSON.stringify(msg) is only computed when trace is enabled — it runs for // every outbound message otherwise, and trace is disabled by default. // Optional-chained because test logger stubs don't implement isLevelEnabled. diff --git a/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts b/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts index c8b00332fe..4c83cec6a3 100644 --- a/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts +++ b/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts @@ -681,3 +681,13 @@ describe("subscription lifecycle", () => { expect(h.agentUpdates()).toEqual([{ kind: "remove", agentId: "a" }]); }); }); + +test("an idle session skips agent hydration while preserving workspace updates", async () => { + const h = buildHarness(); + const agent = { ...h.managed("unobserved"), workspaceId: "workspace" }; + // No payload is registered: attempting to hydrate this agent would fail. + await h.service.forwardLiveAgent(agent); + expect(h.loggedErrors).toEqual([]); + expect(h.agentUpdates()).toEqual([]); + expect(h.workspaceUpdates).toEqual([agent.workspaceId]); +}); diff --git a/packages/server/src/server/session/agent-updates/agent-updates-service.ts b/packages/server/src/server/session/agent-updates/agent-updates-service.ts index 61919378c5..a4b1579688 100644 --- a/packages/server/src/server/session/agent-updates/agent-updates-service.ts +++ b/packages/server/src/server/session/agent-updates/agent-updates-service.ts @@ -390,6 +390,18 @@ export function createAgentUpdatesService(deps: AgentUpdatesServiceDeps): AgentU } function forwardLiveAgent(agent: ManagedAgent): Promise { + if (!subscription) { + const workspaceId = agent.workspaceId; + return workspaceId + ? enqueueAgentUpdate(agent.id, async () => { + try { + await deps.emitWorkspaceUpdateForWorkspaceId(workspaceId); + } catch (error) { + deps.logger.error({ err: error }, "Failed to emit workspace update"); + } + }) + : Promise.resolve(); + } const payload = toAgentPayload(agent); return enqueueAgentUpdate(payload.id, () => emitLiveAgentUpdate(payload)); } diff --git a/packages/server/src/server/session/provider/provider-catalog-session.test.ts b/packages/server/src/server/session/provider/provider-catalog-session.test.ts index 653f46917c..b72a5142d9 100644 --- a/packages/server/src/server/session/provider/provider-catalog-session.test.ts +++ b/packages/server/src/server/session/provider/provider-catalog-session.test.ts @@ -54,6 +54,7 @@ function makeSubsystem(options: MakeOptions = {}) { isProviderVisibleToClient: (provider) => visible.has(provider), supportsCustomModeIcons: () => options.supportsCustomModeIcons ?? false, supportsProviderSnapshotReferences: () => false, + wantsSnapshotChanges: () => true, supportsCompactProviderSnapshots: () => options.supportsCompactProviderSnapshots ?? false, listProviderAvailability: async () => [], listDraftFeatures: async () => [], @@ -395,6 +396,7 @@ it("announces shared content without retransmitting models or hashing discovery isProviderVisibleToClient: () => true, supportsCustomModeIcons: () => true, supportsCompactProviderSnapshots: () => true, + wantsSnapshotChanges: () => true, supportsProviderSnapshotReferences: () => references, listProviderAvailability: async () => [], listDraftFeatures: async () => [], diff --git a/packages/server/src/server/session/provider/provider-catalog-session.ts b/packages/server/src/server/session/provider/provider-catalog-session.ts index f82f1ac56c..7175cabbc9 100644 --- a/packages/server/src/server/session/provider/provider-catalog-session.ts +++ b/packages/server/src/server/session/provider/provider-catalog-session.ts @@ -48,6 +48,7 @@ export interface ProviderCatalogSessionHost { supportsCustomModeIcons(): boolean; supportsCompactProviderSnapshots(): boolean; supportsProviderSnapshotReferences(): boolean; + wantsSnapshotChanges(): boolean; listProviderAvailability(): Promise; listDraftFeatures(config: AgentSessionConfig): Promise; } @@ -82,6 +83,7 @@ export class ProviderCatalogSession { start(): void { const handleProviderSnapshotChange = (transition: ProviderSnapshotTransition) => { + if (!this.host.wantsSnapshotChanges()) return; const previous = this.visibleSnapshot(transition.previous); const current = this.visibleSnapshot(transition.current); if (sameSnapshotRecords(previous.records, current.records)) return; diff --git a/packages/server/src/server/voice-local-agent.e2e.test.ts b/packages/server/src/server/voice-local-agent.e2e.test.ts index 9dbe49d74a..29841b7938 100644 --- a/packages/server/src/server/voice-local-agent.e2e.test.ts +++ b/packages/server/src/server/voice-local-agent.e2e.test.ts @@ -32,7 +32,7 @@ function makeErrorHandler(reject: (error: Error) => void) { } function makeSpeakToolHandler(resolve: (value: string) => void) { - return (msg: SessionMessage<"agent_stream">) => { + return (msg: SessionMessage<"agent_stream" | "agent.timeline.replacement">) => { if (msg.type !== "agent_stream") return; if (msg.payload.event.type !== "timeline") return; const item = msg.payload.event.item; @@ -131,7 +131,10 @@ function waitForSignal( }); const speakToolPromise = waitForSignal(120000, (resolve, reject) => { - const offStream = ctx.client.on("agent_stream", makeSpeakToolHandler(resolve)); + const offStream = ctx.client.subscribeAgentTimeline( + targetAgent.id, + makeSpeakToolHandler(resolve), + ); const offError = ctx.client.on("activity_log", makeErrorHandler(reject)); return () => { offStream(); diff --git a/packages/server/src/server/voice-roundtrip.e2e.test.ts b/packages/server/src/server/voice-roundtrip.e2e.test.ts index 8808421b3d..97096c9938 100644 --- a/packages/server/src/server/voice-roundtrip.e2e.test.ts +++ b/packages/server/src/server/voice-roundtrip.e2e.test.ts @@ -225,7 +225,7 @@ for (const targetProvider of [ const timelineToolAgentIds = new Set(); const activityErrors: string[] = []; - const offStream = ctx.client.on("agent_stream", (message) => { + const offStream = ctx.client.subscribeAgentTimeline(voiceAgentId, (message) => { if (message.type !== "agent_stream") { return; } diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 8fba3c27ac..4d56418152 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1755,6 +1755,7 @@ export class VoiceAssistantWebSocketServer { forgeProviders: true, // COMPAT(selectiveAgentTimeline): added in v0.1.106, remove after 2027-01-12. selectiveAgentTimeline: true, + explicitEventSubscriptions: true, // COMPAT(canonicalSubmittedPrompts): added in v0.2.6, remove gate after 2027-01-30. canonicalSubmittedPrompts: true, // COMPAT(stableProjectIdentity): added in v0.1.109, remove gate after 2027-01-15. diff --git a/public-docs/plugins/v0.8/reference.md b/public-docs/plugins/v0.8/reference.md index 2cad2a163a..d8166f3204 100644 --- a/public-docs/plugins/v0.8/reference.md +++ b/public-docs/plugins/v0.8/reference.md @@ -1478,7 +1478,7 @@ export default function contribute(server: PluginServerContext) { Inputs and outputs are validated on both sides. RPC names start with a lowercase letter and contain lowercase letters, numbers, dots, hyphens, or underscores. `useRpc()` returns a typed async function. Use TanStack Query for request state, caching, and mutations. -Backend handlers receive the same `PaseoApi` as `{ paseo }`. Their connection belongs to the subprocess and closes when the plugin stops. Backend code can use Node APIs and dependencies installed in the plugin directory. +Backend handlers receive the same `PaseoApi` as `{ paseo }`. Their connection belongs to the subprocess and closes when the plugin stops. It does not subscribe to timelines or catalog events until plugin code subscribes. Follow the [SDK event contract](../../sdk/events.md) for cleanup and timeline replacements. Backend code can use Node APIs and dependencies installed in the plugin directory. ## Debug backend output diff --git a/public-docs/sdk/events.md b/public-docs/sdk/events.md index 36384554e5..dafba6bcb3 100644 --- a/public-docs/sdk/events.md +++ b/public-docs/sdk/events.md @@ -10,7 +10,7 @@ category: TypeScript SDK Subscriptions report changes after they happen. Fetch an initial snapshot first, then apply updates to it. -Every `subscribe()` method returns a local unsubscribe function. It removes your callback; it does not stop or archive the underlying resource. +Every `subscribe()` method returns an unsubscribe function. Timeline, project, and provider subscriptions establish network demand and restore it after reconnect. Unsubscribing releases that demand when its last listener leaves. Agent and workspace directories use the explicit `list({ subscribe })` bootstrap shown below. Unsubscribing never stops or archives the underlying resource. ## Follow one agent's status @@ -39,17 +39,25 @@ The handle updates its properties and `current()` value before it calls your han ## Follow timeline events ```ts -const unsubscribe = agent.timeline.subscribe(({ event, timestamp }) => { +const unsubscribe = agent.timeline.subscribe((update) => { + const { event } = update; + if (event.type === "replacement") { + // Previously fetched history belongs to an old epoch. Fetch the page your UI needs. + void agent.timeline.refetch().then((page) => console.log(page.entries)); + return; + } if (event.type === "timeline" && event.item.type === "assistant_message") { process.stdout.write(event.item.text); } if (event.type === "turn_completed") { - console.log(`\nCompleted at ${timestamp}`); + console.log("\nTurn completed"); } }); ``` +Await `unsubscribe.ready` before starting work whose events you need to observe. It resolves when the daemon acknowledges the subscription and rejects if establishment fails. Call `unsubscribe()` to release it. + 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. Turn completion comes from `turn_completed`, `turn_failed`, or `turn_canceled`. Do not infer turn completion from an `agent_update` transition to `idle`. diff --git a/public-docs/sdk/reference.md b/public-docs/sdk/reference.md index 0f8bf7200f..85e4d60b9d 100644 --- a/public-docs/sdk/reference.md +++ b/public-docs/sdk/reference.md @@ -108,7 +108,7 @@ 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)` listens for stream events belonging to the agent and returns a local unsubscribe function. +`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. The handler also receives `{ agentId, event: { type: "replacement", epoch } }` when history is replaced; refetch the page you need. See [events](./events.md#follow-timeline-events). ## `client.projects`