Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
19 changes: 19 additions & 0 deletions src/lib/errors.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*/
Expand Down
2 changes: 2 additions & 0 deletions src/lib/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,8 @@ export {
AmbiguousAgentSessionError,
WorkspaceNotFoundError,
DataNotFoundError,
DuplicateMessageConflictError,
isDuplicateMessageConflictError,
isAmbiguousAgentSessionError,
isSessionNotFoundError,
isWorkspaceNotFoundError,
Expand Down
104 changes: 92 additions & 12 deletions src/lib/parser.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -97,16 +99,21 @@ export async function scanJsonlFile<TState>(

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 };
Expand Down Expand Up @@ -478,9 +485,72 @@ export async function parseSessionFile(filePath: string): Promise<ParseResult<Me
// Session Metadata Extraction
// =============================================================================

/** Hash canonical JSON incrementally without retaining serialized message bodies. */
function hashJsonValue(hash: Hash, value: unknown): void {
if (Array.isArray(value)) {
hash.update('[');
for (const item of value) {
hashJsonValue(hash, item);
hash.update(',');
}
hash.update(']');
} else if (typeof value === 'object' && value !== null) {
const record = value as Record<string, unknown>;
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<string, string>,
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<string, unknown>;
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;
Expand Down Expand Up @@ -570,9 +640,12 @@ function updateMetadataFromEntry(metadata: SessionMetadata, entry: RawSessionEnt
*/
export function extractMetadata(entries: RawSessionEntry[]): SessionMetadata {
const metadata = createEmptySessionMetadata();
const seen = new Map<string, string>();

for (const entry of entries) {
updateMetadataFromEntry(metadata, entry);
if (acceptMessageEntry(entry, seen)) {
updateMetadataFromEntry(metadata, entry);
}
}

return metadata;
Expand All @@ -586,8 +659,11 @@ export function extractMetadata(entries: RawSessionEntry[]): SessionMetadata {
export async function parseSessionMetadata(
filePath: string
): Promise<ParseResult<SessionMetadata>> {
const seen = new Map<string, string>();
return scanJsonlFile(filePath, createEmptySessionMetadata(), (entry, metadata) => {
updateMetadataFromEntry(metadata, entry);
if (acceptMessageEntry(entry, seen, filePath)) {
updateMetadataFromEntry(metadata, entry);
}
});
}

Expand Down Expand Up @@ -647,12 +723,14 @@ export function extractExplicitAgentIds(entries: RawSessionEntry[]): string[] {
export async function parseSessionSummary(
filePath: string
): Promise<ParseResult<SessionSummaryScanResult>> {
const seen = new Map<string, string>();
const initialState = {
metadata: createEmptySessionMetadata(),
explicitAgentIds: new Set<string>(),
};

const { data, warnings } = await scanJsonlFile(filePath, initialState, (entry, state) => {
if (!acceptMessageEntry(entry, seen, filePath)) return;
updateMetadataFromEntry(state.metadata, entry);
collectExplicitAgentIdsFromEntry(entry, state.explicitAgentIds);
});
Expand All @@ -674,13 +752,15 @@ export async function parseSessionSummary(
export async function parseSessionFileWithMetadata(
filePath: string
): Promise<ParseResult<SessionFileScanResult>> {
const seen = new Map<string, string>();
const initialState = {
messages: [] as Message[],
metadata: createEmptySessionMetadata(),
explicitAgentIds: new Set<string>(),
};

const { data, warnings } = await scanJsonlFile(filePath, initialState, (entry, state) => {
if (!acceptMessageEntry(entry, seen, filePath)) return;
updateMetadataFromEntry(state.metadata, entry);
collectExplicitAgentIdsFromEntry(entry, state.explicitAgentIds);

Expand Down
88 changes: 88 additions & 0 deletions tests/integration/session-deduplication.test.ts
Original file line number Diff line number Diff line change
@@ -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,
});
});
});
Loading
Loading