From 7d838a0dfb227149d51ae56d89ad1b99fdf94ef6 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 17 Jul 2026 12:23:59 +0200 Subject: [PATCH] fix selective timeline compatibility races --- packages/app/src/contexts/session-context.tsx | 6 +- .../app/src/hooks/use-agent-initialization.ts | 3 +- packages/app/src/runtime/host-runtime.test.ts | 71 +++++++++++++++++++ packages/app/src/runtime/host-runtime.ts | 11 ++- .../fetch-agent-timeline-once.test.ts | 30 ++++++++ .../src/timeline/fetch-agent-timeline-once.ts | 31 ++++++++ .../src/timeline/viewed-timeline-sync.test.ts | 18 +++++ .../app/src/timeline/viewed-timeline-sync.ts | 29 +++++++- .../src/utils/agent-directory-sync.test.ts | 17 +++++ .../app/src/utils/agent-directory-sync.ts | 1 - .../selective-timeline-delivery.e2e.test.ts | 14 ++++ packages/server/src/server/session.test.ts | 38 ++++++++++ packages/server/src/server/session.ts | 12 +++- .../server/src/server/websocket-server.ts | 6 ++ 14 files changed, 277 insertions(+), 10 deletions(-) create mode 100644 packages/app/src/timeline/fetch-agent-timeline-once.test.ts create mode 100644 packages/app/src/timeline/fetch-agent-timeline-once.ts diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 0199b2cfa..645b2ccd2 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -21,6 +21,7 @@ import { } from "@/timeline/session-stream-reducers"; import { useCreateFlowStore } from "@/stores/create-flow-store"; import { isTimelineCatchUpComplete } from "@/timeline/timeline-sync-plan"; +import { fetchAgentTimelineOnce } from "@/timeline/fetch-agent-timeline-once"; import { createViewedTimelineSync, type ViewedTimelineSync } from "@/timeline/viewed-timeline-sync"; import type { AgentAttachment, SessionOutboundMessage } from "@getpaseo/protocol/messages"; import { parseServerInfoStatusPayload } from "@getpaseo/protocol/messages"; @@ -1071,7 +1072,8 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const initKey = getInitKey(serverId, agentId); if (session?.agentAuthoritativeHistoryApplied.get(agentId) !== true) { if (!getInitDeferred(initKey)) { - createInitDeferred(initKey, request.direction ?? "tail"); + const deferred = createInitDeferred(initKey, request.direction ?? "tail"); + void deferred.promise.catch(() => undefined); } refreshAgentInitializationTimeout({ key: initKey, @@ -1081,7 +1083,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider setAgentInitializing(agentId, true); } try { - const page = await client.fetchAgentTimeline(agentId, request); + const page = await fetchAgentTimelineOnce(client, agentId, request); if (getInitDeferred(initKey)) { refreshAgentInitializationTimeout({ key: initKey, agentId, setAgentInitializing }); } diff --git a/packages/app/src/hooks/use-agent-initialization.ts b/packages/app/src/hooks/use-agent-initialization.ts index b9a76bc99..af357deed 100644 --- a/packages/app/src/hooks/use-agent-initialization.ts +++ b/packages/app/src/hooks/use-agent-initialization.ts @@ -10,6 +10,7 @@ import { rejectInitDeferred, refreshInitTimeout, } from "@/utils/agent-initialization"; +import { fetchAgentTimelineOnce } from "@/timeline/fetch-agent-timeline-once"; import { planInitialAgentTimelineSync, planTimelineTailFetch } from "@/timeline/timeline-sync-plan"; import { i18n } from "@/i18n/i18next"; @@ -68,7 +69,7 @@ export function ensureAgentIsInitialized(input: EnsureAgentIsInitializedInput): return deferred.promise; } - client.fetchAgentTimeline(agentId, timelineRequest).catch((error) => { + fetchAgentTimelineOnce(client, agentId, timelineRequest).catch((error) => { setAgentInitializing(agentId, false); rejectInitDeferred(key, error instanceof Error ? error : new Error(String(error))); }); diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index cf570292d..7a19bc67e 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -2153,6 +2153,77 @@ describe("HostRuntimeStore", () => { useSessionStore.getState().clearSession(host.serverId); }); + it("uses legacy GitHub attachments when draining a queue for an old daemon", async () => { + const host = makeHost({ serverId: "srv_legacy_queue_attachment" }); + const fakeClient = new FakeDaemonClient(); + const store = new HostRuntimeStore({ + deps: { + createClient: () => fakeClient as unknown as DaemonClient, + connectToDaemon: async () => ({ + client: fakeClient as unknown as DaemonClient, + serverId: host.serverId, + hostname: null, + }), + getClientId: async () => "cid_legacy_queue_attachment", + }, + }); + const sessionStore = useSessionStore.getState(); + sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1); + sessionStore.updateSessionServerInfo(host.serverId, { + serverId: host.serverId, + hostname: null, + version: "0.1.105", + features: { forgeSearch: false }, + }); + sessionStore.setQueuedMessages( + host.serverId, + new Map([ + [ + "agent", + [ + { + id: "queued-legacy-attachment", + text: "review this", + attachments: [ + { + kind: "github_pr" as const, + item: { + kind: "change_request" as const, + number: 42, + title: "Compatibility fix", + url: "https://github.com/acme/repo/pull/42", + state: "open" as const, + body: "Details", + labels: [], + baseRefName: "main", + headRefName: "fix", + }, + }, + ], + }, + ], + ], + ]), + ); + + store.drainQueuedAgentMessage(host.serverId, "agent"); + await fakeClient.waitForSentMessages(1); + + expect(fakeClient.sentAgentMessages[0]?.[2]?.attachments).toEqual([ + { + type: "github_pr", + mimeType: "application/github-pr", + number: 42, + title: "Compatibility fix", + url: "https://github.com/acme/repo/pull/42", + body: "Details", + baseRefName: "main", + headRefName: "fix", + }, + ]); + sessionStore.clearSession(host.serverId); + }); + it("applies buffered stale side effects from the accepted page agent", async () => { const host = makeHost({ serverId: "srv_buffered_stale_side_effects" }); const fakeClient = new FakeDaemonClient(); diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 6d80f87ea..16568cd2e 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -62,7 +62,10 @@ import { import { mountBrowserAutomationDaemonClientHandler } from "@/browser-automation/handler"; import { schedulesQueryBaseKey } from "@/schedules/aggregated-schedules"; import { sendQueuedComposerMessageNow } from "@/composer/actions"; -import { splitComposerAttachmentsForSubmit } from "@/composer/attachments/submit"; +import { + resolveComposerAttachmentSubmitFormat, + splitComposerAttachmentsForSubmit, +} from "@/composer/attachments/submit"; import { encodeImages } from "@/utils/encode-images"; export type HostRuntimeConnectionStatus = "idle" | "connecting" | "online" | "offline" | "error"; @@ -2141,7 +2144,11 @@ export class HostRuntimeStore { write: (update) => useSessionStore.getState().setQueuedMessages(serverId, update), }, submitMessage: async ({ text, attachments }) => { - const wirePayload = splitComposerAttachmentsForSubmit(attachments); + const supportsForgeAttachments = + useSessionStore.getState().sessions[serverId]?.serverInfo?.features?.forgeSearch === true; + const wirePayload = splitComposerAttachmentsForSubmit(attachments, { + format: resolveComposerAttachmentSubmitFormat({ supportsForgeAttachments }), + }); const images = await encodeImages(wirePayload.images); await client.sendAgentMessage(agentId, text, { messageId: next.id, diff --git a/packages/app/src/timeline/fetch-agent-timeline-once.test.ts b/packages/app/src/timeline/fetch-agent-timeline-once.test.ts new file mode 100644 index 000000000..1ae97e028 --- /dev/null +++ b/packages/app/src/timeline/fetch-agent-timeline-once.test.ts @@ -0,0 +1,30 @@ +import { expect, test } from "vitest"; +import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; +import { fetchAgentTimelineOnce } from "./fetch-agent-timeline-once"; + +type TimelinePage = Awaited>; + +test("concurrent identical timeline reads share one request", async () => { + let resolvePage: (page: TimelinePage) => void = () => {}; + const page = new Promise((resolve) => { + resolvePage = resolve; + }); + const requests: Array<{ agentId: string; direction: string }> = []; + const client = { + fetchAgentTimeline: async ( + agentId: string, + request: Parameters[1], + ) => { + requests.push({ agentId, direction: request?.direction ?? "tail" }); + return page; + }, + }; + const request = { direction: "tail" as const, limit: 100, projection: "projected" as const }; + + const first = fetchAgentTimelineOnce(client, "agent", request); + const second = fetchAgentTimelineOnce(client, "agent", request); + resolvePage({ hasNewer: false } as TimelinePage); + + await expect(Promise.all([first, second])).resolves.toHaveLength(2); + expect(requests).toEqual([{ agentId: "agent", direction: "tail" }]); +}); diff --git a/packages/app/src/timeline/fetch-agent-timeline-once.ts b/packages/app/src/timeline/fetch-agent-timeline-once.ts new file mode 100644 index 000000000..e0d67b489 --- /dev/null +++ b/packages/app/src/timeline/fetch-agent-timeline-once.ts @@ -0,0 +1,31 @@ +import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; + +type TimelineClient = Pick; +type TimelineRequest = Parameters[1]; +type TimelinePage = Awaited>; + +const inFlightByClient = new WeakMap>>(); + +export function fetchAgentTimelineOnce( + client: TimelineClient, + agentId: string, + request: TimelineRequest, +): Promise { + let inFlight = inFlightByClient.get(client); + if (!inFlight) { + inFlight = new Map(); + inFlightByClient.set(client, inFlight); + } + + const key = `${agentId}:${JSON.stringify(request)}`; + const existing = inFlight.get(key); + if (existing) return existing; + + const fetch = client.fetchAgentTimeline(agentId, request); + inFlight.set(key, fetch); + const clear = () => { + if (inFlight.get(key) === fetch) inFlight.delete(key); + }; + void fetch.then(clear, clear); + return fetch; +} diff --git a/packages/app/src/timeline/viewed-timeline-sync.test.ts b/packages/app/src/timeline/viewed-timeline-sync.test.ts index 367675710..511650f62 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.test.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.test.ts @@ -345,6 +345,24 @@ test("gap recovery supersedes completed catch-up and pages through the current t ]); }); +test("repeated recovery for the same running gap reuses the in-flight fetch", async () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + const membership = await world.nextMembership(); + membership.succeed(); + const initial = await world.nextFetch("agent-a"); + initial.respond({ hasNewer: false }); + + const cursor = { epoch: "epoch-agent-a", endSeq: 10 }; + world.sync.recoverGap("agent-a", cursor); + const gapPage = await world.nextFetch("agent-a"); + world.sync.recoverGap("agent-a", cursor); + + world.expectNoPendingFetch(); + gapPage.respond({ hasNewer: false }); +}); + test("membership failure autonomously retries without another visibility declaration", async () => { const world = new TimelineWorld(); world.sync.setConnected(true); diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index 3a40069b5..f0f2c0870 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -39,9 +39,34 @@ type CatchUpStatus = "running" | "complete" | "error"; interface CatchUpState { generation: number; status: CatchUpStatus; + request?: ProjectedTimelineForwardFetchPlan; cancelRetry?: () => void; } +function isSameCatchUpRequest( + left: ProjectedTimelineForwardFetchPlan | undefined, + right: ProjectedTimelineForwardFetchPlan | undefined, +): boolean { + if (!left || !right || left.direction !== right.direction) return false; + if (left.direction !== "after" || right.direction !== "after") return true; + return left.cursor.epoch === right.cursor.epoch && left.cursor.seq === right.cursor.seq; +} + +function shouldKeepCurrentCatchUp(input: { + current: CatchUpState | undefined; + request: ProjectedTimelineForwardFetchPlan | undefined; + supersede: boolean; +}): boolean { + if (!input.current) return false; + if (input.supersede) { + return ( + input.current.status === "running" && + isSameCatchUpRequest(input.current.request, input.request) + ); + } + return input.current.status === "running" || input.current.status === "complete"; +} + function normalizeAgentIds(agentIds: string[]): string[] { return [...new Set(agentIds)].filter(Boolean).sort(); } @@ -138,13 +163,13 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed return; } const current = catchUps.get(agentId); - if (!supersede && (current?.status === "running" || current?.status === "complete")) { + if (shouldKeepCurrentCatchUp({ current, request, supersede })) { return; } current?.cancelRetry?.(); const generation = (catchUpGenerations.get(agentId) ?? 0) + 1; catchUpGenerations.set(agentId, generation); - catchUps.set(agentId, { generation, status: "running" }); + catchUps.set(agentId, { generation, status: "running", request }); pendingGaps.delete(agentId); const cursor = ports.readCursor(agentId); const nextRequest = diff --git a/packages/app/src/utils/agent-directory-sync.test.ts b/packages/app/src/utils/agent-directory-sync.test.ts index a136a1af0..6b1c0910e 100644 --- a/packages/app/src/utils/agent-directory-sync.test.ts +++ b/packages/app/src/utils/agent-directory-sync.test.ts @@ -65,6 +65,23 @@ function permission(id: string): AgentPermissionRequest { } describe("replaceFetchedAgentDirectory", () => { + it("preserves timeline initialization while replacing directory state", () => { + const serverId = "server-initializing"; + const store = useSessionStore.getState(); + store.initializeSession(serverId, null as unknown as DaemonClient); + store.setInitializingAgents(serverId, new Map([["agent", true]])); + + replaceFetchedAgentDirectory({ + serverId, + entries: [createEntry(createAgentPayload({ id: "agent" }))], + }); + + expect(useSessionStore.getState().sessions[serverId]?.initializingAgents.get("agent")).toBe( + true, + ); + store.clearSession(serverId); + }); + it("re-derives parentAgentId every time an agent snapshot is ingested", () => { const serverId = "server-1"; const store = useSessionStore.getState(); diff --git a/packages/app/src/utils/agent-directory-sync.ts b/packages/app/src/utils/agent-directory-sync.ts index 30039faa6..dfbff645d 100644 --- a/packages/app/src/utils/agent-directory-sync.ts +++ b/packages/app/src/utils/agent-directory-sync.ts @@ -180,7 +180,6 @@ export function replaceFetchedAgentDirectory(input: { store.setAgentLastActivityBatch(lastActivityByAgentId); store.setPendingPermissions(input.serverId, new Map(pendingPermissions)); - store.setInitializingAgents(input.serverId, new Map()); store.setHasHydratedAgents(input.serverId, true); return { agents: fetchedAgents }; } diff --git a/packages/server/src/server/selective-timeline-delivery.e2e.test.ts b/packages/server/src/server/selective-timeline-delivery.e2e.test.ts index 52246e04e..8933def32 100644 --- a/packages/server/src/server/selective-timeline-delivery.e2e.test.ts +++ b/packages/server/src/server/selective-timeline-delivery.e2e.test.ts @@ -137,6 +137,20 @@ async function connect(input: { clientId: string; selective: boolean }): Promise return connected; } +test("subscription acknowledgements stay on the requesting socket of a retained session", async () => { + const legacy = await connect({ clientId: "shared-client", selective: false }); + const capable = await connect({ clientId: "shared-client", selective: true }); + legacy.clear(); + capable.clear(); + + await capable.client.setAgentTimelineSubscription(["agent-a"]); + await capable.barrier("targeted-subscription-ack"); + + expect( + legacy.messages.some((message) => message.type === "agent.timeline.set_subscription.response"), + ).toBe(false); +}); + test("real WebSocket sessions enforce selective delivery, retained resets, downgrade, and dedicated attention", async () => { const legacy = await connect({ clientId: "legacy-client", selective: false }); let capable = await connect({ clientId: "capable-client", selective: true }); diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index 1df991386..8f0054dc6 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -311,6 +311,7 @@ interface SessionForTestOptions { daemonRuntimeConfig?: SessionOptions["daemonRuntimeConfig"]; downloadTokenStore?: SessionOptions["downloadTokenStore"]; messages?: unknown[]; + targetedMessages?: Array<{ source: object; message: SessionOutboundMessage }>; binaryMessages?: Uint8Array[]; } @@ -348,6 +349,12 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session { return new Session({ clientId: "test-client", onMessage: (message) => messages.push(message), + ...(options.targetedMessages + ? { + onMessageToSource: (source: object, message: SessionOutboundMessage) => + options.targetedMessages?.push({ source, message }), + } + : {}), onBinaryMessage: createBinaryMessageHandler(options.binaryMessages), logger, downloadTokenStore: options.downloadTokenStore ?? asDownloadTokenStore(), @@ -4578,6 +4585,37 @@ test("replaces a capable session's complete viewed timeline set", async () => { ]); }); +test("acknowledges a timeline subscription only to its socket source", async () => { + const messages: SessionOutboundMessage[] = []; + const targetedMessages: Array<{ source: object; message: SessionOutboundMessage }> = []; + const session = createSessionForTest({ messages, targetedMessages }); + const capableSocket = {}; + session.updateClientCapabilities({ selective_agent_timeline: true }, capableSocket); + + await session.handleMessage( + { + type: "agent.timeline.set_subscription.request", + agentIds: ["agent-a"], + requestId: "timeline-subscription-targeted", + }, + capableSocket, + ); + + expect(messages).toEqual([]); + expect(targetedMessages).toEqual([ + { + source: capableSocket, + message: { + type: "agent.timeline.set_subscription.response", + payload: { + agentIds: ["agent-a"], + requestId: "timeline-subscription-targeted", + }, + }, + }, + ]); +}); + test("unions viewed timelines across socket sources and removes detached sources", async () => { const messages: SessionOutboundMessage[] = []; const agentEventListeners: Array<(event: AgentManagerEvent) => void> = []; diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index de82f0f60..9d0f23800 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -429,6 +429,7 @@ export interface SessionOptions { appVersion?: string | null; clientCapabilities?: Record | null; onMessage: (msg: SessionOutboundMessage) => void; + onMessageToSource?: (source: object, msg: SessionOutboundMessage) => void; onBinaryMessage?: (frame: Uint8Array) => void; getTransportBufferedAmount?: () => number | null; onLifecycleIntent?: (intent: SessionLifecycleIntent) => void; @@ -558,6 +559,9 @@ export class Session { private clientCapabilities: ReadonlySet; private readonly sessionId: string; private readonly onMessage: (msg: SessionOutboundMessage) => void; + private readonly onMessageToSource: + | ((source: object, msg: SessionOutboundMessage) => void) + | null; private readonly onBinaryMessage: ((frame: Uint8Array) => void) | null; private readonly getTransportBufferedAmount: () => number | null; private readonly onLifecycleIntent: ((intent: SessionLifecycleIntent) => void) | null; @@ -629,6 +633,7 @@ export class Session { appVersion, clientCapabilities, onMessage, + onMessageToSource, onBinaryMessage, getTransportBufferedAmount, onLifecycleIntent, @@ -679,6 +684,7 @@ export class Session { this.clientCapabilities = parseClientCapabilities(clientCapabilities); this.sessionId = uuidv4(); this.onMessage = onMessage; + this.onMessageToSource = onMessageToSource ?? null; this.onBinaryMessage = onBinaryMessage ?? null; this.getTransportBufferedAmount = getTransportBufferedAmount ?? (() => 0); this.onLifecycleIntent = onLifecycleIntent ?? null; @@ -1618,10 +1624,12 @@ export class Session { ) { this.replaceAgentTimelineSubscription(source, agentIds); } - this.emit({ + const response: SessionOutboundMessage = { type: "agent.timeline.set_subscription.response", payload: { agentIds, requestId: msg.requestId }, - }); + }; + if (source && this.onMessageToSource) this.onMessageToSource(source, response); + else this.emit(response); return undefined; } case "agent.fork_context.request": diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 9c0647a90..62b6b03f4 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -988,6 +988,12 @@ export class VoiceAssistantWebSocketServer { } this.sendToConnection(connection, wrapSessionMessage(msg)); }, + onMessageToSource: (source, msg) => { + if (!connection || !connection.sockets.has(source as WebSocketLike)) { + return; + } + this.sendToClient(source as WebSocketLike, wrapSessionMessage(msg)); + }, onBinaryMessage: (frame) => { if (!connection) { return;