Fold streaming text deltas at the chat replay boundary

apps/ade-cli/src/services/sync/syncHostService.test.ts

@@ -6283,6 +6283,125 @@ describe("initial hydration priority", () => {
     }
   });
 
+  it("folds replay deltas only for peers that declared the capability", async () => {
+    // The unit tests pin the fold itself; this pins the wiring — that the gate
+    // actually gates, and that a client without the capability still receives
+    // every individual delta.
+    const { projectRoot, cleanup } = createTempProjectRoot();
+    const transcriptPath = path.join(projectRoot, "transcripts", "folded-chat.chat.jsonl");
+    fs.mkdirSync(path.dirname(transcriptPath), { recursive: true });
+    const deltas = ["Because the", " comparison", " is visual"].map((text, index) => ({
+      sessionId: "folded-chat",
+      timestamp: `2026-08-09T04:50:5${index}.000Z`,
+      sequence: index + 1,
+      event: { type: "text", text, messageId: "msg_fold", itemId: "msg_fold", turnId: "turn_fold" },
+    })) as unknown as AgentChatEventEnvelope[];
+    fs.writeFileSync(transcriptPath, `${deltas.map((e) => JSON.stringify(e)).join("\n")}\n`, "utf8");
+
+    const base = createHostArgs(projectRoot, []);
+    const host = createSyncHostService({
+      ...base,
+      projectId: "project-1",
+      discoveryEnabled: false,
+      pollIntervalMs: 100,
+      db: {
+        sync: {
+          getSiteId: () => "site-folded-replay",
+          getDbVersion: () => 0,
+          exportChangesSince: () => [],
+          applyChanges: () => ({ appliedCount: 0 }),
+          discardUnpublishedChangesForTables: () => {},
+        },
+      },
+      sessionService: {
+        list: () => [],
+        get: (sessionId: string) => sessionId === "folded-chat"
+          ? { id: sessionId, transcriptPath, status: "running" }
+          : null,
+        readTranscriptTail: async () => "",
+      },
+      agentChatService: {
+        subscribeToEvents: vi.fn().mockReturnValue(() => {}),
+        getChatEventHistory: vi.fn(() => ({
+          sessionId: "folded-chat",
+          events: deltas,
+          truncated: false,
+          transcriptTruncated: false,
+          windowTruncated: false,
+          sessionFound: true,
+        })),
+        getSessionSummary: vi.fn(() => Promise.resolve({ status: "idle" })),
+      },
+    } as unknown as Parameters<typeof createSyncHostService>[0]);
+
+    const clients: WebSocket[] = [];
+    const openClient = async (
+      port: number,
+      deviceId: string,
+      capabilities: string[],
+      requestId: string,
+    ) => {
+      const ws = new WebSocket(`ws://127.0.0.1:${port}`);
+      clients.push(ws);
+      const tracked = trackClientEnvelopes(ws);
+      await new Promise<void>((resolve, reject) => {
+        ws.once("open", () => resolve());
+        ws.once("error", reject);
+      });
+      ws.send(encodeSyncEnvelope({
+        type: "hello",
+        requestId: `${deviceId}-hello`,
+        payload: {
+          peer: {
+            deviceId,
+            deviceName: deviceId,
+            platform: "macOS",
+            deviceType: "browser",
+            siteId: `${deviceId}-site`,
+            dbVersion: 0,
+            capabilities,
+          },
+          auth: { kind: "bootstrap", token: host.getBootstrapToken() },
+        },
+      }));
+      ws.send(encodeSyncEnvelope({
+        type: "chat_subscribe",
+        requestId,
+        projectId: "project-1",
+        payload: { sessionId: "folded-chat" },
+      }));
+      return tracked;
+    };
+
+    try {
+      const port = await host.waitUntilListening();
+      const folding = await openClient(port, "peer-folding", ["foldedReplay"], "folding-subscribe");
+      const legacy = await openClient(port, "peer-legacy", [], "legacy-subscribe");
+
+      const foldedSnapshot = await waitForEnvelope(folding.envelopes, "chat_subscribe", "folding-subscribe");
+      const legacySnapshot = await waitForEnvelope(legacy.envelopes, "chat_subscribe", "legacy-subscribe");
+      const foldedEvents = (foldedSnapshot.payload as { events: AgentChatEventEnvelope[] }).events;
+      const legacyEvents = (legacySnapshot.payload as { events: AgentChatEventEnvelope[] }).events;
+
+      expect(legacyEvents).toHaveLength(3);
+      expect(foldedEvents).toHaveLength(1);
+      expect((foldedEvents[0]!.event as unknown as { text: string }).text)
+        .toBe("Because the comparison is visual");
+      // The folded run carries the LAST delta's sequence, so a consumer cannot
+      // watermark inside it.
+      expect(foldedEvents[0]!.sequence).toBe(3);
+      // And it is exactly what the ungated peer derives by concatenating.
+      expect(legacyEvents.map((e) => (e.event as unknown as { text: string }).text).join(""))
+        .toBe((foldedEvents[0]!.event as unknown as { text: string }).text);
+    } finally {
+      for (const ws of clients) {
+        try { ws.close(); } catch { /* ignore */ }
+      }
+      await host.dispose();
+      cleanup();
+    }
+  });
+
   it("hydrates the selected browser chat without replaying historical CRDT rows", async () => {
     const { projectRoot, cleanup } = createTempProjectRoot();
     const transcriptPath = path.join(projectRoot, "transcripts", "selected-chat.chat.jsonl");

apps/ade-cli/src/services/sync/syncHostService.ts

@@ -103,13 +103,15 @@ import type {
 import {
   SYNC_APPLICATION_COMPRESSION_THRESHOLD_BYTES,
   SYNC_COMPACT_INVALIDATION_V1_CAPABILITY,
+  SYNC_FOLDED_REPLAY_CAPABILITY,
   SYNC_INVALIDATION_BATCH_MAX_ENVELOPE_BYTES,
   SYNC_INVALIDATION_BATCH_MAX_TABLES,
   SYNC_INVALIDATION_TABLE_MAX_BYTES,
   SYNC_INVALIDATION_ONLY_V1_CAPABILITY,
   SYNC_RELAY_REAUTHORIZE_V1_CAPABILITY,
 } from "../../../../desktop/src/shared/types";
 import { parseAgentChatTranscript } from "../../../../desktop/src/shared/chatTranscript";
+import { foldChatEventEnvelopesForReplay } from "../../../../desktop/src/shared/chatReplayFold";
 import { readTranscriptHistoryPage } from "../../../../desktop/src/main/services/chat/chatTranscriptHistoryPager";
 import type { Logger } from "../../../../desktop/src/main/services/logging/logger";
 import type { ProductAnalyticsService } from "../../../../desktop/src/main/services/analytics/productAnalyticsService";
@@ -8084,6 +8086,26 @@ export function createSyncHostService(args: SyncHostServiceArgs) {
             ?? (history?.truncated === true && tailStartOffset > 0);
         }
         events = events.map(compactChatEventEnvelopeForSync);
+        // Fold streaming deltas into the message they belong to. Snapshot-only
+        // and capability-gated: the replay-buffer resume path below stays
+        // unfolded because its per-event `seq` monotonicity is load-bearing for
+        // the client's `seq <= lastSeq` drop rule, and it carries only a small
+        // recent gap. `sourceEvents` keeps the pre-fold envelopes so delivery
+        // bookkeeping still marks every collapsed delta as sent.
+        let sourceEvents: AgentChatEventEnvelope[] = events;
+        if (peer.metadata?.capabilities?.includes(SYNC_FOLDED_REPLAY_CAPABILITY)) {
+          const folded = foldChatEventEnvelopesForReplay(events);
+          if (folded.foldedAwayCount > 0) {
+            args.logger.debug("sync_host.chat_replay_folded", {
+              sessionId,
+              beforeCount: events.length,
+              afterCount: folded.events.length,
+              foldedAwayCount: folded.foldedAwayCount,
+            });
+          }
+          sourceEvents = folded.sources;
+          events = folded.events;
+        }
         peer.chatTranscriptOffsets.set(sessionId, hydrationStartOffset);
         peer.chatTranscriptScanOffsets.delete(sessionId);
         const snapshot: SyncChatSubscribeSnapshotPayload = {
@@ -8097,7 +8119,10 @@ export function createSyncHostService(args: SyncHostServiceArgs) {
           ...(await resolveLiveStatusFields()),
         };
         sendRequired(peer, "chat_subscribe", snapshot, envelope.requestId);
-        for (const event of events) {
+        // Mark the PRE-fold envelopes: a collapsed delta still has its own
+        // delivery key, and leaving it unmarked lets the transcript pump
+        // re-send it as a separate event the client would render twice.
+        for (const event of sourceEvents) {
           markChatEventSent(peer, event);
         }
         hydrationSucceeded = true;