diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 3c321c86..e1998265 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -2382,14 +2382,26 @@ export class CodexAcpServer { } } -function mergeHistoryUpdates( +export function mergeHistoryUpdates( responseItemFallbackUpdates: UpdateSessionEvent[], threadUpdates: UpdateSessionEvent[], ): UpdateSessionEvent[] { const merged: UpdateSessionEvent[] = []; const seen = new Set(); + const fallbackIndicesByKey = new Map(); + const fallbackIndicesByContentKey = new Map(); let fallbackIndex = 0; + for (let index = 0; index < responseItemFallbackUpdates.length; index += 1) { + const update = responseItemFallbackUpdates[index]!; + indexHistoryUpdate(fallbackIndicesByKey, historyUpdateKey(update), index); + indexHistoryUpdate( + fallbackIndicesByContentKey, + historyUpdateContentKey(update), + index, + ); + } + const pushUpdate = (update: UpdateSessionEvent) => { const key = historyUpdateKey(update); if (key && seen.has(key)) { @@ -2408,13 +2420,15 @@ function mergeHistoryUpdates( return; } - const matchIndex = responseItemFallbackUpdates.findIndex((update, index) => ( - index >= fallbackIndex - && ( - (targetKey !== null && historyUpdateKey(update) === targetKey) - || (targetContentKey !== null && historyUpdateContentKey(update) === targetContentKey) - ) - )); + const keyMatchIndex = nextHistoryUpdateIndex( + fallbackIndicesByKey.get(targetKey ?? ""), + fallbackIndex, + ); + const contentMatchIndex = nextHistoryUpdateIndex( + fallbackIndicesByContentKey.get(targetContentKey ?? ""), + fallbackIndex, + ); + const matchIndex = minHistoryUpdateIndex(keyMatchIndex, contentMatchIndex); if (matchIndex === -1) { return; } @@ -2439,6 +2453,41 @@ function mergeHistoryUpdates( return merged; } +function indexHistoryUpdate( + target: Map, + key: string | null, + index: number, +): void { + if (key === null) return; + const indices = target.get(key); + if (indices) { + indices.push(index); + } else { + target.set(key, [index]); + } +} + +function nextHistoryUpdateIndex(indices: number[] | undefined, minimum: number): number { + if (!indices || indices.length === 0) return -1; + let low = 0; + let high = indices.length; + while (low < high) { + const middle = low + Math.floor((high - low) / 2); + if (indices[middle]! < minimum) { + low = middle + 1; + } else { + high = middle; + } + } + return low < indices.length ? indices[low]! : -1; +} + +function minHistoryUpdateIndex(left: number, right: number): number { + if (left === -1) return right; + if (right === -1) return left; + return Math.min(left, right); +} + function historyUpdateKey(update: UpdateSessionEvent): string | null { switch (update.sessionUpdate) { case "user_message_chunk": diff --git a/src/__tests__/HistoryUpdateMerge.test.ts b/src/__tests__/HistoryUpdateMerge.test.ts new file mode 100644 index 00000000..343176da --- /dev/null +++ b/src/__tests__/HistoryUpdateMerge.test.ts @@ -0,0 +1,49 @@ +import {describe, expect, it} from "vitest"; +import type {UpdateSessionEvent} from "../ACPSessionConnection"; +import {mergeHistoryUpdates} from "../CodexAcpServer"; + +describe("mergeHistoryUpdates", () => { + it("uses the earliest content or exact-key match", () => { + const earlierContentMatch = message("fallback-id", "same content", "fallback-content"); + const laterExactMatch = message("target-id", "same content", "fallback-exact"); + const target = message("target-id", "same content", "thread"); + + expect(mergeHistoryUpdates( + [earlierContentMatch, laterExactMatch], + [target], + )).toEqual([target]); + }); + + it("does not revisit matches before the advancing fallback cursor", () => { + const targetA = message("message-a", "alpha", "thread-a"); + const between = thought("thought-between", "keep between matches"); + const targetB = message("message-b", "beta", "thread-b"); + + expect(mergeHistoryUpdates( + [ + message("message-a", "alpha", "fallback-a"), + between, + message("message-b", "beta", "fallback-b"), + message("message-a", "alpha", "fallback-a-late"), + ], + [targetA, targetB], + )).toEqual([targetA, between, targetB]); + }); +}); + +function message(messageId: string, text: string, source: string): UpdateSessionEvent { + return { + sessionUpdate: "agent_message_chunk", + messageId, + content: {type: "text", text}, + _meta: {source}, + }; +} + +function thought(messageId: string, text: string): UpdateSessionEvent { + return { + sessionUpdate: "agent_thought_chunk", + messageId, + content: {type: "text", text}, + }; +}