fix(server): restarts keep queued steering and stop duplicate launches

apps/server/src/mcp/McpSessionRegistry.test.ts

@@ -59,6 +59,26 @@ it.effect("stores only a token hash, resolves the bearer token, and revokes by t
   }),
 );
 
+it.effect("keeps the request namespace across credential reissue and splits it per thread", () =>
+  Effect.gen(function* () {
+    const registry = yield* makeRegistry(() => 1_000);
+    const request = {
+      threadId: ThreadId.make("thread-restarted"),
+      providerInstanceId: ProviderInstanceId.make("codex"),
+    };
+    const scopeOf = (issued: { readonly config: { readonly authorizationHeader: string } }) =>
+      registry.resolve(issued.config.authorizationHeader.replace(/^Bearer\s+/, ""));
+    const first = yield* scopeOf(yield* registry.issue(request));
+    const second = yield* scopeOf(yield* registry.issue(request));
+    const other = yield* scopeOf(
+      yield* registry.issue({ ...request, threadId: ThreadId.make("thread-other") }),
+    );
+    expect(second?.thread.providerSessionId).not.toBe(first?.thread.providerSessionId);
+    expect(second?.requestNamespace).toBe(first?.requestNamespace);
+    expect(other?.requestNamespace).not.toBe(first?.requestNamespace);
+  }),
+);
+
 it.effect("always grants pull-requests and gates browser and device access independently", () =>
   Effect.gen(function* () {
     const registry = yield* makeRegistry(() => 1_000);

apps/server/src/mcp/McpSessionRegistry.ts

@@ -113,7 +113,8 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* (
       const browserToolsAvailable = request.browserToolsAvailable ?? true;
       const scope: McpInvocationContext.McpThreadInvocationScope = {
         environmentId,
-        requestNamespace: providerSessionId,
+        // Retrying an accepted write after a restart must reuse its receipt.
+        requestNamespace: `thread:${encodeURIComponent(request.threadId)}:${encodeURIComponent(request.providerInstanceId)}`,
         thread: {
           threadId: ThreadId.make(request.threadId),
           providerSessionId,

apps/server/src/mcp/toolkits/project/handlers.test.ts

@@ -101,6 +101,97 @@ it.effect("attributes a launched thread's first message to the calling thread",
   }),
 );
 
+it.effect(
+  "a launch retried with the same clientRequestId keeps its command after session rotation",
+  () =>
+    Effect.gen(function* () {
+      const sourceThreadId = ThreadId.make("source-thread");
+      const projectId = ProjectId.make("project");
+      const providerInstanceId = ProviderInstanceId.make("codex");
+      const modelSelection = { instanceId: providerInstanceId, model: "gpt-5" };
+      const caller = {
+        id: sourceThreadId,
+        projectId,
+        providerInstanceId,
+        modelSelection,
+        runtimeMode: "full-access",
+        interactionMode: "default",
+        activeRunId: "active-run",
+        archivedAt: null,
+        deletedAt: null,
+      } as OrchestrationV2ThreadShell;
+      const launches: Array<ThreadLaunch.ThreadLaunchInput> = [];
+      const launch = (clientRequestId: string | undefined, requestNamespace: string) =>
+        Effect.gen(function* () {
+          const dependencies = Layer.mergeAll(
+            ServerSettings.layerTest(),
+            NodeCrypto.layer,
+            Layer.succeed(McpInvocationContext.McpInvocationContext, {
+              environmentId: EnvironmentId.make("environment"),
+              requestNamespace,
+              thread: {
+                threadId: sourceThreadId,
+                providerSessionId: "session",
+                providerInstanceId,
+              },
+              client: undefined,
+              issuedAt: 0,
+              capabilities: new Set(["orchestration" as const]),
+            }),
+            Layer.mock(ThreadManagement.ThreadManagementService)({
+              getThreadShell: () => Effect.succeed(caller),
+            }),
+            Layer.mock(ThreadLaunch.ThreadLaunchService)({
+              launch: (input) => {
+                launches.push(input);
+                return Effect.succeed({
+                  threadId: input.threadId,
+                  projection: {
+                    thread: { id: input.threadId, projectId, modelSelection },
+                    runs: [],
+                  },
+                  resumed: false,
+                } as unknown as ThreadLaunch.ThreadLaunchResult);
+              },
+            }),
+            Layer.mock(Project.ProjectService)({}),
+            Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
+              namedProjectsRoot: "/projects",
+            }),
+            Layer.mock(GitVcsDriver.GitVcsDriver)({}),
+            NodeServices.layer,
+            ServerConfig.layerTest(process.cwd(), { prefix: "t3-launch-retry-" }).pipe(
+              Layer.provide(NodeServices.layer),
+            ),
+          );
+          const toolkit = yield* ProjectToolkit.pipe(
+            Effect.provide(
+              McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer).pipe(
+                Layer.provide(dependencies),
+              ),
+            ),
+          );
+          yield* toolkit
+            .handle("t3_thread_launch", {
+              title: "Audit",
+              message: "Review the change",
+              ...(clientRequestId === undefined ? {} : { clientRequestId }),
+            })
+            .pipe(Stream.unwrap, Stream.runCollect, Effect.provide(dependencies));
+        });
+      // The namespace is stable per thread, so a rotated session still matches.
+      yield* launch("retry-launch", "thread:source-thread:codex");
+      yield* launch("retry-launch", "thread:source-thread:codex");
+      expect(launches[1]?.commandId).toBe(launches[0]?.commandId);
+      expect(launches[1]?.threadId).toBe(launches[0]?.threadId);
+      expect(launches[1]?.initialMessage?.messageId).toBe(launches[0]?.initialMessage?.messageId);
+      yield* launch("retry-launch", "thread:other-thread:codex");
+      yield* launch(undefined, "thread:source-thread:codex");
+      yield* launch(undefined, "thread:source-thread:codex");
+      expect(new Set(launches.map((input) => input.threadId)).size).toBe(4);
+    }),
+);
+
 it.effect("launches a scratch thread into the Scratch project", () =>
   Effect.gen(function* () {
     const sourceThreadId = ThreadId.make("source-thread");

apps/server/src/mcp/toolkits/project/handlers.ts

@@ -1,4 +1,10 @@
-import { MessageId, ThreadId, OrchestratorMcpFailure, ProjectId } from "@t3tools/contracts";
+import {
+  CommandId,
+  MessageId,
+  ThreadId,
+  OrchestratorMcpFailure,
+  ProjectId,
+} from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
 import * as Option from "effect/Option";
@@ -57,7 +63,12 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, {
       Effect.gen(function* () {
         const context = yield* readCaller();
         const { caller } = context;
-        const commandId = yield* newCommandId();
+        const commandId =
+          input.clientRequestId === undefined
+            ? yield* newCommandId()
+            : CommandId.make(
+                `command:mcp:${encodeURIComponent(context.scope.requestNamespace)}:thread-launch:${encodeURIComponent(input.clientRequestId)}`,
+              );
         const threadId = ThreadId.make(commandId);
         const messageId = MessageId.make(commandId);
         const attachments = input.attachments ?? [];

apps/server/src/mcp/toolkits/project/tools.ts

@@ -104,8 +104,9 @@ const ProjectCloneTool = Tool.make("t3_project_clone", {
 const ThreadLaunchTool = Tool.make("t3_thread_launch", {
   ...shared,
   description:
-    'Create an ordinary TOP-LEVEL thread with an explicit workspace binding before its agent starts. Use this when the user requests independent work, a new thread, or a PR stack in its own worktree; use delegate_task for child subagents. Set workspaceStrategy to {type:"worktree",baseRef:"parent-branch",branch:"new-branch",startFromOrigin:false} for a new worktree based on local commits, or {type:"existing_worktree",worktreePath:"/absolute/path",branch:"existing-branch"} to use an existing checkout. For upstream commits, set startFromOrigin:true. Omitted workspaceStrategy means the project root, NOT the caller\'s worktree. Omit projectId/modelSelection/modes to inherit those settings from the calling thread; a caller outside a T3 thread must pass projectId and gets the project\'s default model. Set scratch:true instead of projectId for a thread without a project: it runs in a fresh folder of its own, outside any repository. Put the task in message. Do not ask the agent to create its own worktree via shell: that does not update the thread binding. Each call creates a new launch with no retry key; retain threadId and use t3_thread_read/t3_thread_wait to follow preparation. To link a thread for the user, write `[title](t3-thread://v1/<threadId>)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread\'s current title. After errors or lost responses, inspect t3_thread_list before retrying. Attachments must be pending uploads. The new thread may not run with broader runtime or interaction modes than the caller: the calling T3 thread\'s own modes, or the permission mode an outside agent was approved with.',
+    'Create an ordinary TOP-LEVEL thread with an explicit workspace binding before its agent starts. Use this when the user requests independent work, a new thread, or a PR stack in its own worktree; use delegate_task for child subagents. Set workspaceStrategy to {type:"worktree",baseRef:"parent-branch",branch:"new-branch",startFromOrigin:false} for a new worktree based on local commits, or {type:"existing_worktree",worktreePath:"/absolute/path",branch:"existing-branch"} to use an existing checkout. For upstream commits, set startFromOrigin:true. Omitted workspaceStrategy means the project root, NOT the caller\'s worktree. Omit projectId/modelSelection/modes to inherit those settings from the calling thread; a caller outside a T3 thread must pass projectId and gets the project\'s default model. Set scratch:true instead of projectId for a thread without a project: it runs in a fresh folder of its own, outside any repository. Put the task in message. Do not ask the agent to create its own worktree via shell: that does not update the thread binding. Pass a stable clientRequestId to retry the same launch after errors or restarts; retain threadId and use t3_thread_read/t3_thread_wait to follow preparation. To link a thread for the user, write `[title](t3-thread://v1/<threadId>)` with the threadId exactly as returned, not URL-encoded; T3 Code shows the thread\'s current title. After errors or lost responses, inspect t3_thread_list before retrying. Attachments must be pending uploads. The new thread may not run with broader runtime or interaction modes than the caller: the calling T3 thread\'s own modes, or the permission mode an outside agent was approved with.',
   parameters: Schema.Struct({
+    clientRequestId: Schema.optional(TrimmedNonEmptyString.check(Schema.isMaxLength(256))),
     projectId: Schema.optional(ProjectId),
     scratch: Schema.optional(
       Schema.Boolean.annotate({

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts

@@ -3458,6 +3458,154 @@ describe("CodexAdapterV2 post-settle continuation", () => {
     ),
   );
 
+  it.effect.each(["active-goal", "no-goal"] as const)(
+    "a continuation Codex rejects as EmptyInput retries only on an active goal (%s)",
+    (state) =>
+      Effect.scoped(
+        Effect.gen(function* () {
+          const scenario = "codex-restart-empty-input";
+          const nativeThreadId = "native-restart-empty-input";
+          const nativeTurnId = "turn-restart-empty-input";
+          const prompt = "Continue where you left off.";
+          const preamble = codexReplayPreamble({
+            nativeThreadId,
+            nativeTurnId,
+            prompt: "unused",
+          }).slice(0, 5);
+          const turnParams = {
+            threadId: nativeThreadId,
+            cwd: "/workspace",
+            model: "gpt-5.4",
+            approvalPolicy: "never",
+            approvalsReviewer: "user",
+            sandboxPolicy: { type: "dangerFullAccess" },
+            summary: "detailed",
+          };
+          const transcript = makeCodexReplayTranscript({
+            scenario,
+            entries: [
+              ...preamble,
+              {
+                type: "expect_outbound",
+                label: "resume",
+                frame: {
+                  id: 3,
+                  method: "thread/resume",
+                  params: {
+                    threadId: nativeThreadId,
+                    excludeTurns: true,
+                    config: CodexAdapterV2.CODEX_THREAD_CONFIG,
+                  },
+                },
+              },
+              {
+                type: "emit_inbound",
+                label: "resume",
+                frame: { id: 3, result: { thread: { id: nativeThreadId, updatedAt: 1782622450 } } },
+              },
+              {
+                type: "expect_outbound",
+                label: "empty continue",
+                frame: { id: 4, method: "turn/start", params: { ...turnParams, input: [] } },
+              },
+              {
+                type: "emit_inbound",
+                label: "empty continue",
+                frame: {
+                  id: 4,
+                  error: { code: -32603, message: "failed to submit turn input: EmptyInput" },
+                },
+              },
+              {
+                type: "expect_outbound",
+                label: "read goal",
+                frame: { id: 5, method: "thread/goal/get", params: { threadId: nativeThreadId } },
+              },
+              {
+                type: "emit_inbound",
+                label: "read goal",
+                frame: {
+                  id: 5,
+                  result: {
+                    goal:
+                      state === "no-goal"
+                        ? null
+                        : {
+                            threadId: nativeThreadId,
+                            objective: "Finish the work",
+                            status: "active",
+                            tokenBudget: null,
+                            tokensUsed: 10,
+                            timeUsedSeconds: 3,
+                            createdAt: 1782622440,
+                            updatedAt: 1782622450,
+                          },
+                  },
+                },
+              },
+              ...(state === "active-goal"
+                ? [
+                    {
+                      type: "expect_outbound" as const,
+                      label: "prompted continue",
+                      frame: {
+                        id: 6,
+                        method: "turn/start",
+                        params: { ...turnParams, input: [{ type: "text", text: prompt }] },
+                      },
+                    },
+                    {
+                      type: "emit_inbound" as const,
+                      label: "prompted continue",
+                      frame: {
+                        id: 6,
+                        result: {
+                          turn: makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }),
+                        },
+                      },
+                    },
+                  ]
+                : []),
+            ],
+          });
+          const harness = yield* makeCodexReplayHarness(transcript);
+          const resumed = yield* harness.runtime.resumeThread({
+            providerThread: harness.providerThread,
+            modelSelection: CODEX_TEST_MODEL_SELECTION,
+            runtimePolicy: CODEX_TEST_RUNTIME_POLICY,
+          });
+          const started = yield* harness.runtime
+            .startTurn({
+              ...makeCodexTestTurnInput({
+                threadId: harness.threadId,
+                providerThread: resumed,
+                now: yield* DateTime.now,
+                attemptId: RunAttemptId.make("attempt-restart-empty-input"),
+                text: prompt,
+              }),
+              restartContinuationOfRunId: RunId.make("run-before-restart"),
+            })
+            .pipe(Effect.result);
+          if (state === "active-goal") {
+            assert.equal(started._tag, "Success");
+            return;
+          }
+          assert.equal(started._tag, "Failure");
+          if (started._tag === "Failure") {
+            assert.nestedPropertyVal(
+              started.failure,
+              "cause.errorMessage",
+              "failed to submit turn input: EmptyInput",
+            );
+          }
+        }).pipe(
+          Effect.provide(
+            Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
+          ),
+        ),
+      ),
+  );
+
   it.effect("resolves retryable app-server errors on resumed provider activity", () =>
     Effect.scoped(
       Effect.gen(function* () {

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts

@@ -6614,11 +6614,32 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
                 : null;
             if (goalCommand !== null) return runGoalCommand(turnInput, goalCommand);
             return Effect.gen(function* () {
-              yield* startNativeTurn(
-                turnInput,
+              const input =
                 turnInput.restartContinuationOfRunId === undefined
                   ? yield* toCodexInput(turnInput)
-                  : [],
+                  : [];
+              yield* startNativeTurn(turnInput, input).pipe(
+                Effect.catchTags({
+                  CodexAppServerRequestError: (cause) => {
+                    // Codex rejects an empty continuation while a goal turn is
+                    // still active. Send the continue prompt as real input.
+                    if (
+                      input.length !== 0 ||
+                      cause.code !== -32603 ||
+                      cause.errorMessage !== "failed to submit turn input: EmptyInput"
+                    )
+                      return Effect.fail(cause);
+                    return Effect.gen(function* () {
+                      const current = yield* client
+                        .request("thread/goal/get", {
+                          threadId: yield* getNativeThreadId(turnInput.providerThread),
+                        })
+                        .pipe(Effect.orElseSucceed(() => null));
+                      if (current?.goal?.status !== "active") return yield* Effect.fail(cause);
+                      yield* startNativeTurn(turnInput, yield* toCodexInput(turnInput));
+                    });
+                  },
+                }),
               );
             }).pipe(
               Effect.mapError(