fix(server): resume active turns when T3 Code reopens

apps/server/src/provider/Layers/ClaudeAdapter.test.ts

@@ -4247,6 +4247,52 @@ describe("ClaudeAdapterLive", () => {
     );
   });
 
+  it.effect("rejects an echoed Claude resume cursor when the SDK starts a fresh session", () => {
+    const harness = makeHarness();
+    return Effect.gen(function* () {
+      const adapter = yield* ClaudeAdapter;
+      const persistedResumeCursor = {
+        threadId: RESUME_THREAD_ID,
+        resume: "550e8400-e29b-41d4-a716-446655440000",
+        turnCount: 3,
+      };
+      const session = yield* adapter.startSession({
+        threadId: RESUME_THREAD_ID,
+        provider: ProviderDriverKind.make("claudeAgent"),
+        resumeCursor: persistedResumeCursor,
+        runtimeMode: "full-access",
+      });
+      assert.notEqual(adapter.isSameResumeCursor, undefined);
+      if (adapter.isSameResumeCursor === undefined) return;
+
+      const verification = yield* adapter
+        .isSameResumeCursor(RESUME_THREAD_ID, persistedResumeCursor, session.resumeCursor)
+        .pipe(Effect.forkChild);
+      harness.query.emit({
+        type: "system",
+        subtype: "init",
+        apiKeySource: "none",
+        claude_code_version: "test",
+        cwd: "/tmp/claude-adapter-test",
+        tools: [],
+        mcp_servers: [],
+        model: SYNTHETIC_CLAUDE_STANDARD_MODEL,
+        permissionMode: "bypassPermissions",
+        slash_commands: [],
+        output_style: "default",
+        skills: [],
+        plugins: [],
+        session_id: "7368d0c7-40a3-4d8a-bcc1-ac80c49f2719",
+        uuid: "fresh-init",
+      } as unknown as SDKMessage);
+
+      assert.equal(yield* Fiber.join(verification), false);
+    }).pipe(
+      Effect.provideService(Random.Random, makeDeterministicRandomService()),
+      Effect.provide(harness.layer),
+    );
+  });
+
   it.effect("preserves durable resume ids across Claude resume hooks", () => {
     const harness = makeHarness();
     return Effect.gen(function* () {
@@ -4259,7 +4305,7 @@ describe("ClaudeAdapterLive", () => {
         Effect.forkChild,
       );
 
-      yield* adapter.startSession({
+      const session = yield* adapter.startSession({
         threadId: RESUME_THREAD_ID,
         provider: ProviderDriverKind.make("claudeAgent"),
         resumeCursor: {
@@ -4331,6 +4377,17 @@ describe("ClaudeAdapterLive", () => {
           }
         | undefined;
       assert.equal(resumeCursor?.resume, durableSessionId);
+      assert.notEqual(adapter.isSameResumeCursor, undefined);
+      if (adapter.isSameResumeCursor !== undefined) {
+        assert.equal(
+          yield* adapter.isSameResumeCursor(
+            RESUME_THREAD_ID,
+            session.resumeCursor,
+            activeSessions[0]?.resumeCursor,
+          ),
+          true,
+        );
+      }
     }).pipe(
       Effect.provideService(Random.Random, makeDeterministicRandomService()),
       Effect.provide(harness.layer),

apps/server/src/provider/Layers/ClaudeAdapter.ts

@@ -69,6 +69,7 @@ import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
 import * as FileSystem from "effect/FileSystem";
 import * as Fiber from "effect/Fiber";
+import * as Option from "effect/Option";
 import * as Path from "effect/Path";
 import * as Queue from "effect/Queue";
 import * as Ref from "effect/Ref";
@@ -291,6 +292,7 @@ interface ClaudeSessionContext {
    * effort override inherit this. */
   currentEffort: string | undefined;
   resumeSessionId: string | undefined;
+  readonly confirmedResumeSessionId: Deferred.Deferred<string>;
   readonly pendingApprovals: Map<ApprovalRequestId, PendingApproval>;
   readonly pendingUserInputs: Map<ApprovalRequestId, PendingUserInput>;
   readonly turns: Array<{
@@ -2047,6 +2049,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
     const nextThreadId = message.session_id;
     context.resumeSessionId = message.session_id;
     yield* updateResumeCursor(context);
+    yield* Deferred.succeed(context.confirmedResumeSessionId, message.session_id);
 
     if (context.lastThreadStartedId !== nextThreadId) {
       context.lastThreadStartedId = nextThreadId;
@@ -3886,6 +3889,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
       const runPromise = Effect.runPromiseWith(runtimeContext);
 
       const promptQueue = yield* Queue.unbounded<PromptQueueItem>();
+      const confirmedResumeSessionId = yield* Deferred.make<string>();
       const prompt = Stream.fromQueue(promptQueue).pipe(
         Stream.filter((item) => item.type === "message"),
         Stream.map((item) => item.message),
@@ -4461,6 +4465,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
         currentApiModelId: apiModelId,
         currentEffort: effectiveEffort ?? undefined,
         resumeSessionId: sessionId,
+        confirmedResumeSessionId,
         pendingApprovals,
         pendingUserInputs,
         turns: [],
@@ -4804,6 +4809,24 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
       sessionModelSwitch: "in-session",
     },
     startSession,
+    isSameResumeCursor: (threadId, persisted, recovered) =>
+      Effect.gen(function* () {
+        const persistedResume = readClaudeResumeState(persisted)?.resume;
+        const recoveredResume = readClaudeResumeState(recovered)?.resume;
+        const context = sessions.get(threadId);
+        if (
+          persistedResume === undefined ||
+          recoveredResume === undefined ||
+          persistedResume !== recoveredResume ||
+          context === undefined
+        ) {
+          return false;
+        }
+        const confirmed = yield* Deferred.await(context.confirmedResumeSessionId).pipe(
+          Effect.timeoutOption("10 seconds"),
+        );
+        return Option.isSome(confirmed) && confirmed.value === persistedResume;
+      }),
     sendTurn,
     interruptTurn,
     readThread,

apps/server/src/provider/Layers/CodexAdapter.ts

@@ -2324,6 +2324,12 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
       promptlessTurnContinuation: true,
     },
     startSession,
+    isSameResumeCursor: (_threadId, persisted, recovered) =>
+      Effect.succeed(
+        isCodexResumeCursorSchema(persisted) &&
+          isCodexResumeCursorSchema(recovered) &&
+          persisted.threadId === recovered.threadId,
+      ),
     sendTurn,
     interruptTurn,
     readThread,

apps/server/src/provider/Layers/CursorAdapter.ts

@@ -1205,6 +1205,15 @@ export function makeCursorAdapter(
       provider: PROVIDER,
       capabilities: { sessionModelSwitch: "in-session" },
       startSession,
+      isSameResumeCursor: (_threadId, persisted, recovered) => {
+        const persistedSession = parseCursorResume(persisted);
+        const recoveredSession = parseCursorResume(recovered);
+        return Effect.succeed(
+          persistedSession !== undefined &&
+            recoveredSession !== undefined &&
+            persistedSession.sessionId === recoveredSession.sessionId,
+        );
+      },
       sendTurn,
       interruptTurn,
       readThread,

apps/server/src/provider/Layers/GrokAdapter.ts

@@ -2114,6 +2114,15 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte
       provider: PROVIDER,
       capabilities: { sessionModelSwitch: "in-session" },
       startSession,
+      isSameResumeCursor: (_threadId, persisted, recovered) => {
+        const persistedSession = parseGrokResume(persisted);
+        const recoveredSession = parseGrokResume(recovered);
+        return Effect.succeed(
+          persistedSession !== undefined &&
+            recoveredSession !== undefined &&
+            persistedSession.sessionId === recoveredSession.sessionId,
+        );
+      },
       sendTurn,
       interruptTurn,
       readThread,

apps/server/src/provider/Layers/OpenCodeAdapter.ts

@@ -3267,6 +3267,15 @@ export function makeOpenCodeAdapter(
         sessionModelSwitch: "in-session",
       },
       startSession,
+      isSameResumeCursor: (_threadId, persisted, recovered) => {
+        const persistedSession = parseOpenCodeResume(persisted);
+        const recoveredSession = parseOpenCodeResume(recovered);
+        return Effect.succeed(
+          persistedSession !== undefined &&
+            recoveredSession !== undefined &&
+            persistedSession.sessionId === recoveredSession.sessionId,
+        );
+      },
       sendTurn,
       interruptTurn,
       respondToRequest,

apps/server/src/provider/Layers/ProviderService.test.ts

@@ -246,6 +246,19 @@ function makeFakeCodexAdapter(provider: ProviderDriverKind = CODEX_DRIVER) {
       ...(provider === CODEX_DRIVER ? { promptlessTurnContinuation: true } : {}),
     },
     startSession,
+    isSameResumeCursor: (_threadId, persisted, recovered) => {
+      const readOpaque = (value: unknown) =>
+        typeof value === "object" &&
+        value !== null &&
+        !Array.isArray(value) &&
+        typeof (value as { opaque?: unknown }).opaque === "string"
+          ? (value as { opaque: string }).opaque
+          : undefined;
+      const persistedOpaque = readOpaque(persisted);
+      return Effect.succeed(
+        persistedOpaque !== undefined && persistedOpaque === readOpaque(recovered),
+      );
+    },
     sendTurn,
     interruptTurn,
     respondToRequest,
@@ -1464,11 +1477,14 @@ routing.layer("ProviderServiceLive routing", (it) => {
       routing.codex.startSession.mockClear();
       routing.codex.sendTurn.mockClear();
 
-      yield* provider.sendTurn({
-        threadId: initial.threadId,
-        input: "resume",
-        attachments: [],
-      });
+      yield* provider.sendTurn(
+        {
+          threadId: initial.threadId,
+          input: "resume",
+          attachments: [],
+        },
+        { requireResumeCursor: initial.resumeCursor },
+      );
 
       assert.equal(routing.codex.startSession.mock.calls.length, 1);
       const resumedStartInput = routing.codex.startSession.mock.calls[0]?.[0];
@@ -1489,6 +1505,47 @@ routing.layer("ProviderServiceLive routing", (it) => {
     }),
   );
 
+  it.effect("rejects automatic continuation when recovery starts a fresh conversation", () =>
+    Effect.gen(function* () {
+      const provider = yield* ProviderService.ProviderService;
+      const threadId = asThreadId("thread-strict-resume-mismatch");
+      const initial = yield* provider.startSession(threadId, {
+        provider: CODEX_DRIVER,
+        providerInstanceId: codexInstanceId,
+        threadId,
+        cwd: "/tmp/project-strict-resume",
+        runtimeMode: "full-access",
+      });
+
+      yield* routing.codex.stopAll();
+      routing.codex.startSession.mockClear();
+      routing.codex.sendTurn.mockClear();
+      routing.codex.stopSession.mockClear();
+      routing.codex.startSession.mockImplementationOnce(() =>
+        Effect.succeed({
+          ...initial,
+          resumeCursor: { opaque: "fresh-provider-conversation" },
+        }),
+      );
+
+      const failure = yield* provider
+        .sendTurn(
+          {
+            threadId,
+            input: "Continue where you left off.",
+            attachments: [],
+          },
+          { requireResumeCursor: initial.resumeCursor },
+        )
+        .pipe(Effect.flip);
+
+      assert.instanceOf(failure, ProviderValidationError);
+      assert.match(failure.message, /did not resume the persisted conversation/u);
+      assert.equal(routing.codex.sendTurn.mock.calls.length, 0);
+      assert.deepStrictEqual(routing.codex.stopSession.mock.calls, [[threadId]]);
+    }),
+  );
+
   it.effect("recovers stale claudeAgent sessions for sendTurn using persisted cwd", () =>
     Effect.gen(function* () {
       const provider = yield* ProviderService.ProviderService;