Skip to content

Commit 0dfff95

Browse files
jalehmanjalehman
committed
perf(sessions): find injected transcript duplicates with a reverse early-exit scan
1 parent 643b563 commit 0dfff95

4 files changed

Lines changed: 83 additions & 16 deletions

File tree

‎src/config/sessions/session-accessor.test.ts‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import {
1818
cleanupSessionLifecycleArtifacts,
1919
commitReplySessionInitialization,
2020
createSessionEntryWithTranscript,
21+
findTranscriptEvent,
2122
listSessionEntries,
2223
loadReplySessionInitializationSnapshot,
2324
loadSessionEntry,
@@ -135,6 +136,35 @@ describe("session accessor file-backed seam", () => {
135136
expect(missing).toEqual([]);
136137
});
137138

139+
it("finds the newest matching transcript event without loading the whole file", async () => {
140+
const header = { type: "session", id: "session-find", timestamp: 1 };
141+
const older = { type: "message", id: "m1", message: { role: "assistant", tag: "old" } };
142+
const newer = { type: "message", id: "m2", message: { role: "assistant", tag: "new" } };
143+
fs.writeFileSync(
144+
transcriptPath,
145+
`${JSON.stringify(header)}\n${JSON.stringify(older)}\n${JSON.stringify(newer)}\n`,
146+
"utf-8",
147+
);
148+
149+
const seen: unknown[] = [];
150+
const found = await findTranscriptEvent(
151+
{ sessionFile: transcriptPath, sessionId: "session-find" },
152+
(event) => {
153+
seen.push(event);
154+
return (event as { type?: string }).type === "message";
155+
},
156+
);
157+
// Newest-first with early exit: the older message is never visited.
158+
expect(found).toEqual({ event: newer });
159+
expect(seen).toEqual([newer]);
160+
161+
const missing = await findTranscriptEvent(
162+
{ sessionFile: path.join(tempDir, "missing.jsonl"), sessionId: "session-find" },
163+
() => true,
164+
);
165+
expect(missing).toBeUndefined();
166+
});
167+
138168
it("opens a borrowed read view with raw exact-key probes and deferred enumeration", async () => {
139169
const mixedKey = "agent:main:matrix:channel:!RoomAbC:example.org";
140170
await upsertSessionEntry(

‎src/config/sessions/session-accessor.ts‎

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,10 @@ import { resolveSessionTranscriptFile } from "./transcript-file-resolve.js";
9696
import { createSessionTranscriptHeader } from "./transcript-header.js";
9797
import { serializeJsonlLine, writeJsonlLines } from "./transcript-jsonl.js";
9898
import { replayRecentUserAssistantMessages } from "./transcript-replay.js";
99-
import { streamSessionTranscriptLines } from "./transcript-stream.js";
99+
import {
100+
streamSessionTranscriptLines,
101+
streamSessionTranscriptLinesReverse,
102+
} from "./transcript-stream.js";
100103
import {
101104
scanSessionTranscriptTree,
102105
selectSessionTranscriptTreePathNodes,
@@ -2089,6 +2092,30 @@ export async function appendTranscriptMessage<TMessage>(
20892092
});
20902093
}
20912094

2095+
/**
2096+
* Finds the newest transcript record accepted by the matcher. Reads newest-first
2097+
* with early exit so hot append-path lookups never materialize the whole
2098+
* transcript; missing transcripts match nothing. The match is wrapped so parsed
2099+
* falsy records stay distinguishable from "no match".
2100+
*/
2101+
export async function findTranscriptEvent(
2102+
scope: SessionTranscriptReadScope,
2103+
match: (event: TranscriptEvent) => boolean,
2104+
): Promise<{ event: TranscriptEvent } | undefined> {
2105+
const target = resolveSessionTranscriptReadTarget(scope);
2106+
for await (const line of streamSessionTranscriptLinesReverse(target.sessionFile)) {
2107+
try {
2108+
const event = JSON.parse(line) as TranscriptEvent;
2109+
if (match(event)) {
2110+
return { event };
2111+
}
2112+
} catch {
2113+
// Malformed lines are skipped, matching transcript index tolerance.
2114+
}
2115+
}
2116+
return undefined;
2117+
}
2118+
20922119
/** Reads parsed transcript records from an explicit or derived transcript target. */
20932120
export async function loadTranscriptEvents(
20942121
scope: SessionTranscriptReadScope,

‎src/gateway/server-methods/chat-transcript-inject.ts‎

Lines changed: 18 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22
// preserving agent-session parent links and transcript update notifications.
33
import type { SessionManager } from "../../agents/sessions/session-manager.js";
44
import {
5-
loadTranscriptEvents,
5+
findTranscriptEvent,
66
persistSessionTranscriptTurn,
77
type TranscriptEvent,
88
} from "../../config/sessions/session-accessor.js";
@@ -92,24 +92,28 @@ async function findInjectedAssistantMessageByIdempotencyKey(params: {
9292
}
9393
// The in-lock duplicate check resolves through the already-resolved
9494
// sessionFile when present so the lookup reads the file being appended to.
95-
const events = await loadTranscriptEvents({
96-
...(params.target.agentId ? { agentId: params.target.agentId } : {}),
97-
...(params.target.sessionFile ? { sessionFile: params.target.sessionFile } : {}),
98-
sessionId: params.target.sessionId,
99-
sessionKey: params.target.sessionKey,
100-
...(params.target.storePath ? { storePath: params.target.storePath } : {}),
101-
});
102-
const event = events.toReversed().find((candidate) => {
103-
const message = transcriptEventMessage(candidate);
104-
return message?.role === "assistant" && message.idempotencyKey === params.idempotencyKey;
105-
});
106-
const message = event ? transcriptEventMessage(event) : undefined;
95+
// findTranscriptEvent scans newest-first with early exit, keeping the hot
96+
// idempotent append path from materializing the whole transcript.
97+
const found = await findTranscriptEvent(
98+
{
99+
...(params.target.agentId ? { agentId: params.target.agentId } : {}),
100+
...(params.target.sessionFile ? { sessionFile: params.target.sessionFile } : {}),
101+
sessionId: params.target.sessionId,
102+
sessionKey: params.target.sessionKey,
103+
...(params.target.storePath ? { storePath: params.target.storePath } : {}),
104+
},
105+
(candidate) => {
106+
const message = transcriptEventMessage(candidate);
107+
return message?.role === "assistant" && message.idempotencyKey === params.idempotencyKey;
108+
},
109+
);
110+
const message = found ? transcriptEventMessage(found.event) : undefined;
107111
if (!message) {
108112
return undefined;
109113
}
110114
// Legacy shipped transcripts can carry assistant rows without top-level ids;
111115
// fall back to the idempotency key so re-issued aborts still dedupe there.
112-
const messageId = (event ? transcriptEventId(event) : undefined) ?? params.idempotencyKey;
116+
const messageId = (found ? transcriptEventId(found.event) : undefined) ?? params.idempotencyKey;
113117
return { messageId, message };
114118
}
115119

‎src/gateway/server-methods/chat.ts‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ import {
6969
SESSION_ROUTING_CHANGED_ERROR_REASON,
7070
} from "../../config/sessions/main-session.js";
7171
import {
72+
findTranscriptEvent,
7273
loadTranscriptEvents,
7374
patchSessionEntry,
7475
type SessionTranscriptWriteScope,
@@ -1865,7 +1866,12 @@ async function transcriptExists(scope: SessionTranscriptWriteScope): Promise<boo
18651866
if (!sessionId) {
18661867
return false;
18671868
}
1868-
return (await loadTranscriptEvents({ ...scope, sessionId }).catch(() => [])).length > 0;
1869+
// Existence probe: the newest-first matcher returns on the first record, so
1870+
// this reads one transcript line instead of materializing the whole file.
1871+
const found = await findTranscriptEvent({ ...scope, sessionId }, () => true).catch(
1872+
() => undefined,
1873+
);
1874+
return found !== undefined;
18691875
}
18701876

18711877
async function appendAssistantTranscriptMessage(params: {

0 commit comments

Comments
 (0)