jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Fix duplicated and out-of-order agent chat messages (#2185)

Browse Source 
* fix(app): keep agent timelines ordered during catch-up

Projected catch-up pages can overlap live rows and arrive after optimistic
prompts. Reconcile canonical projections before the existing stream reducer,
and retain selective subscriptions briefly across quick view switches.

* refactor(app): make optimistic catch-up consumption explicit

* fix(app): preserve catch-up message order

* fix(app): isolate delayed catch-up history

* fix(app): anchor catch-up before live turns

* fix(app): stage complete catch-up history

* fix(app): keep timeline arrays Hermes-safe

* refactor(app): preserve prompt positions during catch-up
This commit is contained in:
Mohamed Boudra
2026-07-18 13:49:32 +02:00
committed by GitHub
parent 99da5736db
commit 98f6611362
6 changed files with 1325 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/timeline-sync.md
+30
View File
@@ -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`
packages/app/src/contexts/session-context.tsx
+26 -7
View File
@@ -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,
packages/app/src/timeline/session-stream-reducers.test.ts
+619 -3
View File
@@ -107,7 +107,10 @@ function makeStreamReducerEvent(
};
}
function makeAssistantItem(text: string, id = `assistant-${text.length}`): StreamItem {
function makeAssistantItem(
text: string,
id = `assistant-${text.length}`,
): Extract<StreamItem, { kind: "assistant_message" }> {
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,
packages/app/src/timeline/session-stream-reducers.ts
+383 -33
View File
@@ -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<TimelineUnit> } {
let tail = params.tail;
let head = params.head;
const reconciledUnits = new Set<TimelineUnit>();
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<StreamItem[]>(
(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;
}
}
packages/app/src/timeline/viewed-timeline-sync.test.ts
+186 -9
View File
@@ -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<T> {
promise: Promise<T>;
@@ -34,6 +37,7 @@ interface TimelineFetch {
class TimelineWorld {
readonly errors: string[] = [];
readonly sync = createViewedTimelineSync({
initialDeliveryMode: "selective",
setSubscription: async (agentIds) => {
const result = deferred<void>();
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<string, { epoch: string; startSeq: number; endSeq: number }>();
private readonly authoritativeHistory = new Set<string>();
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();
});
packages/app/src/timeline/viewed-timeline-sync.ts
+81 -15
View File
@@ -12,6 +12,7 @@ interface TimelinePageResult {
}
interface ViewedTimelineSyncPorts {
initialDeliveryMode: TimelineDeliveryMode;
setSubscription(agentIds: string[]): Promise<void>;
readCursor(agentId: string): AgentTimelineCursorState | undefined;
hasAuthoritativeHistory(agentId: string): boolean;
@@ -20,9 +21,11 @@ interface ViewedTimelineSyncPorts {
request: ProjectedTimelineForwardFetchPlan,
): Promise<TimelinePageResult>;
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<string, CatchUpState>();
const catchUpGenerations = new Map<string, number>();
const pendingGaps = new Map<string, ProjectedTimelineForwardFetchPlan>();
const lingeringRemovals = new Map<string, () => 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<void> => {
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();