refactor(acp): make reply projector single-dispatch (#120560)

src/auto-reply/reply/acp-projector.test.ts

@@ -112,13 +112,6 @@ function emitTool(projector: Projector, event: Omit<EventOf<"tool_call">, "type"
   return projector.onEvent({ type: "tool_call", ...event });
 }
 
-function finishTurn(
-  projector: Projector,
-  event: EventOf<"done"> | EventOf<"error"> = { type: "done" },
-) {
-  return projector.onEvent(event);
-}
-
 async function runHiddenBoundaryCase(params: {
   streamOverrides?: Record<string, unknown>;
   toolCallId: string;
@@ -230,7 +223,7 @@ describe("createAcpReplyProjector", () => {
     await emitText(projector, "done");
     allowToolSummaries = false;
 
-    await finishTurn(projector);
+    await projector.flush(true);
 
     expect(deliveries).toEqual([{ kind: "final", text: "done" }]);
   });
@@ -258,25 +251,11 @@ describe("createAcpReplyProjector", () => {
     await emitText(projector, "I don't");
     allowToolSummaries = false;
 
-    await finishTurn(projector);
+    await projector.flush(true);
 
     expect(deliveries).toEqual([{ kind: "final", text: "fallback.\n\nI don't" }]);
   });
 
-  it("does not suppress identical short text across terminal turn boundaries", async () => {
-    const { deliveries, projector } = createStreamHarness("live");
-
-    await emitText(projector, "A");
-    await finishTurn(projector, { type: "done", stopReason: "end_turn" });
-    await emitText(projector, "A");
-    await finishTurn(projector, { type: "done", stopReason: "end_turn" });
-
-    expect(blockDeliveries(deliveries)).toEqual([
-      { kind: "block", text: "A" },
-      { kind: "block", text: "A" },
-    ]);
-  });
-
   it("flushes staggered live text deltas after idle gaps", async () => {
     vi.useFakeTimers();
     try {
@@ -343,7 +322,7 @@ describe("createAcpReplyProjector", () => {
     await emitText(projector, " now?");
     expect(deliveries).toStrictEqual([]);
 
-    await finishTurn(projector);
+    await projector.flush(true);
     expect(deliveries).toHaveLength(3);
     expect(deliveries[0]).toEqual({
       kind: "tool",
@@ -366,7 +345,7 @@ describe("createAcpReplyProjector", () => {
     });
     expect(deliveries).toStrictEqual([]);
 
-    await finishTurn(projector, { type: "error", message: "turn failed" });
+    await projector.flush(true);
     expect(deliveries).toHaveLength(2);
     expect(deliveries[0]).toEqual({
       kind: "tool",

src/auto-reply/reply/acp-projector.ts

@@ -187,15 +187,13 @@ export function createAcpReplyProjector(params: {
     accountId: params.accountId,
     deliveryMode: settings.deliveryMode,
   });
-  const createTurnBlockReplyPipeline = () =>
-    createBlockReplyPipeline({
-      onBlockReply: async (payload) => {
-        await params.deliver("block", payload);
-      },
-      timeoutMs: ACP_BLOCK_REPLY_TIMEOUT_MS,
-      coalescing: settings.deliveryMode === "live" ? undefined : streaming.coalescing,
-    });
-  let blockReplyPipeline = createTurnBlockReplyPipeline();
+  const blockReplyPipeline = createBlockReplyPipeline({
+    onBlockReply: async (payload) => {
+      await params.deliver("block", payload);
+    },
+    timeoutMs: ACP_BLOCK_REPLY_TIMEOUT_MS,
+    coalescing: settings.deliveryMode === "live" ? undefined : streaming.coalescing,
+  });
   const chunker = new EmbeddedBlockChunker(streaming.chunking);
   const liveIdleFlushMs = Math.max(streaming.coalescing.idleMs, ACP_LIVE_IDLE_FLUSH_FLOOR_MS);
 
@@ -267,23 +265,6 @@ export function createAcpReplyProjector(params: {
     }, liveIdleFlushMs);
   };
 
-  const resetTurnState = () => {
-    clearLiveIdleTimer();
-    blockReplyPipeline.stop();
-    blockReplyPipeline = createTurnBlockReplyPipeline();
-    emittedOutputChars = 0;
-    truncationNoticeEmitted = false;
-    lastStatusHash = undefined;
-    lastToolHash = undefined;
-    lastUsageTuple = undefined;
-    lastVisibleOutputTail = undefined;
-    pendingHiddenBoundary = false;
-    liveBufferText = "";
-    finalOnlyOutputText = "";
-    pendingToolDeliveries.length = 0;
-    toolLifecycleById.clear();
-  };
-
   const flushBufferedToolDeliveries = async (force: boolean) => {
     if (!(settings.deliveryMode === "final_only" && force)) {
       return;
@@ -434,6 +415,7 @@ export function createAcpReplyProjector(params: {
     );
   };
 
+  // One projector serves one dispatch; terminal settlement belongs to tryDispatchAcpReply.
   const onEvent = async (event: AcpRuntimeEvent): Promise<void> => {
     params.onProgress?.();
     if (event.type === "text_delta") {
@@ -514,12 +496,6 @@ export function createAcpReplyProjector(params: {
         return;
       }
       await emitToolSummary(event);
-      return;
-    }
-
-    if (event.type === "done" || event.type === "error") {
-      await flush(true);
-      resetTurnState();
     }
   };