diff --git a/packages/paperclip-runner/devtools/browser/src/live/use-live-console.test.ts b/packages/paperclip-runner/devtools/browser/src/live/use-live-console.test.ts index 43d7b46fac2..8542950c832 100644 --- a/packages/paperclip-runner/devtools/browser/src/live/use-live-console.test.ts +++ b/packages/paperclip-runner/devtools/browser/src/live/use-live-console.test.ts @@ -6,14 +6,14 @@ import { createLiveEventCursor, } from "./use-live-console"; -function event(sourceEventId: string): PrpEvent { - return { sourceEventId } as PrpEvent; +function event(sourceEventId: string, sourceSeq: number): PrpEvent { + return { sourceEventId, sourceSeq } as PrpEvent; } describe("Live console reconnect cursor", () => { it("does not advance past an unseen event when replay delivers a duplicate", () => { - const first = event("source:event-1"); - const second = event("source:event-2"); + const first = event("source:event-1", 1); + const second = event("source:event-2", 2); const cursor = createLiveEventCursor([first], 1); expect(acceptLiveEvent(cursor, first)).toBe(false); @@ -22,4 +22,12 @@ describe("Live console reconnect cursor", () => { expect(acceptLiveEvent(cursor, second)).toBe(true); expect(cursor.cursor).toBe(2); }); + + it("uses the durable source sequence after the server compacts retained events", () => { + const retained = event("source:event-4097", 4097); + const cursor = createLiveEventCursor([], 4096); + + expect(acceptLiveEvent(cursor, retained)).toBe(true); + expect(cursor.cursor).toBe(4097); + }); }); diff --git a/packages/paperclip-runner/devtools/browser/src/live/use-live-console.ts b/packages/paperclip-runner/devtools/browser/src/live/use-live-console.ts index a2c6a903a9d..4d90f21916b 100644 --- a/packages/paperclip-runner/devtools/browser/src/live/use-live-console.ts +++ b/packages/paperclip-runner/devtools/browser/src/live/use-live-console.ts @@ -42,7 +42,7 @@ export function createLiveEventCursor( export function acceptLiveEvent(cursor: LiveEventCursor, event: PrpEvent): boolean { if (cursor.seenSourceEventIds.has(event.sourceEventId)) return false; cursor.seenSourceEventIds.add(event.sourceEventId); - cursor.cursor += 1; + cursor.cursor = Math.max(cursor.cursor, event.sourceSeq); return true; } diff --git a/packages/paperclip-runner/src/mock-core/live-console-demo-server.test.ts b/packages/paperclip-runner/src/mock-core/live-console-demo-server.test.ts index 391c770c1c6..54b30cc5481 100644 --- a/packages/paperclip-runner/src/mock-core/live-console-demo-server.test.ts +++ b/packages/paperclip-runner/src/mock-core/live-console-demo-server.test.ts @@ -14,7 +14,11 @@ import type { PersistedHarnessSession, } from "../contracts/harness-driver.js"; import type { PrpEvent } from "../protocol/replay-contract.js"; -import { LiveConsoleDemoServer } from "./live-console-demo-server.js"; +import { + liveEventCursor, + liveEventsAfterCursor, + LiveConsoleDemoServer, +} from "./live-console-demo-server.js"; const OPAQUE_BROWSER_CREDENTIALS = { clientSecret: "opaque-client-secret-value", @@ -26,6 +30,20 @@ const OPAQUE_BROWSER_CREDENTIALS = { cookie: "opaque-cookie-value", } as const; +describe("live console durable cursors", () => { + it("replays unseen retained events after positional compaction", () => { + const retained = [ + { sourceEventId: "event-4095", sourceSeq: 4095 }, + { sourceEventId: "event-4096", sourceSeq: 4096 }, + { sourceEventId: "event-4097", sourceSeq: 4097 }, + ] as PrpEvent[]; + + expect(liveEventsAfterCursor(retained, 4095).map((event) => event.sourceSeq)) + .toEqual([4096, 4097]); + expect(liveEventCursor(retained)).toBe(4097); + }); +}); + class Queue implements AsyncIterable { #values: T[] = []; #waiters: Array<(value: IteratorResult) => void> = []; diff --git a/packages/paperclip-runner/src/mock-core/live-console-demo-server.ts b/packages/paperclip-runner/src/mock-core/live-console-demo-server.ts index 28035f52a1f..71cd3909e4d 100644 --- a/packages/paperclip-runner/src/mock-core/live-console-demo-server.ts +++ b/packages/paperclip-runner/src/mock-core/live-console-demo-server.ts @@ -91,6 +91,17 @@ interface BrowserCreateBody { startTurn?: boolean; } +export function liveEventsAfterCursor( + events: readonly PrpEvent[], + after: number, +): PrpEvent[] { + return events.filter((event) => event.sourceSeq > after); +} + +export function liveEventCursor(events: readonly PrpEvent[]): number { + return events.at(-1)?.sourceSeq ?? 0; +} + function record(value: unknown): Record { return typeof value === "object" && value !== null && !Array.isArray(value) ? (value as Record) @@ -653,8 +664,8 @@ export class LiveConsoleDemoServer { const url = new URL(request.url ?? "/", "http://liveConsole.invalid"); const after = Math.max(0, Number.parseInt(url.searchParams.get("after") ?? "0", 10) || 0); json(response, 200, { - events: entry.events.slice(after), - cursor: entry.events.length, + events: liveEventsAfterCursor(entry.events, after), + cursor: liveEventCursor(entry.events), replay: after === 0, }); return; @@ -678,7 +689,7 @@ export class LiveConsoleDemoServer { // last event, and Node would otherwise hold the response open with no // bytes sent, leaving the browser stuck in CONNECTING. response.write(": stream open\n\n"); - for (const event of entry.events.slice(after)) { + for (const event of liveEventsAfterCursor(entry.events, after)) { const output = serializeBrowserJson(event); if (!output.overflow) response.write(`data: ${output.serialized}\n\n`); } @@ -882,7 +893,7 @@ export class LiveConsoleDemoServer { pendingRequests: entry.session.pendingRuntimeRequests?.() ?? [], goal: harnessSnapshot.goal ?? null, lineage: entry.session.lineage?.() ?? [], - cursor: entry.events.length, + cursor: liveEventCursor(entry.events), snapshot, }; }