🤖 fix: wake stream-ends no longer falsely settle active workspace turns

src/node/services/taskService.test.ts

@@ -4870,6 +4870,198 @@ describe("TaskService", () => {
     expect(snapshot?.error).toBeUndefined();
   });
 
+  test("uncorrelated wake stream-end with live continuation evidence keeps the handle active", async () => {
+    // Sub-agent progress reports and bash-monitor wakes dispatch new streams
+    // inside the child while the watched delegated turn still runs. When such a
+    // wake stream ends uncorrelated (no workspace-turn muxMetadata) and more
+    // work is still queued or streaming, settling interrupted would report a
+    // false terminal while the child keeps working; defer to the real terminal.
+    const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest({
+      hasPendingBashMonitorWakeContinuation: mock(() => true),
+    });
+
+    const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", {
+      muxMetadata: {
+        type: "workspace-turn-task",
+        taskHandleId: created.taskId,
+        ownerWorkspaceId: parentId,
+        turnId: "turn",
+      },
+    });
+    expect((await historyService.appendToHistory(created.workspaceId, prompt)).success).toBe(true);
+    const wakeOutput = createMuxMessage(
+      "msg_subagent_wake_stream",
+      "assistant",
+      "Sub-agent survey update",
+      { model: "anthropic:claude-opus-4-6", agentId: "exec", finishReason: "stop" }
+    );
+    expect((await historyService.appendToHistory(created.workspaceId, wakeOutput)).success).toBe(
+      true
+    );
+
+    await handleTaskServiceStreamEndForTest(taskService, {
+      type: "stream-end",
+      workspaceId: created.workspaceId,
+      messageId: "msg_subagent_wake_stream",
+      metadata: {
+        model: "anthropic:claude-opus-4-6",
+        agentId: "exec",
+        finishReason: "stop",
+      },
+      parts: [{ type: "text", text: "Sub-agent survey update" }],
+    });
+
+    const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId);
+    expect(snapshot).toMatchObject({ status: "running", workspaceId: created.workspaceId });
+    expect(snapshot?.error).toBeUndefined();
+  });
+
+  test("idle uncorrelated wake stream-end settles the delegated turn from the wake output", async () => {
+    // When a synthetic wake's stream is the delegated turn's last activity and
+    // nothing else is queued or streaming, that end IS the turn outcome.
+    // Ignoring it unconditionally would strand the handle as running until
+    // restart recovery instead of delivering a real report to the owner.
+    const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest();
+
+    const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", {
+      muxMetadata: {
+        type: "workspace-turn-task",
+        taskHandleId: created.taskId,
+        ownerWorkspaceId: parentId,
+        turnId: "turn",
+      },
+    });
+    expect((await historyService.appendToHistory(created.workspaceId, prompt)).success).toBe(true);
+    const wakeOutput = createMuxMessage(
+      "msg_subagent_wake_stream",
+      "assistant",
+      "Sub-agent survey update",
+      { model: "anthropic:claude-opus-4-6", agentId: "exec", finishReason: "stop" }
+    );
+    expect((await historyService.appendToHistory(created.workspaceId, wakeOutput)).success).toBe(
+      true
+    );
+
+    await handleTaskServiceStreamEndForTest(taskService, {
+      type: "stream-end",
+      workspaceId: created.workspaceId,
+      messageId: "msg_subagent_wake_stream",
+      metadata: {
+        model: "anthropic:claude-opus-4-6",
+        agentId: "exec",
+        finishReason: "stop",
+      },
+      parts: [{ type: "text", text: "Sub-agent survey update" }],
+    });
+
+    const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId);
+    expect(snapshot).toMatchObject({
+      status: "completed",
+      workspaceId: created.workspaceId,
+      reportMarkdown: "Sub-agent survey update",
+    });
+  });
+
+  test("compaction-preserved correlation anchors the turn for post-compaction wake ends", async () => {
+    // Auto-compaction hides the correlated prompt behind a summary boundary;
+    // the preserved pendingFollowUp correlation must still anchor the turn so
+    // a later uncorrelated wake end does not supersede it.
+    const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest();
+
+    const compactionSummary = createMuxMessage("compaction-summary", "user", "Compacted context", {
+      muxMetadata: {
+        type: "compaction-summary",
+        pendingFollowUp: {
+          model: "anthropic:claude-opus-4-6",
+          agentId: "exec",
+          text: "Continue the delegated work",
+          workspaceTurnMetadata: {
+            type: "workspace-turn-task",
+            taskHandleId: created.taskId,
+            ownerWorkspaceId: parentId,
+            turnId: "turn",
+          },
+        },
+      },
+    });
+    expect(
+      (await historyService.appendToHistory(created.workspaceId, compactionSummary)).success
+    ).toBe(true);
+    const wakeOutput = createMuxMessage(
+      "msg_post_compaction_wake",
+      "assistant",
+      "Post-compaction wake output",
+      { model: "anthropic:claude-opus-4-6", agentId: "exec", finishReason: "stop" }
+    );
+    expect((await historyService.appendToHistory(created.workspaceId, wakeOutput)).success).toBe(
+      true
+    );
+
+    await handleTaskServiceStreamEndForTest(taskService, {
+      type: "stream-end",
+      workspaceId: created.workspaceId,
+      messageId: "msg_post_compaction_wake",
+      metadata: {
+        model: "anthropic:claude-opus-4-6",
+        agentId: "exec",
+        finishReason: "stop",
+      },
+      parts: [{ type: "text", text: "Post-compaction wake output" }],
+    });
+
+    const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId);
+    expect(snapshot?.status).not.toBe("interrupted");
+    expect(snapshot).toMatchObject({ status: "completed" });
+  });
+
+  test("manual user input still supersedes an active workspace turn on uncorrelated stream-end", async () => {
+    // Only manual (non-synthetic) user rows between the turn prompt and the
+    // uncorrelated stream-end prove the workspace was redirected away from the
+    // delegated turn; the handle must settle interrupted in that case.
+    const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest();
+
+    const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", {
+      muxMetadata: {
+        type: "workspace-turn-task",
+        taskHandleId: created.taskId,
+        ownerWorkspaceId: parentId,
+        turnId: "turn",
+      },
+    });
+    expect((await historyService.appendToHistory(created.workspaceId, prompt)).success).toBe(true);
+    const manualInput = createMuxMessage("manual-input", "user", "Stop that, do something else");
+    expect((await historyService.appendToHistory(created.workspaceId, manualInput)).success).toBe(
+      true
+    );
+    const redirectOutput = createMuxMessage(
+      "msg_redirect_stream",
+      "assistant",
+      "Working on the new request",
+      { model: "anthropic:claude-opus-4-6", agentId: "exec", finishReason: "stop" }
+    );
+    expect(
+      (await historyService.appendToHistory(created.workspaceId, redirectOutput)).success
+    ).toBe(true);
+
+    await handleTaskServiceStreamEndForTest(taskService, {
+      type: "stream-end",
+      workspaceId: created.workspaceId,
+      messageId: "msg_redirect_stream",
+      metadata: {
+        model: "anthropic:claude-opus-4-6",
+        agentId: "exec",
+        finishReason: "stop",
+      },
+      parts: [{ type: "text", text: "Working on the new request" }],
+    });
+
+    const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId);
+    expect(snapshot).toMatchObject({
+      status: "interrupted",
+      error: "Workspace turn superseded by an uncorrelated workspace stream-end",
+    });
+  });
+
   test("compaction stream-end does not advance a running persistent child toward recovery", async () => {
     const config = await createTestConfig(rootDir);
     const projectPath = path.join(rootDir, "repo");