mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 09:41:04 +01:00
fix(runner): keep live cursors stable after compaction
Use durable PRP source sequences for browser replay and reconnect cursors so trimming the retained event window cannot skip unseen events. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
2fa2ea60da
commit
fdc9bf4936
4 files changed
+47
-10
No files matched your search
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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<T> implements AsyncIterable<T> {
|
||||
#values: T[] = [];
|
||||
#waiters: Array<(value: IteratorResult<T>) => void> = [];
|
||||
|
||||
@@ -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<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value)
|
||||
? (value as Record<string, unknown>)
|
||||
@@ -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,
|
||||
};
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user