fix(agents): dedupe transcript rewrite replay

CaptainTimon · CaptainTimon · commit ee0a371f0fad · 2026-05-08T02:42:43.000+02:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -160,6 +160,7 @@ Docs: https://docs.openclaw.ai
 
 ### Fixes
 
+- Agents/transcript: dedupe exact replayed transcript entries during overflow-recovery rewrites so cloned user prompts, compactions, and bootstrap context records do not keep amplifying session JSONL growth. Refs #69208 and fixes #66443. Thanks @BradGroux.
 - Gateway/watch: leave `OPENCLAW_TRACE_SYNC_IO` disabled by default in `pnpm gateway:watch:raw` so watch mode avoids noisy Node sync-I/O stack traces unless explicitly requested.
 - Providers: preserve non-OK `text/event-stream` response bodies so provider HTTP errors keep their JSON detail instead of collapsing to generic streaming failures. Fixes #78180.
 - Gateway/auth: make explicit `trusted-proxy` mode fail closed instead of accepting local password fallback credentials after trusted-proxy identity checks fail. Fixes #78684.
diff --git a/src/agents/pi-embedded-runner/transcript-rewrite.test.ts b/src/agents/pi-embedded-runner/transcript-rewrite.test.ts
@@ -271,6 +271,115 @@ describe("rewriteTranscriptEntriesInSessionManager", () => {
       content: [{ type: "text", text: "summarized" }],
     });
   });
+
+  it("dedupes exact replayed user payloads when rewriting a polluted branch", () => {
+    const sessionManager = SessionManager.inMemory();
+    const heartbeatPrompt = "Read HEARTBEAT.md if it exists (workspace context)...";
+    const replayedUserMessage = asAppendMessage({
+      role: "user",
+      content: heartbeatPrompt,
+      timestamp: 1,
+    });
+    const entryIds = appendSessionMessages(sessionManager, [
+      replayedUserMessage,
+      replayedUserMessage,
+      replayedUserMessage,
+      asAppendMessage({
+        role: "assistant",
+        content: createTextContent("ack"),
+        timestamp: 2,
+      }),
+    ]);
+
+    const result = rewriteTranscriptEntriesInSessionManager({
+      sessionManager,
+      replacements: [
+        {
+          entryId: entryIds[0],
+          message: replayedUserMessage as AgentMessage,
+        },
+      ],
+    });
+
+    expect(result.changed).toBe(true);
+    const branchMessages = getBranchMessages(sessionManager);
+    expect(branchMessages.map((message) => message.role)).toEqual(["user", "assistant"]);
+    expect(branchMessages[0]).toMatchObject({
+      role: "user",
+      content: heartbeatPrompt,
+      timestamp: 1,
+    });
+  });
+
+  it("preserves repeated user text when message timestamps differ", () => {
+    const sessionManager = SessionManager.inMemory();
+    const entryIds = appendSessionMessages(sessionManager, [
+      asAppendMessage({ role: "user", content: "yes", timestamp: 1 }),
+      asAppendMessage({ role: "user", content: "yes", timestamp: 2 }),
+      asAppendMessage({
+        role: "assistant",
+        content: createTextContent("ok"),
+        timestamp: 3,
+      }),
+    ]);
+
+    const result = rewriteTranscriptEntriesInSessionManager({
+      sessionManager,
+      replacements: [
+        {
+          entryId: entryIds[0],
+          message: { role: "user", content: "yes", timestamp: 1 } as AgentMessage,
+        },
+      ],
+    });
+
+    expect(result.changed).toBe(true);
+    expect(getBranchMessages(sessionManager).map((message) => message.role)).toEqual([
+      "user",
+      "user",
+      "assistant",
+    ]);
+  });
+
+  it("dedupes replayed bootstrap custom entries and cloned compactions", () => {
+    const sessionManager = SessionManager.inMemory();
+    const entryIds = appendSessionMessages(sessionManager, [
+      asAppendMessage({ role: "user", content: "start", timestamp: 1 }),
+      asAppendMessage({
+        role: "assistant",
+        content: createTextContent("kept"),
+        timestamp: 2,
+      }),
+    ]);
+    sessionManager.appendCustomEntry("openclaw:bootstrap-context:full", { hash: "same" });
+    sessionManager.appendCustomEntry("openclaw:bootstrap-context:full", { hash: "same" });
+    const firstCompactionId = sessionManager.appendCompaction("summary", entryIds[1], 9_000, {
+      readFiles: ["HEARTBEAT.md"],
+    });
+    sessionManager.appendCompaction("summary", firstCompactionId, 9_000, {
+      readFiles: ["HEARTBEAT.md"],
+    });
+
+    const result = rewriteTranscriptEntriesInSessionManager({
+      sessionManager,
+      replacements: [
+        {
+          entryId: entryIds[0],
+          message: { role: "user", content: "start", timestamp: 1 } as AgentMessage,
+        },
+      ],
+    });
+
+    expect(result.changed).toBe(true);
+    const branch = sessionManager.getBranch();
+    expect(
+      branch.filter(
+        (entry) =>
+          entry.type === "custom" && entry.customType === "openclaw:bootstrap-context:full",
+      ),
+    ).toHaveLength(1);
+    expect(branch.filter((entry) => entry.type === "compaction")).toHaveLength(1);
+  });
 });
 
 describe("rewriteTranscriptEntriesInSessionFile", () => {
@@ -344,4 +453,49 @@ describe("rewriteTranscriptEntriesInSessionFile", () => {
       openSpy.mockRestore();
     }
   });
+
+  it("persists only one copy of exact replayed user payloads during file rewrite", async () => {
+    const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-transcript-rewrite-"));
+    const sessionManager = SessionManager.create(dir, dir);
+    const replayedUserMessage = asAppendMessage({
+      role: "user",
+      content: "Read HEARTBEAT.md if it exists (workspace context)...",
+      timestamp: 1,
+    });
+    const entryIds = appendSessionMessages(sessionManager, [
+      replayedUserMessage,
+      replayedUserMessage,
+      replayedUserMessage,
+      asAppendMessage({
+        role: "assistant",
+        content: createTextContent("ack"),
+        timestamp: 2,
+      }),
+    ]);
+    const sessionFile = sessionManager.getSessionFile();
+    expect(sessionFile).toBeTruthy();
+    if (!sessionFile) {
+      throw new Error("expected persisted session file");
+    }
+
+    const result = await rewriteTranscriptEntriesInSessionFile({
+      sessionFile,
+      sessionKey: "agent:main:main:heartbeat",
+      request: {
+        replacements: [
+          {
+            entryId: entryIds[0],
+            message: replayedUserMessage as AgentMessage,
+          },
+        ],
+      },
+    });
+
+    expect(result.changed).toBe(true);
+    const rewrittenSession = SessionManager.open(sessionFile);
+    expect(getBranchMessages(rewrittenSession).map((message) => message.role)).toEqual([
+      "user",
+      "assistant",
+    ]);
+  });
 });
diff --git a/src/agents/pi-embedded-runner/transcript-rewrite.ts b/src/agents/pi-embedded-runner/transcript-rewrite.ts
@@ -1,3 +1,4 @@
+import { createHash } from "node:crypto";
 import type { AgentMessage } from "@mariozechner/pi-agent-core";
 import { SessionManager } from "@mariozechner/pi-coding-agent";
 import type {
@@ -27,6 +28,40 @@ function estimateMessageBytes(message: AgentMessage): number {
   return Buffer.byteLength(JSON.stringify(message), "utf8");
 }
 
+function hashJson(value: unknown): string {
+  return createHash("sha256")
+    .update(JSON.stringify(value ?? null))
+    .digest("hex");
+}
+
+function computeBranchEntryReplayIdentity(entry: SessionBranchEntry): string | null {
+  if (entry.type === "message") {
+    return `message:${hashJson(entry.message)}`;
+  }
+  if (entry.type === "compaction") {
+    return `compaction:${entry.tokensBefore}:${hashJson({
+      summary: entry.summary,
+      details: entry.details,
+      fromHook: entry.fromHook,
+    })}`;
+  }
+  if (entry.type === "custom") {
+    return `custom:${entry.customType}:${hashJson(entry.data)}`;
+  }
+  if (entry.type === "custom_message") {
+    return `custom_message:${entry.customType}:${hashJson({
+      content: entry.content,
+      details: entry.details,
+      display: entry.display,
+    })}`;
+  }
+  return null;
+}
+
+function computeMessageReplayIdentity(message: AgentMessage): string {
+  return `message:${hashJson(message)}`;
+}
+
 function remapEntryId(
   entryId: string | null | undefined,
   rewrittenEntryIds: ReadonlyMap<string, string>,
@@ -227,19 +262,35 @@ export function rewriteTranscriptEntriesInSessionManager(params: {
   // re-running persistence hooks or size truncation on replayed messages.
   const appendMessage = getRawSessionAppendMessage(params.sessionManager);
   const rewrittenEntryIds = new Map<string, string>();
+  const emittedReplayIdentities = new Map<string, string>();
   for (let index = matchedIndices[0]; index < branch.length; index++) {
     const entry = branch[index];
     const replacement = entry.type === "message" ? replacementsById.get(entry.id) : undefined;
-    const newEntryId =
-      replacement === undefined
-        ? appendBranchEntry({
-            sessionManager: params.sessionManager,
-            entry,
-            rewrittenEntryIds,
-            appendMessage,
-          })
-        : appendMessage(replacement as Parameters<typeof params.sessionManager.appendMessage>[0]);
+    if (replacement === undefined) {
+      const replayIdentity = computeBranchEntryReplayIdentity(entry);
+      const existingEntryId =
+        replayIdentity === null ? undefined : emittedReplayIdentities.get(replayIdentity);
+      if (existingEntryId !== undefined) {
+        rewrittenEntryIds.set(entry.id, existingEntryId);
+        continue;
+      }
+      const newEntryId = appendBranchEntry({
+        sessionManager: params.sessionManager,
+        entry,
+        rewrittenEntryIds,
+        appendMessage,
+      });
+      rewrittenEntryIds.set(entry.id, newEntryId);
+      if (replayIdentity !== null) {
+        emittedReplayIdentities.set(replayIdentity, newEntryId);
+      }
+      continue;
+    }
+    const newEntryId = appendMessage(
+      replacement as Parameters<typeof params.sessionManager.appendMessage>[0],
+    );
     rewrittenEntryIds.set(entry.id, newEntryId);
+    emittedReplayIdentities.set(computeMessageReplayIdentity(replacement), newEntryId);
   }
 
   return {
@@ -328,19 +379,34 @@ export function rewriteTranscriptEntriesInState(params: {
 
   const appendedEntries: SessionBranchEntry[] = [];
   const rewrittenEntryIds = new Map<string, string>();
+  const emittedReplayIdentities = new Map<string, string>();
   for (let index = matchedIndices[0]; index < branch.length; index++) {
     const entry = branch[index];
     const replacement = entry.type === "message" ? replacementsById.get(entry.id) : undefined;
-    const newEntry =
-      replacement === undefined
-        ? appendTranscriptStateBranchEntry({
-            state: params.state,
-            entry,
-            rewrittenEntryIds,
-          })
-        : params.state.appendMessage(replacement);
+    if (replacement === undefined) {
+      const replayIdentity = computeBranchEntryReplayIdentity(entry);
+      const existingEntryId =
+        replayIdentity === null ? undefined : emittedReplayIdentities.get(replayIdentity);
+      if (existingEntryId !== undefined) {
+        rewrittenEntryIds.set(entry.id, existingEntryId);
+        continue;
+      }
+      const newEntry = appendTranscriptStateBranchEntry({
+        state: params.state,
+        entry,
+        rewrittenEntryIds,
+      });
+      rewrittenEntryIds.set(entry.id, newEntry.id);
+      appendedEntries.push(newEntry);
+      if (replayIdentity !== null) {
+        emittedReplayIdentities.set(replayIdentity, newEntry.id);
+      }
+      continue;
+    }
+    const newEntry = params.state.appendMessage(replacement);
     rewrittenEntryIds.set(entry.id, newEntry.id);
     appendedEntries.push(newEntry);
+    emittedReplayIdentities.set(computeMessageReplayIdentity(replacement), newEntry.id);
   }
 
   return {
