fix(pxi): stop re-flushing tool outputs the server already holds

The eager flush re-posted client tool outputs the server had already
persisted — on a seeded transcript every resolved output is durable by
definition, yet the flush sent them again on the next evaluation.

Track the IDs the server holds and skip them. The set is fed only from
sound sources: the transcript the server sent, and the persistedToolOutputIds
the acknowledgement itself names. Reading them off the client's live copy
of a message would mark outputs the server never received, and a wrongly
marked output is never flushed again.

This is a pre-existing defect, independent of mutation approval; it is
stacked on that branch only because it was found while building it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: axiomofjoy

js/app/src/agent/chat/__tests__/toolOutputFlush.test.ts

@@ -1,5 +1,6 @@
 import { createClientToolTimingRecorder } from "@phoenix/agent/chat/clientToolTimings";
 import { flushToolOutputs } from "@phoenix/agent/chat/toolOutputFlush";
+import { createTranscriptPersistenceCoordinator } from "@phoenix/agent/chat/transcriptPersistence";
 import type { AgentUIMessage } from "@phoenix/agent/chat/types";
 
 const FLUSH_URL = "/v1/agent_sessions/session-1/tool_outputs";
@@ -74,8 +75,9 @@ describe("flushToolOutputs", () => {
     expect(body.toolOutputs[0].toolCallId).toBe("call-1");
   });
 
-  it("re-posts every resolved output on each call; the endpoint dedupes", async () => {
+  it("skips outputs marked synced and marks flushed outputs in place", async () => {
     const fetchMock = vi.fn().mockResolvedValue(okResponse());
+    const syncedToolOutputIds = new Set<string>();
 
     flushToolOutputs({
       message: assistantMessage([
@@ -84,6 +86,7 @@ describe("flushToolOutputs", () => {
       ]),
       flushUrl: FLUSH_URL,
       fetch: fetchMock,
+      syncedToolOutputIds,
     });
     flushToolOutputs({
       message: assistantMessage([
@@ -93,6 +96,7 @@ describe("flushToolOutputs", () => {
       ]),
       flushUrl: FLUSH_URL,
       fetch: fetchMock,
+      syncedToolOutputIds,
     });
     await settle();
 
@@ -102,7 +106,137 @@ describe("flushToolOutputs", () => {
       secondBody.toolOutputs.map(
         (toolOutput: { toolCallId: string }) => toolOutput.toolCallId
       )
-    ).toEqual(["call-1", "call-2"]);
+    ).toEqual(["call-2"]);
+    expect(syncedToolOutputIds).toEqual(new Set(["call-1", "call-2"]));
+  });
+
+  it("does not post at all when every resolved output is already synced", async () => {
+    const fetchMock = vi.fn().mockResolvedValue(okResponse());
+
+    flushToolOutputs({
+      message: assistantMessage([
+        resolvedToolPart("call-1"),
+        pendingToolPart("call-2"),
+      ]),
+      flushUrl: FLUSH_URL,
+      fetch: fetchMock,
+      syncedToolOutputIds: new Set(["call-1"]),
+    });
+    await settle();
+
+    expect(fetchMock).not.toHaveBeenCalled();
+  });
+
+  it("unmarks flushed outputs when the post fails so a retry can re-flush", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockRejectedValueOnce(new Error("offline"))
+      .mockResolvedValue(okResponse());
+    const syncedToolOutputIds = new Set<string>();
+    const message = assistantMessage([
+      resolvedToolPart("call-1"),
+      pendingToolPart("call-2"),
+    ]);
+
+    flushToolOutputs({
+      message,
+      flushUrl: FLUSH_URL,
+      fetch: fetchMock,
+      syncedToolOutputIds,
+    });
+    await settle();
+    expect(syncedToolOutputIds.size).toBe(0);
+
+    flushToolOutputs({
+      message,
+      flushUrl: FLUSH_URL,
+      fetch: fetchMock,
+      syncedToolOutputIds,
+    });
+    await settle();
+
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(syncedToolOutputIds).toEqual(new Set(["call-1"]));
+  });
+
+  it("unmarks flushed outputs on a non-2xx response", async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: false } as Response);
+    const syncedToolOutputIds = new Set<string>();
+
+    flushToolOutputs({
+      message: assistantMessage([
+        resolvedToolPart("call-1"),
+        pendingToolPart("call-2"),
+      ]),
+      flushUrl: FLUSH_URL,
+      fetch: fetchMock,
+      syncedToolOutputIds,
+    });
+    await settle();
+
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(syncedToolOutputIds.size).toBe(0);
+  });
+
+  it("still retries after a failed flush when the ack does not name the output", async () => {
+    // A failed flush unmarks its IDs so a later flush can retry. The
+    // transcript-persisted ack used to re-mark them from the client's own copy
+    // of the message, cancelling that retry for an output the server never
+    // received. The ack now names only what it wrote, so an output it does not
+    // name stays flushable.
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce({ ok: false, status: 500 } as Response)
+      .mockResolvedValue(okResponse());
+    const coordinator = createTranscriptPersistenceCoordinator();
+    const message = assistantMessage([
+      resolvedToolPart("call-1"),
+      pendingToolPart("call-2"),
+    ]);
+
+    flushToolOutputs({
+      message,
+      flushUrl: FLUSH_URL,
+      fetch: fetchMock,
+      syncedToolOutputIds: coordinator.syncedToolOutputIds,
+    });
+    await settle();
+    expect(coordinator.syncedToolOutputIds.size).toBe(0);
+
+    // The turn persisted without call-1's output, so the ack names nothing.
+    // Its output resolved client-side after the server took its snapshot.
+    coordinator.acknowledge({ messageId: "assistant-1" });
+    coordinator.markToolOutputsPersisted([]);
+
+    flushToolOutputs({
+      message,
+      flushUrl: FLUSH_URL,
+      fetch: fetchMock,
+      syncedToolOutputIds: coordinator.syncedToolOutputIds,
+    });
+    await settle();
+
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(coordinator.syncedToolOutputIds).toEqual(new Set(["call-1"]));
+  });
+
+  it("skips an output the ack named as persisted", async () => {
+    const fetchMock = vi.fn().mockResolvedValue(okResponse());
+    const coordinator = createTranscriptPersistenceCoordinator();
+
+    coordinator.markToolOutputsPersisted(["call-1"]);
+    flushToolOutputs({
+      message: assistantMessage([
+        resolvedToolPart("call-1"),
+        pendingToolPart("call-2"),
+      ]),
+      flushUrl: FLUSH_URL,
+      fetch: fetchMock,
+      syncedToolOutputIds: coordinator.syncedToolOutputIds,
+    });
+    await settle();
+
+    expect(fetchMock).not.toHaveBeenCalled();
   });
 
   it("does not flush when every tool call has resolved", async () => {

js/app/src/agent/chat/__tests__/transcriptPersistence.test.ts

@@ -1,6 +1,10 @@
 import { describe, expect, it } from "vitest";
 
-import { createTranscriptPersistenceCoordinator } from "../transcriptPersistence";
+import {
+  createTranscriptPersistenceCoordinator,
+  resolvedClientToolOutputIds,
+} from "../transcriptPersistence";
+import type { AgentUIMessage } from "../types";
 
 describe("createTranscriptPersistenceCoordinator", () => {
   it("waits for the matching assistant message to be persisted", async () => {
@@ -66,4 +70,78 @@ describe("createTranscriptPersistenceCoordinator", () => {
       coordinator.waitForMessage({ messageId: "assistant-1" })
     ).resolves.toBe(true);
   });
+
+  it("marks only the tool outputs it is given", () => {
+    const coordinator = createTranscriptPersistenceCoordinator();
+
+    coordinator.markToolOutputsPersisted(["tool-call-a", "tool-call-b"]);
+
+    expect(coordinator.syncedToolOutputIds).toEqual(
+      new Set(["tool-call-a", "tool-call-b"])
+    );
+  });
+
+  it("marks nothing when the acknowledgement names no outputs", () => {
+    const coordinator = createTranscriptPersistenceCoordinator();
+
+    // An older server, or a turn that persisted no client outputs. Either way
+    // the flush must stay free to send them.
+    coordinator.markToolOutputsPersisted(undefined);
+    coordinator.markToolOutputsPersisted([]);
+
+    expect(coordinator.syncedToolOutputIds.size).toBe(0);
+  });
+});
+
+describe("resolvedClientToolOutputIds", () => {
+  it("picks resolved client tool outputs off a server-provided message", () => {
+    const message = {
+      id: "assistant-1",
+      role: "assistant",
+      parts: [
+        {
+          type: "tool-edit_prompt",
+          toolCallId: "tool-call-resolved",
+          state: "output-available",
+          input: {},
+          output: { applied: true },
+          callProviderMetadata: {
+            phoenix: { toolExecutionEnvironment: "client" },
+          },
+        },
+        {
+          type: "tool-edit_prompt",
+          toolCallId: "tool-call-pending",
+          state: "input-available",
+          input: {},
+          callProviderMetadata: {
+            phoenix: { toolExecutionEnvironment: "client" },
+          },
+        },
+        {
+          type: "tool-bash",
+          toolCallId: "tool-call-server",
+          state: "output-available",
+          input: {},
+          output: "done",
+        },
+      ],
+    } as AgentUIMessage;
+
+    // Server-executed and still-pending calls are not the client's to flush.
+    expect(resolvedClientToolOutputIds(message)).toEqual([
+      "tool-call-resolved",
+    ]);
+  });
+
+  it("ignores non-assistant and missing messages", () => {
+    expect(resolvedClientToolOutputIds(undefined)).toEqual([]);
+    expect(
+      resolvedClientToolOutputIds({
+        id: "user-1",
+        role: "user",
+        parts: [{ type: "text", text: "hello" }],
+      } as AgentUIMessage)
+    ).toEqual([]);
+  });
 });

js/app/src/agent/chat/createAgentSessionChat.ts

@@ -18,7 +18,10 @@ import {
 } from "@phoenix/agent/chat/rehydratePendingToolCalls";
 import { shouldSendAutomaticallyAfterToolOutput } from "@phoenix/agent/chat/shouldSendAutomatically";
 import { flushToolOutputs } from "@phoenix/agent/chat/toolOutputFlush";
-import { createTranscriptPersistenceCoordinator } from "@phoenix/agent/chat/transcriptPersistence";
+import {
+  createTranscriptPersistenceCoordinator,
+  resolvedClientToolOutputIds,
+} from "@phoenix/agent/chat/transcriptPersistence";
 import { createTurnCompletionGate } from "@phoenix/agent/chat/turnCompletion";
 import type {
   AgentUIMessage,
@@ -111,6 +114,11 @@ export function createAgentSessionChat({
   // racing its own in-flight change.
   let lastAssertedModelSelection: AgentModelSelection | null = null;
   const transcriptPersistence = createTranscriptPersistenceCoordinator();
+  seedMessages.forEach((message) =>
+    transcriptPersistence.markToolOutputsPersisted(
+      resolvedClientToolOutputIds(message)
+    )
+  );
   const turnCompletionGate = createTurnCompletionGate({
     getShouldSendAutomatically: (messages) =>
       shouldSendAutomaticallyAfterToolOutput({
@@ -238,6 +246,12 @@ export function createAgentSessionChat({
         });
       } else if (dataPart.type === "data-transcript-persisted") {
         transcriptPersistence.acknowledge(dataPart.data);
+        // The server names what it wrote. Deriving this from the local copy of
+        // the message instead would mark outputs that resolved after the
+        // server's snapshot, suppressing their flush forever.
+        transcriptPersistence.markToolOutputsPersisted(
+          dataPart.data.persistedToolOutputIds
+        );
       }
     },
     sendAutomaticallyWhen: async ({ messages }) => {
@@ -253,6 +267,7 @@ export function createAgentSessionChat({
             toolTimings,
             locallyInterruptedToolCallIds:
               store.getState().locallyInterruptedToolCallIds,
+            syncedToolOutputIds: transcriptPersistence.syncedToolOutputIds,
           });
         }
         return false;

js/app/src/agent/chat/toolOutputFlush.ts

@@ -21,6 +21,7 @@ export function flushToolOutputs({
   fetch: fetchFn,
   toolTimings = null,
   locallyInterruptedToolCallIds = {},
+  syncedToolOutputIds = null,
 }: {
   /** The transcript's trailing assistant message. */
   message: AgentUIMessage;
@@ -32,6 +33,8 @@ export function flushToolOutputs({
   toolTimings?: ClientToolTimingRecorder | null;
   /** Tool calls this client resolved as interrupted; suppresses the flush. */
   locallyInterruptedToolCallIds?: LocallyInterruptedToolCallIds;
+  /** Tool-call IDs the server already holds; skipped, and updated in place. */
+  syncedToolOutputIds?: Set<string> | null;
 }): void {
   const enrichedMessage = enrichMessageWithClientToolMetadata({
     message,
@@ -41,19 +44,35 @@ export function flushToolOutputs({
   const toolOutputs = getFlushableClientToolOutputs({
     message: enrichedMessage,
     locallyInterruptedToolCallIds,
-  });
+  }).filter((part) => !syncedToolOutputIds?.has(part.toolCallId));
   if (toolOutputs.length === 0) {
     return;
   }
   const body: SubmitToolOutputsRequestBody = {
     toolOutputs,
     lastMessageId: message.id,
   };
+  // Marked before the response lands so an overlapping evaluation (each
+  // approval response re-runs sendAutomaticallyWhen) doesn't double-post.
+  const flushedToolCallIds = toolOutputs.map((part) => part.toolCallId);
+  for (const toolCallId of flushedToolCallIds) {
+    syncedToolOutputIds?.add(toolCallId);
+  }
   void fetchFn(flushUrl, {
     method: "POST",
     headers: { "Content-Type": "application/json" },
     body: JSON.stringify(body),
-  }).catch(() => {
-    // Benign: the chat continuation re-carries resolved outputs.
-  });
+  })
+    .then((response) => {
+      if (!response.ok) {
+        throw new Error(`Tool output flush failed: ${response.status}`);
+      }
+    })
+    .catch(() => {
+      // Benign: the chat continuation re-carries resolved outputs. Unmark so
+      // a later flush may retry before that continuation happens.
+      for (const toolCallId of flushedToolCallIds) {
+        syncedToolOutputIds?.delete(toolCallId);
+      }
+    });
 }