authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-18 23:27:22-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-19 00:35:32-08:00
log79382b272c02f3a16e4ea77c3c67103d4c14414e
tree7afdd7c74ce16c6ba02fd90c041d240b1d65f7ef
parent90e199d6a7cc667b5c94d08e4be61504a18a1968

Emit checkpoint captured events after turn checkpointing

- emit `checkpoint/captured` from `ProviderManager` after checkpoint init/capture - add backend test coverage for the new checkpoint-captured notification flow - refresh provider queries on `checkpoint/captured` and only re-derive diffs on `turn/completed`

4 files changed, 111 insertions(+), 12 deletions(-)

apps/server/src/providerManager.test.ts+76
......@@ -296,6 +296,82 @@ describe("ProviderManager", () => {
296296 manager.dispose();
297297 });
298298
299 it("emits checkpoint/captured after turn completion checkpoint capture", async () => {
300 const manager = new ProviderManager();
301 const events: Array<{ method: string; payload?: unknown; threadId?: string }> = [];
302 manager.on("event", (event) => {
303 events.push({
304 method: event.method,
305 ...(event.payload !== undefined ? { payload: event.payload } : {}),
306 ...(event.threadId ? { threadId: event.threadId } : {}),
307 });
308 });
309
310 const internals = manager as unknown as {
311 onCodexEvent: (event: {
312 id: string;
313 kind: "notification";
314 provider: "codex";
315 sessionId: string;
316 createdAt: string;
317 method: string;
318 threadId: string;
319 }) => void;
320 codex: {
321 hasSession: (sessionId: string) => boolean;
322 readThread: (sessionId: string) => Promise<{
323 threadId: string;
324 turns: Array<{ id: string; items: unknown[] }>;
325 }>;
326 };
327 filesystemCheckpointStore: {
328 captureCheckpoint: (input: {
329 cwd: string;
330 threadId: string;
331 turnCount: number;
332 }) => Promise<void>;
333 };
334 sessionCheckpointCwds: Map<string, string>;
335 };
336
337 internals.codex.hasSession = () => true;
338 internals.codex.readThread = async () => ({
339 threadId: "thr_1",
340 turns: [{ id: "turn_1", items: [] }],
341 });
342 const captureCheckpoint = vi.fn(async () => undefined);
343 internals.filesystemCheckpointStore.captureCheckpoint = captureCheckpoint;
344 internals.sessionCheckpointCwds.set("sess_1", "/repo");
345
346 internals.onCodexEvent({
347 id: "evt-turn-complete",
348 kind: "notification",
349 provider: "codex",
350 sessionId: "sess_1",
351 createdAt: "2026-02-19T00:00:00.000Z",
352 method: "turn/completed",
353 threadId: "thr_1",
354 });
355
356 await vi.waitFor(() => {
357 expect(captureCheckpoint).toHaveBeenCalledWith({
358 cwd: "/repo",
359 threadId: "thr_1",
360 turnCount: 1,
361 });
362 expect(events.some((event) => event.method === "checkpoint/captured")).toBe(true);
363 });
364
365 const checkpointCapturedEvent = events.find((event) => event.method === "checkpoint/captured");
366 expect(checkpointCapturedEvent?.threadId).toBe("thr_1");
367 expect(checkpointCapturedEvent?.payload).toEqual({
368 threadId: "thr_1",
369 turnCount: 1,
370 });
371
372 manager.dispose();
373 });
374
299375 it("lazily initializes filesystem checkpoints before diffing when cache is missing", async () => {
300376 const manager = new ProviderManager();
301377 const internals = manager as unknown as {
apps/server/src/providerManager.ts+30-5
......@@ -477,11 +477,19 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {
477477
478478 const checkpointCwd = this.sessionCheckpointCwds.get(event.sessionId);
479479 if (!checkpointCwd) {
480 void this.getOrInitializeFilesystemCheckpointCwd(event.sessionId).catch((error) => {
481 const message =
482 error instanceof Error ? error.message : "Failed to initialize filesystem checkpoints.";
483 this.emitFilesystemCheckpointError(event.sessionId, message, event.threadId);
484 });
480 void this.getOrInitializeFilesystemCheckpointCwd(event.sessionId)
481 .then(async (initializedCwd) => {
482 if (!initializedCwd || !this.codex.hasSession(event.sessionId)) {
483 return;
484 }
485 const snapshot = await this.codex.readThread(event.sessionId);
486 this.emitCheckpointCaptured(event.sessionId, snapshot.threadId, snapshot.turns.length);
487 })
488 .catch((error) => {
489 const message =
490 error instanceof Error ? error.message : "Failed to initialize filesystem checkpoints.";
491 this.emitFilesystemCheckpointError(event.sessionId, message, event.threadId);
492 });
485493 return;
486494 }
487495
......@@ -496,12 +504,29 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {
496504 threadId: snapshot.threadId,
497505 turnCount: snapshot.turns.length,
498506 });
507 this.emitCheckpointCaptured(event.sessionId, snapshot.threadId, snapshot.turns.length);
499508 }).catch((error) => {
500509 const message = error instanceof Error ? error.message : "Failed to capture checkpoint.";
501510 this.emitFilesystemCheckpointError(event.sessionId, message, event.threadId);
502511 });
503512 }
504513
514 private emitCheckpointCaptured(sessionId: string, threadId: string, turnCount: number): void {
515 this.emit("event", {
516 id: randomUUID(),
517 kind: "notification",
518 provider: "codex",
519 sessionId,
520 createdAt: new Date().toISOString(),
521 method: "checkpoint/captured",
522 threadId,
523 payload: {
524 threadId,
525 turnCount,
526 },
527 });
528 }
529
505530 private emitFilesystemCheckpointError(
506531 sessionId: string,
507532 message: string,
apps/web/src/routes/__root.tsx+4
......@@ -6,6 +6,7 @@ import { AnchoredToastProvider, ToastProvider } from "../components/ui/toast";
66import { isElectron } from "../env";
77import { useNativeApi } from "../hooks/useNativeApi";
88import { invalidateGitQueries } from "../lib/gitReactQuery";
9import { providerQueryKeys } from "../lib/providerReactQuery";
910import { DEFAULT_MODEL } from "../model-logic";
1011import { useStore } from "../store";
1112import { onServerWelcome } from "../wsNativeApi";
......@@ -57,6 +58,9 @@ function EventRouter() {
5758 if (event.method === "turn/completed") {
5859 void invalidateGitQueries(queryClient);
5960 }
61 if (event.method === "checkpoint/captured") {
62 void queryClient.invalidateQueries({ queryKey: providerQueryKeys.all });
63 }
6064 if (!activeThreadId) return;
6165 dispatch({
6266 type: "APPLY_EVENT",
apps/web/src/store.ts+1-7
......@@ -891,13 +891,7 @@ export function reducer(state: AppState, action: Action): AppState {
891891 ...state,
892892 threads: updateThread(state.threads, target.id, (t) => {
893893 const nextEvents = [event, ...t.events];
894 const itemType = asString(asObject(asObject(event.payload)?.item)?.type);
895 const normalizedItemType = itemType?.replace(/[_-]/g, "").toLowerCase();
896 const shouldRederiveDiffs =
897 event.method === "turn/completed" ||
898 event.method === "turn/diff/updated" ||
899 (event.method === "item/completed" &&
900 (normalizedItemType === "agentmessage" || normalizedItemType === "filechange"));
894 const shouldRederiveDiffs = event.method === "turn/completed";
901895 const turnDiffSummaries = shouldRederiveDiffs
902896 ? mergeTurnDiffSummaries(t.turnDiffSummaries, deriveTurnDiffSummaries(nextEvents))
903897 : t.turnDiffSummaries;