puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-05 16:04:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(timeline): resume agents from the latest history

Browse Source 
Use a bounded tail probe on resume so long background runs appear atomically instead of replaying every missed page. Preserve unchanged views, reconcile local/live rows, and keep skipped history reachable through backward pagination.
This commit is contained in:
Mohamed Boudra
2026-07-31 11:36:31 +02:00
parent 142d4d2535
commit 3f63df678a
16 changed files with 654 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/contexts/session-context.tsx
+22 -19
View File
@@ -351,6 +351,9 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
const setAgentTimelineCursor = useSessionStore((state) => state.setAgentTimelineCursor);
const setInitializingAgents = useSessionStore((state) => state.setInitializingAgents);
const bumpHistorySyncGeneration = useSessionStore((state) => state.bumpHistorySyncGeneration);
const markAgentHistorySynchronized = useSessionStore(
(state) => state.markAgentHistorySynchronized,
);
const applyAgentTimelineResponseState = useSessionStore(
(state) => state.applyAgentTimelineResponseState,
);
@@ -619,14 +622,18 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
return;
}
applyAgentTimelineResponseState(serverId, agentId, {
items: result.tail,
head: result.head,
range: result.cursorChanged ? (result.cursor ?? null) : (currentCursor ?? null),
older: payload.hasOlder ? "available" : "none",
synchronized: shouldMarkAuthoritativeHistoryApplied,
acknowledgedClientMessageIds: result.acknowledgedClientMessageIds,
});
if (result.commit === "discard") {
markAgentHistorySynchronized(serverId, agentId);
} else {
applyAgentTimelineResponseState(serverId, agentId, {
items: result.tail,
head: result.head,
range: result.cursorChanged ? (result.cursor ?? null) : (currentCursor ?? null),
older: result.older,
synchronized: shouldMarkAuthoritativeHistoryApplied,
acknowledgedClientMessageIds: result.acknowledgedClientMessageIds,
});
}
executeTimelineSideEffects({
sideEffects: result.sideEffects,
@@ -643,7 +650,13 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
setInitializingAgents,
});
},
[applyAgentTimelineResponseState, recoverTimelineGap, serverId, setInitializingAgents],
[
applyAgentTimelineResponseState,
markAgentHistorySynchronized,
recoverTimelineGap,
serverId,
setInitializingAgents,
],
);
useEffect(() => {
@@ -654,16 +667,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
const sync = createViewedTimelineSync({
initialDeliveryMode,
setSubscription: (agentIds) => client.setAgentTimelineSubscription(agentIds),
readCursor: (agentId) => {
const timeline = selectAgentTimelineState(
useSessionStore.getState().sessions[serverId],
agentId,
);
return timeline.status === "synced" ? (timeline.range ?? undefined) : undefined;
},
hasAuthoritativeHistory: (agentId) =>
selectAgentTimelineState(useSessionStore.getState().sessions[serverId], agentId).status ===
"synced",
fetchPage: async (agentId, request) => {
const session = useSessionStore.getState().sessions[serverId];
const initKey = getInitKey(serverId, agentId);
packages/app/src/hooks/use-agent-initialization.test.ts
+3 -4
View File
@@ -49,7 +49,7 @@ afterEach(() => {
});
describe("ensureAgentIsInitialized", () => {
it("requests bounded projected catch-up after the current cursor when authoritative history is loaded", () => {
it("requests a bounded projected tail when authoritative history is loaded", () => {
const client = new FakeDaemonClient();
const runtime = new FakeTimelineRuntime();
useSessionStore.getState().initializeSession(serverId, client as never);
@@ -74,14 +74,13 @@ describe("ensureAgentIsInitialized", () => {
serverId,
agentId,
request: {
direction: "after",
cursor: { epoch: "epoch-1", seq: 42 },
direction: "tail",
limit: TIMELINE_FETCH_PAGE_SIZE,
projection: "projected",
},
},
]);
expect(getInitDeferred(getInitKey(serverId, agentId))?.requestDirection).toBe("after");
expect(getInitDeferred(getInitKey(serverId, agentId))?.requestDirection).toBe("tail");
});
it("requests a bounded projected tail when no authoritative cursor is available", () => {
packages/app/src/hooks/use-agent-initialization.ts
+3 -7
View File
@@ -1,7 +1,7 @@
import { useCallback, useMemo } from "react";
import { useTranslation } from "react-i18next";
import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
import { selectAgentTimelineState, useSessionStore } from "@/stores/session-store";
import { useSessionStore } from "@/stores/session-store";
import {
createInitDeferred,
getInitDeferred,
@@ -11,7 +11,7 @@ import {
refreshInitTimeout,
} from "@/utils/agent-initialization";
import { getHostRuntimeStore, type HostRuntimeStore } from "@/runtime/host-runtime";
import { planInitialAgentTimelineSync, planTimelineTailFetch } from "@/timeline/timeline-sync-plan";
import { planTimelineTailFetch } from "@/timeline/timeline-sync-plan";
import { i18n } from "@/i18n/i18next";
export type SetAgentInitializing = (agentId: string, initializing: boolean) => void;
@@ -51,11 +51,7 @@ export function ensureAgentIsInitialized(input: EnsureAgentIsInitializedInput):
return existing.promise;
}
const session = useSessionStore.getState().sessions[serverId];
const timeline = selectAgentTimelineState(session, agentId);
const cursor = timeline.status === "synced" ? (timeline.range ?? undefined) : undefined;
const hasAuthoritativeHistory = timeline.status === "synced";
const timelineRequest = planInitialAgentTimelineSync({ cursor, hasAuthoritativeHistory });
const timelineRequest = planTimelineTailFetch();
const deferred = createInitDeferred(key, timelineRequest.direction);
refreshAgentInitializationTimeout({ key, agentId, setAgentInitializing });
packages/app/src/stores/session-store.test.ts
+31
View File
@@ -134,6 +134,37 @@ describe("agent timeline state", () => {
).toEqual({ status: "synced", items: [], range: null, older: "none" });
});
it("preserves older availability when an applied response leaves it unchanged", () => {
initializeTestSession();
const store = useSessionStore.getState();
store.applyAgentTimelineResponseState("test-server", "agent-1", {
items: [],
head: [],
range: { epoch: "epoch-1", startSeq: 200, endSeq: 240 },
older: "available",
synchronized: true,
acknowledgedClientMessageIds: [],
});
store.applyAgentTimelineResponseState("test-server", "agent-1", {
items: [],
head: [],
range: { epoch: "epoch-1", startSeq: 200, endSeq: 240 },
older: "unchanged",
synchronized: false,
acknowledgedClientMessageIds: [],
});
expect(
selectAgentTimelineState(useSessionStore.getState().sessions["test-server"], "agent-1"),
).toEqual({
status: "synced",
items: [],
range: { epoch: "epoch-1", startSeq: 200, endSeq: 240 },
older: "available",
});
});
it("lets a later running directory transition raise but never lower an open turn", () => {
initializeTestSession();
const store = useSessionStore.getState();
packages/app/src/stores/session-store.ts
+4 -2
View File
@@ -554,7 +554,7 @@ interface SessionStoreActions {
items: StreamItem[];
head: StreamItem[];
range: AgentTimelineCursorState | null;
older: "available" | "none";
older: "available" | "none" | "unchanged";
synchronized: boolean;
acknowledgedClientMessageIds: string[];
},
@@ -1519,7 +1519,9 @@ export const useSessionStore = create<SessionStore>()(
if (state.range) nextCursor.set(agentId, state.range);
else nextCursor.delete(agentId);
const nextHasOlder = new Map(session.agentTimelineHasOlder);
nextHasOlder.set(agentId, state.older === "available");
if (state.older !== "unchanged") {
nextHasOlder.set(agentId, state.older === "available");
}
const nextAuthoritative = new Map(session.agentAuthoritativeHistoryApplied);
const nextSyncGeneration = new Map(session.agentHistorySyncGeneration);
const currentSubmissions = session.messageSubmissions.get(agentId) ?? [];
packages/app/src/timeline/session-stream-reducers.test.ts
+213
View File
@@ -197,6 +197,7 @@ const baseTimelineInput: ProcessTimelineResponseInput = {
direction: "after",
reset: false,
epoch: "epoch-1",
window: { minSeq: 1, maxSeq: 0, nextSeq: 1 },
startCursor: null,
endCursor: null,
entries: [],
@@ -228,6 +229,178 @@ const baseStreamInput: ProcessAgentStreamEventInput = {
// ---------------------------------------------------------------------------
describe("processTimelineResponse", () => {
it("discards an unchanged resume tail without replacing timeline state", () => {
const currentTail = [makeAssistantItem("existing tail", "existing-tail")];
const currentHead = [makeAssistantItem("existing head", "existing-head")];
const result = processTimelineResponse({
...baseTimelineInput,
currentTail,
currentHead,
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 40 },
payload: {
...baseTimelineInput.payload,
direction: "tail",
window: { minSeq: 1, maxSeq: 40, nextSeq: 41 },
startCursor: { seq: 1 },
endCursor: { seq: 40 },
entries: [makeTimelineEntry(1, "existing tail", "assistant_message", 40)],
},
});
expect(result.commit).toBe("discard");
expect(result.tail).toBe(currentTail);
expect(result.head).toBe(currentHead);
expect(result.cursorChanged).toBe(false);
expect(result.sideEffects).toEqual([{ type: "flush_pending_updates" }]);
});
it("appends only rows newer than the cursor from an overlapping resume tail", () => {
const loaded: StreamItem = {
kind: "user_message",
id: "loaded-30",
text: "already loaded",
timestamp: new Date(1030),
timelineCursor: { epoch: "epoch-1", seq: 30 },
};
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [loaded],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 40 },
payload: {
...baseTimelineInput.payload,
direction: "tail",
window: { minSeq: 1, maxSeq: 42, nextSeq: 43 },
startCursor: { seq: 30 },
endCursor: { seq: 42 },
entries: [
makeTimelineEntry(30, "already loaded", "user_message"),
makeTimelineEntry(41, "new 41", "user_message"),
makeTimelineEntry(42, "new 42", "user_message"),
],
},
});
expect(result.commit).toBe("apply");
expect(result.tail[0]).toBe(loaded);
expect(getUserTexts(result.tail)).toEqual(["already loaded", "new 41", "new 42"]);
expect(result.cursor).toEqual({ epoch: "epoch-1", startSeq: 1, endSeq: 42 });
expect(result.sideEffects).toEqual([{ type: "flush_pending_updates" }]);
});
it("atomically replaces history when a resume tail leaves a middle gap", () => {
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [
{
kind: "user_message",
id: "stale-history",
text: "stale history",
timestamp: new Date(1040),
timelineCursor: { epoch: "epoch-1", seq: 40 },
},
],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 40 },
payload: {
...baseTimelineInput.payload,
direction: "tail",
window: { minSeq: 1, maxSeq: 240, nextSeq: 241 },
startCursor: { seq: 200 },
endCursor: { seq: 240 },
hasOlder: true,
entries: [
makeTimelineEntry(200, "latest 200", "user_message"),
makeTimelineEntry(240, "latest 240", "user_message"),
],
},
});
expect(result.commit).toBe("apply");
expect(getUserTexts(result.tail)).toEqual(["latest 200", "latest 240"]);
expect(result.cursor).toEqual({ epoch: "epoch-1", startSeq: 200, endSeq: 240 });
expect(result.sideEffects).toEqual([{ type: "flush_pending_updates" }]);
});
it("reconciles in-flight live rows and an unresolved submission during gap replacement", () => {
const unresolved = makeSubmittedUserMessage("unresolved prompt", "client-unresolved");
const coveredLive: StreamItem = {
...makeAssistantItem("covered live", "covered-live"),
timelineCursor: { epoch: "epoch-1", seq: 235 },
};
const newerLive: StreamItem = {
...makeAssistantItem("newer live", "newer-live"),
timelineCursor: { epoch: "epoch-1", seq: 245 },
};
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [makeAssistantItem("stale history", "stale-history"), unresolved],
currentHead: [coveredLive, newerLive],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 40 },
payload: {
...baseTimelineInput.payload,
direction: "tail",
window: { minSeq: 1, maxSeq: 240, nextSeq: 241 },
startCursor: { seq: 200 },
endCursor: { seq: 240 },
hasOlder: true,
entries: [makeTimelineEntry(240, "canonical tail")],
},
});
expect(getAssistantTexts(result.tail)).toEqual(["canonical tail"]);
expect(getUserTexts(result.tail)).toEqual(["unresolved prompt"]);
expect(result.head).toEqual([newerLive]);
});
it("replaces destructively when a resume tail has a new epoch", () => {
const acknowledged: StreamItem = {
...makeSubmittedUserMessage("acknowledged old prompt", "client-acknowledged"),
messageId: "provider-acknowledged",
};
const sending = makeSubmittedUserMessage("still sending", "client-sending");
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [acknowledged, sending],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 40 },
sendingClientMessageIds: ["client-sending"],
payload: {
...baseTimelineInput.payload,
direction: "tail",
epoch: "epoch-2",
window: { minSeq: 1, maxSeq: 1, nextSeq: 2 },
startCursor: { seq: 1 },
endCursor: { seq: 1 },
entries: [makeTimelineEntry(1, "new epoch")],
},
});
expect(getAssistantTexts(result.tail)).toEqual(["new epoch"]);
expect(getUserTexts(result.tail)).toEqual(["still sending"]);
expect(result.cursor).toEqual({ epoch: "epoch-2", startSeq: 1, endSeq: 1 });
});
it("replaces destructively when the same epoch rewinds behind the local cursor", () => {
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [makeAssistantItem("future history", "future-history")],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 40 },
payload: {
...baseTimelineInput.payload,
direction: "tail",
window: { minSeq: 1, maxSeq: 10, nextSeq: 11 },
startCursor: { seq: 1 },
endCursor: { seq: 10 },
entries: [makeTimelineEntry(10, "rewound history")],
},
});
expect(getAssistantTexts(result.tail)).toEqual(["rewound history"]);
expect(result.cursor).toEqual({ epoch: "epoch-1", startSeq: 1, endSeq: 10 });
});
it("preserves the canonical end cursor on a projected assistant message", () => {
const result = processTimelineResponse({
...baseTimelineInput,
@@ -2266,6 +2439,7 @@ describe("processTimelineResponse", () => {
// No new items appended (all dropped as stale)
expect(result.tail).toBe(baseTimelineInput.currentTail);
expect(result.cursorChanged).toBe(false);
expect(result.older).toBe("unchanged");
});
it("prepends older before-cursor entries and only expands the start cursor", () => {
@@ -2310,6 +2484,43 @@ describe("processTimelineResponse", () => {
startSeq: 1,
endSeq: 5,
});
expect(result.older).toBe("none");
});
it("drops a stale before page anchored before a resume-tail replacement", () => {
const currentTail: StreamItem[] = [
{
kind: "user_message",
id: "current-200",
text: "current-200",
timestamp: new Date(3200),
timelineCursor: { epoch: "epoch-1", seq: 200 },
},
];
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 200,
endSeq: 240,
};
const result = processTimelineResponse({
...baseTimelineInput,
currentTail,
currentCursor: existingCursor,
payload: {
...baseTimelineInput.payload,
direction: "before",
epoch: "epoch-1",
startCursor: { seq: 1 },
endCursor: { seq: 39 },
entries: [makeTimelineEntry(39, "stale older page", "user_message")],
},
});
expect(result.tail).toBe(currentTail);
expect(result.cursor).toBe(existingCursor);
expect(result.cursorChanged).toBe(false);
expect(result.older).toBe("unchanged");
});
it("does not reconcile an active submitted user message from a before-page response", () => {
@@ -3344,6 +3555,7 @@ describe("processAgentStreamEvents", () => {
direction: "tail",
reset: false,
epoch: "epoch-1",
window: { minSeq: 186, maxSeq: 186, nextSeq: 187 },
startCursor: { seq: 186 },
endCursor: { seq: 186 },
entries: [makeTimelineEntry(186, seq186Text)],
@@ -3387,6 +3599,7 @@ describe("processAgentStreamEvents", () => {
direction: "tail",
reset: false,
epoch: "epoch-1",
window: { minSeq: 10, maxSeq: 10, nextSeq: 11 },
startCursor: { seq: 10 },
endCursor: { seq: 10 },
entries: [makeTimelineEntry(10, "Call-site API — exactly one primitive. Not")],
packages/app/src/timeline/session-stream-reducers.ts
+97 -19
View File
@@ -75,6 +75,7 @@ export interface ProcessTimelineResponseInput {
direction: TimelineDirection;
reset: boolean;
epoch: string;
window: { minSeq: number; maxSeq: number; nextSeq: number };
startCursor: { seq: number } | null;
endCursor: { seq: number } | null;
entries: TimelineResponseEntry[];
@@ -92,10 +93,12 @@ export interface ProcessTimelineResponseInput {
}
export interface ProcessTimelineResponseOutput {
commit: "apply" | "discard";
tail: StreamItem[];
head: StreamItem[];
cursor: TimelineCursor | null | undefined;
cursorChanged: boolean;
older: "available" | "none" | "unchanged";
initResolution: "resolve" | "reject" | null;
clearInitializing: boolean;
error: string | null;
@@ -116,6 +119,7 @@ interface TimelinePathResult {
head: StreamItem[];
cursor: TimelineCursor | null | undefined;
cursorChanged: boolean;
older: "available" | "none" | "unchanged";
sideEffects: TimelineReducerSideEffect[];
acknowledgedClientMessageIds: string[];
}
@@ -177,6 +181,39 @@ function deriveBootstrapTailTimelinePolicy({
};
}
type ResumeTailPolicy =
| { kind: "not_resume" }
| { kind: "discard" }
| { kind: "append" }
| { kind: "replace"; preserveContinuity: boolean };
function deriveResumeTailPolicy(input: {
direction: TimelineDirection;
reset: boolean;
epoch: string;
windowMaxSeq: number;
pageStartSeq: number | null;
currentCursor: TimelineCursor | undefined;
bootstrapReplace: boolean;
}): ResumeTailPolicy {
if (input.direction !== "tail" || input.reset || input.bootstrapReplace || !input.currentCursor) {
return { kind: "not_resume" };
}
if (input.currentCursor.epoch !== input.epoch) {
return { kind: "replace", preserveContinuity: false };
}
if (input.windowMaxSeq === input.currentCursor.endSeq) {
return { kind: "discard" };
}
if (input.windowMaxSeq < input.currentCursor.endSeq) {
return { kind: "replace", preserveContinuity: false };
}
if (input.pageStartSeq !== null && input.pageStartSeq <= input.currentCursor.endSeq + 1) {
return { kind: "append" };
}
return { kind: "replace", preserveContinuity: true };
}
function shouldResolveTimelineInit({
hasActiveInitDeferred,
hasNewer,
@@ -255,6 +292,7 @@ function applyTimelineReplacePath(args: {
head,
cursor,
cursorChanged: true,
older: payload.hasOlder ? "available" : "none",
sideEffects,
acknowledgedClientMessageIds,
};
@@ -340,7 +378,7 @@ function acceptOlderTimelineUnits(args: {
if (
responseStartSeq === undefined ||
responseEndSeq === undefined ||
responseEndSeq >= currentCursor.startSeq
responseEndSeq !== currentCursor.startSeq - 1
) {
return { acceptedUnits: [], cursor: currentCursor, gapCursor: null };
}
@@ -733,6 +771,7 @@ function applyTimelineIncrementalPath(args: {
let nextHead = currentHead;
let nextCursor: TimelineCursor | null | undefined = currentCursor;
let cursorChanged = false;
let older: TimelinePathResult["older"] = "unchanged";
const sideEffects: TimelineReducerSideEffect[] = [];
let acknowledgedClientMessageIds: string[] = [];
@@ -742,6 +781,7 @@ function applyTimelineIncrementalPath(args: {
head: nextHead,
cursor: nextCursor,
cursorChanged,
older,
sideEffects,
acknowledgedClientMessageIds,
};
@@ -762,6 +802,7 @@ function applyTimelineIncrementalPath(args: {
if (acceptedUnits.length > 0) {
if (payload.direction === "before") {
older = payload.hasOlder ? "available" : "none";
const olderTail = hydrateStreamState(
acceptedUnits.map(({ event, timestamp, seqEnd }) => ({
event,
@@ -805,6 +846,7 @@ function applyTimelineIncrementalPath(args: {
head: nextHead,
cursor: nextCursor,
cursorChanged,
older,
sideEffects,
acknowledgedClientMessageIds,
};
@@ -829,10 +871,12 @@ export function processTimelineResponse(
// ------------------------------------------------------------------
if (payload.error) {
return {
commit: "apply",
tail: currentTail,
head: currentHead,
cursor: currentCursor,
cursorChanged: false,
older: "unchanged",
initResolution: hasActiveInitDeferred ? "reject" : null,
clearInitializing: isInitializing,
error: payload.error,
@@ -885,24 +929,56 @@ export function processTimelineResponse(
});
const replace = bootstrapPolicy.replace;
const sideEffects: TimelineReducerSideEffect[] = [];
const timelineResult = replace
? applyTimelineReplacePath({
timelineUnits,
payload,
bootstrapPolicy,
currentTail,
currentHead,
sendingClientMessageIds,
preserveContinuity: currentCursor?.epoch === payload.epoch || !payload.reset,
toHydratedEvents,
})
: applyTimelineIncrementalPath({
timelineUnits,
payload,
currentTail,
currentHead,
currentCursor,
});
const resumeTailPolicy = deriveResumeTailPolicy({
direction: payload.direction,
reset: payload.reset,
epoch: payload.epoch,
windowMaxSeq: payload.window.maxSeq,
pageStartSeq: payload.startCursor?.seq ?? null,
currentCursor,
bootstrapReplace: replace,
});
const discard = resumeTailPolicy.kind === "discard";
let timelineResult: TimelinePathResult;
if (discard) {
timelineResult = {
tail: currentTail,
head: currentHead,
cursor: currentCursor,
cursorChanged: false,
older: "unchanged",
sideEffects: [],
acknowledgedClientMessageIds: [],
};
} else if (replace || resumeTailPolicy.kind === "replace") {
const isResumeReplacement = resumeTailPolicy.kind === "replace";
timelineResult = applyTimelineReplacePath({
timelineUnits,
payload,
bootstrapPolicy: isResumeReplacement
? { replace: true, catchUpCursor: null }
: bootstrapPolicy,
currentTail,
currentHead,
sendingClientMessageIds,
preserveContinuity: isResumeReplacement
? resumeTailPolicy.preserveContinuity
: currentCursor?.epoch === payload.epoch || !payload.reset,
toHydratedEvents,
});
} else {
const incrementalUnits =
resumeTailPolicy.kind === "append" && currentCursor
? timelineUnits.filter((unit) => unit.seqEnd > currentCursor.endSeq)
: timelineUnits;
timelineResult = applyTimelineIncrementalPath({
timelineUnits: incrementalUnits,
payload,
currentTail,
currentHead,
currentCursor,
});
}
const nextTail = timelineResult.tail;
const nextHead = timelineResult.head;
@@ -934,10 +1010,12 @@ export function processTimelineResponse(
const initResolution: "resolve" | "reject" | null = shouldResolveDeferredInit ? "resolve" : null;
return {
commit: discard ? "discard" : "apply",
tail: nextTail,
head: nextHead,
cursor: nextCursor,
cursorChanged,
older: timelineResult.older,
initResolution,
clearInitializing,
error: null,
packages/app/src/timeline/timeline-sync-plan.test.ts
+3 -44
View File
@@ -2,54 +2,13 @@ import { describe, expect, test } from "vitest";
import { TIMELINE_FETCH_PAGE_SIZE } from "@/timeline/timeline-fetch-policy";
import {
isTimelineCatchUpComplete,
planInitialAgentTimelineSync,
planResumeTimelineSync,
planTimelineOlderFetch,
planTimelineTailFetch,
} from "./timeline-sync-plan";
describe("timeline sync planning", () => {
test("initial open without an authoritative cursor loads a bounded tail page", () => {
const plan = planInitialAgentTimelineSync({
cursor: undefined,
hasAuthoritativeHistory: false,
});
expect(plan).toEqual({
direction: "tail",
limit: TIMELINE_FETCH_PAGE_SIZE,
projection: "projected",
});
});
test("initial open with an authoritative cursor catches up after the cursor", () => {
const plan = planInitialAgentTimelineSync({
cursor: { epoch: "epoch-1", startSeq: 1, endSeq: 42 },
hasAuthoritativeHistory: true,
});
expect(plan).toEqual({
direction: "after",
cursor: { epoch: "epoch-1", seq: 42 },
limit: TIMELINE_FETCH_PAGE_SIZE,
projection: "projected",
});
});
test("resume with a cursor catches up after the cursor", () => {
const plan = planResumeTimelineSync({
cursor: { epoch: "epoch-1", startSeq: 1, endSeq: 100 },
});
expect(plan).toEqual({
direction: "after",
cursor: { epoch: "epoch-1", seq: 100 },
limit: TIMELINE_FETCH_PAGE_SIZE,
projection: "projected",
});
});
test("resume without a cursor loads a bounded tail page", () => {
const plan = planResumeTimelineSync({ cursor: undefined });
test("agent synchronization checks a bounded tail page", () => {
const plan = planTimelineTailFetch();
expect(plan).toEqual({
direction: "tail",
packages/app/src/timeline/timeline-sync-plan.ts
-27
View File
@@ -5,12 +5,6 @@ export interface TimelineSyncCursor {
seq: number;
}
export interface AgentTimelineCursorRange {
epoch: string;
startSeq: number;
endSeq: number;
}
export interface ProjectedTimelineTailFetchPlan {
direction: "tail";
limit: number;
@@ -40,27 +34,6 @@ export type ProjectedTimelineForwardFetchPlan =
| ProjectedTimelineTailFetchPlan
| ProjectedTimelineAfterFetchPlan;
export function planInitialAgentTimelineSync(input: {
cursor: AgentTimelineCursorRange | undefined;
hasAuthoritativeHistory: boolean;
}): ProjectedTimelineForwardFetchPlan {
if (input.hasAuthoritativeHistory && input.cursor) {
return planTimelineCatchUpAfter({ epoch: input.cursor.epoch, seq: input.cursor.endSeq });
}
return planTimelineTailFetch();
}
export function planResumeTimelineSync(input: {
cursor: AgentTimelineCursorRange | undefined;
}): ProjectedTimelineForwardFetchPlan {
if (input.cursor) {
return planTimelineCatchUpAfter({ epoch: input.cursor.epoch, seq: input.cursor.endSeq });
}
return planTimelineTailFetch();
}
export function planTimelineCatchUpAfter(cursor: TimelineSyncCursor) {
return {
direction: "after",
packages/app/src/timeline/viewed-timeline-sync.test.ts
+28 -15
View File
@@ -48,8 +48,6 @@ class TimelineWorld {
this.releaseMembershipWaiter();
return result.promise;
},
readCursor: (agentId) => this.cursors.get(agentId),
hasAuthoritativeHistory: (agentId) => this.authoritativeHistory.has(agentId),
fetchPage: async (agentId, request) => {
const result = deferred<{
hasNewer: boolean;
@@ -95,21 +93,10 @@ class TimelineWorld {
agentId: string;
resolve(fetch: TimelineFetch): void;
}> = [];
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 scheduled: Array<{ task: () => void; delayMs: number }> = [];
private readonly retryWaiters: Array<(retry: () => void) => void> = [];
setCursor(agentId: string, endSeq: number): void {
this.cursors.set(agentId, { epoch: `epoch-${agentId}`, startSeq: 1, endSeq });
this.authoritativeHistory.add(agentId);
}
setLiveCursor(agentId: string, endSeq: number): void {
this.cursors.set(agentId, { epoch: `epoch-${agentId}`, startSeq: 1, endSeq });
}
nextMembership(): Promise<MembershipRequest> {
const request = this.memberships.shift();
if (request) return Promise.resolve(request);
@@ -178,9 +165,8 @@ test("keeps hidden agents subscribed for thirty seconds", () => {
expect(VIEWED_TIMELINE_UNSUBSCRIBE_GRACE_MS).toBe(30_000);
});
test("uses a tail fetch when a live cursor is not authoritative", async () => {
test("uses a tail fetch when an agent becomes visible", async () => {
const world = new TimelineWorld();
world.setLiveCursor("agent-a", 9);
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const membership = await world.nextMembership();
@@ -191,6 +177,30 @@ test("uses a tail fetch when a live cursor is not authoritative", async () => {
fetch.respond({ hasNewer: false });
});
test("a gap absorbed by a running tail is recovered after the tail completes", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const membership = await world.nextMembership();
membership.succeed();
const tail = await world.nextFetch("agent-a");
world.sync.recoverGap("agent-a", { epoch: "epoch-agent-a", endSeq: 9 });
world.expectNoPendingFetch();
tail.respond({ hasNewer: false });
const recovery = await world.nextFetch("agent-a");
expect(recovery.request).toEqual({
direction: "after",
cursor: { epoch: "epoch-agent-a", seq: 9 },
limit: 40,
projection: "projected",
});
recovery.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
});
test("unchanged visible-set publication does not cancel paged catch-up", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
@@ -364,6 +374,7 @@ test("gap recovery supersedes completed catch-up and pages through the current t
membership.succeed();
const initial = await world.nextFetch("agent-a");
initial.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
world.sync.recoverGap("agent-a", { epoch: "epoch-agent-a", endSeq: 10 });
const gapPage = await world.nextFetch("agent-a");
@@ -395,6 +406,7 @@ test("repeated recovery for the same running gap reuses the in-flight fetch", as
membership.succeed();
const initial = await world.nextFetch("agent-a");
initial.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
const cursor = { epoch: "epoch-agent-a", endSeq: 10 };
world.sync.recoverGap("agent-a", cursor);
@@ -656,6 +668,7 @@ test("legacy delivery skips subscription RPCs while retaining visibility catch-u
world.expectNoPendingMembership();
const initial = await world.nextFetch("agent-a");
initial.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
world.sync.recoverGap("agent-a", { epoch: "epoch-agent-a", endSeq: 10 });
const recovery = await world.nextFetch("agent-a");
packages/app/src/timeline/viewed-timeline-sync.ts
+33 -20
View File
@@ -1,8 +1,6 @@
import type { AgentTimelineCursorState } from "@/stores/session-store";
import {
planInitialAgentTimelineSync,
planResumeTimelineSync,
planTimelineCatchUpAfter,
planTimelineTailFetch,
type ProjectedTimelineForwardFetchPlan,
} from "./timeline-sync-plan";
@@ -14,8 +12,6 @@ interface TimelinePageResult {
interface ViewedTimelineSyncPorts {
initialDeliveryMode: TimelineDeliveryMode;
setSubscription(agentIds: string[]): Promise<void>;
readCursor(agentId: string): AgentTimelineCursorState | undefined;
hasAuthoritativeHistory(agentId: string): boolean;
fetchPage(
agentId: string,
request: ProjectedTimelineForwardFetchPlan,
@@ -62,19 +58,29 @@ function isSameCatchUpRequest(
return left.cursor.epoch === right.cursor.epoch && left.cursor.seq === right.cursor.seq;
}
function shouldKeepCurrentCatchUp(input: {
type CatchUpDecision = "keep" | "keep-and-park" | "replace";
function decideCatchUp(input: {
current: CatchUpState | undefined;
request: ProjectedTimelineForwardFetchPlan | undefined;
request: ProjectedTimelineForwardFetchPlan;
supersede: boolean;
}): boolean {
if (!input.current) return false;
}): CatchUpDecision {
if (!input.current) return "replace";
if (input.supersede) {
return (
if (
input.current.status === "running" &&
isSameCatchUpRequest(input.current.request, input.request)
);
) {
return "keep";
}
if (input.current.status === "running" && input.current.request?.direction === "tail") {
return "keep-and-park";
}
return "replace";
}
return input.current.status === "running" || input.current.status === "complete";
return input.current.status === "running" || input.current.status === "complete"
? "keep"
: "replace";
}
function normalizeAgentIds(agentIds: string[]): string[] {
@@ -89,6 +95,8 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
const sources = new Map<string, string[]>();
const catchUps = new Map<string, CatchUpState>();
const catchUpGenerations = new Map<string, number>();
// Recovery owed but not runnable yet: disconnected, unacknowledged, or parked behind a tail.
// Acknowledgement and tail completion are the only drain points.
const pendingGaps = new Map<string, ProjectedTimelineForwardFetchPlan>();
const lingeringRemovals = new Map<string, () => void>();
const visibilityCatchUpPending = new Set<string>();
@@ -174,6 +182,11 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
throw new Error(`Timeline page for ${agentId} hasNewer without an end cursor`);
}
catchUps.set(agentId, { generation, status: "complete" });
const pendingGap = pendingGaps.get(agentId);
if (pendingGap) {
startCatchUp(agentId, { request: pendingGap, supersede: true });
return;
}
setVisibilityCatchUpReady(agentId);
} catch (error) {
if (catchUps.get(agentId)?.generation === generation) {
@@ -201,21 +214,21 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
if (request) pendingGaps.set(agentId, request);
return;
}
const nextRequest = request ?? planTimelineTailFetch();
const current = catchUps.get(agentId);
if (shouldKeepCurrentCatchUp({ current, request, supersede })) {
const decision = decideCatchUp({ current, request: nextRequest, supersede });
if (decision === "keep-and-park") {
pendingGaps.set(agentId, nextRequest);
return;
}
if (decision === "keep") {
return;
}
current?.cancelRetry?.();
const generation = (catchUpGenerations.get(agentId) ?? 0) + 1;
catchUpGenerations.set(agentId, generation);
catchUps.set(agentId, { generation, status: "running", request });
catchUps.set(agentId, { generation, status: "running", request: nextRequest });
pendingGaps.delete(agentId);
const cursor = ports.readCursor(agentId);
const nextRequest =
request ??
(ports.hasAuthoritativeHistory(agentId)
? planResumeTimelineSync({ cursor })
: planInitialAgentTimelineSync({ cursor, hasAuthoritativeHistory: false }));
void fetchUntilCurrent(agentId, generation, nextRequest);
};