Skip to content

Commit cea9e7b

Browse files
rynfarrynfar
committed
fix(server): fail closed on relative rollback
1 parent 9ade80d commit cea9e7b

30 files changed

Lines changed: 305 additions & 519 deletions

‎apps/server/integration/TestProviderAdapter.integration.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -498,6 +498,7 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter
498498
provider,
499499
capabilities: {
500500
sessionModelSwitch: "in-session",
501+
conversationRollback: "relative",
501502
},
502503
startSession,
503504
sendTurn,

‎apps/server/integration/orchestrationEngine.integration.test.ts‎

Lines changed: 50 additions & 68 deletions
Original file line numberDiff line numberDiff line change
@@ -704,7 +704,7 @@ it.live("records failed turn runtime state and checkpoint status as error", () =
704704
),
705705
);
706706

707-
it.live("reverts to an earlier checkpoint and trims checkpoint projections + git refs", () =>
707+
it.live("fails closed before checkpoint, workspace, or provider rollback mutation", () =>
708708
withHarness((harness) =>
709709
Effect.gen(function* () {
710710
yield* seedProjectAndThread(harness);
@@ -763,9 +763,11 @@ it.live("reverts to an earlier checkpoint and trims checkpoint projections + git
763763
createdAt: "2026-02-24T10:04:59.900Z",
764764
});
765765

766-
yield* harness.waitForThread(
767-
THREAD_ID,
768-
(entry) => entry.session?.threadId === "thread-1" && entry.checkpoints.length === 1,
766+
yield* harness.waitForReceipt(
767+
(receipt): receipt is TurnProcessingQuiescedReceipt =>
768+
receipt.type === "turn.processing.quiesced" &&
769+
receipt.threadId === THREAD_ID &&
770+
receipt.checkpointTurnCount === 1,
769771
);
770772

771773
yield* harness.adapterHarness!.queueTurnResponse(THREAD_ID, {
@@ -822,13 +824,11 @@ it.live("reverts to an earlier checkpoint and trims checkpoint projections + git
822824
createdAt: "2026-02-24T10:05:00.900Z",
823825
});
824826

825-
yield* harness.waitForThread(
826-
THREAD_ID,
827-
(entry) =>
828-
entry.latestTurn?.turnId === "turn-2" &&
829-
entry.checkpoints.length === 2 &&
830-
entry.activities.some((activity) => activity.turnId === "turn-2"),
831-
8000,
827+
yield* harness.waitForReceipt(
828+
(receipt): receipt is TurnProcessingQuiescedReceipt =>
829+
receipt.type === "turn.processing.quiesced" &&
830+
receipt.threadId === THREAD_ID &&
831+
receipt.checkpointTurnCount === 2,
832832
);
833833

834834
yield* harness.engine.dispatch({
@@ -839,50 +839,32 @@ it.live("reverts to an earlier checkpoint and trims checkpoint projections + git
839839
createdAt: nowIso(),
840840
});
841841

842-
yield* harness.waitForDomainEvent((event) => event.type === "thread.reverted");
843-
const revertedThread = yield* harness.waitForThread(
844-
THREAD_ID,
845-
(entry) =>
846-
entry.checkpoints.length === 1 && entry.checkpoints[0]?.checkpointTurnCount === 1,
847-
);
848-
assert.equal(revertedThread.checkpoints[0]?.checkpointTurnCount, 1);
849-
assert.deepEqual(
850-
revertedThread.messages.map((message) => ({ role: message.role, text: message.text })),
851-
[
852-
{ role: "user", text: "First edit" },
853-
{ role: "assistant", text: "Updated README to v2.\n" },
854-
],
855-
);
856-
assert.equal(
857-
revertedThread.activities.some((activity) => activity.turnId === "turn-2"),
858-
false,
859-
);
860-
assert.equal(
861-
revertedThread.activities.some(
862-
(activity) => activity.turnId === "turn-1" && activity.kind === "tool.started",
863-
),
864-
true,
865-
);
842+
yield* harness.drainCheckpointReactor;
843+
const snapshot = yield* harness.snapshotQuery.getSnapshot();
844+
const unchangedThread = snapshot.threads.find((entry) => entry.id === THREAD_ID);
845+
846+
assert.equal(unchangedThread?.checkpoints.length, 2);
847+
assert.equal(unchangedThread?.latestTurn?.turnId, "turn-2");
866848
assert.equal(
867-
revertedThread.activities.some(
868-
(activity) => activity.turnId === "turn-1" && activity.kind === "tool.completed",
849+
unchangedThread?.activities.some(
850+
(activity) => activity.kind === "checkpoint.revert.failed",
869851
),
870852
true,
871853
);
872854
assert.equal(
873855
NodeFS.readFileSync(NodePath.join(harness.workspaceDir, "README.md"), "utf8"),
874-
"v2\n",
856+
"v3\n",
875857
);
876858
assert.equal(
877859
gitRefExists(harness.workspaceDir, checkpointRefForThreadTurn(THREAD_ID, 2)),
878-
false,
860+
true,
879861
);
880-
assert.deepEqual(harness.adapterHarness!.getRollbackCalls(THREAD_ID), [1]);
862+
assert.deepEqual(harness.adapterHarness!.getRollbackCalls(THREAD_ID), []);
881863

882864
const checkpointRows = yield* harness.checkpointRepository.listByThreadId({
883865
threadId: THREAD_ID,
884866
});
885-
assert.equal(checkpointRows.length, 1);
867+
assert.equal(checkpointRows.length, 2);
886868
}),
887869
),
888870
);
@@ -902,15 +884,10 @@ it.live(
902884
createdAt: nowIso(),
903885
});
904886

905-
const thread = yield* harness.waitForThread(THREAD_ID, (entry) =>
906-
entry.activities.some(
907-
(activity) =>
908-
activity.kind === "checkpoint.revert.failed" &&
909-
typeof activity.payload === "object" &&
910-
activity.payload !== null,
911-
),
912-
);
913-
const failureActivity = thread.activities.find(
887+
yield* harness.drainCheckpointReactor;
888+
const snapshot = yield* harness.snapshotQuery.getSnapshot();
889+
const thread = snapshot.threads.find((entry) => entry.id === THREAD_ID);
890+
const failureActivity = thread?.activities.find(
914891
(activity) => activity.kind === "checkpoint.revert.failed",
915892
);
916893
assert.equal(failureActivity !== undefined, true);
@@ -1292,7 +1269,7 @@ it.live("forwards thread.turn.interrupt to claudeAgent provider sessions", () =>
12921269
),
12931270
);
12941271

1295-
it.live("reverts claudeAgent turns and rolls back provider conversation state", () =>
1272+
it.live("fails closed for relative claudeAgent rollback", () =>
12961273
withHarness(
12971274
(harness) =>
12981275
Effect.gen(function* () {
@@ -1350,10 +1327,11 @@ it.live("reverts claudeAgent turns and rolls back provider conversation state",
13501327
},
13511328
});
13521329

1353-
yield* harness.waitForThread(
1354-
THREAD_ID,
1355-
(entry) =>
1356-
entry.latestTurn?.turnId === "turn-1" && entry.session?.threadId === "thread-1",
1330+
yield* harness.waitForReceipt(
1331+
(receipt): receipt is TurnProcessingQuiescedReceipt =>
1332+
receipt.type === "turn.processing.quiesced" &&
1333+
receipt.threadId === THREAD_ID &&
1334+
receipt.checkpointTurnCount === 1,
13571335
);
13581336

13591337
yield* harness.adapterHarness!.queueTurnResponse(THREAD_ID, {
@@ -1404,12 +1382,11 @@ it.live("reverts claudeAgent turns and rolls back provider conversation state",
14041382
text: "Second Claude edit",
14051383
});
14061384

1407-
yield* harness.waitForThread(
1408-
THREAD_ID,
1409-
(entry) =>
1410-
entry.latestTurn?.turnId === "turn-2" &&
1411-
entry.checkpoints.length === 2 &&
1412-
entry.session?.providerName === "claudeAgent",
1385+
yield* harness.waitForReceipt(
1386+
(receipt): receipt is TurnProcessingQuiescedReceipt =>
1387+
receipt.type === "turn.processing.quiesced" &&
1388+
receipt.threadId === THREAD_ID &&
1389+
receipt.checkpointTurnCount === 2,
14131390
);
14141391

14151392
yield* harness.engine.dispatch({
@@ -1420,21 +1397,26 @@ it.live("reverts claudeAgent turns and rolls back provider conversation state",
14201397
createdAt: nowIso(),
14211398
});
14221399

1423-
const revertedThread = yield* harness.waitForThread(
1424-
THREAD_ID,
1425-
(entry) =>
1426-
entry.checkpoints.length === 1 && entry.checkpoints[0]?.checkpointTurnCount === 1,
1400+
yield* harness.drainCheckpointReactor;
1401+
const snapshot = yield* harness.snapshotQuery.getSnapshot();
1402+
const unchangedThread = snapshot.threads.find((entry) => entry.id === THREAD_ID);
1403+
1404+
assert.equal(unchangedThread?.checkpoints.length, 2);
1405+
assert.equal(
1406+
unchangedThread?.activities.some(
1407+
(activity) => activity.kind === "checkpoint.revert.failed",
1408+
),
1409+
true,
14271410
);
1428-
assert.equal(revertedThread.checkpoints[0]?.checkpointTurnCount, 1);
14291411
assert.equal(
14301412
gitRefExists(harness.workspaceDir, checkpointRefForThreadTurn(THREAD_ID, 1)),
14311413
true,
14321414
);
14331415
assert.equal(
14341416
gitRefExists(harness.workspaceDir, checkpointRefForThreadTurn(THREAD_ID, 2)),
1435-
false,
1417+
true,
14361418
);
1437-
assert.deepEqual(harness.adapterHarness!.getRollbackCalls(THREAD_ID), [1]);
1419+
assert.deepEqual(harness.adapterHarness!.getRollbackCalls(THREAD_ID), []);
14381420
}),
14391421
CLAUDE_AGENT_PROVIDER,
14401422
),

0 commit comments

Comments
 (0)