From 818ab23b50fd642b27a1c50804415b7bd857d5b1 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 12 Jul 2026 22:39:15 +0200 Subject: [PATCH] fix(sync): preserve selective delivery boundaries Keep selective timeline capability app-owned, union viewed sets across shared sockets, and reconcile directory side effects from accepted state. Visibility and archive suppression now follow their existing authoritative boundaries. --- packages/app/src/contexts/session-context.tsx | 13 ++- ...workspace-directory-reconciliation.test.ts | 22 +++++ .../workspace-directory-reconciliation.ts | 8 +- packages/app/src/runtime/host-runtime.test.ts | 30 ++++++- packages/app/src/runtime/host-runtime.ts | 13 ++- packages/client/src/daemon-client.test.ts | 2 +- packages/client/src/daemon-client.ts | 1 - .../agent/providers/opencode-agent.test.ts | 8 +- packages/server/src/server/session.test.ts | 86 +++++++++++++++++++ packages/server/src/server/session.ts | 40 +++++++-- .../websocket-server.relay-reconnect.test.ts | 4 + .../server/src/server/websocket-server.ts | 3 +- 12 files changed, 207 insertions(+), 23 deletions(-) diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 55d0898465..89c6b8c945 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -4,6 +4,7 @@ import { AppState } from "react-native"; import { useQueryClient } from "@tanstack/react-query"; import { useTranslation } from "react-i18next"; import { useClientActivity } from "@/hooks/use-client-activity"; +import { useAppVisible } from "@/hooks/use-app-visible"; import { usePushTokenRegistration } from "@/hooks/use-push-token-registration"; import { clearArchiveAgentPending } from "@/hooks/use-archive-agent"; import { @@ -533,6 +534,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const audioOutputBuffersRef = useRef>(new Map()); const activeAudioGroupsRef = useRef>(new Set()); const workspaceHydrationRef = useRef(null); + const isAppVisible = useAppVisible(); const applyWorkspaceUpdatePayload = useCallback( (payload: WorkspaceUpdatePayload) => { @@ -564,7 +566,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider useEffect(() => { const subscription = AppState.addEventListener("change", (nextState) => { appStateRef.current = nextState; - viewedTimelineSyncRef.current?.setActive(getIsAppActivelyVisible(nextState)); }); return () => { @@ -572,6 +573,10 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider }; }, []); + useEffect(() => { + viewedTimelineSyncRef.current?.setActive(isAppVisible); + }, [isAppVisible]); + const hydrateWorkspaces = useCallback( async (options?: { subscribe?: boolean; isCancelled?: () => boolean }) => { if (!client || !isConnected) { @@ -618,7 +623,11 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider setWorkspaces( serverId, - reconcileWorkspaceDirectory({ snapshot: snapshot.workspaces, deltas: snapshot.deltas }), + reconcileWorkspaceDirectory({ + serverId, + snapshot: snapshot.workspaces, + deltas: snapshot.deltas, + }), ); setEmptyProjects(serverId, snapshot.emptyProjects.values()); setHasHydratedWorkspaces(serverId, true); diff --git a/packages/app/src/contexts/workspace-directory-reconciliation.test.ts b/packages/app/src/contexts/workspace-directory-reconciliation.test.ts index a9d6d85083..851f784c12 100644 --- a/packages/app/src/contexts/workspace-directory-reconciliation.test.ts +++ b/packages/app/src/contexts/workspace-directory-reconciliation.test.ts @@ -1,8 +1,14 @@ import { expect, it } from "vitest"; import type { WorkspaceDescriptorPayload } from "@getpaseo/protocol/messages"; import { normalizeWorkspaceDescriptor } from "@/stores/session-store"; +import { + clearWorkspaceArchivePending, + markWorkspaceArchivePending, +} from "./session-workspace-upserts"; import { reconcileWorkspaceDirectory } from "./workspace-directory-reconciliation"; +const SERVER_ID = "workspace-directory-reconciliation"; + function workspace(id: string, title: string): WorkspaceDescriptorPayload { return { id, @@ -25,6 +31,7 @@ function workspace(id: string, title: string): WorkspaceDescriptorPayload { it("keeps workspace upserts and removals received during later pages", () => { const result = reconcileWorkspaceDirectory({ + serverId: SERVER_ID, snapshot: new Map([ ["updated", normalizeWorkspaceDescriptor(workspace("updated", "snapshot"))], ["removed", normalizeWorkspaceDescriptor(workspace("removed", "snapshot"))], @@ -39,3 +46,18 @@ it("keeps workspace upserts and removals received during later pages", () => { ["updated", "live"], ]); }); + +it("does not restore a locally archiving workspace from a buffered upsert", () => { + markWorkspaceArchivePending({ serverId: SERVER_ID, workspaceId: "archiving" }); + try { + const result = reconcileWorkspaceDirectory({ + serverId: SERVER_ID, + snapshot: new Map(), + deltas: [{ kind: "upsert", workspace: workspace("archiving", "live") }], + }); + + expect(result.has("archiving")).toBe(false); + } finally { + clearWorkspaceArchivePending({ serverId: SERVER_ID, workspaceId: "archiving" }); + } +}); diff --git a/packages/app/src/contexts/workspace-directory-reconciliation.ts b/packages/app/src/contexts/workspace-directory-reconciliation.ts index 2fcb5e3af2..4be9cdd9ff 100644 --- a/packages/app/src/contexts/workspace-directory-reconciliation.ts +++ b/packages/app/src/contexts/workspace-directory-reconciliation.ts @@ -1,9 +1,11 @@ import type { SessionOutboundMessage } from "@getpaseo/protocol/messages"; import { normalizeWorkspaceDescriptor, type WorkspaceDescriptor } from "@/stores/session-store"; +import { shouldSuppressWorkspaceForLocalArchive } from "./session-workspace-upserts"; type WorkspaceDelta = Extract["payload"]; export function reconcileWorkspaceDirectory(input: { + serverId: string; snapshot: ReadonlyMap; deltas: readonly WorkspaceDelta[]; }): Map { @@ -13,7 +15,11 @@ export function reconcileWorkspaceDirectory(input: { workspaces.delete(delta.id); } else { const workspace = normalizeWorkspaceDescriptor(delta.workspace); - workspaces.set(workspace.id, workspace); + if (shouldSuppressWorkspaceForLocalArchive({ serverId: input.serverId, workspace })) { + workspaces.delete(workspace.id); + } else { + workspaces.set(workspace.id, workspace); + } } } return workspaces; diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index f679b1f18d..db64b571d4 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -1761,6 +1761,12 @@ describe("HostRuntimeStore", () => { updatedAt: "2026-07-12T10:00:00.000Z", title: "snapshot", }); + const recoveredAgent = makeFetchAgentsEntry({ + id: "agent-recovered", + cwd: "/repo", + updatedAt: "2026-07-12T10:00:00.000Z", + title: "before refresh", + }); fakeClient.fetchAgentsResponses.push( makeFetchAgentsPayload({ entries: [pageOneAgent], @@ -1781,9 +1787,16 @@ describe("HostRuntimeStore", () => { getClientId: async () => "cid_paged_delta", }, }); - useSessionStore - .getState() - .initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1); + const sessionStore = useSessionStore.getState(); + sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1); + sessionStore.setAgents( + host.serverId, + new Map([[recoveredAgent.agent.id, replicaAgent(recoveredAgent.agent, host.serverId)]]), + ); + sessionStore.setAgentTimelineCursor( + host.serverId, + new Map([[recoveredAgent.agent.id, { epoch: "epoch", startSeq: 10, endSeq: 20 }]]), + ); store.syncHosts([host]); await fakeClient.waitForFetches(2); @@ -1792,6 +1805,11 @@ describe("HostRuntimeStore", () => { agent: { ...pageOneAgent.agent, title: "live" }, project: pageOneAgent.project, }); + fakeClient.agentUpdate({ + kind: "upsert", + agent: { ...recoveredAgent.agent, title: "recovered by delta" }, + project: recoveredAgent.project, + }); finishPageTwo( makeFetchAgentsPayload({ entries: [ @@ -1812,7 +1830,13 @@ describe("HostRuntimeStore", () => { ).toEqual([ ["agent-a", "live"], ["agent-b", null], + ["agent-recovered", "recovered by delta"], ]); + expect( + useSessionStore + .getState() + .sessions[host.serverId]?.agentTimelineCursor.get(recoveredAgent.agent.id), + ).toEqual({ epoch: "epoch", startSeq: 10, endSeq: 20 }); const agentB = makeFetchAgentsEntry({ id: "agent-b", diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 1cf0f46f94..4b38b48dd7 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -554,6 +554,10 @@ function createDefaultDeps(): HostRuntimeControllerDeps { }, } : undefined; + const appCapabilities = { + [CLIENT_CAPS.selectiveAgentTimeline]: true, + ...browserAutomationCapabilities, + }; return { createClient: ({ host, connection, clientId, runtimeGeneration }) => { @@ -564,7 +568,7 @@ function createDefaultDeps(): HostRuntimeControllerDeps { clientType: "mobile" as const, appVersion: resolveAppVersion() ?? undefined, runtimeGeneration, - ...(browserAutomationCapabilities ? { capabilities: browserAutomationCapabilities } : {}), + capabilities: appCapabilities, }; if (connection.type === "directSocket" || connection.type === "directPipe") { return new DaemonClient({ @@ -602,7 +606,7 @@ function createDefaultDeps(): HostRuntimeControllerDeps { connectToDaemon(connection, { ...(host.serverId ? { serverId: host.serverId } : {}), ...(timeoutMs !== undefined ? { timeoutMs } : {}), - ...(browserAutomationCapabilities ? { capabilities: browserAutomationCapabilities } : {}), + capabilities: appCapabilities, }), getClientId: () => getOrCreateClientId(), mountClientHandlers: ({ client, host }) => { @@ -2327,6 +2331,7 @@ export class HostRuntimeStore { this.applyAgentDirectoryCommitSideEffects({ serverId: input.serverId, previous, + committedEntries: reconciled.entries, directory: { entries: stampedSnapshot, deltas: transaction.deltas, @@ -2379,6 +2384,7 @@ export class HostRuntimeStore { this.applyAgentDirectoryCommitSideEffects({ serverId: input.serverId, previous, + committedEntries: reconciled.entries, directory, stoppedRunningAgentIds: reconciled.stoppedRunningAgentIds, }); @@ -2390,10 +2396,11 @@ export class HostRuntimeStore { private applyAgentDirectoryCommitSideEffects(input: { serverId: string; previous: ReadonlyMap; + committedEntries: FetchAgentsEntry[]; directory: AgentDirectoryFetchResult; stoppedRunningAgentIds: string[]; }): void { - const snapshotAgentIds = new Set(input.directory.entries.map((entry) => entry.agent.id)); + const snapshotAgentIds = new Set(input.committedEntries.map((entry) => entry.agent.id)); for (const agentId of input.previous.keys()) { if (!snapshotAgentIds.has(agentId)) { applyAgentDirectoryDelta({ serverId: input.serverId, delta: { kind: "remove", agentId } }); diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index 383cdde5ad..ad41bdddac 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -186,6 +186,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(); }); test("advertises consumer-provided browser automation capabilities", async () => { @@ -647,7 +648,6 @@ test("advertises client capabilities in hello", async () => { custom_mode_icons: true, provider_subagents: true, reasoning_merge_enum: true, - selective_agent_timeline: true, terminal_reflowable_snapshot: true, browser_host: { supportedCommands: ["list_tabs"], diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index aa270ca76c..91bd350a34 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -5108,7 +5108,6 @@ export class DaemonClient { protocolVersion: 1, capabilities: { [CLIENT_CAPS.customModeIcons]: true, - [CLIENT_CAPS.selectiveAgentTimeline]: true, [CLIENT_CAPS.reasoningMergeEnum]: true, [CLIENT_CAPS.terminalReflowableSnapshot]: true, [CLIENT_CAPS.providerSubagents]: true, diff --git a/packages/server/src/server/agent/providers/opencode-agent.test.ts b/packages/server/src/server/agent/providers/opencode-agent.test.ts index 2cccfcbeec..829422ea71 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -3033,11 +3033,11 @@ describe("OpenCode provider subagent contract", () => { event: { type: "timeline", id: "ses_child_with_history", - item: { + item: expect.objectContaining({ type: "assistant_message", text: "Persisted child result.", messageId: "msg_child_history", - }, + }), timestamp: "1970-01-01T00:00:02.000Z", }, }), @@ -3111,11 +3111,11 @@ describe("OpenCode provider subagent contract", () => { event: expect.objectContaining({ type: "timeline", id: "ses_child_hydrating", - item: { + item: expect.objectContaining({ type: "assistant_message", text: "Hydration did not lose this.", messageId: "msg_child_hydrating", - }, + }), }), }), ), diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index 78e4220c24..f0f96a1307 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -21,6 +21,7 @@ import { Session } from "./session.js"; import { DownloadTokenStore } from "./file-download/token-store.js"; import { StructuredAgentFallbackError } from "./agent/agent-response-loop.js"; import type { StoredAgentRecord } from "./agent/agent-storage.js"; +import type { AgentManagerEvent } from "./agent/agent-manager.js"; import type { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js"; import type { SessionOptions } from "./session.js"; import type { SessionInboundMessage, SessionOutboundMessage } from "./messages.js"; @@ -4576,6 +4577,91 @@ test("replaces a capable session's complete viewed timeline set", async () => { ]); }); +test("unions viewed timelines across socket sources and removes detached sources", async () => { + const messages: SessionOutboundMessage[] = []; + const agentEventListeners: Array<(event: AgentManagerEvent) => void> = []; + const session = createSessionForTest({ + messages, + agentManager: { + subscribe: vi.fn((listener: (event: AgentManagerEvent) => void) => { + agentEventListeners.push(listener); + return () => {}; + }), + }, + }); + session.updateClientCapabilities({ selective_agent_timeline: true }); + const firstSocket = {}; + const secondSocket = {}; + + await session.handleMessage( + { + type: "agent.timeline.set_subscription.request", + agentIds: ["agent-a"], + requestId: "timeline-subscription-a", + }, + firstSocket, + ); + await session.handleMessage( + { + type: "agent.timeline.set_subscription.request", + agentIds: ["agent-b"], + requestId: "timeline-subscription-b", + }, + secondSocket, + ); + messages.length = 0; + + if (agentEventListeners.length === 0) throw new Error("Agent event listener was not installed"); + const forward = (event: AgentManagerEvent) => { + for (const listener of agentEventListeners) listener(event); + }; + forward({ + type: "agent_stream", + agentId: "agent-a", + event: { + type: "timeline", + provider: "mock", + item: { type: "assistant_message", messageId: "message-a", text: "A" }, + }, + }); + forward({ + type: "agent_stream", + agentId: "agent-b", + event: { + type: "timeline", + provider: "mock", + item: { type: "assistant_message", messageId: "message-b", text: "B" }, + }, + }); + expect(messages.filter((message) => message.type === "agent_stream")).toHaveLength(2); + + session.clearAgentTimelineSubscription(firstSocket); + messages.length = 0; + forward({ + type: "agent_stream", + agentId: "agent-a", + event: { + type: "timeline", + provider: "mock", + item: { type: "assistant_message", messageId: "message-a-2", text: "detached A" }, + }, + }); + forward({ + type: "agent_stream", + agentId: "agent-b", + event: { + type: "timeline", + provider: "mock", + item: { type: "assistant_message", messageId: "message-b-2", text: "retained B" }, + }, + }); + expect( + messages.flatMap((message) => + message.type === "agent_stream" ? [message.payload.agentId] : [], + ), + ).toEqual(["agent-b"]); +}); + describe("agent config setters", () => { test("set_agent_mode_request: success emits accepted response carrying the notice", async () => { const messages: SessionOutboundMessage[] = []; diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index efffedc795..c8eb93e3bb 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -584,6 +584,8 @@ export class Session { private readonly pushTokenStore: PushTokenStore; private unsubscribeAgentEvents: (() => void) | null = null; private viewedTimelineAgentIds = new Set(); + private readonly viewedTimelineAgentIdsBySource = new Map>(); + private readonly defaultTimelineSubscriptionSource = {}; private unsubscribeTerminalWorkspaceContributionEvents: (() => void) | null = null; private readonly agentUpdates: AgentUpdatesService; private workspaceUpdatesSubscription: WorkspaceUpdatesSubscriptionState | null = null; @@ -966,11 +968,32 @@ export class Session { updateClientCapabilities(capabilities: Record | null): void { this.clientCapabilities = parseClientCapabilities(capabilities); - if (this.supports(CLIENT_CAPS.selectiveAgentTimeline)) { + if (!this.supports(CLIENT_CAPS.selectiveAgentTimeline)) { + this.viewedTimelineAgentIdsBySource.clear(); this.viewedTimelineAgentIds.clear(); } } + clearAgentTimelineSubscription(source: object): void { + if (!this.viewedTimelineAgentIdsBySource.delete(source)) return; + this.rebuildViewedTimelineAgentIds(); + } + + private replaceAgentTimelineSubscription(source: object | undefined, agentIds: string[]): void { + const subscriptionSource = source ?? this.defaultTimelineSubscriptionSource; + if (agentIds.length === 0) this.viewedTimelineAgentIdsBySource.delete(subscriptionSource); + else this.viewedTimelineAgentIdsBySource.set(subscriptionSource, new Set(agentIds)); + this.rebuildViewedTimelineAgentIds(); + } + + private rebuildViewedTimelineAgentIds(): void { + const viewedAgentIds = new Set(); + for (const agentIds of this.viewedTimelineAgentIdsBySource.values()) { + for (const agentId of agentIds) viewedAgentIds.add(agentId); + } + this.viewedTimelineAgentIds = viewedAgentIds; + } + supports(capability: ClientCapability): boolean { return this.clientCapabilities.has(capability); } @@ -1412,7 +1435,7 @@ export class Session { /** * Main entry point for processing session messages */ - public async handleMessage(msg: SessionInboundMessage): Promise { + public async handleMessage(msg: SessionInboundMessage, source?: object): Promise { this.inflightRequests++; if (this.inflightRequests > this.peakInflightRequests) { this.peakInflightRequests = this.inflightRequests; @@ -1426,7 +1449,7 @@ export class Session { "agent.session.inbound", ); try { - await this.dispatchInboundMessage(msg); + await this.dispatchInboundMessage(msg, source); } catch (error) { const err = error instanceof Error ? error : new Error(String(error)); this.sessionLogger.error({ err }, "Error handling message"); @@ -1464,12 +1487,12 @@ export class Session { } } - private async dispatchInboundMessage(msg: SessionInboundMessage): Promise { + private async dispatchInboundMessage(msg: SessionInboundMessage, source?: object): Promise { const promise = this.dispatchVoiceAndControlMessage(msg) ?? this.dispatchAgentRewindMessage(msg) ?? this.dispatchAgentRelationshipMessage(msg) ?? - this.dispatchAgentTimelineMessage(msg) ?? + this.dispatchAgentTimelineMessage(msg, source) ?? this.dispatchAgentLifecycleMessage(msg) ?? this.dispatchAgentConfigMessage(msg) ?? this.dispatchCheckoutMessage(msg) ?? @@ -1551,7 +1574,10 @@ export class Session { } } - private dispatchAgentTimelineMessage(msg: SessionInboundMessage): Promise | undefined { + private dispatchAgentTimelineMessage( + msg: SessionInboundMessage, + source?: object, + ): Promise | undefined { switch (msg.type) { case "fetch_agent_timeline_request": return this.handleFetchAgentTimelineRequest(msg); @@ -1562,7 +1588,7 @@ export class Session { case "agent.timeline.set_subscription.request": { const agentIds = [...new Set(msg.agentIds)].sort(); if (this.supports(CLIENT_CAPS.selectiveAgentTimeline)) { - this.viewedTimelineAgentIds = new Set(agentIds); + this.replaceAgentTimelineSubscription(source, agentIds); } this.emit({ type: "agent.timeline.set_subscription.response", diff --git a/packages/server/src/server/websocket-server.relay-reconnect.test.ts b/packages/server/src/server/websocket-server.relay-reconnect.test.ts index be49b8dae5..cd077f0375 100644 --- a/packages/server/src/server/websocket-server.relay-reconnect.test.ts +++ b/packages/server/src/server/websocket-server.relay-reconnect.test.ts @@ -52,6 +52,10 @@ const sessionMock = vi.hoisted(() => { handleMessage = vi.fn(async () => {}); handleBinaryFrame = vi.fn((_frame: unknown) => {}); supports = vi.fn((capability: string) => this.args.clientCapabilities?.[capability] === true); + updateClientCapabilities = vi.fn((capabilities: Record | null) => { + this.args.clientCapabilities = capabilities; + }); + clearAgentTimelineSubscription = vi.fn(); getClientActivity = vi.fn(() => null); getSessionId = vi.fn(() => "mock-session-id"); resetPeakInflight = vi.fn(() => {}); diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 58a558246f..68cd8025bc 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1395,6 +1395,7 @@ export class VoiceAssistantWebSocketServer { this.sessions.delete(ws); connection.sockets.delete(ws); + connection.session.clearAgentTimelineSubscription(ws); this.socketIdentities.delete(ws); if (connection.sockets.size === 0) { @@ -1763,7 +1764,7 @@ export class VoiceAssistantWebSocketServer { } const startMs = performance.now(); - await activeConnection.session.handleMessage(message.message); + await activeConnection.session.handleMessage(message.message, ws); const durationMs = performance.now() - startMs; this.recordRequestLatency(message.message.type, durationMs);