From 716ba2b8e42cbf469d2fa050dd4a3e429db39ddb Mon Sep 17 00:00:00 2001 From: borui Date: Wed, 23 Sep 2026 08:03:31 +0000 Subject: [PATCH] fix(parser): deduplicate repeated transcript UUIDs (#18) --- README.md | 18 ++ src/lib/errors.ts | 19 ++ src/lib/index.ts | 2 + src/lib/parser.ts | 104 +++++++- .../integration/session-deduplication.test.ts | 88 +++++++ tests/unit/parser-deduplication.test.ts | 237 ++++++++++++++++++ 6 files changed, 456 insertions(+), 12 deletions(-) create mode 100644 tests/integration/session-deduplication.test.ts create mode 100644 tests/unit/parser-deduplication.test.ts diff --git a/README.md b/README.md index e468b16..b66be0e 100644 --- a/README.md +++ b/README.md @@ -355,6 +355,24 @@ console.log(result); // { successCount: 2, failedCount: 0, errors: [] } ``` +### Duplicate Messages + +Session reads and summary counts deduplicate `user`, `assistant`, and `progress` +records with the same non-empty UUID within each transcript file. The first +occurrence is retained. Object key order and the top-level `origin` field are +ignored when comparing records; all other fields, including unknown payloads, +are compared. Identical text under different UUIDs remains separate messages. + +Conflicting records with the same UUID cause a `DuplicateMessageConflictError` +with `messageUuid` and `filePath` properties. This also applies to summary scans, +so callers can detect conflicts before consuming incorrect counts. The exported +`isDuplicateMessageConflictError()` guard identifies these errors. + +Records without UUIDs, summaries, and file-history snapshots remain independent +records. Raw JSONL parsing preserves every record. Summary scans retain only +UUIDs and fixed-size fingerprints for duplicate detection, rather than complete +message bodies. + ### Custom Data Path ```typescript diff --git a/src/lib/errors.ts b/src/lib/errors.ts index 5aba4aa..7156c1e 100644 --- a/src/lib/errors.ts +++ b/src/lib/errors.ts @@ -60,10 +60,29 @@ export class AmbiguousAgentSessionError extends Error { } } +/** Error raised when one transcript contains conflicting records for a message UUID. */ +export class DuplicateMessageConflictError extends Error { + readonly name = 'DuplicateMessageConflictError' as const; + + constructor( + readonly messageUuid: string, + readonly filePath?: string + ) { + super(`Conflicting duplicate message UUID ${messageUuid}${filePath ? ` in ${filePath}` : ''}`); + Object.setPrototypeOf(this, DuplicateMessageConflictError.prototype); + } +} + // ============================================================================= // Type Guards // ============================================================================= +export function isDuplicateMessageConflictError( + error: unknown +): error is DuplicateMessageConflictError { + return error instanceof DuplicateMessageConflictError; +} + /** * Check if error is a SessionNotFoundError. */ diff --git a/src/lib/index.ts b/src/lib/index.ts index 63c49fd..1681243 100644 --- a/src/lib/index.ts +++ b/src/lib/index.ts @@ -75,6 +75,8 @@ export { AmbiguousAgentSessionError, WorkspaceNotFoundError, DataNotFoundError, + DuplicateMessageConflictError, + isDuplicateMessageConflictError, isAmbiguousAgentSessionError, isSessionNotFoundError, isWorkspaceNotFoundError, diff --git a/src/lib/parser.ts b/src/lib/parser.ts index 3cffe81..030991d 100644 --- a/src/lib/parser.ts +++ b/src/lib/parser.ts @@ -9,6 +9,8 @@ import { createReadStream } from 'fs'; import { createInterface } from 'readline'; +import { createHash, type Hash } from 'crypto'; +import { DuplicateMessageConflictError } from './errors.js'; import type { RawSessionEntry, RawMessage, @@ -97,16 +99,21 @@ export async function scanJsonlFile( let lineNumber = 0; - for await (const line of rl) { - lineNumber++; - const result = parseJsonLine(line, lineNumber); - - if (result.entry) { - visitEntry(result.entry, initialState); - } else if (result.warning && result.warning.error !== 'Empty line') { - // Only track non-empty line warnings - warnings.push(result.warning); + try { + for await (const line of rl) { + lineNumber++; + const result = parseJsonLine(line, lineNumber); + + if (result.entry) { + visitEntry(result.entry, initialState); + } else if (result.warning && result.warning.error !== 'Empty line') { + // Only track non-empty line warnings + warnings.push(result.warning); + } } + } finally { + rl.close(); + fileStream.destroy(); } return { data: initialState, warnings }; @@ -478,9 +485,72 @@ export async function parseSessionFile(filePath: string): Promise; + hash.update('{'); + for (const key of Object.keys(record).sort()) { + if (record[key] === undefined) continue; + hash.update(JSON.stringify(key)); + hash.update(':'); + hashJsonValue(hash, record[key]); + hash.update(','); + } + hash.update('}'); + } else { + hash.update(JSON.stringify(value) ?? 'null'); + } +} + +/** + * Deduplicate transcript messages within one file, before transformation. + * Retain only UUIDs and fixed-size fingerprints, including for summary scans. + * Only top-level origin is provenance; preserve unknown fields in comparisons + * so transformation cannot hide conflicting payloads. Non-transcript records + * and messages without usable UUIDs remain independent records. + */ +function acceptMessageEntry( + entry: RawSessionEntry, + seen: Map, + filePath?: string +): boolean { + if ( + !['user', 'assistant', 'progress'].includes(entry.type) || + typeof entry.uuid !== 'string' || + entry.uuid.length === 0 + ) { + return true; + } + + const semanticEntry = { ...entry } as Record; + delete semanticEntry.origin; + const hash = createHash('sha256'); + hashJsonValue(hash, semanticEntry); + const fingerprint = hash.digest('hex'); + const previous = seen.get(entry.uuid); + + if (previous !== undefined) { + if (previous !== fingerprint) { + throw new DuplicateMessageConflictError(entry.uuid, filePath); + } + return false; + } + + seen.set(entry.uuid, fingerprint); + return true; +} + /** * Extract session metadata from raw entries (summary, version, etc.). - * Reads only the necessary fields without full message parsing. + * Validates duplicate UUIDs without retaining full message bodies. */ export interface SessionMetadata { summary: string | null; @@ -570,9 +640,12 @@ function updateMetadataFromEntry(metadata: SessionMetadata, entry: RawSessionEnt */ export function extractMetadata(entries: RawSessionEntry[]): SessionMetadata { const metadata = createEmptySessionMetadata(); + const seen = new Map(); for (const entry of entries) { - updateMetadataFromEntry(metadata, entry); + if (acceptMessageEntry(entry, seen)) { + updateMetadataFromEntry(metadata, entry); + } } return metadata; @@ -586,8 +659,11 @@ export function extractMetadata(entries: RawSessionEntry[]): SessionMetadata { export async function parseSessionMetadata( filePath: string ): Promise> { + const seen = new Map(); return scanJsonlFile(filePath, createEmptySessionMetadata(), (entry, metadata) => { - updateMetadataFromEntry(metadata, entry); + if (acceptMessageEntry(entry, seen, filePath)) { + updateMetadataFromEntry(metadata, entry); + } }); } @@ -647,12 +723,14 @@ export function extractExplicitAgentIds(entries: RawSessionEntry[]): string[] { export async function parseSessionSummary( filePath: string ): Promise> { + const seen = new Map(); const initialState = { metadata: createEmptySessionMetadata(), explicitAgentIds: new Set(), }; const { data, warnings } = await scanJsonlFile(filePath, initialState, (entry, state) => { + if (!acceptMessageEntry(entry, seen, filePath)) return; updateMetadataFromEntry(state.metadata, entry); collectExplicitAgentIdsFromEntry(entry, state.explicitAgentIds); }); @@ -674,6 +752,7 @@ export async function parseSessionSummary( export async function parseSessionFileWithMetadata( filePath: string ): Promise> { + const seen = new Map(); const initialState = { messages: [] as Message[], metadata: createEmptySessionMetadata(), @@ -681,6 +760,7 @@ export async function parseSessionFileWithMetadata( }; const { data, warnings } = await scanJsonlFile(filePath, initialState, (entry, state) => { + if (!acceptMessageEntry(entry, seen, filePath)) return; updateMetadataFromEntry(state.metadata, entry); collectExplicitAgentIdsFromEntry(entry, state.explicitAgentIds); diff --git a/tests/integration/session-deduplication.test.ts b/tests/integration/session-deduplication.test.ts new file mode 100644 index 0000000..fa436ec --- /dev/null +++ b/tests/integration/session-deduplication.test.ts @@ -0,0 +1,88 @@ +import { afterEach, describe, expect, it } from 'vitest'; +import { + computeTokenStats, + exportSessionToJson, + exportSessionToMarkdown, + getSession, + listSessions, + searchSessions, +} from '../../src/lib/index.js'; +import { + buildSimpleSessionJson, + cleanupTempClaudeData, + createTempClaudeData, + writeProjectSessionFile, +} from '../helpers/agent-linking.js'; + +const tempDirs: string[] = []; +afterEach(async () => { + await Promise.all(tempDirs.splice(0).map(cleanupTempClaudeData)); +}); + +describe('deduplicated session consumers', () => { + it('uses unique messages for listing, details, search, exports, stats, and agent links', async () => { + const { dataPath, projectsPath } = await createTempClaudeData('cch-dedup-consumers-'); + tempDirs.push(dataPath); + const sessionId = '11111111-1111-1111-1111-111111111111'; + const agentId = 'linked123'; + const timestamp = '2026-09-01T00:00:00.000Z'; + const entries = buildSimpleSessionJson(sessionId, '/test/project', 'UniqueToken', timestamp) + .split('\n') + .map((line) => JSON.parse(line)); + entries[0].timestamp = timestamp; + entries[1].toolUseResult = { agentId }; + const repeated = [ + entries[0], + entries[1], + entries[2], + entries[1], + { ...entries[2], origin: 'replay' }, + ]; + await writeProjectSessionFile( + projectsPath, + '-test-project', + `${sessionId}.jsonl`, + repeated.map((entry) => JSON.stringify(entry)).join('\n') + ); + const agentLines = buildSimpleSessionJson( + sessionId, + '/test/project', + 'Agent reply', + timestamp + ).split('\n'); + await writeProjectSessionFile( + projectsPath, + '-test-project', + `${sessionId}/subagents/agent-${agentId}.jsonl`, + [...agentLines, ...agentLines.slice(1)].join('\n') + ); + const config = { dataPath }; + const listed = await listSessions(config); + expect(listed.data).toHaveLength(2); + expect(listed.data.map((session) => session.messageCount)).toEqual([2, 2]); + const session = await getSession(sessionId, config); + expect(session.messageCount).toBe(2); + expect(session.messages.filter((message) => message.type !== 'summary')).toHaveLength(2); + expect(session.agentIds).toEqual([agentId]); + expect(await getSession(sessionId, config)).toEqual(session); + const agent = await getSession(`agent-${agentId}`, config); + expect(agent.messageCount).toBe(2); + expect(agent.messages).toHaveLength(3); + const results = await searchSessions('UniqueToken', config); + expect(results.pagination.total).toBe(2); + expect(results.data).toHaveLength(2); + const exported = JSON.parse(await exportSessionToJson(sessionId, config)); + expect(exported.messageCount).toBe(2); + expect(exported.messages).toHaveLength(3); + const markdown = await exportSessionToMarkdown(sessionId, config); + expect(markdown.match(/Prompt for UniqueToken/g)).toHaveLength(1); + expect(markdown.match(/Response for UniqueToken/g)).toHaveLength(1); + expect(computeTokenStats(session.messages)).toEqual({ + inputTokens: 1, + outputTokens: 1, + cacheCreationInputTokens: 0, + cacheReadInputTokens: 0, + totalTokens: 2, + }); + }); +}); diff --git a/tests/unit/parser-deduplication.test.ts b/tests/unit/parser-deduplication.test.ts new file mode 100644 index 0000000..7f6cece --- /dev/null +++ b/tests/unit/parser-deduplication.test.ts @@ -0,0 +1,237 @@ +import { afterEach, describe, expect, it } from 'vitest'; +import { mkdtemp, rm, writeFile } from 'fs/promises'; +import { tmpdir } from 'os'; +import { join } from 'path'; +import { + extractMetadata, + parseJsonlFile, + parseSessionFile, + parseSessionFileWithMetadata, + parseSessionMetadata, + parseSessionSummary, +} from '../../src/lib/parser.js'; +import { + DuplicateMessageConflictError, + isDuplicateMessageConflictError, +} from '../../src/lib/index.js'; +import type { RawSessionEntry } from '../../src/lib/types.js'; + +const user: RawSessionEntry = { + type: 'user', + uuid: 'user-1', + parentUuid: null, + timestamp: '2026-09-01T00:00:00.000Z', + sessionId: 'session-1', + cwd: '/tmp/example', + message: { role: 'user', content: 'Hello' }, +}; +const assistant: RawSessionEntry = { + type: 'assistant', + uuid: 'assistant-1', + parentUuid: 'user-1', + timestamp: '2026-09-01T00:00:01.000Z', + sessionId: 'session-1', + message: { + role: 'assistant', + model: 'test-model', + content: [ + { type: 'tool_use', id: 'tool-1', name: 'Read', input: { path: '/tmp/file', limit: 1 } }, + ], + usage: { + input_tokens: 10, + output_tokens: 5, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + }, +}; +const tempDirs: string[] = []; +async function writeEntries(entries: RawSessionEntry[]): Promise { + const dir = await mkdtemp(join(tmpdir(), 'cch-dedup-')); + tempDirs.push(dir); + const path = join(dir, 'session.jsonl'); + await writeFile(path, entries.map((entry) => JSON.stringify(entry)).join('\n')); + return path; +} +afterEach(async () => { + await Promise.all(tempDirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true }))); +}); + +async function expectCount(entries: RawSessionEntry[], count: number): Promise { + const path = await writeEntries(entries); + const detail = await parseSessionFileWithMetadata(path); + expect(detail.data.metadata.messageCount).toBe(count); + expect((await parseSessionMetadata(path)).data.messageCount).toBe(count); + expect((await parseSessionSummary(path)).data.metadata).toEqual(detail.data.metadata); + expect(extractMetadata(entries)).toEqual(detail.data.metadata); +} + +describe('session message deduplication', () => { + it('keeps the first occurrence in order and leaves raw parsing unchanged', async () => { + const progress = { ...user, type: 'progress', uuid: 'progress-1' }; + const entries = [user, assistant, user, progress, assistant, progress]; + const path = await writeEntries(entries); + expect((await parseJsonlFile(path)).data).toEqual(entries); + const parsed = await parseSessionFile(path); + expect(parsed.warnings).toEqual([]); + expect(parsed.data.map((message) => message.uuid)).toEqual([ + 'user-1', + 'assistant-1', + 'progress-1', + ]); + expect(await parseSessionFile(path)).toEqual(parsed); + await expectCount(entries, 3); + }); + + it('ignores only top-level origin and object key order, including nested tool input', async () => { + const repeated = JSON.parse(JSON.stringify(assistant)); + repeated.origin = { type: 'replay' }; + repeated.message.content[0].input = { limit: 1, path: '/tmp/file' }; + const entries = [assistant, repeated]; + await expectCount(entries, 1); + expect((await parseSessionFile(await writeEntries(entries))).data).toHaveLength(1); + }); + + it('preserves identical text with different UUIDs', async () => { + await expectCount([user, { ...user, uuid: 'user-2' }], 2); + }); + + it('does not collapse absent or empty UUIDs', async () => { + const entries = [ + { ...user, uuid: undefined }, + { ...user, uuid: undefined }, + { ...user, uuid: '' }, + { ...user, uuid: '' }, + ]; + await expectCount(entries, 4); + expect((await parseSessionFile(await writeEntries(entries))).data).toHaveLength(4); + }); + + it('deduplicates before generating fallback timestamps', async () => { + await expectCount( + [ + { ...user, timestamp: undefined }, + { ...user, timestamp: undefined }, + ], + 1 + ); + }); + + it('preserves summary and snapshot updates without using their IDs as transcript UUIDs', async () => { + const snapshot = { + type: 'file-history-snapshot', + uuid: user.uuid, + messageId: user.uuid, + snapshot: { messageId: user.uuid!, timestamp: user.timestamp!, trackedFileBackups: {} }, + }; + const entries = [ + user, + { type: 'summary', uuid: user.uuid, summary: 'First' }, + snapshot, + { type: 'summary', uuid: user.uuid, summary: 'Updated' }, + snapshot, + user, + ]; + const result = await parseSessionFileWithMetadata(await writeEntries(entries)); + expect(result.data.messages).toHaveLength(5); + expect(result.data.metadata.summary).toBe('Updated'); + await expectCount(entries, 1); + }); + + it('preserves explicit agent links on duplicated records', async () => { + const entry = { ...user, toolUseResult: { agentId: 'linked-agent' } }; + const path = await writeEntries([entry, entry]); + expect((await parseSessionSummary(path)).data.explicitAgentIds).toEqual(['linked-agent']); + expect((await parseSessionFileWithMetadata(path)).data.explicitAgentIds).toEqual([ + 'linked-agent', + ]); + }); + + it.each([ + ['content', { ...user, message: { role: 'user', content: 'Changed' } }], + ['parent UUID', { ...user, parentUuid: 'different-parent' }], + ['timestamp', { ...user, timestamp: '2026-09-02T00:00:00.000Z' }], + ['message type', { ...user, type: 'progress' }], + ['unknown semantic field', { ...user, customPayload: 'different' }], + ['agent link', { ...user, toolUseResult: { agentId: 'different-agent' } }], + [ + 'content discarded by transformation', + { ...user, message: { role: 'user', content: [{ type: 'image', source: 'different' }] } }, + ], + ])('rejects conflicts in %s in every semantic parsing path', async (_field, changed) => { + const entries = [user, changed as RawSessionEntry]; + const path = await writeEntries(entries); + for (const parse of [ + parseSessionFile, + parseSessionFileWithMetadata, + parseSessionMetadata, + parseSessionSummary, + ]) { + await expect(parse(path)).rejects.toMatchObject({ + name: 'DuplicateMessageConflictError', + messageUuid: user.uuid, + filePath: path, + }); + } + expect(() => extractMetadata(entries)).toThrow(DuplicateMessageConflictError); + }); + + it.each(['tool input', 'usage', 'model', 'nested origin', 'content order'])( + 'rejects assistant conflicts in %s', + async (field) => { + const changed = JSON.parse(JSON.stringify(assistant)); + if (field === 'tool input') changed.message.content[0].input.path = '/different'; + if (field === 'usage') changed.message.usage.output_tokens++; + if (field === 'model') changed.message.model = 'other-model'; + if (field === 'nested origin') changed.message.content[0].input.origin = 'semantic input'; + if (field === 'content order') + changed.message.content.unshift({ type: 'text', text: 'Additional content' }); + const path = await writeEntries([assistant, changed]); + await expect(parseSessionFile(path)).rejects.toThrow(DuplicateMessageConflictError); + await expect(parseSessionSummary(path)).rejects.toThrow(DuplicateMessageConflictError); + } + ); + + it('scopes UUID tracking to each file', async () => { + const first = await writeEntries([user, user]); + const second = await writeEntries([ + { ...user, message: { role: 'user', content: 'Another session' } }, + ]); + expect((await parseSessionFile(first)).data).toHaveLength(1); + expect((await parseSessionFile(second)).data).toHaveLength(1); + }); + + it('exports a typed conflict error and guard', () => { + const error = new DuplicateMessageConflictError('msg-1', '/tmp/session.jsonl'); + expect(error).toBeInstanceOf(Error); + expect(error.message).toContain('msg-1'); + expect(error.message).toContain('/tmp/session.jsonl'); + expect(isDuplicateMessageConflictError(error)).toBe(true); + expect(isDuplicateMessageConflictError(new Error())).toBe(false); + expect(isDuplicateMessageConflictError(null)).toBe(false); + }); + + it('detects conflicting fields even when transformation discards both payloads', async () => { + const entries = ['first-image', 'second-image'].map((source) => ({ + ...user, + message: { role: 'user', content: [{ type: 'image', source }] }, + })); + const path = await writeEntries(entries); + await expect(parseSessionFile(path)).rejects.toThrow(DuplicateMessageConflictError); + await expect(parseSessionSummary(path)).rejects.toThrow(DuplicateMessageConflictError); + }); + + it('treats array order as meaningful', async () => { + const content = [ + { type: 'text', text: 'First' }, + { type: 'text', text: 'Second' }, + ]; + const entries = [content, [...content].reverse()].map((blocks) => ({ + ...assistant, + message: { ...assistant.message!, content: blocks }, + })); + const path = await writeEntries(entries); + await expect(parseSessionFile(path)).rejects.toThrow(DuplicateMessageConflictError); + await expect(parseSessionMetadata(path)).rejects.toThrow(DuplicateMessageConflictError); + }); +});