authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-09-04 22:36:32-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 22:36:32-07:00
log7a089b2b2449c5c146e1a7d554a31e6d55924d3f
tree8298be14595bc34194e37a84dd7c618e15085fc0
parentf47a3fe90354a3dfa14276e3699541d552426276
signature Signed by PGP key B5690EEEBB952194

fix(server): capture turn checkpoints after all edits finish (#9841)

7 files changed, 453 insertions(+), 212 deletions(-)

apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+162
...@@ -12,6 +12,7 @@ import {...@@ -12,6 +12,7 @@ import {
12} from "@t3tools/contracts";12} from "@t3tools/contracts";
13import {13import {
14 CommandId,14 CommandId,
15 CheckpointRef,
15 DEFAULT_PROVIDER_INTERACTION_MODE,16 DEFAULT_PROVIDER_INTERACTION_MODE,
16 EventId,17 EventId,
17 MessageId,18 MessageId,
...@@ -667,6 +668,167 @@ describe("CheckpointReactor", () => {...@@ -667,6 +668,167 @@ describe("CheckpointReactor", () => {
667 }),668 }),
668 );669 );
669670
671 effectIt.effect.each(["turn.completed", "turn.aborted"] as const)(
672 "captures every edit after a mid-turn diff update on %s",
673 (terminalEventType) =>
674 Effect.gen(function* () {
675 const harness = yield* Effect.promise(() =>
676 createHarness({ seedFilesystemCheckpoints: false }),
677 );
678 const threadId = ThreadId.make("thread-1");
679 const turnId = asTurnId("turn-1");
680 const assistantMessageId = MessageId.make("assistant:mid-turn");
681 const createdAt = "2026-01-01T00:00:00.000Z";
682 yield* harness.engine.dispatch({
683 type: "thread.session.set",
684 commandId: CommandId.make("cmd-mid-turn-running"),
685 threadId,
686 session: {
687 threadId,
688 status: "running",
689 providerName: "codex",
690 runtimeMode: "approval-required",
691 activeTurnId: turnId,
692 lastError: null,
693 updatedAt: createdAt,
694 },
695 createdAt,
696 });
697 harness.provider.emit({
698 type: "turn.started",
699 eventId: EventId.make("evt-mid-turn-start"),
700 provider: ProviderDriverKind.make("codex"),
701 createdAt,
702 threadId,
703 turnId,
704 });
705 expect(yield* harness.nextReceipt).toMatchObject({
706 type: "checkpoint.baseline.captured",
707 });
708
709 NodeFS.writeFileSync(NodePath.join(harness.cwd, "early.ts"), "export const early = 1;\n");
710 yield* harness.engine.dispatch({
711 type: "thread.turn.diff.complete",
712 commandId: CommandId.make("cmd-mid-turn-diff"),
713 threadId,
714 turnId,
715 completedAt: createdAt,
716 checkpointRef: CheckpointRef.make("provider-diff:mid-turn"),
717 assistantMessageId,
718 status: "missing",
719 files: [],
720 checkpointTurnCount: 1,
721 createdAt,
722 });
723 yield* Effect.promise(harness.drain);
724
725 NodeFS.writeFileSync(NodePath.join(harness.cwd, "late.ts"), "export const late = 2;\n");
726 yield* harness.engine.dispatch({
727 type: "thread.session.set",
728 commandId: CommandId.make("cmd-mid-turn-settled"),
729 threadId,
730 session: {
731 threadId,
732 status: terminalEventType === "turn.aborted" ? "interrupted" : "ready",
733 providerName: "codex",
734 runtimeMode: "approval-required",
735 activeTurnId: null,
736 lastError: null,
737 updatedAt: createdAt,
738 },
739 createdAt,
740 });
741 harness.provider.emit({
742 eventId: EventId.make("evt-mid-turn-complete"),
743 provider: ProviderDriverKind.make("codex"),
744 createdAt,
745 threadId,
746 turnId,
747 ...(terminalEventType === "turn.completed"
748 ? { type: "turn.completed", payload: { state: "completed" } }
749 : { type: "turn.aborted", payload: { reason: "Interrupted by user." } }),
750 });
751 yield* Effect.promise(harness.drain);
752 expect(gitRefExists(harness.cwd, checkpointRefForThreadTurn(threadId, 1))).toBe(true);
753 expect(yield* harness.nextReceipt).toMatchObject({
754 type: "checkpoint.diff.finalized",
755 turnId,
756 checkpointTurnCount: 1,
757 });
758 expect(yield* harness.nextReceipt).toMatchObject({
759 type: "turn.processing.quiesced",
760 turnId,
761 });
762 yield* Effect.promise(harness.drain);
763 const thread = (yield* Effect.promise(harness.readModel)).threads.find(
764 (entry) => entry.id === threadId,
765 );
766 expect(thread?.checkpoints).toHaveLength(1);
767 expect(thread?.checkpoints[0]?.status).toBe("ready");
768 expect(thread?.latestTurn?.state).toBe(
769 terminalEventType === "turn.aborted" ? "interrupted" : "completed",
770 );
771 expect(thread?.checkpoints[0]?.assistantMessageId).toBe(assistantMessageId);
772 expect(thread?.checkpoints[0]?.files.map((file) => file.path)).toEqual([
773 "early.ts",
774 "late.ts",
775 ]);
776 expect(
777 gitShowFileAtRef(harness.cwd, checkpointRefForThreadTurn(threadId, 1), "late.ts"),
778 ).toBe("export const late = 2;\n");
779
780 const followUpTurnId = asTurnId("turn-2");
781 harness.provider.emit({
782 type: "turn.started",
783 eventId: EventId.make("evt-follow-up-start"),
784 provider: ProviderDriverKind.make("codex"),
785 createdAt,
786 threadId,
787 turnId: followUpTurnId,
788 });
789 harness.provider.emit({
790 type: "turn.completed",
791 eventId: EventId.make("evt-follow-up-complete"),
792 provider: ProviderDriverKind.make("codex"),
793 createdAt,
794 threadId,
795 turnId: followUpTurnId,
796 payload: { state: "completed" },
797 });
798 expect(yield* harness.nextReceipt).toMatchObject({
799 type: "checkpoint.diff.finalized",
800 turnId: followUpTurnId,
801 checkpointTurnCount: 2,
802 });
803 const followUp = (yield* Effect.promise(harness.readModel)).threads.find(
804 (entry) => entry.id === threadId,
805 );
806 expect(
807 followUp?.checkpoints.find((checkpoint) => checkpoint.turnId === followUpTurnId),
808 ).toMatchObject({ checkpointTurnCount: 2, files: [] });
809 }),
810 );
811
812 it("does not capture an aborted turn without a matching start or active session", async () => {
813 const harness = await createHarness({ seedFilesystemCheckpoints: false });
814 harness.provider.emit({
815 type: "turn.aborted",
816 eventId: EventId.make("evt-untracked-abort"),
817 provider: ProviderDriverKind.make("codex"),
818 createdAt: "2026-01-01T00:00:00.000Z",
819 threadId: ThreadId.make("thread-1"),
820 turnId: asTurnId("turn-untracked"),
821 payload: { reason: "Interrupted before the turn started." },
822 });
823 await harness.drain();
824
825 const thread = (await harness.readModel()).threads.find((entry) => entry.id === "thread-1");
826 expect(thread?.checkpoints).toEqual([]);
827 expect(
828 gitRefExists(harness.cwd, checkpointRefForThreadTurn(ThreadId.make("thread-1"), 1)),
829 ).toBe(false);
830 });
831
670 it("refreshes local git status state on turn completion using the session cwd", async () => {832 it("refreshes local git status state on turn completion using the session cwd", async () => {
671 const gitStatusRefreshCalls: string[] = [];833 const gitStatusRefreshCalls: string[] = [];
672 const harness = await createHarness({834 const harness = await createHarness({
apps/server/src/orchestration/Layers/CheckpointReactor.ts+16-91
...@@ -216,9 +216,7 @@ const make = Effect.gen(function* () {...@@ -216,9 +216,7 @@ const make = Effect.gen(function* () {
216 return cwd;216 return cwd;
217 });217 });
218218
219 // Shared tail for both capture paths: creates the git checkpoint ref, diffs219 // Capture the completed turn's files, then publish its summary and receipts.
220 // it against the previous turn, then dispatches the domain events to update
221 // the orchestration read model.
222 const captureAndDispatchCheckpoint = Effect.fn("captureAndDispatchCheckpoint")(function* (input: {220 const captureAndDispatchCheckpoint = Effect.fn("captureAndDispatchCheckpoint")(function* (input: {
223 readonly threadId: ThreadId;221 readonly threadId: ThreadId;
224 readonly turnId: TurnId;222 readonly turnId: TurnId;
...@@ -353,9 +351,9 @@ const make = Effect.gen(function* () {...@@ -353,9 +351,9 @@ const make = Effect.gen(function* () {
353 });351 });
354 });352 });
355353
356 // Captures a real git checkpoint when a turn completes via a runtime event.354 // Capture the files left by a completed or interrupted turn.
357 const captureCheckpointFromTurnCompletion = Effect.fn("captureCheckpointFromTurnCompletion")(355 const captureCheckpointFromTurnCompletion = Effect.fn("captureCheckpointFromTurnCompletion")(
358 function* (event: Extract<ProviderRuntimeEvent, { type: "turn.completed" }>) {356 function* (event: Extract<ProviderRuntimeEvent, { type: "turn.completed" | "turn.aborted" }>) {
359 const turnId = toTurnId(event.turnId);357 const turnId = toTurnId(event.turnId);
360 if (!turnId) {358 if (!turnId) {
361 return;359 return;
...@@ -412,75 +410,16 @@ const make = Effect.gen(function* () {...@@ -412,75 +410,16 @@ const make = Effect.gen(function* () {
412 thread,410 thread,
413 cwd: checkpointCwd,411 cwd: checkpointCwd,
414 turnCount: nextTurnCount,412 turnCount: nextTurnCount,
415 status: checkpointStatusFromRuntime(event.payload.state),413 status:
416 assistantMessageId: undefined,414 event.type === "turn.aborted"
415 ? "ready"
416 : checkpointStatusFromRuntime(event.payload.state),
417 assistantMessageId: existingPlaceholder?.assistantMessageId ?? undefined,
417 createdAt: event.createdAt,418 createdAt: event.createdAt,
418 });419 });
419 },420 },
420 );421 );
421422
422 // Captures a real git checkpoint when a placeholder checkpoint (status "missing")
423 // is detected via a domain event. This replaces the placeholder with a real
424 // git-ref-based checkpoint.
425 //
426 // ProviderRuntimeIngestion creates placeholder checkpoints on turn.diff.updated
427 // events from the Codex runtime. This handler fires when the corresponding
428 // domain event arrives, allowing the reactor to capture the actual filesystem
429 // state into a git ref and dispatch a replacement checkpoint.
430 const captureCheckpointFromPlaceholder = Effect.fn("captureCheckpointFromPlaceholder")(function* (
431 event: Extract<OrchestrationEvent, { type: "thread.turn-diff-completed" }>,
432 ) {
433 const { threadId, turnId, checkpointTurnCount, status } = event.payload;
434
435 // Only replace placeholders; skip events from our own real captures.
436 if (status !== "missing") {
437 return;
438 }
439
440 const thread = yield* resolveThreadDetail(threadId);
441 if (!thread) {
442 yield* Effect.logWarning("checkpoint capture from placeholder skipped: thread not found", {
443 threadId,
444 });
445 return;
446 }
447
448 // If a real checkpoint already exists for this turn, skip.
449 if (
450 thread.checkpoints.some(
451 (checkpoint) => checkpoint.turnId === turnId && checkpoint.status !== "missing",
452 )
453 ) {
454 yield* Effect.logDebug(
455 "checkpoint capture from placeholder skipped: real checkpoint already exists",
456 { threadId, turnId },
457 );
458 return;
459 }
460
461 const projects = yield* resolveThreadProjects(thread.projectId);
462 const checkpointCwd = yield* resolveCheckpointCwd({
463 threadId,
464 thread,
465 projects,
466 preferSessionRuntime: true,
467 });
468 if (!checkpointCwd) {
469 return;
470 }
471
472 yield* captureAndDispatchCheckpoint({
473 threadId,
474 turnId,
475 thread,
476 cwd: checkpointCwd,
477 turnCount: checkpointTurnCount,
478 status: "ready",
479 assistantMessageId: event.payload.assistantMessageId ?? undefined,
480 createdAt: event.payload.completedAt,
481 });
482 });
483
484 const ensurePreTurnBaselineFromTurnStart = Effect.fn("ensurePreTurnBaselineFromTurnStart")(423 const ensurePreTurnBaselineFromTurnStart = Effect.fn("ensurePreTurnBaselineFromTurnStart")(
485 function* (event: Extract<ProviderRuntimeEvent, { type: "turn.started" }>) {424 function* (event: Extract<ProviderRuntimeEvent, { type: "turn.started" }>) {
486 const turnId = toTurnId(event.turnId);425 const turnId = toTurnId(event.turnId);
...@@ -876,25 +815,6 @@ const make = Effect.gen(function* () {...@@ -876,25 +815,6 @@ const make = Effect.gen(function* () {
876 );815 );
877 return;816 return;
878 }817 }
879
880 // When ProviderRuntimeIngestion creates a placeholder checkpoint (status "missing")
881 // from a turn.diff.updated runtime event, capture the real git checkpoint to
882 // replace it. ProviderService broadcasts runtime events to each subscriber.
883 // This domain-event path also captures checkpoints from turn diff updates.
884 if (event.type === "thread.turn-diff-completed") {
885 yield* captureCheckpointFromPlaceholder(event).pipe(
886 Effect.catch((error) =>
887 Effect.flatMap(nowIso, (createdAt) =>
888 appendCaptureFailureActivity({
889 threadId: event.payload.threadId,
890 turnId: event.payload.turnId,
891 detail: error.message,
892 createdAt,
893 }).pipe(Effect.catch(() => Effect.void)),
894 ),
895 ),
896 );
897 }
898 });818 });
899819
900 const processRuntimeEvent = Effect.fn("processRuntimeEvent")(function* (820 const processRuntimeEvent = Effect.fn("processRuntimeEvent")(function* (
...@@ -939,7 +859,13 @@ const make = Effect.gen(function* () {...@@ -939,7 +859,13 @@ const make = Effect.gen(function* () {
939 pending.delete(event.threadId);859 pending.delete(event.threadId);
940 yield* pullRequests.refreshAfterTurn;860 yield* pullRequests.refreshAfterTurn;
941 }861 }
942 if (event.type === "turn.aborted") return;862 if (
863 event.type === "turn.aborted" &&
864 !isTrackedTurn &&
865 !sameId(thread?.session?.activeTurnId, turnId)
866 ) {
867 return;
868 }
943 yield* captureCheckpointFromTurnCompletion(event).pipe(869 yield* captureCheckpointFromTurnCompletion(event).pipe(
944 Effect.catch((error) =>870 Effect.catch((error) =>
945 Effect.flatMap(nowIso, (createdAt) =>871 Effect.flatMap(nowIso, (createdAt) =>
...@@ -987,8 +913,7 @@ const make = Effect.gen(function* () {...@@ -987,8 +913,7 @@ const make = Effect.gen(function* () {
987 if (913 if (
988 event.type !== "thread.turn-start-requested" &&914 event.type !== "thread.turn-start-requested" &&
989 event.type !== "thread.message-sent" &&915 event.type !== "thread.message-sent" &&
990 event.type !== "thread.checkpoint-revert-requested" &&916 event.type !== "thread.checkpoint-revert-requested"
991 event.type !== "thread.turn-diff-completed"
992 ) {917 ) {
993 return Effect.void;918 return Effect.void;
994 }919 }
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+4-1
...@@ -1573,7 +1573,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti...@@ -1573,7 +1573,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
1573 yield* projectionTurnRepository.upsertByTurnId({1573 yield* projectionTurnRepository.upsertByTurnId({
1574 ...existingTurn.value,1574 ...existingTurn.value,
1575 assistantMessageId: event.payload.assistantMessageId,1575 assistantMessageId: event.payload.assistantMessageId,
1576 state: turnStillRunning ? existingTurn.value.state : nextState,1576 state:
1577 turnStillRunning || existingTurn.value.state === "interrupted"
1578 ? existingTurn.value.state
1579 : nextState,
1577 checkpointTurnCount: event.payload.checkpointTurnCount,1580 checkpointTurnCount: event.payload.checkpointTurnCount,
1578 checkpointRef: event.payload.checkpointRef,1581 checkpointRef: event.payload.checkpointRef,
1579 checkpointStatus: event.payload.status,1582 checkpointStatus: event.payload.status,
apps/server/src/orchestration/projector.test.ts+215-90
...@@ -7,6 +7,7 @@ import {...@@ -7,6 +7,7 @@ import {
7 type OrchestrationEvent,7 type OrchestrationEvent,
8} from "@t3tools/contracts";8} from "@t3tools/contracts";
9import * as Effect from "effect/Effect";9import * as Effect from "effect/Effect";
10import { it as effectIt } from "@effect/vitest";
10import { describe, expect, it } from "vite-plus/test";11import { describe, expect, it } from "vite-plus/test";
1112
12import { createEmptyReadModel, projectEvent } from "./projector.ts";13import { createEmptyReadModel, projectEvent } from "./projector.ts";
...@@ -239,110 +240,234 @@ describe("orchestration projector", () => {...@@ -239,110 +240,234 @@ describe("orchestration projector", () => {
239 expect(next.threads).toEqual([]);240 expect(next.threads).toEqual([]);
240 });241 });
241242
242 it("tracks latest turn id from session lifecycle events", async () => {243 effectIt.effect.each([
243 const createdAt = "2026-02-23T08:00:00.000Z";244 ["ready", "completed"],
244 const startedAt = "2026-02-23T08:00:05.000Z";245 ["interrupted", "interrupted"],
245 const model = createEmptyReadModel(createdAt);246 ] as const)(
246247 "preserves the turn state after a %s session captures its checkpoint",
247 const afterCreate = await Effect.runPromise(248 ([status, state]) =>
248 projectEvent(249 Effect.gen(function* () {
249 model,250 const createdAt = "2026-02-23T08:00:00.000Z";
250 makeEvent({251 const startedAt = "2026-02-23T08:00:05.000Z";
251 sequence: 1,252 const model = createEmptyReadModel(createdAt);
252 type: "thread.created",
253 aggregateKind: "thread",
254 aggregateId: "thread-1",
255 occurredAt: createdAt,
256 commandId: "cmd-create",
257 payload: {
258 threadId: "thread-1",
259 projectId: "project-1",
260 title: "demo",
261 modelSelection: {
262 provider: ProviderDriverKind.make("codex"),
263 model: "gpt-5.3-codex",
264 },
265 runtimeMode: "full-access",
266 branch: null,
267 worktreePath: null,
268 createdAt,
269 updatedAt: createdAt,
270 },
271 }),
272 ),
273 );
274253
275 const settledAt = "2026-02-23T08:01:00.000Z";254 const afterCreate = yield* projectEvent(
276 const [afterRunning, afterReady] = await Effect.runPromise(255 model,
277 Effect.flatMap(
278 projectEvent(
279 afterCreate,
280 makeEvent({256 makeEvent({
281 sequence: 2,257 sequence: 1,
282 type: "thread.session-set",258 type: "thread.created",
283 aggregateKind: "thread",259 aggregateKind: "thread",
284 aggregateId: "thread-1",260 aggregateId: "thread-1",
285 occurredAt: startedAt,261 occurredAt: createdAt,
286 commandId: "cmd-running",262 commandId: "cmd-create",
287 payload: {263 payload: {
288 threadId: "thread-1",264 threadId: "thread-1",
289 session: {265 projectId: "project-1",
290 threadId: "thread-1",266 title: "demo",
291 status: "running",267 modelSelection: {
292 providerName: "codex",268 provider: ProviderDriverKind.make("codex"),
293 providerSessionId: "session-1",269 model: "gpt-5.3-codex",
294 providerThreadId: "provider-thread-1",
295 runtimeMode: "approval-required",
296 activeTurnId: "turn-1",
297 lastError: null,
298 updatedAt: startedAt,
299 },270 },
271 runtimeMode: "full-access",
272 branch: null,
273 worktreePath: null,
274 createdAt,
275 updatedAt: createdAt,
300 },276 },
301 }),277 }),
302 ),278 );
303 (running) =>279
304 Effect.map(280 const settledAt = "2026-02-23T08:01:00.000Z";
305 projectEvent(281 const [afterRunning, afterReady] = yield* Effect.flatMap(
306 running,282 projectEvent(
307 makeEvent({283 afterCreate,
308 sequence: 3,284 makeEvent({
309 type: "thread.session-set",285 sequence: 2,
310 aggregateKind: "thread",286 type: "thread.session-set",
311 aggregateId: "thread-1",287 aggregateKind: "thread",
312 occurredAt: settledAt,288 aggregateId: "thread-1",
313 commandId: "cmd-ready",289 occurredAt: startedAt,
314 payload: {290 commandId: "cmd-running",
291 payload: {
292 threadId: "thread-1",
293 session: {
315 threadId: "thread-1",294 threadId: "thread-1",
316 session: {295 status: "running",
296 providerName: "codex",
297 providerSessionId: "session-1",
298 providerThreadId: "provider-thread-1",
299 runtimeMode: "approval-required",
300 activeTurnId: "turn-1",
301 lastError: null,
302 updatedAt: startedAt,
303 },
304 },
305 }),
306 ),
307 (running) =>
308 Effect.map(
309 projectEvent(
310 running,
311 makeEvent({
312 sequence: 3,
313 type: "thread.session-set",
314 aggregateKind: "thread",
315 aggregateId: "thread-1",
316 occurredAt: settledAt,
317 commandId: "cmd-ready",
318 payload: {
317 threadId: "thread-1",319 threadId: "thread-1",
318 status: "ready",320 session: {
319 providerName: "codex",321 threadId: "thread-1",
320 providerSessionId: "session-1",322 status,
321 providerThreadId: "provider-thread-1",323 providerName: "codex",
322 runtimeMode: "approval-required",324 providerSessionId: "session-1",
323 activeTurnId: null,325 providerThreadId: "provider-thread-1",
324 lastError: null,326 runtimeMode: "approval-required",
325 updatedAt: settledAt,327 activeTurnId: null,
328 lastError: null,
329 updatedAt: settledAt,
330 },
326 },331 },
327 },332 }),
328 }),333 ),
334 (ready) => [running, ready] as const,
329 ),335 ),
330 (ready) => [running, ready] as const,336 );
331 ),
332 ),
333 );
334337
335 const thread = afterRunning.threads[0];338 const thread = afterRunning.threads[0];
336 expect(thread?.latestTurn?.turnId).toBe("turn-1");339 expect(thread?.latestTurn?.turnId).toBe("turn-1");
337 expect(thread?.session?.status).toBe("running");340 expect(thread?.session?.status).toBe("running");
338341
339 // Leaving the "running" session status settles the running turn with the342 // Leaving the "running" session status settles the running turn with the
340 // session timestamp as the turn end.343 // session timestamp as the turn end.
341 const settledThread = afterReady.threads[0];344 const settledThread = afterReady.threads[0];
342 expect(settledThread?.latestTurn?.turnId).toBe("turn-1");345 expect(settledThread?.latestTurn?.turnId).toBe("turn-1");
343 expect(settledThread?.latestTurn?.state).toBe("completed");346 expect(settledThread?.latestTurn?.state).toBe(state);
344 expect(settledThread?.latestTurn?.completedAt).toBe(settledAt);347 expect(settledThread?.latestTurn?.completedAt).toBe(settledAt);
345 });348
349 const captured = yield* projectEvent(
350 afterReady,
351 makeEvent({
352 sequence: 4,
353 type: "thread.turn-diff-completed",
354 aggregateKind: "thread",
355 aggregateId: "thread-1",
356 occurredAt: settledAt,
357 commandId: "cmd-final-checkpoint",
358 payload: {
359 threadId: "thread-1",
360 turnId: "turn-1",
361 checkpointTurnCount: 1,
362 checkpointRef: "refs/t3/checkpoints/thread-1/turn/1",
363 status: "ready",
364 files: [],
365 assistantMessageId: "assistant:turn-1",
366 completedAt: settledAt,
367 },
368 }),
369 );
370 expect(captured.threads[0]?.latestTurn?.state).toBe(state);
371 expect(captured.threads[0]?.checkpoints[0]?.status).toBe("ready");
372 }),
373 );
374
375 effectIt.effect.each([null, "ready", "interrupted", "stopped"] as const)(
376 "replaces a missing checkpoint without inventing interruption for a %s session",
377 (sessionStatus) =>
378 Effect.gen(function* () {
379 const now = "2026-09-04T23:00:00.000Z";
380 const threadId = "thread-placeholder";
381 const event = (sequence: number, type: OrchestrationEvent["type"], payload: unknown) =>
382 makeEvent({
383 sequence,
384 type,
385 payload,
386 aggregateKind: "thread",
387 aggregateId: threadId,
388 occurredAt: now,
389 commandId: `placeholder-${sequence}`,
390 });
391 let model = yield* projectEvent(
392 createEmptyReadModel(now),
393 event(1, "thread.created", {
394 threadId,
395 projectId: "project-1",
396 title: "Placeholder",
397 modelSelection: { instanceId: "codex", model: "test" },
398 runtimeMode: "full-access",
399 interactionMode: "default",
400 branch: null,
401 worktreePath: null,
402 createdAt: now,
403 updatedAt: now,
404 }),
405 );
406 const checkpoint = {
407 threadId,
408 turnId: "turn-placeholder",
409 checkpointTurnCount: 1,
410 checkpointRef: "provider-diff:placeholder",
411 files: [],
412 assistantMessageId: "assistant:placeholder",
413 completedAt: now,
414 };
415 if (sessionStatus === "interrupted" || sessionStatus === "stopped") {
416 model = yield* projectEvent(
417 model,
418 event(2, "thread.session-set", {
419 threadId,
420 session: {
421 threadId,
422 status: "running",
423 providerName: "codex",
424 runtimeMode: "full-access",
425 activeTurnId: "turn-placeholder",
426 lastError: null,
427 updatedAt: now,
428 },
429 }),
430 );
431 }
432 model = yield* projectEvent(
433 model,
434 event(3, "thread.turn-diff-completed", {
435 ...checkpoint,
436 status: "missing",
437 }),
438 );
439 if (sessionStatus !== null) {
440 model = yield* projectEvent(
441 model,
442 event(4, "thread.session-set", {
443 threadId,
444 session: {
445 threadId,
446 status: sessionStatus,
447 providerName: "codex",
448 runtimeMode: "full-access",
449 activeTurnId: null,
450 lastError: null,
451 updatedAt: now,
452 },
453 }),
454 );
455 }
456 model = yield* projectEvent(
457 model,
458 event(5, "thread.turn-diff-completed", {
459 ...checkpoint,
460 status: "ready",
461 checkpointRef: "refs/t3/checkpoints/thread-placeholder/turn/1",
462 }),
463 );
464 expect(model.threads[0]?.latestTurn?.state).toBe(
465 sessionStatus === "interrupted" || sessionStatus === "stopped"
466 ? "interrupted"
467 : "completed",
468 );
469 }),
470 );
346471
347 it("updates canonical thread runtime mode from thread.runtime-mode-set", async () => {472 it("updates canonical thread runtime mode from thread.runtime-mode-set", async () => {
348 const createdAt = "2026-02-23T08:00:00.000Z";473 const createdAt = "2026-02-23T08:00:00.000Z";
apps/server/src/orchestration/projector.ts+6-2
...@@ -64,7 +64,7 @@ function retainThreadActivities(activities: OrchestrationThread["activities"]) {...@@ -64,7 +64,7 @@ function retainThreadActivities(activities: OrchestrationThread["activities"]) {
6464
65function checkpointStatusToLatestTurnState(status: "ready" | "missing" | "error") {65function checkpointStatusToLatestTurnState(status: "ready" | "missing" | "error") {
66 if (status === "error") return "error" as const;66 if (status === "error") return "error" as const;
67 if (status === "missing") return "interrupted" as const;67 // Match SQL and client projections: a missing git ref is not an interruption.
68 return "completed" as const;68 return "completed" as const;
69}69}
7070
...@@ -746,7 +746,11 @@ export function projectEvent(...@@ -746,7 +746,11 @@ export function projectEvent(
746 ? thread.latestTurn746 ? thread.latestTurn
747 : {747 : {
748 turnId: payload.turnId,748 turnId: payload.turnId,
749 state: checkpointStatusToLatestTurnState(payload.status),749 state:
750 thread.latestTurn?.turnId === payload.turnId &&
751 thread.latestTurn.state === "interrupted"
752 ? "interrupted"
753 : checkpointStatusToLatestTurnState(payload.status),
750 requestedAt:754 requestedAt:
751 thread.latestTurn?.turnId === payload.turnId755 thread.latestTurn?.turnId === payload.turnId
752 ? thread.latestTurn.requestedAt756 ? thread.latestTurn.requestedAt
packages/client-runtime/src/state/threadReducer.test.ts+46-27
...@@ -1129,33 +1129,52 @@ describe("applyThreadDetailEvent", () => {...@@ -1129,33 +1129,52 @@ describe("applyThreadDetailEvent", () => {
1129 });1129 });
11301130
1131 describe("thread.turn-diff-completed", () => {1131 describe("thread.turn-diff-completed", () => {
1132 it("adds a checkpoint and updates latestTurn", () => {1132 it.each([null, "interrupted"] as const)(
1133 const result = applyThreadDetailEvent(baseThread, {1133 "adds a checkpoint without replacing a %s turn outcome",
1134 ...baseEventFields,1134 (previousState) => {
1135 sequence: 13,1135 const result = applyThreadDetailEvent(
1136 occurredAt: "2026-04-01T12:00:00.000Z",1136 {
1137 aggregateKind: "thread",1137 ...baseThread,
1138 aggregateId: ThreadId.make("thread-1"),1138 latestTurn:
1139 type: "thread.turn-diff-completed",1139 previousState === null
1140 payload: {1140 ? null
1141 threadId: ThreadId.make("thread-1"),1141 : {
1142 turnId: TurnId.make("turn-1"),1142 turnId: TurnId.make("turn-1"),
1143 checkpointTurnCount: 1,1143 state: previousState,
1144 checkpointRef: CheckpointRef.make("ref-1"),1144 requestedAt: "2026-04-01T11:00:00.000Z",
1145 status: "ready",1145 startedAt: "2026-04-01T11:00:00.000Z",
1146 files: [],1146 completedAt: "2026-04-01T12:00:00.000Z",
1147 assistantMessageId: MessageId.make("msg-3"),1147 assistantMessageId: null,
1148 completedAt: "2026-04-01T12:00:00.000Z",1148 },
1149 },1149 },
1150 });1150 {
11511151 ...baseEventFields,
1152 expect(result.kind).toBe("updated");1152 sequence: 13,
1153 if (result.kind === "updated") {1153 occurredAt: "2026-04-01T12:00:00.000Z",
1154 expect(result.thread.checkpoints).toHaveLength(1);1154 aggregateKind: "thread",
1155 expect(result.thread.latestTurn?.turnId).toBe("turn-1");1155 aggregateId: ThreadId.make("thread-1"),
1156 expect(result.thread.latestTurn?.state).toBe("completed");1156 type: "thread.turn-diff-completed",
1157 }1157 payload: {
1158 });1158 threadId: ThreadId.make("thread-1"),
1159 turnId: TurnId.make("turn-1"),
1160 checkpointTurnCount: 1,
1161 checkpointRef: CheckpointRef.make("ref-1"),
1162 status: "ready",
1163 files: [],
1164 assistantMessageId: MessageId.make("msg-3"),
1165 completedAt: "2026-04-01T12:00:00.000Z",
1166 },
1167 },
1168 );
1169
1170 expect(result.kind).toBe("updated");
1171 if (result.kind === "updated") {
1172 expect(result.thread.checkpoints).toHaveLength(1);
1173 expect(result.thread.latestTurn?.turnId).toBe("turn-1");
1174 expect(result.thread.latestTurn?.state).toBe(previousState ?? "completed");
1175 }
1176 },
1177 );
1159 });1178 });
11601179
1161 describe("thread.reverted", () => {1180 describe("thread.reverted", () => {
packages/client-runtime/src/state/threadReducer.ts+4-1
...@@ -521,7 +521,10 @@ export function applyThreadDetailEvent(...@@ -521,7 +521,10 @@ export function applyThreadDetailEvent(
521 (thread.latestTurn === null || thread.latestTurn.turnId === event.payload.turnId)521 (thread.latestTurn === null || thread.latestTurn.turnId === event.payload.turnId)
522 ? {522 ? {
523 turnId: event.payload.turnId,523 turnId: event.payload.turnId,
524 state: checkpointStatusToTurnState(event.payload.status),524 state:
525 thread.latestTurn?.state === "interrupted"
526 ? "interrupted"
527 : checkpointStatusToTurnState(event.payload.status),
525 requestedAt: thread.latestTurn?.requestedAt ?? event.payload.completedAt,528 requestedAt: thread.latestTurn?.requestedAt ?? event.payload.completedAt,
526 startedAt: thread.latestTurn?.startedAt ?? event.payload.completedAt,529 startedAt: thread.latestTurn?.startedAt ?? event.payload.completedAt,
527 completedAt: event.payload.completedAt,530 completedAt: event.payload.completedAt,