[task-service] 🤖 fix: preserve turns across synthetic wake ends

src/node/services/workspaceTurnManager.test.ts

@@ -18,12 +18,13 @@ import { Ok, Err, type Result } from "@/common/types/result";
 import { DEFAULT_TASK_SETTINGS } from "@/common/types/tasks";
 import type { SendMessageError } from "@/common/types/errors";
 import type { ErrorEvent, StreamEndEvent } from "@/common/types/stream";
-import { createMuxMessage } from "@/common/types/message";
+import { createMuxMessage, type MuxMessageMetadata } from "@/common/types/message";
 import type { WorkspaceMetadata } from "@/common/types/workspace";
 import type { AIService } from "@/node/services/aiService";
 import type {
   WorkspaceHost,
   BackgroundableForegroundWaiter,
+  QueueCutAttributionSnapshot,
   WorkspaceTurnManagerHost,
 } from "@/node/services/taskWorkspaceSeam";
 import type { InitStateManager } from "@/node/services/initStateManager";
@@ -362,6 +363,63 @@ describe("WorkspaceTurnManager", () => {
     };
   }
 
+  async function finalizeWorkspaceTurnStreamEndForTest(
+    taskService: WorkspaceTurnManager,
+    event: StreamEndEvent
+  ): Promise<boolean> {
+    const internal = taskService as unknown as {
+      captureQueueCutAttributionSnapshot: (workspaceId: string) => QueueCutAttributionSnapshot;
+      finalizeWorkspaceTurnFromStreamEnd: (
+        event: StreamEndEvent,
+        queueCutSnapshot: QueueCutAttributionSnapshot
+      ) => Promise<boolean>;
+    };
+    return await internal.finalizeWorkspaceTurnFromStreamEnd(
+      event,
+      internal.captureQueueCutAttributionSnapshot(event.workspaceId)
+    );
+  }
+
+  async function appendUncorrelatedWakeHistory(params: {
+    historyService: HistoryService;
+    parentId: string;
+    taskId: string;
+    workspaceId: string;
+    inputSynthetic: boolean;
+  }): Promise<StreamEndEvent> {
+    const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", {
+      muxMetadata: workspaceTurnMuxMetadata(params.parentId, params.taskId),
+    });
+    expect((await params.historyService.appendToHistory(params.workspaceId, prompt)).success).toBe(
+      true
+    );
+    const input = createMuxMessage("wake-input", "user", "Continue after a wake", {
+      synthetic: params.inputSynthetic,
+    });
+    expect((await params.historyService.appendToHistory(params.workspaceId, input)).success).toBe(
+      true
+    );
+    const output = createMuxMessage("wake-output", "assistant", "Wake result", {
+      model: "anthropic:claude-opus-4-6",
+      agentId: "exec",
+      finishReason: "stop",
+    });
+    expect((await params.historyService.appendToHistory(params.workspaceId, output)).success).toBe(
+      true
+    );
+    return {
+      type: "stream-end",
+      workspaceId: params.workspaceId,
+      messageId: output.id,
+      metadata: {
+        model: "anthropic:claude-opus-4-6",
+        agentId: "exec",
+        finishReason: "stop",
+      },
+      parts: [{ type: "text", text: "Wake result" }],
+    };
+  }
+
   async function createWorkspaceLifecycleHarness(
     options: {
       archived?: boolean;
@@ -3837,6 +3895,157 @@ describe("WorkspaceTurnManager", () => {
     expect(snapshot).toMatchObject({ status: "running", workspaceId: created.workspaceId });
   });
 
+  test("uncorrelated synthetic wake end leaves the active handle running", async () => {
+    const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest();
+    const event = await appendUncorrelatedWakeHistory({
+      historyService,
+      parentId,
+      taskId: created.taskId,
+      workspaceId: created.workspaceId,
+      inputSynthetic: true,
+    });
+
+    expect(await finalizeWorkspaceTurnStreamEndForTest(taskService, event)).toBe(true);
+    expect(await workspaceTurnSnapshot(taskService, parentId, created.taskId)).toMatchObject({
+      status: "running",
+    });
+
+    const correlatedFinal: StreamEndEvent = {
+      ...event,
+      messageId: "real-final",
+      metadata: {
+        ...event.metadata,
+        muxMetadata: workspaceTurnMuxMetadata(parentId, created.taskId),
+      },
+      parts: [{ type: "text", text: "Real final result" }],
+    };
+    expect(await finalizeWorkspaceTurnStreamEndForTest(taskService, correlatedFinal)).toBe(true);
+    expect(await workspaceTurnSnapshot(taskService, parentId, created.taskId)).toMatchObject({
+      status: "completed",
+      messageId: "real-final",
+      reportMarkdown: "Real final result",
+    });
+  });
+
+  test("compaction-preserved turn anchor ignores an uncorrelated synthetic wake end", async () => {
+    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: workspaceTurnMuxMetadata(parentId, created.taskId),
+        },
+      },
+    });
+    expect(
+      (await historyService.appendToHistory(created.workspaceId, compactionSummary)).success
+    ).toBe(true);
+    const wakeInput = createMuxMessage("wake-input", "user", "Continue after a wake", {
+      synthetic: true,
+    });
+    expect((await historyService.appendToHistory(created.workspaceId, wakeInput)).success).toBe(
+      true
+    );
+    const wakeOutput = createMuxMessage("wake-output", "assistant", "Wake result", {
+      model: "anthropic:claude-opus-4-6",
+      agentId: "exec",
+      finishReason: "stop",
+    });
+    expect((await historyService.appendToHistory(created.workspaceId, wakeOutput)).success).toBe(
+      true
+    );
+
+    expect(
+      await finalizeWorkspaceTurnStreamEndForTest(taskService, {
+        type: "stream-end",
+        workspaceId: created.workspaceId,
+        messageId: wakeOutput.id,
+        metadata: {
+          model: "anthropic:claude-opus-4-6",
+          agentId: "exec",
+          finishReason: "stop",
+        },
+        parts: [{ type: "text", text: "Wake result" }],
+      })
+    ).toBe(true);
+    expect(await workspaceTurnSnapshot(taskService, parentId, created.taskId)).toMatchObject({
+      status: "running",
+    });
+  });
+
+  test("manual user input still supersedes an active workspace turn on uncorrelated end", async () => {
+    const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest();
+    const event = await appendUncorrelatedWakeHistory({
+      historyService,
+      parentId,
+      taskId: created.taskId,
+      workspaceId: created.workspaceId,
+      inputSynthetic: false,
+    });
+
+    expect(await finalizeWorkspaceTurnStreamEndForTest(taskService, event)).toBe(true);
+    expect(await workspaceTurnSnapshot(taskService, parentId, created.taskId)).toMatchObject({
+      status: "interrupted",
+      messageId: "wake-output",
+      error: "Workspace turn superseded by an uncorrelated workspace stream-end",
+    });
+  });
+
+  test("malformed user-triggered auto-compaction still supersedes an active workspace turn", async () => {
+    const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest();
+    const prompt = createMuxMessage("turn-prompt", "user", "Summarize the repo", {
+      muxMetadata: workspaceTurnMuxMetadata(parentId, created.taskId),
+    });
+    expect((await historyService.appendToHistory(created.workspaceId, prompt)).success).toBe(true);
+    const compactionRequest = createMuxMessage(
+      "auto-compaction",
+      "user",
+      "Compacting before a new user prompt",
+      {
+        synthetic: true,
+        muxMetadata: {
+          type: "compaction-request",
+          rawCommand: "/compact",
+          parsed: null,
+          source: "auto-compaction",
+        } as unknown as MuxMessageMetadata,
+      }
+    );
+    expect(
+      (await historyService.appendToHistory(created.workspaceId, compactionRequest)).success
+    ).toBe(true);
+    const wakeOutput = createMuxMessage("wake-output", "assistant", "Wake result", {
+      model: "anthropic:claude-opus-4-6",
+      agentId: "exec",
+      finishReason: "stop",
+    });
+    expect((await historyService.appendToHistory(created.workspaceId, wakeOutput)).success).toBe(
+      true
+    );
+
+    expect(
+      await finalizeWorkspaceTurnStreamEndForTest(taskService, {
+        type: "stream-end",
+        workspaceId: created.workspaceId,
+        messageId: wakeOutput.id,
+        metadata: {
+          model: "anthropic:claude-opus-4-6",
+          agentId: "exec",
+          finishReason: "stop",
+        },
+        parts: [{ type: "text", text: "Wake result" }],
+      })
+    ).toBe(true);
+    expect(await workspaceTurnSnapshot(taskService, parentId, created.taskId)).toMatchObject({
+      status: "interrupted",
+      messageId: wakeOutput.id,
+      error: "Workspace turn superseded by an uncorrelated workspace stream-end",
+    });
+  });
+
   test("getWorkspaceTurnSnapshot recovers stale completed handles from matching history", async () => {
     const { parentId, taskService, historyService, created } = await startWorkspaceTurnForTest();
     const appendResult = await historyService.appendToHistory(

src/node/services/workspaceTurnManager.ts

@@ -52,6 +52,7 @@ import {
 } from "@/common/types/backgroundWorkAttention";
 import {
   createMuxMessage,
+  getCompactionFollowUpContent,
   parseWorkspaceTurnTaskCorrelation,
   type MuxMessage,
   type MuxMessageMetadata,
@@ -281,6 +282,22 @@ const WORKSPACE_TURN_STALE_RESTART_ERROR = "Workspace turn interrupted after res
 const WORKSPACE_TURN_SUPERSEDED_BY_NEW_INPUT_ERROR =
   "Workspace turn superseded by new input in the target workspace; the workspace continues under that input and this delegated turn will not report";
 
+/** A human-authored child input that redirects the delegated turn. */
+function isManualChildWorkspaceInput(message: MuxMessage): boolean {
+  if (message.role !== "user") {
+    return false;
+  }
+  if (message.metadata?.synthetic !== true) {
+    return true;
+  }
+  const muxMetadata = message.metadata.muxMetadata;
+  return (
+    muxMetadata?.type === "compaction-request" &&
+    muxMetadata.source === "auto-compaction" &&
+    getCompactionFollowUpContent(muxMetadata)?.dispatchOptions?.source !== "internal-resume"
+  );
+}
+
 /**
  * Reason prefix persisted when the owner's OWN follow-up turn (task
  * kind="workspace", mode="existing", tool-end dispatch) cut its active
@@ -4217,37 +4234,27 @@ export class WorkspaceTurnManager {
     };
   }
 
-  private async isStreamEndBeforeWorkspaceTurnPrompt(
+  private isWorkspaceTurnAnchorForRecord(
     record: WorkspaceTurnTaskHandleRecord,
-    event: StreamEndEvent
-  ): Promise<boolean> {
-    const historyResult = await this.historyService.getHistoryFromLatestBoundary(event.workspaceId);
-    if (!historyResult.success) {
-      log.warn("Could not compare uncorrelated stream-end history for workspace turn", {
-        workspaceId: event.workspaceId,
-        handleId: record.handleId,
-        error: historyResult.error,
-      });
-      return false;
+    message: MuxMessage
+  ): boolean {
+    const muxMetadata = message.metadata?.muxMetadata;
+    if (muxMetadata?.type === "workspace-turn-task") {
+      return (
+        muxMetadata.taskHandleId === record.handleId &&
+        muxMetadata.ownerWorkspaceId === record.ownerWorkspaceId &&
+        muxMetadata.turnId === record.turnId
+      );
     }
-
-    let streamEndIndex = -1;
-    let promptIndex = -1;
-    for (const [index, message] of historyResult.data.entries()) {
-      if (message.id === event.messageId) {
-        streamEndIndex = index;
-      }
-      const metadata = this.getWorkspaceTurnMetadataFromValue(message.metadata?.muxMetadata);
-      if (
-        metadata?.taskHandleId === record.handleId &&
-        metadata.ownerWorkspaceId === record.ownerWorkspaceId &&
-        metadata.turnId === record.turnId
-      ) {
-        promptIndex = index;
-      }
+    if (muxMetadata?.type === "compaction-summary") {
+      const preserved = muxMetadata.pendingFollowUp?.workspaceTurnMetadata;
+      return (
+        preserved?.taskHandleId === record.handleId &&
+        preserved.ownerWorkspaceId === record.ownerWorkspaceId &&
+        preserved.turnId === record.turnId
+      );
     }
-
-    return streamEndIndex !== -1 && promptIndex !== -1 && streamEndIndex < promptIndex;
+    return false;
   }
 
   private async interruptWorkspaceTurnFromUncorrelatedStreamEnd(
@@ -4281,7 +4288,33 @@ export class WorkspaceTurnManager {
       return true;
     }
 
-    if (await this.isStreamEndBeforeWorkspaceTurnPrompt(record, event)) {
+    const historyResult = await this.historyService.getHistoryFromLatestBoundary(event.workspaceId);
+    if (!historyResult.success) {
+      log.warn("Could not compare uncorrelated stream-end history for workspace turn", {
+        workspaceId: event.workspaceId,
+        handleId: record.handleId,
+        error: historyResult.error,
+      });
+      await this.settleWorkspaceTurnSupersededFromUncorrelatedStreamEnd(record, event);
+      return true;
+    }
+
+    let streamEndIndex = -1;
+    let turnAnchorIndex = -1;
+    for (const [index, message] of historyResult.data.entries()) {
+      if (message.id === event.messageId) {
+        streamEndIndex = index;
+      }
+      if (this.isWorkspaceTurnAnchorForRecord(record, message)) {
+        turnAnchorIndex = index;
+      }
+    }
+
+    if (streamEndIndex === -1 || turnAnchorIndex === -1) {
+      await this.settleWorkspaceTurnSupersededFromUncorrelatedStreamEnd(record, event);
+      return true;
+    }
+    if (streamEndIndex < turnAnchorIndex) {
       log.debug("Ignoring stale uncorrelated stream-end before queued workspace turn prompt", {
         workspaceId: event.workspaceId,
         taskHandleId: record.handleId,
@@ -4290,6 +4323,19 @@ export class WorkspaceTurnManager {
       return true;
     }
 
+    const hasManualSupersessionInput = historyResult.data
+      .slice(turnAnchorIndex + 1, streamEndIndex)
+      .some(isManualChildWorkspaceInput);
+    if (hasManualSupersessionInput) {
+      await this.settleWorkspaceTurnSupersededFromUncorrelatedStreamEnd(record, event);
+    }
+    return true;
+  }
+
+  private async settleWorkspaceTurnSupersededFromUncorrelatedStreamEnd(
+    record: WorkspaceTurnTaskHandleRecord,
+    event: StreamEndEvent
+  ): Promise<void> {
     const error = "Workspace turn superseded by an uncorrelated workspace stream-end";
     const next: WorkspaceTurnTaskHandleRecord = {
       ...record,
@@ -4298,12 +4344,12 @@ export class WorkspaceTurnManager {
       messageId: event.messageId,
       error,
     };
+    delete next.deferredMessageIds;
     await this.settleWorkspaceTurn({
       record,
       next,
       waiterSettlement: { status: "error", error: new Error(error) },
     });
-    return true;
   }
 
   /**