Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
60 changes: 39 additions & 21 deletions apps/server/src/orchestration/Layers/CheckpointReactor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1916,27 +1916,45 @@ describe("CheckpointReactor", () => {
});
});

it("appends an error activity when revert is requested without an active session", async () => {
const harness = await createHarness({ hasSession: false });
const createdAt = "2026-01-01T00:00:00.000Z";

await Effect.runPromise(
harness.engine.dispatch({
type: "thread.checkpoint.revert",
commandId: CommandId.make("cmd-revert-no-session"),
threadId: ThreadId.make("thread-1"),
turnCount: 1,
createdAt,
}),
);
it.each([false, true])(
"reverts without an active session using project cwd fallback: %s",
async (useProjectCwd) => {
const harness = await createHarness({
hasSession: false,
...(useProjectCwd ? { threadWorktreePath: null } : {}),
});
const createdAt = "2026-01-01T00:00:00.000Z";

const thread = await waitForThread(harness.readModel, (entry) =>
entry.activities.some((activity) => activity.kind === "checkpoint.revert.failed"),
);
await Effect.runPromise(
harness.engine.dispatch({
type: "thread.turn.diff.complete",
commandId: CommandId.make("cmd-diff-before-session-recovery"),
threadId: ThreadId.make("thread-1"),
turnId: asTurnId("turn-1"),
completedAt: createdAt,
checkpointRef: checkpointRefForThreadTurn(ThreadId.make("thread-1"), 1),
status: "ready",
files: [],
checkpointTurnCount: 1,
createdAt,
}),
);
await Effect.runPromise(
harness.engine.dispatch({
type: "thread.checkpoint.revert",
commandId: CommandId.make("cmd-revert-no-session"),
threadId: ThreadId.make("thread-1"),
turnCount: 0,
createdAt,
}),
);

expect(thread.activities.some((activity) => activity.kind === "checkpoint.revert.failed")).toBe(
true,
);
expect(harness.provider.rollbackConversation).not.toHaveBeenCalled();
});
await waitForEvent(harness.engine, (event) => event.type === "thread.reverted");
expect(harness.provider.rollbackConversation).toHaveBeenCalledWith({
threadId: ThreadId.make("thread-1"),
numTurns: 1,
});
expect(NodeFS.readFileSync(NodePath.join(harness.cwd, "README.md"), "utf8")).toBe("v1\n");
},
);
});
28 changes: 12 additions & 16 deletions apps/server/src/orchestration/Layers/CheckpointReactor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -699,21 +699,17 @@ const make = Effect.gen(function* () {
return;
}

const sessionRuntime = yield* resolveSessionRuntimeForThread(event.payload.threadId);
if (Option.isNone(sessionRuntime)) {
yield* appendRevertFailureActivity({
threadId: event.payload.threadId,
turnCount: event.payload.turnCount,
detail: "No active provider session with workspace cwd is bound to this thread.",
createdAt: now,
}).pipe(Effect.catch(() => Effect.void));
return;
}
if (!(yield* checkpointStore.isGitRepository(sessionRuntime.value.cwd))) {
const checkpointCwd = yield* resolveCheckpointCwd({
threadId: event.payload.threadId,
thread,
projects: yield* resolveThreadProjects(thread.projectId),
preferSessionRuntime: true,
});
if (!checkpointCwd) {
yield* appendRevertFailureActivity({
threadId: event.payload.threadId,
turnCount: event.payload.turnCount,
detail: "Checkpoints are unavailable because this project is not a git repository.",
detail: "Checkpoint workspace is unavailable or is not a git repository.",
createdAt: now,
}).pipe(Effect.catch(() => Effect.void));
return;
Expand Down Expand Up @@ -754,7 +750,7 @@ const make = Effect.gen(function* () {
yield* providerService.assertConversationRollbackSupported(event.payload.threadId);

const restored = yield* checkpointStore.restoreCheckpoint({
cwd: sessionRuntime.value.cwd,
cwd: checkpointCwd,
checkpointRef: targetCheckpointRef,
fallbackToHead: event.payload.turnCount === 0,
});
Expand All @@ -770,12 +766,12 @@ const make = Effect.gen(function* () {

// Refresh the workspace entry index so the @-mention file picker
// reflects the reverted filesystem state.
yield* workspaceEntries.refresh(sessionRuntime.value.cwd);
yield* workspaceEntries.refresh(checkpointCwd);

const rolledBackTurns = Math.max(0, currentTurnCount - event.payload.turnCount);
if (rolledBackTurns > 0) {
yield* providerService.rollbackConversation({
threadId: sessionRuntime.value.threadId,
threadId: event.payload.threadId,
numTurns: rolledBackTurns,
});
}
Expand All @@ -789,7 +785,7 @@ const make = Effect.gen(function* () {

if (staleCheckpointRefs.length > 0) {
yield* checkpointStore.deleteCheckpointRefs({
cwd: sessionRuntime.value.cwd,
cwd: checkpointCwd,
checkpointRefs: staleCheckpointRefs,
});
}
Expand Down
87 changes: 82 additions & 5 deletions apps/server/src/provider/Layers/ClaudeAdapter.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -170,8 +170,11 @@ function makeHarness(config?: {
readonly instanceId?: ProviderInstanceId;
readonly scopedLimitNames?: ClaudeAdapterLiveOptions["scopedLimitNames"];
readonly environment?: ClaudeAdapterLiveOptions["environment"];
readonly getSessionMessages?: ClaudeAdapterLiveOptions["getSessionMessages"];
readonly forkSession?: ClaudeAdapterLiveOptions["forkSession"];
}) {
const query = new FakeClaudeQuery();
const queries = [query];
let createInput:
| {
readonly prompt: AsyncIterable<SDKUserMessage>;
Expand All @@ -184,9 +187,12 @@ function makeHarness(config?: {
...(config?.instanceId ? { instanceId: config.instanceId } : {}),
...(config?.scopedLimitNames ? { scopedLimitNames: config.scopedLimitNames } : {}),
modelCatalog: Effect.succeed(SYNTHETIC_CLAUDE_MODEL_CATALOG),
...(config?.getSessionMessages ? { getSessionMessages: config.getSessionMessages } : {}),
...(config?.forkSession ? { forkSession: config.forkSession } : {}),
createQuery: (input) => {
if (createInput && config?.getSessionMessages) queries.push(new FakeClaudeQuery());
createInput = input;
return query;
return queries.at(-1)!;
},
...(config?.nativeEventLogger
? {
Expand Down Expand Up @@ -6257,9 +6263,65 @@ describe("ClaudeAdapterLive", () => {
});

it.effect(
"supports rollbackThread by trimming in-memory turns and preserving earlier turns",
"forks Claude history at the retained turn and resets to a fresh session at zero",
() => {
const harness = makeHarness();
const forkCalls: Array<Parameters<NonNullable<ClaudeAdapterLiveOptions["forkSession"]>>> = [];
const harness = makeHarness({
forkSession: async (...args) => {
forkCalls.push(args);
return { sessionId: "550e8400-e29b-41d4-a716-446655440020" };
},
getSessionMessages: async () => [
{
type: "user",
uuid: "user-1",
session_id: "550e8400-e29b-41d4-a716-446655440010",
parent_tool_use_id: null,
parent_agent_id: null,
message: { content: "first" },
},
{
type: "assistant",
uuid: "assistant-1",
session_id: "550e8400-e29b-41d4-a716-446655440010",
parent_tool_use_id: null,
parent_agent_id: null,
message: { content: [] },
},
{
type: "user",
uuid: "tool-result-1",
session_id: "550e8400-e29b-41d4-a716-446655440010",
parent_tool_use_id: null,
parent_agent_id: null,
message: { content: [{ type: "tool_result" }] },
},
{
type: "assistant",
uuid: "assistant-1-final",
session_id: "550e8400-e29b-41d4-a716-446655440010",
parent_tool_use_id: null,
parent_agent_id: null,
message: { content: [] },
},
{
type: "user",
uuid: "user-2",
session_id: "550e8400-e29b-41d4-a716-446655440010",
parent_tool_use_id: null,
parent_agent_id: null,
message: { content: "second" },
},
{
type: "assistant",
uuid: "assistant-2",
session_id: "550e8400-e29b-41d4-a716-446655440010",
parent_tool_use_id: null,
parent_agent_id: null,
message: { content: [] },
},
],
});
return Effect.gen(function* () {
const adapter = yield* ClaudeAdapter;

Expand All @@ -6285,7 +6347,7 @@ describe("ClaudeAdapterLive", () => {
subtype: "success",
is_error: false,
errors: [],
session_id: "sdk-session-rollback",
session_id: "550e8400-e29b-41d4-a716-446655440010",
uuid: "result-first",
} as unknown as SDKMessage);

Expand All @@ -6311,7 +6373,7 @@ describe("ClaudeAdapterLive", () => {
subtype: "success",
is_error: false,
errors: [],
session_id: "sdk-session-rollback",
session_id: "550e8400-e29b-41d4-a716-446655440010",
uuid: "result-second",
} as unknown as SDKMessage);

Expand All @@ -6331,6 +6393,21 @@ describe("ClaudeAdapterLive", () => {
const threadAfterRollback = yield* adapter.readThread(session.threadId);
assert.equal(threadAfterRollback.turns.length, 1);
assert.equal(threadAfterRollback.turns[0]?.id, firstTurn.turnId);
assert.equal(harness.query.closeCalls, 1);
const forkOptions = harness.getLastCreateQueryInput()?.options;
assert.deepEqual(forkCalls, [
["550e8400-e29b-41d4-a716-446655440010", { upToMessageId: "assistant-1-final" }],
]);
assert.equal(forkOptions?.resume, "550e8400-e29b-41d4-a716-446655440020");
assert.equal(forkOptions?.resumeSessionAt, undefined);
assert.equal(forkOptions?.forkSession, undefined);

yield* adapter.rollbackThread(session.threadId, 2);
const resetOptions = harness.getLastCreateQueryInput()?.options;
assert.equal(resetOptions?.resume, undefined);
assert.equal(resetOptions?.resumeSessionAt, undefined);
assert.equal(resetOptions?.forkSession, undefined);
assert.ok(resetOptions?.sessionId);
}).pipe(
Effect.provideService(Random.Random, makeDeterministicRandomService()),
Effect.provide(harness.layer),
Expand Down
Loading
Loading