fix(server): integrate peer links with current main

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

@@ -95,6 +95,7 @@ const clientLaunchHarness = (input: {
       },
     }),
     Layer.mock(ThreadLaunch.ThreadLaunchService)({
+      getAcceptedLaunch: () => Effect.succeedNone,
       checkWorktreeBase: input.checkWorktreeBase ?? (() => Effect.void),
       launch: (launch) => {
         input.launched.push(launch);

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

@@ -116,6 +116,19 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, {
                 ),
               )).projectId
             : yield* resolveProjectId(context, input.projectId);
+        // Accepted creates skip mutable base/default-model preflights.
+        // Message intake still handles their replay and preparation.
+        const acceptedLaunch =
+          input.clientRequestId === undefined
+            ? undefined
+            : Option.getOrUndefined(
+                yield* ThreadLaunch.ThreadLaunchService.pipe(
+                  Effect.flatMap((launches) =>
+                    launches.getAcceptedLaunch({ commandId, threadId, projectId }),
+                  ),
+                  Effect.mapError(unavailable),
+                ),
+              );
         const readProject = Project.ProjectService.pipe(
           Effect.flatMap((projects) => projects.getById(projectId)),
           Effect.mapError(unavailable),
@@ -131,7 +144,9 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, {
         const launchedModel =
           launched === null || launched.deletedAt !== null ? undefined : launched.modelSelection;
         const replaying = launchedModel !== undefined;
-        if (!replaying && input.workspaceStrategy?.type === "existing_worktree") {
+        // A retry may bind a workspace to an idle accepted launch, so membership
+        // is checked on every launch, unlike the checks below.
+        if (input.workspaceStrategy?.type === "existing_worktree") {
           const project = yield* readProject;
           if (project === undefined)
             return yield* new OrchestratorMcpFailure({
@@ -142,7 +157,7 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, {
         }
         // Reject unresolved base refs before accepting a launch; worktree provisioning happens later.
         const strategy = input.workspaceStrategy;
-        if (!replaying && strategy?.type === "worktree")
+        if (acceptedLaunch === undefined && !replaying && strategy?.type === "worktree")
           yield* ThreadLaunch.ThreadLaunchService.pipe(
             Effect.flatMap((launches) =>
               launches.checkWorktreeBase({ projectId, workspaceStrategy: strategy }),
@@ -176,6 +191,7 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, {
               )
             : undefined;
         const modelSelection =
+          acceptedLaunch?.modelSelection ??
           input.modelSelection ??
           launchedModel ??
           caller?.modelSelection ??

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -12193,6 +12193,33 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
             }),
         ),
       );
+      if (
+        command.type === "message.dispatch" &&
+        command.dispatchMode?.type === "defer_start" &&
+        (yield* DispatchModeLimit) !== undefined
+      ) {
+        const runCreated = storedEvents.find(
+          (stored) => stored.event.type === "run.created",
+        )?.event;
+        if (runCreated?.type === "run.created") {
+          const current = yield* projectionStore
+            .getThreadRecords(command.threadId, ["runs"])
+            .pipe(
+              Effect.mapError(
+                (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }),
+              ),
+            );
+          // Replaying this receipt can resume workspace preparation, so it must
+          // respect the caller's current ceiling just like a fresh dispatch.
+          if (
+            current.runs.some(
+              (run) => run.id === runCreated.payload.id && run.status === "preparing",
+            )
+          ) {
+            yield* refuseAboveDispatchModeLimit(command, command.threadId, current.thread);
+          }
+        }
+      }
       if (command.type === "queue.resume" || command.type === "queued-run.start") {
         yield* mapDispatchError(command)(
           startNextQueuedRun(

apps/server/src/orchestration-v2/ThreadLaunchService.test.ts

@@ -15,6 +15,7 @@ import {
   ComposerContextId,
   type ChatAttachment,
   CommandId,
+  EnvironmentId,
   DEFAULT_SERVER_SETTINGS,
   GitCommandError,
   MessageId,
@@ -59,6 +60,10 @@ import * as EffectOutbox from "./EffectOutbox.ts";
 import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
 import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
 import * as ThreadLaunch from "./ThreadLaunchService.ts";
+import * as McpInvocationContext from "../mcp/McpInvocationContext.ts";
+import * as McpToolAccess from "../mcp/McpToolAccess.ts";
+import * as ProjectHandlers from "../mcp/toolkits/project/handlers.ts";
+import { ProjectToolkit } from "../mcp/toolkits/project/tools.ts";
 import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
 import * as ThreadManagement from "./ThreadManagementService.ts";
 import * as ThreadTitleRegeneration from "./ThreadTitleRegenerationService.ts";
@@ -114,6 +119,7 @@ const claudeAdapter = {
 } as ProviderAdapter.ProviderAdapterV2["Service"];
 
 interface HarnessOptions {
+  readonly getProject?: ProjectService.ProjectService["Service"]["getById"];
   readonly closeTerminal?: TerminalManager.TerminalManager["Service"]["close"];
   readonly adapters?: ReadonlyArray<ProviderAdapter.ProviderAdapterV2["Service"]>;
   readonly managedFolders?: Layer.Layer<ManagedProjectFolders.ManagedProjectFolders>;
@@ -179,14 +185,16 @@ function makeHarness(options: HarnessOptions = {}) {
       bootstrap: () => Effect.die("unused"),
       update: () => Effect.die("unused"),
       delete: () => Effect.die("unused"),
-      getById: (id) =>
-        Effect.succeed(
-          id === projectId
-            ? Option.some(project)
-            : id === otherProjectId
-              ? Option.some(otherProject)
-              : Option.none(),
-        ),
+      getById:
+        options.getProject ??
+        ((id) =>
+          Effect.succeed(
+            id === projectId
+              ? Option.some(project)
+              : id === otherProjectId
+                ? Option.some(otherProject)
+                : Option.none(),
+          )),
       getByWorkspaceRoot: () => Effect.succeedSome(project),
       snapshot: Effect.die("unused"),
       getShell: () => Effect.die("unused"),
@@ -1386,6 +1394,180 @@ it.effect("replays a server-allocated launch", () =>
   }),
 );
 
+it.effect("MCP replays accepted launches and checks worktree membership", () =>
+  Effect.gen(function* () {
+    const setupEntered = yield* Deferred.make<void>();
+    const allowSetup = yield* Deferred.make<void>();
+    let changed = false;
+    const harness = makeHarness({
+      getProject: (id) =>
+        Effect.sync(() =>
+          id === projectId
+            ? Option.some({ ...project, defaultModelSelection: changed ? null : modelSelection })
+            : id === otherProjectId
+              ? Option.some(otherProject)
+              : Option.none(),
+        ),
+      hasCommit: ({ refName }) => Effect.sync(() => refName === "HEAD" || !changed),
+      git: { hasRefNamed: () => Effect.succeed(false) },
+      runSetup: () =>
+        Deferred.succeed(setupEntered, undefined).pipe(
+          Effect.andThen(Deferred.await(allowSetup)),
+          Effect.as({ status: "no-script" as const }),
+        ),
+    });
+    const context = McpInvocationContext.McpInvocationContext.of({
+      environmentId: EnvironmentId.make("launch-retry"),
+      requestNamespace: "client:launch-retry",
+      thread: undefined,
+      client: { sessionId: "launch-retry", label: "Test", access: "full-access" },
+      issuedAt: 0,
+      capabilities: new Set(["orchestration" as const]),
+    });
+    const dependencies = Layer.mergeAll(
+      harness.layer,
+      NodeCrypto.layer,
+      DocumentText.layer.pipe(Layer.provideMerge(NodeServices.layer)),
+      Layer.succeed(McpInvocationContext.McpInvocationContext, context),
+      Layer.mock(GitVcsDriver.GitVcsDriver)({
+        listWorktreePaths: () => Effect.succeed([project.workspaceRoot, "/repo-worktrees/feature"]),
+      }),
+      ServerConfig.layerTest(process.cwd(), { prefix: "t3-launch-retry-" }).pipe(
+        Layer.provide(NodeServices.layer),
+      ),
+    );
+    yield* Effect.gen(function* () {
+      const toolkit = yield* ProjectToolkit.pipe(
+        Effect.provide(McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer)),
+      );
+      const input = {
+        title: "Only once",
+        projectId,
+        clientRequestId: "lost-response",
+        runtimeMode: "approval-required" as const,
+        message: "Do the work",
+        workspaceStrategy: { type: "worktree" as const, baseRef: "temporary-base" },
+      };
+      const launch = (params: Parameters<typeof toolkit.handle<"t3_thread_launch">>[1]) =>
+        toolkit.handle("t3_thread_launch", params).pipe(
+          Stream.unwrap,
+          Stream.runCollect,
+          Effect.map((results) => results.at(-1)?.result),
+        );
+      const fs = yield* FileSystem.FileSystem;
+      const outsidePath = yield* fs.makeTempDirectoryScoped({ prefix: "launch-unregistered-" });
+      const outsideStrategy = { type: "existing_worktree" as const, worktreePath: outsidePath };
+      const refusal = {
+        code: "invalid_request",
+        message:
+          "worktreePath must be one of the project's git worktrees. t3_worktree_list shows them.",
+      };
+      const idleInput = {
+        title: "Idle accepted launch",
+        projectId,
+        clientRequestId: "idle-retry",
+        workspaceStrategy: { type: "root" as const },
+      };
+      const idleLaunch = yield* launch(idleInput);
+      assert.include(idleLaunch, { runId: null, status: null });
+      assert(idleLaunch !== undefined && "threadId" in idleLaunch);
+      assert.include(
+        yield* launch({
+          ...idleInput,
+          message: "Run outside the project",
+          workspaceStrategy: outsideStrategy,
+        }),
+        refusal,
+      );
+      const threads = yield* ThreadManagement.ThreadManagementService;
+      const idle = yield* threads.getThreadProjection(idleLaunch.threadId);
+      assert.equal(idle.messages.length, 0);
+      assert.equal(idle.runs.length, 0);
+      assert.equal(idle.thread.worktreePath, null);
+      assert.equal(harness.createWorktree.mock.calls.length, 0);
+      assert.equal(harness.runSetup.mock.calls.length, 0);
+
+      const first = yield* launch(input);
+      assert(first !== undefined && "threadId" in first);
+      assert.notProperty(first, "code");
+      yield* Deferred.await(setupEntered);
+      changed = true;
+
+      assert.include(yield* launch({ ...input, workspaceStrategy: outsideStrategy }), refusal);
+
+      assert.deepEqual(yield* launch(input), first);
+      assert.deepEqual(yield* launch({ ...input, modelSelection: claudeModelSelection }), first);
+      assert.include(yield* launch({ ...input, projectId: otherProjectId }), {
+        code: "orchestration_error",
+      });
+      assert.include(
+        yield* launch(input).pipe(
+          Effect.provideService(McpInvocationContext.McpInvocationContext, {
+            ...context,
+            client: { ...context.client!, access: "read-only" },
+          }),
+        ),
+        { code: "capability_denied" },
+      );
+      assert.include(yield* launch({ ...input, clientRequestId: "fresh" }), {
+        code: "invalid_request",
+      });
+      assert.include(
+        yield* launch({
+          ...input,
+          clientRequestId: "fresh-root",
+          workspaceStrategy: { type: "root" },
+        }),
+        { code: "invalid_request" },
+      );
+      const threadId = first.threadId;
+      assert.equal((yield* threads.getThreadShell(threadId))?.runtimeMode, "approval-required");
+      yield* threads.dispatch({
+        type: "thread.runtime-mode.set",
+        commandId: CommandId.make("owner:raise-launch-modes"),
+        threadId,
+        runtimeMode: "full-access",
+      });
+      assert.include(
+        yield* launch(input).pipe(
+          Effect.provideService(McpInvocationContext.McpInvocationContext, {
+            ...context,
+            client: { ...context.client!, access: "approval-required" },
+          }),
+        ),
+        { code: "orchestration_error" },
+      );
+      const projection = yield* threads.getThreadProjection(threadId);
+      assert.equal(projection.thread.runtimeMode, "full-access");
+      assert.equal(projection.messages.length, 1);
+      assert.equal(projection.runs.length, 1);
+      assert.equal(projection.runs[0]?.status, "preparing");
+      assert.deepEqual(projection.runs[0]?.workspacePreparation, input.workspaceStrategy);
+      assert.equal(projection.thread.worktreePath, "/repo-worktrees/feature");
+      assert.equal(harness.createWorktree.mock.calls.length, 1);
+      assert.equal(harness.runSetup.mock.calls.length, 1);
+      yield* Deferred.succeed(allowSetup, undefined);
+      yield* threads.streamStoredEventsFrom({ threadId }).pipe(
+        Stream.filter(
+          (stored) =>
+            stored.commandId === CommandId.make(`${threadId}:release`) &&
+            stored.event.type === "run.updated",
+        ),
+        Stream.runHead,
+      );
+      assert.include(
+        yield* launch(input).pipe(
+          Effect.provideService(McpInvocationContext.McpInvocationContext, {
+            ...context,
+            client: { ...context.client!, access: "approval-required" },
+          }),
+        ),
+        { threadId, status: "starting" },
+      );
+    }).pipe(Effect.provide(dependencies));
+  }),
+);
+
 it.effect("replays an accepted launch after the project disables the instance", () => {
   const harness = makeHarness();
   return Effect.gen(function* () {
@@ -1742,19 +1924,38 @@ it.effect("schedules an accepted preparing message exactly once across concurren
         text: "Resume preparation",
         attachments: [],
         modelSelection: input.modelSelection,
-        dispatchMode: { type: "defer_start" },
+        dispatchMode: { type: "defer_start", workspaceStrategy: input.workspaceStrategy },
         createdBy: input.createdBy,
         creationSource: input.creationSource,
       });
       const preparing = yield* threads.getThreadProjection(input.threadId);
       assert.equal(preparing.runs[0]?.status, "preparing");
 
-      const [first, second] = yield* Effect.all([launches.launch(input), launches.launch(input)], {
+      const retry = { ...input, workspaceStrategy: { type: "root" as const } };
+      const [first, second] = yield* Effect.all([launches.launch(retry), launches.launch(retry)], {
         concurrency: "unbounded",
       });
-      yield* Deferred.await(setupEntered);
+      const prepared = yield* Effect.raceFirst(
+        Deferred.await(setupEntered).pipe(Effect.as(true)),
+        threads.streamStoredEventsFrom({ threadId: input.threadId }).pipe(
+          Stream.filter(
+            (stored) =>
+              stored.commandId === CommandId.make(input.commandId + ":release") &&
+              stored.event.type === "run.updated",
+          ),
+          Stream.runHead,
+          Effect.as(false),
+        ),
+      );
+      assert.isTrue(prepared, "the accepted worktree setup must run before releasing the run");
       assert.isTrue(first.resumed);
       assert.isTrue(second.resumed);
+      assert.equal(harness.createWorktree.mock.calls.length, 1);
+      assert.equal(harness.runSetup.mock.calls[0]?.[0].worktreePath, "/repo-worktrees/feature");
+      assert.deepEqual(
+        (yield* threads.getThreadProjection(input.threadId)).runs[0]?.workspacePreparation,
+        input.workspaceStrategy,
+      );
       assert.equal(harness.runSetup.mock.calls.length, 1);
       yield* Deferred.succeed(allowSetup, undefined);
     }).pipe(Effect.provide(harness.layer));
@@ -1867,6 +2068,7 @@ it.effect("shared intake preserves durable attachment bytes after a lost launch
               }),
             ),
           ),
+        getAcceptedLaunch: launches.getAcceptedLaunch,
         checkWorktreeBase: launches.checkWorktreeBase,
         retryPreparation: launches.retryPreparation,
         prepareRun: launches.prepareRun,