Skip to content
Closed
Show file tree
Hide file tree
Changes from 6 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
259 changes: 255 additions & 4 deletions apps/server/src/orchestration/Layers/CheckpointReactor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@ function createProviderServiceHarness(
const rollbackConversation = vi.fn(
(_input: { readonly threadId: ThreadId; readonly numTurns: number }) => Effect.void,
);
let activeTurnId: TurnId | undefined;

const unsupported = <A>() =>
Effect.die(new Error("Unsupported provider call in test")) as Effect.Effect<A, never>;
Expand All @@ -101,6 +102,7 @@ function createProviderServiceHarness(
cwd: sessionCwd,
createdAt: now,
updatedAt: now,
...(activeTurnId ? { activeTurnId } : {}),
},
] satisfies ReadonlyArray<ProviderSession>)
: Effect.succeed([] as ReadonlyArray<ProviderSession>);
Expand Down Expand Up @@ -139,6 +141,9 @@ function createProviderServiceHarness(
service,
rollbackConversation,
emit,
setActiveTurnId: (turnId: TurnId | undefined) => {
activeTurnId = turnId;
},
};
}

Expand All @@ -147,21 +152,24 @@ async function waitForThread(
readonly threads: ReadonlyArray<{
readonly id: ThreadId;
readonly latestTurn: { readonly turnId: string } | null;
readonly checkpoints: ReadonlyArray<{ readonly checkpointTurnCount: number }>;
readonly checkpoints: ReadonlyArray<{
readonly checkpointTurnCount: number;
readonly completedAt: string;
}>;
readonly activities: ReadonlyArray<{ readonly kind: string }>;
}>;
}>,
predicate: (thread: {
latestTurn: { turnId: string } | null;
checkpoints: ReadonlyArray<{ checkpointTurnCount: number }>;
checkpoints: ReadonlyArray<{ checkpointTurnCount: number; completedAt: string }>;
activities: ReadonlyArray<{ kind: string }>;
}) => boolean,
timeoutMs = 15_000,
) {
const deadline = (await Effect.runPromise(Clock.currentTimeMillis)) + timeoutMs;
const poll = async (): Promise<{
latestTurn: { turnId: string } | null;
checkpoints: ReadonlyArray<{ checkpointTurnCount: number }>;
checkpoints: ReadonlyArray<{ checkpointTurnCount: number; completedAt: string }>;
activities: ReadonlyArray<{ kind: string }>;
}> => {
const snapshot = await readModel();
Expand All @@ -180,7 +188,7 @@ async function waitForThread(

async function waitForEvent(
engine: OrchestrationEngineShape,
predicate: (event: { type: string }) => boolean,
predicate: (event: { type: string; occurredAt: string }) => boolean,
timeoutMs = 15_000,
) {
const deadline = (await Effect.runPromise(Clock.currentTimeMillis)) + timeoutMs;
Expand Down Expand Up @@ -339,6 +347,7 @@ describe("CheckpointReactor", () => {
Layer.provideMerge(orchestrationLayer),
Layer.provideMerge(projectionSnapshotLayer),
Layer.provideMerge(RuntimeReceiptBusLive),
Layer.provideMerge(SqlitePersistenceMemory),
Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),
Layer.provideMerge(vcsStatusBroadcasterLayer),
Layer.provideMerge(CheckpointStore.layer.pipe(Layer.provide(VcsDriverRegistry.layer))),
Expand All @@ -364,6 +373,10 @@ describe("CheckpointReactor", () => {
scope = await Effect.runPromise(Scope.make("sequential"));
await Effect.runPromise(reactor.start().pipe(Scope.provide(scope)));
const drain = () => Effect.runPromise(reactor.drain);
const captureCheckpoint = (input: CheckpointStore.CaptureCheckpointInput) =>
runtime!.runPromise(checkpointStore.captureCheckpoint(input));
const dispatch = (command: Parameters<OrchestrationEngineShape["dispatch"]>[0]) =>
runtime!.runPromise(engine.dispatch(command));

const createdAt = "2026-01-01T00:00:00.000Z";
await Effect.runPromise(
Expand Down Expand Up @@ -449,6 +462,8 @@ describe("CheckpointReactor", () => {
engine,
readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()),
provider,
captureCheckpoint,
dispatch,
cwd,
drain,
};
Expand Down Expand Up @@ -530,6 +545,60 @@ describe("CheckpointReactor", () => {
).toBe("v2\n");
});

it("replaces a legacy mid-turn checkpoint with the filesystem state at completion", async () => {
const harness = await createHarness({ seedFilesystemCheckpoints: false });
const threadId = ThreadId.make("thread-1");
const turnId = asTurnId("turn-1");
const checkpointRef = checkpointRefForThreadTurn(threadId, 1);
const midTurnAt = "2026-01-01T00:00:10.000Z";
const completedAt = "2026-01-01T00:00:30.000Z";

await harness.captureCheckpoint({
cwd: harness.cwd,
checkpointRef: checkpointRefForThreadTurn(threadId, 0),
});
NodeFS.writeFileSync(NodePath.join(harness.cwd, "README.md"), "mid-turn\n", "utf8");
await harness.captureCheckpoint({ cwd: harness.cwd, checkpointRef });
await harness.dispatch({
type: "thread.turn.diff.complete",
commandId: CommandId.make("cmd-legacy-mid-turn-diff"),
threadId,
turnId,
completedAt: midTurnAt,
checkpointRef,
status: "ready",
files: [{ path: "README.md", kind: "modified", additions: 1, deletions: 1 }],
checkpointTurnCount: 1,
createdAt: midTurnAt,
});
await harness.drain();
expect(gitShowFileAtRef(harness.cwd, checkpointRef, "README.md")).toBe("mid-turn\n");

NodeFS.writeFileSync(NodePath.join(harness.cwd, "README.md"), "final\n", "utf8");
harness.provider.emit({
type: "turn.completed",
eventId: EventId.make("evt-turn-completed-after-legacy-checkpoint"),
provider: ProviderDriverKind.make("codex"),
createdAt: completedAt,
threadId,
turnId,
payload: { state: "completed" },
});

await waitForEvent(
harness.engine,
(event) => event.type === "thread.turn-diff-completed" && event.occurredAt === completedAt,
);
const thread = await waitForThread(
harness.readModel,
(entry) =>
entry.checkpoints.length === 1 && entry.checkpoints[0]?.completedAt === completedAt,
);

expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1);
expect(gitShowFileAtRef(harness.cwd, checkpointRef, "README.md")).toBe("final\n");
});

it("refreshes local git status state on turn completion using the session cwd", async () => {
const gitStatusRefreshCalls: string[] = [];
const harness = await createHarness({
Expand Down Expand Up @@ -704,6 +773,188 @@ describe("CheckpointReactor", () => {
expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1);
});

it("preserves the active turn when a stale turn.started event arrives", async () => {
const harness = await createHarness({ seedFilesystemCheckpoints: false });
const threadId = ThreadId.make("thread-1");
const activeTurnId = asTurnId("turn-active");
const completedAt = "2026-01-01T00:00:10.000Z";

harness.provider.emit({
type: "turn.started",
eventId: EventId.make("evt-turn-started-active"),
provider: ProviderDriverKind.make("codex"),
createdAt: "2026-01-01T00:00:00.000Z",
threadId,
turnId: activeTurnId,
});
await waitForGitRefExists(harness.cwd, checkpointRefForThreadTurn(threadId, 0));

harness.provider.emit({
type: "turn.started",
eventId: EventId.make("evt-turn-started-stale"),
provider: ProviderDriverKind.make("codex"),
createdAt: "2026-01-01T00:00:05.000Z",
threadId,
turnId: asTurnId("turn-stale"),
});
NodeFS.writeFileSync(NodePath.join(harness.cwd, "README.md"), "active turn\n", "utf8");
harness.provider.emit({
type: "turn.completed",
eventId: EventId.make("evt-turn-completed-active-after-stale-start"),
provider: ProviderDriverKind.make("codex"),
createdAt: completedAt,
threadId,
turnId: activeTurnId,
payload: { state: "completed" },
});

await waitForEvent(
harness.engine,
(event) => event.type === "thread.turn-diff-completed" && event.occurredAt === completedAt,
);
const thread = await waitForThread(
harness.readModel,
(entry) => entry.checkpoints[0]?.completedAt === completedAt,
);
expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1);
});

it("tracks a new turn when an accepted steer supersedes the active turn", async () => {
const harness = await createHarness({ seedFilesystemCheckpoints: false });
const threadId = ThreadId.make("thread-1");
const oldTurnId = asTurnId("turn-steered-over");
const newTurnId = asTurnId("turn-from-steer");
const createdAt = "2026-01-01T00:00:00.000Z";
const completedAt = "2026-01-01T00:00:10.000Z";

await harness.dispatch({
type: "thread.session.set",
commandId: CommandId.make("cmd-session-set-before-steer"),
threadId,
session: {
threadId,
status: "running",
providerName: "codex",
runtimeMode: "approval-required",
activeTurnId: oldTurnId,
lastError: null,
updatedAt: createdAt,
},
createdAt,
});
harness.provider.emit({
type: "turn.started",
eventId: EventId.make("evt-turn-started-before-steer"),
provider: ProviderDriverKind.make("codex"),
createdAt,
threadId,
turnId: oldTurnId,
});
await waitForGitRefExists(harness.cwd, checkpointRefForThreadTurn(threadId, 0));

await harness.dispatch({
type: "thread.turn.start",
commandId: CommandId.make("cmd-turn-start-steer"),
threadId,
message: {
messageId: MessageId.make("msg-steer"),
role: "user",
text: "Use the new direction",
attachments: [],
},
interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
runtimeMode: "approval-required",
createdAt: "2026-01-01T00:00:05.000Z",
});
harness.provider.setActiveTurnId(newTurnId);
harness.provider.emit({
type: "turn.started",
eventId: EventId.make("evt-turn-started-from-steer"),
provider: ProviderDriverKind.make("codex"),
createdAt: "2026-01-01T00:00:05.000Z",
threadId,
turnId: newTurnId,
});
NodeFS.writeFileSync(NodePath.join(harness.cwd, "README.md"), "steered turn\n", "utf8");
harness.provider.emit({
type: "turn.completed",
eventId: EventId.make("evt-turn-completed-from-steer"),
provider: ProviderDriverKind.make("codex"),
createdAt: completedAt,
threadId,
turnId: newTurnId,
payload: { state: "completed" },
});

await waitForEvent(
harness.engine,
(event) => event.type === "thread.turn-diff-completed" && event.occurredAt === completedAt,
);
const thread = await waitForThread(
harness.readModel,
(entry) => entry.checkpoints[0]?.completedAt === completedAt,
);
expect(thread.latestTurn?.turnId).toBe(newTurnId);
});

it("captures a completion after the read model advances to a follow-up turn", async () => {
const harness = await createHarness({ seedFilesystemCheckpoints: false });
const threadId = ThreadId.make("thread-1");
const completedTurnId = asTurnId("turn-completed");
const completedAt = "2026-01-01T00:00:10.000Z";

harness.provider.emit({
type: "turn.started",
eventId: EventId.make("evt-turn-started-before-follow-up"),
provider: ProviderDriverKind.make("codex"),
createdAt: "2026-01-01T00:00:00.000Z",
threadId,
turnId: completedTurnId,
});
await waitForGitRefExists(harness.cwd, checkpointRefForThreadTurn(threadId, 0));

await harness.dispatch({
type: "thread.session.set",
commandId: CommandId.make("cmd-session-advanced-to-follow-up"),
threadId,
session: {
threadId,
status: "running",
providerName: "codex",
runtimeMode: "approval-required",
activeTurnId: asTurnId("turn-follow-up"),
lastError: null,
updatedAt: "2026-01-01T00:00:20.000Z",
},
createdAt: "2026-01-01T00:00:20.000Z",
});
NodeFS.writeFileSync(NodePath.join(harness.cwd, "README.md"), "completed turn\n", "utf8");

harness.provider.emit({
type: "turn.completed",
eventId: EventId.make("evt-turn-completed-after-follow-up-started"),
provider: ProviderDriverKind.make("codex"),
createdAt: completedAt,
threadId,
turnId: completedTurnId,
payload: { state: "completed" },
});

await waitForEvent(
harness.engine,
(event) => event.type === "thread.turn-diff-completed" && event.occurredAt === completedAt,
);
const thread = await waitForThread(
harness.readModel,
(entry) => entry.checkpoints[0]?.completedAt === completedAt,
);

expect(thread.checkpoints).toHaveLength(1);
expect(
gitShowFileAtRef(harness.cwd, checkpointRefForThreadTurn(threadId, 1), "README.md"),
).toBe("completed turn\n");
});

it("captures pre-turn and completion checkpoints for claude runtime events", async () => {
const harness = await createHarness({
seedFilesystemCheckpoints: false,
Expand Down
Loading
Loading