fix: harden block stream dedupe

src/agents/pi-embedded-subscribe.test.ts

@@ -231,6 +231,99 @@ describe("subscribeEmbeddedPiSession", () => {
     expect(subscription.assistantTexts).toEqual(["Hello block"]);
   });
 
+  it("does not duplicate when text_end repeats full content", () => {
+    let handler: ((evt: unknown) => void) | undefined;
+    const session: StubSession = {
+      subscribe: (fn) => {
+        handler = fn;
+        return () => {};
+      },
+    };
+
+    const onBlockReply = vi.fn();
+
+    const subscription = subscribeEmbeddedPiSession({
+      session: session as unknown as Parameters<
+        typeof subscribeEmbeddedPiSession
+      >[0]["session"],
+      runId: "run",
+      onBlockReply,
+      blockReplyBreak: "text_end",
+    });
+
+    handler?.({
+      type: "message_update",
+      message: { role: "assistant" },
+      assistantMessageEvent: {
+        type: "text_delta",
+        delta: "Good morning!",
+      },
+    });
+
+    handler?.({
+      type: "message_update",
+      message: { role: "assistant" },
+      assistantMessageEvent: {
+        type: "text_end",
+        content: "Good morning!",
+      },
+    });
+
+    expect(onBlockReply).toHaveBeenCalledTimes(1);
+    expect(subscription.assistantTexts).toEqual(["Good morning!"]);
+  });
+
+  it("does not duplicate block chunks when text_end repeats full content", () => {
+    let handler: ((evt: unknown) => void) | undefined;
+    const session: StubSession = {
+      subscribe: (fn) => {
+        handler = fn;
+        return () => {};
+      },
+    };
+
+    const onBlockReply = vi.fn();
+
+    subscribeEmbeddedPiSession({
+      session: session as unknown as Parameters<
+        typeof subscribeEmbeddedPiSession
+      >[0]["session"],
+      runId: "run",
+      onBlockReply,
+      blockReplyBreak: "text_end",
+      blockReplyChunking: {
+        minChars: 5,
+        maxChars: 40,
+        breakPreference: "newline",
+      },
+    });
+
+    const fullText = "First line\nSecond line\nThird line\n";
+
+    handler?.({
+      type: "message_update",
+      message: { role: "assistant" },
+      assistantMessageEvent: {
+        type: "text_delta",
+        delta: fullText,
+      },
+    });
+
+    const callsAfterDelta = onBlockReply.mock.calls.length;
+    expect(callsAfterDelta).toBeGreaterThan(0);
+
+    handler?.({
+      type: "message_update",
+      message: { role: "assistant" },
+      assistantMessageEvent: {
+        type: "text_end",
+        content: fullText,
+      },
+    });
+
+    expect(onBlockReply).toHaveBeenCalledTimes(callsAfterDelta);
+  });
+
   it("streams soft chunks with paragraph preference", () => {
     let handler: ((evt: unknown) => void) | undefined;
     const session: StubSession = {

src/agents/pi-embedded-subscribe.ts

@@ -259,6 +259,11 @@ export function subscribeEmbeddedPiSession(params: {
     if (!blockChunking) return;
     const minChars = Math.max(1, Math.floor(blockChunking.minChars));
     const maxChars = Math.max(minChars, Math.floor(blockChunking.maxChars));
+    if (force && blockBuffer.length > 0 && blockBuffer.length <= maxChars) {
+      emitBlockChunk(blockBuffer);
+      blockBuffer = "";
+      return;
+    }
     if (blockBuffer.length < minChars && !force) return;
     while (blockBuffer.length >= minChars || (force && blockBuffer.length > 0)) {
       const breakIdx = pickBreakIndex(blockBuffer);
@@ -437,12 +442,30 @@ export function subscribeEmbeddedPiSession(params: {
             evtType === "text_start" ||
             evtType === "text_end"
           ) {
-            const chunk =
+            const delta =
               typeof assistantRecord?.delta === "string"
                 ? assistantRecord.delta
-                : typeof assistantRecord?.content === "string"
-                  ? assistantRecord.content
-                  : "";
+                : "";
+            const content =
+              typeof assistantRecord?.content === "string"
+                ? assistantRecord.content
+                : "";
+            let chunk = "";
+            if (evtType === "text_delta") {
+              chunk = delta;
+            } else if (evtType === "text_start" || evtType === "text_end") {
+              if (delta) {
+                chunk = delta;
+              } else if (content) {
+                if (content.startsWith(deltaBuffer)) {
+                  chunk = content.slice(deltaBuffer.length);
+                } else if (deltaBuffer.startsWith(content)) {
+                  chunk = "";
+                } else if (!deltaBuffer.includes(content)) {
+                  chunk = content;
+                }
+              }
+            }
             if (chunk) {
               deltaBuffer += chunk;
               blockBuffer += chunk;