diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 020ee0ed4..dfd56a46a 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -51,9 +51,39 @@ When a client resumes with a known cursor, it catches up after that cursor to co When a client resumes without a cursor, it fetches the latest tail page. +## Selective and legacy delivery + +The app chooses one delivery policy from `server_info.features.selectiveAgentTimeline`: + +- Selective daemons receive the union of agents visible in every pane. Additions subscribe and + catch up immediately. Removals stay subscribed for a short grace period so quick tab and pane + switches do not repeatedly unsubscribe and catch up. Backgrounding, disconnecting, and disposal + clear that grace immediately. +- Legacy daemons keep globally streaming agent timelines. Visibility still triggers the existing + authoritative catch-up, but the app does not issue selective-subscription RPCs. + +This policy is owned by `viewed-timeline-sync.ts`; downstream reducers do not branch on daemon +version. + +## Projected pages reconcile with live presentation + +A projected page is canonical state, not a sequence of live deltas. One projected item can overlap +rows already received live—for example, a tool call retained at its original display position while +its completion advances `seqEnd`, followed by a merged assistant message. The app uses +`sourceSeqRanges` to replace overlapping assistant and reasoning projections before applying the +remaining page through the existing stream reducer. It must not append full projected text to a +live prefix. + +Optimistic user prompts are presentation state rather than canonical history. Incremental catch-up +temporarily separates them, applies canonical entries, lets canonical user rows reconcile through +the existing optimistic-message rules, then restores any unmatched prompts after the caught-up +history. This keeps late history before a newly submitted prompt without duplicating an +acknowledged prompt. + ## Relevant code - Server live stream forwarding: `packages/server/src/server/session.ts` - App sync planning: `packages/app/src/timeline/timeline-sync-plan.ts` +- App viewed-agent synchronization: `packages/app/src/timeline/viewed-timeline-sync.ts` - App stream/timeline reducer: `packages/app/src/timeline/session-stream-reducers.ts` - Session wiring: `packages/app/src/contexts/session-context.tsx` diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 31720b34d..abde80934 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -20,7 +20,11 @@ import { } from "@/timeline/session-stream-reducers"; import { useCreateFlowStore } from "@/stores/create-flow-store"; import { isTimelineCatchUpComplete } from "@/timeline/timeline-sync-plan"; -import { createViewedTimelineSync, type ViewedTimelineSync } from "@/timeline/viewed-timeline-sync"; +import { + createViewedTimelineSync, + type TimelineDeliveryMode, + type ViewedTimelineSync, +} from "@/timeline/viewed-timeline-sync"; import type { AgentAttachment, SessionOutboundMessage } from "@getpaseo/protocol/messages"; import { parseServerInfoStatusPayload } from "@getpaseo/protocol/messages"; import { @@ -79,6 +83,11 @@ interface BufferedAudioChunk { id: string; } +// COMPAT(selectiveAgentTimeline): added in v0.1.106, remove after 2027-01-12. +function getTimelineDeliveryMode(selectiveAgentTimeline?: boolean): TimelineDeliveryMode { + return selectiveAgentTimeline ? "selective" : "legacy"; +} + function decodeBase64Chunk(base64: string): Uint8Array { return Buffer.from(base64, "base64"); } @@ -712,7 +721,11 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider useEffect(() => { const setAgentInitializing = createSetAgentInitializing(serverId, setInitializingAgents); + const initialDeliveryMode = getTimelineDeliveryMode( + client.getLastServerInfoMessage()?.features?.selectiveAgentTimeline, + ); const sync = createViewedTimelineSync({ + initialDeliveryMode, setSubscription: (agentIds) => client.setAgentTimelineSubscription(agentIds), readCursor: (agentId) => useSessionStore.getState().sessions[serverId]?.agentTimelineCursor.get(agentId), @@ -723,7 +736,8 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider fetchPage: async (agentId, request) => { const session = useSessionStore.getState().sessions[serverId]; const initKey = getInitKey(serverId, agentId); - if (session?.agentAuthoritativeHistoryApplied.get(agentId) !== true) { + const shouldInitialize = session?.agentAuthoritativeHistoryApplied.get(agentId) !== true; + if (shouldInitialize) { if (!getInitDeferred(initKey)) { const deferred = createInitDeferred(initKey, request.direction ?? "tail"); void deferred.promise.catch(() => undefined); @@ -737,21 +751,23 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider } try { const page = await getHostRuntimeStore().fetchAgentTimeline(serverId, agentId, request); - if (getInitDeferred(initKey)) { + if (shouldInitialize && getInitDeferred(initKey)) { refreshAgentInitializationTimeout({ key: initKey, agentId, setAgentInitializing }); } return page; } catch (error) { - setAgentInitializing(agentId, false); - rejectInitDeferred(initKey, error instanceof Error ? error : new Error(String(error))); + if (shouldInitialize) { + setAgentInitializing(agentId, false); + rejectInitDeferred(initKey, error instanceof Error ? error : new Error(String(error))); + } throw error; } }, reportError: (error) => { console.warn("[Session] viewed timeline synchronization failed", { serverId, error }); }, - scheduleRetry: (retry) => { - const timeout = setTimeout(retry, 1_000); + schedule: (task, delayMs) => { + const timeout = setTimeout(task, delayMs); return () => clearTimeout(timeout); }, }); @@ -855,6 +871,9 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider if (message.type !== "status") return; const serverInfo = parseServerInfoStatusPayload(message.payload); if (serverInfo) { + viewedTimelineSyncRef.current?.setDeliveryMode( + getTimelineDeliveryMode(serverInfo.features?.selectiveAgentTimeline), + ); updateSessionServerInfo(serverId, { serverId: serverInfo.serverId, hostname: serverInfo.hostname, diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index e0026371c..05048e2d5 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -107,7 +107,10 @@ function makeStreamReducerEvent( }; } -function makeAssistantItem(text: string, id = `assistant-${text.length}`): StreamItem { +function makeAssistantItem( + text: string, + id = `assistant-${text.length}`, +): Extract { return { kind: "assistant_message", id, @@ -556,7 +559,7 @@ describe("processTimelineResponse", () => { item: { type: "user_message", text: "sent while catching up", - messageId: "canonical-after", + messageId: "optimistic-after", }, }, ], @@ -565,10 +568,125 @@ describe("processTimelineResponse", () => { const userMessages = result.tail.filter((item) => item.kind === "user_message"); expect(userMessages).toHaveLength(1); - expect(userMessages[0]?.id).toBe("canonical-after"); + expect(userMessages[0]?.id).toBe("optimistic-after"); expect(userMessages[0]?.optimistic).toBeUndefined(); }); + it("reconciles multiple optimistic user messages in canonical order", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [ + makeOptimisticUserMessage("first prompt", "optimistic-first"), + makeOptimisticUserMessage("second prompt", "optimistic-second"), + ], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 }, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 2 }, + endCursor: { seq: 3 }, + entries: [ + { + ...makeTimelineEntry(2, "first prompt", "user_message"), + item: { type: "user_message", text: "first prompt", messageId: "optimistic-first" }, + }, + { + ...makeTimelineEntry(3, "second prompt", "user_message"), + item: { + type: "user_message", + text: "second prompt", + messageId: "optimistic-second", + }, + }, + ], + }, + }); + + expect( + result.tail + .filter((item) => item.kind === "user_message") + .map((item) => ({ id: item.id, text: item.text, optimistic: item.optimistic })), + ).toEqual([ + { id: "optimistic-first", text: "first prompt", optimistic: undefined }, + { id: "optimistic-second", text: "second prompt", optimistic: undefined }, + ]); + }); + + it("keeps a tail optimistic prompt before a reconciled live assistant head", () => { + const prompt = makeOptimisticUserMessage("new prompt", "optimistic-new-prompt"); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [prompt], + currentHead: [ + { + ...makeAssistantItem("Hel", "answer-1"), + messageId: "answer-1", + }, + ], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 2 }, + payload: { + ...baseTimelineInput.payload, + direction: "after", + epoch: "epoch-1", + startCursor: { seq: 3 }, + endCursor: { seq: 3 }, + entries: [ + { + ...makeTimelineEntry(2, "Hello", "assistant_message", 3), + sourceSeqRanges: [{ startSeq: 2, endSeq: 3 }], + item: { + type: "assistant_message", + text: "Hello", + messageId: "answer-1", + }, + }, + ], + }, + }); + + expect( + [...result.tail, ...result.head] + .filter((item) => item.kind === "assistant_message" || item.kind === "user_message") + .map((item) => item.text), + ).toEqual(["new prompt", "Hello"]); + }); + + it("keeps a tail optimistic prompt before a live head flushed by catch-up", () => { + const prompt = makeOptimisticUserMessage("new prompt", "optimistic-new-prompt"); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [prompt], + currentHead: [ + { + ...makeAssistantItem("Live response", "answer-1"), + messageId: "answer-1", + }, + ], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 2 }, + payload: { + ...baseTimelineInput.payload, + direction: "after", + epoch: "epoch-1", + startCursor: { seq: 3 }, + endCursor: { seq: 3 }, + entries: [ + makeToolCallTimelineEntry(3, "call-1", "running", { + type: "read", + filePath: "/tmp/example.ts", + }), + ], + }, + }); + + expect( + [...result.tail, ...result.head] + .filter((item) => item.kind === "assistant_message" || item.kind === "user_message") + .map((item) => item.text), + ).toEqual(["new prompt", "Live response"]); + }); + it("keeps an active assistant head live when an incremental fetch accepts same-turn assistant text", () => { const existingCursor: TimelineCursor = { epoch: "epoch-1", @@ -597,6 +715,504 @@ describe("processTimelineResponse", () => { }); }); + it("does not replay an assistant prefix when catch-up completes an earlier tool call", () => { + const live = processAgentStreamEvents({ + events: [ + makeStreamReducerEvent(makeToolCallTimelineEvent("call-1"), 1), + makeStreamReducerEvent(makeAssistantTimelineEvent("Hel", "answer-1"), 2), + ], + currentTail: [], + currentHead: [], + currentCursor: undefined, + currentAgent: null, + }); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: live.tail, + currentHead: live.head, + currentCursor: live.cursor ?? undefined, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 3 }, + endCursor: { seq: 4 }, + entries: [ + { + ...makeToolCallTimelineEntry(1, "call-1", "completed", { + type: "read", + filePath: "/tmp/example.ts", + }), + seqEnd: 4, + sourceSeqRanges: [ + { startSeq: 1, endSeq: 1 }, + { startSeq: 4, endSeq: 4 }, + ], + }, + { + ...makeTimelineEntry(2, "Hello", "assistant_message", 3), + sourceSeqRanges: [{ startSeq: 2, endSeq: 3 }], + item: { + type: "assistant_message", + text: "Hello", + messageId: "answer-1", + }, + }, + ], + }, + }); + + expect(getAssistantTexts([...result.tail, ...result.head])).toEqual(["Hello"]); + }); + + it("reconciles an identified projection with an overlapping anonymous live prefix", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + currentHead: [makeAssistantItem("Hel")], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 2 }, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 3 }, + endCursor: { seq: 3 }, + entries: [ + { + ...makeTimelineEntry(2, "Hello", "assistant_message", 3), + sourceSeqRanges: [{ startSeq: 2, endSeq: 3 }], + item: { + type: "assistant_message", + text: "Hello", + messageId: "answer-1", + }, + }, + ], + }, + }); + + const assistants = [...result.tail, ...result.head].filter( + (item) => item.kind === "assistant_message", + ); + expect(assistants).toHaveLength(1); + expect(assistants[0]).toMatchObject({ text: "Hello", messageId: "answer-1" }); + }); + + it("replaces every promoted assistant block when reconciling a projected message", () => { + const live = processAgentStreamEvents({ + events: [makeStreamReducerEvent(makeAssistantTimelineEvent("First paragraph.\n\nSec"), 2)], + currentTail: [], + currentHead: [], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 }, + currentAgent: null, + }); + expect(getAssistantTexts(live.tail)).toHaveLength(1); + expect(getAssistantTexts(live.head)).toHaveLength(1); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: live.tail, + currentHead: live.head, + currentCursor: live.cursor ?? undefined, + payload: { + ...baseTimelineInput.payload, + direction: "after", + epoch: "epoch-1", + startCursor: { seq: 3 }, + endCursor: { seq: 3 }, + entries: [ + { + ...makeTimelineEntry( + 2, + "First paragraph.\n\nSecond paragraph.", + "assistant_message", + 3, + ), + sourceSeqRanges: [{ startSeq: 2, endSeq: 3 }], + item: { + type: "assistant_message", + text: "First paragraph.\n\nSecond paragraph.", + }, + }, + ], + }, + }); + + expect(getAssistantTexts([...result.tail, ...result.head])).toEqual([ + "First paragraph.\n\nSecond paragraph.", + ]); + }); + + it("does not replay a reasoning prefix when catch-up completes an earlier tool call", () => { + const live = processAgentStreamEvents({ + events: [ + makeStreamReducerEvent(makeToolCallTimelineEvent("call-1"), 1), + makeStreamReducerEvent(makeTimelineEvent("Thi", "reasoning"), 2), + ], + currentTail: [], + currentHead: [], + currentCursor: undefined, + currentAgent: null, + }); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: live.tail, + currentHead: live.head, + currentCursor: live.cursor ?? undefined, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 3 }, + endCursor: { seq: 4 }, + entries: [ + { + ...makeToolCallTimelineEntry(1, "call-1", "completed", { + type: "read", + filePath: "/tmp/example.ts", + }), + seqEnd: 4, + sourceSeqRanges: [ + { startSeq: 1, endSeq: 1 }, + { startSeq: 4, endSeq: 4 }, + ], + }, + { + ...makeTimelineEntry(2, "Thinking", "reasoning", 3), + sourceSeqRanges: [{ startSeq: 2, endSeq: 3 }], + }, + ], + }, + }); + + const thoughts = [...result.tail, ...result.head].filter((item) => item.kind === "thought"); + expect(thoughts).toHaveLength(1); + expect(thoughts[0]?.text).toBe("Thinking"); + }); + + it("keeps delayed catch-up history before a newly submitted prompt", () => { + const prompt = makeOptimisticUserMessage("New prompt", "new-prompt"); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [makeAssistantItem("Earlier answer", "earlier-answer"), prompt], + currentHead: [], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 2 }, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 3 }, + endCursor: { seq: 3 }, + entries: [makeTimelineEntry(3, "Missed answer")], + }, + }); + + expect( + [...result.tail, ...result.head] + .filter((item) => item.kind === "assistant_message" || item.kind === "user_message") + .map((item) => item.text), + ).toEqual(["Earlier answer", "Missed answer", "New prompt"]); + }); + + it("keeps delayed catch-up history between a live head and its unmatched head prompt", () => { + const prompt = makeOptimisticUserMessage("New prompt", "new-prompt"); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [], + currentHead: [makeAssistantItem("Live answer", "live-answer"), prompt], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 2 }, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 3 }, + endCursor: { seq: 3 }, + entries: [ + makeToolCallTimelineEntry(3, "missed-call", "completed", { + type: "read", + filePath: "/tmp/missed.ts", + }), + ], + }, + }); + + expect([...result.tail, ...result.head].map((item) => item.kind)).toEqual([ + "assistant_message", + "tool_call", + "user_message", + ]); + }); + + it("keeps delayed catch-up history between a live head and its acknowledged head prompt", () => { + const prompt = makeOptimisticUserMessage("New prompt", "new-prompt"); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [], + currentHead: [makeAssistantItem("Live answer", "live-answer"), prompt], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 2 }, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 3 }, + endCursor: { seq: 4 }, + entries: [ + makeToolCallTimelineEntry(3, "missed-call", "completed", { + type: "read", + filePath: "/tmp/missed.ts", + }), + { + ...makeTimelineEntry(4, "New prompt", "user_message"), + item: { + type: "user_message", + text: "New prompt", + messageId: "new-prompt", + }, + }, + ], + }, + }); + + expect([...result.tail, ...result.head].map((item) => item.kind)).toEqual([ + "assistant_message", + "tool_call", + "user_message", + ]); + expect( + [...result.tail, ...result.head] + .filter((item) => item.kind === "user_message") + .map((item) => item.optimistic), + ).toEqual([undefined]); + }); + + it("keeps unrelated delayed history before the prompt and its live response", () => { + const prompt = makeOptimisticUserMessage("New prompt", "new-prompt"); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [makeAssistantItem("Earlier answer", "earlier-answer"), prompt], + currentHead: [ + { + ...makeAssistantItem("Live response", "live-response"), + messageId: "live-response", + }, + ], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 2 }, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 3 }, + endCursor: { seq: 4 }, + entries: [ + makeTimelineEntry(3, "Missed answer"), + { + ...makeTimelineEntry(4, "Remote prompt", "user_message"), + item: { + type: "user_message", + text: "Remote prompt", + messageId: "remote-prompt", + }, + }, + ], + }, + }); + + expect( + [...result.tail, ...result.head] + .filter((item) => item.kind === "assistant_message" || item.kind === "user_message") + .map((item) => item.text), + ).toEqual(["Earlier answer", "Missed answer", "Remote prompt", "New prompt", "Live response"]); + }); + + it("keeps delayed history before a prompt whose live answer has promoted blocks", () => { + const prompt = makeOptimisticUserMessage("New prompt", "new-prompt"); + const live = processAgentStreamEvents({ + events: [ + makeStreamReducerEvent( + makeAssistantTimelineEvent("First paragraph.\n\nSecond paragraph", "live-response"), + 2, + ), + ], + currentTail: [prompt], + currentHead: [], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 }, + currentAgent: null, + }); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: live.tail, + currentHead: live.head, + currentCursor: live.cursor ?? undefined, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 3 }, + endCursor: { seq: 3 }, + entries: [makeTimelineEntry(3, "Missed answer")], + }, + }); + + expect( + [...result.tail, ...result.head] + .filter((item) => item.kind === "assistant_message" || item.kind === "user_message") + .map((item) => item.text), + ).toEqual(["Missed answer", "New prompt", "First paragraph.", "Second paragraph"]); + }); + + it("keeps delayed tool history before a prompt whose live answer has promoted blocks", () => { + const prompt = makeOptimisticUserMessage("New prompt", "new-prompt"); + const live = processAgentStreamEvents({ + events: [ + makeStreamReducerEvent( + makeAssistantTimelineEvent("First paragraph.\n\nSecond paragraph", "live-response"), + 2, + ), + ], + currentTail: [prompt], + currentHead: [], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 }, + currentAgent: null, + }); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: live.tail, + currentHead: live.head, + currentCursor: live.cursor ?? undefined, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 3 }, + endCursor: { seq: 3 }, + entries: [ + makeToolCallTimelineEntry(3, "missed-call", "completed", { + type: "read", + filePath: "/tmp/missed.ts", + }), + ], + }, + }); + + expect([...result.tail, ...result.head].map((item) => item.kind)).toEqual([ + "tool_call", + "user_message", + "assistant_message", + "assistant_message", + ]); + }); + + it("matches a local optimistic prompt after an unrelated remote user row", () => { + const prompt = makeOptimisticUserMessage("Local prompt", "local-prompt"); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [prompt], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 }, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 2 }, + endCursor: { seq: 3 }, + entries: [ + { + ...makeTimelineEntry(2, "Remote prompt", "user_message"), + item: { + type: "user_message", + text: "Remote prompt", + messageId: "remote-prompt", + }, + }, + { + ...makeTimelineEntry(3, "Local prompt", "user_message"), + item: { + type: "user_message", + text: "Local prompt", + messageId: "local-prompt", + }, + }, + ], + }, + }); + + expect( + result.tail + .filter((item) => item.kind === "user_message") + .map((item) => ({ text: item.text, optimistic: item.optimistic })), + ).toEqual([ + { text: "Remote prompt", optimistic: undefined }, + { text: "Local prompt", optimistic: undefined }, + ]); + }); + + it("keeps an unmatched optimistic prompt when catch-up contains only a remote user row", () => { + const prompt = makeOptimisticUserMessage("Local prompt", "local-prompt"); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [prompt], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 }, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 2 }, + endCursor: { seq: 2 }, + entries: [ + { + ...makeTimelineEntry(2, "Remote prompt", "user_message"), + item: { + type: "user_message", + text: "Remote prompt", + messageId: "remote-prompt", + }, + }, + ], + }, + }); + + expect( + result.tail + .filter((item) => item.kind === "user_message") + .map((item) => ({ text: item.text, optimistic: item.optimistic })), + ).toEqual([ + { text: "Remote prompt", optimistic: undefined }, + { text: "Local prompt", optimistic: true }, + ]); + }); + + it("does not match equal prompt text when canonical message ids differ", () => { + const prompt = makeOptimisticUserMessage("continue", "local-prompt"); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [prompt], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 }, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 2 }, + endCursor: { seq: 2 }, + entries: [ + { + ...makeTimelineEntry(2, "continue", "user_message"), + item: { + type: "user_message", + text: "continue", + messageId: "remote-prompt", + }, + }, + ], + }, + }); + + expect( + result.tail + .filter((item) => item.kind === "user_message") + .map((item) => ({ id: item.id, optimistic: item.optimistic })), + ).toEqual([ + { id: "remote-prompt", optimistic: undefined }, + { id: "local-prompt", optimistic: true }, + ]); + }); + it("hydrates a fetched in-progress tool call as one item and streams the next update on top", () => { const fetched = processTimelineResponse({ ...baseTimelineInput, diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index c10f91c8b..854832762 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -2,9 +2,10 @@ import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages"; import type { AgentLifecycleStatus } from "@getpaseo/protocol/agent-lifecycle"; import type { Agent } from "@/stores/session-store"; import { useSessionStore } from "@/stores/session-store"; -import type { StreamItem, UserMessageItem } from "@/types/stream"; +import type { AssistantMessageItem, StreamItem, UserMessageItem } from "@/types/stream"; import { applyStreamEvent, + flushHeadToTail, hydrateStreamState, isAgentToolCallItem, mergeAgentToolCallItem, @@ -544,6 +545,378 @@ function replaceLiveAssistantWithProjectedText(params: { return next; } +interface OptimisticUserMessagePosition { + item: UserMessageItem; + placement: "tail" | "head"; + index: number; +} + +function findOptimisticUserMessage(params: { + tail: StreamItem[]; + head: StreamItem[]; +}): OptimisticUserMessagePosition | null { + const tailIndex = params.tail.findIndex( + (item) => item.kind === "user_message" && item.optimistic, + ); + const tailItem = params.tail[tailIndex]; + if (tailItem?.kind === "user_message") { + return { item: tailItem, placement: "tail", index: tailIndex }; + } + + const headIndex = params.head.findIndex( + (item) => item.kind === "user_message" && item.optimistic, + ); + const headItem = params.head[headIndex]; + return headItem?.kind === "user_message" + ? { item: headItem, placement: "head", index: headIndex } + : null; +} + +function replaceOptimisticUserMessage(params: { + tail: StreamItem[]; + head: StreamItem[]; + position: OptimisticUserMessagePosition; + precedingItems: StreamItem[]; + replacement: StreamItem[]; +}): { tail: StreamItem[]; head: StreamItem[] } { + const { position } = params; + const items = [...params.precedingItems, ...params.replacement]; + if (position.placement === "tail") { + return { + tail: [ + ...params.tail.slice(0, position.index), + ...items, + ...params.tail.slice(position.index + 1), + ], + head: params.head, + }; + } + return { + tail: params.tail, + head: [ + ...params.head.slice(0, position.index), + ...items, + ...params.head.slice(position.index + 1), + ], + }; +} + +function reconcileOverlappingProjectedAssistant(params: { + tail: StreamItem[]; + head: StreamItem[]; + unit: TimelineUnit; + epoch: string; + currentEndSeq: number; +}): { tail: StreamItem[]; head: StreamItem[]; reconciled: boolean } { + const { unit } = params; + if ( + unit.event.type !== "timeline" || + unit.event.item.type !== "assistant_message" || + !unit.sourceSeqRanges.some( + (range) => range.startSeq <= params.currentEndSeq && range.endSeq > params.currentEndSeq, + ) + ) { + return { tail: params.tail, head: params.head, reconciled: false }; + } + + const projectedText = unit.event.item.text; + const projectedMessageId = unit.event.item.messageId; + const matches = (item: StreamItem) => { + if (item.kind !== "assistant_message") return false; + if (projectedMessageId && item.messageId) return item.messageId === projectedMessageId; + return projectedText.startsWith(item.text); + }; + const findMatch = (items: StreamItem[]) => { + const index = items.findLastIndex(matches); + const current = items[index]; + return current?.kind === "assistant_message" ? { current, index } : null; + }; + + const headMatch = findMatch(params.head); + const tailMatch = headMatch ? null : findMatch(params.tail); + const match = headMatch ?? tailMatch; + if (!match) { + return { tail: params.tail, head: params.head, reconciled: false }; + } + + const blockGroupId = match.current.blockGroupId; + const messageId = projectedMessageId ?? match.current.messageId; + const replacement: AssistantMessageItem = { + kind: "assistant_message", + id: blockGroupId ?? match.current.id, + ...(messageId !== undefined ? { messageId } : {}), + text: projectedText, + timestamp: unit.timestamp, + timelineCursor: { epoch: params.epoch, seq: unit.seqEnd }, + }; + const belongsToBlockGroup = (item: StreamItem) => + blockGroupId !== undefined && + item.kind === "assistant_message" && + item.blockGroupId === blockGroupId; + const removeBlockGroup = (items: StreamItem[]) => + blockGroupId !== undefined ? items.filter((item) => !belongsToBlockGroup(item)) : items; + const replaceMatch = (items: StreamItem[], index: number) => { + if (!blockGroupId) { + const next = [...items]; + next[index] = replacement; + return next; + } + const next: StreamItem[] = []; + let inserted = false; + for (const item of items) { + if (!belongsToBlockGroup(item)) { + next.push(item); + } else if (!inserted) { + next.push(replacement); + inserted = true; + } + } + return next; + }; + + if (headMatch) { + return { + tail: removeBlockGroup(params.tail), + head: replaceMatch(params.head, headMatch.index), + reconciled: true, + }; + } + return { + tail: replaceMatch(params.tail, match.index), + head: removeBlockGroup(params.head), + reconciled: true, + }; +} + +function reconcileOverlappingProjectedReasoning(params: { + tail: StreamItem[]; + head: StreamItem[]; + unit: TimelineUnit; + currentEndSeq: number; +}): { tail: StreamItem[]; head: StreamItem[]; reconciled: boolean } { + const { unit } = params; + if ( + unit.event.type !== "timeline" || + unit.event.item.type !== "reasoning" || + !unit.sourceSeqRanges.some( + (range) => range.startSeq <= params.currentEndSeq && range.endSeq > params.currentEndSeq, + ) + ) { + return { tail: params.tail, head: params.head, reconciled: false }; + } + + const projectedText = unit.event.item.text; + const replaceIn = (items: StreamItem[]): StreamItem[] | null => { + const index = items.findLastIndex( + (item) => item.kind === "thought" && projectedText.startsWith(item.text), + ); + const current = items[index]; + if (!current || current.kind !== "thought") return null; + const next = [...items]; + next[index] = { + ...current, + text: projectedText, + timestamp: unit.timestamp, + status: "loading", + }; + return next; + }; + + const nextHead = replaceIn(params.head); + if (nextHead) return { tail: params.tail, head: nextHead, reconciled: true }; + const nextTail = replaceIn(params.tail); + return nextTail + ? { tail: nextTail, head: params.head, reconciled: true } + : { tail: params.tail, head: params.head, reconciled: false }; +} + +function reconcileOverlappingProjectedStreamItems(params: { + tail: StreamItem[]; + head: StreamItem[]; + units: TimelineUnit[]; + epoch: string; + currentEndSeq: number | undefined; +}): { tail: StreamItem[]; head: StreamItem[]; reconciledUnits: Set } { + let tail = params.tail; + let head = params.head; + const reconciledUnits = new Set(); + if (params.currentEndSeq === undefined) return { tail, head, reconciledUnits }; + + for (const unit of params.units) { + let reconciled = reconcileOverlappingProjectedAssistant({ + tail, + head, + unit, + epoch: params.epoch, + currentEndSeq: params.currentEndSeq, + }); + if (!reconciled.reconciled) { + reconciled = reconcileOverlappingProjectedReasoning({ + tail, + head, + unit, + currentEndSeq: params.currentEndSeq, + }); + } + tail = reconciled.tail; + head = reconciled.head; + if (reconciled.reconciled) reconciledUnits.add(unit); + } + return { tail, head, reconciledUnits }; +} + +function matchesOptimisticUserMessage(params: { + unit: TimelineUnit; + optimistic: UserMessageItem; +}): boolean { + const { event } = params.unit; + if (event.type !== "timeline" || event.item.type !== "user_message") { + return false; + } + if (event.item.messageId !== undefined) { + return event.item.messageId === params.optimistic.id; + } + return event.item.text.length > 0 && event.item.text === params.optimistic.text; +} + +function acknowledgeOptimisticUserMessage(params: { + tail: StreamItem[]; + head: StreamItem[]; + unit: TimelineUnit; + epoch: string; + optimistic: OptimisticUserMessagePosition; + precedingItems: StreamItem[]; +}): { tail: StreamItem[]; head: StreamItem[] } { + const { event, timestamp, seqEnd } = params.unit; + const timelineCursor = { epoch: params.epoch, seq: seqEnd }; + const acknowledged = reduceStreamUpdate([params.optimistic.item], event, timestamp, { + source: "canonical", + timelineCursor, + }); + return replaceOptimisticUserMessage({ + tail: params.tail, + head: params.head, + position: params.optimistic, + precedingItems: params.precedingItems, + replacement: acknowledged, + }); +} + +function applyCanonicalForwardUnit(params: { + tail: StreamItem[]; + head: StreamItem[]; + unit: TimelineUnit; + epoch: string; +}): { tail: StreamItem[]; head: StreamItem[] } { + const { event, timestamp, seqEnd } = params.unit; + const timelineCursor = { epoch: params.epoch, seq: seqEnd }; + if (params.head.length === 0) { + return { + tail: reduceStreamUpdate(params.tail, event, timestamp, { + source: "canonical", + timelineCursor, + }), + head: params.head, + }; + } + const replacedHead = replaceLiveAssistantWithProjectedText({ + head: params.head, + event, + timestamp, + timelineCursor, + }); + if (replacedHead) return { tail: params.tail, head: replacedHead }; + + const applied = applyStreamEvent({ + tail: params.tail, + head: params.head, + event, + timestamp, + source: "canonical", + timelineCursor, + }); + return { tail: applied.tail, head: applied.head }; +} + +function applyAcceptedForwardTimelineUnits(params: { + units: TimelineUnit[]; + epoch: string; + currentTail: StreamItem[]; + currentHead: StreamItem[]; + currentEndSeq: number | undefined; +}): { tail: StreamItem[]; head: StreamItem[] } { + const reconciled = reconcileOverlappingProjectedStreamItems({ + tail: params.currentTail, + head: params.currentHead, + units: params.units, + epoch: params.epoch, + currentEndSeq: params.currentEndSeq, + }); + let tail = reconciled.tail; + let head = reconciled.head; + let delayedHistoryTail: StreamItem[] = []; + let delayedHistoryHead: StreamItem[] = []; + + for (const unit of params.units) { + if (reconciled.reconciledUnits.has(unit)) continue; + const nextOptimistic = findOptimisticUserMessage({ tail, head }); + if (nextOptimistic && matchesOptimisticUserMessage({ unit, optimistic: nextOptimistic.item })) { + const precedingItems = flushHeadToTail(delayedHistoryTail, delayedHistoryHead); + delayedHistoryTail = []; + delayedHistoryHead = []; + const applied = acknowledgeOptimisticUserMessage({ + tail, + head, + unit, + epoch: params.epoch, + optimistic: nextOptimistic, + precedingItems, + }); + tail = applied.tail; + head = applied.head; + continue; + } + if (nextOptimistic) { + if ( + unit.event.type === "timeline" && + (unit.event.item.type === "assistant_message" || unit.event.item.type === "reasoning") + ) { + delayedHistoryHead = reduceStreamUpdate(delayedHistoryHead, unit.event, unit.timestamp, { + source: "canonical", + timelineCursor: { epoch: params.epoch, seq: unit.seqEnd }, + }); + continue; + } + const applied = applyStreamEvent({ + tail: delayedHistoryTail, + head: delayedHistoryHead, + event: unit.event, + timestamp: unit.timestamp, + source: "canonical", + timelineCursor: { epoch: params.epoch, seq: unit.seqEnd }, + }); + delayedHistoryTail = applied.tail; + delayedHistoryHead = applied.head; + continue; + } + const applied = applyCanonicalForwardUnit({ tail, head, unit, epoch: params.epoch }); + tail = applied.tail; + head = applied.head; + } + + const remainingOptimistic = findOptimisticUserMessage({ tail, head }); + if (!remainingOptimistic) return { tail, head }; + const precedingItems = flushHeadToTail(delayedHistoryTail, delayedHistoryHead); + if (precedingItems.length === 0) return { tail, head }; + return replaceOptimisticUserMessage({ + tail, + head, + position: remainingOptimistic, + precedingItems, + replacement: [remainingOptimistic.item], + }); +} + function applyTimelineIncrementalPath(args: { timelineUnits: TimelineUnit[]; payload: ProcessTimelineResponseInput["payload"]; @@ -586,39 +959,16 @@ function applyTimelineIncrementalPath(args: { { source: "canonical" }, ); nextTail = mergePrependedCanonicalTail(olderTail, currentTail); - } else if (currentHead.length > 0) { - for (const { event, timestamp, seqEnd } of acceptedUnits) { - const timelineCursor = { epoch: payload.epoch, seq: seqEnd }; - const replacedHead = replaceLiveAssistantWithProjectedText({ - head: nextHead, - event, - timestamp, - timelineCursor, - }); - if (replacedHead) { - nextHead = replacedHead; - continue; - } - const applied = applyStreamEvent({ - tail: nextTail, - head: nextHead, - event, - timestamp, - source: "canonical", - timelineCursor, - }); - nextTail = applied.tail; - nextHead = applied.head; - } } else { - nextTail = acceptedUnits.reduce( - (state, { event, timestamp, seqEnd }) => - reduceStreamUpdate(state, event, timestamp, { - source: "canonical", - timelineCursor: { epoch: payload.epoch, seq: seqEnd }, - }), - currentTail, - ); + const applied = applyAcceptedForwardTimelineUnits({ + units: acceptedUnits, + epoch: payload.epoch, + currentTail: nextTail, + currentHead: nextHead, + currentEndSeq: currentCursor?.endSeq, + }); + nextTail = applied.tail; + nextHead = applied.head; } } diff --git a/packages/app/src/timeline/viewed-timeline-sync.test.ts b/packages/app/src/timeline/viewed-timeline-sync.test.ts index 511650f62..2e7fa6dd1 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.test.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.test.ts @@ -1,6 +1,9 @@ import { expect, test } from "vitest"; import type { ProjectedTimelineForwardFetchPlan } from "./timeline-sync-plan"; -import { createViewedTimelineSync } from "./viewed-timeline-sync"; +import { + createViewedTimelineSync, + VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS, +} from "./viewed-timeline-sync"; interface Deferred { promise: Promise; @@ -34,6 +37,7 @@ interface TimelineFetch { class TimelineWorld { readonly errors: string[] = []; readonly sync = createViewedTimelineSync({ + initialDeliveryMode: "selective", setSubscription: async (agentIds) => { const result = deferred(); this.memberships.push({ @@ -69,13 +73,17 @@ class TimelineWorld { const waiter = this.errorWaiters.shift(); if (waiter) waiter(this.errors.at(-1) ?? ""); }, - scheduleRetry: (retry) => { - this.retries.push(retry); + schedule: (task, delayMs) => { + const scheduled = { task, delayMs }; + this.scheduled.push(scheduled); const waiter = this.retryWaiters.shift(); - if (waiter) waiter(this.retries.shift()!); + if (waiter && delayMs === 1_000) { + this.scheduled.splice(this.scheduled.indexOf(scheduled), 1); + waiter(task); + } return () => { - const index = this.retries.indexOf(retry); - if (index >= 0) this.retries.splice(index, 1); + const index = this.scheduled.indexOf(scheduled); + if (index >= 0) this.scheduled.splice(index, 1); }; }, }); @@ -90,7 +98,7 @@ class TimelineWorld { private readonly cursors = new Map(); private readonly authoritativeHistory = new Set(); private readonly errorWaiters: Array<(message: string) => void> = []; - private readonly retries: Array<() => void> = []; + private readonly scheduled: Array<{ task: () => void; delayMs: number }> = []; private readonly retryWaiters: Array<(retry: () => void) => void> = []; setCursor(agentId: string, endSeq: number): void { @@ -129,11 +137,25 @@ class TimelineWorld { } nextRetry(): Promise<() => void> { - const retry = this.retries.shift(); - if (retry) return Promise.resolve(retry); + const index = this.scheduled.findIndex((entry) => entry.delayMs === 1_000); + if (index >= 0) return Promise.resolve(this.scheduled.splice(index, 1)[0].task); return new Promise((resolve) => this.retryWaiters.push(resolve)); } + runUnsubscribeGrace(): void { + const index = this.scheduled.findIndex( + (entry) => entry.delayMs === VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS, + ); + expect(index).toBeGreaterThanOrEqual(0); + this.scheduled.splice(index, 1)[0].task(); + } + + expectNoPendingUnsubscribe(): void { + expect( + this.scheduled.filter((entry) => entry.delayMs === VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS), + ).toEqual([]); + } + private releaseMembershipWaiter(): void { const waiter = this.membershipWaiters.shift(); if (!waiter) return; @@ -211,6 +233,7 @@ test("membership changes during acknowledgement never catch up the stale set", a const staleMembership = await world.nextMembership(); world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]); + world.runUnsubscribeGrace(); staleMembership.succeed(); const currentMembership = await world.nextMembership(); currentMembership.succeed(); @@ -236,6 +259,7 @@ test("removing one agent during paging cancels only that agent", async () => { ]); world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]); + world.runUnsubscribeGrace(); const replacement = await world.nextMembership(); agentA.respond({ hasNewer: true, seq: 4 }); agentB.respond({ hasNewer: true, seq: 7 }); @@ -283,7 +307,9 @@ test("overlapping sources deduplicate membership and source removal preserves re world.sync.replaceVisibleAgentIds("left-route", []); world.expectNoPendingMembership(); + world.expectNoPendingUnsubscribe(); world.sync.replaceVisibleAgentIds("right-route", ["agent-b"]); + world.runUnsubscribeGrace(); const remaining = await world.nextMembership(); remaining.succeed(); @@ -417,6 +443,7 @@ test("stale membership retry cannot overwrite a newer effective set", async () = const staleRetry = await world.nextRetry(); world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]); + world.runUnsubscribeGrace(); const current = await world.nextMembership(); staleRetry(); current.succeed(); @@ -446,3 +473,153 @@ test("membership retry cannot run while disconnected", async () => { expect(restored.agentIds).toEqual(["agent-a"]); }); + +test("quickly returning to an agent cancels its pending unsubscribe without another catch-up", async () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + const membership = await world.nextMembership(); + membership.succeed(); + const initialCatchUp = await world.nextFetch("agent-a"); + initialCatchUp.respond({ hasNewer: false }); + + world.sync.replaceVisibleAgentIds("workspace", []); + world.expectNoPendingMembership(); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + + world.expectNoPendingUnsubscribe(); + world.expectNoPendingMembership(); + world.expectNoPendingFetch(); +}); + +test("unsubscribe grace expiry removes the agent exactly once", async () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + const membership = await world.nextMembership(); + membership.succeed(); + const catchUp = await world.nextFetch("agent-a"); + catchUp.respond({ hasNewer: false }); + + world.sync.replaceVisibleAgentIds("workspace", []); + world.runUnsubscribeGrace(); + const unsubscribe = await world.nextMembership(); + unsubscribe.succeed(); + + expect(unsubscribe.agentIds).toEqual([]); + world.expectNoPendingUnsubscribe(); + world.expectNoPendingMembership(); +}); + +test("a new visible agent subscribes immediately while the previous agent lingers", async () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + const initialMembership = await world.nextMembership(); + initialMembership.succeed(); + const initialCatchUp = await world.nextFetch("agent-a"); + initialCatchUp.respond({ hasNewer: false }); + + world.sync.replaceVisibleAgentIds("workspace", ["agent-b"]); + const expandedMembership = await world.nextMembership(); + expandedMembership.succeed(); + const agentBCatchUp = await world.nextFetch("agent-b"); + agentBCatchUp.respond({ hasNewer: false }); + + expect(expandedMembership.agentIds).toEqual(["agent-a", "agent-b"]); + world.runUnsubscribeGrace(); + const settledMembership = await world.nextMembership(); + settledMembership.succeed(); + expect(settledMembership.agentIds).toEqual(["agent-b"]); +}); + +test("backgrounding clears pending unsubscribe grace and membership immediately", async () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + const membership = await world.nextMembership(); + membership.succeed(); + const catchUp = await world.nextFetch("agent-a"); + catchUp.respond({ hasNewer: false }); + + world.sync.replaceVisibleAgentIds("workspace", []); + world.sync.setActive(false); + const unsubscribe = await world.nextMembership(); + unsubscribe.succeed(); + + expect(unsubscribe.agentIds).toEqual([]); + world.expectNoPendingUnsubscribe(); +}); + +test("disconnecting cancels pending unsubscribe grace without publishing on the closed socket", async () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + const membership = await world.nextMembership(); + membership.succeed(); + const catchUp = await world.nextFetch("agent-a"); + catchUp.respond({ hasNewer: false }); + + world.sync.replaceVisibleAgentIds("workspace", []); + world.sync.setConnected(false); + + world.expectNoPendingUnsubscribe(); + world.expectNoPendingMembership(); +}); + +test("disposing cancels pending unsubscribe grace", async () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + const membership = await world.nextMembership(); + membership.succeed(); + const catchUp = await world.nextFetch("agent-a"); + catchUp.respond({ hasNewer: false }); + + world.sync.replaceVisibleAgentIds("workspace", []); + world.sync.dispose(); + + world.expectNoPendingUnsubscribe(); + world.expectNoPendingMembership(); +}); + +test("legacy delivery skips subscription RPCs while retaining visibility catch-up and gap recovery", async () => { + const world = new TimelineWorld(); + world.sync.setDeliveryMode("legacy"); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + world.sync.setConnected(true); + + world.expectNoPendingMembership(); + const initial = await world.nextFetch("agent-a"); + initial.respond({ hasNewer: false }); + + world.sync.recoverGap("agent-a", { epoch: "epoch-agent-a", endSeq: 10 }); + const recovery = await world.nextFetch("agent-a"); + recovery.respond({ hasNewer: false }); + + expect(recovery.request).toEqual({ + direction: "after", + cursor: { epoch: "epoch-agent-a", seq: 10 }, + limit: 100, + projection: "projected", + }); +}); + +test("switching from legacy to selective delivery publishes membership and catches up once", async () => { + const world = new TimelineWorld(); + world.sync.setDeliveryMode("legacy"); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + world.sync.setConnected(true); + const legacyCatchUp = await world.nextFetch("agent-a"); + legacyCatchUp.respond({ hasNewer: false }); + + world.sync.setDeliveryMode("selective"); + const membership = await world.nextMembership(); + membership.succeed(); + const catchUp = await world.nextFetch("agent-a"); + catchUp.respond({ hasNewer: false }); + + expect(membership.agentIds).toEqual(["agent-a"]); + world.expectNoPendingMembership(); + world.expectNoPendingFetch(); +}); diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index f0f2c0870..4aa8a26dd 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -12,6 +12,7 @@ interface TimelinePageResult { } interface ViewedTimelineSyncPorts { + initialDeliveryMode: TimelineDeliveryMode; setSubscription(agentIds: string[]): Promise; readCursor(agentId: string): AgentTimelineCursorState | undefined; hasAuthoritativeHistory(agentId: string): boolean; @@ -20,9 +21,11 @@ interface ViewedTimelineSyncPorts { request: ProjectedTimelineForwardFetchPlan, ): Promise; reportError(error: unknown): void; - scheduleRetry(retry: () => void): () => void; + schedule(task: () => void, delayMs: number): () => void; } +export type TimelineDeliveryMode = "legacy" | "selective"; + export interface ViewedTimelineUiBridge { replaceVisibleAgentIds(sourceId: string, agentIds: string[]): void; } @@ -30,10 +33,14 @@ export interface ViewedTimelineUiBridge { export interface ViewedTimelineSync extends ViewedTimelineUiBridge { setActive(active: boolean): void; setConnected(connected: boolean): void; + setDeliveryMode(mode: TimelineDeliveryMode): void; recoverGap(agentId: string, cursor: { epoch: string; endSeq: number }): void; dispose(): void; } +const RETRY_DELAY_MS = 1_000; +export const VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS = 5_000; + type CatchUpStatus = "running" | "complete" | "error"; interface CatchUpState { @@ -80,8 +87,10 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed const catchUps = new Map(); const catchUpGenerations = new Map(); const pendingGaps = new Map(); + const lingeringRemovals = new Map void>(); let active = true; let connected = false; + let deliveryMode = ports.initialDeliveryMode; let disposed = false; let desired: string[] = []; let acknowledged: string[] = []; @@ -91,7 +100,9 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed let membershipNeedsRetry = false; let cancelMembershipRetry: (() => void) | null = null; - const effectiveAgentIds = () => (active ? normalizeAgentIds([...sources.values()].flat()) : []); + const visibleAgentIds = () => (active ? normalizeAgentIds([...sources.values()].flat()) : []); + const effectiveAgentIds = () => + normalizeAgentIds([...visibleAgentIds(), ...lingeringRemovals.keys()]); const isAcknowledged = (agentId: string) => acknowledged.includes(agentId); const isDesired = (agentId: string) => desired.includes(agentId); @@ -139,11 +150,11 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed catchUps.set(agentId, { generation, status: "complete" }); } catch (error) { if (catchUps.get(agentId)?.generation === generation) { - const cancelRetry = ports.scheduleRetry(() => { + const cancelRetry = ports.schedule(() => { const current = catchUps.get(agentId); if (current?.generation !== generation || current.status !== "error") return; startCatchUp(agentId); - }); + }, RETRY_DELAY_MS); catchUps.set(agentId, { generation, status: "error", cancelRetry }); ports.reportError(error); } @@ -188,7 +199,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed }; const reconcileLatestMembership = async (): Promise => { - if (disposed || !connected) return; + if (disposed || !connected || deliveryMode !== "selective") return; const generation = membershipGeneration; const requested = desired; if (!membershipNeedsRetry && sameAgentIds(requested, acknowledged)) return; @@ -198,7 +209,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed } catch (error) { membershipNeedsRetry = true; cancelMembershipRetry?.(); - cancelMembershipRetry = ports.scheduleRetry(() => { + cancelMembershipRetry = ports.schedule(() => { cancelMembershipRetry = null; if ( disposed || @@ -209,13 +220,13 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed return; } void reconcileMembership(); - }); + }, RETRY_DELAY_MS); ports.reportError(error); return; } cancelMembershipRetry?.(); cancelMembershipRetry = null; - if (disposed || !connected) return; + if (disposed || !connected || deliveryMode !== "selective") return; acknowledged = requested; if (generation !== membershipGeneration) { await reconcileLatestMembership(); @@ -236,12 +247,13 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed await reconcileLatestMembership(); } finally { reconciling = false; - if (reconcileRequested && !disposed && connected) { + if (reconcileRequested && !disposed && connected && deliveryMode === "selective") { reconcileRequested = false; void reconcileMembership(); } else if ( !disposed && connected && + deliveryMode === "selective" && !membershipNeedsRetry && !sameAgentIds(desired, acknowledged) ) { @@ -256,10 +268,9 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed } }; - const publishEffectiveMembership = () => { - const nextDesired = effectiveAgentIds(); + const commitDesiredMembership = (nextDesired: string[]) => { if (sameAgentIds(nextDesired, desired)) { - if (membershipNeedsRetry) void reconcileMembership(); + if (deliveryMode === "selective" && membershipNeedsRetry) void reconcileMembership(); retryFailedCatchUps(); return; } @@ -271,25 +282,60 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed cancelMembershipRetry = null; desired = nextDesired; membershipGeneration += 1; + if (deliveryMode === "legacy") { + acknowledged = connected ? desired : []; + if (connected) startAcknowledgedCatchUps(); + return; + } void reconcileMembership(); }; + const clearLingeringRemovals = () => { + for (const cancel of lingeringRemovals.values()) cancel(); + lingeringRemovals.clear(); + }; + + const publishVisibleMembership = (allowGrace: boolean) => { + const visible = visibleAgentIds(); + for (const agentId of visible) { + lingeringRemovals.get(agentId)?.(); + lingeringRemovals.delete(agentId); + } + + if (allowGrace && connected && active && deliveryMode === "selective") { + for (const agentId of desired) { + if (visible.includes(agentId) || lingeringRemovals.has(agentId)) continue; + const cancel = ports.schedule(() => { + lingeringRemovals.delete(agentId); + commitDesiredMembership(effectiveAgentIds()); + }, VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS); + lingeringRemovals.set(agentId, cancel); + } + } else { + clearLingeringRemovals(); + } + + commitDesiredMembership(effectiveAgentIds()); + }; + return { replaceVisibleAgentIds(sourceId, agentIds) { const normalized = normalizeAgentIds(agentIds); if (normalized.length === 0) sources.delete(sourceId); else sources.set(sourceId, normalized); - publishEffectiveMembership(); + publishVisibleMembership(true); }, setActive(nextActive) { if (active === nextActive) return; active = nextActive; - publishEffectiveMembership(); + publishVisibleMembership(false); }, setConnected(nextConnected) { if (connected === nextConnected) return; connected = nextConnected; if (!connected) { + clearLingeringRemovals(); + commitDesiredMembership(visibleAgentIds()); cancelMembershipRetry?.(); cancelMembershipRetry = null; acknowledged = []; @@ -298,7 +344,26 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed return; } membershipGeneration += 1; - void reconcileMembership(); + if (deliveryMode === "legacy") { + acknowledged = desired; + startAcknowledgedCatchUps(); + } else { + void reconcileMembership(); + } + }, + setDeliveryMode(nextMode) { + if (deliveryMode === nextMode) return; + deliveryMode = nextMode; + clearLingeringRemovals(); + cancelMembershipRetry?.(); + cancelMembershipRetry = null; + membershipNeedsRetry = false; + membershipGeneration += 1; + for (const agentId of desired) cancelCatchUp(agentId); + desired = visibleAgentIds(); + acknowledged = deliveryMode === "legacy" && connected ? desired : []; + if (deliveryMode === "selective" && connected) void reconcileMembership(); + else if (connected) startAcknowledgedCatchUps(); }, recoverGap(agentId, cursor) { if (!isDesired(agentId)) return; @@ -309,6 +374,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed }, dispose() { disposed = true; + clearLingeringRemovals(); cancelMembershipRetry?.(); cancelMembershipRetry = null; sources.clear();