diff --git a/src/node/services/workspaceTurnManager.test.ts b/src/node/services/workspaceTurnManager.test.ts
index 20588945ad..87e5be7514 100644
--- a/src/node/services/workspaceTurnManager.test.ts
+++ b/src/node/services/workspaceTurnManager.test.ts
@@ -21,6 +21,7 @@ import type { ErrorEvent, StreamEndEvent } from "@/common/types/stream";
 import { createMuxMessage } from "@/common/types/message";
 import type { WorkspaceMetadata } from "@/common/types/workspace";
 import type { AIService } from "@/node/services/aiService";
+import type { StreamManager } from "@/node/services/streamManager";
 import type {
   WorkspaceHost,
   BackgroundableForegroundWaiter,
@@ -250,12 +251,25 @@ function createWorkspaceTurnManagerHost(
   };
 }
 
+function activeStreamInfo(muxMetadata: unknown, messageId: string) {
+  return {
+    messageId,
+    model: "test-model",
+    historySequence: 0,
+    startTime: 0,
+    parts: [],
+    toolCompletionTimestamps: new Map(),
+    muxMetadata,
+  };
+}
+
 function createWorkspaceTurnManagerHarness(
   config: Config,
   overrides?: {
     aiService?: AIService;
     workspaceService?: WorkspaceHost;
     initStateManager?: InitStateManager;
+    streamManager?: StreamManager;
   }
 ): {
   historyService: HistoryService;
@@ -284,7 +298,8 @@ function createWorkspaceTurnManagerHarness(
     workspaceService,
     initStateManager,
     taskHost,
-    terminalAttentionStore
+    terminalAttentionStore,
+    overrides?.streamManager
   );
 
   return {
@@ -319,9 +334,11 @@ describe("WorkspaceTurnManager", () => {
       hasPendingQueuedOrPreparingTurn?: ReturnType<typeof mock>;
       hasPendingBashMonitorWakeContinuation?: ReturnType<typeof mock>;
       hasPendingWorkspaceTurnContinuation?: ReturnType<typeof mock>;
+      hasQueuedWorkspaceTurn?: ReturnType<typeof mock>;
       getQueueCutCutter?: ReturnType<typeof mock>;
       hasPendingAutoRetry?: ReturnType<typeof mock>;
       waitForPendingStreamErrorRecoveryDecision?: ReturnType<typeof mock>;
+      streamManager?: StreamManager;
     } = {}
   ) {
     const config = await createTestConfig(rootDir);
@@ -336,6 +353,7 @@ describe("WorkspaceTurnManager", () => {
     const { historyService, taskService, taskHost } = createWorkspaceTurnManagerHarness(config, {
       aiService: aiMocks.aiService,
       workspaceService: workspaceMocks.workspaceService,
+      streamManager: options.streamManager,
     });
 
     const created = await taskService.createWorkspaceTurn({
@@ -437,10 +455,14 @@ describe("WorkspaceTurnManager", () => {
       taskService as unknown as {
         activeWorkspaceTurnHandleByWorkspaceId: Map<
           string,
-          { handleId: string; ownerWorkspaceId: string }
+          { handleId: string; ownerWorkspaceId: string; accepted: boolean }
         >;
       }
-    ).activeWorkspaceTurnHandleByWorkspaceId.set(workspaceId, { handleId, ownerWorkspaceId });
+    ).activeWorkspaceTurnHandleByWorkspaceId.set(workspaceId, {
+      handleId,
+      ownerWorkspaceId,
+      accepted: true,
+    });
   }
 
   test("workspace lifecycle archives only parent-owned created workspace turns", async () => {
@@ -3476,6 +3498,10 @@ describe("WorkspaceTurnManager", () => {
         (workspaceId: string, handleId: string) =>
           workspaceId === "childworkspace" && handleId === "wst_secondhandle"
       ),
+      hasPendingWorkspaceTurnContinuation: mock(
+        (workspaceId: string, metadata: ReturnType<typeof workspaceTurnMuxMetadata>) =>
+          workspaceId === "childworkspace" && metadata.taskHandleId === "wst_secondhandle"
+      ),
       isBusyForMessage,
       hasQueuedMessages,
     });
@@ -3560,7 +3586,97 @@ describe("WorkspaceTurnManager", () => {
     expect(aiMocks.stopStream).not.toHaveBeenCalled();
   });
 
-  test("createWorkspaceTurn reserves a slot before queueing a manually busy existing workspace", async () => {
+  test("createWorkspaceTurn keeps queued handles live during send preflight", async () => {
+    const config = await createTestConfig(rootDir);
+    stubStableIds(config, ["firsthandle", "firstturn", "secondhandle", "secondturn"]);
+    const { parentId, projectPath } = await saveLocalParentWorkspace(config, rootDir);
+    const createWorkspace = makeWorkspaceTurnCreateMock(config, projectPath);
+    const secondSendStarted = Promise.withResolvers<void>();
+    const releaseSecondSend = Promise.withResolvers<void>();
+    let sendCallCount = 0;
+    let queueContainsSecondHandle = false;
+    const sendMessage = mock(
+      async (..._args: unknown[]): Promise<Result<void, SendMessageError>> => {
+        sendCallCount += 1;
+        if (sendCallCount === 2) {
+          secondSendStarted.resolve();
+          await releaseSecondSend.promise;
+          queueContainsSecondHandle = true;
+        }
+        return Ok(undefined);
+      }
+    );
+    const busyWorkspaceIds = new Set<string>();
+    const workspaceMocks = createWorkspaceServiceMocks({
+      create: createWorkspace,
+      sendMessage,
+      isBusyForMessage: mock((workspaceId: string) => busyWorkspaceIds.has(workspaceId)),
+      hasQueuedMessages: mock((workspaceId: string) => busyWorkspaceIds.has(workspaceId)),
+      hasQueuedWorkspaceTurn: mock(
+        (workspaceId: string, handleId: string) =>
+          queueContainsSecondHandle &&
+          workspaceId === "childworkspace" &&
+          handleId === "wst_secondhandle"
+      ),
+    });
+    const aiMocks = createAIServiceMocks(config, {
+      isStreaming: mock((workspaceId: string) => busyWorkspaceIds.has(workspaceId)),
+    });
+    const streamManager = {
+      getStreamInfo: mock((workspaceId: string) =>
+        busyWorkspaceIds.has(workspaceId)
+          ? activeStreamInfo(
+              workspaceTurnMuxMetadata(parentId, "wst_firsthandle", "firstturn"),
+              "first-message"
+            )
+          : undefined
+      ),
+    } as unknown as StreamManager;
+    const { taskService } = createWorkspaceTurnManagerHarness(config, {
+      aiService: aiMocks.aiService,
+      workspaceService: workspaceMocks.workspaceService,
+      streamManager,
+    });
+
+    const first = await taskService.createWorkspaceTurn({
+      ownerWorkspaceId: parentId,
+      prompt: "First prompt",
+      title: "Workspace turn",
+      workspace: { mode: "new" },
+    });
+    expect(first.success).toBe(true);
+    busyWorkspaceIds.add("childworkspace");
+
+    const secondPromise = taskService.createWorkspaceTurn({
+      ownerWorkspaceId: parentId,
+      prompt: "Queued prompt",
+      title: "Follow-up",
+      workspace: { mode: "existing", workspaceId: "childworkspace" },
+    });
+    await secondSendStarted.promise;
+
+    let snapshot: Awaited<ReturnType<typeof workspaceTurnSnapshot>>;
+    try {
+      snapshot = await workspaceTurnSnapshot(taskService, parentId, "wst_secondhandle");
+    } finally {
+      releaseSecondSend.resolve();
+    }
+    const second = await secondPromise;
+
+    expect(second.success).toBe(true);
+    expect(snapshot).toMatchObject({
+      handleId: "wst_secondhandle",
+      status: "queued",
+      workspaceId: "childworkspace",
+    });
+    expect(await workspaceTurnSnapshot(taskService, parentId, "wst_secondhandle")).toMatchObject({
+      handleId: "wst_secondhandle",
+      status: "queued",
+      workspaceId: "childworkspace",
+    });
+  });
+
+  test("createWorkspaceTurn preserves ownership evidence while releasing stale capacity", async () => {
     const config = await createTestConfig(rootDir);
     stubStableIds(config, ["queuedhandle", "queuedturn"]);
     const { parentId, projectPath } = await saveLocalParentWorkspace(config, rootDir);
@@ -3592,12 +3708,22 @@ describe("WorkspaceTurnManager", () => {
       sendMessage,
       isBusyForMessage: mock((workspaceId: string) => workspaceId === "childworkspace"),
     });
-    const aiMocks = createAIServiceMocks(config, {
-      isStreaming: mock((workspaceId: string) => workspaceId === "otherworkspace"),
-    });
+    let unrelatedActivityObserved = false;
+    const streamManager = {
+      getStreamInfo: mock((workspaceId: string) => {
+        if (workspaceId !== "otherworkspace" || unrelatedActivityObserved) {
+          return undefined;
+        }
+        unrelatedActivityObserved = true;
+        return activeStreamInfo(
+          workspaceTurnMuxMetadata(parentId, "wst_unrelated", "unrelatedturn"),
+          "unrelated-message"
+        );
+      }),
+    } as unknown as StreamManager;
     const { taskService } = createWorkspaceTurnManagerHarness(config, {
-      aiService: aiMocks.aiService,
       workspaceService: workspaceMocks.workspaceService,
+      streamManager,
     });
     const taskHandleStore = (taskService as unknown as { taskHandleStore: TaskHandleStore })
       .taskHandleStore;
@@ -3610,14 +3736,19 @@ describe("WorkspaceTurnManager", () => {
         createdWorkspace: true,
       })
     );
+    // The unrelated activity disappears before settlement. The first observation still
+    // transfers disposable ownership and releases this stale handle's task slot.
     await taskHandleStore.upsertWorkspaceTurn(
       workspaceTurnRecord(parentId, "otherworkspace", "wst_other", "running", {
         turnId: "otherturn",
         createdAt,
         updatedAt: createdAt,
         createdWorkspace: true,
+        disposableWorkspace: true,
+        deferredMessageIds: ["assistant-deferred"],
       })
     );
+    markWorkspaceTurnActive(taskService, "otherworkspace", "wst_other", parentId);
 
     const result = await taskService.createWorkspaceTurn({
       ownerWorkspaceId: parentId,
@@ -3626,10 +3757,16 @@ describe("WorkspaceTurnManager", () => {
       workspace: { mode: "existing", workspaceId: "childworkspace" },
     });
 
-    expect(result.success).toBe(false);
-    if (result.success) return;
-    expect(result.error).toContain("maxParallelAgentTasks exceeded");
-    expect(sendMessage).not.toHaveBeenCalled();
+    expect(result.success).toBe(true);
+    if (!result.success) return;
+    expect(result.data.taskId).toBe("wst_queuedhandle");
+    expect(sendMessage).toHaveBeenCalledTimes(1);
+    expect(await workspaceTurnSnapshot(taskService, parentId, "wst_other")).toMatchObject({
+      status: "interrupted",
+      error: "Workspace turn interrupted after restart",
+      disposableWorkspace: false,
+    });
+    expect(workspaceMocks.remove).not.toHaveBeenCalled();
   });
 
   test("createWorkspaceTurn counts active workspace turns across all owners", async () => {
@@ -3715,9 +3852,19 @@ describe("WorkspaceTurnManager", () => {
       );
       return cfg;
     });
-    const isStreaming = mock((workspaceId: string) => workspaceId === reawakenedTaskId);
-    const { aiService } = createAIServiceMocks(config, { isStreaming });
-    const { taskService, taskHost } = createWorkspaceTurnManagerHarness(config, { aiService });
+    const streamManager = {
+      getStreamInfo: mock((workspaceId: string) =>
+        workspaceId === reawakenedTaskId
+          ? activeStreamInfo(
+              workspaceTurnMuxMetadata(parentId, "wst_reawakened_quota", "turn-reawakened-quota"),
+              "reawakened-message"
+            )
+          : undefined
+      ),
+    } as unknown as StreamManager;
+    const { taskService, taskHost } = createWorkspaceTurnManagerHarness(config, {
+      streamManager,
+    });
     const taskHandleStore = (taskService as unknown as { taskHandleStore: TaskHandleStore })
       .taskHandleStore;
     await taskHandleStore.upsertWorkspaceTurn(
@@ -3760,30 +3907,147 @@ describe("WorkspaceTurnManager", () => {
     });
   });
 
-  test("active workspace turn count keeps startup-retrying handles live", async () => {
-    const hasPendingQueuedOrPreparingTurn = mock(
-      (workspaceId: string) => workspaceId === "childworkspace"
-    );
+  test("active workspace turn count keeps correlated continuation activity live", async () => {
+    let ownerWorkspaceId = "";
+    let activity:
+      | "stream"
+      | "compaction-direct"
+      | "compaction-inherited"
+      | "queued"
+      | "queued-behind"
+      | "auto-retry"
+      | "monitor-wake" = "stream";
+    const hasPendingWorkspaceTurnContinuation = mock(
+      (workspaceId: string, metadata: ReturnType<typeof workspaceTurnMuxMetadata>) =>
+        activity === "queued" &&
+        workspaceId === "childworkspace" &&
+        metadata.taskHandleId === "wst_handle" &&
+        metadata.ownerWorkspaceId === ownerWorkspaceId &&
+        metadata.turnId === "turn"
+    );
+    const hasQueuedWorkspaceTurn = mock(
+      (workspaceId: string, handleId: string) =>
+        activity === "queued-behind" &&
+        workspaceId === "childworkspace" &&
+        handleId === "wst_handle"
+    );
+    const streamManager = {
+      getStreamInfo: mock((workspaceId: string) => {
+        if (workspaceId !== "childworkspace") {
+          return undefined;
+        }
+        const correlation = workspaceTurnMuxMetadata(ownerWorkspaceId, "wst_handle", "turn");
+        if (activity === "stream") {
+          return activeStreamInfo(correlation, "correlated-message");
+        }
+        if (activity === "compaction-direct" || activity === "compaction-inherited") {
+          return activeStreamInfo(
+            {
+              type: "compaction-request",
+              rawCommand: "/compact",
+              source: "auto-compaction",
+              parsed: {
+                followUpContent: {
+                  text: "Continue",
+                  model: "anthropic:claude-opus-4-6",
+                  agentId: "exec",
+                  ...(activity === "compaction-direct"
+                    ? { muxMetadata: correlation }
+                    : { workspaceTurnMetadata: correlation }),
+                },
+              },
+            },
+            "compaction-message"
+          );
+        }
+        return undefined;
+      }),
+    } as unknown as StreamManager;
     const { parentId, taskService } = await startWorkspaceTurnForTest({
-      hasPendingQueuedOrPreparingTurn,
+      hasPendingWorkspaceTurnContinuation,
+      hasQueuedWorkspaceTurn,
+      hasPendingAutoRetry: mock(() => activity === "auto-retry"),
+      hasPendingBashMonitorWakeContinuation: mock(() => activity === "monitor-wake"),
+      streamManager,
     });
+    ownerWorkspaceId = parentId;
     const internal = taskService as unknown as {
       activeWorkspaceTurnHandleByWorkspaceId: Map<
         string,
-        { handleId: string; ownerWorkspaceId: string }
+        { handleId: string; ownerWorkspaceId: string; accepted: boolean }
       >;
       countActiveWorkspaceTurns: () => Promise<number>;
     };
 
-    internal.activeWorkspaceTurnHandleByWorkspaceId.clear();
-    expect(await internal.countActiveWorkspaceTurns()).toBe(1);
-    expect(hasPendingQueuedOrPreparingTurn).toHaveBeenCalledWith("childworkspace");
+    for (const nextActivity of [
+      "stream",
+      "compaction-direct",
+      "compaction-inherited",
+      "queued",
+      "queued-behind",
+      "auto-retry",
+      "monitor-wake",
+    ] as const) {
+      activity = nextActivity;
+      internal.activeWorkspaceTurnHandleByWorkspaceId.clear();
+      if (nextActivity === "auto-retry" || nextActivity === "monitor-wake") {
+        markWorkspaceTurnActive(taskService, "childworkspace", "wst_handle", parentId);
+      }
+      expect(await internal.countActiveWorkspaceTurns()).toBe(1);
+    }
+    expect(hasPendingWorkspaceTurnContinuation).toHaveBeenCalledWith(
+      "childworkspace",
+      workspaceTurnMuxMetadata(parentId, "wst_handle", "turn")
+    );
+    expect(hasQueuedWorkspaceTurn).toHaveBeenCalledWith("childworkspace", "wst_handle");
 
     const snapshot = await workspaceTurnSnapshot(taskService, parentId);
     expect(snapshot).toMatchObject({ status: "running", workspaceId: "childworkspace" });
     expect(snapshot?.error).toBeUndefined();
   });
 
+  for (const continuation of ["auto-retry", "monitor-wake"] as const) {
+    test(
+      "active workspace turn count rejects unrelated " + continuation + " activity",
+      async () => {
+        const { parentId, taskService, created } = await startWorkspaceTurnForTest({
+          hasPendingAutoRetry: mock(
+            (workspaceId: string) =>
+              continuation === "auto-retry" && workspaceId === "childworkspace"
+          ),
+          hasPendingBashMonitorWakeContinuation: mock(
+            (workspaceId: string) =>
+              continuation === "monitor-wake" && workspaceId === "childworkspace"
+          ),
+        });
+        markWorkspaceTurnActive(taskService, created.workspaceId, "wst_other", "other-owner");
+        const internal = taskService as unknown as {
+          countActiveWorkspaceTurns: () => Promise<number>;
+        };
+
+        expect(await internal.countActiveWorkspaceTurns()).toBe(0);
+        expect(await workspaceTurnSnapshot(taskService, parentId)).toMatchObject({
+          status: "interrupted",
+          error: "Workspace turn interrupted after restart",
+        });
+      }
+    );
+  }
+
+  test("active workspace turn count preserves mock streams without StreamInfo", async () => {
+    const isStreaming = mock((workspaceId: string) => workspaceId === "childworkspace");
+    const { parentId, taskService } = await startWorkspaceTurnForTest({ isStreaming });
+    const internal = taskService as unknown as {
+      countActiveWorkspaceTurns: () => Promise<number>;
+    };
+
+    expect(await internal.countActiveWorkspaceTurns()).toBe(1);
+    expect(await workspaceTurnSnapshot(taskService, parentId)).toMatchObject({
+      status: "running",
+      workspaceId: "childworkspace",
+    });
+  });
+
   test("getWorkspaceTurnSnapshot settles stale active handles before returning", async () => {
     const { parentId, taskService } = await startWorkspaceTurnForTest();
     const internal = taskService as unknown as {
@@ -4136,12 +4400,13 @@ describe("WorkspaceTurnManager", () => {
   });
 
   test("mode=existing tool-end follow-up reports the same-owner turn it may supersede", async () => {
-    const hasPendingQueuedOrPreparingTurn = mock(
-      (workspaceId: string) => workspaceId === "childworkspace"
+    const hasPendingWorkspaceTurnContinuation = mock(
+      (workspaceId: string, metadata: ReturnType<typeof workspaceTurnMuxMetadata>) =>
+        workspaceId === "childworkspace" && metadata.taskHandleId === "wst_handle2"
     );
     const { parentId, taskService, workspaceMocks } = await startWorkspaceTurnForTest({
       stableIds: ["handle", "turn", "handle2", "turn2", "handle3", "turn3"],
-      hasPendingQueuedOrPreparingTurn,
+      hasPendingWorkspaceTurnContinuation,
     });
     workspaceMocks.isBusyForMessage.mockImplementation(
       (workspaceId: string) => workspaceId === "childworkspace"
@@ -4196,12 +4461,13 @@ describe("WorkspaceTurnManager", () => {
     // queued, C supersedes B (not A) at B's first boundary — and B's own
     // settlement wake is suppressed, so C's announcement is the only place
     // B's interruption can surface.
-    const hasPendingQueuedOrPreparingTurn = mock(
-      (workspaceId: string) => workspaceId === "childworkspace"
+    const hasPendingWorkspaceTurnContinuation = mock(
+      (workspaceId: string, metadata: ReturnType<typeof workspaceTurnMuxMetadata>) =>
+        workspaceId === "childworkspace" && metadata.taskHandleId === "wst_handle2"
     );
     const { parentId, taskService, workspaceMocks } = await startWorkspaceTurnForTest({
       stableIds: ["handle", "turn", "handle2", "turn2", "handle3", "turn3"],
-      hasPendingQueuedOrPreparingTurn,
+      hasPendingWorkspaceTurnContinuation,
     });
     workspaceMocks.isBusyForMessage.mockImplementation(
       (workspaceId: string) => workspaceId === "childworkspace"
@@ -5559,20 +5825,34 @@ describe("WorkspaceTurnManager", () => {
     },
   ]) {
     test(scenario.name, async () => {
+      let ownerWorkspaceId = "";
       let retryDecisionAwaited = false;
       const pending = mock(
         (workspaceId: string) => retryDecisionAwaited && workspaceId === "childworkspace"
       );
+      const hasPendingWorkspaceTurnContinuation = mock(
+        (workspaceId: string, metadata: ReturnType<typeof workspaceTurnMuxMetadata>) =>
+          scenario.pending === "queued" &&
+          retryDecisionAwaited &&
+          workspaceId === "childworkspace" &&
+          metadata.taskHandleId === "wst_handle" &&
+          metadata.ownerWorkspaceId === ownerWorkspaceId &&
+          metadata.turnId === "turn"
+      );
       const waitForPendingStreamErrorRecoveryDecision = mock((): Promise<void> => {
         retryDecisionAwaited = true;
         return Promise.resolve();
       });
       const { parentId, taskService } = await startWorkspaceTurnForTest({
         ...(scenario.pending === "queued"
-          ? { hasPendingQueuedOrPreparingTurn: pending }
+          ? {
+              hasPendingQueuedOrPreparingTurn: pending,
+              hasPendingWorkspaceTurnContinuation,
+            }
           : { hasPendingAutoRetry: pending }),
         waitForPendingStreamErrorRecoveryDecision,
       });
+      ownerWorkspaceId = parentId;
 
       await taskService.finalizeWorkspaceTurnFromStreamError(scenario.event);
 
diff --git a/src/node/services/workspaceTurnManager.ts b/src/node/services/workspaceTurnManager.ts
index b74f3b9d6e..9dd2c76ae7 100644
--- a/src/node/services/workspaceTurnManager.ts
+++ b/src/node/services/workspaceTurnManager.ts
@@ -52,6 +52,7 @@ import {
 } from "@/common/types/backgroundWorkAttention";
 import {
   createMuxMessage,
+  getCompactionFollowUpContent,
   parseWorkspaceTurnTaskCorrelation,
   type MuxMessage,
   type MuxMessageMetadata,
@@ -296,6 +297,17 @@ const WORKSPACE_TURN_SUPERSEDED_BY_NEW_INPUT_ERROR =
 const WORKSPACE_TURN_SUPERSEDED_BY_OWNER_FOLLOW_UP_ERROR_PREFIX =
   "Workspace turn superseded by follow-up turn ";
 
+interface WorkspaceTurnRuntimeActivity {
+  hasAnyActivity: boolean;
+  hasCorrelatedActivity: boolean;
+  hasUncorrelatedActivity: boolean;
+}
+
+interface WorkspaceTurnLiveness {
+  isLive: boolean;
+  runtimeActivity: WorkspaceTurnRuntimeActivity;
+}
+
 function buildOwnerFollowUpSupersededError(successorHandleId: string): string {
   return (
     `${WORKSPACE_TURN_SUPERSEDED_BY_OWNER_FOLLOW_UP_ERROR_PREFIX}${successorHandleId} from the ` +
@@ -442,6 +454,7 @@ export class WorkspaceTurnManager {
     string,
     { handleId: string; ownerWorkspaceId: string; accepted: boolean }
   >();
+  private readonly workspaceTurnCreationReservationsByWorkspaceId = new Map<string, Set<string>>();
   private lastWorkspaceTurnCreatedAtMs = 0;
   private readonly taskHandleStore: TaskHandleStore;
 
@@ -473,6 +486,22 @@ export class WorkspaceTurnManager {
     return this.activeWorkspaceTurnHandleByWorkspaceId.get(workspaceId);
   }
 
+  private reserveWorkspaceTurnCreation(workspaceId: string, handleId: string) {
+    const reservations =
+      this.workspaceTurnCreationReservationsByWorkspaceId.get(workspaceId) ?? new Set<string>();
+    reservations.add(handleId);
+    this.workspaceTurnCreationReservationsByWorkspaceId.set(workspaceId, reservations);
+
+    return {
+      [Symbol.dispose]: () => {
+        reservations.delete(handleId);
+        if (reservations.size === 0) {
+          this.workspaceTurnCreationReservationsByWorkspaceId.delete(workspaceId);
+        }
+      },
+    };
+  }
+
   async markWorkspaceTurnBackgroundWorkNotifyOnTerminal(
     taskId: string,
     ownerWorkspaceId: string
@@ -1298,6 +1327,10 @@ export class WorkspaceTurnManager {
     // mutex → lifecycle edge of the global lock order (task-tree → this.mutex →
     // workspaceLifecycleLocks; see the workspaceLifecycleLocks declaration), with sorted keys
     // preventing lifecycle-key cycles between concurrent owner/target pairs.
+    // Keep the persisted handle live while sendMessage completes pricing, settings, and queue
+    // admission. The queue does not expose correlation until that preflight finishes.
+    using _creationReservation = this.reserveWorkspaceTurnCreation(targetWorkspaceId, handleId);
+
     const isArchivedInConfig = (workspaceId: string): boolean => {
       const entry = findWorkspaceEntry(this.config.loadConfigOrDefault(), workspaceId);
       return (
@@ -2573,12 +2606,15 @@ export class WorkspaceTurnManager {
       }
     }
 
-    if (
-      isActiveWorkspaceTurnTaskStatus(record.status) &&
-      !(await this.isLiveWorkspaceTurn(record))
-    ) {
-      await this.settleStaleWorkspaceTurn(record);
-      return await this.taskHandleStore.getWorkspaceTurn(record.ownerWorkspaceId, record.handleId);
+    if (isActiveWorkspaceTurnTaskStatus(record.status)) {
+      const liveness = await this.getWorkspaceTurnLiveness(record);
+      if (!liveness.isLive) {
+        await this.settleStaleWorkspaceTurn(record, liveness.runtimeActivity);
+        return await this.taskHandleStore.getWorkspaceTurn(
+          record.ownerWorkspaceId,
+          record.handleId
+        );
+      }
     }
 
     if (
@@ -3761,8 +3797,9 @@ export class WorkspaceTurnManager {
       if (record.workspaceId !== workspaceId || !this.isActiveWorkspaceTurn(record)) {
         continue;
       }
-      if (!(await this.isLiveWorkspaceTurn(record))) {
-        await this.settleStaleWorkspaceTurn(record);
+      const liveness = await this.getWorkspaceTurnLiveness(record);
+      if (!liveness.isLive) {
+        await this.settleStaleWorkspaceTurn(record, liveness.runtimeActivity);
         continue;
       }
       return record;
@@ -3800,54 +3837,167 @@ export class WorkspaceTurnManager {
     return (await this.listActiveWorkspaceTurnTaskIdsForOwner(record.workspaceId)).length > 0;
   }
 
-  private async isLiveWorkspaceTurn(record: WorkspaceTurnTaskHandleRecord): Promise<boolean> {
-    const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId);
-    const hasRuntimeActivity =
-      this.aiService.isStreaming(record.workspaceId) ||
-      this.workspaceService.hasPendingQueuedOrPreparingTurn(record.workspaceId);
-    if (hasRuntimeActivity) {
-      return true;
+  private getRuntimeWorkspaceTurnMetadataFromValue(
+    value: unknown
+  ): { taskHandleId: string; ownerWorkspaceId: string; turnId: string } | undefined {
+    const direct = parseWorkspaceTurnTaskCorrelation(value);
+    if (direct != null) {
+      return direct;
+    }
+    if (value == null || typeof value !== "object") {
+      return undefined;
+    }
+    const compaction = value as MuxMessageMetadata;
+    if (compaction.type !== "compaction-request") {
+      return undefined;
+    }
+    const followUpContent = getCompactionFollowUpContent(compaction);
+    return (
+      parseWorkspaceTurnTaskCorrelation(followUpContent?.muxMetadata) ??
+      parseWorkspaceTurnTaskCorrelation(followUpContent?.workspaceTurnMetadata) ??
+      undefined
+    );
+  }
+
+  private getWorkspaceTurnRuntimeActivity(
+    record: WorkspaceTurnTaskHandleRecord
+  ): WorkspaceTurnRuntimeActivity {
+    const activeStreamInfo = this.streamManager?.getStreamInfo(record.workspaceId);
+    const activeStreamCorrelation = this.getRuntimeWorkspaceTurnMetadataFromValue(
+      activeStreamInfo?.muxMetadata
+    );
+    const hasActiveStream =
+      this.aiService.isStreaming(record.workspaceId) || activeStreamInfo != null;
+    const hasPendingQueuedOrPreparingTurn = this.workspaceService.hasPendingQueuedOrPreparingTurn(
+      record.workspaceId
+    );
+    const hasCorrelatedStream =
+      hasActiveStream &&
+      activeStreamCorrelation?.taskHandleId === record.handleId &&
+      activeStreamCorrelation.ownerWorkspaceId === record.ownerWorkspaceId &&
+      activeStreamCorrelation.turnId === record.turnId;
+    const hasCorrelatedQueuedOrPreparingTurn =
+      this.workspaceService.hasPendingWorkspaceTurnContinuation(
+        record.workspaceId,
+        this.buildWorkspaceTurnMuxMetadata(record)
+      ) || this.workspaceService.hasQueuedWorkspaceTurn(record.workspaceId, record.handleId);
+    const creationReservations = this.workspaceTurnCreationReservationsByWorkspaceId.get(
+      record.workspaceId
+    );
+    const hasCorrelatedCreationReservation = creationReservations?.has(record.handleId) === true;
+    const hasUncorrelatedCreationReservation =
+      creationReservations != null &&
+      creationReservations.size > 0 &&
+      !hasCorrelatedCreationReservation;
+    const activeRegistration = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId);
+    const hasActiveRegistration =
+      activeRegistration?.handleId === record.handleId &&
+      activeRegistration.ownerWorkspaceId === record.ownerWorkspaceId;
+    const hasPendingAutoRetry = this.workspaceService.hasPendingAutoRetry(record.workspaceId);
+    const hasPendingBashMonitorWake = this.workspaceService.hasPendingBashMonitorWakeContinuation(
+      record.workspaceId
+    );
+    const hasCorrelatedRetryOrWake =
+      hasActiveRegistration && (hasPendingAutoRetry || hasPendingBashMonitorWake);
+    const hasUncorrelatedRetryOrWake =
+      !hasActiveRegistration && (hasPendingAutoRetry || hasPendingBashMonitorWake);
+
+    return {
+      hasAnyActivity:
+        hasActiveStream ||
+        hasPendingQueuedOrPreparingTurn ||
+        hasCorrelatedQueuedOrPreparingTurn ||
+        creationReservations != null ||
+        hasPendingAutoRetry ||
+        hasPendingBashMonitorWake,
+      hasCorrelatedActivity:
+        hasCorrelatedStream ||
+        hasCorrelatedQueuedOrPreparingTurn ||
+        hasCorrelatedCreationReservation ||
+        hasCorrelatedRetryOrWake,
+      // A missing StreamInfo is ambiguous because MockAiStreamPlayer reports only through
+      // AIService.isStreaming. Preserve the active-map fallback for that test/runtime path.
+      hasUncorrelatedActivity:
+        (hasActiveStream && activeStreamInfo != null && !hasCorrelatedStream) ||
+        (hasPendingQueuedOrPreparingTurn && !hasCorrelatedQueuedOrPreparingTurn) ||
+        hasUncorrelatedCreationReservation ||
+        hasUncorrelatedRetryOrWake,
+    };
+  }
+
+  private async getWorkspaceTurnLiveness(
+    record: WorkspaceTurnTaskHandleRecord
+  ): Promise<WorkspaceTurnLiveness> {
+    const runtimeActivity = this.getWorkspaceTurnRuntimeActivity(record);
+    if (runtimeActivity.hasCorrelatedActivity) {
+      return { isLive: true, runtimeActivity };
+    }
+    // Only positive evidence of unrelated activity can invalidate the active-map fallback.
+    // Mock streams can report busy without exposing StreamInfo correlation.
+    if (runtimeActivity.hasUncorrelatedActivity) {
+      return { isLive: false, runtimeActivity };
     }
 
+    const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId);
     const isActiveHandle =
       active?.handleId === record.handleId && active.ownerWorkspaceId === record.ownerWorkspaceId;
     if (!isActiveHandle) {
-      return false;
+      return { isLive: false, runtimeActivity };
     }
 
     if ((record.deferredMessageIds?.length ?? 0) === 0) {
-      return true;
+      return { isLive: true, runtimeActivity };
     }
 
     // A deferred workspace-turn stream-end was waiting for background work. Once there is no
     // live stream/queued retry and no active descendant/workflow/nested turn left, the in-memory
     // handle is stale and should be recovered from the deferred history instead of blocking forever.
-    return await this.hasActiveWorkspaceTurnDeferredBlockers(record);
+    return {
+      isLive: await this.hasActiveWorkspaceTurnDeferredBlockers(record),
+      runtimeActivity,
+    };
   }
 
-  private async settleStaleWorkspaceTurn(record: WorkspaceTurnTaskHandleRecord): Promise<void> {
+  private async settleStaleWorkspaceTurn(
+    record: WorkspaceTurnTaskHandleRecord,
+    observedRuntimeActivity: WorkspaceTurnRuntimeActivity
+  ): Promise<void> {
     if (!isActiveWorkspaceTurnTaskStatus(record.status)) {
       return;
     }
+    const runtimeActivity = this.getWorkspaceTurnRuntimeActivity(record);
+    if (runtimeActivity.hasCorrelatedActivity) {
+      return;
+    }
+    // Preserve the ownership evidence that made the caller classify this handle as stale.
+    // The unrelated stream or queue entry can finish before settlement acquires its lock.
+    const disposableOwnershipTransferred =
+      record.disposableWorkspace &&
+      (observedRuntimeActivity.hasUncorrelatedActivity || runtimeActivity.hasUncorrelatedActivity);
     const recovered = await this.recoverTerminalWorkspaceTurnFromHistory(record);
     if (recovered != null) {
+      const next = disposableOwnershipTransferred
+        ? { ...recovered, disposableWorkspace: false }
+        : recovered;
       await this.settleWorkspaceTurn({
         record,
-        next: recovered,
+        next,
         waiterSettlement:
-          recovered.status === "completed"
-            ? { status: "completed", result: this.buildWorkspaceTurnWaitResult(recovered) }
-            : { status: "error", error: new Error(recovered.error ?? "Workspace turn failed") },
+          next.status === "completed"
+            ? { status: "completed", result: this.buildWorkspaceTurnWaitResult(next) }
+            : { status: "error", error: new Error(next.error ?? "Workspace turn failed") },
+        disposableOwnershipTransferred,
       });
       return;
     }
 
     // Same-process deferred stream-ends can be observed before the final assistant message is
-    // readable from history. Keep the handle alive in that narrow window; after restart the active
-    // map is empty, so unrecoverable deferred handles still settle terminally instead of leaking.
+    // readable from history. Keep the handle alive unless unrelated activity owns the runtime.
     const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId);
     if (
       (record.deferredMessageIds?.length ?? 0) > 0 &&
+      !observedRuntimeActivity.hasUncorrelatedActivity &&
+      !runtimeActivity.hasUncorrelatedActivity &&
       active?.handleId === record.handleId &&
       active.ownerWorkspaceId === record.ownerWorkspaceId
     ) {
@@ -3859,6 +4009,7 @@ export class WorkspaceTurnManager {
       status: "interrupted",
       updatedAt: getIsoNow(),
       error: WORKSPACE_TURN_STALE_RESTART_ERROR,
+      ...(disposableOwnershipTransferred ? { disposableWorkspace: false } : {}),
     };
     await this.settleWorkspaceTurn({
       record,
@@ -3867,6 +4018,7 @@ export class WorkspaceTurnManager {
         status: "error",
         error: new Error(WORKSPACE_TURN_STALE_RESTART_ERROR),
       },
+      disposableOwnershipTransferred,
     });
   }
 
@@ -3885,8 +4037,9 @@ export class WorkspaceTurnManager {
       if (!this.isActiveWorkspaceTurn(record)) {
         continue;
       }
-      if (!(await this.isLiveWorkspaceTurn(record))) {
-        await this.settleStaleWorkspaceTurn(record);
+      const liveness = await this.getWorkspaceTurnLiveness(record);
+      if (!liveness.isLive) {
+        await this.settleStaleWorkspaceTurn(record, liveness.runtimeActivity);
         continue;
       }
       if (record.status === "queued") {
@@ -3911,8 +4064,9 @@ export class WorkspaceTurnManager {
     const taskIds: string[] = [];
     for (const record of records) {
       if (isActiveWorkspaceTurnTaskStatus(record.status)) {
-        if (!(await this.isLiveWorkspaceTurn(record))) {
-          await this.settleStaleWorkspaceTurn(record);
+        const liveness = await this.getWorkspaceTurnLiveness(record);
+        if (!liveness.isLive) {
+          await this.settleStaleWorkspaceTurn(record, liveness.runtimeActivity);
           continue;
         }
         taskIds.push(record.handleId);