diff --git a/apps/ade-cli/src/services/sync/syncHostService.test.ts b/apps/ade-cli/src/services/sync/syncHostService.test.ts index 653fda717..7fb60dc0d 100644 --- a/apps/ade-cli/src/services/sync/syncHostService.test.ts +++ b/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[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((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"); diff --git a/apps/ade-cli/src/services/sync/syncHostService.ts b/apps/ade-cli/src/services/sync/syncHostService.ts index 52c58017b..0dcd9ba2a 100644 --- a/apps/ade-cli/src/services/sync/syncHostService.ts +++ b/apps/ade-cli/src/services/sync/syncHostService.ts @@ -103,6 +103,7 @@ 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, @@ -110,6 +111,7 @@ import { 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; diff --git a/apps/desktop/src/shared/chatReplayFold.test.ts b/apps/desktop/src/shared/chatReplayFold.test.ts new file mode 100644 index 000000000..16e4d080c --- /dev/null +++ b/apps/desktop/src/shared/chatReplayFold.test.ts @@ -0,0 +1,442 @@ +import { describe, expect, it } from "vitest"; +import type { AgentChatEventEnvelope } from "./types"; +import { + FOLDABLE_CHAT_EVENT_TYPES, + foldChatEventEnvelopesForReplay, + isCleanTextAppend, + isFoldableChatEventType, +} from "./chatReplayFold"; + +const SESSION = "session-1"; + +function envelope( + event: Record, + sequence: number, + timestamp = `2026-08-09T00:00:${String(sequence % 60).padStart(2, "0")}.000Z`, +): AgentChatEventEnvelope { + return { sessionId: SESSION, timestamp, sequence, event } as unknown as AgentChatEventEnvelope; +} + +function textEvent(text: string, seq: number, itemId = "msg_1", turnId = "turn_1"): AgentChatEventEnvelope { + return envelope({ type: "text", text, messageId: `mid:${itemId}`, itemId, turnId }, seq); +} + +/** Desktop `mergeStreamingText` (chatTranscriptRows.ts), verbatim. */ +function desktopMerge(existing: string, incoming: string): string { + if (!existing.length) return incoming; + if (!incoming.length) return existing; + if (incoming.startsWith(existing)) return incoming; + return `${existing}${incoming}`; +} + +/** + * The subset of iOS `mergeWorkStreamingText` that can fire before its overlap + * scan. If none of these hit, iOS reaches `existing + incoming` too. + */ +function iosMergeAgreesWithConcat(existing: string, incoming: string): boolean { + if (!existing.length || !incoming.length) return false; + if (existing === incoming) return false; + if (incoming.startsWith(existing)) return false; + if (existing.startsWith(incoming)) return false; + const te = existing.trim(); + const ti = incoming.trim(); + if (ti.length && te.endsWith(ti)) return false; + if (te.length && ti.startsWith(te)) return false; + if (existing.endsWith(incoming)) return false; + const max = Math.min(existing.length, incoming.length, 64); + for (let n = max; n > 0; n -= 1) if (existing.endsWith(incoming.slice(0, n))) return false; + return true; +} + +/** What a client derives from an unfolded run, using the desktop merge. */ +function clientFoldText(events: AgentChatEventEnvelope[]): string { + let acc = ""; + for (const e of events) { + acc = desktopMerge(acc, (e.event as unknown as { text: string }).text); + } + return acc; +} + +describe("fold scope", () => { + it("folds only the types it can prove, and nothing else", () => { + expect(Object.keys(FOLDABLE_CHAT_EVENT_TYPES).sort()).toEqual(["reasoning", "text"]); + for (const type of ["plan", "command", "file_change", "tool_call", "tool_result", "context_usage"]) { + expect(isFoldableChatEventType(type)).toBe(false); + } + }); + + it("passes an unknown event type through untouched rather than guessing", () => { + const input = [ + envelope({ type: "some_future_event", text: "a", itemId: "x", turnId: "t" }, 1), + envelope({ type: "some_future_event", text: "b", itemId: "x", turnId: "t" }, 2), + ]; + const { events, foldedAwayCount } = foldChatEventEnvelopesForReplay(input); + expect(foldedAwayCount).toBe(0); + expect(events).toEqual(input); + }); + + it("carries malformed envelopes through instead of faulting", () => { + // Real transcripts contain lines with no `event` (legacy writes and + // splice-repaired tails); this crashed the first implementation. + const input = [ + { sessionId: SESSION, timestamp: "2026-08-09T00:00:00.000Z", sequence: 1 }, + { sessionId: SESSION, timestamp: "2026-08-09T00:00:01.000Z", sequence: 2, event: null }, + textEvent("a ", 3), + textEvent("b", 4), + ] as unknown as AgentChatEventEnvelope[]; + const { events, foldedAwayCount } = foldChatEventEnvelopesForReplay(input); + expect(foldedAwayCount).toBe(1); + expect(events).toHaveLength(3); + expect((events[2]!.event as unknown as { text: string }).text).toBe("a b"); + }); + + it("leaves every non-text type in the union unfolded", () => { + // Named explicitly: these all merge field-wise or append a payload field, + // and desktop/iOS do it differently, so keep-last would change rendering. + const input = [ + envelope({ type: "plan", steps: [{ text: "a" }], turnId: "t", itemId: "p" }, 1), + envelope({ type: "plan", steps: [], explanation: "why", turnId: "t", itemId: "p" }, 2), + envelope({ type: "command", command: "ls", output: "one", itemId: "c", turnId: "t" }, 3), + envelope({ type: "command", command: "ls", output: "two", itemId: "c", turnId: "t" }, 4), + envelope({ type: "file_change", path: "a.ts", diff: "+1", itemId: "f", turnId: "t" }, 5), + envelope({ type: "tool_call", tool: "grep", itemId: "tc", turnId: "t" }, 6), + envelope({ type: "tool_result", tool: "grep", itemId: "tc", turnId: "t", status: "completed" }, 7), + envelope({ type: "context_usage", usage: { used: 10 }, turnId: "t" }, 8), + ]; + const { events, foldedAwayCount } = foldChatEventEnvelopesForReplay(input); + expect(foldedAwayCount).toBe(0); + expect(events).toEqual(input); + }); +}); + +describe("fold equivalence: text", () => { + it("collapses a clean-append run into one event with the same text", () => { + const input = [ + textEvent("Because the comparison", 1), + textEvent(" is fundamentally visual", 2), + textEvent(", I am using the in-app", 3), + textEvent(" browser skill", 4), + ]; + const { events, foldedAwayCount } = foldChatEventEnvelopesForReplay(input); + expect(events).toHaveLength(1); + expect(foldedAwayCount).toBe(3); + expect((events[0]!.event as unknown as { text: string }).text).toBe(clientFoldText(input)); + }); + + it("carries the LAST delta's sequence and timestamp so a watermark cannot land mid-run", () => { + const input = [textEvent("a ", 10), textEvent("b ", 11), textEvent("c", 12)]; + const { events } = foldChatEventEnvelopesForReplay(input); + expect(events[0]!.sequence).toBe(12); + expect(events[0]!.timestamp).toBe(input[2]!.timestamp); + }); + + it("emits the folded run at the position of its FIRST event", () => { + const input = [ + envelope({ type: "status", turnStatus: "active", turnId: "turn_1" }, 1), + textEvent("hello ", 2), + textEvent("world", 3), + envelope({ type: "done", turnId: "turn_1", status: "completed" }, 4), + ]; + const { events } = foldChatEventEnvelopesForReplay(input); + expect(events.map((e) => (e.event as unknown as { type: string }).type)) + .toEqual(["status", "text", "done"]); + }); + + it("keeps separate messages and separate turns apart", () => { + const input = [ + textEvent("one", 1, "msg_a", "turn_1"), + textEvent("two", 2, "msg_b", "turn_1"), + textEvent("three", 3, "msg_a", "turn_2"), + ]; + const { events, foldedAwayCount } = foldChatEventEnvelopesForReplay(input); + expect(foldedAwayCount).toBe(0); + expect(events).toHaveLength(3); + }); + + it("does not fold two messages interleaved with each other", () => { + // iOS merges by item id across gaps; desktop merges only into the previous + // row and would keep four rows. They diverge, so nothing folds. + const a = [textEvent("alpha ", 1, "msg_a"), textEvent("beta", 3, "msg_a")]; + const b = [textEvent("gamma ", 2, "msg_b"), textEvent("delta", 4, "msg_b")]; + const input = [a[0]!, b[0]!, a[1]!, b[1]!]; + const { events, foldedAwayCount } = foldChatEventEnvelopesForReplay(input); + expect(foldedAwayCount).toBe(0); + expect(events).toEqual(input); + }); + + it("does not fold deltas without a stable id — clients merge those by adjacency", () => { + const input = [ + envelope({ type: "text", text: "a", turnId: "t" }, 1), + envelope({ type: "text", text: "b", turnId: "t" }, 2), + ]; + const { events, foldedAwayCount } = foldChatEventEnvelopesForReplay(input); + expect(foldedAwayCount).toBe(0); + expect(events).toEqual(input); + }); + + it("does not fold text deltas separated by a non-text event", () => { + // Desktop merges into rows[rows.length - 1] and only when that row is a + // text row, so a tool call between two deltas ends the run. Folding across + // it would move the tool call after the whole message. + const input = [ + textEvent("A", 1), + envelope({ type: "tool_call", tool: "grep", itemId: "tc", turnId: "turn_1" }, 2), + textEvent("B", 3), + ]; + const { events, foldedAwayCount } = foldChatEventEnvelopesForReplay(input); + expect(foldedAwayCount).toBe(0); + expect(events.map((e) => (e.event as unknown as { type: string }).type)) + .toEqual(["text", "tool_call", "text"]); + }); + + it("resumes folding after the interruption without merging across it", () => { + const input = [ + textEvent("A", 1), + textEvent("B", 2), + envelope({ type: "tool_call", tool: "grep", itemId: "tc", turnId: "turn_1" }, 3), + textEvent("C", 4), + textEvent("D", 5), + ]; + const { events } = foldChatEventEnvelopesForReplay(input); + expect(events.map((e) => (e.event as unknown as { text?: string }).text ?? "tool")) + .toEqual(["AB", "tool", "CD"]); + }); + + it("folds reasoning on the same rule as text", () => { + const input = [ + envelope({ type: "reasoning", text: "first ", itemId: "r1", turnId: "t" }, 1), + envelope({ type: "reasoning", text: "second", itemId: "r1", turnId: "t" }, 2), + ]; + const { events, foldedAwayCount } = foldChatEventEnvelopesForReplay(input); + expect(foldedAwayCount).toBe(1); + expect((events[0]!.event as unknown as { text: string }).text).toBe("first second"); + }); +}); + +describe("fold equivalence: the cases where clients disagree stay unfolded", () => { + it.each([ + ["full-text replay (incoming repeats existing plus more)", "hello", "hello world"], + ["exact duplicate", "hello", "hello"], + ["existing already contains incoming as prefix", "hello world", "hello"], + ["repeated tail", "the answer is", " is"], + ["boundary overlap", "abcdef", "defghi"], + ["empty incoming", "hello", ""], + ])("%s is not a clean append", (_label, existing, incoming) => { + expect(isCleanTextAppend(existing, incoming)).toBe(false); + }); + + it("stops the run instead of guessing when a delta replays the message", () => { + const input = [textEvent("hello", 1), textEvent(" world", 2), textEvent("hello world!", 3)]; + const { events, foldedAwayCount } = foldChatEventEnvelopesForReplay(input); + // First two fold; the replay-shaped third is left for the client's own merge. + expect(foldedAwayCount).toBe(1); + expect(events).toHaveLength(2); + expect((events[0]!.event as unknown as { text: string }).text).toBe("hello world"); + expect((events[1]!.event as unknown as { text: string }).text).toBe("hello world!"); + }); + + it("every pair this module folds is one where desktop and iOS provably agree", () => { + const pairs: Array<[string, string]> = [ + ["Because the comparison", " is fundamentally visual"], + ["hello", " world"], + ["a", "b"], + ["hello", "hello world"], + ["hello", "hello"], + ["the answer is", " is"], + ["abcdef", "defghi"], + ["x", ""], + ["", "y"], + ]; + for (const [existing, incoming] of pairs) { + if (!isCleanTextAppend(existing, incoming)) continue; + expect(iosMergeAgreesWithConcat(existing, incoming)).toBe(true); + expect(desktopMerge(existing, incoming)).toBe(existing + incoming); + } + }); +}); + +describe("delivery bookkeeping", () => { + it("returns every pre-fold envelope as a source so collapsed deltas are still marked sent", () => { + const input = [textEvent("a ", 1), textEvent("b ", 2), textEvent("c", 3)]; + const { events, sources } = foldChatEventEnvelopesForReplay(input); + expect(events).toHaveLength(1); + expect(sources).toEqual(input); + }); +}); + +describe("sequence identity regression (ADE transcript sequence collision class)", () => { + /** + * The documented failure: a host rehydration restarts `eventSequence` at 1, + * so two different events share a `sessionId:sequence` pair and the second is + * discarded as a replay — which is how AskUserQuestion cards silently vanished + * on iOS. A fold that reused one sequence across distinct messages, or that + * left a run's sequence pointing at its first delta, would manufacture the + * same collision. This asserts it cannot. + */ + const deliveryKey = (e: AgentChatEventEnvelope): string => + `${e.sessionId}:${e.sequence ?? -1}:${e.timestamp}:${(e.event as unknown as { type: string }).type}`; + + it("never emits two folded events sharing a delivery key", () => { + const input = [ + textEvent("one ", 1, "msg_a"), + textEvent("two", 2, "msg_a"), + textEvent("three ", 3, "msg_b"), + textEvent("four", 4, "msg_b"), + ]; + const { events } = foldChatEventEnvelopesForReplay(input); + const keys = events.map(deliveryKey); + expect(new Set(keys).size).toBe(keys.length); + }); + + it("survives a rehydration that restarts sequences at 1 without collapsing distinct messages", () => { + // Two host epochs, both numbering from 1, replayed into one snapshot. + const epochOne = [ + textEvent("before ", 1, "msg_old", "turn_old"), + textEvent("restart", 2, "msg_old", "turn_old"), + ]; + const epochTwo = [ + textEvent("after ", 1, "msg_new", "turn_new"), + textEvent("restart", 2, "msg_new", "turn_new"), + ]; + const { events } = foldChatEventEnvelopesForReplay([...epochOne, ...epochTwo]); + expect(events).toHaveLength(2); + // Same reused sequence number, but the events stay distinct and neither + // message absorbed the other's text. + expect(events[0]!.sequence).toBe(2); + expect(events[1]!.sequence).toBe(2); + expect((events[0]!.event as unknown as { text: string }).text).toBe("before restart"); + expect((events[1]!.event as unknown as { text: string }).text).toBe("after restart"); + }); + + it("a folded run's sequence is >= every sequence it absorbed", () => { + const input = [textEvent("a ", 5), textEvent("b ", 9), textEvent("c", 17)]; + const { events } = foldChatEventEnvelopesForReplay(input); + expect(events[0]!.sequence).toBe(17); + for (const source of input) { + expect(events[0]!.sequence! >= source.sequence!).toBe(true); + } + }); +}); + +describe("cumulative-delta runtimes (desktop mergeStreamingText has two semantics)", () => { + /** + * `mergeStreamingText` is five lines but two behaviors: `incoming.startsWith( + * existing)` REPLACES (a runtime re-sending the whole message so far), and + * everything else CONCATENATES. Folding a cumulative run as if it were + * incremental would duplicate the text quadratically, and the client could + * not detect it — after the fold there is one event and nothing to compare + * against. These fixtures pin that the fold refuses that shape. + * + * Empirically no runtime on this machine emits cumulative text deltas + * (316,506 delta pairs across the 40 largest transcripts, zero cumulative), + * but the renderer branch exists, so the guard is pinned rather than argued. + */ + + /** Fold, then apply the client merge — what a folded peer actually renders. */ + function renderAfterFold(input: AgentChatEventEnvelope[]): string { + return clientFoldText(foldChatEventEnvelopesForReplay(input).events); + } + + it("never folds a fully cumulative run — output is the input", () => { + const input = [ + textEvent("Hello", 1), + textEvent("Hello world", 2), + textEvent("Hello world and more", 3), + ]; + const { events, foldedAwayCount } = foldChatEventEnvelopesForReplay(input); + expect(foldedAwayCount).toBe(0); + expect(events).toEqual(input); + // The decisive assertion: no duplication, and identical to unfolded. + expect(renderAfterFold(input)).toBe("Hello world and more"); + expect(renderAfterFold(input)).toBe(clientFoldText(input)); + }); + + it("a cumulative run that would duplicate quadratically renders once", () => { + // 8 cumulative deltas of a growing reply. Concatenating them would yield + // ~8x the text; replacing yields the final message exactly once. + const full = "The quick brown fox jumps over the lazy dog"; + const input = full.split(" ").map((_, index) => + textEvent(full.split(" ").slice(0, index + 1).join(" "), index + 1)); + const rendered = renderAfterFold(input); + expect(rendered).toBe(full); + expect(rendered.length).toBe(full.length); + expect(foldChatEventEnvelopesForReplay(input).foldedAwayCount).toBe(0); + }); + + it("mixed incremental-then-cumulative renders identically folded and unfolded", () => { + const input = [ + textEvent("Because the", 1), + textEvent(" comparison", 2), + textEvent(" is visual", 3), + // The runtime switches to re-sending the whole message so far. + textEvent("Because the comparison is visual, so", 4), + textEvent("Because the comparison is visual, so I checked", 5), + ]; + expect(renderAfterFold(input)).toBe(clientFoldText(input)); + expect(renderAfterFold(input)).toBe("Because the comparison is visual, so I checked"); + }); + + it("folded and unfolded render identically across every fixture shape", () => { + // The general equivalence claim, asserted rather than argued. + const shapes: Array<[string, string[]]> = [ + ["incremental", ["Hello", " world", " again"]], + ["cumulative", ["Hel", "Hello", "Hello world"]], + ["duplicate delta", ["Hello", "Hello", " world"]], + ["repeated tail", ["the answer is", " is", " 42"]], + ["boundary overlap", ["abcdef", "defghi", "ghijkl"]], + ["single delta", ["only one"]], + ["empty interleaved", ["Hello", "", " world"]], + ["whitespace-only delta", ["Hello", " ", "world"]], + ["cumulative then incremental", ["Hi", "Hi there", " friend"]], + ]; + for (const [label, texts] of shapes) { + const input = texts.map((text, index) => textEvent(text, index + 1)); + expect(renderAfterFold(input), `${label} must render identically`) + .toBe(clientFoldText(input)); + } + }); +}); + +describe("seam composition (a message straddling the page cut)", () => { + /** + * Turn-anchored paging guarantees the loaded span CONTAINS a user_message, + * not that it STARTS on one — the window top is still a byte cut, so one + * message's deltas can straddle it. The older half then arrives as individual + * deltas through byte-paged history while the newer half arrives inside a + * FOLDED snapshot. Those two halves must still compose to the same text. + * + * They do because concatenation is associative: folding (C+D) first and + * appending it to A+B gives the same string as folding nothing. This pins + * that, since it is the one place a folded event meets an unfolded one. + */ + it("composes a folded newer half with unfolded older deltas", () => { + const older = [textEvent("Because ", 1), textEvent("the comparison ", 2)]; + const newer = [textEvent("is fundamentally ", 3), textEvent("visual", 4)]; + + const folded = foldChatEventEnvelopesForReplay(newer); + expect(folded.events).toHaveLength(1); + + // What the client renders from [older deltas..., folded newer half]. + const composed = clientFoldText([...older, ...folded.events]); + // What it renders today, with nothing folded anywhere. + const unfolded = clientFoldText([...older, ...newer]); + + expect(composed).toBe(unfolded); + expect(composed).toBe("Because the comparison is fundamentally visual"); + }); + + it("composes when the seam falls at every possible delta boundary", () => { + const parts = ["Because ", "the ", "comparison ", "is ", "visual"]; + const all = parts.map((text, index) => textEvent(text, index + 1)); + const expected = parts.join(""); + for (let cut = 0; cut <= all.length; cut += 1) { + const older = all.slice(0, cut); + const newer = all.slice(cut); + const composed = clientFoldText([ + ...older, + ...foldChatEventEnvelopesForReplay(newer).events, + ]); + expect(composed, `seam after ${cut} delta(s)`).toBe(expected); + } + }); +}); diff --git a/apps/desktop/src/shared/chatReplayFold.ts b/apps/desktop/src/shared/chatReplayFold.ts new file mode 100644 index 000000000..27876ffed --- /dev/null +++ b/apps/desktop/src/shared/chatReplayFold.ts @@ -0,0 +1,230 @@ +import type { AgentChatEventEnvelope } from "./types"; + +/** + * Collapse superseded streaming rows at the replay boundary. + * + * A 30-second reply is persisted as hundreds of individual `text` delta rows, + * and replay re-sends every one of them for the client to re-fold. On a real + * 87.85 MiB thread from this machine, 68,081 `text` events averaged 798 bytes + * each while carrying 10.1 bytes of text — a 1.27% payload. One measured event + * spent 833 bytes to deliver the word " and". The rest is the identifier stack + * (a ~130-char composite `messageId`, a ~160-char `provenance.messageId` + * repeating it plus a UUID, then threadId/turnId/itemId/sessionId/timestamp/ + * sequence), re-sent in full per delta. + * + * ## Why this folds so little of the event union + * + * The obvious approach — fold every type with a `logicalItemId` — is wrong + * here, because the two clients do not fold text the same way. + * + * Desktop merges a text delta inline (chatTranscriptRows.ts, the `text` branch + * of the row reducer): it concatenates UNCONDITIONALLY, and only into + * `rows[rows.length - 1]`, and only when that row is itself a text row passing + * `shouldMergeTextRows`. iOS merges through `mergeWorkStreamingText` + * (WorkErrorAndMessageHelpers.swift) — ~50 lines of replay-shape detection, + * trimmed prefix/suffix checks and an overlap scan — and finds its target by + * searching the message list for the item id, so it merges ACROSS intervening + * events. Two consequences, and this module obeys both: + * + * 1. Only provably-clean appends fold. For deltas that overlap or repeat, iOS + * collapses and desktop concatenates, so they already render differently + * today; folding those would silently pick a winner. + * 2. Only ADJACENT deltas fold. Desktop ends a text row at the first non-text + * event, so folding across a tool call would move that tool call after the + * whole message instead of leaving it between the two halves. Folding only + * adjacent runs is exactly what desktop produces and no more than iOS + * already does. + * + * Types other than text/reasoning are deliberately NOT folded yet, for the same + * reason rather than for lack of value: + * - `command` merges its `output` and `file_change` merges its `diff` through + * `mergeStreamingText` (chatTranscriptRows.ts:634, called at :959 and :984 + * — its ONLY two call sites; it is not on the text path). That function + * has two semantics: `incoming.startsWith(existing)` REPLACES, everything + * else CONCATENATES. A runtime emitting a growing diff or growing command + * output per event is exactly the cumulative shape that branch exists to + * collapse, so anyone extending the fold to those types must handle it — + * the fixtures in this module's test file are all text-shaped and would + * not catch a cumulative diff. + * - `plan` is a field-wise merge with fallbacks + * (`mergePlanTranscriptEvent`, chatTranscriptRows.ts:488): an event with + * empty `steps` preserves the previous steps, so keep-last loses them. + * Each is foldable under its own provable-agreement predicate; that is follow-up + * work, not a guess to make here. + * + * Every other event type passes through untouched — an unrecognized type is + * never folded on a guess. + * + * ## Invariant: a folded run must never overlap a byte-paged range + * + * Event identity is content-derived — `agentChatEventIdentityKey` + * (shared/chatHistoryMerge.ts) is `timestamp#type#JSON.stringify(event)` — + * so a folded run and the individual deltas it replaces have completely + * different identities and will NOT dedupe against each other. If a reader ever + * received both, it would render the same text twice with no path to detect it. + * + * That is safe today only because the two spans are disjoint: the + * `chat_subscribe` snapshot covers `[tailStartOffset, EOF)` and byte-paged + * history covers strictly below `tailStartOffset`. This module is therefore + * called from exactly one place — the snapshot path in `syncHostService` — and + * must stay that way. Never fold anything below `tailStartOffset`, and never + * let a folded snapshot overlap a byte-paged range. (Contract agreed with the + * lane that owns page-cut placement and renderer paging.) + */ + +/** Event types this module knows how to fold, and how. */ +export const FOLDABLE_CHAT_EVENT_TYPES = { + text: "clean-append", + reasoning: "clean-append", +} as const; + +export type FoldableChatEventType = keyof typeof FOLDABLE_CHAT_EVENT_TYPES; + +export function isFoldableChatEventType(type: string): type is FoldableChatEventType { + return Object.prototype.hasOwnProperty.call(FOLDABLE_CHAT_EVENT_TYPES, type); +} + +/** + * True when concatenation is what every client's merge produces for this pair. + * + * Desktop concatenates text deltas unconditionally, so for text this predicate + * is stricter than desktop needs — it simply folds less. It exists for iOS, + * whose `mergeWorkStreamingText` special-cases an empty side, equality, a + * prefix relationship in either direction, and repeated or overlapping tails. + * Rejecting all of those leaves only disjoint appends, where both clients + * return `existing + incoming`. + */ +export function isCleanTextAppend(existing: string, incoming: string): boolean { + if (!existing.length || !incoming.length) return false; + if (existing === incoming) return false; + if (incoming.startsWith(existing) || existing.startsWith(incoming)) return false; + const trimmedExisting = existing.trim(); + const trimmedIncoming = incoming.trim(); + if (trimmedIncoming.length > 0 && trimmedExisting.endsWith(trimmedIncoming)) return false; + if (trimmedExisting.length > 0 && trimmedIncoming.startsWith(trimmedExisting)) return false; + if (existing.endsWith(incoming)) return false; + // A shared boundary substring is where iOS's overlap scan and desktop's plain + // concatenation diverge, so any overlap between the tail of one and the head + // of the other disqualifies the pair. + return !hasBoundaryOverlap(existing, incoming); +} + +/** Longest suffix of `existing` that is also a prefix of `incoming`, bounded. */ +function hasBoundaryOverlap(existing: string, incoming: string): boolean { + const max = Math.min(existing.length, incoming.length, 64); + for (let length = max; length > 0; length -= 1) { + if (existing.endsWith(incoming.slice(0, length))) return true; + } + return false; +} + +function readText(event: Record): string | null { + return typeof event.text === "string" ? event.text : null; +} + +/** + * Grouping key. Must match what the clients treat as one message: iOS uses + * `workAssistantMessageStableId` (messageId, else itemId) and desktop uses + * `logicalItemId ?? itemId` inside a turn. Requiring all of them to agree keeps + * a fold from spanning what either client would render as two messages. + */ +function foldGroupKey(envelope: AgentChatEventEnvelope): string | null { + const event = envelope.event as unknown as Record; + const type = typeof event.type === "string" ? event.type : ""; + const messageId = typeof event.messageId === "string" ? event.messageId.trim() : ""; + const itemId = typeof event.itemId === "string" ? event.itemId.trim() : ""; + const logicalItemId = typeof event.logicalItemId === "string" ? event.logicalItemId.trim() : ""; + const turnId = typeof event.turnId === "string" ? event.turnId.trim() : ""; + const stable = messageId || logicalItemId || itemId; + // Without a stable id the clients merge on adjacency instead, which depends + // on surrounding rows and is not reproducible from the event alone. + if (!stable) return null; + // NUL-separated: ids may legitimately contain spaces or colons, so a + // printable separator could let two distinct groups collide on one key. + // Written as an escape so this file stays reviewable text, not binary. + return [type, turnId, stable, logicalItemId || itemId].join("\u0000"); +} + +export type FoldedChatReplay = { + /** Events to put on the wire, in original render order. */ + events: AgentChatEventEnvelope[]; + /** + * Every source envelope that went into `events`, including the ones folded + * away. Delivery bookkeeping must mark all of these as sent, or the + * transcript pump re-sends the collapsed deltas individually and the client + * renders them twice. + */ + sources: AgentChatEventEnvelope[]; + foldedAwayCount: number; +}; + +/** + * Fold a replay snapshot. Order is preserved: a folded run is emitted at the + * position of its FIRST event, because that is where both clients place the + * message, and carries the LAST event's `sequence` and `timestamp` so a + * consumer that watermarks on the snapshot cannot land inside a collapsed run. + */ +export function foldChatEventEnvelopesForReplay( + envelopes: readonly AgentChatEventEnvelope[], +): FoldedChatReplay { + const events: AgentChatEventEnvelope[] = []; + let foldedAwayCount = 0; + // At most ONE run is open, and only while the deltas are adjacent in the + // stream. Desktop merges a text delta into `rows[rows.length - 1]` and only + // when that row is itself a text row, so a tool call landing between two + // deltas of one message ends the run there and starts a second row. Folding + // across the gap would move the tool call after the whole message. iOS + // merges by item id across gaps, so folding only adjacent runs is correct + // for both: it is what desktop produces, and no more than iOS already does. + let openRun: { index: number; key: string; text: string } | null = null; + + for (const envelope of envelopes) { + const event = envelope?.event as unknown as Record | undefined; + // Real transcripts contain lines with no `event` at all (legacy writes and + // splice-repaired tails). Replay must carry them through untouched rather + // than fault on them. + if (!event || typeof event !== "object") { + openRun = null; + events.push(envelope); + continue; + } + const type = typeof event.type === "string" ? event.type : ""; + if (!isFoldableChatEventType(type)) { + openRun = null; + events.push(envelope); + continue; + } + const incoming = readText(event); + const key = foldGroupKey(envelope); + if (incoming == null || key == null) { + openRun = null; + events.push(envelope); + continue; + } + // A different message, or a merge the clients would not all agree on, + // starts a fresh run instead of extending this one. + if ( + openRun == null + || openRun.key !== key + || !isCleanTextAppend(openRun.text, incoming) + ) { + openRun = { index: events.length, key, text: incoming }; + events.push(envelope); + continue; + } + const merged = openRun.text + incoming; + const previous = events[openRun.index]!; + events[openRun.index] = { + ...previous, + // The last delta's identity: a snapshot consumer that tracks progress on + // sequence must not stop inside the run. + sequence: envelope.sequence ?? previous.sequence, + timestamp: envelope.timestamp ?? previous.timestamp, + event: { ...(previous.event as object), text: merged } as AgentChatEventEnvelope["event"], + }; + openRun.text = merged; + foldedAwayCount += 1; + } + + return { events, sources: [...envelopes], foldedAwayCount }; +} diff --git a/apps/desktop/src/shared/types/sync.ts b/apps/desktop/src/shared/types/sync.ts index 1f0958ede..872dcb7a0 100644 --- a/apps/desktop/src/shared/types/sync.ts +++ b/apps/desktop/src/shared/types/sync.ts @@ -132,6 +132,12 @@ export const SYNC_CHUNKED_ENVELOPES_CAPABILITY = "chunkedEnvelopes"; * Peers that do not declare it keep the base64 wire byte for byte. */ export const SYNC_BINARY_ENVELOPES_CAPABILITY = "binaryEnvelopes"; +/** + * Hello capability a client declares when it accepts a chat replay snapshot + * whose streaming deltas have been folded into the message they belong to. + * Clients that do not declare it receive every individual delta, unchanged. + */ +export const SYNC_FOLDED_REPLAY_CAPABILITY = "foldedReplay"; export type SyncPayloadEncoding = "json" | "base64"; diff --git a/apps/ios/ADE/Services/SyncService.swift b/apps/ios/ADE/Services/SyncService.swift index 699c043af..2a89e0352 100644 --- a/apps/ios/ADE/Services/SyncService.swift +++ b/apps/ios/ADE/Services/SyncService.swift @@ -15498,7 +15498,7 @@ final class SyncService: ObservableObject { "deviceType": "phone", "siteId": database.localSiteId(), "dbVersion": latestRemoteDbVersion, - "capabilities": ["changesetAck", "chunkedEnvelopes", "relayReauthorizeV1", "binaryEnvelopes"], + "capabilities": ["changesetAck", "chunkedEnvelopes", "relayReauthorizeV1", "binaryEnvelopes", "foldedReplay"], ] if let appVersion = (info["CFBundleShortVersionString"] as? String)? .trimmingCharacters(in: .whitespacesAndNewlines), diff --git a/apps/ios/ADETests/SyncEnvelopeChunkAssemblerTests.swift b/apps/ios/ADETests/SyncEnvelopeChunkAssemblerTests.swift index e5f77918b..936a8dcee 100644 --- a/apps/ios/ADETests/SyncEnvelopeChunkAssemblerTests.swift +++ b/apps/ios/ADETests/SyncEnvelopeChunkAssemblerTests.swift @@ -213,4 +213,62 @@ final class SyncEnvelopeChunkAssemblerTests: XCTestCase { assembler.reset() XCTAssertNil(assembler.add(chunkId: "e", index: 1, total: 2, part: base64("2"))) } + // MARK: - Binary envelope container + + private func binaryFrame(header: [String: Any], body: Data) -> Data { + let headerJSON = try! JSONSerialization.data(withJSONObject: header) + var frame = Data("ADE1".utf8) + var length = UInt32(headerJSON.count).bigEndian + withUnsafeBytes(of: &length) { frame.append(contentsOf: $0) } + frame.append(headerJSON) + frame.append(body) + return frame + } + + func testDecodesBinaryContainerHeaderAndBody() { + let frame = binaryFrame(header: ["version": 1, "type": "chat_event"], body: Data("raw-bytes".utf8)) + XCTAssertTrue(SyncBinaryFrame.isBinaryFrame(frame)) + let decoded = SyncBinaryFrame.decode(frame) + XCTAssertEqual(decoded?.header["type"] as? String, "chat_event") + XCTAssertEqual(decoded?.body, Data("raw-bytes".utf8)) + } + + func testTextDeliveredAsDataIsNotTreatedAsBinary() { + // Transports do deliver text frames as data; only the magic marks a + // binary envelope, so a JSON frame must stay on the text path. + let json = Data("{\"version\":1,\"type\":\"chat_event\"}".utf8) + XCTAssertFalse(SyncBinaryFrame.isBinaryFrame(json)) + XCTAssertNil(SyncBinaryFrame.decode(json)) + } + + func testRejectsTruncatedAndOversizedHeaderLengths() { + let frame = binaryFrame(header: ["version": 1, "type": "chat_event"], body: Data("body".utf8)) + XCTAssertNil(SyncBinaryFrame.decode(frame.prefix(6))) + + var lying = frame + var huge = UInt32(SyncBinaryFrame.maxHeaderBytes + 1).bigEndian + withUnsafeBytes(of: &huge) { bytes in + for (offset, byte) in bytes.enumerated() { lying[lying.startIndex + 4 + offset] = byte } + } + XCTAssertNil(SyncBinaryFrame.decode(lying)) + } + + func testDecodesFromASlicedDataBuffer() { + // URLSession hands back Data slices with a non-zero startIndex; absolute + // indexing into one is how a container decoder silently reads garbage. + let frame = binaryFrame(header: ["version": 1, "type": "chat_event"], body: Data("sliced".utf8)) + let padded = Data([0xEE, 0xEE]) + frame + let slice = padded.dropFirst(2) + XCTAssertTrue(SyncBinaryFrame.isBinaryFrame(slice)) + XCTAssertEqual(SyncBinaryFrame.decode(slice)?.body, Data("sliced".utf8)) + } + + func testBinaryChunkPartsReassembleIntoData() { + var assembler = SyncEnvelopeChunkAssembler() + XCTAssertNil(assembler.addBinary(chunkId: "bin", index: 0, total: 2, part: Data([0x01, 0x02]))) + XCTAssertEqual( + assembler.addBinary(chunkId: "bin", index: 1, total: 2, part: Data([0x03])), + Data([0x01, 0x02, 0x03]) + ) + } } diff --git a/docs/features/sync-and-multi-device/README.md b/docs/features/sync-and-multi-device/README.md index 748031753..d8f8a7176 100644 --- a/docs/features/sync-and-multi-device/README.md +++ b/docs/features/sync-and-multi-device/README.md @@ -416,11 +416,18 @@ running sync authority). │ project_catalog/project_switch/project actions, │ │ command / command_ack / command_result, │ │ envelope_chunk │ -│ - negotiated deflate+base64 above 512 B; no offer keeps the │ -│ exact legacy encoder (gzip at 4 KB, or web JSON) │ +│ - negotiated deflate above 512 B; no offer keeps the exact │ +│ legacy encoder (gzip at 4 KB, or web JSON) │ +│ - compressed payloads ride a binary frame ("ADE1" magic, u32 │ +│ header length, header JSON, raw bytes) for peers declaring │ +│ "binaryEnvelopes"; everyone else keeps base64-in-JSON │ +│ - permessage-deflate on both listeners; when a peer negotiates │ +│ it the application codec is skipped (stacking measures worse │ +│ than either layer alone) │ │ - decode capped at 25 MB; reassembly capped and expires at 30 s│ │ - encoded envelopes >720 KB sliced bidirectionally after the │ -│ host confirms the peer's "chunkedEnvelopes" capability │ +│ host confirms the peer's "chunkedEnvelopes" capability; │ +│ binary peers slice into binary chunks (no base64 re-tax) │ └──────────────────────────────────────────────────────────────────┘ │ ▼ @@ -1073,6 +1080,18 @@ Canonical files (`apps/ade-cli/src/services/sync/`): `tool_result.structured` and `tool_result.toolResultMeta` from the wire entirely because no client decodes them — see [chat → Persisted transcript](../chat/transcript-and-turns.md#persisted-transcript)), + replay delta folding (`foldChatEventEnvelopesForReplay` in + `apps/desktop/src/shared/chatReplayFold.ts`, applied to the full + `chat_subscribe` snapshot for peers declaring `foldedReplay`: consecutive + streaming `text`/`reasoning` deltas of one message collapse into a single + event. Only provably-clean appends fold — desktop `mergeStreamingText` and + iOS `mergeWorkStreamingText` already disagree on overlapping or repeated + deltas, so folding those would pick a winner and change what one client + renders. A folded run sits at its first delta's position and carries the last + delta's sequence, and delivery bookkeeping marks the pre-fold envelopes so the + transcript pump cannot re-send a collapsed delta. The replay-buffer resume + path is deliberately not folded: its per-event `seq` monotonicity drives the + client's `seq <= lastSeq` drop rule), the mobile changeset diet (`MOBILE_CHANGESET_EXCLUDED_TABLES`: tables the phone never reads from a changeset — `attempt_transcripts`, `operations`, @@ -1275,6 +1294,9 @@ Canonical files (`apps/ade-cli/src/services/sync/`): rehearsal, controller-to-authority swap). On iOS, an equivalent Swift implementation lives in `apps/ios/ADE/Services/SyncService.swift`. - `syncProtocol.ts` — canonical Node envelope codec and protocol boundary. +- `syncBinaryFrame.ts` — binary envelope container ("ADE1" magic, u32 header + length, header JSON, raw compressed body) plus the magic sniff that keeps a + text frame delivered as binary data on the text path. It retains the legacy gzip threshold (`DEFAULT_SYNC_COMPRESSION_THRESHOLD_BYTES = 4 * 1024`) for peers that omit compression negotiation, and supports negotiated zlib-wrapped diff --git a/docs/features/sync-and-multi-device/ios-companion.md b/docs/features/sync-and-multi-device/ios-companion.md index 1a6c207ca..6a4acf914 100644 --- a/docs/features/sync-and-multi-device/ios-companion.md +++ b/docs/features/sync-and-multi-device/ios-companion.md @@ -884,9 +884,17 @@ Sources: `apps/ios/ADE/Services/SyncService.swift` and converge there. Overlapping wake-ups are coalesced, delayed path tasks carry a connection-generation guard, and an automatic reconnect never tears down an already-live connection. The socket declares the - `chunkedEnvelopes` capability, offers zlib-wrapped `deflate` in - `hello.compression`, and sets a 32 MiB `maximumMessageSize` receive - budget. The phone does not use either new wire behavior until the host + `chunkedEnvelopes`, `binaryEnvelopes`, and `foldedReplay` capabilities, + offers zlib-wrapped `deflate` in `hello.compression`, and sets a 32 MiB + `maximumMessageSize` receive budget. `binaryEnvelopes` lets the host send a + compressed payload as a binary frame instead of base64 inside JSON, which is + how the phone avoids the +33% the base64 tax costs on exactly the bytes + compression just removed; `SyncBinaryFrame` in `SyncService.swift` decodes + it, and a data frame without the `ADE1` magic still takes the text path. + iOS is the only peer that needs this, because `URLSessionWebSocketTask` + cannot negotiate permessage-deflate the way browser and Node peers do. + `foldedReplay` lets a `chat_subscribe` snapshot arrive with each message's + streaming deltas already collapsed into one event. The phone does not use either new wire behavior until the host confirms it in `hello_ok`, so an older host stays on the exact legacy path. Relay candidates first append `ready=2`. A new Worker sends `accepted/v2` before bridge setup and `ready/v2` only after both pipe and validated local