fix: prevent duplicate agent event emission

src/agents/pi-embedded-subscribe.handlers.lifecycle.ts

@@ -24,6 +24,11 @@ export function handleAutoCompactionStart(ctx: EmbeddedPiSubscribeContext) {
   ctx.state.compactionInFlight = true;
   ctx.ensureCompactionPromise();
   ctx.log.debug(`embedded run compaction start: runId=${ctx.params.runId}`);
+  emitAgentEvent({
+    runId: ctx.params.runId,
+    stream: "compaction",
+    data: { phase: "start" },
+  });
   void ctx.params.onAgentEvent?.({
     stream: "compaction",
     data: { phase: "start" },
@@ -43,6 +48,11 @@ export function handleAutoCompactionEnd(
   } else {
     ctx.maybeResolveCompactionWait();
   }
+  emitAgentEvent({
+    runId: ctx.params.runId,
+    stream: "compaction",
+    data: { phase: "end", willRetry },
+  });
   void ctx.params.onAgentEvent?.({
     stream: "compaction",
     data: { phase: "end", willRetry },

src/agents/pi-embedded-subscribe.handlers.messages.ts

@@ -109,34 +109,34 @@ export function handleMessageUpdate(
     .trim();
   if (next && next !== ctx.state.lastStreamedAssistant) {
     const previousText = ctx.state.lastStreamedAssistant ?? "";
-    ctx.state.lastStreamedAssistant = next;
     const { text: cleanedText, mediaUrls } = parseReplyDirectives(next);
     const { text: previousCleanedText } = parseReplyDirectives(previousText);
-    const deltaText = cleanedText.startsWith(previousCleanedText)
-      ? cleanedText.slice(previousCleanedText.length)
-      : cleanedText;
-    emitAgentEvent({
-      runId: ctx.params.runId,
-      stream: "assistant",
-      data: {
-        text: cleanedText,
-        delta: deltaText,
-        mediaUrls: mediaUrls?.length ? mediaUrls : undefined,
-      },
-    });
-    void ctx.params.onAgentEvent?.({
-      stream: "assistant",
-      data: {
-        text: cleanedText,
-        delta: deltaText,
-        mediaUrls: mediaUrls?.length ? mediaUrls : undefined,
-      },
-    });
-    if (ctx.params.onPartialReply && ctx.state.shouldEmitPartialReplies) {
-      void ctx.params.onPartialReply({
-        text: cleanedText,
-        mediaUrls: mediaUrls?.length ? mediaUrls : undefined,
+    if (cleanedText.startsWith(previousCleanedText)) {
+      const deltaText = cleanedText.slice(previousCleanedText.length);
+      ctx.state.lastStreamedAssistant = next;
+      emitAgentEvent({
+        runId: ctx.params.runId,
+        stream: "assistant",
+        data: {
+          text: cleanedText,
+          delta: deltaText,
+          mediaUrls: mediaUrls?.length ? mediaUrls : undefined,
+        },
       });
+      void ctx.params.onAgentEvent?.({
+        stream: "assistant",
+        data: {
+          text: cleanedText,
+          delta: deltaText,
+          mediaUrls: mediaUrls?.length ? mediaUrls : undefined,
+        },
+      });
+      if (ctx.params.onPartialReply && ctx.state.shouldEmitPartialReplies) {
+        void ctx.params.onPartialReply({
+          text: cleanedText,
+          mediaUrls: mediaUrls?.length ? mediaUrls : undefined,
+        });
+      }
     }
   }
 

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

@@ -184,4 +184,40 @@ describe("subscribeEmbeddedPiSession", () => {
     expect(payloads[1]?.text).toBe("Hello world");
     expect(payloads[1]?.delta).toBe(" world");
   });
+
+  it("skips agent events when cleaned text rewinds mid-stream", () => {
+    let handler: ((evt: unknown) => void) | undefined;
+    const session: StubSession = {
+      subscribe: (fn) => {
+        handler = fn;
+        return () => {};
+      },
+    };
+
+    const onAgentEvent = vi.fn();
+
+    subscribeEmbeddedPiSession({
+      session: session as unknown as Parameters<typeof subscribeEmbeddedPiSession>[0]["session"],
+      runId: "run",
+      onAgentEvent,
+    });
+
+    handler?.({ type: "message_start", message: { role: "assistant" } });
+    handler?.({
+      type: "message_update",
+      message: { role: "assistant" },
+      assistantMessageEvent: { type: "text_delta", delta: "MEDIA:" },
+    });
+    handler?.({
+      type: "message_update",
+      message: { role: "assistant" },
+      assistantMessageEvent: { type: "text_delta", delta: " https://example.com/a.png\nCaption" },
+    });
+
+    const payloads = onAgentEvent.mock.calls
+      .map((call) => call[0]?.data as Record<string, unknown> | undefined)
+      .filter((value): value is Record<string, unknown> => Boolean(value));
+    expect(payloads).toHaveLength(1);
+    expect(payloads[0]?.text).toBe("MEDIA:");
+  });
 });

src/agents/pi-embedded-subscribe.subscribe-embedded-pi-session.waits-multiple-compaction-retries-before-resolving.test.ts

@@ -1,4 +1,5 @@
 import { describe, expect, it, vi } from "vitest";
+import { onAgentEvent } from "../infra/agent-events.js";
 import { subscribeEmbeddedPiSession } from "./pi-embedded-subscribe.js";
 
 type StubSession = {
@@ -54,6 +55,44 @@ describe("subscribeEmbeddedPiSession", () => {
     await waitPromise;
     expect(resolved).toBe(true);
   });
+
+  it("emits compaction events on the agent event bus", async () => {
+    let handler: ((evt: unknown) => void) | undefined;
+    const session: StubSession = {
+      subscribe: (fn) => {
+        handler = fn;
+        return () => {};
+      },
+    };
+
+    const events: Array<{ phase: string; willRetry?: boolean }> = [];
+    const stop = onAgentEvent((evt) => {
+      if (evt.runId !== "run-compaction") return;
+      if (evt.stream !== "compaction") return;
+      const phase = typeof evt.data?.phase === "string" ? evt.data.phase : "";
+      events.push({
+        phase,
+        willRetry: typeof evt.data?.willRetry === "boolean" ? evt.data.willRetry : undefined,
+      });
+    });
+
+    subscribeEmbeddedPiSession({
+      session: session as unknown as Parameters<typeof subscribeEmbeddedPiSession>[0]["session"],
+      runId: "run-compaction",
+    });
+
+    handler?.({ type: "auto_compaction_start" });
+    handler?.({ type: "auto_compaction_end", willRetry: true });
+    handler?.({ type: "auto_compaction_end", willRetry: false });
+
+    stop();
+
+    expect(events).toEqual([
+      { phase: "start" },
+      { phase: "end", willRetry: true },
+      { phase: "end", willRetry: false },
+    ]);
+  });
   it("emits tool summaries at tool start when verbose is on", async () => {
     let handler: ((evt: unknown) => void) | undefined;
     const session: StubSession = {