From 93de80b4633b552498e6f368635d671f47af0dee Mon Sep 17 00:00:00 2001 From: Adam Firestone Date: Thu, 13 Aug 2026 11:13:39 -0500 Subject: [PATCH] fix(claude): retract turns by target identity Prefer the concrete orchestration turn id over stale Claude lifetime counts while preserving count-only validation. Persist the rebased resume cursor and cover retry, restart, and foreign-target safety. --- .../Layers/TurnRetractionReactor.test.ts | 9 +- .../src/provider/Layers/ClaudeAdapter.test.ts | 178 ++++++++++++++++++ .../src/provider/Layers/ClaudeAdapter.ts | 83 +++++++- .../provider/Layers/ProviderService.test.ts | 67 +++++++ .../src/provider/Layers/ProviderService.ts | 19 ++ 5 files changed, 350 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration/Layers/TurnRetractionReactor.test.ts b/apps/server/src/orchestration/Layers/TurnRetractionReactor.test.ts index 99309c9e2f91..8df6cf98cd8b 100644 --- a/apps/server/src/orchestration/Layers/TurnRetractionReactor.test.ts +++ b/apps/server/src/orchestration/Layers/TurnRetractionReactor.test.ts @@ -78,7 +78,7 @@ type MutableState = { sessionStatus: OrchestrationSessionStatus | null; activeTurnId: TurnId | null; historyTurnCount: number; - rollbackTargetTurnId: TurnId | undefined; + readonly rollbackTargetTurnIds: Array; filesystemRestored: boolean; failRollbackAfterEffect: boolean; failRestoreAfterEffect: boolean; @@ -118,7 +118,7 @@ function makeState(providerSendState: ProjectionTurnRetraction["providerSendStat sessionStatus: providerSendState === "claimed" ? "running" : null, activeTurnId: providerSendState === "claimed" ? TURN_ID : null, historyTurnCount: 2, - rollbackTargetTurnId: undefined, + rollbackTargetTurnIds: [], filesystemRestored: false, failRollbackAfterEffect: false, failRestoreAfterEffect: false, @@ -352,7 +352,7 @@ async function startHarness( rollbackConversationTo: ({ retainedTurnCount, targetTurnId }) => Effect.gen(function* () { state.order.push("rollback"); - state.rollbackTargetTurnId = targetTurnId; + state.rollbackTargetTurnIds.push(targetTurnId); if (state.terminalRollbackFailure) { return yield* new ProviderValidationError({ operation: "ProviderService.rollbackConversationTo", @@ -492,7 +492,7 @@ it("drives claimed convergence from interrupt through a settlement event", async expect(state.order).toEqual(["interrupt", "rollback", "restore", "complete"]); expect(state.historyTurnCount).toBe(1); - expect(state.rollbackTargetTurnId).toBe(TURN_ID); + expect(state.rollbackTargetTurnIds).toEqual([TURN_ID]); expect(state.row.status).toBe("completed"); await stopHarness(harness); }); @@ -614,6 +614,7 @@ it("repeats absolute provider rollback harmlessly after a post-rollback crash", expect(state.row.status).toBe("completed"); expect(state.historyTurnCount).toBe(1); expect(state.order.filter((entry) => entry === "rollback")).toHaveLength(2); + expect(state.rollbackTargetTurnIds).toEqual([TURN_ID, TURN_ID]); expect(state.order.slice(-3)).toEqual(["rollback", "restore", "complete"]); await stopHarness(harness); }); diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index d4bd64bc3b93..0d6b2f1c1d12 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -20,6 +20,7 @@ import { type RuntimeMode, ThreadId, ProviderInstanceId, + TurnId, } from "@t3tools/contracts"; import { createModelSelection } from "@t3tools/shared/model"; import { assert, describe, it } from "@effect/vitest"; @@ -4024,6 +4025,183 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("rolls a concrete target across a stale logical turn boundary", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const resumeSessionId = "550e8400-e29b-41d4-a716-446655440000"; + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + resumeCursor: { + threadId: THREAD_ID, + resume: resumeSessionId, + resumeSessionAt: "assistant-before-restart", + turnCount: 12, + }, + }); + + const retainedTurn = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "retained", + attachments: [], + }); + const retainedCompletedFiber = yield* Stream.filter( + adapter.streamEvents, + (event) => event.type === "turn.completed", + ).pipe(Stream.runHead, Effect.forkChild); + harness.query.emit({ + type: "assistant", + session_id: resumeSessionId, + uuid: "assistant-retained", + parent_tool_use_id: null, + message: { + id: "assistant-message-retained", + content: [{ type: "text", text: "retained response" }], + }, + } as unknown as SDKMessage); + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: resumeSessionId, + uuid: "result-retained", + } as unknown as SDKMessage); + yield* Fiber.join(retainedCompletedFiber); + + const targetTurn = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "target", + attachments: [], + }); + + assert.isDefined(adapter.validateRollbackThreadTo); + assert.isDefined(adapter.rollbackThreadTo); + if (!adapter.validateRollbackThreadTo || !adapter.rollbackThreadTo) return; + + // The target identity is authoritative even though the persisted T3 + // boundary is ahead of this older Claude cursor. + yield* adapter.validateRollbackThreadTo(session.threadId, 22, targetTurn.turnId); + + const completedFiber = yield* Stream.filter( + adapter.streamEvents, + (event) => event.type === "turn.completed", + ).pipe(Stream.runHead, Effect.forkChild); + harness.query.emit({ + type: "assistant", + session_id: resumeSessionId, + uuid: "assistant-target", + parent_tool_use_id: null, + message: { + id: "assistant-message-target", + content: [{ type: "text", text: "target response" }], + }, + } as unknown as SDKMessage); + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: resumeSessionId, + uuid: "result-target", + } as unknown as SDKMessage); + yield* Fiber.join(completedFiber); + + const rolledBack = yield* adapter.rollbackThreadTo(session.threadId, 22, targetTurn.turnId); + assert.deepEqual( + rolledBack.turns.map((turn) => turn.id), + [retainedTurn.turnId], + ); + assert.deepEqual((yield* adapter.listSessions())[0]?.resumeCursor, { + threadId: THREAD_ID, + resume: resumeSessionId, + resumeSessionAt: "assistant-retained", + turnCount: 22, + }); + + // Models a crash after provider rollback but before the reactor records + // completion: retrying the same concrete target is harmless. + const repeated = yield* adapter.rollbackThreadTo(session.threadId, 22, targetTurn.turnId); + assert.deepEqual( + repeated.turns.map((turn) => turn.id), + [retainedTurn.turnId], + ); + assert.equal( + ((yield* adapter.listSessions())[0]?.resumeCursor as { turnCount?: number } | undefined) + ?.turnCount, + 22, + ); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("rejects a rollback target that differs from the active Claude turn", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ + threadId: session.threadId, + input: "active", + attachments: [], + }); + + assert.isDefined(adapter.validateRollbackThreadTo); + if (!adapter.validateRollbackThreadTo) return; + const validation = yield* adapter + .validateRollbackThreadTo(session.threadId, 0, TurnId.make("foreign-turn")) + .pipe(Effect.result); + assert.equal(validation._tag, "Failure"); + if (validation._tag === "Failure") { + assert.equal(validation.failure._tag, "ProviderAdapterValidationError"); + assert.match(validation.failure.message, /does not match rollback target/); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("rejects an absent target until its stale boundary is rebased", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + resumeCursor: { + threadId: THREAD_ID, + resume: "550e8400-e29b-41d4-a716-446655440000", + resumeSessionAt: "assistant-stale", + turnCount: 12, + }, + }); + + assert.isDefined(adapter.validateRollbackThreadTo); + if (!adapter.validateRollbackThreadTo) return; + const validation = yield* adapter + .validateRollbackThreadTo(session.threadId, 22, TurnId.make("missing-target")) + .pipe(Effect.result); + assert.equal(validation._tag, "Failure"); + if (validation._tag === "Failure") { + assert.equal(validation.failure._tag, "ProviderAdapterValidationError"); + assert.match(validation.failure.message, /12 turns, below retained boundary 22/); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("recycles after an absolute rollback already at its retained boundary", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index beceaf1d32bd..3fd47df860c9 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -4768,6 +4768,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const validateRollbackBoundary = Effect.fn("validateClaudeRollbackBoundary")(function* ( threadId: ThreadId, retainedTurnCount: number, + targetTurnId?: TurnId, ) { const context = yield* requireSession(threadId); if (!Number.isInteger(retainedTurnCount) || retainedTurnCount < 0) { @@ -4777,6 +4778,44 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( issue: "retainedTurnCount must be an integer >= 0.", }); } + if (targetTurnId !== undefined) { + if (context.turnState !== undefined && context.turnState.turnId !== targetTurnId) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "rollbackThreadTo", + issue: `Active provider turn '${context.turnState.turnId}' does not match rollback target '${targetTurnId}'.`, + }); + } + const targetIndex = context.turns.findIndex((turn) => turn.id === targetTurnId); + const targetIsActive = context.turnState?.turnId === targetTurnId; + if (targetIndex < 0 && !targetIsActive) { + const lifetimeTurnCount = context.sessionBaseTurnCount + context.turns.length; + if (lifetimeTurnCount < retainedTurnCount) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "rollbackThreadTo", + issue: `Provider history has ${lifetimeTurnCount} turns, below retained boundary ${retainedTurnCount}.`, + }); + } + if (lifetimeTurnCount > retainedTurnCount) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "rollbackThreadTo", + issue: `Rollback target '${targetTurnId}' is unavailable while provider history remains above retained boundary ${retainedTurnCount}.`, + }); + } + return; + } + const retainedSessionTurnCount = targetIndex >= 0 ? targetIndex : context.turns.length; + if (retainedSessionTurnCount > retainedTurnCount) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "rollbackThreadTo", + issue: `Provider rollback would retain ${retainedSessionTurnCount} session turns, above retained boundary ${retainedTurnCount}.`, + }); + } + return; + } const lifetimeTurnCount = context.sessionBaseTurnCount + context.turns.length; if (lifetimeTurnCount < retainedTurnCount) { return yield* new ProviderAdapterValidationError({ @@ -4792,9 +4831,49 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const rollbackThreadTo: NonNullable = Effect.fn( "rollbackThreadTo", - )(function* (threadId, retainedTurnCount) { - yield* validateRollbackBoundary(threadId, retainedTurnCount); + )(function* (threadId, retainedTurnCount, targetTurnId) { + yield* validateRollbackBoundary(threadId, retainedTurnCount, targetTurnId); const context = yield* requireSession(threadId); + if (targetTurnId !== undefined) { + const targetIndex = context.turns.findIndex((turn) => turn.id === targetTurnId); + const nextLength = targetIndex >= 0 ? targetIndex : context.turns.length; + const retainedTurns = context.turns.slice(0, nextLength); + const nextSessionBaseTurnCount = retainedTurnCount - nextLength; + + // A concrete orchestration turn id is stronger rollback evidence than + // Claude's logical lifetime count, which can lag after older resume + // cursors or projection repairs. Remove the target and every later local + // turn, then rebase the cursor watermark to T3's retained boundary. + // If the target is already absent, preserve all visible local turns: a + // prior attempt may have completed before its receipt was persisted. + const snapshot = yield* applyRollback(context, nextLength, nextSessionBaseTurnCount); + if (snapshot.turns.some((turn) => turn.id === targetTurnId)) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "thread/rollback-to", + detail: `Provider history still contains retracted turn '${targetTurnId}'.`, + }); + } + if ( + snapshot.turns.length !== retainedTurns.length || + snapshot.turns.some((turn, index) => turn.id !== retainedTurns[index]?.id) + ) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "thread/rollback-to", + detail: `Provider rollback for '${targetTurnId}' did not preserve the preceding turn history.`, + }); + } + const resultingLifetimeTurnCount = context.sessionBaseTurnCount + snapshot.turns.length; + if (resultingLifetimeTurnCount !== retainedTurnCount) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "thread/rollback-to", + detail: `Expected ${retainedTurnCount} retained turns, found ${resultingLifetimeTurnCount}.`, + }); + } + return snapshot; + } const lifetimeTurnCount = context.sessionBaseTurnCount + context.turns.length; const delta = lifetimeTurnCount - retainedTurnCount; const sessionLocalTurnCount = context.turns.length; diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 7334cd019725..9f2bc19fa23c 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -196,6 +196,15 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) { Effect.succeed({ threadId, turns: [] }), ); + const rollbackThreadTo = vi.fn( + ( + threadId: ThreadId, + _retainedTurnCount: number, + _targetTurnId?: TurnId, + ): Effect.Effect<{ threadId: ThreadId; turns: readonly [] }, ProviderAdapterError> => + Effect.succeed({ threadId, turns: [] }), + ); + const stopAll = vi.fn( (): Effect.Effect => Effect.sync(() => { @@ -218,6 +227,7 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) { hasSession, readThread, rollbackThread, + rollbackThreadTo, stopAll, get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); @@ -253,6 +263,7 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) { hasSession, readThread, rollbackThread, + rollbackThreadTo, stopAll, }; } @@ -854,6 +865,62 @@ it.effect( ); routing.layer("ProviderServiceLive routing", (it) => { + it.effect("persists the provider resume cursor after absolute rollback", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + const threadId = asThreadId("thread-rollback-cursor"); + const targetTurnId = asTurnId("turn-rollback-cursor"); + const session = yield* provider.startSession(threadId, { + provider: CLAUDE_AGENT_DRIVER, + providerInstanceId: claudeAgentInstanceId, + threadId, + runtimeMode: "full-access", + }); + const rolledBackCursor = { + threadId, + resume: "resume-after-rollback", + resumeSessionAt: "assistant-before-target", + turnCount: 22, + }; + routing.claude.rollbackThreadTo.mockImplementationOnce((adapterThreadId) => + Effect.sync(() => { + routing.claude.updateSession(adapterThreadId, (existing) => ({ + ...existing, + status: "ready", + resumeCursor: rolledBackCursor, + })); + return { threadId: adapterThreadId, turns: [] }; + }), + ); + + yield* provider.rollbackConversationTo({ + threadId: session.threadId, + retainedTurnCount: 22, + targetTurnId, + }); + + const persisted = yield* directory.getBinding(threadId); + assert.equal(Option.isSome(persisted), true); + if (Option.isSome(persisted)) { + assert.deepEqual(persisted.value.resumeCursor, rolledBackCursor); + assert.equal( + persisted.value.runtimePayload.lastRuntimeEvent, + "provider.rollbackConversationTo", + ); + } + assert.deepEqual(routing.claude.rollbackThreadTo.mock.calls.at(-1), [ + threadId, + 22, + targetTurnId, + ]); + yield* provider.stopSession({ threadId }); + routing.claude.startSession.mockClear(); + routing.claude.rollbackThreadTo.mockClear(); + routing.claude.stopSession.mockClear(); + }), + ); + it.effect("routes provider operations and rollback conversation", () => Effect.gen(function* () { const provider = yield* ProviderService.ProviderService; diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 4e807541f483..0414d05d67ca 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -1125,6 +1125,25 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( } } + // Rollback can change the provider's native resume cursor. Persist it + // before reporting success so recovery never resumes the removed turn. + const rolledBackSession = (yield* routed.adapter.listSessions()).find( + (session) => session.threadId === routed.threadId, + ); + if (rolledBackSession !== undefined) { + yield* upsertSessionBinding( + { + ...rolledBackSession, + providerInstanceId: routed.instanceId, + }, + input.threadId, + { + lastRuntimeEvent: "provider.rollbackConversationTo", + lastRuntimeEventAt: yield* nowIso, + }, + ); + } + yield* analytics.record("provider.conversation.rolled_back", { provider: routed.adapter.provider, retainedTurns: input.retainedTurnCount,