fix(server): keep Claude sessions up while child tasks run

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

@@ -2,6 +2,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices";
 import { expect, it } from "@effect/vitest";
 import { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
 import { HttpServer } from "effect/unstable/http";
 import * as NetAddress from "effect/unstable/net/NetAddress";
 
@@ -163,3 +164,47 @@ it.effect("does not keep credentials of other threads alive", () =>
     expect(yield* registry.resolve(token)).toBeUndefined();
   }),
 );
+
+/**
+ * A replacement credential can be issued without revoking the token a live
+ * provider query is still presenting.
+ */
+function retainsExistingThreadCredentialUntilThatSessionIsRevoked() {
+  const layer = McpSessionRegistry.layer.pipe(
+    Layer.provide(Layer.succeed(HttpServer.HttpServer, fakeHttpServer)),
+    Layer.provide(Layer.succeed(ServerEnvironment.ServerEnvironment, fakeEnvironment)),
+    Layer.provide(NodeServices.layer),
+  );
+  return Effect.gen(function* () {
+    const registry = yield* McpSessionRegistry.McpSessionRegistry;
+    const threadId = ThreadId.make("thread-retain");
+    const first = yield* McpSessionRegistry.issueActiveMcpCredential({
+      threadId,
+      providerInstanceId: ProviderInstanceId.make("claudeAgent"),
+      capabilities: new Set(["preview"]),
+    });
+    const second = yield* McpSessionRegistry.issueActiveMcpCredential({
+      threadId,
+      providerInstanceId: ProviderInstanceId.make("claudeAgent"),
+      capabilities: new Set(["preview"]),
+      retainExisting: true,
+    });
+    if (first === undefined || second === undefined) {
+      return yield* Effect.die("expected MCP credentials");
+    }
+    const firstToken = first.config.authorizationHeader.replace(/^Bearer\s+/, "");
+    const secondToken = second.config.authorizationHeader.replace(/^Bearer\s+/, "");
+
+    expect((yield* registry.resolve(firstToken))?.threadId).toBe(threadId);
+    expect((yield* registry.resolve(secondToken))?.threadId).toBe(threadId);
+
+    yield* McpSessionRegistry.revokeActiveMcpProviderSession(second.config.providerSessionId);
+    expect((yield* registry.resolve(firstToken))?.threadId).toBe(threadId);
+    expect(yield* registry.resolve(secondToken)).toBeUndefined();
+  }).pipe(Effect.provide(layer));
+}
+
+it.effect(
+  "retains an existing thread credential when the replacement asks to keep it",
+  retainsExistingThreadCredentialUntilThatSessionIsRevoked,
+);

apps/server/src/mcp/McpSessionRegistry.ts

@@ -16,6 +16,11 @@ export interface McpCredentialRequest {
   readonly threadId: ThreadId;
   readonly providerInstanceId: ProviderInstanceId;
   readonly capabilities: ReadonlySet<McpInvocationContext.McpCapability>;
+  /**
+   * Keep credentials already issued for this thread. A refused session
+   * replacement still needs the token the running query baked in.
+   */
+  readonly retainExisting?: boolean;
 }
 
 export interface McpIssuedCredential {
@@ -223,14 +228,31 @@ const make = Effect.acquireRelease(
 
 export const layer = Layer.effect(McpSessionRegistry, make);
 
+/**
+ * Issue the MCP credential a provider session presents to `/mcp`.
+ * Replaces any credential already issued for the thread unless
+ * `retainExisting` is set, which keeps the live query's token valid until
+ * the caller revokes it.
+ */
 export const issueActiveMcpCredential = (
   request: McpCredentialRequest,
 ): Effect.Effect<McpIssuedCredential | undefined> =>
   activeMcpSessionRegistry
-    ? activeMcpSessionRegistry
-        .revokeThread(request.threadId)
-        .pipe(Effect.andThen(activeMcpSessionRegistry.issue(request)))
-    : Effect.sync((): McpIssuedCredential | undefined => undefined);
+    ? (request.retainExisting === true
+        ? Effect.void
+        : activeMcpSessionRegistry.revokeThread(request.threadId)
+      ).pipe(Effect.andThen(activeMcpSessionRegistry.issue(request)))
+    : Effect.succeed(undefined);
+
+/**
+ * Revoke one issued MCP credential without touching the thread's other tokens.
+ * Drops a rejected replacement credential, or the credential a stopped query
+ * was using, while a live query keeps its own.
+ */
+export const revokeActiveMcpProviderSession = (providerSessionId: string): Effect.Effect<void> =>
+  activeMcpSessionRegistry
+    ? activeMcpSessionRegistry.revokeProviderSession(providerSessionId)
+    : Effect.void;
 
 /**
  * Refreshes the liveness of a thread's MCP credential. Called on every provider

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

@@ -313,6 +313,178 @@ async function readPromptMessages(
 }
 
 const THREAD_ID = ThreadId.make("thread-claude-1");
+
+/**
+ * Claude adapter layer that records every query `startSession` creates.
+ */
+function makeQueryRecordingClaudeLayer() {
+  const queries: Array<FakeClaudeQuery> = [];
+  const layer = Layer.effect(
+    ClaudeAdapter,
+    Effect.gen(function* () {
+      const claudeConfig = decodeClaudeSettings({});
+      return yield* makeClaudeAdapter(claudeConfig, {
+        createQuery: () => {
+          const query = new FakeClaudeQuery();
+          queries.push(query);
+          return query;
+        },
+      });
+    }),
+  ).pipe(
+    Layer.provideMerge(ServerConfig.layerTest("/tmp/claude-adapter-test", "/tmp")),
+    Layer.provideMerge(ServerSettingsService.layerTest()),
+    Layer.provideMerge(NodeServices.layer),
+  );
+  return { queries, layer };
+}
+
+/**
+ * Refuses Claude session replacement while a child task is still running.
+ */
+function refusesClaudeSessionReplacementWhileLiveTasks() {
+  const harness = makeQueryRecordingClaudeLayer();
+  return Effect.gen(function* () {
+    const adapter = yield* ClaudeAdapter;
+    const taskStartedFiber = yield* adapter.streamEvents.pipe(
+      Stream.filter((event) => event.type === "task.started"),
+      Stream.take(1),
+      Stream.runCollect,
+      Effect.forkChild,
+    );
+    const firstSession = yield* adapter.startSession({
+      threadId: THREAD_ID,
+      provider: ProviderDriverKind.make("claudeAgent"),
+      runtimeMode: "full-access",
+    });
+    yield* adapter.sendTurn({
+      threadId: firstSession.threadId,
+      input: "spawn agents",
+      attachments: [],
+    });
+    harness.queries[0]?.emit({
+      type: "system",
+      subtype: "task_started",
+      task_id: "task-live",
+      description: "Agent A",
+      task_type: "local_agent",
+      uuid: "task-live-uuid",
+      session_id: "sdk-session",
+    } as unknown as SDKMessage);
+    yield* Fiber.join(taskStartedFiber);
+
+    const warningFiber = yield* adapter.streamEvents.pipe(
+      Stream.filter((event) => event.type === "runtime.warning"),
+      Stream.take(1),
+      Stream.runCollect,
+      Effect.forkChild,
+    );
+    const replaceError = yield* adapter
+      .startSession({
+        threadId: THREAD_ID,
+        provider: ProviderDriverKind.make("claudeAgent"),
+        runtimeMode: "approval-required",
+        resumeCursor: firstSession.resumeCursor,
+      })
+      .pipe(Effect.flip);
+    const warnings = Array.from(yield* Fiber.join(warningFiber));
+    const activeSessions = yield* adapter.listSessions();
+
+    assert.equal(replaceError._tag, "ProviderAdapterRequestError");
+    if (replaceError._tag !== "ProviderAdapterRequestError") {
+      return;
+    }
+    assert.equal(replaceError.method, "startSession");
+    assert.match(replaceError.detail, /live task/);
+    assert.equal(harness.queries.length, 1);
+    assert.equal(harness.queries[0]?.closeCalls, 0);
+    assert.equal(yield* adapter.hasSession(THREAD_ID), true);
+    assert.equal(activeSessions.length, 1);
+    assert.equal(activeSessions[0]?.threadId, firstSession.threadId);
+    assert.equal(activeSessions[0]?.runtimeMode, "full-access");
+    assert.equal(warnings.length, 1);
+    assert.equal(warnings[0]?.type, "runtime.warning");
+    if (warnings[0]?.type === "runtime.warning") {
+      assert.match(warnings[0].payload.message, /live task/);
+    }
+  }).pipe(
+    Effect.provideService(Random.Random, makeDeterministicRandomService()),
+    Effect.provide(harness.layer),
+  );
+}
+
+/**
+ * Replaces a Claude session once its child tasks have finished.
+ */
+function replacesClaudeSessionAfterLiveTasksFinish() {
+  const harness = makeQueryRecordingClaudeLayer();
+  return Effect.gen(function* () {
+    const adapter = yield* ClaudeAdapter;
+    const taskStartedFiber = yield* adapter.streamEvents.pipe(
+      Stream.filter((event) => event.type === "task.started"),
+      Stream.take(1),
+      Stream.runCollect,
+      Effect.forkChild,
+    );
+    const firstSession = yield* adapter.startSession({
+      threadId: THREAD_ID,
+      provider: ProviderDriverKind.make("claudeAgent"),
+      runtimeMode: "full-access",
+    });
+    yield* adapter.sendTurn({
+      threadId: firstSession.threadId,
+      input: "spawn agents",
+      attachments: [],
+    });
+    harness.queries[0]?.emit({
+      type: "system",
+      subtype: "task_started",
+      task_id: "task-done",
+      description: "Agent A",
+      task_type: "local_agent",
+      uuid: "task-done-uuid",
+      session_id: "sdk-session",
+    } as unknown as SDKMessage);
+    yield* Fiber.join(taskStartedFiber);
+
+    const taskCompletedFiber = yield* adapter.streamEvents.pipe(
+      Stream.filter((event) => event.type === "task.completed"),
+      Stream.take(1),
+      Stream.runCollect,
+      Effect.forkChild,
+    );
+    harness.queries[0]?.emit({
+      type: "system",
+      subtype: "task_notification",
+      task_id: "task-done",
+      status: "completed",
+      summary: "done",
+      uuid: "task-done-notify",
+      session_id: "sdk-session",
+    } as unknown as SDKMessage);
+    yield* Fiber.join(taskCompletedFiber);
+
+    const secondSession = yield* adapter.startSession({
+      threadId: THREAD_ID,
+      provider: ProviderDriverKind.make("claudeAgent"),
+      runtimeMode: "approval-required",
+      resumeCursor: firstSession.resumeCursor,
+    });
+    const activeSessions = yield* adapter.listSessions();
+
+    assert.equal(harness.queries.length, 2);
+    assert.equal(harness.queries[0]?.closeCalls, 1);
+    assert.equal(harness.queries[1]?.closeCalls, 0);
+    assert.equal(yield* adapter.hasSession(THREAD_ID), true);
+    assert.equal(activeSessions.length, 1);
+    assert.equal(activeSessions[0]?.runtimeMode, "approval-required");
+    assert.deepEqual(activeSessions[0]?.resumeCursor, secondSession.resumeCursor);
+  }).pipe(
+    Effect.provideService(Random.Random, makeDeterministicRandomService()),
+    Effect.provide(harness.layer),
+  );
+}
+
 const RESUME_THREAD_ID = ThreadId.make("thread-claude-resume");
 const SYNTHETIC_SUBAGENT_MODEL = "claude-synthetic-subagent[expanded]";
 const CLAUDE_ORIGINAL_SESSION_ID = "550e8400-e29b-41d4-a716-446655440010";
@@ -4186,6 +4358,16 @@ describe("ClaudeAdapterLive", () => {
     );
   });
 
+  it.effect(
+    "does not replace a session while live tasks are still running",
+    refusesClaudeSessionReplacementWhileLiveTasks,
+  );
+
+  it.effect(
+    "replaces a session after its live tasks have finished",
+    replacesClaudeSessionAfterLiveTasksFinish,
+  );
+
   it.effect("stopSession does not throw into the SDK prompt consumer", () => {
     // The SDK consumes user messages via `for await (... of prompt)`.
     // Stopping a session must end that loop cleanly — not throw an error.

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

@@ -2064,6 +2064,17 @@ function sdkNativeItemId(message: SDKMessage): string | undefined {
   return undefined;
 }
 
+/**
+ * True when replacing the Claude query would stop child tasks that are still
+ * running. That stop records them completed and suppresses `session.exited`.
+ */
+function shouldBlockClaudeSessionReplacementForLiveTasks(existingContext: {
+  readonly stopped: boolean;
+  readonly liveTaskIds: ReadonlySet<string>;
+}): boolean {
+  return !existingContext.stopped && existingContext.liveTaskIds.size > 0;
+}
+
 export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
   claudeSettings: ClaudeSettings,
   options?: ClaudeAdapterLiveOptions,
@@ -4391,7 +4402,17 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
     return Effect.succeed(context);
   };
 
+  /**
+   * Start or replace the Claude query for a thread.
+   * Refuses replacement while child tasks are still running so the open
+   * query and those tasks stay in place.
+   */
   const startSession: ClaudeAdapterShape["startSession"] = Effect.fn("startSession")(
+    /**
+     * Start or replace the Claude query for a thread.
+     * Refuses replacement while child tasks are still running so the open
+     * query and those tasks stay in place.
+     */
     function* (input) {
       const modelCatalog = yield* modelCatalogEffect;
       if (input.provider !== undefined && input.provider !== PROVIDER) {
@@ -4404,6 +4425,25 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
 
       const existingContext = sessions.get(input.threadId);
       if (existingContext) {
+        // Fail instead of handing back the current session: callers bind the
+        // requested runtime mode and send the turn on whatever comes back.
+        // Stopping here would mark live tasks completed without session.exited.
+        if (shouldBlockClaudeSessionReplacementForLiveTasks(existingContext)) {
+          const liveTaskCount = existingContext.liveTaskIds.size;
+          yield* emitRuntimeWarning(
+            existingContext,
+            `Claude session replacement blocked: ${liveTaskCount} live task(s) are still running.`,
+            {
+              existingSessionStatus: existingContext.session.status,
+              liveTaskCount,
+            },
+          );
+          return yield* new ProviderAdapterRequestError({
+            provider: PROVIDER,
+            method: "startSession",
+            detail: `Claude session replacement is blocked: ${liveTaskCount} live task(s) are still running. Retry after background work finishes.`,
+          });
+        }
         yield* Effect.logWarning("claude.session.replacing", {
           threadId: input.threadId,
           existingSessionStatus: existingContext.session.status,