authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-07 22:20:16-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-07 22:20:16-07:00
logc28cd905572e2a28934a6229d88bcf2975963032
treea27e5fc6d9824129a9f0eff6fcc3f1c2b81c4774
parent0bfd9dd19d0c9418a936190525ec8f4c4fdae09b
signature Signed by PGP key B5690EEEBB952194

fix(server): a message sent during a rollback no longer undoes it (#17079)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

5 files changed, 95 insertions(+), 61 deletions(-)

apps/server/src/orchestration-v2/CheckpointRestoreSafety.test.ts+2
......@@ -17,6 +17,7 @@ import * as Option from "effect/Option";
1717import * as Path from "effect/Path";
1818import { symlinksSupported } from "@t3tools/shared/testing/symlinks";
1919import * as CheckpointRollbackService from "./CheckpointRollbackService.ts";
20import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
2021import { CheckpointServiceV2 } from "./CheckpointService.ts";
2122import { EventSinkV2 } from "./EventSink.ts";
2223import * as IdAllocator from "./IdAllocator.ts";
......@@ -102,6 +103,7 @@ it.effect.each([
102103 Layer.provide(
103104 Layer.mergeAll(
104105 NodeServices.layer,
106 ThreadCommandExecutor.layer,
105107 IdAllocator.layer,
106108 Layer.mock(ProjectStore.ProjectStoreV2)({
107109 get: () => Effect.succeed(Option.some({ workspaceRoot: parent } as never)),
apps/server/src/orchestration-v2/CheckpointRollbackService.test.ts+20-1
......@@ -20,6 +20,8 @@ import { resolveCodexRollbackTurnCount } from "./Adapters/CodexAdapterV2.ts";
2020import { isCheckpointRestoreIsolated } from "./CheckpointRestoreSafety.ts";
2121import * as CheckpointService from "./CheckpointService.ts";
2222import * as CheckpointRollbackService from "./CheckpointRollbackService.ts";
23import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
24import * as KeyedLock from "@t3tools/shared/KeyedLock";
2325import * as EventSink from "./EventSink.ts";
2426import * as IdAllocator from "./IdAllocator.ts";
2527import * as ProjectionStore from "./ProjectionStore.ts";
......@@ -64,6 +66,7 @@ it.effect("rejects a non-ready checkpoint before opening a session or restoring
6466 const layerTest = layerCheckpointRollbackService.pipe(
6567 Layer.provide(
6668 Layer.mergeAll(
69 ThreadCommandExecutor.layer,
6770 Layer.mock(CheckpointService.CheckpointServiceV2)({ restore }),
6871 Layer.mock(EventSink.EventSinkV2)({}),
6972 IdAllocator.layer,
......@@ -141,6 +144,7 @@ it.effect("rejects a rollback when another provider thread became active", () =>
141144 const layerTest = layerCheckpointRollbackService.pipe(
142145 Layer.provide(
143146 Layer.mergeAll(
147 ThreadCommandExecutor.layer,
144148 Layer.mock(CheckpointService.CheckpointServiceV2)({ restore }),
145149 Layer.mock(EventSink.EventSinkV2)({}),
146150 IdAllocator.layer,
......@@ -221,6 +225,7 @@ it.effect("rejects a rollback when provider selection changed before execution",
221225 const layerTest = layerCheckpointRollbackService.pipe(
222226 Layer.provide(
223227 Layer.mergeAll(
228 ThreadCommandExecutor.layer,
224229 Layer.mock(CheckpointService.CheckpointServiceV2)({ restore }),
225230 Layer.mock(EventSink.EventSinkV2)({}),
226231 IdAllocator.layer,
......@@ -292,6 +297,7 @@ it.effect("reports a missing provider turn as a structured rollback failure", ()
292297 const layerTest = layerCheckpointRollbackService.pipe(
293298 Layer.provide(
294299 Layer.mergeAll(
300 ThreadCommandExecutor.layer,
295301 Layer.mock(CheckpointService.CheckpointServiceV2)({ restore }),
296302 Layer.mock(EventSink.EventSinkV2)({}),
297303 IdAllocator.layer,
......@@ -352,6 +358,15 @@ it.effect.each([
352358 const checkpointId = CheckpointId.make("rewind-start");
353359 const scopeId = CheckpointScopeId.make("rewind-scope");
354360 const calls: string[] = [];
361 let lockedThreads = Effect.succeed<ReadonlyArray<ThreadId>>([]);
362 const layerThreadCommands = Layer.effect(
363 ThreadCommandExecutor.ThreadCommandExecutor,
364 Effect.tap(KeyedLock.make<ThreadId>(), (lock) =>
365 Effect.sync(() => {
366 lockedThreads = lock.activeKeys;
367 }),
368 ),
369 );
355370 const providerThread = {
356371 id: providerThreadId,
357372 providerSessionId,
......@@ -390,6 +405,7 @@ it.effect.each([
390405 const layerTest = layerCheckpointRollbackService.pipe(
391406 Layer.provide(
392407 Layer.mergeAll(
408 layerThreadCommands,
393409 Layer.mock(CheckpointService.CheckpointServiceV2)({
394410 restore: () =>
395411 Effect.sync(() => {
......@@ -398,7 +414,10 @@ it.effect.each([
398414 }),
399415 Layer.mock(EventSink.EventSinkV2)({
400416 write: ({ events }) =>
401 Effect.sync(() => {
417 Effect.map(lockedThreads, (locked) => {
418 // A thread command that planned against the old provider thread
419 // must not commit after this and undo the rollback.
420 assert.include(locked, threadId);
402421 assert.ok(
403422 events.some(
404423 (event) => event.type === "run.updated" && event.payload.status === "rolled_back",
apps/server/src/orchestration-v2/CheckpointRollbackService.ts+71-60
......@@ -26,6 +26,7 @@ import * as ProjectStore from "./ProjectStore.ts";
2626import type { ProviderAdapterV2RollbackTarget } from "./ProviderAdapter.ts";
2727import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts";
2828import { RuntimePolicyV2 } from "./RuntimePolicy.ts";
29import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
2930
3031export const ROLLBACK_FAILED_MESSAGE =
3132 "The provider could not roll back this conversation. Try again; if it keeps failing, check the provider and server logs.";
......@@ -91,6 +92,7 @@ export const layer: Layer.Layer<
9192 | FileSystem.FileSystem
9293 | Path.Path
9394 | ProjectStore.ProjectStoreV2
95 | ThreadCommandExecutor.ThreadCommandExecutor
9496> = Layer.effect(
9597 CheckpointRollbackServiceV2,
9698 Effect.gen(function* () {
......@@ -103,6 +105,7 @@ export const layer: Layer.Layer<
103105 const fileSystem = yield* FileSystem.FileSystem;
104106 const projects = yield* ProjectStore.ProjectStoreV2;
105107 const path = yield* Path.Path;
108 const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor;
106109
107110 const execute = Effect.fn("orchestrationV2.checkpointRollback.execute")(function* (input: {
108111 readonly threadId: ThreadId;
......@@ -274,72 +277,80 @@ export const layer: Layer.Layer<
274277 yield* checkpoints.deleteStaleRefs({ scope, checkpoints: staleCheckpoints });
275278 }
276279
277 const now = yield* DateTime.now;
278 const makeEvent = <Event extends OrchestrationV2DomainEvent>(event: Omit<Event, "id">) =>
279 Effect.map(
280 ids.allocate.event({ threadId: event.threadId }),
281 (id) =>
282 ({
283 ...event,
284 id,
285 }) as Event,
286 );
287 const events: Array<OrchestrationV2DomainEvent> = [];
288 events.push(
289 yield* makeEvent({
290 type: "provider-thread.updated",
291 threadId: input.threadId,
292 driver: providerThread.driver,
293 providerInstanceId: providerThread.providerInstanceId,
294 occurredAt: now,
295 payload: {
296 ...snapshot.providerThread,
297 lastRunOrdinal: targetOrdinal === 0 ? null : targetOrdinal,
298 updatedAt: now,
299 },
300 }),
301 );
302 for (const staleCheckpoint of staleCheckpoints) {
303 events.push(
304 yield* makeEvent({
305 type: "checkpoint.captured",
306 threadId: input.threadId,
307 ...(staleCheckpoint.runId === null ? {} : { runId: staleCheckpoint.runId }),
308 nodeId: staleCheckpoint.nodeId,
309 providerInstanceId: providerThread.providerInstanceId,
310 occurredAt: now,
311 payload: { ...staleCheckpoint, status: "stale" },
312 }),
313 );
314 }
315 for (const run of runsToRollback) {
316 const rootNode = projection.nodes.find((candidate) => candidate.id === run.rootNodeId);
317 events.push(
318 yield* makeEvent({
319 type: "run.updated",
320 threadId: input.threadId,
321 runId: run.id,
322 ...(rootNode === undefined ? {} : { nodeId: rootNode.id }),
323 providerInstanceId: run.providerInstanceId,
324 occurredAt: now,
325 payload: { ...run, status: "rolled_back", completedAt: now },
326 }),
327 );
328 if (rootNode !== undefined) {
280 // Thread commands write full provider thread rows under this lock. Without
281 // it, a message sent during the rollback can plan against the old native
282 // session and commit after this, undoing the provider's fork.
283 yield* threadCommands.withLock(
284 input.threadId,
285 Effect.gen(function* () {
286 const now = yield* DateTime.now;
287 const makeEvent = <Event extends OrchestrationV2DomainEvent>(event: Omit<Event, "id">) =>
288 Effect.map(
289 ids.allocate.event({ threadId: event.threadId }),
290 (id) =>
291 ({
292 ...event,
293 id,
294 }) as Event,
295 );
296 const events: Array<OrchestrationV2DomainEvent> = [];
329297 events.push(
330298 yield* makeEvent({
331 type: "node.updated",
299 type: "provider-thread.updated",
332300 threadId: input.threadId,
333 runId: run.id,
334 nodeId: rootNode.id,
335 providerInstanceId: run.providerInstanceId,
301 driver: providerThread.driver,
302 providerInstanceId: providerThread.providerInstanceId,
336303 occurredAt: now,
337 payload: { ...rootNode, status: "rolled_back", completedAt: now },
304 payload: {
305 ...snapshot.providerThread,
306 lastRunOrdinal: targetOrdinal === 0 ? null : targetOrdinal,
307 updatedAt: now,
308 },
338309 }),
339310 );
340 }
341 }
342 yield* eventSink.write({ events });
311 for (const staleCheckpoint of staleCheckpoints) {
312 events.push(
313 yield* makeEvent({
314 type: "checkpoint.captured",
315 threadId: input.threadId,
316 ...(staleCheckpoint.runId === null ? {} : { runId: staleCheckpoint.runId }),
317 nodeId: staleCheckpoint.nodeId,
318 providerInstanceId: providerThread.providerInstanceId,
319 occurredAt: now,
320 payload: { ...staleCheckpoint, status: "stale" },
321 }),
322 );
323 }
324 for (const run of runsToRollback) {
325 const rootNode = projection.nodes.find((candidate) => candidate.id === run.rootNodeId);
326 events.push(
327 yield* makeEvent({
328 type: "run.updated",
329 threadId: input.threadId,
330 runId: run.id,
331 ...(rootNode === undefined ? {} : { nodeId: rootNode.id }),
332 providerInstanceId: run.providerInstanceId,
333 occurredAt: now,
334 payload: { ...run, status: "rolled_back", completedAt: now },
335 }),
336 );
337 if (rootNode !== undefined) {
338 events.push(
339 yield* makeEvent({
340 type: "node.updated",
341 threadId: input.threadId,
342 runId: run.id,
343 nodeId: rootNode.id,
344 providerInstanceId: run.providerInstanceId,
345 occurredAt: now,
346 payload: { ...rootNode, status: "rolled_back", completedAt: now },
347 }),
348 );
349 }
350 }
351 yield* eventSink.write({ events });
352 }),
353 );
343354 });
344355
345356 return CheckpointRollbackServiceV2.of({
apps/server/src/orchestration-v2/runtimeLayer.ts+1
......@@ -182,6 +182,7 @@ const layerCheckpointRollbackServiceProvided = CheckpointRollbackService.layer.p
182182 IdAllocator.layer,
183183 ProjectionStore.layer,
184184 layerProviderSessionManagerProvided,
185 ThreadCommandExecutor.layer,
185186 layerRuntimePolicyProvided,
186187 ),
187188 ),
apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts+1
......@@ -395,6 +395,7 @@ export function layerWithRegistry<Error>(
395395 IdAllocator.layer,
396396 layerStores,
397397 layerProviderSessionManagerProvided,
398 ThreadCommandExecutor.layer,
398399 layerRuntime,
399400 ),
400401 ),