From 02298d226c8a9abb46f4fdb4638cdaefe229b777 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 12 Jul 2026 23:13:27 +0200 Subject: [PATCH] fix(sync): preserve mixed-client delivery guarantees --- packages/app/src/panels/agent-panel.tsx | 3 +- packages/server/src/server/session.test.ts | 18 ++++++ packages/server/src/server/session.ts | 56 +++++++++++++++---- .../server/src/server/websocket-server.ts | 3 +- 4 files changed, 68 insertions(+), 12 deletions(-) diff --git a/packages/app/src/panels/agent-panel.tsx b/packages/app/src/panels/agent-panel.tsx index b6b9f1ebb6..cb3920f604 100644 --- a/packages/app/src/panels/agent-panel.tsx +++ b/packages/app/src/panels/agent-panel.tsx @@ -907,7 +907,7 @@ function ChatAgentContent({ if (!agentId) { return; } - if (agentState.id) { + if (agentState.id && hasAppliedAuthoritativeHistory) { if (missingAgentState.kind === "resolving" || missingAgentState.kind === "not_found") { setMissingAgentState(reconcileMissingAgentStateWithPresentAgent); } @@ -964,6 +964,7 @@ function ChatAgentContent({ }); }, [ agentState.id, + hasAppliedAuthoritativeHistory, agentId, client, ensureAgentIsInitialized, diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index f0f96a1307..685388b133 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -4592,6 +4592,8 @@ test("unions viewed timelines across socket sources and removes detached sources session.updateClientCapabilities({ selective_agent_timeline: true }); const firstSocket = {}; const secondSocket = {}; + session.updateClientCapabilities({ selective_agent_timeline: true }, firstSocket); + session.updateClientCapabilities({ selective_agent_timeline: true }, secondSocket); await session.handleMessage( { @@ -4635,6 +4637,22 @@ test("unions viewed timelines across socket sources and removes detached sources }); expect(messages.filter((message) => message.type === "agent_stream")).toHaveLength(2); + const legacySocket = {}; + session.updateClientCapabilities(null, legacySocket); + messages.length = 0; + forward({ + type: "agent_stream", + agentId: "agent-not-viewed", + event: { + type: "timeline", + provider: "mock", + item: { type: "assistant_message", messageId: "message-legacy", text: "legacy" }, + }, + }); + expect(messages.some((message) => message.type === "agent_stream")).toBe(true); + + session.clearAgentTimelineSubscription(legacySocket); + session.clearAgentTimelineSubscription(firstSocket); messages.length = 0; forward({ diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index c8eb93e3bb..71ca860cb6 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -585,6 +585,7 @@ export class Session { private unsubscribeAgentEvents: (() => void) | null = null; private viewedTimelineAgentIds = new Set(); private readonly viewedTimelineAgentIdsBySource = new Map>(); + private readonly selectiveTimelineCapabilityBySource = new Map(); private readonly defaultTimelineSubscriptionSource = {}; private unsubscribeTerminalWorkspaceContributionEvents: (() => void) | null = null; private readonly agentUpdates: AgentUpdatesService; @@ -966,17 +967,25 @@ export class Session { } } - updateClientCapabilities(capabilities: Record | null): void { + updateClientCapabilities(capabilities: Record | null, source?: object): void { this.clientCapabilities = parseClientCapabilities(capabilities); - if (!this.supports(CLIENT_CAPS.selectiveAgentTimeline)) { + if (source) { + this.selectiveTimelineCapabilityBySource.set( + source, + this.supports(CLIENT_CAPS.selectiveAgentTimeline), + ); + } + if (!source && !this.supports(CLIENT_CAPS.selectiveAgentTimeline)) { this.viewedTimelineAgentIdsBySource.clear(); this.viewedTimelineAgentIds.clear(); } } clearAgentTimelineSubscription(source: object): void { - if (!this.viewedTimelineAgentIdsBySource.delete(source)) return; - this.rebuildViewedTimelineAgentIds(); + this.selectiveTimelineCapabilityBySource.delete(source); + if (this.viewedTimelineAgentIdsBySource.delete(source)) { + this.rebuildViewedTimelineAgentIds(); + } } private replaceAgentTimelineSubscription(source: object | undefined, agentIds: string[]): void { @@ -994,6 +1003,16 @@ export class Session { this.viewedTimelineAgentIds = viewedAgentIds; } + private usesSelectiveTimelineDelivery(): boolean { + if (this.selectiveTimelineCapabilityBySource.size === 0) { + return this.supports(CLIENT_CAPS.selectiveAgentTimeline); + } + for (const capable of this.selectiveTimelineCapabilityBySource.values()) { + if (!capable) return false; + } + return true; + } + supports(capability: ClientCapability): boolean { return this.clientCapabilities.has(capability); } @@ -1308,10 +1327,7 @@ export class Session { "agent.session.forward_stream", ); - if ( - this.supports(CLIENT_CAPS.selectiveAgentTimeline) && - serializedEvent.type === "attention_required" - ) { + if (this.usesSelectiveTimelineDelivery() && serializedEvent.type === "attention_required") { this.emit({ type: "agent_attention_required", payload: { @@ -1325,7 +1341,7 @@ export class Session { }, }); } else if ( - !this.supports(CLIENT_CAPS.selectiveAgentTimeline) || + !this.usesSelectiveTimelineDelivery() || this.viewedTimelineAgentIds.has(event.agentId) ) { this.emit({ @@ -1587,7 +1603,12 @@ export class Session { return this.handleProviderSubagentTimelineRequest(msg); case "agent.timeline.set_subscription.request": { const agentIds = [...new Set(msg.agentIds)].sort(); - if (this.supports(CLIENT_CAPS.selectiveAgentTimeline)) { + if ( + source + ? (this.selectiveTimelineCapabilityBySource.get(source) ?? + this.supports(CLIENT_CAPS.selectiveAgentTimeline)) + : this.supports(CLIENT_CAPS.selectiveAgentTimeline) + ) { this.replaceAgentTimelineSubscription(source, agentIds); } this.emit({ @@ -4590,6 +4611,7 @@ export class Session { "fetch_workspaces_response_ready", ); const snapshot = this.buildBootstrapSnapshot(payload.entries); + this.seedWorkspaceSubscriptionSnapshot(subscriptionId, payload.entries); this.emit({ type: "fetch_workspaces_response", @@ -4650,6 +4672,20 @@ export class Session { return { snapshotByWorkspaceId }; } + private seedWorkspaceSubscriptionSnapshot( + subscriptionId: string | null, + entries: FetchWorkspacesResponseEntry[], + ): void { + const subscription = this.workspaceUpdatesSubscription; + if (!subscriptionId || subscription?.subscriptionId !== subscriptionId) return; + for (const entry of entries) { + subscription.lastEmittedByWorkspaceId.set(entry.id, { + kind: "upsert", + workspace: entry, + }); + } + } + private async registerWorkspaceForImportedAgent( workspace: PersistedWorkspaceRecord, ): Promise { diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 68cd8025bc..b8b9b1d8e3 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1093,6 +1093,7 @@ export class VoiceAssistantWebSocketServer { sockets: new Set([ws]), externalDisconnectCleanupTimeout: null, }; + session.updateClientCapabilities(clientCapabilities, ws); return connection; } @@ -1166,7 +1167,7 @@ export class VoiceAssistantWebSocketServer { // COMPAT(selectiveAgentTimeline): added in v0.1.106. Every capable resumed // hello resets membership before server_info so stale retained-session // state cannot leak. Remove after 2027-01-12. - existing.session.updateClientCapabilities(newClientCapabilities); + existing.session.updateClientCapabilities(newClientCapabilities, ws); if ( JSON.stringify(existing.clientCapabilities ?? null) !== JSON.stringify(newClientCapabilities ?? null)