Resume cached timelines without replaying their tail (#3329)
* perf(timeline): resume cached history from its exact range Warm launches were repainting cached history and then downloading the same bounded tail to rediscover that nothing changed. Persist synchronization authority only when the complete current canonical window can be encoded losslessly; otherwise retain the existing display-only fallback. * test(timeline): hide replica setup behind E2E helper * fix(timeline): bound persisted resume overflow * test(timeline): cover workspace eviction catch-up order * test(timeline): read persisted range from IndexedDB
This commit is contained in:
+15
-9
@@ -85,16 +85,22 @@ recomposition while the runtime still owns the same directory snapshot and timel
|
|||||||
Removing the host from the registry is the destructive boundary: it stops the runtime and clears the
|
Removing the host from the registry is the destructive boundary: it stops the runtime and clears the
|
||||||
session and host-scoped setup state together.
|
session and host-scoped setup state together.
|
||||||
|
|
||||||
The durable replica cache is a display cache, not a synchronization checkpoint. Its timeline record
|
The durable replica cache persists synchronization authority only when it can store the complete
|
||||||
contains only the focused `agentId` and a truncated item tail. It never persists a cursor, epoch,
|
current canonical window losslessly. The stored range describes those exact items: `startSeq` drives
|
||||||
older-history availability, authority status, or sync generation because those facts would describe
|
older pagination and `endSeq` drives forward catch-up. Restore paints the items immediately, requests
|
||||||
the complete source dataset rather than the truncated display dataset.
|
`after endSeq`, and requests `before startSeq` when the user loads older history.
|
||||||
|
|
||||||
Restoring that cache produces a painted timeline: the items may render immediately, but the first
|
The first resume request is bounded. If it reports more newer history, fetch one latest bounded tail
|
||||||
daemon timeline request is still `tail`. A successful tail response atomically establishes canonical
|
instead of replaying every missed page. Live gap recovery still pages forward until current.
|
||||||
items, range, and older-history availability. Live rows received between cache paint and that tail
|
|
||||||
response stay in the separate live head, do not advance a cursor or trigger gap recovery, and are
|
If the canonical window exceeds the cache item limit, contains a discontiguous retained range, has a
|
||||||
reconciled with the authoritative tail and subsequent catch-up.
|
live head, or includes presentation data the cache cannot encode losslessly, persistence drops the
|
||||||
|
range and keeps a display-only tail. Restore then uses the ordinary bounded `tail` bootstrap. Never
|
||||||
|
slice items while retaining the pre-slice range; that falsely certifies discarded source rows.
|
||||||
|
|
||||||
|
Live rows received between cache paint and catch-up stay in the separate live head and reconcile with
|
||||||
|
the authoritative range through the existing forward-page path. The cache does not persist sync
|
||||||
|
generation or unreconciled local submissions.
|
||||||
|
|
||||||
Every daemon-derived live item carries its timeline epoch and sequence position. Bootstrap
|
Every daemon-derived live item carries its timeline epoch and sequence position. Bootstrap
|
||||||
replacement keeps only positioned rows newer than the page it installs, while unresolved local
|
replacement keeps only positioned rows newer than the page it installs, while unresolved local
|
||||||
|
|||||||
@@ -30,6 +30,12 @@ import { buildHostWorkspaceRoute } from "@/utils/host-routes";
|
|||||||
import { delayBrowserAgentCreatedStatus } from "../support/helpers/new-workspace";
|
import { delayBrowserAgentCreatedStatus } from "../support/helpers/new-workspace";
|
||||||
import { installDaemonWebSocketGate } from "../support/helpers/daemon-websocket-gate";
|
import { installDaemonWebSocketGate } from "../support/helpers/daemon-websocket-gate";
|
||||||
import { selectModel } from "../support/helpers/app";
|
import { selectModel } from "../support/helpers/app";
|
||||||
|
import { observeTimelineSubscriptions } from "../support/helpers/timeline-delivery";
|
||||||
|
import {
|
||||||
|
expectResumeOverflowFallsBackToOneTail,
|
||||||
|
rememberTimelineRequestCounts,
|
||||||
|
} from "../support/helpers/timeline-resume";
|
||||||
|
import { workspaceDeckEntryLocator } from "../support/helpers/workspace-ui";
|
||||||
import {
|
import {
|
||||||
scrollTimelineToNewestLoadedEdge,
|
scrollTimelineToNewestLoadedEdge,
|
||||||
scrollTimelineUntilOlderHistoryIsReachable,
|
scrollTimelineUntilOlderHistoryIsReachable,
|
||||||
@@ -336,6 +342,64 @@ async function expectInterruptedTurnOrderAfterReconnect(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function expectHiddenStreamingSubmissionOrderAfterWorkspaceEviction(
|
||||||
|
page: Page,
|
||||||
|
testInfo: { workerIndex: number },
|
||||||
|
): Promise<void> {
|
||||||
|
const subscriptions = observeTimelineSubscriptions(page);
|
||||||
|
const gate = await installDaemonWebSocketGate(page);
|
||||||
|
const target = await seedMockAgentWorkspace({
|
||||||
|
repoPrefix: `submission-hidden-stream-${testInfo.workerIndex}-`,
|
||||||
|
title: "Hidden streaming submission",
|
||||||
|
model: "ten-second-stream",
|
||||||
|
});
|
||||||
|
const evictionAgents = await Promise.all(
|
||||||
|
Array.from({ length: 3 }, (_unused, index) =>
|
||||||
|
seedMockAgentWorkspace({
|
||||||
|
repoPrefix: `submission-workspace-eviction-${testInfo.workerIndex}-${index}-`,
|
||||||
|
title: `Workspace eviction ${index + 1}`,
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
);
|
||||||
|
const prompt = "Keep this hidden image prompt before its streaming output.";
|
||||||
|
const targetDeckEntry = workspaceDeckEntryLocator(page, getServerId(), target.workspaceId);
|
||||||
|
|
||||||
|
try {
|
||||||
|
await openAgentRoute(page, target);
|
||||||
|
await expectComposerVisible(page);
|
||||||
|
await subscriptions.waitForSubscribedAgents([target.agentId]);
|
||||||
|
|
||||||
|
const userMessageCount = gate.getAgentStreamItemCount("user_message");
|
||||||
|
gate.setAgentStreamSuppressed(true);
|
||||||
|
const promptRow = await submitMessageWithImage(page, prompt);
|
||||||
|
await gate.waitForAgentStreamItem("user_message", userMessageCount + 1);
|
||||||
|
|
||||||
|
for (const evictionAgent of evictionAgents) {
|
||||||
|
await openAgentRoute(page, evictionAgent);
|
||||||
|
await expectComposerVisible(page);
|
||||||
|
}
|
||||||
|
await expect(targetDeckEntry).toHaveCount(0);
|
||||||
|
await subscriptions.waitForSubscribedAgents([evictionAgents[2]!.agentId]);
|
||||||
|
gate.setAgentStreamSuppressed(false);
|
||||||
|
|
||||||
|
await target.client.waitForFinish(target.agentId, 30_000);
|
||||||
|
const requestsBeforeReturn = rememberTimelineRequestCounts(gate);
|
||||||
|
await openAgentRoute(page, target);
|
||||||
|
await expectComposerVisible(page);
|
||||||
|
await subscriptions.waitForSubscribedAgents([target.agentId]);
|
||||||
|
|
||||||
|
const response = page.getByText("(end of synthetic stream)", { exact: true }).last();
|
||||||
|
await expect(promptRow).toBeVisible();
|
||||||
|
await expect(response).toBeVisible();
|
||||||
|
await expectRenderedBefore(promptRow, response);
|
||||||
|
expectResumeOverflowFallsBackToOneTail(gate, requestsBeforeReturn);
|
||||||
|
} finally {
|
||||||
|
gate.setAgentStreamSuppressed(false);
|
||||||
|
gate.restore();
|
||||||
|
await Promise.all([...evictionAgents.map((agent) => agent.cleanup()), target.cleanup()]);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async function expectCompletedSubmissionClearsAfterMissedRunningTransition(
|
async function expectCompletedSubmissionClearsAfterMissedRunningTransition(
|
||||||
page: Page,
|
page: Page,
|
||||||
testInfo: { workerIndex: number },
|
testInfo: { workerIndex: number },
|
||||||
@@ -1013,6 +1077,13 @@ test.describe("Agent message submission", () => {
|
|||||||
await expectInterruptedTurnOrderAfterReconnect(page, testInfo);
|
await expectInterruptedTurnOrderAfterReconnect(page, testInfo);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("keeps a streaming hidden submission before its output after workspace eviction", async ({
|
||||||
|
page,
|
||||||
|
}, testInfo) => {
|
||||||
|
test.setTimeout(120_000);
|
||||||
|
await expectHiddenStreamingSubmissionOrderAfterWorkspaceEviction(page, testInfo);
|
||||||
|
});
|
||||||
|
|
||||||
test("clears an attachment-only submission when canonical history arrives after a missed running transition", async ({
|
test("clears an attachment-only submission when canonical history arrives after a missed running transition", async ({
|
||||||
page,
|
page,
|
||||||
}, testInfo) => {
|
}, testInfo) => {
|
||||||
|
|||||||
@@ -26,9 +26,41 @@ import {
|
|||||||
expectTimelineViewportAnchoredAfterPrepend,
|
expectTimelineViewportAnchoredAfterPrepend,
|
||||||
userNavigatesTimelineToHistoryStartWithKeyboard,
|
userNavigatesTimelineToHistoryStartWithKeyboard,
|
||||||
userScrollsTimelineToHistoryStart,
|
userScrollsTimelineToHistoryStart,
|
||||||
|
waitForPersistedCanonicalTimelineRange,
|
||||||
} from "../support/helpers/timeline-pagination";
|
} from "../support/helpers/timeline-pagination";
|
||||||
|
import { trackAgentTimelineRequests } from "../support/helpers/agent-timeline-gate";
|
||||||
|
import { seedMockAgentWorkspace } from "../support/helpers/mock-agent";
|
||||||
|
|
||||||
test.describe("Agent timeline pagination", () => {
|
test.describe("Agent timeline pagination", () => {
|
||||||
|
test("resumes a persisted canonical timeline after its exact end", async ({ page }) => {
|
||||||
|
const prompt = "timeline persisted range resumes without tail replay";
|
||||||
|
const agent = await seedMockAgentWorkspace({
|
||||||
|
repoPrefix: "timeline-persisted-range-",
|
||||||
|
title: "Persisted timeline range",
|
||||||
|
initialPrompt: prompt,
|
||||||
|
});
|
||||||
|
try {
|
||||||
|
await agent.client.waitForFinish(agent.agentId, 15_000);
|
||||||
|
await openAgentTimeline(page, agent);
|
||||||
|
await expectTimelinePromptVisible(page, prompt);
|
||||||
|
const range = await waitForPersistedCanonicalTimelineRange(page, agent.agentId);
|
||||||
|
|
||||||
|
const tracker = await trackAgentTimelineRequests(page, agent.agentId);
|
||||||
|
await page.reload();
|
||||||
|
await expectTimelinePromptVisible(page, prompt);
|
||||||
|
|
||||||
|
const request = await tracker.nextRequest();
|
||||||
|
expect(request).toEqual({
|
||||||
|
direction: "after",
|
||||||
|
cursor: { epoch: range.epoch, seq: range.endSeq },
|
||||||
|
});
|
||||||
|
await tracker.waitForResponse();
|
||||||
|
expect(tracker.requests().some((entry) => entry.direction === "tail")).toBe(false);
|
||||||
|
} finally {
|
||||||
|
await agent.cleanup();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
test("keeps a fixed history-start gutter before, during, and after pagination", async ({
|
test("keeps a fixed history-start gutter before, during, and after pagination", async ({
|
||||||
page,
|
page,
|
||||||
}) => {
|
}) => {
|
||||||
|
|||||||
@@ -13,9 +13,10 @@ import {
|
|||||||
import {
|
import {
|
||||||
commitTimelineTurnsWhileDisconnected,
|
commitTimelineTurnsWhileDisconnected,
|
||||||
disconnectViewedTimeline,
|
disconnectViewedTimeline,
|
||||||
expectOneTailCheckWithoutCatchUpPages,
|
expectOneResumeCheckWithoutTail,
|
||||||
|
expectResumeOverflowFallsBackToOneTail,
|
||||||
rememberTimelineRequestCounts,
|
rememberTimelineRequestCounts,
|
||||||
restoreViewedTimelineWithHeldTail,
|
restoreViewedTimelineWithHeldResponse,
|
||||||
} from "../support/helpers/timeline-resume";
|
} from "../support/helpers/timeline-resume";
|
||||||
|
|
||||||
test.describe("Agent timeline resume", () => {
|
test.describe("Agent timeline resume", () => {
|
||||||
@@ -29,11 +30,11 @@ test.describe("Agent timeline resume", () => {
|
|||||||
const requests = rememberTimelineRequestCounts(gate);
|
const requests = rememberTimelineRequestCounts(gate);
|
||||||
await disconnectViewedTimeline(page, gate);
|
await disconnectViewedTimeline(page, gate);
|
||||||
const background = await commitTimelineTurnsWhileDisconnected(agent, 24);
|
const background = await commitTimelineTurnsWhileDisconnected(agent, 24);
|
||||||
await restoreViewedTimelineWithHeldTail(page, gate);
|
await restoreViewedTimelineWithHeldResponse(page, gate);
|
||||||
|
|
||||||
await expectTimelinePromptVisible(page, background.newestPrompt);
|
await expectTimelinePromptVisible(page, background.newestPrompt);
|
||||||
await expectTimelinePromptNotMounted(page, background.firstPrompt);
|
await expectTimelinePromptNotMounted(page, background.firstPrompt);
|
||||||
expectOneTailCheckWithoutCatchUpPages(gate, requests);
|
expectResumeOverflowFallsBackToOneTail(gate, requests);
|
||||||
await scrollTimelineUntilOlderHistoryIsReachable(page, background.firstPrompt);
|
await scrollTimelineUntilOlderHistoryIsReachable(page, background.firstPrompt);
|
||||||
} finally {
|
} finally {
|
||||||
gate.restore();
|
gate.restore();
|
||||||
@@ -53,10 +54,10 @@ test.describe("Agent timeline resume", () => {
|
|||||||
const requests = rememberTimelineRequestCounts(gate);
|
const requests = rememberTimelineRequestCounts(gate);
|
||||||
|
|
||||||
await disconnectViewedTimeline(page, gate);
|
await disconnectViewedTimeline(page, gate);
|
||||||
await restoreViewedTimelineWithHeldTail(page, gate);
|
await restoreViewedTimelineWithHeldResponse(page, gate);
|
||||||
|
|
||||||
await expectTimelinePresentationUnchanged(page, presentation);
|
await expectTimelinePresentationUnchanged(page, presentation);
|
||||||
expectOneTailCheckWithoutCatchUpPages(gate, requests);
|
expectOneResumeCheckWithoutTail(gate, requests);
|
||||||
} finally {
|
} finally {
|
||||||
gate.restore();
|
gate.restore();
|
||||||
await agent.cleanup();
|
await agent.cleanup();
|
||||||
|
|||||||
@@ -31,6 +31,82 @@ export interface PromptJumpRequestTracker {
|
|||||||
requests(): Array<{ cursorSeq: number | null; limit: number | null; mergeWindow: boolean }>;
|
requests(): Array<{ cursorSeq: number | null; limit: number | null; mergeWindow: boolean }>;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export interface TimelineRequestTracker {
|
||||||
|
nextRequest(): Promise<{
|
||||||
|
direction: string | null;
|
||||||
|
cursor: { epoch: string; seq: number } | null;
|
||||||
|
}>;
|
||||||
|
requests(): Array<{
|
||||||
|
direction: string | null;
|
||||||
|
cursor: { epoch: string; seq: number } | null;
|
||||||
|
}>;
|
||||||
|
waitForResponse(): Promise<void>;
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function trackAgentTimelineRequests(
|
||||||
|
page: Page,
|
||||||
|
agentId: string,
|
||||||
|
): Promise<TimelineRequestTracker> {
|
||||||
|
type Request = ReturnType<TimelineRequestTracker["requests"]>[number];
|
||||||
|
const seen: Request[] = [];
|
||||||
|
const waiters: Array<(request: Request) => void> = [];
|
||||||
|
let responseSeen = false;
|
||||||
|
let resolveResponse: (() => void) | null = null;
|
||||||
|
const response = new Promise<void>((resolve) => {
|
||||||
|
resolveResponse = resolve;
|
||||||
|
});
|
||||||
|
await page.routeWebSocket(daemonWsRoutePattern(), (ws) => {
|
||||||
|
const server = ws.connectToServer();
|
||||||
|
ws.onMessage((message) => {
|
||||||
|
const sessionMessage = getSessionMessage(message);
|
||||||
|
if (
|
||||||
|
sessionMessage?.type === "fetch_agent_timeline_request" &&
|
||||||
|
sessionMessage.agentId === agentId
|
||||||
|
) {
|
||||||
|
const rawCursor = sessionMessage.cursor;
|
||||||
|
const cursor =
|
||||||
|
rawCursor &&
|
||||||
|
typeof rawCursor === "object" &&
|
||||||
|
typeof (rawCursor as { epoch?: unknown }).epoch === "string" &&
|
||||||
|
typeof (rawCursor as { seq?: unknown }).seq === "number"
|
||||||
|
? {
|
||||||
|
epoch: (rawCursor as { epoch: string }).epoch,
|
||||||
|
seq: (rawCursor as { seq: number }).seq,
|
||||||
|
}
|
||||||
|
: null;
|
||||||
|
const request = {
|
||||||
|
direction: typeof sessionMessage.direction === "string" ? sessionMessage.direction : null,
|
||||||
|
cursor,
|
||||||
|
};
|
||||||
|
seen.push(request);
|
||||||
|
waiters.shift()?.(request);
|
||||||
|
}
|
||||||
|
server.send(message);
|
||||||
|
});
|
||||||
|
server.onMessage((message) => {
|
||||||
|
const sessionMessage = getSessionMessage(message);
|
||||||
|
const payload = sessionMessage ? getPayload(sessionMessage) : null;
|
||||||
|
if (
|
||||||
|
sessionMessage?.type === "fetch_agent_timeline_response" &&
|
||||||
|
payload?.agentId === agentId
|
||||||
|
) {
|
||||||
|
responseSeen = true;
|
||||||
|
resolveResponse?.();
|
||||||
|
}
|
||||||
|
ws.send(message);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
return {
|
||||||
|
nextRequest() {
|
||||||
|
const request = seen[0];
|
||||||
|
if (request) return Promise.resolve(request);
|
||||||
|
return new Promise((resolve) => waiters.push(resolve));
|
||||||
|
},
|
||||||
|
requests: () => [...seen],
|
||||||
|
waitForResponse: () => (responseSeen ? Promise.resolve() : response),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
export async function trackPromptJumpRequests(
|
export async function trackPromptJumpRequests(
|
||||||
page: Page,
|
page: Page,
|
||||||
agentId: string,
|
agentId: string,
|
||||||
|
|||||||
@@ -5,10 +5,16 @@ const STORE_NAME = "key-value";
|
|||||||
const STORAGE_KEY = "@paseo:replica-cache";
|
const STORAGE_KEY = "@paseo:replica-cache";
|
||||||
|
|
||||||
export interface ReplicaCacheRecord {
|
export interface ReplicaCacheRecord {
|
||||||
|
version?: number;
|
||||||
hosts?: Array<{
|
hosts?: Array<{
|
||||||
timeline?: {
|
timeline?: {
|
||||||
agentId?: string;
|
agentId?: string;
|
||||||
items?: Array<Record<string, unknown>>;
|
items?: Array<Record<string, unknown>>;
|
||||||
|
range?: {
|
||||||
|
endSeq?: number;
|
||||||
|
epoch?: string;
|
||||||
|
startSeq?: number;
|
||||||
|
} | null;
|
||||||
} | null;
|
} | null;
|
||||||
}>;
|
}>;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -40,6 +40,12 @@ interface TimelineViewportSnapshot {
|
|||||||
scrollTop: number;
|
scrollTop: number;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
interface PersistedCanonicalTimelineRange {
|
||||||
|
epoch: string;
|
||||||
|
startSeq: number;
|
||||||
|
endSeq: number;
|
||||||
|
}
|
||||||
|
|
||||||
export interface TimelinePresentationSnapshot {
|
export interface TimelinePresentationSnapshot {
|
||||||
marker: string;
|
marker: string;
|
||||||
position: TimelinePromptPositionSnapshot;
|
position: TimelinePromptPositionSnapshot;
|
||||||
@@ -298,6 +304,32 @@ export async function reloadAgentTimelineFromPersistedReplica(
|
|||||||
await expectTimelinePromptVisible(page, agent.newestPrompt);
|
await expectTimelinePromptVisible(page, agent.newestPrompt);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export async function waitForPersistedCanonicalTimelineRange(
|
||||||
|
page: Page,
|
||||||
|
agentId: string,
|
||||||
|
): Promise<PersistedCanonicalTimelineRange> {
|
||||||
|
const readRange = async () => {
|
||||||
|
const cache = await readReplicaCache(page);
|
||||||
|
if (cache?.version !== 6) return null;
|
||||||
|
const range = cache.hosts?.find((host) => host.timeline?.agentId === agentId)?.timeline?.range;
|
||||||
|
if (
|
||||||
|
typeof range?.epoch !== "string" ||
|
||||||
|
typeof range.startSeq !== "number" ||
|
||||||
|
typeof range.endSeq !== "number"
|
||||||
|
) {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
return { epoch: range.epoch, startSeq: range.startSeq, endSeq: range.endSeq };
|
||||||
|
};
|
||||||
|
|
||||||
|
await expect.poll(readRange).not.toBeNull();
|
||||||
|
const range = await readRange();
|
||||||
|
if (!range) {
|
||||||
|
throw new Error(`Persisted canonical timeline range is missing for ${agentId}`);
|
||||||
|
}
|
||||||
|
return range;
|
||||||
|
}
|
||||||
|
|
||||||
export async function holdNextOlderTimelinePage(
|
export async function holdNextOlderTimelinePage(
|
||||||
page: Page,
|
page: Page,
|
||||||
agent: LongTimelineAgent,
|
agent: LongTimelineAgent,
|
||||||
|
|||||||
@@ -40,12 +40,20 @@ export function rememberTimelineRequestCounts(gate: DaemonWebSocketGate): Timeli
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
export function expectOneTailCheckWithoutCatchUpPages(
|
export function expectOneResumeCheckWithoutTail(
|
||||||
gate: DaemonWebSocketGate,
|
gate: DaemonWebSocketGate,
|
||||||
before: TimelineRequestCounts,
|
before: TimelineRequestCounts,
|
||||||
): void {
|
): void {
|
||||||
|
expect(gate.getTimelineRequestCount("after") - before.after).toBe(1);
|
||||||
|
expect(gate.getTimelineRequestCount("tail") - before.tail).toBe(0);
|
||||||
|
}
|
||||||
|
|
||||||
|
export function expectResumeOverflowFallsBackToOneTail(
|
||||||
|
gate: DaemonWebSocketGate,
|
||||||
|
before: TimelineRequestCounts,
|
||||||
|
): void {
|
||||||
|
expect(gate.getTimelineRequestCount("after") - before.after).toBe(1);
|
||||||
expect(gate.getTimelineRequestCount("tail") - before.tail).toBe(1);
|
expect(gate.getTimelineRequestCount("tail") - before.tail).toBe(1);
|
||||||
expect(gate.getTimelineRequestCount("after") - before.after).toBe(0);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function disconnectViewedTimeline(
|
export async function disconnectViewedTimeline(
|
||||||
@@ -56,7 +64,7 @@ export async function disconnectViewedTimeline(
|
|||||||
await expectReconnectingToastVisible(page);
|
await expectReconnectingToastVisible(page);
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function restoreViewedTimelineWithHeldTail(
|
export async function restoreViewedTimelineWithHeldResponse(
|
||||||
page: Page,
|
page: Page,
|
||||||
gate: DaemonWebSocketGate,
|
gate: DaemonWebSocketGate,
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
|
|||||||
@@ -19,7 +19,10 @@ import {
|
|||||||
type TimelineReducerSideEffect,
|
type TimelineReducerSideEffect,
|
||||||
} from "@/timeline/session-stream-reducers";
|
} from "@/timeline/session-stream-reducers";
|
||||||
import { useCreateFlowStore } from "@/stores/create-flow-store";
|
import { useCreateFlowStore } from "@/stores/create-flow-store";
|
||||||
import { isTimelineResumeSnapshotAuthoritative } from "@/timeline/timeline-sync-plan";
|
import {
|
||||||
|
isTimelineResumeSnapshotAuthoritative,
|
||||||
|
planTimelineTailFetch,
|
||||||
|
} from "@/timeline/timeline-sync-plan";
|
||||||
import {
|
import {
|
||||||
createViewedTimelineSync,
|
createViewedTimelineSync,
|
||||||
type TimelineDeliveryMode,
|
type TimelineDeliveryMode,
|
||||||
@@ -33,6 +36,7 @@ import {
|
|||||||
type AgentAttentionNotificationPayload,
|
type AgentAttentionNotificationPayload,
|
||||||
type NotificationPermissionRequest,
|
type NotificationPermissionRequest,
|
||||||
} from "@getpaseo/protocol/agent-attention-notification";
|
} from "@getpaseo/protocol/agent-attention-notification";
|
||||||
|
|
||||||
import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
|
import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
|
||||||
import type { AgentSessionConfig } from "@getpaseo/protocol/agent-types";
|
import type { AgentSessionConfig } from "@getpaseo/protocol/agent-types";
|
||||||
import type { GitSetupOptions } from "@getpaseo/protocol/messages";
|
import type { GitSetupOptions } from "@getpaseo/protocol/messages";
|
||||||
@@ -67,6 +71,20 @@ import { applyCheckoutStatusUpdateFromEvent } from "@/git/checkout-status-cache"
|
|||||||
import { useProviderSubagentStore } from "@/subagents/provider-store";
|
import { useProviderSubagentStore } from "@/subagents/provider-store";
|
||||||
import { revalidateSessionAfterResume } from "@/contexts/session-resume-revalidation";
|
import { revalidateSessionAfterResume } from "@/contexts/session-resume-revalidation";
|
||||||
|
|
||||||
|
type TimelineResponsePayload = Extract<
|
||||||
|
SessionOutboundMessage,
|
||||||
|
{ type: "fetch_agent_timeline_response" }
|
||||||
|
>["payload"];
|
||||||
|
|
||||||
|
function consumeForcedTimelineTailReplacement(
|
||||||
|
payload: TimelineResponsePayload,
|
||||||
|
replacements: Set<string>,
|
||||||
|
): TimelineResponsePayload {
|
||||||
|
if (payload.direction !== "tail") return payload;
|
||||||
|
if (!replacements.delete(payload.agentId)) return payload;
|
||||||
|
return { ...payload, reset: true };
|
||||||
|
}
|
||||||
|
|
||||||
// Re-export types from session-store and draft-store for backward compatibility
|
// Re-export types from session-store and draft-store for backward compatibility
|
||||||
export type { DraftInput } from "@/stores/draft-store";
|
export type { DraftInput } from "@/stores/draft-store";
|
||||||
export type {
|
export type {
|
||||||
@@ -350,6 +368,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
|||||||
const _sessionStateTimeoutRef = useRef<ReturnType<typeof setTimeout> | null>(null);
|
const _sessionStateTimeoutRef = useRef<ReturnType<typeof setTimeout> | null>(null);
|
||||||
const attentionNotifiedRef = useRef<Map<string, number>>(new Map());
|
const attentionNotifiedRef = useRef<Map<string, number>>(new Map());
|
||||||
const appStateRef = useRef(AppState.currentState);
|
const appStateRef = useRef(AppState.currentState);
|
||||||
|
const forcedTimelineTailReplacements = useRef(new Set<string>());
|
||||||
const viewedTimelineSyncRef = useRef<ViewedTimelineSync | null>(null);
|
const viewedTimelineSyncRef = useRef<ViewedTimelineSync | null>(null);
|
||||||
const audioOutputBuffersRef = useRef<Map<string, BufferedAudioChunk[]>>(new Map());
|
const audioOutputBuffersRef = useRef<Map<string, BufferedAudioChunk[]>>(new Map());
|
||||||
const activeAudioGroupsRef = useRef<Set<string>>(new Set());
|
const activeAudioGroupsRef = useRef<Set<string>>(new Set());
|
||||||
@@ -528,12 +547,11 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
|||||||
);
|
);
|
||||||
|
|
||||||
const applyTimelineResponse = useCallback(
|
const applyTimelineResponse = useCallback(
|
||||||
(
|
(receivedPayload: TimelineResponsePayload) => {
|
||||||
payload: Extract<
|
const payload = consumeForcedTimelineTailReplacement(
|
||||||
SessionOutboundMessage,
|
receivedPayload,
|
||||||
{ type: "fetch_agent_timeline_response" }
|
forcedTimelineTailReplacements.current,
|
||||||
>["payload"],
|
);
|
||||||
) => {
|
|
||||||
const agentId = payload.agentId;
|
const agentId = payload.agentId;
|
||||||
const initKey = getInitKey(serverId, agentId);
|
const initKey = getInitKey(serverId, agentId);
|
||||||
const shouldMarkAuthoritativeHistoryApplied = isTimelineResumeSnapshotAuthoritative({
|
const shouldMarkAuthoritativeHistoryApplied = isTimelineResumeSnapshotAuthoritative({
|
||||||
@@ -642,6 +660,15 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
|||||||
const sync = createViewedTimelineSync({
|
const sync = createViewedTimelineSync({
|
||||||
initialDeliveryMode,
|
initialDeliveryMode,
|
||||||
setSubscription: (agentIds) => client.setAgentTimelineSubscription(agentIds),
|
setSubscription: (agentIds) => client.setAgentTimelineSubscription(agentIds),
|
||||||
|
readCursor: (agentId) => {
|
||||||
|
const timeline = selectAgentTimelineState(
|
||||||
|
useSessionStore.getState().sessions[serverId],
|
||||||
|
agentId,
|
||||||
|
);
|
||||||
|
return timeline.status === "synced" && timeline.range
|
||||||
|
? { epoch: timeline.range.epoch, endSeq: timeline.range.endSeq }
|
||||||
|
: undefined;
|
||||||
|
},
|
||||||
fetchPage: async (agentId, request) => {
|
fetchPage: async (agentId, request) => {
|
||||||
const session = useSessionStore.getState().sessions[serverId];
|
const session = useSessionStore.getState().sessions[serverId];
|
||||||
const initKey = getInitKey(serverId, agentId);
|
const initKey = getInitKey(serverId, agentId);
|
||||||
@@ -672,6 +699,18 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
|
|||||||
throw error;
|
throw error;
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
fetchLatestTail: async (agentId) => {
|
||||||
|
forcedTimelineTailReplacements.current.add(agentId);
|
||||||
|
try {
|
||||||
|
return await getHostRuntimeStore().fetchAgentTimeline(
|
||||||
|
serverId,
|
||||||
|
agentId,
|
||||||
|
planTimelineTailFetch(),
|
||||||
|
);
|
||||||
|
} finally {
|
||||||
|
forcedTimelineTailReplacements.current.delete(agentId);
|
||||||
|
}
|
||||||
|
},
|
||||||
reportError: (error) => {
|
reportError: (error) => {
|
||||||
console.warn("[Session] viewed timeline synchronization failed", { serverId, error });
|
console.warn("[Session] viewed timeline synchronization failed", { serverId, error });
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -49,7 +49,7 @@ afterEach(() => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
describe("ensureAgentIsInitialized", () => {
|
describe("ensureAgentIsInitialized", () => {
|
||||||
it("requests a bounded projected tail when authoritative history is loaded", () => {
|
it("catches up after loaded authoritative history", () => {
|
||||||
const client = new FakeDaemonClient();
|
const client = new FakeDaemonClient();
|
||||||
const runtime = new FakeTimelineRuntime();
|
const runtime = new FakeTimelineRuntime();
|
||||||
useSessionStore.getState().initializeSession(serverId, client as never);
|
useSessionStore.getState().initializeSession(serverId, client as never);
|
||||||
@@ -74,13 +74,14 @@ describe("ensureAgentIsInitialized", () => {
|
|||||||
serverId,
|
serverId,
|
||||||
agentId,
|
agentId,
|
||||||
request: {
|
request: {
|
||||||
direction: "tail",
|
direction: "after",
|
||||||
|
cursor: { epoch: "epoch-1", seq: 42 },
|
||||||
limit: TIMELINE_FETCH_PAGE_SIZE,
|
limit: TIMELINE_FETCH_PAGE_SIZE,
|
||||||
projection: "projected",
|
projection: "projected",
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
]);
|
]);
|
||||||
expect(getInitDeferred(getInitKey(serverId, agentId))?.requestDirection).toBe("tail");
|
expect(getInitDeferred(getInitKey(serverId, agentId))?.requestDirection).toBe("after");
|
||||||
});
|
});
|
||||||
|
|
||||||
it("requests a bounded projected tail when no authoritative cursor is available", () => {
|
it("requests a bounded projected tail when no authoritative cursor is available", () => {
|
||||||
@@ -127,6 +128,8 @@ describe("ensureAgentIsInitialized", () => {
|
|||||||
timestamp: new Date("2026-07-27T10:00:00.000Z"),
|
timestamp: new Date("2026-07-27T10:00:00.000Z"),
|
||||||
},
|
},
|
||||||
],
|
],
|
||||||
|
range: null,
|
||||||
|
hasOlder: false,
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -151,6 +154,52 @@ describe("ensureAgentIsInitialized", () => {
|
|||||||
]);
|
]);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("catches up after the restored canonical replica range", () => {
|
||||||
|
const client = new FakeDaemonClient();
|
||||||
|
const runtime = new FakeTimelineRuntime();
|
||||||
|
useSessionStore.getState().restoreSessionReplica(serverId, {
|
||||||
|
agents: new Map(),
|
||||||
|
workspaces: new Map(),
|
||||||
|
projects: new Map(),
|
||||||
|
timeline: {
|
||||||
|
agentId,
|
||||||
|
items: [
|
||||||
|
{
|
||||||
|
kind: "assistant_message",
|
||||||
|
id: "canonical-item",
|
||||||
|
text: "Canonical before restart",
|
||||||
|
timelineCursor: { epoch: "epoch-1", seq: 12 },
|
||||||
|
timestamp: new Date("2026-07-27T10:00:00.000Z"),
|
||||||
|
},
|
||||||
|
],
|
||||||
|
range: { epoch: "epoch-1", startSeq: 10, endSeq: 12 },
|
||||||
|
hasOlder: true,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
void ensureAgentIsInitialized({
|
||||||
|
serverId,
|
||||||
|
agentId,
|
||||||
|
client: client as never,
|
||||||
|
runtime,
|
||||||
|
setAgentInitializing: bindSetAgentInitializing(),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(runtime.requests).toEqual([
|
||||||
|
{
|
||||||
|
serverId,
|
||||||
|
agentId,
|
||||||
|
request: {
|
||||||
|
direction: "after",
|
||||||
|
cursor: { epoch: "epoch-1", seq: 12 },
|
||||||
|
limit: TIMELINE_FETCH_PAGE_SIZE,
|
||||||
|
projection: "projected",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
]);
|
||||||
|
expect(getInitDeferred(getInitKey(serverId, agentId))?.requestDirection).toBe("after");
|
||||||
|
});
|
||||||
|
|
||||||
it("times out initialization after 65 seconds", async () => {
|
it("times out initialization after 65 seconds", async () => {
|
||||||
vi.useFakeTimers();
|
vi.useFakeTimers();
|
||||||
const client = new FakeDaemonClient();
|
const client = new FakeDaemonClient();
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
import { useCallback, useMemo } from "react";
|
import { useCallback, useMemo } from "react";
|
||||||
import { useTranslation } from "react-i18next";
|
import { useTranslation } from "react-i18next";
|
||||||
import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
|
import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
|
||||||
import { useSessionStore } from "@/stores/session-store";
|
import { selectAgentTimelineState, useSessionStore } from "@/stores/session-store";
|
||||||
import {
|
import {
|
||||||
createInitDeferred,
|
createInitDeferred,
|
||||||
getInitDeferred,
|
getInitDeferred,
|
||||||
@@ -11,7 +11,7 @@ import {
|
|||||||
refreshInitTimeout,
|
refreshInitTimeout,
|
||||||
} from "@/utils/agent-initialization";
|
} from "@/utils/agent-initialization";
|
||||||
import { getHostRuntimeStore, type HostRuntimeStore } from "@/runtime/host-runtime";
|
import { getHostRuntimeStore, type HostRuntimeStore } from "@/runtime/host-runtime";
|
||||||
import { planTimelineTailFetch } from "@/timeline/timeline-sync-plan";
|
import { planTimelineResumeFetch, planTimelineTailFetch } from "@/timeline/timeline-sync-plan";
|
||||||
import { i18n } from "@/i18n/i18next";
|
import { i18n } from "@/i18n/i18next";
|
||||||
|
|
||||||
export type SetAgentInitializing = (agentId: string, initializing: boolean) => void;
|
export type SetAgentInitializing = (agentId: string, initializing: boolean) => void;
|
||||||
@@ -51,7 +51,10 @@ export function ensureAgentIsInitialized(input: EnsureAgentIsInitializedInput):
|
|||||||
return existing.promise;
|
return existing.promise;
|
||||||
}
|
}
|
||||||
|
|
||||||
const timelineRequest = planTimelineTailFetch();
|
const timeline = selectAgentTimelineState(useSessionStore.getState().sessions[serverId], agentId);
|
||||||
|
const timelineRequest = planTimelineResumeFetch(
|
||||||
|
timeline.status === "synced" ? timeline.range : null,
|
||||||
|
);
|
||||||
|
|
||||||
const deferred = createInitDeferred(key, timelineRequest.direction);
|
const deferred = createInitDeferred(key, timelineRequest.direction);
|
||||||
refreshAgentInitializationTimeout({ key, agentId, setAgentInitializing });
|
refreshAgentInitializationTimeout({ key, agentId, setAgentInitializing });
|
||||||
|
|||||||
@@ -111,6 +111,31 @@ function message(id: string, text: string): StreamItem {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function toolCall(): StreamItem {
|
||||||
|
return {
|
||||||
|
kind: "tool_call",
|
||||||
|
id: "tool-1",
|
||||||
|
timelineCursor: { epoch: "epoch-1", seq: 12 },
|
||||||
|
timestamp: new Date("2026-07-18T08:02:00.000Z"),
|
||||||
|
payload: {
|
||||||
|
source: "agent",
|
||||||
|
data: {
|
||||||
|
provider: "codex",
|
||||||
|
callId: "call-1",
|
||||||
|
name: "shell",
|
||||||
|
status: "completed",
|
||||||
|
error: null,
|
||||||
|
detail: {
|
||||||
|
type: "shell",
|
||||||
|
command: "npm test",
|
||||||
|
output: "passed",
|
||||||
|
exitCode: 0,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
function seedSession(): void {
|
function seedSession(): void {
|
||||||
const store = useSessionStore.getState();
|
const store = useSessionStore.getState();
|
||||||
store.initializeSession(SERVER_ID, null);
|
store.initializeSession(SERVER_ID, null);
|
||||||
@@ -209,7 +234,7 @@ describe("ReplicaCache", () => {
|
|||||||
cache.setHosts([]);
|
cache.setHosts([]);
|
||||||
});
|
});
|
||||||
|
|
||||||
it("restores a displayable stale replica without claiming remote hydration", async () => {
|
it("restores the exact persisted canonical timeline window", async () => {
|
||||||
const storage = new MemoryStorage();
|
const storage = new MemoryStorage();
|
||||||
const writer = new ReplicaCache(storage);
|
const writer = new ReplicaCache(storage);
|
||||||
writer.setHosts([SERVER_ID]);
|
writer.setHosts([SERVER_ID]);
|
||||||
@@ -237,11 +262,65 @@ describe("ReplicaCache", () => {
|
|||||||
expect(session.workspaces.get("workspace-1")?.statusEnteredAt).toBeInstanceOf(Date);
|
expect(session.workspaces.get("workspace-1")?.statusEnteredAt).toBeInstanceOf(Date);
|
||||||
expect(session.workspaces.get("workspace-1")?.worktreeSlug).toBe("owned-worktree");
|
expect(session.workspaces.get("workspace-1")?.worktreeSlug).toBe("owned-worktree");
|
||||||
expect(session.agentStreamTail.get("agent-1")).toEqual([message("message-1", "Cached")]);
|
expect(session.agentStreamTail.get("agent-1")).toEqual([message("message-1", "Cached")]);
|
||||||
expect(session.agentAuthoritativeHistoryApplied).toEqual(new Map());
|
expect(session.agentAuthoritativeHistoryApplied).toEqual(new Map([["agent-1", true]]));
|
||||||
expect(session.agentTimelineCursor).toEqual(new Map());
|
expect(session.agentTimelineCursor).toEqual(
|
||||||
expect(session.agentTimelineHasOlder).toEqual(new Map());
|
new Map([["agent-1", { epoch: "epoch-1", startSeq: 1, endSeq: 12 }]]),
|
||||||
expect(session.agentTimelineHasNewer).toEqual(new Map());
|
);
|
||||||
|
expect(session.agentTimelineHasOlder).toEqual(new Map([["agent-1", true]]));
|
||||||
|
expect(session.agentTimelineHasNewer).toEqual(new Map([["agent-1", false]]));
|
||||||
expect(session.agentHistorySyncGeneration).toEqual(new Map());
|
expect(session.agentHistorySyncGeneration).toEqual(new Map());
|
||||||
|
expect(selectAgentTimelineState(session, "agent-1")).toEqual({
|
||||||
|
status: "synced",
|
||||||
|
items: [message("message-1", "Cached")],
|
||||||
|
range: { epoch: "epoch-1", startSeq: 1, endSeq: 12 },
|
||||||
|
older: "available",
|
||||||
|
newer: "none",
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it("restores tool calls inside an authoritative cached window", async () => {
|
||||||
|
const storage = new MemoryStorage();
|
||||||
|
const writer = new ReplicaCache(storage);
|
||||||
|
writer.setHosts([SERVER_ID]);
|
||||||
|
seedSession();
|
||||||
|
useSessionStore.getState().setAgentStreamTail(SERVER_ID, new Map([["agent-1", [toolCall()]]]));
|
||||||
|
await writer.flush();
|
||||||
|
|
||||||
|
useSessionStore.getState().clearSession(SERVER_ID);
|
||||||
|
const reader = new ReplicaCache(storage);
|
||||||
|
reader.setHosts([SERVER_ID]);
|
||||||
|
await reader.restore();
|
||||||
|
|
||||||
|
const session = useSessionStore.getState().sessions[SERVER_ID];
|
||||||
|
expect(session?.agentStreamTail.get("agent-1")).toEqual([toolCall()]);
|
||||||
|
expect(session?.agentTimelineCursor.get("agent-1")).toEqual({
|
||||||
|
epoch: "epoch-1",
|
||||||
|
startSeq: 1,
|
||||||
|
endSeq: 12,
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it("restores display-only when retained items do not reach the stored range end", async () => {
|
||||||
|
const storage = new MemoryStorage();
|
||||||
|
const writer = new ReplicaCache(storage);
|
||||||
|
writer.setHosts([SERVER_ID]);
|
||||||
|
seedSession();
|
||||||
|
useSessionStore
|
||||||
|
.getState()
|
||||||
|
.setAgentTimelineCursor(
|
||||||
|
SERVER_ID,
|
||||||
|
new Map([["agent-1", { epoch: "epoch-1", startSeq: 1, endSeq: 13 }]]),
|
||||||
|
);
|
||||||
|
await writer.flush();
|
||||||
|
|
||||||
|
useSessionStore.getState().clearSession(SERVER_ID);
|
||||||
|
const reader = new ReplicaCache(storage);
|
||||||
|
reader.setHosts([SERVER_ID]);
|
||||||
|
await reader.restore();
|
||||||
|
|
||||||
|
const session = useSessionStore.getState().sessions[SERVER_ID];
|
||||||
|
expect(session?.agentStreamTail.get("agent-1")).toEqual([message("message-1", "Cached")]);
|
||||||
|
expect(session?.agentTimelineCursor).toEqual(new Map());
|
||||||
expect(selectAgentTimelineState(session, "agent-1")).toEqual({
|
expect(selectAgentTimelineState(session, "agent-1")).toEqual({
|
||||||
status: "painted",
|
status: "painted",
|
||||||
items: [message("message-1", "Cached")],
|
items: [message("message-1", "Cached")],
|
||||||
@@ -264,9 +343,10 @@ describe("ReplicaCache", () => {
|
|||||||
normalizeWorkspaceDescriptor(workspace("workspace-2", "project-2", "/repo/other")),
|
normalizeWorkspaceDescriptor(workspace("workspace-2", "project-2", "/repo/other")),
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
const secondTimeline = Array.from({ length: 60 }, (_, index) =>
|
const secondTimeline = Array.from({ length: 60 }, (_, index) => ({
|
||||||
message(`message-${index}`, `Second ${index}`),
|
...message(`message-${index}`, `Second ${index}`),
|
||||||
);
|
timelineCursor: { epoch: "epoch-2", seq: index + 1 },
|
||||||
|
}));
|
||||||
store.setAgentStreamTail(
|
store.setAgentStreamTail(
|
||||||
SERVER_ID,
|
SERVER_ID,
|
||||||
new Map([
|
new Map([
|
||||||
@@ -274,6 +354,12 @@ describe("ReplicaCache", () => {
|
|||||||
["agent-2", secondTimeline],
|
["agent-2", secondTimeline],
|
||||||
]),
|
]),
|
||||||
);
|
);
|
||||||
|
store.setAgentTimelineCursor(
|
||||||
|
SERVER_ID,
|
||||||
|
new Map([["agent-2", { epoch: "epoch-2", startSeq: 1, endSeq: 60 }]]),
|
||||||
|
);
|
||||||
|
store.setAgentTimelineHasOlder(SERVER_ID, new Map([["agent-2", true]]));
|
||||||
|
store.setAgentAuthoritativeHistoryApplied(SERVER_ID, "agent-2", true);
|
||||||
store.setFocusedAgentId(SERVER_ID, "agent-2");
|
store.setFocusedAgentId(SERVER_ID, "agent-2");
|
||||||
await cache.flush();
|
await cache.flush();
|
||||||
|
|
||||||
@@ -293,13 +379,23 @@ describe("ReplicaCache", () => {
|
|||||||
]);
|
]);
|
||||||
expect(Array.from(timelines?.keys() ?? [])).toEqual(["agent-2"]);
|
expect(Array.from(timelines?.keys() ?? [])).toEqual(["agent-2"]);
|
||||||
expect(timelines?.get("agent-2")).toEqual(secondTimeline.slice(-50));
|
expect(timelines?.get("agent-2")).toEqual(secondTimeline.slice(-50));
|
||||||
|
expect(session?.agentTimelineCursor.has("agent-2")).toBe(false);
|
||||||
|
expect(selectAgentTimelineState(session, "agent-2")).toEqual({
|
||||||
|
status: "painted",
|
||||||
|
items: secondTimeline.slice(-50),
|
||||||
|
});
|
||||||
|
|
||||||
const persisted = JSON.parse(storage.values.get("@paseo:replica-cache") ?? "null") as {
|
const persisted = JSON.parse(storage.values.get("@paseo:replica-cache") ?? "null") as {
|
||||||
version: number;
|
version: number;
|
||||||
hosts: Array<{ timeline: Record<string, unknown> | null }>;
|
hosts: Array<{ timeline: Record<string, unknown> | null }>;
|
||||||
};
|
};
|
||||||
expect(persisted.version).toBe(5);
|
expect(persisted.version).toBe(6);
|
||||||
expect(Object.keys(persisted.hosts[0]?.timeline ?? {}).sort()).toEqual(["agentId", "items"]);
|
expect(Object.keys(persisted.hosts[0]?.timeline ?? {}).sort()).toEqual([
|
||||||
|
"agentId",
|
||||||
|
"hasOlder",
|
||||||
|
"items",
|
||||||
|
"range",
|
||||||
|
]);
|
||||||
});
|
});
|
||||||
|
|
||||||
it("persists reconciled rows without caching unreconciled local presentations", async () => {
|
it("persists reconciled rows without caching unreconciled local presentations", async () => {
|
||||||
@@ -404,12 +500,12 @@ describe("ReplicaCache", () => {
|
|||||||
expect(Object.keys(useSessionStore.getState().sessions).sort()).toEqual(["host-a", "host-c"]);
|
expect(Object.keys(useSessionStore.getState().sessions).sort()).toEqual(["host-a", "host-c"]);
|
||||||
});
|
});
|
||||||
|
|
||||||
it("rejects and clears version 4 cache data before overwriting it on flush", async () => {
|
it("rejects and clears version 5 cache data before overwriting it on flush", async () => {
|
||||||
const storage = new MemoryStorage();
|
const storage = new MemoryStorage();
|
||||||
storage.values.set(
|
storage.values.set(
|
||||||
"@paseo:replica-cache",
|
"@paseo:replica-cache",
|
||||||
JSON.stringify({
|
JSON.stringify({
|
||||||
version: 4,
|
version: 5,
|
||||||
hosts: [
|
hosts: [
|
||||||
{
|
{
|
||||||
serverId: SERVER_ID,
|
serverId: SERVER_ID,
|
||||||
@@ -435,7 +531,7 @@ describe("ReplicaCache", () => {
|
|||||||
|
|
||||||
expect(useSessionStore.getState().sessions[SERVER_ID]).toBeUndefined();
|
expect(useSessionStore.getState().sessions[SERVER_ID]).toBeUndefined();
|
||||||
expect(JSON.parse(storage.values.get("@paseo:replica-cache") ?? "null")).toEqual({
|
expect(JSON.parse(storage.values.get("@paseo:replica-cache") ?? "null")).toEqual({
|
||||||
version: 5,
|
version: 6,
|
||||||
hosts: [],
|
hosts: [],
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
import { Buffer } from "buffer";
|
import { Buffer } from "buffer";
|
||||||
import { z } from "zod";
|
import { z } from "zod";
|
||||||
import { AgentStatusSchema } from "@getpaseo/protocol/messages";
|
import { AgentStatusSchema, AgentTimelineItemPayloadSchema } from "@getpaseo/protocol/messages";
|
||||||
import { AgentProviderSchema } from "@getpaseo/protocol/provider-manifest";
|
import { AgentProviderSchema } from "@getpaseo/protocol/provider-manifest";
|
||||||
import {
|
import {
|
||||||
normalizeProjectDescriptor,
|
normalizeProjectDescriptor,
|
||||||
@@ -9,6 +9,7 @@ import {
|
|||||||
useSessionStore,
|
useSessionStore,
|
||||||
type Agent,
|
type Agent,
|
||||||
type SessionReplica,
|
type SessionReplica,
|
||||||
|
type SessionReplicaTimeline,
|
||||||
type SessionState,
|
type SessionState,
|
||||||
type ProjectDescriptor,
|
type ProjectDescriptor,
|
||||||
type WorkspaceDescriptor,
|
type WorkspaceDescriptor,
|
||||||
@@ -17,7 +18,7 @@ import { isUnreconciledLocalUserMessage, type StreamItem } from "@/types/stream"
|
|||||||
import { normalizeAgentSnapshot } from "@/utils/agent-snapshots";
|
import { normalizeAgentSnapshot } from "@/utils/agent-snapshots";
|
||||||
|
|
||||||
const STORAGE_KEY = "@paseo:replica-cache";
|
const STORAGE_KEY = "@paseo:replica-cache";
|
||||||
const CACHE_VERSION = 5;
|
const CACHE_VERSION = 6;
|
||||||
const PERSIST_DELAY_MS = 750;
|
const PERSIST_DELAY_MS = 750;
|
||||||
const MAX_TIMELINE_ITEMS = 50;
|
const MAX_TIMELINE_ITEMS = 50;
|
||||||
const MAX_CACHE_BYTES = 32 * 1024 * 1024;
|
const MAX_CACHE_BYTES = 32 * 1024 * 1024;
|
||||||
@@ -91,6 +92,12 @@ const StoredTimelineItemSchema = z.discriminatedUnion("kind", [
|
|||||||
trigger: z.enum(["auto", "manual"]).optional(),
|
trigger: z.enum(["auto", "manual"]).optional(),
|
||||||
preTokens: z.number().nonnegative().optional(),
|
preTokens: z.number().nonnegative().optional(),
|
||||||
}),
|
}),
|
||||||
|
z.strictObject({
|
||||||
|
...TimelineItemBaseShape,
|
||||||
|
kind: z.literal("tool_call"),
|
||||||
|
provider: AgentProviderSchema,
|
||||||
|
item: AgentTimelineItemPayloadSchema.refine((item) => item.type === "tool_call"),
|
||||||
|
}),
|
||||||
]);
|
]);
|
||||||
|
|
||||||
const AgentCapabilitiesSchema = z.strictObject({
|
const AgentCapabilitiesSchema = z.strictObject({
|
||||||
@@ -246,6 +253,14 @@ const StoredProjectSchema = z.strictObject({
|
|||||||
const StoredTimelineSchema = z.strictObject({
|
const StoredTimelineSchema = z.strictObject({
|
||||||
agentId: z.string(),
|
agentId: z.string(),
|
||||||
items: z.array(StoredTimelineItemSchema),
|
items: z.array(StoredTimelineItemSchema),
|
||||||
|
range: z
|
||||||
|
.strictObject({
|
||||||
|
epoch: z.string(),
|
||||||
|
startSeq: z.number().int().nonnegative(),
|
||||||
|
endSeq: z.number().int().nonnegative(),
|
||||||
|
})
|
||||||
|
.nullable(),
|
||||||
|
hasOlder: z.boolean(),
|
||||||
});
|
});
|
||||||
|
|
||||||
const StoredHostSchema = z.strictObject({
|
const StoredHostSchema = z.strictObject({
|
||||||
@@ -275,6 +290,8 @@ interface ReplicaInput {
|
|||||||
projects: ReadonlyMap<string, ProjectDescriptor>;
|
projects: ReadonlyMap<string, ProjectDescriptor>;
|
||||||
focusedAgent: Agent | undefined;
|
focusedAgent: Agent | undefined;
|
||||||
timelineItems: StreamItem[] | undefined;
|
timelineItems: StreamItem[] | undefined;
|
||||||
|
timelineRange: SessionReplicaTimeline["range"];
|
||||||
|
timelineHasOlder: boolean;
|
||||||
}
|
}
|
||||||
|
|
||||||
export interface ReplicaCacheStorage {
|
export interface ReplicaCacheStorage {
|
||||||
@@ -294,6 +311,8 @@ function deserializeTimeline(stored: StoredHost["timeline"]): SessionReplica["ti
|
|||||||
return {
|
return {
|
||||||
agentId: stored.agentId,
|
agentId: stored.agentId,
|
||||||
items: stored.items.map(deserializeTimelineItem),
|
items: stored.items.map(deserializeTimelineItem),
|
||||||
|
range: stored.range,
|
||||||
|
hasOlder: stored.hasOlder,
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -351,7 +370,23 @@ function serializeTimelineItem(item: StreamItem): StoredTimelineItem | null {
|
|||||||
...(item.preTokens !== undefined ? { preTokens: item.preTokens } : {}),
|
...(item.preTokens !== undefined ? { preTokens: item.preTokens } : {}),
|
||||||
};
|
};
|
||||||
case "tool_call":
|
case "tool_call":
|
||||||
return null;
|
if (item.payload.source !== "agent") return null;
|
||||||
|
const storedTool = AgentTimelineItemPayloadSchema.safeParse({
|
||||||
|
type: "tool_call",
|
||||||
|
callId: item.payload.data.callId,
|
||||||
|
name: item.payload.data.name,
|
||||||
|
status: item.payload.data.status,
|
||||||
|
error: item.payload.data.error,
|
||||||
|
detail: item.payload.data.detail,
|
||||||
|
...(item.payload.data.metadata ? { metadata: item.payload.data.metadata } : {}),
|
||||||
|
});
|
||||||
|
if (!storedTool.success || storedTool.data.type !== "tool_call") return null;
|
||||||
|
return {
|
||||||
|
...base,
|
||||||
|
kind: item.kind,
|
||||||
|
provider: item.payload.data.provider,
|
||||||
|
item: storedTool.data,
|
||||||
|
};
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -404,6 +439,28 @@ function deserializeTimelineItem(item: StoredTimelineItem): StreamItem {
|
|||||||
...(item.trigger ? { trigger: item.trigger } : {}),
|
...(item.trigger ? { trigger: item.trigger } : {}),
|
||||||
...(item.preTokens !== undefined ? { preTokens: item.preTokens } : {}),
|
...(item.preTokens !== undefined ? { preTokens: item.preTokens } : {}),
|
||||||
};
|
};
|
||||||
|
case "tool_call": {
|
||||||
|
const tool = item.item;
|
||||||
|
if (tool.type !== "tool_call") {
|
||||||
|
throw new Error("Stored tool call contains a non-tool timeline item");
|
||||||
|
}
|
||||||
|
return {
|
||||||
|
...base,
|
||||||
|
kind: item.kind,
|
||||||
|
payload: {
|
||||||
|
source: "agent",
|
||||||
|
data: {
|
||||||
|
provider: item.provider,
|
||||||
|
callId: tool.callId,
|
||||||
|
name: tool.name,
|
||||||
|
status: tool.status,
|
||||||
|
error: tool.error,
|
||||||
|
detail: tool.detail,
|
||||||
|
...(tool.metadata ? { metadata: tool.metadata } : {}),
|
||||||
|
},
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -529,13 +586,28 @@ function serializeProject(project: ProjectDescriptor): StoredProject {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function isTimelineItemStoredLosslessly(item: StreamItem): boolean {
|
||||||
|
switch (item.kind) {
|
||||||
|
case "user_message":
|
||||||
|
return (item.images?.length ?? 0) === 0 && (item.attachments?.length ?? 0) === 0;
|
||||||
|
case "activity_log":
|
||||||
|
return item.metadata === undefined;
|
||||||
|
case "tool_call":
|
||||||
|
return item.payload.source === "agent";
|
||||||
|
default:
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
function replicaInputsEqual(left: ReplicaInput, right: ReplicaInput): boolean {
|
function replicaInputsEqual(left: ReplicaInput, right: ReplicaInput): boolean {
|
||||||
return (
|
return (
|
||||||
left.agents === right.agents &&
|
left.agents === right.agents &&
|
||||||
left.workspaces === right.workspaces &&
|
left.workspaces === right.workspaces &&
|
||||||
left.projects === right.projects &&
|
left.projects === right.projects &&
|
||||||
left.focusedAgent === right.focusedAgent &&
|
left.focusedAgent === right.focusedAgent &&
|
||||||
left.timelineItems === right.timelineItems
|
left.timelineItems === right.timelineItems &&
|
||||||
|
left.timelineRange === right.timelineRange &&
|
||||||
|
left.timelineHasOlder === right.timelineHasOlder
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -550,14 +622,33 @@ function selectReplicaInput(session: SessionState, agentId: string | null): Repl
|
|||||||
projects: session.projects,
|
projects: session.projects,
|
||||||
focusedAgent: agent,
|
focusedAgent: agent,
|
||||||
timelineItems: timeline.status === "cold" ? undefined : timeline.items,
|
timelineItems: timeline.status === "cold" ? undefined : timeline.items,
|
||||||
|
timelineRange: timeline.status === "synced" ? timeline.range : null,
|
||||||
|
timelineHasOlder: timeline.status === "synced" && timeline.older === "available",
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
function serializeHost(serverId: string, input: ReplicaInput, directorySync?: unknown): StoredHost {
|
function serializeHost(serverId: string, input: ReplicaInput, directorySync?: unknown): StoredHost {
|
||||||
const items = input.timelineItems
|
const canonicalItems = input.timelineItems?.filter(
|
||||||
?.filter((item) => item.kind !== "user_message" || !isUnreconciledLocalUserMessage(item))
|
(item) => item.kind !== "user_message" || !isUnreconciledLocalUserMessage(item),
|
||||||
.map(serializeTimelineItem)
|
);
|
||||||
.filter((item) => item !== null);
|
const items = canonicalItems
|
||||||
|
? canonicalItems.map(serializeTimelineItem).filter((item) => item !== null)
|
||||||
|
: undefined;
|
||||||
|
const range = input.timelineRange;
|
||||||
|
const canPersistCoverage =
|
||||||
|
range !== null &&
|
||||||
|
range.retainedRanges === undefined &&
|
||||||
|
canonicalItems !== undefined &&
|
||||||
|
canonicalItems.length <= MAX_TIMELINE_ITEMS &&
|
||||||
|
items?.length === canonicalItems.length &&
|
||||||
|
canonicalItems.every(
|
||||||
|
(item) =>
|
||||||
|
isTimelineItemStoredLosslessly(item) &&
|
||||||
|
item.timelineCursor?.epoch === range.epoch &&
|
||||||
|
item.timelineCursor.seq >= range.startSeq &&
|
||||||
|
item.timelineCursor.seq <= range.endSeq,
|
||||||
|
) &&
|
||||||
|
canonicalItems.some((item) => item.timelineCursor?.seq === range.endSeq);
|
||||||
return {
|
return {
|
||||||
serverId,
|
serverId,
|
||||||
agents: Array.from(input.agents.values(), serializeAgent),
|
agents: Array.from(input.agents.values(), serializeAgent),
|
||||||
@@ -569,6 +660,10 @@ function serializeHost(serverId: string, input: ReplicaInput, directorySync?: un
|
|||||||
? {
|
? {
|
||||||
agentId: input.focusedAgent.id,
|
agentId: input.focusedAgent.id,
|
||||||
items: items.slice(-MAX_TIMELINE_ITEMS),
|
items: items.slice(-MAX_TIMELINE_ITEMS),
|
||||||
|
range: canPersistCoverage
|
||||||
|
? { epoch: range.epoch, startSeq: range.startSeq, endSeq: range.endSeq }
|
||||||
|
: null,
|
||||||
|
hasOlder: canPersistCoverage ? input.timelineHasOlder : false,
|
||||||
}
|
}
|
||||||
: null,
|
: null,
|
||||||
...(directorySync ? { directorySync } : {}),
|
...(directorySync ? { directorySync } : {}),
|
||||||
@@ -783,8 +878,21 @@ export class ReplicaCache {
|
|||||||
if (session.focusedAgentId) {
|
if (session.focusedAgentId) {
|
||||||
this.lastFocusedAgentIds.set(serverId, session.focusedAgentId);
|
this.lastFocusedAgentIds.set(serverId, session.focusedAgentId);
|
||||||
}
|
}
|
||||||
const input = selectReplicaInput(session, this.lastFocusedAgentIds.get(serverId) ?? null);
|
const focusedAgentId = this.lastFocusedAgentIds.get(serverId) ?? null;
|
||||||
const previous = this.capturedInputs.get(serverId);
|
const previous = this.capturedInputs.get(serverId);
|
||||||
|
const selected = selectReplicaInput(session, focusedAgentId);
|
||||||
|
const hasTimelineHead =
|
||||||
|
focusedAgentId !== null && (session.agentStreamHead.get(focusedAgentId)?.length ?? 0) > 0;
|
||||||
|
let input = selected;
|
||||||
|
if (hasTimelineHead && previous && selected.timelineItems === previous.timelineItems) {
|
||||||
|
input = {
|
||||||
|
...selected,
|
||||||
|
timelineRange: previous.timelineRange,
|
||||||
|
timelineHasOlder: previous.timelineHasOlder,
|
||||||
|
};
|
||||||
|
} else if (hasTimelineHead) {
|
||||||
|
input = { ...selected, timelineRange: null, timelineHasOlder: false };
|
||||||
|
}
|
||||||
if (previous && replicaInputsEqual(previous, input)) return false;
|
if (previous && replicaInputsEqual(previous, input)) return false;
|
||||||
|
|
||||||
this.capturedInputs.set(serverId, input);
|
this.capturedInputs.set(serverId, input);
|
||||||
|
|||||||
@@ -308,6 +308,8 @@ export interface AgentTimelineCursorState {
|
|||||||
export interface SessionReplicaTimeline {
|
export interface SessionReplicaTimeline {
|
||||||
agentId: string;
|
agentId: string;
|
||||||
items: StreamItem[];
|
items: StreamItem[];
|
||||||
|
range: AgentTimelineCursorState | null;
|
||||||
|
hasOlder: boolean;
|
||||||
}
|
}
|
||||||
|
|
||||||
export interface SessionReplica {
|
export interface SessionReplica {
|
||||||
@@ -789,6 +791,16 @@ export const useSessionStore = create<SessionStore>()(
|
|||||||
const tasks = latestTasksFromStream(timeline.items);
|
const tasks = latestTasksFromStream(timeline.items);
|
||||||
if (tasks.length > 0) agentTasks.set(timeline.agentId, tasks);
|
if (tasks.length > 0) agentTasks.set(timeline.agentId, tasks);
|
||||||
}
|
}
|
||||||
|
const agentTimelineCursor = new Map<string, AgentTimelineCursorState>();
|
||||||
|
const agentTimelineHasOlder = new Map<string, boolean>();
|
||||||
|
const agentTimelineHasNewer = new Map<string, boolean>();
|
||||||
|
const agentAuthoritativeHistoryApplied = new Map<string, boolean>();
|
||||||
|
if (timeline?.range) {
|
||||||
|
agentTimelineCursor.set(timeline.agentId, timeline.range);
|
||||||
|
agentTimelineHasOlder.set(timeline.agentId, timeline.hasOlder);
|
||||||
|
agentTimelineHasNewer.set(timeline.agentId, false);
|
||||||
|
agentAuthoritativeHistoryApplied.set(timeline.agentId, true);
|
||||||
|
}
|
||||||
const agentLastActivity = new Map(prev.agentLastActivity);
|
const agentLastActivity = new Map(prev.agentLastActivity);
|
||||||
for (const agent of replica.agents.values()) {
|
for (const agent of replica.agents.values()) {
|
||||||
agentLastActivity.set(agent.id, agent.lastActivityAt);
|
agentLastActivity.set(agent.id, agent.lastActivityAt);
|
||||||
@@ -806,6 +818,10 @@ export const useSessionStore = create<SessionStore>()(
|
|||||||
hasWorkspaceDirectorySnapshot: true,
|
hasWorkspaceDirectorySnapshot: true,
|
||||||
agentStreamTail,
|
agentStreamTail,
|
||||||
agentTasks,
|
agentTasks,
|
||||||
|
agentTimelineCursor,
|
||||||
|
agentTimelineHasOlder,
|
||||||
|
agentTimelineHasNewer,
|
||||||
|
agentAuthoritativeHistoryApplied,
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
agentLastActivity,
|
agentLastActivity,
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import {
|
|||||||
isTimelineResumeSnapshotAuthoritative,
|
isTimelineResumeSnapshotAuthoritative,
|
||||||
planTimelineOlderFetch,
|
planTimelineOlderFetch,
|
||||||
planTimelinePromptJump,
|
planTimelinePromptJump,
|
||||||
|
planTimelineResumeFetch,
|
||||||
planTimelineTailFetch,
|
planTimelineTailFetch,
|
||||||
} from "./timeline-sync-plan";
|
} from "./timeline-sync-plan";
|
||||||
|
|
||||||
@@ -19,6 +20,15 @@ describe("timeline sync planning", () => {
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("a restored canonical range catches up after its exact end", () => {
|
||||||
|
expect(planTimelineResumeFetch({ epoch: "epoch-1", endSeq: 49 })).toEqual({
|
||||||
|
direction: "after",
|
||||||
|
cursor: { epoch: "epoch-1", seq: 49 },
|
||||||
|
limit: TIMELINE_FETCH_PAGE_SIZE,
|
||||||
|
projection: "projected",
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
test("older history loads one bounded page before the start cursor", () => {
|
test("older history loads one bounded page before the start cursor", () => {
|
||||||
const plan = planTimelineOlderFetch({ epoch: "epoch-1", seq: 25 });
|
const plan = planTimelineOlderFetch({ epoch: "epoch-1", seq: 25 });
|
||||||
|
|
||||||
|
|||||||
@@ -51,6 +51,14 @@ export function planTimelineTailFetch() {
|
|||||||
} as const;
|
} as const;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export function planTimelineResumeFetch(
|
||||||
|
range: { epoch: string; endSeq: number } | null | undefined,
|
||||||
|
): ProjectedTimelineForwardFetchPlan {
|
||||||
|
return range
|
||||||
|
? planTimelineCatchUpAfter({ epoch: range.epoch, seq: range.endSeq })
|
||||||
|
: planTimelineTailFetch();
|
||||||
|
}
|
||||||
|
|
||||||
export function planTimelineOlderFetch(cursor: TimelineSyncCursor) {
|
export function planTimelineOlderFetch(cursor: TimelineSyncCursor) {
|
||||||
return {
|
return {
|
||||||
direction: "before",
|
direction: "before",
|
||||||
|
|||||||
@@ -33,6 +33,7 @@ interface TimelineFetch {
|
|||||||
|
|
||||||
class TimelineWorld {
|
class TimelineWorld {
|
||||||
readonly errors: string[] = [];
|
readonly errors: string[] = [];
|
||||||
|
readonly cursors = new Map<string, { epoch: string; endSeq: number }>();
|
||||||
readonly sync = createViewedTimelineSync({
|
readonly sync = createViewedTimelineSync({
|
||||||
initialDeliveryMode: "selective",
|
initialDeliveryMode: "selective",
|
||||||
setSubscription: async (agentIds) => {
|
setSubscription: async (agentIds) => {
|
||||||
@@ -45,6 +46,7 @@ class TimelineWorld {
|
|||||||
this.releaseMembershipWaiter();
|
this.releaseMembershipWaiter();
|
||||||
return result.promise;
|
return result.promise;
|
||||||
},
|
},
|
||||||
|
readCursor: (agentId) => this.cursors.get(agentId),
|
||||||
fetchPage: async (agentId, request) => {
|
fetchPage: async (agentId, request) => {
|
||||||
const result = deferred<{
|
const result = deferred<{
|
||||||
hasNewer: boolean;
|
hasNewer: boolean;
|
||||||
@@ -63,6 +65,12 @@ class TimelineWorld {
|
|||||||
this.releaseFetchWaiters();
|
this.releaseFetchWaiters();
|
||||||
return result.promise;
|
return result.promise;
|
||||||
},
|
},
|
||||||
|
fetchLatestTail: (agentId) =>
|
||||||
|
this.fetchTimeline(agentId, {
|
||||||
|
direction: "tail",
|
||||||
|
limit: 40,
|
||||||
|
projection: "projected",
|
||||||
|
}),
|
||||||
reportError: (error) => {
|
reportError: (error) => {
|
||||||
this.errors.push(error instanceof Error ? error.message : String(error));
|
this.errors.push(error instanceof Error ? error.message : String(error));
|
||||||
const waiter = this.errorWaiters.shift();
|
const waiter = this.errorWaiters.shift();
|
||||||
@@ -94,6 +102,25 @@ class TimelineWorld {
|
|||||||
private readonly scheduled: Array<{ task: () => void; delayMs: number }> = [];
|
private readonly scheduled: Array<{ task: () => void; delayMs: number }> = [];
|
||||||
private readonly retryWaiters: Array<(retry: () => void) => void> = [];
|
private readonly retryWaiters: Array<(retry: () => void) => void> = [];
|
||||||
|
|
||||||
|
private fetchTimeline(
|
||||||
|
agentId: string,
|
||||||
|
request: ProjectedTimelineForwardFetchPlan,
|
||||||
|
): Promise<{ hasNewer: boolean; endCursor: { epoch: string; seq: number } | null }> {
|
||||||
|
const result = deferred<{
|
||||||
|
hasNewer: boolean;
|
||||||
|
endCursor: { epoch: string; seq: number } | null;
|
||||||
|
}>();
|
||||||
|
this.fetches.push({
|
||||||
|
agentId,
|
||||||
|
request,
|
||||||
|
respond: ({ hasNewer, seq = 1 }) =>
|
||||||
|
result.resolve({ hasNewer, endCursor: { epoch: `epoch-${agentId}`, seq } }),
|
||||||
|
fail: (message) => result.reject(new Error(message)),
|
||||||
|
});
|
||||||
|
this.releaseFetchWaiters();
|
||||||
|
return result.promise;
|
||||||
|
}
|
||||||
|
|
||||||
nextMembership(): Promise<MembershipRequest> {
|
nextMembership(): Promise<MembershipRequest> {
|
||||||
const request = this.memberships.shift();
|
const request = this.memberships.shift();
|
||||||
if (request) return Promise.resolve(request);
|
if (request) return Promise.resolve(request);
|
||||||
@@ -166,6 +193,41 @@ test("uses a tail fetch when an agent becomes visible", async () => {
|
|||||||
fetch.respond({ hasNewer: false });
|
fetch.respond({ hasNewer: false });
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("catches up after the restored cursor when an agent becomes visible", async () => {
|
||||||
|
const world = new TimelineWorld();
|
||||||
|
world.cursors.set("agent-a", { epoch: "epoch-agent-a", endSeq: 42 });
|
||||||
|
world.sync.setConnected(true);
|
||||||
|
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
|
||||||
|
const membership = await world.nextMembership();
|
||||||
|
membership.succeed();
|
||||||
|
|
||||||
|
const fetch = await world.nextFetch("agent-a");
|
||||||
|
expect(fetch.request).toEqual({
|
||||||
|
direction: "after",
|
||||||
|
cursor: { epoch: "epoch-agent-a", seq: 42 },
|
||||||
|
limit: 40,
|
||||||
|
projection: "projected",
|
||||||
|
});
|
||||||
|
fetch.respond({ hasNewer: false });
|
||||||
|
});
|
||||||
|
|
||||||
|
test("falls back to the latest tail when a restored cursor has more than one catch-up page", async () => {
|
||||||
|
const world = new TimelineWorld();
|
||||||
|
world.cursors.set("agent-a", { epoch: "epoch-agent-a", endSeq: 42 });
|
||||||
|
world.sync.setConnected(true);
|
||||||
|
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
|
||||||
|
const membership = await world.nextMembership();
|
||||||
|
membership.succeed();
|
||||||
|
|
||||||
|
const probe = await world.nextFetch("agent-a");
|
||||||
|
expect(probe.request.direction).toBe("after");
|
||||||
|
probe.respond({ hasNewer: true, seq: 82 });
|
||||||
|
|
||||||
|
const fallback = await world.nextFetch("agent-a");
|
||||||
|
expect(fallback.request).toEqual({ direction: "tail", limit: 40, projection: "projected" });
|
||||||
|
fallback.respond({ hasNewer: false });
|
||||||
|
});
|
||||||
|
|
||||||
test("a gap absorbed by a running tail is recovered after the tail completes", async () => {
|
test("a gap absorbed by a running tail is recovered after the tail completes", async () => {
|
||||||
const world = new TimelineWorld();
|
const world = new TimelineWorld();
|
||||||
world.sync.setConnected(true);
|
world.sync.setConnected(true);
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
import {
|
import {
|
||||||
planTimelineCatchUpAfter,
|
planTimelineCatchUpAfter,
|
||||||
planTimelineTailFetch,
|
planTimelineResumeFetch,
|
||||||
type ProjectedTimelineForwardFetchPlan,
|
type ProjectedTimelineForwardFetchPlan,
|
||||||
} from "./timeline-sync-plan";
|
} from "./timeline-sync-plan";
|
||||||
|
|
||||||
@@ -12,10 +12,12 @@ interface TimelinePageResult {
|
|||||||
interface ViewedTimelineSyncPorts {
|
interface ViewedTimelineSyncPorts {
|
||||||
initialDeliveryMode: TimelineDeliveryMode;
|
initialDeliveryMode: TimelineDeliveryMode;
|
||||||
setSubscription(agentIds: string[]): Promise<void>;
|
setSubscription(agentIds: string[]): Promise<void>;
|
||||||
|
readCursor(agentId: string): { epoch: string; endSeq: number } | undefined;
|
||||||
fetchPage(
|
fetchPage(
|
||||||
agentId: string,
|
agentId: string,
|
||||||
request: ProjectedTimelineForwardFetchPlan,
|
request: ProjectedTimelineForwardFetchPlan,
|
||||||
): Promise<TimelinePageResult>;
|
): Promise<TimelinePageResult>;
|
||||||
|
fetchLatestTail(agentId: string): Promise<TimelinePageResult>;
|
||||||
reportError(error: unknown): void;
|
reportError(error: unknown): void;
|
||||||
schedule(task: () => void, delayMs: number): () => void;
|
schedule(task: () => void, delayMs: number): () => void;
|
||||||
}
|
}
|
||||||
@@ -136,6 +138,12 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
|
|||||||
|
|
||||||
const isAcknowledged = (agentId: string) => acknowledged.includes(agentId);
|
const isAcknowledged = (agentId: string) => acknowledged.includes(agentId);
|
||||||
const isDesired = (agentId: string) => desired.includes(agentId);
|
const isDesired = (agentId: string) => desired.includes(agentId);
|
||||||
|
const ownsCatchUp = (agentId: string, generation: number) =>
|
||||||
|
!disposed &&
|
||||||
|
connected &&
|
||||||
|
isDesired(agentId) &&
|
||||||
|
isAcknowledged(agentId) &&
|
||||||
|
catchUps.get(agentId)?.generation === generation;
|
||||||
|
|
||||||
const notifyListeners = () => {
|
const notifyListeners = () => {
|
||||||
for (const listener of listeners) listener();
|
for (const listener of listeners) listener();
|
||||||
@@ -168,30 +176,26 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
|
|||||||
agentId: string,
|
agentId: string,
|
||||||
generation: number,
|
generation: number,
|
||||||
request: ProjectedTimelineForwardFetchPlan,
|
request: ProjectedTimelineForwardFetchPlan,
|
||||||
|
fallbackToLatestTailOnOverflow: boolean,
|
||||||
): Promise<void> => {
|
): Promise<void> => {
|
||||||
if (
|
if (!ownsCatchUp(agentId, generation)) return;
|
||||||
disposed ||
|
|
||||||
!connected ||
|
|
||||||
!isDesired(agentId) ||
|
|
||||||
!isAcknowledged(agentId) ||
|
|
||||||
catchUps.get(agentId)?.generation !== generation
|
|
||||||
) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const page = await ports.fetchPage(agentId, request);
|
const page = await ports.fetchPage(agentId, request);
|
||||||
if (
|
if (!ownsCatchUp(agentId, generation)) return;
|
||||||
disposed ||
|
|
||||||
!connected ||
|
|
||||||
!isDesired(agentId) ||
|
|
||||||
!isAcknowledged(agentId) ||
|
|
||||||
catchUps.get(agentId)?.generation !== generation
|
|
||||||
) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
if (page.hasNewer && page.endCursor) {
|
if (page.hasNewer && page.endCursor) {
|
||||||
await fetchUntilCurrent(agentId, generation, planTimelineCatchUpAfter(page.endCursor));
|
if (fallbackToLatestTailOnOverflow) {
|
||||||
|
await ports.fetchLatestTail(agentId);
|
||||||
|
catchUps.set(agentId, { generation, status: "complete" });
|
||||||
|
setVisibilityCatchUpReady(agentId);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
await fetchUntilCurrent(
|
||||||
|
agentId,
|
||||||
|
generation,
|
||||||
|
planTimelineCatchUpAfter(page.endCursor),
|
||||||
|
false,
|
||||||
|
);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
if (page.hasNewer) {
|
if (page.hasNewer) {
|
||||||
@@ -230,7 +234,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
|
|||||||
if (request) pendingCatchUps.set(agentId, request);
|
if (request) pendingCatchUps.set(agentId, request);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const nextRequest = request ?? planTimelineTailFetch();
|
const nextRequest = request ?? planTimelineResumeFetch(ports.readCursor(agentId));
|
||||||
const current = catchUps.get(agentId);
|
const current = catchUps.get(agentId);
|
||||||
const decision = decideCatchUp({ current, request: nextRequest, supersede });
|
const decision = decideCatchUp({ current, request: nextRequest, supersede });
|
||||||
if (decision === "keep-and-park") {
|
if (decision === "keep-and-park") {
|
||||||
@@ -245,7 +249,12 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
|
|||||||
catchUpGenerations.set(agentId, generation);
|
catchUpGenerations.set(agentId, generation);
|
||||||
catchUps.set(agentId, { generation, status: "running", request: nextRequest });
|
catchUps.set(agentId, { generation, status: "running", request: nextRequest });
|
||||||
pendingCatchUps.delete(agentId);
|
pendingCatchUps.delete(agentId);
|
||||||
void fetchUntilCurrent(agentId, generation, nextRequest);
|
void fetchUntilCurrent(
|
||||||
|
agentId,
|
||||||
|
generation,
|
||||||
|
nextRequest,
|
||||||
|
request === undefined && nextRequest.direction === "after",
|
||||||
|
);
|
||||||
};
|
};
|
||||||
|
|
||||||
const startAcknowledgedCatchUps = () => {
|
const startAcknowledgedCatchUps = () => {
|
||||||
|
|||||||
Reference in New Issue
Block a user