fix(claude): retract turns by target identity

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<TurnId | undefined>;
   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);
 });

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* () {

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<ClaudeAdapterShape["rollbackThreadTo"]> = 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;