History

Add durable provider event catch-up by sequence

- Persist provider event `seq` metadata and return persisted events from ingestion - Add `providers.catchUp` WS/API contract and server persistence query - Replay missing provider events on web reconnect using ordered sequence catch-up - Extend server/contracts tests for sequence ordering and catch-up behavior

Julius Marminge committed 8 months ago

Commit f8aadcf1 parent b9bb89c13 files · +400 −22Browse files

Unified · Split

13 files
  1. apps/server/src/persistenceService.test.ts +26 −4
  2. apps/server/src/persistenceService.ts +105 −3
  3. apps/server/src/providerManager.test.ts +52 −0
  4. apps/server/src/providerManager.ts +26 −13
  5. apps/server/src/wsServer.test.ts +52 −0
  6. apps/server/src/wsServer.ts +3 −0
  7. apps/web/src/routes/__root.tsx +72 −2
  8. apps/web/src/wsNativeApi.ts +1 −0
  9. packages/contracts/src/ipc.ts +3 −0
  10. packages/contracts/src/provider.test.ts +42 −0
  11. packages/contracts/src/provider.ts +13 −0
  12. packages/contracts/src/ws.test.ts +4 −0
  13. packages/contracts/src/ws.ts +1 −0
apps/server/src/persistenceService.test.ts+26−4
@@ -102,7 +102,7 @@ describe("PersistenceService", () => { · 101 unmodified lines
102 sessionId: "sess-1",102 sessionId: "sess-1",
103 };103 };
104 104
105 service.ingestProviderEvent({105 const event1 = service.ingestProviderEvent({
106 ...baseEvent,106 ...baseEvent,
107 id: "evt-1",107 id: "evt-1",
108 createdAt: iso(10),108 createdAt: iso(10),
@@ -115,8 +115,9 @@ describe("PersistenceService", () => { · 6 unmodified lines
115 },115 },
116 },116 },
117 });117 });
118 expect(event1?.seq).toBeGreaterThan(0);
118 119
119 service.ingestProviderEvent({120 const event2 = service.ingestProviderEvent({
120 ...baseEvent,121 ...baseEvent,
121 id: "evt-2",122 id: "evt-2",
122 createdAt: iso(20),123 createdAt: iso(20),
@@ -128,8 +129,9 @@ describe("PersistenceService", () => { · 5 unmodified lines
128 delta: "hi",129 delta: "hi",
129 },130 },
130 });131 });
132 expect((event2?.seq ?? 0) > (event1?.seq ?? 0)).toBe(true);
131 133
132 service.ingestProviderEvent({134 const event3 = service.ingestProviderEvent({
133 ...baseEvent,135 ...baseEvent,
134 id: "evt-3",136 id: "evt-3",
135 createdAt: iso(30),137 createdAt: iso(30),
@@ -142,8 +144,9 @@ describe("PersistenceService", () => { · 6 unmodified lines
142 },144 },
143 },145 },
144 });146 });
147 expect((event3?.seq ?? 0) > (event2?.seq ?? 0)).toBe(true);
145 148
146 service.ingestProviderEvent({149 const event4 = service.ingestProviderEvent({
147 ...baseEvent,150 ...baseEvent,
148 id: "evt-4",151 id: "evt-4",
149 createdAt: iso(40),152 createdAt: iso(40),
@@ -156,6 +159,7 @@ describe("PersistenceService", () => { · 6 unmodified lines
156 },159 },
157 },160 },
158 });161 });
162 expect((event4?.seq ?? 0) > (event3?.seq ?? 0)).toBe(true);
159 163
160 const snapshot = service.loadSnapshot();164 const snapshot = service.loadSnapshot();
161 expect(snapshot.projects).toHaveLength(1);165 expect(snapshot.projects).toHaveLength(1);
@@ -179,6 +183,24 @@ describe("PersistenceService", () => { · 17 unmodified lines
179 expect(catchUp.events.some((event) => event.eventType === "project.upsert")).toBe(true);183 expect(catchUp.events.some((event) => event.eventType === "project.upsert")).toBe(true);
180 expect(catchUp.events.some((event) => event.eventType === "thread.upsert")).toBe(true);184 expect(catchUp.events.some((event) => event.eventType === "thread.upsert")).toBe(true);
181 expect(catchUp.events.some((event) => event.eventType === "message.upsert")).toBe(true);185 expect(catchUp.events.some((event) => event.eventType === "message.upsert")).toBe(true);
186
187 const providerCatchUp = service.providerCatchUp({ afterSeq: 0 });
188 expect(providerCatchUp.lastProviderSeq).toBe(event4?.seq ?? 0);
189 expect(providerCatchUp.events.map((event) => event.id)).toEqual([
190 "evt-1",
191 "evt-2",
192 "evt-3",
193 "evt-4",
194 ]);
195 const providerSeqs = providerCatchUp.events.map((event) => event.seq ?? 0);
196 for (let index = 1; index < providerSeqs.length; index += 1) {
197 expect(providerSeqs[index]).toBeGreaterThan(providerSeqs[index - 1] ?? Number.NEGATIVE_INFINITY);
198 }
199
200 const providerCatchUpAfterSecond = service.providerCatchUp({
201 afterSeq: event2?.seq ?? 0,
202 });
203 expect(providerCatchUpAfterSecond.events.map((event) => event.id)).toEqual(["evt-3", "evt-4"]);
182 } finally {204 } finally {
183 service.close();205 service.close();
184 }206 }
apps/server/src/persistenceService.ts+105−3
@@ -16,6 +16,8 @@ import { · 15 unmodified lines
16 type ProjectRemoveInput,16 type ProjectRemoveInput,
17 type ProjectUpdateScriptsInput,17 type ProjectUpdateScriptsInput,
18 type ProjectUpdateScriptsResult,18 type ProjectUpdateScriptsResult,
19 type ProviderCatchUpInput,
20 type ProviderCatchUpResult,
19 type ProviderEvent,21 type ProviderEvent,
20 type ProviderSendTurnInput,22 type ProviderSendTurnInput,
21 type StateBootstrapResult,23 type StateBootstrapResult,
@@ -44,6 +46,9 @@ import { · 22 unmodified lines
44 projectRemoveInputSchema,46 projectRemoveInputSchema,
45 projectScriptsSchema,47 projectScriptsSchema,
46 projectUpdateScriptsInputSchema,48 projectUpdateScriptsInputSchema,
49 providerCatchUpInputSchema,
50 providerCatchUpResultSchema,
51 providerEventSchema,
47 stateBootstrapResultSchema,52 stateBootstrapResultSchema,
48 stateCatchUpInputSchema,53 stateCatchUpInputSchema,
49 stateCatchUpResultSchema,54 stateCatchUpResultSchema,
@@ -94,6 +99,25 @@ interface StateEventRow { · 44 unmodified lines
94interface ProviderEventInsertResult {99interface ProviderEventInsertResult {
95 inserted: boolean;100 inserted: boolean;
96 runtimeThreadId: string | null;101 runtimeThreadId: string | null;
102 seq: number;
103}
104
105interface ProviderEventRow {
106 seq: number;
107 id: string;
108 session_id: string;
109 provider: string;
110 kind: string;
111 method: string;
112 thread_id: string | null;
113 turn_id: string | null;
114 item_id: string | null;
115 request_id: string | null;
116 request_kind: string | null;
117 text_delta: string | null;
118 message: string | null;
119 payload_json: string | null;
120 created_at: string;
97}121}
98 122
99export interface PersistenceServiceOptions {123export interface PersistenceServiceOptions {
@@ -772,6 +796,43 @@ export class PersistenceService extends EventEmitter<PersistenceServiceEvents> { · 672 unmodified lines
772 });796 });
773 }797 }
774 798
799 providerCatchUp(raw: ProviderCatchUpInput): ProviderCatchUpResult {
800 const input = providerCatchUpInputSchema.parse(raw);
801 const rows = this.db
802 .prepare(
803 `SELECT
804 seq,
805 id,
806 session_id,
807 provider,
808 kind,
809 method,
810 thread_id,
811 turn_id,
812 item_id,
813 request_id,
814 request_kind,
815 text_delta,
816 message,
817 payload_json,
818 created_at
819 FROM provider_events
820 WHERE seq > ?
821 ORDER BY seq ASC
822 LIMIT ?;`,
823 )
824 .all(input.afterSeq, input.limit) as unknown as ProviderEventRow[];
825
826 const events = rows
827 .map((row) => this.parseProviderEventRow(row))
828 .filter((event): event is ProviderEvent => event !== null);
829
830 return providerCatchUpResultSchema.parse({
831 events,
832 lastProviderSeq: this.readLastProviderSeq(),
833 });
834 }
835
775 listMessages(raw: StateListMessagesInput): StateListMessagesResult {836 listMessages(raw: StateListMessagesInput): StateListMessagesResult {
776 const input = stateListMessagesInputSchema.parse(raw);837 const input = stateListMessagesInputSchema.parse(raw);
777 const rows = this.db838 const rows = this.db
@@ -867,12 +928,18 @@ export class PersistenceService extends EventEmitter<PersistenceServiceEvents> { · 89 unmodified lines
867 });928 });
868 }929 }
869 930
870 ingestProviderEvent(event: ProviderEvent): void {931 ingestProviderEvent(event: ProviderEvent): ProviderEvent | null {
932 let persistedEvent: ProviderEvent | null = null;
871 this.withTransaction((pendingEvents) => {933 this.withTransaction((pendingEvents) => {
872 const insertResult = this.insertProviderEvent(event);934 const insertResult = this.insertProviderEvent(event);
873 if (!insertResult.inserted) {935 if (!insertResult.inserted) {
874 return;936 return;
875 }937 }
938 persistedEvent = providerEventSchema.parse({
939 ...event,
940 seq: insertResult.seq,
941 ...(insertResult.runtimeThreadId ? { threadId: insertResult.runtimeThreadId } : {}),
942 });
876 943
877 const localThreadId = this.resolveThreadIdForEvent(event, insertResult.runtimeThreadId);944 const localThreadId = this.resolveThreadIdForEvent(event, insertResult.runtimeThreadId);
878 if (!localThreadId) {945 if (!localThreadId) {
@@ -1138,6 +1205,7 @@ export class PersistenceService extends EventEmitter<PersistenceServiceEvents> { · 259 unmodified lines
1138 }1205 }
1139 }1206 }
1140 });1207 });
1208 return persistedEvent;
1141 }1209 }
1142 1210
1143 persistTurnDiffSummaryFromCheckpoint(input: {1211 persistTurnDiffSummaryFromCheckpoint(input: {
@@ -1360,6 +1428,13 @@ export class PersistenceService extends EventEmitter<PersistenceServiceEvents> { · 216 unmodified lines
1360 return row?.seq ?? 0;1428 return row?.seq ?? 0;
1361 }1429 }
1362 1430
1431 private readLastProviderSeq(): number {
1432 const row = this.db
1433 .prepare("SELECT COALESCE(MAX(seq), 0) AS seq FROM provider_events;")
1434 .get() as { seq: number } | undefined;
1435 return row?.seq ?? 0;
1436 }
1437
1363 private withTransaction<T>(fn: (pendingEvents: StateEvent[]) => T): T {1438 private withTransaction<T>(fn: (pendingEvents: StateEvent[]) => T): T {
1364 const pendingEvents: StateEvent[] = [];1439 const pendingEvents: StateEvent[] = [];
1365 const result = this.stateDb.transaction(() => fn(pendingEvents));1440 const result = this.stateDb.transaction(() => fn(pendingEvents));
@@ -1766,13 +1841,40 @@ export class PersistenceService extends EventEmitter<PersistenceServiceEvents> { · 400 unmodified lines
1766 event.message ?? null,1841 event.message ?? null,
1767 payloadJson,1842 payloadJson,
1768 event.createdAt,1843 event.createdAt,
1769 ) as { changes?: number | bigint };1844 ) as { changes?: number | bigint; lastInsertRowid?: number | bigint };
1845 const inserted = toSafeInteger(result.changes, 0) > 0;
1770 return {1846 return {
1771 inserted: toSafeInteger(result.changes, 0) > 0,1847 inserted,
1772 runtimeThreadId: runtimeThreadId ?? null,1848 runtimeThreadId: runtimeThreadId ?? null,
1849 seq: inserted ? toSafeInteger(result.lastInsertRowid, 0) : 0,
1773 };1850 };
1774 }1851 }
1775 1852
1853 private parseProviderEventRow(row: ProviderEventRow): ProviderEvent | null {
1854 const payload = row.payload_json ? this.tryParseJson(row.payload_json) : undefined;
1855 const parsed = providerEventSchema.safeParse({
1856 seq: row.seq,
1857 id: row.id,
1858 kind: row.kind,
1859 provider: row.provider,
1860 sessionId: row.session_id,
1861 createdAt: row.created_at,
1862 method: row.method,
1863 ...(row.thread_id ? { threadId: row.thread_id } : {}),
1864 ...(row.turn_id ? { turnId: row.turn_id } : {}),
1865 ...(row.item_id ? { itemId: row.item_id } : {}),
1866 ...(row.request_id ? { requestId: row.request_id } : {}),
1867 ...(row.request_kind ? { requestKind: row.request_kind } : {}),
1868 ...(row.text_delta !== null ? { textDelta: row.text_delta } : {}),
1869 ...(row.message ? { message: row.message } : {}),
1870 ...(payload !== undefined ? { payload } : {}),
1871 });
1872 if (!parsed.success) {
1873 return null;
1874 }
1875 return parsed.data;
1876 }
1877
1776 private parseJson<T>(json: string, schema: SafeParseSchema<T>): T | null {1878 private parseJson<T>(json: string, schema: SafeParseSchema<T>): T | null {
1777 const candidate = this.tryParseJson(json);1879 const candidate = this.tryParseJson(json);
1778 const parsed = schema.safeParse(candidate);1880 const parsed = schema.safeParse(candidate);
apps/server/src/providerManager.test.ts+52−0
@@ -3,6 +3,58 @@ import { describe, expect, it, vi } from "vitest"; · 2 unmodified lines
3import { ProviderManager } from "./providerManager";3import { ProviderManager } from "./providerManager";
4 4
5describe("ProviderManager", () => {5describe("ProviderManager", () => {
6 it("emits persisted provider events with durable sequence metadata when available", () => {
7 const ingestProviderEvent = vi.fn((event: {
8 id: string;
9 kind: "notification";
10 provider: "codex";
11 sessionId: string;
12 createdAt: string;
13 method: string;
14 threadId: string;
15 }) => ({
16 ...event,
17 seq: 11,
18 }));
19 const persistenceService = {
20 ingestProviderEvent,
21 unbindSession: vi.fn(),
22 };
23 const manager = new ProviderManager({
24 persistenceService: persistenceService as never,
25 });
26 const events: Array<{ id: string; seq?: number }> = [];
27 manager.on("event", (event) => {
28 events.push({ id: event.id, seq: event.seq });
29 });
30
31 const internals = manager as unknown as {
32 onCodexEvent: (event: {
33 id: string;
34 kind: "notification";
35 provider: "codex";
36 sessionId: string;
37 createdAt: string;
38 method: string;
39 threadId: string;
40 }) => void;
41 };
42
43 internals.onCodexEvent({
44 id: "evt-seq-1",
45 kind: "notification",
46 provider: "codex",
47 sessionId: "sess-1",
48 createdAt: new Date().toISOString(),
49 method: "turn/started",
50 threadId: "thread-1",
51 });
52
53 expect(ingestProviderEvent).toHaveBeenCalledTimes(1);
54 expect(events).toEqual([{ id: "evt-seq-1", seq: 11 }]);
55 manager.dispose();
56 });
57
6 it("detaches provider event listener and ends thread log streams on dispose", () => {58 it("detaches provider event listener and ends thread log streams on dispose", () => {
7 const manager = new ProviderManager();59 const manager = new ProviderManager();
8 const internals = manager as unknown as {60 const internals = manager as unknown as {
apps/server/src/providerManager.ts+26−13
@@ -180,18 +180,11 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> { · 179 unmodified lines
180 if (this.disposed) {180 if (this.disposed) {
181 return;181 return;
182 }182 }
183 183 const emittedEvent = this.publishProviderEvent(event);
184 this.routeEventToThreadLog(event);184 if (!emittedEvent) {
185 try {185 return;
186 this.persistenceService?.ingestProviderEvent(event);
187 } catch {
188 // Persistence failures should not break provider streaming.
189 }
190 if (event.method === "session/closed" || event.method === "session/exited") {
191 this.persistenceService?.unbindSession(event.sessionId);
192 }186 }
193 this.emit("event", event);187 this.maybeCaptureFilesystemCheckpoint(emittedEvent);
194 this.maybeCaptureFilesystemCheckpoint(event);
195 };188 };
196 189
197 constructor(options: ProviderManagerOptions = {}) {190 constructor(options: ProviderManagerOptions = {}) {
@@ -546,7 +539,7 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> { · 348 unmodified lines
546 }539 }
547 540
548 private emitCheckpointCaptured(sessionId: string, threadId: string, turnCount: number): void {541 private emitCheckpointCaptured(sessionId: string, threadId: string, turnCount: number): void {
549 this.emit("event", {542 this.publishProviderEvent({
550 id: randomUUID(),543 id: randomUUID(),
551 kind: "notification",544 kind: "notification",
552 provider: "codex",545 provider: "codex",
@@ -566,7 +559,7 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> { · 13 unmodified lines
566 message: string,559 message: string,
567 threadId?: string,560 threadId?: string,
568 ): void {561 ): void {
569 this.emit("event", {562 this.publishProviderEvent({
570 id: randomUUID(),563 id: randomUUID(),
571 kind: "error",564 kind: "error",
572 provider: "codex",565 provider: "codex",
@@ -578,6 +571,26 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> { · 5 unmodified lines
578 });571 });
579 }572 }
580 573
574 private publishProviderEvent(event: ProviderEvent): ProviderEvent | null {
575 let emittedEvent = event;
576 try {
577 const persistedEvent = this.persistenceService?.ingestProviderEvent(event);
578 if (this.persistenceService && !persistedEvent) {
579 return null;
580 }
581 emittedEvent = persistedEvent ?? event;
582 } catch {
583 // Persistence failures should not break provider streaming.
584 }
585
586 this.routeEventToThreadLog(emittedEvent);
587 if (emittedEvent.method === "session/closed" || emittedEvent.method === "session/exited") {
588 this.persistenceService?.unbindSession(emittedEvent.sessionId);
589 }
590 this.emit("event", emittedEvent);
591 return emittedEvent;
592 }
593
581 private withFilesystemLock<T>(sessionId: string, fn: () => Promise<T>): Promise<T> {594 private withFilesystemLock<T>(sessionId: string, fn: () => Promise<T>): Promise<T> {
582 const previous = this.filesystemLocks.get(sessionId) ?? Promise.resolve();595 const previous = this.filesystemLocks.get(sessionId) ?? Promise.resolve();
583 const next = previous.catch(() => undefined).then(fn);596 const next = previous.catch(() => undefined).then(fn);
apps/server/src/wsServer.test.ts+52−0
@@ -669,6 +669,58 @@ describe("WebSocket Server", () => { · 668 unmodified lines
669 expect(response.result).toEqual([]);669 expect(response.result).toEqual([]);
670 });670 });
671 671
672 it("returns durable provider catch-up events ordered by seq", async () => {
673 const stateDir = makeTempDir("t3code-ws-provider-catchup-");
674 const persistenceService = new PersistenceService({
675 dbPath: path.join(stateDir, "state.sqlite"),
676 legacyProjectsJsonPath: path.join(stateDir, "projects.json"),
677 });
678
679 const first = persistenceService.ingestProviderEvent({
680 id: "provider-evt-1",
681 kind: "notification",
682 provider: "codex",
683 sessionId: "sess-1",
684 createdAt: "2026-02-20T00:00:00.000Z",
685 method: "turn/started",
686 turnId: "turn-1",
687 payload: { turn: { id: "turn-1" } },
688 });
689 const second = persistenceService.ingestProviderEvent({
690 id: "provider-evt-2",
691 kind: "notification",
692 provider: "codex",
693 sessionId: "sess-1",
694 createdAt: "2026-02-20T00:00:01.000Z",
695 method: "turn/completed",
696 turnId: "turn-1",
697 payload: { turn: { id: "turn-1", status: "completed" } },
698 });
699
700 server = createTestServer({ cwd: "/test", stateDir, persistenceService });
701 await server.start();
702 const addr = server.httpServer.address();
703 const port = typeof addr === "object" && addr !== null ? addr.port : 0;
704
705 const ws = await connectWs(port);
706 connections.push(ws);
707 await waitForMessage(ws);
708
709 const response = await sendRequest(ws, WS_METHODS.providersCatchUp, {
710 afterSeq: first?.seq ?? 0,
711 });
712 expect(response.error).toBeUndefined();
713 expect(response.result).toEqual({
714 events: [
715 expect.objectContaining({
716 id: "provider-evt-2",
717 seq: second?.seq,
718 }),
719 ],
720 lastProviderSeq: second?.seq ?? 0,
721 });
722 });
723
672 it("persists app settings through appSettings RPC methods", async () => {724 it("persists app settings through appSettings RPC methods", async () => {
673 const stateDir = makeTempDir("t3code-ws-app-settings-");725 const stateDir = makeTempDir("t3code-ws-app-settings-");
674 server = createTestServer({ cwd: "/test", stateDir });726 server = createTestServer({ cwd: "/test", stateDir });
apps/server/src/wsServer.ts+3−0
@@ -334,6 +334,9 @@ export function createServer(options: ServerOptions) { · 333 unmodified lines
334 case WS_METHODS.providersListSessions:334 case WS_METHODS.providersListSessions:
335 return providerManager.listSessions();335 return providerManager.listSessions();
336 336
337 case WS_METHODS.providersCatchUp:
338 return persistenceService.providerCatchUp(request.params as never);
339
337 case WS_METHODS.providersListCheckpoints:340 case WS_METHODS.providersListCheckpoints:
338 return providerManager.listCheckpoints(request.params as never);341 return providerManager.listCheckpoints(request.params as never);
339 342
apps/web/src/routes/__root.tsx+72−2
@@ -6,6 +6,7 @@ import { · 5 unmodified lines
6} from "@tanstack/react-router";6} from "@tanstack/react-router";
7import { useEffect, useRef } from "react";7import { useEffect, useRef } from "react";
8import { QueryClient, useQueryClient } from "@tanstack/react-query";8import { QueryClient, useQueryClient } from "@tanstack/react-query";
9import type { ProviderEvent } from "@t3tools/contracts";
9 10
10import { APP_DISPLAY_NAME } from "../branding";11import { APP_DISPLAY_NAME } from "../branding";
11import { Button } from "../components/ui/button";12import { Button } from "../components/ui/button";
@@ -120,6 +121,14 @@ function errorDetails(error: unknown): string { · 108 unmodified lines
120 }121 }
121}122}
122 123
124function readProviderSeq(event: ProviderEvent): number | null {
125 const seq = event.seq;
126 if (typeof seq !== "number" || !Number.isInteger(seq) || seq <= 0) {
127 return null;
128 }
129 return seq;
130}
131
123function StateSyncRouter() {132function StateSyncRouter() {
124 const api = useNativeApi();133 const api = useNativeApi();
125 const { dispatch } = useStore();134 const { dispatch } = useStore();
@@ -131,6 +140,8 @@ function StateSyncRouter() { · 5 unmodified lines
131 });140 });
132 const lastStateSeqRef = useRef(0);141 const lastStateSeqRef = useRef(0);
133 const stateQueueRef = useRef(Promise.resolve());142 const stateQueueRef = useRef(Promise.resolve());
143 const lastProviderSeqRef = useRef(0);
144 const providerQueueRef = useRef(Promise.resolve());
134 145
135 useEffect(() => {146 useEffect(() => {
136 if (!api) return;147 if (!api) return;
@@ -202,7 +213,7 @@ function StateSyncRouter() { · 65 unmodified lines
202 213
203 useEffect(() => {214 useEffect(() => {
204 if (!api) return;215 if (!api) return;
205 return api.providers.onEvent((event) => {216 const applyProviderEvent = (event: ProviderEvent) => {
206 if (event.method === "turn/completed") {217 if (event.method === "turn/completed") {
207 void invalidateGitQueries(queryClient);218 void invalidateGitQueries(queryClient);
208 }219 }
@@ -217,14 +228,73 @@ function StateSyncRouter() { · 8 unmodified lines
217 },228 },
218 });229 });
219 }230 }
220 if (!activeThreadId) return;
221 dispatch({231 dispatch({
222 type: "APPLY_EVENT",232 type: "APPLY_EVENT",
223 event,233 event,
224 activeAssistantItemRef,234 activeAssistantItemRef,
225 activeThreadId,235 activeThreadId,
226 });236 });
237 };
238
239 const replayProviderCatchUp = async (): Promise<void> => {
240 const catchUp = await api.providers.catchUp({
241 afterSeq: lastProviderSeqRef.current,
242 });
243 if (catchUp.events.length === 0) {
244 lastProviderSeqRef.current = Math.max(lastProviderSeqRef.current, catchUp.lastProviderSeq);
245 return;
246 }
247
248 for (const missingEvent of catchUp.events) {
249 const missingSeq = readProviderSeq(missingEvent);
250 if (!missingSeq || missingSeq <= lastProviderSeqRef.current) continue;
251 applyProviderEvent(missingEvent);
252 lastProviderSeqRef.current = missingSeq;
253 }
254
255 if (lastProviderSeqRef.current < catchUp.lastProviderSeq) {
256 await replayProviderCatchUp();
257 }
258 };
259
260 const enqueueProviderWork = (work: () => Promise<void>) => {
261 providerQueueRef.current = providerQueueRef.current.then(work).catch(() => undefined);
262 };
263
264 const unsubscribeWelcome = onServerWelcome(() => {
265 if (lastProviderSeqRef.current <= 0) {
266 return;
267 }
268 enqueueProviderWork(async () => {
269 await replayProviderCatchUp();
270 });
227 });271 });
272
273 const unsubscribeProvider = api.providers.onEvent((event) => {
274 enqueueProviderWork(async () => {
275 const eventSeq = readProviderSeq(event);
276 if (!eventSeq) {
277 applyProviderEvent(event);
278 return;
279 }
280 if (eventSeq <= lastProviderSeqRef.current) {
281 return;
282 }
283 if (lastProviderSeqRef.current > 0 && eventSeq > lastProviderSeqRef.current + 1) {
284 await replayProviderCatchUp();
285 }
286 if (eventSeq <= lastProviderSeqRef.current) {
287 return;
288 }
289 applyProviderEvent(event);
290 lastProviderSeqRef.current = eventSeq;
291 });
292 });
293
294 return () => {
295 unsubscribeWelcome();
296 unsubscribeProvider();
297 };
228 }, [activeThreadId, api, dispatch, queryClient]);298 }, [activeThreadId, api, dispatch, queryClient]);
229 299
230 useEffect(() => {300 useEffect(() => {
apps/web/src/wsNativeApi.ts+1−0
@@ -93,6 +93,7 @@ export function createWsNativeApi(): NativeApi { · 92 unmodified lines
93 listSessions: () => transport.request(WS_METHODS.providersListSessions),93 listSessions: () => transport.request(WS_METHODS.providersListSessions),
94 listCheckpoints: (input) => transport.request(WS_METHODS.providersListCheckpoints, input),94 listCheckpoints: (input) => transport.request(WS_METHODS.providersListCheckpoints, input),
95 getCheckpointDiff: (input) => transport.request(WS_METHODS.providersGetCheckpointDiff, input),95 getCheckpointDiff: (input) => transport.request(WS_METHODS.providersGetCheckpointDiff, input),
96 catchUp: (input) => transport.request(WS_METHODS.providersCatchUp, input),
96 revertToCheckpoint: (input) =>97 revertToCheckpoint: (input) =>
97 transport.request(WS_METHODS.providersRevertToCheckpoint, input),98 transport.request(WS_METHODS.providersRevertToCheckpoint, input),
98 onEvent: (callback) =>99 onEvent: (callback) =>
packages/contracts/src/ipc.ts+3−0
@@ -17,6 +17,8 @@ import type { · 16 unmodified lines
17 GitStatusResult,17 GitStatusResult,
18} from "./git";18} from "./git";
19import type {19import type {
20 ProviderCatchUpInput,
21 ProviderCatchUpResult,
20 ProviderEvent,22 ProviderEvent,
21 ProviderGetCheckpointDiffInput,23 ProviderGetCheckpointDiffInput,
22 ProviderGetCheckpointDiffResult,24 ProviderGetCheckpointDiffResult,
@@ -118,6 +120,7 @@ export interface NativeApi { · 95 unmodified lines
118 getCheckpointDiff: (120 getCheckpointDiff: (
119 input: ProviderGetCheckpointDiffInput,121 input: ProviderGetCheckpointDiffInput,
120 ) => Promise<ProviderGetCheckpointDiffResult>;122 ) => Promise<ProviderGetCheckpointDiffResult>;
123 catchUp: (input: ProviderCatchUpInput) => Promise<ProviderCatchUpResult>;
121 revertToCheckpoint: (124 revertToCheckpoint: (
122 input: ProviderRevertToCheckpointInput,125 input: ProviderRevertToCheckpointInput,
123 ) => Promise<ProviderRevertToCheckpointResult>;126 ) => Promise<ProviderRevertToCheckpointResult>;
packages/contracts/src/provider.test.ts+42−0
@@ -2,6 +2,8 @@ import { describe, expect, it } from "vitest"; · 1 unmodified line
2 2
3import {3import {
4 PROVIDER_SEND_TURN_MAX_ATTACHMENTS,4 PROVIDER_SEND_TURN_MAX_ATTACHMENTS,
5 providerCatchUpInputSchema,
6 providerCatchUpResultSchema,
5 providerCheckpointSchema,7 providerCheckpointSchema,
6 providerEventSchema,8 providerEventSchema,
7 providerGetCheckpointDiffInputSchema,9 providerGetCheckpointDiffInputSchema,
@@ -156,6 +158,19 @@ describe("providerEventSchema", () => { · 148 unmodified lines
156 expect(parsed.method).toBe("item/agentMessage/delta");158 expect(parsed.method).toBe("item/agentMessage/delta");
157 });159 });
158 160
161 it("accepts optional durable sequence metadata", () => {
162 const parsed = providerEventSchema.parse({
163 seq: 42,
164 id: "evt_seq_1",
165 kind: "notification",
166 provider: "codex",
167 sessionId: "sess_1",
168 createdAt: "2026-01-01T00:00:00.000Z",
169 method: "turn/completed",
170 });
171 expect(parsed.seq).toBe(42);
172 });
173
159 it("accepts request approval metadata", () => {174 it("accepts request approval metadata", () => {
160 const parsed = providerEventSchema.parse({175 const parsed = providerEventSchema.parse({
161 id: "evt_2",176 id: "evt_2",
@@ -172,6 +187,33 @@ describe("providerEventSchema", () => { · 10 unmodified lines
172 });187 });
173});188});
174 189
190describe("provider catch-up schemas", () => {
191 it("parses catch-up input defaults", () => {
192 const parsed = providerCatchUpInputSchema.parse({});
193 expect(parsed.afterSeq).toBe(0);
194 expect(parsed.limit).toBe(1_000);
195 });
196
197 it("parses catch-up result payload", () => {
198 const parsed = providerCatchUpResultSchema.parse({
199 events: [
200 {
201 seq: 7,
202 id: "evt_7",
203 kind: "notification",
204 provider: "codex",
205 sessionId: "sess_1",
206 createdAt: "2026-01-01T00:00:00.000Z",
207 method: "turn/started",
208 },
209 ],
210 lastProviderSeq: 9,
211 });
212 expect(parsed.events[0]?.seq).toBe(7);
213 expect(parsed.lastProviderSeq).toBe(9);
214 });
215});
216
175describe("providerRespondToRequestInputSchema", () => {217describe("providerRespondToRequestInputSchema", () => {
176 it("accepts valid decisions", () => {218 it("accepts valid decisions", () => {
177 const parsed = providerRespondToRequestInputSchema.parse({219 const parsed = providerRespondToRequestInputSchema.parse({
packages/contracts/src/provider.ts+13−0
@@ -170,6 +170,11 @@ export const providerGetCheckpointDiffResultSchema = z.object({ · 169 unmodified lines
170 diff: z.string(),170 diff: z.string(),
171});171});
172 172
173export const providerCatchUpInputSchema = z.object({
174 afterSeq: z.number().int().min(0).default(0),
175 limit: z.number().int().min(1).max(5_000).default(1_000),
176});
177
173export const providerRespondToRequestInputSchema = z.object({178export const providerRespondToRequestInputSchema = z.object({
174 sessionId: z.string().min(1),179 sessionId: z.string().min(1),
175 requestId: z.string().min(1),180 requestId: z.string().min(1),
@@ -179,6 +184,7 @@ export const providerRespondToRequestInputSchema = z.object({ · 3 unmodified lines
179export const providerEventKindSchema = z.enum(["session", "notification", "request", "error"]);184export const providerEventKindSchema = z.enum(["session", "notification", "request", "error"]);
180 185
181export const providerEventSchema = z.object({186export const providerEventSchema = z.object({
187 seq: z.number().int().positive().optional(),
182 id: z.string().min(1),188 id: z.string().min(1),
183 kind: providerEventKindSchema,189 kind: providerEventKindSchema,
184 provider: providerKindSchema,190 provider: providerKindSchema,
@@ -195,6 +201,11 @@ export const providerEventSchema = z.object({ · 10 unmodified lines
195 payload: z.unknown().optional(),201 payload: z.unknown().optional(),
196});202});
197 203
204export const providerCatchUpResultSchema = z.object({
205 events: z.array(providerEventSchema),
206 lastProviderSeq: z.number().int().min(0),
207});
208
198export type ProviderKind = z.infer<typeof providerKindSchema>;209export type ProviderKind = z.infer<typeof providerKindSchema>;
199export type ProviderApprovalPolicy = z.infer<typeof providerApprovalPolicySchema>;210export type ProviderApprovalPolicy = z.infer<typeof providerApprovalPolicySchema>;
200export type ProviderSandboxMode = z.infer<typeof providerSandboxModeSchema>;211export type ProviderSandboxMode = z.infer<typeof providerSandboxModeSchema>;
@@ -219,6 +230,8 @@ export type ProviderRevertToCheckpointResult = z.infer< · 18 unmodified lines
219>;230>;
220export type ProviderGetCheckpointDiffInput = z.input<typeof providerGetCheckpointDiffInputSchema>;231export type ProviderGetCheckpointDiffInput = z.input<typeof providerGetCheckpointDiffInputSchema>;
221export type ProviderGetCheckpointDiffResult = z.infer<typeof providerGetCheckpointDiffResultSchema>;232export type ProviderGetCheckpointDiffResult = z.infer<typeof providerGetCheckpointDiffResultSchema>;
233export type ProviderCatchUpInput = z.input<typeof providerCatchUpInputSchema>;
234export type ProviderCatchUpResult = z.infer<typeof providerCatchUpResultSchema>;
222export type ProviderRespondToRequestInput = z.input<typeof providerRespondToRequestInputSchema>;235export type ProviderRespondToRequestInput = z.input<typeof providerRespondToRequestInputSchema>;
223export type ProviderEventKind = z.infer<typeof providerEventKindSchema>;236export type ProviderEventKind = z.infer<typeof providerEventKindSchema>;
224export type ProviderEvent = z.infer<typeof providerEventSchema>;237export type ProviderEvent = z.infer<typeof providerEventSchema>;
packages/contracts/src/ws.test.ts+4−0
@@ -7,4 +7,8 @@ describe("WS thread methods", () => { · 6 unmodified lines
7 expect(WS_METHODS.threadsUpdateTerminalState).toBe("threads.updateTerminalState");7 expect(WS_METHODS.threadsUpdateTerminalState).toBe("threads.updateTerminalState");
8 expect("threadsUpdate" in WS_METHODS).toBe(false);8 expect("threadsUpdate" in WS_METHODS).toBe(false);
9 });9 });
10
11 it("includes provider catch-up method for durable stream resume", () => {
12 expect(WS_METHODS.providersCatchUp).toBe("providers.catchUp");
13 });
10});14});
packages/contracts/src/ws.ts+1−0
@@ -28,6 +28,7 @@ export const WS_METHODS = { · 27 unmodified lines
28 providersListSessions: "providers.listSessions",28 providersListSessions: "providers.listSessions",
29 providersListCheckpoints: "providers.listCheckpoints",29 providersListCheckpoints: "providers.listCheckpoints",
30 providersGetCheckpointDiff: "providers.getCheckpointDiff",30 providersGetCheckpointDiff: "providers.getCheckpointDiff",
31 providersCatchUp: "providers.catchUp",
31 providersRevertToCheckpoint: "providers.revertToCheckpoint",32 providersRevertToCheckpoint: "providers.revertToCheckpoint",
32 33
33 // Project registry methods34 // Project registry methods