mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-30 21:23:04 +00:00
Mohamed Boudra
83f9fba209
* fix(app): keep the draft-create handoff until the new chat is current Creating an agent from the draft composer on web flashed the "Updating messages" panel toast for 40-200ms. That toast means "this chat may be out of date, we are catching up" — a chat this client just created cannot be. The timeline reducer released the create handoff at the first synchronized page, but the visibility catch-up flag only clears once catch-up settles, and a fresh chat always has a parked follow-up "after" page because live rows arrive during the tail fetch. Between the two releases the state machine saw a hydrated chat with no handoff and a pending catch-up, which resolves to the resume toast. #4863 turned that branch from silent to status, which is what made the window visible. Both signals now flip in one synchronous call: createViewedTimelineSync gains an onTimelineCurrent port, called from setVisibilityCatchUpReady — the single place the sync declares a chat current, already covering the parked follow-up and the fetchLatestTail overflow path. The owner implements it by clearing the handoff. The optimistic "running" override therefore lasts one extra round trip. The state machine and sync-status rules are unchanged; they were correct. * fix(app): release the create handoff when a chat leaves timeline demand Closing a new chat's tab before catch-up settled left its sent attempt in useCreateFlowStore forever. Reopening it in the same session then supplied optimistic-create continuity for an authoritative chat: the panel showed it as running and suppressed "Updating messages" while the real catch-up ran — the state this fix exists to protect. An offline or slow host widens that window from milliseconds to minutes. Leaving the demanded set ends the owner's obligation to a chat exactly as reaching current does, so onTimelineCurrent becomes onCatchUpEnded and fires on both. It reports after the membership commit, so the sync never describes itself mid-transition. Disconnect and backgrounding preserve demand and therefore preserve the handoff.
668 lines
22 KiB
TypeScript
668 lines
22 KiB
TypeScript
import { DatabaseSync } from "node:sqlite";
|
|
import { ReplicaCache } from "@/runtime/replica-cache";
|
|
import {
|
|
createSqliteReplicaRowStore,
|
|
type ReplicaSqliteConnection,
|
|
type SqliteValue,
|
|
} from "@/runtime/replica-cache/row-store-sqlite";
|
|
import { afterEach, describe, expect, it } from "vitest";
|
|
import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages";
|
|
import type { CachedTimeline } from "@/runtime/replica-cache";
|
|
import { selectAgentTimelineState, useSessionStore } from "@/stores/session-store";
|
|
import { useCreateFlowStore, type PendingCreateAttempt } from "@/stores/create-flow-store";
|
|
import type { StreamItem } from "@/types/stream";
|
|
import {
|
|
createTimelineReplica,
|
|
createViewedTimelineOwner,
|
|
type TimelineReplicaStorage,
|
|
type ViewedTimelineOwner,
|
|
} from "./viewed-timeline-sync";
|
|
|
|
const SERVER_ID = "timeline-replica-host";
|
|
const AGENT_ID = "agent-1";
|
|
|
|
function item(id: string, text: string, seq: number): StreamItem {
|
|
return {
|
|
kind: "assistant_message",
|
|
id,
|
|
text,
|
|
timestamp: new Date("2026-08-26T10:00:00.000Z"),
|
|
timelineCursor: { epoch: "epoch-1", seq },
|
|
};
|
|
}
|
|
|
|
function cachedTimeline(): CachedTimeline {
|
|
return {
|
|
agentId: AGENT_ID,
|
|
items: [item("cached", "cached", 4)],
|
|
range: { epoch: "epoch-1", startSeq: 1, endSeq: 4 },
|
|
hasOlder: true,
|
|
};
|
|
}
|
|
|
|
function createOwner(storage: TimelineReplicaStorage): ViewedTimelineOwner {
|
|
const replica = createTimelineReplica({
|
|
serverId: SERVER_ID,
|
|
storage,
|
|
prepareAgent: async () => undefined,
|
|
});
|
|
return createViewedTimelineOwner({
|
|
serverId: SERVER_ID,
|
|
replica,
|
|
replaceDemandedAgentIds: () => undefined,
|
|
drainQueuedAgentMessage: () => undefined,
|
|
ports: {
|
|
observe: () => ({ ready: Promise.resolve(), release: async () => undefined }),
|
|
readCursor: () => undefined,
|
|
fetchPage: async () => ({ hasNewer: false, endCursor: null }),
|
|
fetchLatestTail: async () => ({ hasNewer: false, endCursor: null }),
|
|
reportError: () => undefined,
|
|
schedule: () => () => undefined,
|
|
},
|
|
});
|
|
}
|
|
|
|
function applySynced(agentId: string, seq: number): void {
|
|
useSessionStore.getState().applyAgentTimelineResponseState(SERVER_ID, agentId, {
|
|
items: [item(`network-${agentId}`, "network", seq)],
|
|
head: [],
|
|
range: { epoch: "epoch-1", startSeq: 1, endSeq: seq },
|
|
older: "available",
|
|
newer: false,
|
|
synchronized: true,
|
|
acknowledgedClientMessageIds: [],
|
|
});
|
|
}
|
|
|
|
afterEach(() => useSessionStore.getState().clearSession(SERVER_ID));
|
|
|
|
describe("viewed timeline persistence", () => {
|
|
it("shares an in-flight cache preparation with the viewed owner", async () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
let release!: (value: CachedTimeline) => void;
|
|
let reads = 0;
|
|
const read = new Promise<CachedTimeline>((resolve) => {
|
|
release = resolve;
|
|
});
|
|
const replica = createTimelineReplica({
|
|
serverId: SERVER_ID,
|
|
storage: {
|
|
readTimeline: () => {
|
|
reads += 1;
|
|
return read;
|
|
},
|
|
commitTimeline: () => undefined,
|
|
},
|
|
prepareAgent: async () => undefined,
|
|
});
|
|
|
|
const routePreparation = replica.prepare(AGENT_ID);
|
|
const ownerPreparation = replica.prepare(AGENT_ID);
|
|
release(cachedTimeline());
|
|
await Promise.all([routePreparation, ownerPreparation]);
|
|
|
|
expect(reads).toBe(1);
|
|
expect(replica.readCursor(AGENT_ID)).toEqual({ epoch: "epoch-1", endSeq: 4 });
|
|
});
|
|
|
|
it("paints cached history without claiming authoritative synchronization", async () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
const owner = createOwner({
|
|
readTimeline: async () => cachedTimeline(),
|
|
commitTimeline: () => undefined,
|
|
});
|
|
|
|
owner.replaceVisibleAgentIds("test", [AGENT_ID]);
|
|
|
|
await expect
|
|
.poll(() =>
|
|
selectAgentTimelineState(useSessionStore.getState().sessions[SERVER_ID], AGENT_ID),
|
|
)
|
|
.toEqual({ status: "painted", items: cachedTimeline().items });
|
|
owner.dispose();
|
|
});
|
|
|
|
it("reconciles an overlapping projected message against its cached cursor", async () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
const partial = "```mermaid\nflowchart LR\n Start --> Mid";
|
|
const complete = `${partial}dle\n Middle --> Done\n\`\`\``;
|
|
const owner = createOwner({
|
|
readTimeline: async () => ({
|
|
agentId: AGENT_ID,
|
|
items: [item("cached", partial, 4)],
|
|
range: { epoch: "epoch-1", startSeq: 1, endSeq: 4 },
|
|
hasOlder: false,
|
|
}),
|
|
commitTimeline: () => undefined,
|
|
});
|
|
|
|
owner.replaceVisibleAgentIds("test", [AGENT_ID]);
|
|
await expect
|
|
.poll(() =>
|
|
selectAgentTimelineState(useSessionStore.getState().sessions[SERVER_ID], AGENT_ID),
|
|
)
|
|
.toMatchObject({ status: "painted" });
|
|
|
|
owner.applyTimelineResponse({
|
|
requestId: "page-after-cache",
|
|
agentId: AGENT_ID,
|
|
agent: null,
|
|
direction: "after",
|
|
projection: "projected",
|
|
reset: false,
|
|
epoch: "epoch-1",
|
|
window: { minSeq: 1, maxSeq: 5, nextSeq: 6 },
|
|
startCursor: { epoch: "epoch-1", seq: 5 },
|
|
endCursor: { epoch: "epoch-1", seq: 5 },
|
|
entries: [
|
|
{
|
|
provider: "mock",
|
|
item: { type: "assistant_message", text: complete },
|
|
timestamp: "2026-08-26T10:00:00.000Z",
|
|
seqStart: 2,
|
|
seqEnd: 5,
|
|
sourceSeqRanges: [{ startSeq: 2, endSeq: 5 }],
|
|
collapsed: ["assistant_merge"],
|
|
},
|
|
],
|
|
error: null,
|
|
hasNewer: false,
|
|
hasOlder: false,
|
|
staleCursor: false,
|
|
gap: false,
|
|
});
|
|
|
|
const session = useSessionStore.getState().sessions[SERVER_ID];
|
|
expect([
|
|
...(session?.agentStreamTail.get(AGENT_ID) ?? []),
|
|
...(session?.agentStreamHead.get(AGENT_ID) ?? []),
|
|
]).toMatchObject([{ kind: "assistant_message", text: complete }]);
|
|
owner.dispose();
|
|
});
|
|
|
|
it("reopens painted live mutations without persisting authoritative coverage", async () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
let durable: CachedTimeline | undefined = cachedTimeline();
|
|
let pending: CachedTimeline | undefined;
|
|
const storage: TimelineReplicaStorage = {
|
|
readTimeline: async () => durable,
|
|
commitTimeline: (_serverId, _agentId, timeline) => {
|
|
pending = timeline;
|
|
},
|
|
};
|
|
const first = createOwner(storage);
|
|
first.replaceVisibleAgentIds("test", [AGENT_ID]);
|
|
await expect
|
|
.poll(() =>
|
|
selectAgentTimelineState(useSessionStore.getState().sessions[SERVER_ID], AGENT_ID),
|
|
)
|
|
.toMatchObject({ status: "painted" });
|
|
|
|
first.enqueueStreamEvent(AGENT_ID, {
|
|
event: {
|
|
type: "timeline",
|
|
provider: "codex",
|
|
item: { type: "assistant_message", text: "live", messageId: "live" },
|
|
} as AgentStreamEventPayload,
|
|
seq: 5,
|
|
epoch: "epoch-1",
|
|
timestamp: new Date("2026-08-26T10:00:01.000Z"),
|
|
});
|
|
first.flushStreamAgent(AGENT_ID);
|
|
|
|
expect(pending).toMatchObject({ range: null, hasOlder: false });
|
|
expect(pending?.items.map((entry) => entry.id)).toEqual(["cached", expect.any(String)]);
|
|
durable = pending;
|
|
first.dispose();
|
|
useSessionStore.getState().clearSession(SERVER_ID);
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
|
|
const reopened = createOwner(storage);
|
|
reopened.replaceVisibleAgentIds("test", [AGENT_ID]);
|
|
await expect
|
|
.poll(() =>
|
|
selectAgentTimelineState(useSessionStore.getState().sessions[SERVER_ID], AGENT_ID),
|
|
)
|
|
.toMatchObject({
|
|
status: "painted",
|
|
items: [
|
|
expect.objectContaining({ id: "cached" }),
|
|
expect.objectContaining({ text: "live" }),
|
|
],
|
|
});
|
|
reopened.dispose();
|
|
});
|
|
|
|
it("does not let a late cache read overwrite newer network state", async () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
let release!: (value: CachedTimeline) => void;
|
|
const read = new Promise<CachedTimeline>((resolve) => {
|
|
release = resolve;
|
|
});
|
|
const owner = createOwner({
|
|
readTimeline: () => read,
|
|
commitTimeline: () => undefined,
|
|
});
|
|
|
|
owner.replaceVisibleAgentIds("test", [AGENT_ID]);
|
|
applySynced(AGENT_ID, 8);
|
|
release(cachedTimeline());
|
|
|
|
await expect
|
|
.poll(() =>
|
|
selectAgentTimelineState(useSessionStore.getState().sessions[SERVER_ID], AGENT_ID),
|
|
)
|
|
.toMatchObject({ status: "synced", range: { endSeq: 8 } });
|
|
owner.dispose();
|
|
});
|
|
|
|
it("paints cached rows without replacing a live head that arrives during preparation", async () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
let release!: (value: CachedTimeline) => void;
|
|
const read = new Promise<CachedTimeline>((resolve) => {
|
|
release = resolve;
|
|
});
|
|
const replica = createTimelineReplica({
|
|
serverId: SERVER_ID,
|
|
storage: {
|
|
readTimeline: () => read,
|
|
commitTimeline: () => undefined,
|
|
},
|
|
prepareAgent: async () => undefined,
|
|
});
|
|
|
|
const preparation = replica.prepare(AGENT_ID);
|
|
useSessionStore.getState().setAgentStreamState(SERVER_ID, AGENT_ID, {
|
|
head: [item("live", "live", 5)],
|
|
});
|
|
release(cachedTimeline());
|
|
await preparation;
|
|
|
|
expect(replica.readCursor(AGENT_ID)).toEqual({ epoch: "epoch-1", endSeq: 4 });
|
|
expect(
|
|
selectAgentTimelineState(useSessionStore.getState().sessions[SERVER_ID], AGENT_ID),
|
|
).toEqual({ status: "painted", items: cachedTimeline().items });
|
|
expect(useSessionStore.getState().sessions[SERVER_ID]?.agentStreamHead.get(AGENT_ID)).toEqual([
|
|
item("live", "live", 5),
|
|
]);
|
|
});
|
|
|
|
it("reconciles a live head that overlaps the cached canonical timeline", async () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
let release!: (value: CachedTimeline) => void;
|
|
const read = new Promise<CachedTimeline>((resolve) => {
|
|
release = resolve;
|
|
});
|
|
const replica = createTimelineReplica({
|
|
serverId: SERVER_ID,
|
|
storage: {
|
|
readTimeline: () => read,
|
|
commitTimeline: () => undefined,
|
|
},
|
|
prepareAgent: async () => undefined,
|
|
});
|
|
|
|
const preparation = replica.prepare(AGENT_ID);
|
|
useSessionStore.getState().setAgentStreamState(SERVER_ID, AGENT_ID, {
|
|
head: [item("cached", "cached", 4), item("live", "live", 5)],
|
|
});
|
|
release(cachedTimeline());
|
|
await preparation;
|
|
|
|
const session = useSessionStore.getState().sessions[SERVER_ID];
|
|
expect([
|
|
...(session?.agentStreamTail.get(AGENT_ID) ?? []),
|
|
...(session?.agentStreamHead.get(AGENT_ID) ?? []),
|
|
]).toEqual([item("cached", "cached", 4), item("live", "live", 5)]);
|
|
expect(replica.readCursor(AGENT_ID)).toEqual({ epoch: "epoch-1", endSeq: 4 });
|
|
});
|
|
|
|
it("reconciles cached rows with a non-authoritative timeline painted during preparation", async () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
let release!: (value: CachedTimeline) => void;
|
|
const read = new Promise<CachedTimeline>((resolve) => {
|
|
release = resolve;
|
|
});
|
|
const replica = createTimelineReplica({
|
|
serverId: SERVER_ID,
|
|
storage: {
|
|
readTimeline: () => read,
|
|
commitTimeline: () => undefined,
|
|
},
|
|
prepareAgent: async () => undefined,
|
|
});
|
|
|
|
const preparation = replica.prepare(AGENT_ID);
|
|
useSessionStore.getState().applyAgentTimelineResponseState(SERVER_ID, AGENT_ID, {
|
|
items: [item("live", "live", 5)],
|
|
head: [],
|
|
range: null,
|
|
older: "none",
|
|
newer: false,
|
|
synchronized: false,
|
|
acknowledgedClientMessageIds: [],
|
|
});
|
|
release(cachedTimeline());
|
|
await preparation;
|
|
|
|
const session = useSessionStore.getState().sessions[SERVER_ID];
|
|
expect([
|
|
...(session?.agentStreamTail.get(AGENT_ID) ?? []),
|
|
...(session?.agentStreamHead.get(AGENT_ID) ?? []),
|
|
]).toEqual([item("cached", "cached", 4), item("live", "live", 5)]);
|
|
expect(replica.readCursor(AGENT_ID)).toEqual({ epoch: "epoch-1", endSeq: 4 });
|
|
});
|
|
|
|
it("persists accepted live stream commits through the owner", () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
applySynced(AGENT_ID, 8);
|
|
const commits: CachedTimeline[] = [];
|
|
const owner = createOwner({
|
|
readTimeline: async () => undefined,
|
|
commitTimeline: (_serverId, _agentId, timeline) => commits.push(timeline),
|
|
});
|
|
owner.enqueueStreamEvent(AGENT_ID, {
|
|
event: {
|
|
type: "timeline",
|
|
provider: "codex",
|
|
item: { type: "assistant_message", text: "live", messageId: "live" },
|
|
} as AgentStreamEventPayload,
|
|
seq: 9,
|
|
epoch: "epoch-1",
|
|
timestamp: new Date("2026-08-26T10:00:01.000Z"),
|
|
});
|
|
owner.flushStreamAgent(AGENT_ID);
|
|
|
|
expect(commits.at(-1)?.items.at(-1)).toMatchObject({ text: "live" });
|
|
expect(commits.at(-1)?.range?.endSeq).toBe(9);
|
|
owner.dispose();
|
|
});
|
|
|
|
it("applies and persists authoritative pages inside the owner", () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
const keys: string[] = [];
|
|
const owner = createOwner({
|
|
readTimeline: async () => undefined,
|
|
commitTimeline: (_serverId, agentId) => keys.push(agentId),
|
|
});
|
|
|
|
owner.applyTimelineResponse({
|
|
requestId: "page-1",
|
|
agentId: AGENT_ID,
|
|
agent: null,
|
|
direction: "tail",
|
|
projection: "projected",
|
|
reset: false,
|
|
epoch: "epoch-1",
|
|
window: { minSeq: 1, maxSeq: 0, nextSeq: 1 },
|
|
startCursor: null,
|
|
endCursor: null,
|
|
entries: [],
|
|
error: null,
|
|
hasNewer: false,
|
|
hasOlder: false,
|
|
staleCursor: false,
|
|
gap: false,
|
|
});
|
|
|
|
expect(
|
|
selectAgentTimelineState(useSessionStore.getState().sessions[SERVER_ID], AGENT_ID),
|
|
).toMatchObject({ status: "synced" });
|
|
expect(keys).toEqual([AGENT_ID]);
|
|
owner.dispose();
|
|
});
|
|
|
|
it("persists demanded agents independently", () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
const keys: string[] = [];
|
|
const owner = createOwner({
|
|
readTimeline: async () => undefined,
|
|
commitTimeline: (_serverId, agentId) => keys.push(agentId),
|
|
});
|
|
|
|
applySynced(AGENT_ID, 8);
|
|
applySynced("agent-2", 3);
|
|
for (const [agentId, seq] of [
|
|
[AGENT_ID, 9],
|
|
["agent-2", 4],
|
|
] as const) {
|
|
owner.enqueueStreamEvent(agentId, {
|
|
event: {
|
|
type: "timeline",
|
|
provider: "codex",
|
|
item: { type: "assistant_message", text: agentId, messageId: `live-${agentId}` },
|
|
} as AgentStreamEventPayload,
|
|
seq,
|
|
epoch: "epoch-1",
|
|
timestamp: new Date("2026-08-26T10:00:01.000Z"),
|
|
});
|
|
owner.flushStreamAgent(agentId);
|
|
}
|
|
|
|
expect(keys).toEqual([AGENT_ID, "agent-2"]);
|
|
owner.dispose();
|
|
});
|
|
});
|
|
|
|
describe("create handoff lifetime", () => {
|
|
const DRAFT_ID = "draft-1";
|
|
|
|
afterEach(() => useCreateFlowStore.getState().clearAll());
|
|
|
|
function sentCreateHandoff(): void {
|
|
useCreateFlowStore.getState().setPending({
|
|
draftId: DRAFT_ID,
|
|
serverId: SERVER_ID,
|
|
agentId: AGENT_ID,
|
|
clientMessageId: "client-message-1",
|
|
text: "Create this chat.",
|
|
timestamp: 0,
|
|
});
|
|
useCreateFlowStore.getState().markLifecycle({ draftId: DRAFT_ID, lifecycle: "sent" });
|
|
}
|
|
|
|
function createHandoff(): PendingCreateAttempt | undefined {
|
|
return useCreateFlowStore.getState().pendingByDraftId[DRAFT_ID];
|
|
}
|
|
|
|
function syncedTailPage(): Parameters<ViewedTimelineOwner["applyTimelineResponse"]>[0] {
|
|
return {
|
|
requestId: "tail-1",
|
|
agentId: AGENT_ID,
|
|
agent: null,
|
|
direction: "tail",
|
|
projection: "projected",
|
|
reset: false,
|
|
epoch: "epoch-1",
|
|
window: { minSeq: 1, maxSeq: 0, nextSeq: 1 },
|
|
startCursor: null,
|
|
endCursor: null,
|
|
entries: [],
|
|
error: null,
|
|
hasNewer: false,
|
|
hasOlder: false,
|
|
staleCursor: false,
|
|
gap: false,
|
|
};
|
|
}
|
|
|
|
interface HeldCatchUpOwner {
|
|
owner: ViewedTimelineOwner;
|
|
release: () => void;
|
|
}
|
|
|
|
async function ownerWithHeldCatchUp(): Promise<HeldCatchUpOwner> {
|
|
let releaseTailFetch!: () => void;
|
|
let tailFetchStarted = false;
|
|
const tailFetched = new Promise<void>((resolve) => {
|
|
releaseTailFetch = resolve;
|
|
});
|
|
const replica = createTimelineReplica({
|
|
serverId: SERVER_ID,
|
|
storage: { readTimeline: async () => undefined, commitTimeline: () => undefined },
|
|
prepareAgent: async () => undefined,
|
|
});
|
|
const owner = createViewedTimelineOwner({
|
|
serverId: SERVER_ID,
|
|
replica,
|
|
replaceDemandedAgentIds: () => undefined,
|
|
drainQueuedAgentMessage: () => undefined,
|
|
ports: {
|
|
observe: () => ({ ready: Promise.resolve(), release: async () => undefined }),
|
|
readCursor: () => undefined,
|
|
fetchPage: async () => {
|
|
tailFetchStarted = true;
|
|
await tailFetched;
|
|
return { hasNewer: false, endCursor: null };
|
|
},
|
|
fetchLatestTail: async () => ({ hasNewer: false, endCursor: null }),
|
|
reportError: () => undefined,
|
|
schedule: () => () => undefined,
|
|
},
|
|
});
|
|
|
|
owner.setConnected(true);
|
|
owner.replaceVisibleAgentIds("test", [AGENT_ID]);
|
|
await expect.poll(() => tailFetchStarted).toBe(true);
|
|
return { owner, release: releaseTailFetch };
|
|
}
|
|
|
|
it("keeps the handoff through the first synchronized page and releases it when the chat is current", async () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
sentCreateHandoff();
|
|
const { owner, release } = await ownerWithHeldCatchUp();
|
|
|
|
owner.applyTimelineResponse(syncedTailPage());
|
|
expect(
|
|
selectAgentTimelineState(useSessionStore.getState().sessions[SERVER_ID], AGENT_ID),
|
|
).toMatchObject({ status: "synced" });
|
|
expect(createHandoff()).toMatchObject({ agentId: AGENT_ID, lifecycle: "sent" });
|
|
|
|
release();
|
|
await expect.poll(() => createHandoff()).toBeUndefined();
|
|
owner.dispose();
|
|
});
|
|
|
|
it("releases the handoff when the chat's tab closes before catch-up completes", async () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
sentCreateHandoff();
|
|
const { owner, release } = await ownerWithHeldCatchUp();
|
|
|
|
owner.replaceVisibleAgentIds("test", []);
|
|
owner.replaceOpenTabAgentIds([]);
|
|
|
|
expect(createHandoff()).toBeUndefined();
|
|
release();
|
|
owner.dispose();
|
|
});
|
|
});
|
|
|
|
function createSqliteCache() {
|
|
const database = new DatabaseSync(":memory:");
|
|
let beforeRead = async () => {};
|
|
const connection: ReplicaSqliteConnection = {
|
|
async exec(sql) {
|
|
database.exec(sql);
|
|
},
|
|
async run(sql, params = []) {
|
|
database.prepare(sql).run(...params);
|
|
},
|
|
async all<Row>(sql: string, params: readonly SqliteValue[] = []) {
|
|
const rows = database.prepare(sql).all(...params) as Row[];
|
|
if (sql.includes("FROM rows") && sql.includes("WHERE")) await beforeRead();
|
|
return rows;
|
|
},
|
|
async transaction(operation) {
|
|
database.exec("BEGIN IMMEDIATE");
|
|
try {
|
|
await operation(connection);
|
|
database.exec("COMMIT");
|
|
} catch (error) {
|
|
database.exec("ROLLBACK");
|
|
throw error;
|
|
}
|
|
},
|
|
};
|
|
const storage = createSqliteReplicaRowStore({ open: async () => connection }, 1);
|
|
const cache = new ReplicaCache(storage, { clearLegacyCache: async () => {} });
|
|
return {
|
|
cache,
|
|
database,
|
|
holdRead: (operation: () => Promise<void>) => {
|
|
beforeRead = operation;
|
|
},
|
|
};
|
|
}
|
|
|
|
describe("SQLite baseline restoration", () => {
|
|
it.each([false, true])(
|
|
"keeps display-only cached history under a live event (synced=%s)",
|
|
async (synced) => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
const { cache, database, holdRead } = createSqliteCache();
|
|
cache.setHosts([SERVER_ID]);
|
|
cache.commitTimeline(SERVER_ID, AGENT_ID, {
|
|
...cachedTimeline(),
|
|
items: [item("earlier", "earlier", 1), ...cachedTimeline().items],
|
|
range: null,
|
|
});
|
|
await cache.flush();
|
|
holdRead(async () => {
|
|
if (synced) applySynced(AGENT_ID, 8);
|
|
else
|
|
useSessionStore.getState().setAgentStreamState(SERVER_ID, AGENT_ID, {
|
|
head: [item("cached", "cached", 4), item("live", "live", 5)],
|
|
});
|
|
});
|
|
const replica = createTimelineReplica({
|
|
serverId: SERVER_ID,
|
|
storage: cache,
|
|
prepareAgent: async () => {},
|
|
});
|
|
await replica.prepare(AGENT_ID);
|
|
const session = useSessionStore.getState().sessions[SERVER_ID];
|
|
const timeline = selectAgentTimelineState(session, AGENT_ID);
|
|
expect(timeline.status).toBe(synced ? "synced" : "painted");
|
|
expect([
|
|
...(timeline.status === "cold" ? [] : timeline.items),
|
|
...(session?.agentStreamHead.get(AGENT_ID) ?? []),
|
|
]).toEqual(
|
|
synced
|
|
? [item(`network-${AGENT_ID}`, "network", 8)]
|
|
: [item("earlier", "earlier", 1), item("cached", "cached", 4), item("live", "live", 5)],
|
|
);
|
|
expect(replica.readCursor(AGENT_ID)).toBeUndefined();
|
|
database.close();
|
|
},
|
|
);
|
|
|
|
it("starts the timeline disk read while agent preparation is pending", async () => {
|
|
useSessionStore.getState().initializeSession(SERVER_ID, null);
|
|
const { cache, database, holdRead } = createSqliteCache();
|
|
cache.setHosts([SERVER_ID]);
|
|
cache.commitTimeline(SERVER_ID, AGENT_ID, cachedTimeline());
|
|
await cache.flush();
|
|
let readStarted = false;
|
|
let release!: () => void;
|
|
const agentReady = new Promise<void>((resolve) => {
|
|
release = resolve;
|
|
});
|
|
holdRead(async () => {
|
|
readStarted = true;
|
|
});
|
|
const replica = createTimelineReplica({
|
|
serverId: SERVER_ID,
|
|
storage: cache,
|
|
prepareAgent: () => agentReady,
|
|
});
|
|
const preparation = replica.prepare(AGENT_ID);
|
|
try {
|
|
await expect.poll(() => readStarted, { timeout: 500 }).toBe(true);
|
|
} finally {
|
|
release();
|
|
await preparation;
|
|
database.close();
|
|
}
|
|
});
|
|
});
|