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

src/node/services/taskHandleStore.ts

@@ -72,6 +72,12 @@ export interface WorkspaceTurnTaskHandleRecord {
     metadata: StreamEndEvent["metadata"];
   };
   deferredMessageIds?: string[];
+  /**
+   * True only for a terminal settlement produced from an uncorrelated synthetic
+   * wake stream-end (no turn correlation existed on that stream). A later
+   * strictly-correlated stream-end may replace this provisional outcome.
+   */
+  provisionalOutcome?: boolean;
   error?: string;
   /**
    * How the owner workspace's stream-end treats this workspace turn while active.
@@ -118,6 +124,7 @@ const WorkspaceTurnTaskHandleRecordSchema = z
       .passthrough()
       .optional(),
     deferredMessageIds: z.array(z.string().min(1)).optional(),
+    provisionalOutcome: z.boolean().optional(),
     error: z.string().optional(),
     attentionPolicy: BackgroundWorkAttentionPolicySchema.optional(),
     directParentResultDeliveryRequiredAt: z.string().optional(),

src/node/services/taskService.test.ts

@@ -4870,6 +4870,369 @@ 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("history read failure on uncorrelated wake end still settles the handle as interrupted", async () => {
+    // Without history we cannot classify the end, but it is the only stream-end
+    // that can settle the waiter. The supersede fallback preserves the
+    // pre-existing fail-safe instead of stranding the handle as running.
+    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);
+
+    // Inject for every read in this test: handleStreamEnd performs earlier
+    // history reads that would consume a one-shot mock.
+    spyOn(historyService, "getHistoryFromLatestBoundary").mockImplementation(() =>
+      Promise.resolve(Err("Failed to read history from boundary: injected test failure"))
+    );
+
+    await handleTaskServiceStreamEndForTest(taskService, {
+      type: "stream-end",
+      workspaceId: created.workspaceId,
+      messageId: "msg_wake_stream",
+      metadata: {
+        model: "anthropic:claude-opus-4-6",
+        agentId: "exec",
+        finishReason: "stop",
+      },
+      parts: [{ type: "text", text: "Wake output" }],
+    });
+
+    const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId);
+    expect(snapshot).toMatchObject({
+      status: "interrupted",
+      error: "Workspace turn superseded by an uncorrelated workspace stream-end",
+    });
+  });
+
+  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("correlated terminal replaces a provisional idle-wake completion", async () => {
+    // The provisional completion from an uncorrelated wake end must stay
+    // replaceable: a later strictly-correlated stream-end carries the real
+    // outcome and wins over the synthetic wake report.
+    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" }],
+    });
+    expect(await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId)).toMatchObject({
+      status: "completed",
+      reportMarkdown: "Sub-agent survey update",
+    });
+
+    await handleTaskServiceStreamEndForTest(taskService, {
+      type: "stream-end",
+      workspaceId: created.workspaceId,
+      messageId: "msg_real_final",
+      metadata: {
+        model: "anthropic:claude-opus-4-6",
+        agentId: "exec",
+        finishReason: "stop",
+        muxMetadata: {
+          type: "workspace-turn-task",
+          taskHandleId: created.taskId,
+          ownerWorkspaceId: parentId,
+          turnId: "turn",
+        },
+      },
+      parts: [{ type: "text", text: "Real final report" }],
+    });
+
+    expect(await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId)).toMatchObject({
+      status: "completed",
+      reportMarkdown: "Real final report",
+    });
+  });
+
+  test("deferred wake end settles once continuation evidence disappears", async () => {
+    // Continuation signals can vanish without producing another stream-end
+    // (rejected queued send, abandoned retry, removed monitor wake). The
+    // persisted deferred marker must stop the handle counting as live so stale
+    // recovery settles it from the deferred wake output instead of blocking
+    // the owner until restart.
+    let monitorWakePending = true;
+    const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest({
+      hasPendingBashMonitorWakeContinuation: mock(() => monitorWakePending),
+    });
+
+    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" }],
+    });
+    expect(await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId)).toMatchObject({
+      status: "running",
+    });
+
+    const store = (
+      taskService as unknown as {
+        taskHandleStore: {
+          getWorkspaceTurn: (
+            ownerWorkspaceId: string,
+            handleId: string
+          ) => Promise<{ deferredMessageIds?: string[] } | null>;
+        };
+      }
+    ).taskHandleStore;
+    const record = await store.getWorkspaceTurn(parentId, created.taskId);
+    expect(record?.deferredMessageIds).toContain("msg_subagent_wake_stream");
+
+    monitorWakePending = false;
+    expect(await taskService.getWorkspaceTurnSnapshot(parentId, created.taskId)).toMatchObject({
+      status: "completed",
+      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");