History

fix: retain authoritative checkpoint turn counts

Rishin Georgy Anil committed 3 months ago

Commit ad968861 parent c74caeb2 files · +97 −4Browse files

Unified · Split

2 files
  1. apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +75 −0
  2. apps/server/src/orchestration/Layers/CheckpointReactor.ts +22 −4
apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+75−0
@@ -787,6 +787,81 @@ describe("CheckpointReactor", () => { · 786 unmodified lines
787 ).toBe("v2\n");787 ).toBe("v2\n");
788 });788 });
789 789
790 it("keeps the authoritative turn count when a placeholder is processed late", async () => {
791 const harness = await createHarness({ seedFilesystemCheckpoints: false });
792 const threadId = ThreadId.make("thread-1");
793 const createdAt = "2026-01-01T00:00:00.000Z";
794
795 for (const turnNumber of [1, 2]) {
796 const turnId = asTurnId(`late-placeholder-turn-${turnNumber}`);
797 await harness.runPromise(
798 harness.engine.dispatch({
799 type: "thread.turn.start",
800 commandId: CommandId.make(`cmd-late-placeholder-user-${turnNumber}`),
801 threadId,
802 message: {
803 messageId: MessageId.make(`late-placeholder-user-${turnNumber}`),
804 role: "user",
805 text: `User ${turnNumber}`,
806 attachments: [],
807 },
808 runtimeMode: "approval-required",
809 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
810 createdAt,
811 }),
812 );
813 await harness.runPromise(
814 harness.engine.dispatch({
815 type: "thread.message.assistant.delta",
816 commandId: CommandId.make(`cmd-late-placeholder-assistant-delta-${turnNumber}`),
817 threadId,
818 messageId: MessageId.make(`late-placeholder-assistant-${turnNumber}`),
819 delta: `Assistant ${turnNumber}`,
820 turnId,
821 createdAt,
822 }),
823 );
824 await harness.runPromise(
825 harness.engine.dispatch({
826 type: "thread.message.assistant.complete",
827 commandId: CommandId.make(`cmd-late-placeholder-assistant-complete-${turnNumber}`),
828 threadId,
829 messageId: MessageId.make(`late-placeholder-assistant-${turnNumber}`),
830 turnId,
831 createdAt,
832 }),
833 );
834 }
835
836 await harness.runPromise(
837 harness.engine.dispatch({
838 type: "thread.turn.diff.complete",
839 commandId: CommandId.make("cmd-late-placeholder-diff"),
840 threadId,
841 turnId: asTurnId("late-placeholder-turn-1"),
842 completedAt: createdAt,
843 checkpointRef: checkpointRefForThreadTurn(threadId, 1),
844 status: "missing",
845 files: [],
846 checkpointTurnCount: 1,
847 createdAt,
848 }),
849 );
850
851 const thread = await waitForThread(
852 harness.readModel,
853 (entry) =>
854 entry.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 1) &&
855 entry.activities.some((activity) => activity.kind === "checkpoint.captured"),
856 );
857 expect(thread.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 1)).toBe(
858 true,
859 );
860 expect(thread.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 2)).toBe(
861 false,
862 );
863 });
864
790 it("ignores non-v2 checkpoint.captured runtime events", async () => {865 it("ignores non-v2 checkpoint.captured runtime events", async () => {
791 const harness = await createHarness();866 const harness = await createHarness();
792 const createdAt = "2026-01-01T00:00:00.000Z";867 const createdAt = "2026-01-01T00:00:00.000Z";
apps/server/src/orchestration/Layers/CheckpointReactor.ts+22−4
@@ -62,6 +62,25 @@ export function countConversationTurns(messages: ReadonlyArray<{ readonly role: · 61 unmodified lines
62 return messages.reduce((count, message) => count + (message.role === "user" ? 1 : 0), 0);62 return messages.reduce((count, message) => count + (message.role === "user" ? 1 : 0), 0);
63}63}
64 64
65function conversationTurnCountForTurn(
66 messages: ReadonlyArray<{
67 readonly role: string;
68 readonly turnId: TurnId | null;
69 }>,
70 turnId: TurnId,
71): number | undefined {
72 let turnCount = 0;
73 for (const message of messages) {
74 if (message.role === "user") {
75 turnCount += 1;
76 }
77 if (message.role === "assistant" && sameId(message.turnId, turnId)) {
78 return turnCount > 0 ? turnCount : undefined;
79 }
80 }
81 return undefined;
82}
83
65function maxCheckpointTurnCount(84function maxCheckpointTurnCount(
66 checkpoints: ReadonlyArray<{ readonly checkpointTurnCount: number }>,85 checkpoints: ReadonlyArray<{ readonly checkpointTurnCount: number }>,
67): number {86): number {
@@ -412,10 +431,9 @@ const make = Effect.gen(function* () { · 344 unmodified lines
412 (checkpoint) => checkpoint.turnId === turnId && checkpoint.status === "missing",431 (checkpoint) => checkpoint.turnId === turnId && checkpoint.status === "missing",
413 );432 );
414 const currentTurnCount = maxCheckpointTurnCount(thread.checkpoints);433 const currentTurnCount = maxCheckpointTurnCount(thread.checkpoints);
415 const conversationTurnCount = countConversationTurns(thread.messages);
416 const nextTurnCount = existingPlaceholder434 const nextTurnCount = existingPlaceholder
417 ? Math.max(existingPlaceholder.checkpointTurnCount, conversationTurnCount)435 ? existingPlaceholder.checkpointTurnCount
418 : Math.max(currentTurnCount + 1, conversationTurnCount);436 : (conversationTurnCountForTurn(thread.messages, turnId) ?? currentTurnCount + 1);
419 437
420 yield* captureAndDispatchCheckpoint({438 yield* captureAndDispatchCheckpoint({
421 threadId: thread.id,439 threadId: thread.id,
@@ -485,7 +503,7 @@ const make = Effect.gen(function* () { · 63 unmodified lines
485 turnId,503 turnId,
486 thread,504 thread,
487 cwd: checkpointCwd,505 cwd: checkpointCwd,
488 turnCount: Math.max(checkpointTurnCount, countConversationTurns(thread.messages)),506 turnCount: checkpointTurnCount,
489 status: "ready",507 status: "ready",
490 assistantMessageId: event.payload.assistantMessageId ?? undefined,508 assistantMessageId: event.payload.assistantMessageId ?? undefined,
491 createdAt: event.payload.completedAt,509 createdAt: event.payload.completedAt,