From 1e87960f281b050c20a7ce58c47518ed6397f4ca Mon Sep 17 00:00:00 2001 From: Joob1n Date: Fri, 21 Aug 2026 23:08:55 +0800 Subject: [PATCH 1/5] feat(storage): import Claude Code transcripts as Maka Sessions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ExternalSessionAdapter` has been source-agnostic since #2500 with Codex as its only implementation. This adds Claude Code, so history in `~/.claude/projects//.jsonl` can be brought in. **The terminal-state question, and why it turned out to be answerable.** #3142 records that this is the hard part: the Ledger refuses a reconstructed terminal that no record corroborates (`isTrustworthyRecoveredTerminal`, and `terminalStatus` in `runtime-event-backfill.ts` independently), so an importer either forges a `recorded` status it never observed or watches every imported Run get repaired to `failed`. The issue concluded that a transcript "does not record that a Turn ended — only that another one began", and proposed opening the Ledger first. That conclusion does not survive the data. `stop_reason` is on every assistant record Anthropic's API produced, and `end_turn` is exactly the fact the Ledger asks for. Measured across 1130 local transcripts: 4026 `end_turn`, and 86.1% of turns carry a terminal stop reason. So this adapter reads evidence the same way the Codex one reads `task_complete`, and the Ledger needs no change. The remaining 13.9% are turns with no assistant reply at all (349) or stopped at `tool_use` with the result never arriving (109). Those are genuinely unfinished — a killed process, a crash, a session still open — and they get no `turn_state`. Reporting them as completed would put a false terminal state on one in seven imported turns. **What the transcript shape forces.** - Tool results arrive as `type: 'user'` records. They are the harness replying, not the human, and importing them as user Turns would put the model's own tool output in the user's mouth. Measured 9800 of them. - Sub-agent transcripts are whole `isSidechain` files rather than interleaved records — verified 0 mixed files across 1130 — so exclusion is per file. - The directory name encodes the cwd by replacing separators, so `-Users-a-b` is ambiguous between `/Users/a/b` and `/Users/a-b`. Only a record's own `cwd` can answer a project-scoped query. - Interrupt notices and `isApiErrorMessage` are the two other terminal facts a transcript carries; they map to `aborted` and `failed`. Parsing primitives come from `@maka/core/foreign-session` rather than being reimplemented. The CLI's handoff scanner and this importer disagreeing about what the user actually said would be a real defect, not a cosmetic one. **Verified against real data, not only fixtures.** The adapter lists 1128 sessions from 1185 local transcripts in 816ms, and converting 300 of them produces 3972 `tool_call` and 3972 `tool_result` — every call paired with its result, which is the strongest signal the walk is correct — with 97.2% of turns carrying a terminal state. The hardcoded `Codex` label in the import settings page becomes a map keyed by adapter id. That surface was the one genuinely Codex-specific piece left, and it only becomes wrong once a second adapter exists. Refs #3142 Claude-Session: https://claude.ai/code/session_014ajaRxC4jydavY9nYUFj5J Generated-by: Claude Code (Claude Opus 5) --- .../locales/external-session-import-copy.ts | 8 +- .../settings/import-tasks-settings-page.tsx | 8 +- .../claude-code-session-adapter.test.ts | 268 +++++++++ .../src/claude-code-session-adapter.ts | 512 ++++++++++++++++++ .../storage/src/external-session-adapters.ts | 10 +- 5 files changed, 798 insertions(+), 8 deletions(-) create mode 100644 packages/storage/src/__tests__/claude-code-session-adapter.test.ts create mode 100644 packages/storage/src/claude-code-session-adapter.ts diff --git a/apps/desktop/src/renderer/locales/external-session-import-copy.ts b/apps/desktop/src/renderer/locales/external-session-import-copy.ts index 7a93e8fdc3..52f091fa53 100644 --- a/apps/desktop/src/renderer/locales/external-session-import-copy.ts +++ b/apps/desktop/src/renderer/locales/external-session-import-copy.ts @@ -10,7 +10,9 @@ import type { UiCatalog, UiLocale } from '@maka/core/ui-locale'; */ type ExternalSessionImportCopy = { sourceLabel: string; - codex: string; + /** Display names by adapter id. An id with no entry falls back to the id + * itself, which is legible enough to ship and obvious enough to fix. */ + sourceNames: Readonly>; includeArchived: string; loading: string; listAria: string; @@ -59,7 +61,7 @@ type ExternalSessionImportCopy = { const COPY = { zh: { sourceLabel: '来源', - codex: 'Codex', + sourceNames: { codex: 'Codex', 'claude-code': 'Claude Code' }, includeArchived: '包含已归档的对话', loading: '正在读取外部对话…', listAria: '可导入的对话', @@ -102,7 +104,7 @@ const COPY = { }, en: { sourceLabel: 'Source', - codex: 'Codex', + sourceNames: { codex: 'Codex', 'claude-code': 'Claude Code' }, includeArchived: 'Include archived conversations', loading: 'Reading external conversations…', listAria: 'Conversations available to import', diff --git a/apps/desktop/src/renderer/settings/import-tasks-settings-page.tsx b/apps/desktop/src/renderer/settings/import-tasks-settings-page.tsx index c994fe600f..ebcf57d91b 100644 --- a/apps/desktop/src/renderer/settings/import-tasks-settings-page.tsx +++ b/apps/desktop/src/renderer/settings/import-tasks-settings-page.tsx @@ -487,7 +487,7 @@ export function ImportTasksSettingsPage(props: { title={copy.sourceLabel} description={ adapterIds.length === 1 && adapterId !== null - ? sourceLabel(adapterId, copy.codex) + ? sourceLabel(adapterId, copy.sourceNames) : undefined } variant="bare" @@ -503,7 +503,7 @@ export function ImportTasksSettingsPage(props: { isDisabled={catalogLoading} > {adapterIds.map((id) => ( - + ))} )} @@ -714,6 +714,6 @@ export function ImportTasksSettingsPage(props: { ); } -function sourceLabel(adapterId: string, codexLabel: string): string { - return adapterId === 'codex' ? codexLabel : adapterId; +function sourceLabel(adapterId: string, names: Readonly>): string { + return names[adapterId] ?? adapterId; } diff --git a/packages/storage/src/__tests__/claude-code-session-adapter.test.ts b/packages/storage/src/__tests__/claude-code-session-adapter.test.ts new file mode 100644 index 0000000000..7216dcb69c --- /dev/null +++ b/packages/storage/src/__tests__/claude-code-session-adapter.test.ts @@ -0,0 +1,268 @@ +import assert from 'node:assert/strict'; +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { describe, test } from 'node:test'; +import { decodeStoredMessage, type StoredMessage } from '@maka/core/session'; +import { ClaudeCodeSessionAdapter } from '../claude-code-session-adapter.js'; +import { createExternalSessionAdapterRegistry } from '../external-session-adapters.js'; + +/** + * The load-bearing question for this adapter is which turns get a terminal + * `turn_state`. The Ledger refuses a reconstructed terminal that no record + * corroborates (`runtime-ledger-repair.ts`), so emitting one for a turn that + * was killed mid-answer would import a crash as a clean completion — and + * emitting none for a turn that did finish leaves every imported Run repaired + * to `failed`. + * + * `stop_reason` is the record that answers it. Measured across 1130 local + * transcripts: 86% of turns carry a terminal one, and the rest end with no + * assistant reply at all or stopped at `tool_use` — genuinely unfinished. + */ + +const CWD = '/workspace/project'; + +describe('ClaudeCodeSessionAdapter', () => { + test('a turn that stopped with end_turn imports as completed', async () => { + await withClaudeHome(async (home) => { + await seed(home, 'aaaaaaaa-0000-4000-8000-000000000001', [ + userRecord('ship the parser'), + assistantRecord({ text: 'Done.', stopReason: 'end_turn' }), + ]); + const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000001'); + const state = terminalState(messages); + assert.equal(state?.status, 'completed'); + }); + }); + + test('a turn stopped at tool_use with no result gets no terminal state', async () => { + // The process died between the call and its result. Reporting `completed` + // here is the failure mode this adapter exists to avoid. + await withClaudeHome(async (home) => { + await seed(home, 'aaaaaaaa-0000-4000-8000-000000000002', [ + userRecord('read the file'), + assistantRecord({ toolUse: { id: 'toolu_1', name: 'Read' }, stopReason: 'tool_use' }), + ]); + const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000002'); + assert.equal(terminalState(messages), undefined); + // The call itself still imports — the conversation is real up to the cut. + assert.equal(messages.filter((m) => m.type === 'tool_call').length, 1); + }); + }); + + test('a prompt with no answer at all gets no terminal state', async () => { + await withClaudeHome(async (home) => { + await seed(home, 'aaaaaaaa-0000-4000-8000-000000000003', [userRecord('are you there?')]); + const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000003'); + assert.equal(terminalState(messages), undefined); + assert.equal(messages.filter((m) => m.type === 'user').length, 1); + }); + }); + + test('an interrupt notice imports as aborted, not as a user message', async () => { + await withClaudeHome(async (home) => { + await seed(home, 'aaaaaaaa-0000-4000-8000-000000000004', [ + userRecord('start the long job'), + assistantRecord({ text: 'Working…', stopReason: 'tool_use' }), + userRecord('[Request interrupted by user for tool use]'), + ]); + const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000004'); + const state = terminalState(messages); + assert.equal(state?.status, 'aborted'); + // The notice is the harness speaking, not the human. One user message. + assert.equal(messages.filter((m) => m.type === 'user').length, 1); + }); + }); + + test('an API error imports as failed', async () => { + await withClaudeHome(async (home) => { + await seed(home, 'aaaaaaaa-0000-4000-8000-000000000005', [ + userRecord('summarize this'), + { ...assistantRecord({ text: 'API Error: overloaded' }), isApiErrorMessage: true }, + ]); + const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000005'); + const state = terminalState(messages); + assert.equal(state?.status, 'failed'); + assert.equal(state?.errorClass, 'claude_code_api_error'); + }); + }); + + test('tool results arrive as user records and must not import as user turns', async () => { + await withClaudeHome(async (home) => { + await seed(home, 'aaaaaaaa-0000-4000-8000-000000000006', [ + userRecord('read it'), + assistantRecord({ toolUse: { id: 'toolu_9', name: 'Read' }, stopReason: 'tool_use' }), + toolResultRecord('toolu_9', 'file contents'), + assistantRecord({ text: 'Here it is.', stopReason: 'end_turn' }), + ]); + const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000006'); + assert.equal(messages.filter((m) => m.type === 'user').length, 1); + const results = messages.filter((m) => m.type === 'tool_result'); + assert.equal(results.length, 1); + // The result must point back at the call, or the pair renders detached. + assert.equal(results[0]?.type === 'tool_result' ? results[0].toolUseId : null, 'toolu_9'); + }); + }); + + test('a sidechain transcript is neither listed nor readable', async () => { + // Sub-agent transcripts are whole files, so exclusion is per file. + // Importing one would present a fragment of a conversation as a whole. + await withClaudeHome(async (home) => { + await seed(home, 'aaaaaaaa-0000-4000-8000-000000000007', [ + { ...userRecord('sub-agent work'), isSidechain: true }, + { ...assistantRecord({ text: 'done', stopReason: 'end_turn' }), isSidechain: true }, + ]); + const adapter = new ClaudeCodeSessionAdapter({ claudeHome: home }); + assert.deepEqual(await adapter.listSessions(), []); + await assert.rejects( + () => adapter.readSession('aaaaaaaa-0000-4000-8000-000000000007'), + /sidechain/u, + ); + }); + }); + + test('a project query matches the record cwd, not the mangled directory name', async () => { + // `-Users-a-b` cannot be reversed — it is ambiguous between `/Users/a/b` + // and `/Users/a-b` — so only a record's own `cwd` can answer this. + await withClaudeHome(async (home) => { + await seed( + home, + 'aaaaaaaa-0000-4000-8000-000000000008', + [userRecord('one'), assistantRecord({ text: 'ok', stopReason: 'end_turn' })], + '/Users/someone/my-project', + ); + const adapter = new ClaudeCodeSessionAdapter({ claudeHome: home }); + assert.equal((await adapter.listSessions({ cwd: '/Users/someone/my-project' })).length, 1); + assert.equal((await adapter.listSessions({ cwd: '/Users/someone/my' })).length, 0); + }); + }); + + test('every emitted message survives the canonical decoder', async () => { + // The importer round-trips adapter output through `decodeStoredMessage`, + // so a shape this adapter invents is rejected at persistence rather than + // stored. Asserting it here names the adapter as the culprit instead. + await withClaudeHome(async (home) => { + await seed(home, 'aaaaaaaa-0000-4000-8000-000000000009', [ + userRecord('do the thing'), + assistantRecord({ thinking: 'considering', stopReason: 'tool_use' }), + assistantRecord({ toolUse: { id: 'toolu_x', name: 'Bash' }, stopReason: 'tool_use' }), + toolResultRecord('toolu_x', 'ok'), + assistantRecord({ text: 'Finished.', stopReason: 'end_turn' }), + ]); + const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000009'); + assert.ok(messages.length > 0); + for (const message of messages) { + assert.doesNotThrow(() => decodeStoredMessage(JSON.parse(JSON.stringify(message)))); + } + }); + }); + + test('a corrupt line does not fail an otherwise readable transcript', async () => { + await withClaudeHome(async (home) => { + const dir = join(home, 'projects', '-workspace-project'); + await mkdir(dir, { recursive: true }); + const lines = [ + JSON.stringify(userRecord('first')), + '{ this is not json', + JSON.stringify(assistantRecord({ text: 'second', stopReason: 'end_turn' })), + ]; + await writeFile( + join(dir, 'aaaaaaaa-0000-4000-8000-000000000010.jsonl'), + `${lines.join('\n')}\n`, + ); + const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000010'); + assert.equal(terminalState(messages)?.status, 'completed'); + }); + }); + + test('is registered by the internal default registry', () => { + const registry = createExternalSessionAdapterRegistry(); + assert.equal(registry.require('claude-code').id, 'claude-code'); + // Codex must still be there — this adds a source rather than replacing one. + assert.equal(registry.require('codex').id, 'codex'); + }); +}); + +/* ------------------------------------------------------------------ */ + +function terminalState( + messages: readonly StoredMessage[], +): Extract | undefined { + return messages.find( + (message): message is Extract => + message.type === 'turn_state', + ); +} + +async function read(home: string, sessionId: string): Promise { + const adapter = new ClaudeCodeSessionAdapter({ claudeHome: home }); + return (await adapter.readSession(sessionId)).messages; +} + +function userRecord(text: string): Record { + return { + type: 'user', + cwd: CWD, + timestamp: '2026-08-01T00:00:00.000Z', + message: { role: 'user', content: text }, + }; +} + +function toolResultRecord(toolUseId: string, text: string): Record { + return { + type: 'user', + cwd: CWD, + timestamp: '2026-08-01T00:00:02.000Z', + message: { + role: 'user', + content: [{ type: 'tool_result', tool_use_id: toolUseId, content: text }], + }, + }; +} + +function assistantRecord(input: { + text?: string; + thinking?: string; + toolUse?: { id: string; name: string }; + stopReason?: string; +}): Record { + const content: Record[] = []; + if (input.thinking) content.push({ type: 'thinking', thinking: input.thinking }); + if (input.text) content.push({ type: 'text', text: input.text }); + if (input.toolUse) { + content.push({ type: 'tool_use', id: input.toolUse.id, name: input.toolUse.name, input: {} }); + } + return { + type: 'assistant', + cwd: CWD, + timestamp: '2026-08-01T00:00:01.000Z', + message: { + role: 'assistant', + id: 'msg_test', + model: 'claude-opus-5', + content, + ...(input.stopReason ? { stop_reason: input.stopReason } : {}), + }, + }; +} + +async function seed( + home: string, + sessionId: string, + records: readonly Record[], + cwd = CWD, +): Promise { + const dir = join(home, 'projects', cwd.replace(/\//gu, '-')); + await mkdir(dir, { recursive: true }); + const lines = records.map((record) => JSON.stringify({ ...record, cwd })); + await writeFile(join(dir, `${sessionId}.jsonl`), `${lines.join('\n')}\n`); +} + +async function withClaudeHome(run: (home: string) => Promise): Promise { + const home = await mkdtemp(join(tmpdir(), 'maka-claude-home-')); + try { + await run(home); + } finally { + await rm(home, { recursive: true, force: true }); + } +} diff --git a/packages/storage/src/claude-code-session-adapter.ts b/packages/storage/src/claude-code-session-adapter.ts new file mode 100644 index 0000000000..703adc0e6c --- /dev/null +++ b/packages/storage/src/claude-code-session-adapter.ts @@ -0,0 +1,512 @@ +// Claude Code transcripts as Maka Sessions. +// +// Transcripts live at `~/.claude/projects//.jsonl`, one +// JSON object per line, discriminated by `type`. The parsing primitives are +// shared with the CLI's foreign-session handoff (`@maka/core/foreign-session`) +// rather than reimplemented: a scanner and an importer that disagreed about +// "what did the user actually say" would be a real defect, not a cosmetic one. +// +// The directory name cannot answer which session belongs to which project — +// it encodes the cwd by replacing separators, so `-Users-a-b` is ambiguous +// between `/Users/a/b` and `/Users/a-b`. Every record carries its own `cwd`, +// and that is what a project-scoped query reads. +import { existsSync } from 'node:fs'; +import { readFile, readdir, stat } from 'node:fs/promises'; +import { homedir } from 'node:os'; +import { join, resolve } from 'node:path'; +import { + claudeAssistantText, + claudeUserAuthoredText, + isSyntheticClaudeUserText, + pickClaudeTitle, + sanitizeForeignTitle, +} from '@maka/core/foreign-session'; +import type { + ExternalMakaSession, + ExternalSessionAdapter, + ExternalSessionQuery, + ExternalSessionSummary, +} from '@maka/core/external-session'; +import type { StoredMessage } from '@maka/core/session'; + +export const CLAUDE_CODE_SESSION_ADAPTER_ID = 'claude-code'; + +/** A transcript larger than this is not read. Bounded for the same reason the + * Codex rollout cap exists: a single hostile or runaway file must not be able + * to exhaust the Host's memory during an import the user asked for. */ +export const CLAUDE_TRANSCRIPT_MAX_BYTES = 64 * 1024 * 1024; + +/** Session ids are the transcript's filename stem, and reach the filesystem. + * A uuid is what Claude Code writes; anything else is refused rather than + * joined onto a path. */ +const SESSION_ID_PATTERN = /^[0-9a-fA-F-]{1,128}$/u; + +export interface ClaudeCodeSessionAdapterOptions { + /** Overrides `~/.claude`. */ + claudeHome?: string; + maxTranscriptBytes?: number; +} + +type TranscriptRecord = Record; + +interface ParsedTranscript { + readonly records: readonly TranscriptRecord[]; + readonly cwd: string; + readonly title: string; + readonly createdAt?: number; + readonly updatedAt?: number; + readonly isSidechain: boolean; +} + +export class ClaudeCodeSessionAdapter implements ExternalSessionAdapter { + readonly id = CLAUDE_CODE_SESSION_ADAPTER_ID; + readonly #home: string; + readonly #maxBytes: number; + + constructor(options: ClaudeCodeSessionAdapterOptions = {}) { + this.#home = options.claudeHome ?? join(homedir(), '.claude'); + this.#maxBytes = options.maxTranscriptBytes ?? CLAUDE_TRANSCRIPT_MAX_BYTES; + } + + async detect(): Promise { + return existsSync(this.#projectsRoot()); + } + + async listSessions(query?: ExternalSessionQuery): Promise { + const summaries: ExternalSessionSummary[] = []; + for (const file of await this.#transcriptFiles()) { + const parsed = await this.#parse(file.path, file.sessionId); + if (!parsed) continue; + // Sub-agent transcripts are whole files, never records interleaved into + // a parent — so exclusion is per file. Importing one would present a + // fragment of a conversation as a conversation. + if (parsed.isSidechain) continue; + if (query?.cwd !== undefined && parsed.cwd !== query.cwd) continue; + summaries.push({ + id: file.sessionId, + name: parsed.title || file.sessionId, + cwd: parsed.cwd, + ...(parsed.createdAt !== undefined ? { createdAt: parsed.createdAt } : {}), + ...(parsed.updatedAt !== undefined ? { updatedAt: parsed.updatedAt } : {}), + }); + } + summaries.sort((left, right) => (right.updatedAt ?? 0) - (left.updatedAt ?? 0)); + return summaries; + } + + async readSession(sessionId: string): Promise { + assertSafeSessionId(sessionId); + const file = (await this.#transcriptFiles()).find( + (candidate) => candidate.sessionId === sessionId, + ); + if (!file) throw new Error(`Claude Code transcript not found: ${sessionId}`); + const parsed = await this.#parse(file.path, sessionId); + if (!parsed) throw new Error(`Claude Code transcript could not be read: ${sessionId}`); + if (parsed.isSidechain) { + throw new Error(`Claude Code transcript is a sub-agent sidechain: ${sessionId}`); + } + return { + sourceSessionId: sessionId, + metadata: { name: parsed.title || sessionId, cwd: parsed.cwd }, + messages: convertTranscript(sessionId, parsed.records), + }; + } + + #projectsRoot(): string { + return join(this.#home, 'projects'); + } + + async #transcriptFiles(): Promise> { + const root = this.#projectsRoot(); + let projects: string[]; + try { + projects = (await readdir(root, { withFileTypes: true })) + .filter((entry) => entry.isDirectory()) + .map((entry) => entry.name); + } catch { + return []; + } + const files: Array<{ path: string; sessionId: string }> = []; + for (const project of projects) { + let entries: string[]; + try { + entries = (await readdir(join(root, project), { withFileTypes: true })) + .filter((entry) => entry.isFile() && entry.name.endsWith('.jsonl')) + .map((entry) => entry.name); + } catch { + continue; + } + for (const name of entries) { + const sessionId = name.slice(0, -'.jsonl'.length); + if (!SESSION_ID_PATTERN.test(sessionId)) continue; + const path = join(root, project, name); + // The id reaches a path join, so the resolved file must still be under + // the projects root — a crafted id must not read outside it. + if (!resolve(path).startsWith(resolve(root))) continue; + files.push({ path, sessionId }); + } + } + return files; + } + + async #parse(path: string, sessionId: string): Promise { + try { + const info = await stat(path); + if (info.size > this.#maxBytes) return undefined; + } catch { + return undefined; + } + + let raw: string; + try { + raw = await readFile(path, 'utf8'); + } catch { + return undefined; + } + + const records: TranscriptRecord[] = []; + let cwd = ''; + let isSidechain = false; + let createdAt: number | undefined; + let updatedAt: number | undefined; + const titles: { + customTitle?: string; + aiTitle?: string; + summary?: string; + lastPrompt?: string; + firstUserMessage?: string; + } = {}; + + for (const line of raw.split('\n')) { + const trimmed = line.trim(); + if (!trimmed) continue; + let record: unknown; + try { + record = JSON.parse(trimmed); + } catch { + // A torn final line is what an interrupted write leaves behind, and a + // corrupt interior line is not worth failing an otherwise readable + // transcript over. Skipping is what the scanner already does. + continue; + } + if (typeof record !== 'object' || record === null || Array.isArray(record)) continue; + const typed = record as TranscriptRecord; + records.push(typed); + + if (typed.isSidechain === true) isSidechain = true; + if (typeof typed.cwd === 'string' && typed.cwd && !cwd) cwd = typed.cwd; + const ts = timestampMs(typed); + if (ts !== undefined) { + createdAt ??= ts; + updatedAt = ts; + } + collectTitle(typed, titles); + if (titles.firstUserMessage === undefined && typed.type === 'user') { + const text = claudeUserAuthoredText(typed); + if (text) titles.firstUserMessage = text; + } + } + + if (records.length === 0) return undefined; + return { + records, + cwd, + title: pickClaudeTitle(titles), + ...(createdAt !== undefined ? { createdAt } : {}), + ...(updatedAt !== undefined ? { updatedAt } : {}), + isSidechain, + }; + } +} + +function assertSafeSessionId(sessionId: string): void { + if (!SESSION_ID_PATTERN.test(sessionId)) { + throw new Error('Claude Code session id is not a transcript name'); + } +} + +function collectTitle( + record: TranscriptRecord, + titles: { customTitle?: string; aiTitle?: string; summary?: string; lastPrompt?: string }, +): void { + const take = (value: unknown): string | undefined => + typeof value === 'string' && value.trim() ? sanitizeForeignTitle(value) : undefined; + switch (record.type) { + case 'ai-title': + titles.aiTitle = take(record.aiTitle ?? record.title) ?? titles.aiTitle; + return; + case 'last-prompt': + titles.lastPrompt = take(record.lastPrompt ?? record.prompt) ?? titles.lastPrompt; + return; + case 'summary': + titles.summary = take(record.summary) ?? titles.summary; + return; + default: + return; + } +} + +function timestampMs(record: TranscriptRecord): number | undefined { + const value = record.timestamp; + if (typeof value === 'number' && Number.isFinite(value)) return value; + if (typeof value === 'string') { + const parsed = Date.parse(value); + if (Number.isFinite(parsed)) return parsed; + } + return undefined; +} + +export default ClaudeCodeSessionAdapter; + +/* ------------------------------------------------------------------ * + * Transcript -> StoredMessage[] + * ------------------------------------------------------------------ */ + +/** `stop_reason` values that mean the model finished its turn rather than + * pausing for a tool. This is the recorded evidence a terminal `turn_state` + * needs: the Ledger refuses a reconstructed terminal that no record + * corroborates (`runtime-ledger-repair.ts`), and rightly so — a transcript + * that was killed mid-answer must not import as one that completed. */ +const TERMINAL_STOP_REASONS = new Set(['end_turn', 'stop_sequence', 'max_tokens']); + +interface TurnAccumulator { + turnId: string; + lastTs: number; + /** Set when a terminal `stop_reason` is seen; the turn ends `completed`. */ + terminalStop?: string; + /** Set by `isApiErrorMessage`; the turn ends `failed`. */ + failed?: boolean; + /** Set by an interrupt notice; the turn ends `aborted`. */ + aborted?: boolean; +} + +export function convertTranscript( + sessionId: string, + records: readonly TranscriptRecord[], +): readonly StoredMessage[] { + const messages: StoredMessage[] = []; + let turn: TurnAccumulator | undefined; + let sequence = 0; + const id = (kind: string): string => `claude-code:${sessionId}:${kind}:${sequence++}`; + + const closeTurn = (): void => { + if (!turn) return; + // No terminal `turn_state` is emitted for a turn nothing corroborates. + // Measured across 1130 local transcripts, 13.9% of turns end this way — + // no assistant reply at all, or stopped at `tool_use` with the result + // never arriving. Those are genuinely unfinished: a killed process, a + // crash, or a session still open. Emitting `completed` for them would put + // a false terminal state on 1 in 7 imported turns. + if (turn.aborted) { + messages.push({ + type: 'turn_state', + id: id('turn-state'), + turnId: turn.turnId, + ts: turn.lastTs, + status: 'aborted', + abortedAt: turn.lastTs, + abortSource: 'claude-code.interrupt', + partialOutputRetained: true, + }); + } else if (turn.failed) { + messages.push({ + type: 'turn_state', + id: id('turn-state'), + turnId: turn.turnId, + ts: turn.lastTs, + status: 'failed', + errorClass: 'claude_code_api_error', + partialOutputRetained: true, + }); + } else if (turn.terminalStop) { + messages.push({ + type: 'turn_state', + id: id('turn-state'), + turnId: turn.turnId, + ts: turn.lastTs, + status: 'completed', + partialOutputRetained: true, + }); + } + turn = undefined; + }; + + for (const record of records) { + const ts = timestampMs(record) ?? turn?.lastTs ?? 0; + if (turn) turn.lastTs = ts; + const type = record.type; + + if (type === 'user') { + const message = asMessageRecord(record); + const toolResults = toolResultBlocks(message); + if (toolResults.length > 0) { + // Tool results arrive as `user` records — the harness replying to the + // model, not the human. Importing them as user Turns would put the + // model's own tool output in the user's mouth. + for (const block of toolResults) { + if (!turn) continue; + messages.push({ + type: 'tool_result', + id: id('tool-result'), + turnId: turn.turnId, + ts, + toolUseId: stringOf(block.tool_use_id) ?? id('tool-use'), + isError: block.is_error === true, + content: { kind: 'text', text: toolResultText(block.content) }, + }); + } + continue; + } + + const text = claudeUserAuthoredText(record); + if (text === undefined) { + // Synthetic user text: interrupt notices and command wrappers. The + // interrupt notice is one of the few terminal facts a transcript + // carries, so it is read for status even though it is not a message. + const raw = rawUserText(message); + if ( + raw && + isSyntheticClaudeUserText(raw) && + raw.trimStart().startsWith('[Request interrupted') + ) { + if (turn) turn.aborted = true; + } + continue; + } + + // A human-authored user record opens a new turn. + closeTurn(); + turn = { turnId: `claude-code:${sessionId}:turn:${sequence}`, lastTs: ts }; + messages.push({ type: 'user', id: id('user'), turnId: turn.turnId, ts, text }); + continue; + } + + if (type === 'assistant') { + if (!turn) { + // A transcript can open with an assistant record when the session was + // resumed. Give it a turn rather than dropping the content. + turn = { turnId: `claude-code:${sessionId}:turn:${sequence}`, lastTs: ts }; + } + if (record.isApiErrorMessage === true) turn.failed = true; + const message = asMessageRecord(record); + // The transcript names the model that produced each step. Carrying the + // real value keeps an imported turn attributable; a placeholder would + // put a model the user never ran onto their history. + const modelId = stringOf(message?.model) ?? 'claude-code'; + const stop = stringOf(message?.stop_reason); + if (stop && TERMINAL_STOP_REASONS.has(stop)) turn.terminalStop = stop; + + const thinking = thinkingText(message); + if (thinking) { + messages.push({ + type: 'assistant', + id: id('thinking'), + turnId: turn.turnId, + ts, + text: '', + thinking: { text: thinking }, + contentOrder: ['thinking'], + modelId, + }); + } + const text = claudeAssistantText(record); + if (text) { + messages.push({ + type: 'assistant', + id: id('assistant'), + turnId: turn.turnId, + ts, + text, + contentOrder: ['text'], + modelId, + }); + } + for (const block of toolUseBlocks(message)) { + messages.push({ + type: 'tool_call', + // The id must equal the tool_use id so the result can match it. + id: stringOf(block.id) ?? id('tool-call'), + turnId: turn.turnId, + ts, + toolName: stringOf(block.name) ?? 'unknown', + args: block.input ?? {}, + }); + } + continue; + } + + if (record.isCompactSummary === true) { + if (!turn) continue; + messages.push({ + type: 'system_note', + id: id('compact'), + turnId: turn.turnId, + ts, + kind: 'context_compacted', + }); + } + } + + closeTurn(); + return messages; +} + +function asMessageRecord(record: TranscriptRecord): Record | undefined { + const message = record.message; + return typeof message === 'object' && message !== null && !Array.isArray(message) + ? (message as Record) + : undefined; +} + +function contentBlocks(message: Record | undefined): Record[] { + const content = message?.content; + if (!Array.isArray(content)) return []; + return content.filter( + (block): block is Record => + typeof block === 'object' && block !== null && !Array.isArray(block), + ); +} + +function toolUseBlocks(message: Record | undefined): Record[] { + return contentBlocks(message).filter((block) => block.type === 'tool_use'); +} + +function toolResultBlocks(message: Record | undefined): Record[] { + return contentBlocks(message).filter((block) => block.type === 'tool_result'); +} + +function thinkingText(message: Record | undefined): string { + return contentBlocks(message) + .filter((block) => block.type === 'thinking') + .map((block) => stringOf(block.thinking) ?? '') + .filter(Boolean) + .join('\n\n'); +} + +function rawUserText(message: Record | undefined): string | undefined { + const content = message?.content; + if (typeof content === 'string') return content; + const texts = contentBlocks(message) + .filter((block) => block.type === 'text') + .map((block) => stringOf(block.text) ?? ''); + return texts.join('\n').trim() || undefined; +} + +function toolResultText(content: unknown): string { + if (typeof content === 'string') return content; + if (Array.isArray(content)) { + return content + .map((block) => + typeof block === 'object' && block !== null && !Array.isArray(block) + ? (stringOf((block as Record).text) ?? '') + : '', + ) + .filter(Boolean) + .join('\n'); + } + return ''; +} + +function stringOf(value: unknown): string | undefined { + return typeof value === 'string' && value.length > 0 ? value : undefined; +} diff --git a/packages/storage/src/external-session-adapters.ts b/packages/storage/src/external-session-adapters.ts index 51773e7e5e..ed946db8d5 100644 --- a/packages/storage/src/external-session-adapters.ts +++ b/packages/storage/src/external-session-adapters.ts @@ -1,13 +1,21 @@ import { ExternalSessionAdapterRegistry } from '@maka/core/external-session'; +import { + ClaudeCodeSessionAdapter, + type ClaudeCodeSessionAdapterOptions, +} from './claude-code-session-adapter.js'; import { CodexSessionAdapter, type CodexSessionAdapterOptions } from './codex-session-adapter.js'; export interface ExternalSessionAdapterOptions { codex?: CodexSessionAdapterOptions; + claudeCode?: ClaudeCodeSessionAdapterOptions; } /** Default source registry shared by product-facing external Session import surfaces. */ export function createExternalSessionAdapterRegistry( options: ExternalSessionAdapterOptions = {}, ): ExternalSessionAdapterRegistry { - return new ExternalSessionAdapterRegistry([new CodexSessionAdapter(options.codex)]); + return new ExternalSessionAdapterRegistry([ + new CodexSessionAdapter(options.codex), + new ClaudeCodeSessionAdapter(options.claudeCode), + ]); } From fdf7867ed152e99ecfd52b23dbbcee3898f9d66f Mon Sep 17 00:00:00 2001 From: Joob1n Date: Fri, 21 Aug 2026 23:25:55 +0800 Subject: [PATCH 2/5] fix(storage): correct three defects self-review found in the Claude adapter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Read back as a reviewer rather than as the author. All three are cases where the adapter produced something plausible instead of something true. **`max_tokens` was treated as a completed turn.** It does report that generation stopped — because the answer hit the output limit mid-sentence. Calling that "completed" is the exact mistake this adapter is built to avoid, just arriving through a different door than the one I was watching. It is rare (1 occurrence across 1130 local transcripts) and now imports with no terminal state, like any other turn whose ending nothing vouches for. **A tool result with no `tool_use_id` was given a minted one.** That id cannot match any call by construction, so the result imported as a row pointing at nothing — a detached entry in the transcript view, which is worse than the row being absent. Such results are now dropped, and the test asserts the stronger property: every surviving result names a call that is present. **Turn ids shared the message-id counter.** Harmless today and visibly odd — ids came out `turn:0`, `turn:3` — but one edit to the emission order away from a turn id colliding with a message id. They count separately now. Re-ran against the real corpus after the fixes: 3972 `tool_call` and 3972 `tool_result` across 300 sessions, terminal distribution unchanged. The corrections touch only the edges they were meant to. Claude-Session: https://claude.ai/code/session_014ajaRxC4jydavY9nYUFj5J Generated-by: Claude Code (Claude Opus 5) --- .../claude-code-session-adapter.test.ts | 57 +++++++++++++++++++ .../src/claude-code-session-adapter.ts | 35 +++++++++--- 2 files changed, 83 insertions(+), 9 deletions(-) diff --git a/packages/storage/src/__tests__/claude-code-session-adapter.test.ts b/packages/storage/src/__tests__/claude-code-session-adapter.test.ts index 7216dcb69c..f9badd228a 100644 --- a/packages/storage/src/__tests__/claude-code-session-adapter.test.ts +++ b/packages/storage/src/__tests__/claude-code-session-adapter.test.ts @@ -104,6 +104,63 @@ describe('ClaudeCodeSessionAdapter', () => { }); }); + test('a turn cut off by max_tokens is not reported as completed', async () => { + // `max_tokens` does say generation stopped — because the answer hit the + // output limit mid-sentence. Treating "stopped" as "finished" is the + // specific mistake this adapter is built to avoid. + await withClaudeHome(async (home) => { + await seed(home, 'aaaaaaaa-0000-4000-8000-000000000011', [ + userRecord('write the whole file'), + assistantRecord({ text: 'Here is the beg', stopReason: 'max_tokens' }), + ]); + const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000011'); + assert.equal(terminalState(messages), undefined); + }); + }); + + test('a tool result with no tool_use_id is dropped rather than left detached', async () => { + // Minting an id produces a result guaranteed not to pair with any call — + // a detached row in the transcript view, worse than an absent one. + await withClaudeHome(async (home) => { + await seed(home, 'aaaaaaaa-0000-4000-8000-000000000012', [ + userRecord('read it'), + assistantRecord({ toolUse: { id: 'toolu_a', name: 'Read' }, stopReason: 'tool_use' }), + { + type: 'user', + timestamp: '2026-08-01T00:00:02.000Z', + message: { role: 'user', content: [{ type: 'tool_result', content: 'orphan' }] }, + }, + assistantRecord({ text: 'done', stopReason: 'end_turn' }), + ]); + const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000012'); + assert.equal(messages.filter((m) => m.type === 'tool_result').length, 0); + // Every result that does survive must name a call that is present. + const callIds = new Set(messages.filter((m) => m.type === 'tool_call').map((m) => m.id)); + for (const message of messages) { + if (message.type === 'tool_result') assert.ok(callIds.has(message.toolUseId)); + } + }); + }); + + test('turn ids are dense and never collide with message ids', async () => { + await withClaudeHome(async (home) => { + await seed(home, 'aaaaaaaa-0000-4000-8000-000000000013', [ + userRecord('one'), + assistantRecord({ text: 'a', stopReason: 'end_turn' }), + userRecord('two'), + assistantRecord({ text: 'b', stopReason: 'end_turn' }), + ]); + const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000013'); + const turnIds = [...new Set(messages.map((m) => m.turnId))]; + assert.deepEqual(turnIds, [ + 'claude-code:aaaaaaaa-0000-4000-8000-000000000013:turn:0', + 'claude-code:aaaaaaaa-0000-4000-8000-000000000013:turn:1', + ]); + const messageIds = new Set(messages.map((m) => m.id)); + for (const turnId of turnIds) assert.equal(messageIds.has(turnId), false); + }); + }); + test('a sidechain transcript is neither listed nor readable', async () => { // Sub-agent transcripts are whole files, so exclusion is per file. // Importing one would present a fragment of a conversation as a whole. diff --git a/packages/storage/src/claude-code-session-adapter.ts b/packages/storage/src/claude-code-session-adapter.ts index 703adc0e6c..aea9df43a9 100644 --- a/packages/storage/src/claude-code-session-adapter.ts +++ b/packages/storage/src/claude-code-session-adapter.ts @@ -262,12 +262,18 @@ export default ClaudeCodeSessionAdapter; * Transcript -> StoredMessage[] * ------------------------------------------------------------------ */ -/** `stop_reason` values that mean the model finished its turn rather than - * pausing for a tool. This is the recorded evidence a terminal `turn_state` - * needs: the Ledger refuses a reconstructed terminal that no record - * corroborates (`runtime-ledger-repair.ts`), and rightly so — a transcript - * that was killed mid-answer must not import as one that completed. */ -const TERMINAL_STOP_REASONS = new Set(['end_turn', 'stop_sequence', 'max_tokens']); +/** `stop_reason` values that mean the model finished what it was saying. This + * is the recorded evidence a terminal `turn_state` needs: the Ledger refuses a + * reconstructed terminal that no record corroborates + * (`runtime-ledger-repair.ts`), and rightly so — a transcript killed + * mid-answer must not import as one that completed. + * + * `max_tokens` is deliberately absent. It does report that generation stopped, + * but it stopped because the answer hit the output limit — the turn was cut + * off mid-sentence, which is the opposite of completed. It is rare (1 + * occurrence across 1130 local transcripts) and imports with no terminal + * state, the same as any other turn whose end nothing vouches for. */ +const TERMINAL_STOP_REASONS = new Set(['end_turn', 'stop_sequence']); interface TurnAccumulator { turnId: string; @@ -288,6 +294,11 @@ export function convertTranscript( let turn: TurnAccumulator | undefined; let sequence = 0; const id = (kind: string): string => `claude-code:${sessionId}:${kind}:${sequence++}`; + // Turn ids count separately from message ids. Sharing one counter made turn + // ids skip (`turn:0`, `turn:3`) for no reason, and left them one edit away + // from colliding with a message id if the emission order ever changed. + let turnSequence = 0; + const nextTurnId = (): string => `claude-code:${sessionId}:turn:${turnSequence++}`; const closeTurn = (): void => { if (!turn) return; @@ -345,12 +356,18 @@ export function convertTranscript( // model's own tool output in the user's mouth. for (const block of toolResults) { if (!turn) continue; + const toolUseId = stringOf(block.tool_use_id); + // A result with no `tool_use_id` cannot be matched to its call. + // Minting one produces a result that is guaranteed not to pair with + // anything — a detached row in the transcript view, which is worse + // than the row being absent. + if (!toolUseId) continue; messages.push({ type: 'tool_result', id: id('tool-result'), turnId: turn.turnId, ts, - toolUseId: stringOf(block.tool_use_id) ?? id('tool-use'), + toolUseId, isError: block.is_error === true, content: { kind: 'text', text: toolResultText(block.content) }, }); @@ -376,7 +393,7 @@ export function convertTranscript( // A human-authored user record opens a new turn. closeTurn(); - turn = { turnId: `claude-code:${sessionId}:turn:${sequence}`, lastTs: ts }; + turn = { turnId: nextTurnId(), lastTs: ts }; messages.push({ type: 'user', id: id('user'), turnId: turn.turnId, ts, text }); continue; } @@ -385,7 +402,7 @@ export function convertTranscript( if (!turn) { // A transcript can open with an assistant record when the session was // resumed. Give it a turn rather than dropping the content. - turn = { turnId: `claude-code:${sessionId}:turn:${sequence}`, lastTs: ts }; + turn = { turnId: nextTurnId(), lastTs: ts }; } if (record.isApiErrorMessage === true) turn.failed = true; const message = asMessageRecord(record); From 8316b773c6d930cea1b9d90ff6edc922f153242a Mon Sep 17 00:00:00 2001 From: Joob1n Date: Sat, 22 Aug 2026 00:52:49 +0800 Subject: [PATCH 3/5] fix(storage): normalize Claude source identity before emitting messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three findings from @Astro-Han's review. The blocking one was right, and my first attempt to check it was wrong in a way worth recording. **P1 — a transcript line is not a semantic event.** I verified the claim by counting duplicate `tool_call` ids in the output and found zero, which nearly became a reply saying the concern did not reproduce. That check could not have found anything: the converter mints a fresh id per record, so duplicates are impossible by construction. The review had already named this — "equal call/result counts are not sufficient here." Measuring the right thing showed the damage: across the 30 affected transcripts, 1683 assistant messages more than there were responses. One reply rendered as four. Two shapes cause it, both ordinary in resume and recovery histories: - A record can be written twice (3 repeats across 1130 local transcripts). Records now de-duplicate on `record.uuid`. - One response is split across records sharing `message.id`, and the pieces can be separated by the `tool_result` of a call an earlier piece made (374 occurrences across 30 transcripts). Its prose is now emitted once. Folding does not move the later fragments to the first one's position. A `tool_use` that arrives after a result was issued after it, and relocating it would invert cause and effect — so prose folds and calls stay where the log put them. Verified against the corpus first: `text` and `thinking` each appear at most once per `message.id`, so emitting them once loses nothing. Result on the affected transcripts: 1683 extra assistant messages became 280, and the remainder are genuinely separate responses — different `message.id`, different text, one per model reply after a tool returned. **P2 — one transcript per session id, chosen the same way twice.** `listSessions` exposed every file and `readSession` took whichever `.find()` reached first, so a user could select one summary and import another. Both now resolve through one map keyed by session id: newest mtime wins, path breaks a tie, so the choice does not depend on directory iteration order. **P2 — "no terminal state" was not what the system did with it.** Leaving `turn_state` out does not preserve "unfinished". `deriveTurnRecords` falls back to `inferLegacyTurnStatus`, which answers `completed` for any turn holding an assistant message (`session.ts:1250`) and marks it `inferred`; the Ledger then refuses that uncorroborated terminal and the repair path persists `failed / missing_terminal_event`. A transcript that was merely cut short would have imported as internal corruption. An unfinished turn is now recorded as what it is — `aborted` with `abortSource: 'external_session_snapshot'`, distinguishable from a user Stop or a provider abort. `end_turn`, interrupt notices and API errors keep their own evidence. Terminal coverage across 1128 local sessions goes from 97.2% to 100%, with the 32 newly covered turns landing in the snapshot-cutoff bucket rather than in the Ledger's repair path. Tool calls and results remain exactly paired at 3972 each. Claude-Session: https://claude.ai/code/session_014ajaRxC4jydavY9nYUFj5J Generated-by: Claude Code (Claude Opus 5) --- .../claude-code-session-adapter.test.ts | 99 ++++++++++++++- .../src/claude-code-session-adapter.ts | 117 ++++++++++++++++-- 2 files changed, 199 insertions(+), 17 deletions(-) diff --git a/packages/storage/src/__tests__/claude-code-session-adapter.test.ts b/packages/storage/src/__tests__/claude-code-session-adapter.test.ts index f9badd228a..43ae6b9720 100644 --- a/packages/storage/src/__tests__/claude-code-session-adapter.test.ts +++ b/packages/storage/src/__tests__/claude-code-session-adapter.test.ts @@ -35,7 +35,7 @@ describe('ClaudeCodeSessionAdapter', () => { }); }); - test('a turn stopped at tool_use with no result gets no terminal state', async () => { + test('a turn stopped at tool_use with no result is recorded as a snapshot cutoff', async () => { // The process died between the call and its result. Reporting `completed` // here is the failure mode this adapter exists to avoid. await withClaudeHome(async (home) => { @@ -44,17 +44,21 @@ describe('ClaudeCodeSessionAdapter', () => { assistantRecord({ toolUse: { id: 'toolu_1', name: 'Read' }, stopReason: 'tool_use' }), ]); const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000002'); - assert.equal(terminalState(messages), undefined); + const state = terminalState(messages); + assert.equal(state?.status, 'aborted'); + assert.equal(state?.abortSource, 'external_session_snapshot'); // The call itself still imports — the conversation is real up to the cut. assert.equal(messages.filter((m) => m.type === 'tool_call').length, 1); }); }); - test('a prompt with no answer at all gets no terminal state', async () => { + test('a prompt with no answer at all is recorded as a snapshot cutoff', async () => { await withClaudeHome(async (home) => { await seed(home, 'aaaaaaaa-0000-4000-8000-000000000003', [userRecord('are you there?')]); const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000003'); - assert.equal(terminalState(messages), undefined); + const state = terminalState(messages); + assert.equal(state?.status, 'aborted'); + assert.equal(state?.abortSource, 'external_session_snapshot'); assert.equal(messages.filter((m) => m.type === 'user').length, 1); }); }); @@ -114,7 +118,11 @@ describe('ClaudeCodeSessionAdapter', () => { assistantRecord({ text: 'Here is the beg', stopReason: 'max_tokens' }), ]); const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000011'); - assert.equal(terminalState(messages), undefined); + const state = terminalState(messages); + // Cut off, not completed — and named as the snapshot's edge rather than + // as a user Stop. + assert.equal(state?.status, 'aborted'); + assert.equal(state?.abortSource, 'external_session_snapshot'); }); }); @@ -161,6 +169,69 @@ describe('ClaudeCodeSessionAdapter', () => { }); }); + test('a record written twice emits one message, not two', async () => { + // A transcript is an append log: resume and recovery can replay a line. + // Persisting both copies duplicates the prompt in canonical history, and + // re-importing produces the same result — it is not recoverable after the + // fact. + await withClaudeHome(async (home) => { + const prompt = { ...userRecord('build the thing'), uuid: 'u-1' }; + await seed(home, 'aaaaaaaa-0000-4000-8000-000000000014', [ + prompt, + prompt, + assistantRecord({ text: 'Done.', stopReason: 'end_turn' }), + ]); + const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000014'); + assert.equal(messages.filter((m) => m.type === 'user').length, 1); + assert.equal(new Set(messages.map((m) => m.turnId)).size, 1); + }); + }); + + test('one response split across records emits its prose once and keeps calls in place', async () => { + // The real shape, measured 374 times across 30 local transcripts: text and + // a first tool_use, then the result of that call, then a second tool_use + // carrying the same `message.id`. The second call was issued after the + // result, so folding it back to the first record's position would invert + // cause and effect. + await withClaudeHome(async (home) => { + const fragment = (blocks: Parameters[1]) => + assistantFragment('msg_shared', blocks); + // Each fragment repeats the response's text, which is what a naive walk + // turns into three separate replies. + await seed(home, 'aaaaaaaa-0000-4000-8000-000000000015', [ + userRecord('read both files'), + fragment([{ type: 'text', text: 'Reading them now.' }]), + fragment([ + { type: 'text', text: 'Reading them now.' }, + { type: 'tool_use', id: 'toolu_1', name: 'Read', input: {} }, + ]), + toolResultRecord('toolu_1', 'first file'), + fragment([ + { type: 'text', text: 'Reading them now.' }, + { type: 'tool_use', id: 'toolu_2', name: 'Read', input: {} }, + ]), + toolResultRecord('toolu_2', 'second file'), + assistantRecord({ text: 'Both read.', stopReason: 'end_turn' }), + ]); + const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000015'); + + // The split response contributes one visible reply, not three. + const texts = messages.filter( + (m): m is Extract => m.type === 'assistant', + ); + assert.deepEqual( + texts.map((m) => m.text), + ['Reading them now.', 'Both read.'], + ); + + // Both calls survive, and each still precedes its own result. + const order = messages + .filter((m) => m.type === 'tool_call' || m.type === 'tool_result') + .map((m) => (m.type === 'tool_call' ? `call:${m.id}` : `result:${m.toolUseId}`)); + assert.deepEqual(order, ['call:toolu_1', 'result:toolu_1', 'call:toolu_2', 'result:toolu_2']); + }); + }); + test('a sidechain transcript is neither listed nor readable', async () => { // Sub-agent transcripts are whole files, so exclusion is per file. // Importing one would present a fragment of a conversation as a whole. @@ -277,6 +348,24 @@ function toolResultRecord(toolUseId: string, text: string): Record[], +): Record { + return { + type: 'assistant', + cwd: CWD, + timestamp: '2026-08-01T00:00:01.000Z', + message: { + role: 'assistant', + id: messageId, + model: 'claude-opus-5', + content, + stop_reason: 'tool_use', + }, + }; +} + function assistantRecord(input: { text?: string; thinking?: string; diff --git a/packages/storage/src/claude-code-session-adapter.ts b/packages/storage/src/claude-code-session-adapter.ts index aea9df43a9..6725c5e71c 100644 --- a/packages/storage/src/claude-code-session-adapter.ts +++ b/packages/storage/src/claude-code-session-adapter.ts @@ -126,7 +126,12 @@ export class ClaudeCodeSessionAdapter implements ExternalSessionAdapter { } catch { return []; } - const files: Array<{ path: string; sessionId: string }> = []; + // Keyed by session id: the same id can legitimately exist under more than + // one project directory after a workspace move or a resumed session. Two + // files with one id are two candidates for the same source session, and + // list and read must pick the same one or a user selects one summary and + // imports the other. + const bySessionId = new Map(); for (const project of projects) { let entries: string[]; try { @@ -143,10 +148,26 @@ export class ClaudeCodeSessionAdapter implements ExternalSessionAdapter { // The id reaches a path join, so the resolved file must still be under // the projects root — a crafted id must not read outside it. if (!resolve(path).startsWith(resolve(root))) continue; - files.push({ path, sessionId }); + let mtimeMs: number; + try { + mtimeMs = (await stat(path)).mtimeMs; + } catch { + continue; + } + const existing = bySessionId.get(sessionId); + // Newest wins, and the path breaks a tie so the choice does not depend + // on directory iteration order. A resumed session's continuation is + // the copy a user means when they pick that id. + if ( + !existing || + mtimeMs > existing.mtimeMs || + (mtimeMs === existing.mtimeMs && path < existing.path) + ) { + bySessionId.set(sessionId, { path, sessionId, mtimeMs }); + } } } - return files; + return [...bySessionId.values()].map(({ path, sessionId }) => ({ path, sessionId })); } async #parse(path: string, sessionId: string): Promise { @@ -275,6 +296,10 @@ export default ClaudeCodeSessionAdapter; * state, the same as any other turn whose end nothing vouches for. */ const TERMINAL_STOP_REASONS = new Set(['end_turn', 'stop_sequence']); +/** Names the cutoff for a turn the transcript simply stops inside, so a reader + * can tell an imported snapshot's edge from a user's Stop or a provider abort. */ +const EXTERNAL_SNAPSHOT_ABORT_SOURCE = 'external_session_snapshot'; + interface TurnAccumulator { turnId: string; lastTs: number; @@ -286,10 +311,50 @@ interface TurnAccumulator { aborted?: boolean; } +/** + * One semantic event per source record, before any of it becomes a message. + * + * A transcript is an append log, not a list of distinct events. Two shapes + * break a naive one-line-one-message walk, and both occur in ordinary resume + * and recovery histories: + * + * - **A record can be written twice.** Replaying it emits the prompt, turn, or + * tool identity twice. Measured: 3 repeats across 1130 local transcripts — + * rare, and unrecoverable once persisted as canonical history. + * - **One assistant response is split across records sharing `message.id`, + * and the pieces can be separated by the `tool_result` of a call the earlier + * piece made.** Measured: 374 occurrences across 30 transcripts. Emitting + * each piece as its own assistant message turned one reply into four. + * + * Folding must not flatten the second shape into its first record's position: + * a later `tool_use` sharing that id was issued *after* the result in between, + * and moving it earlier would invert cause and effect. So the visible prose of + * a response is emitted once, at its first appearance, while each tool call + * stays where the log put it. Verified against the corpus: `text` and + * `thinking` each appear at most once per `message.id`, so emitting them once + * loses nothing. + */ +function normalizeRecords(records: readonly TranscriptRecord[]): readonly TranscriptRecord[] { + const seenUuids = new Set(); + const normalized: TranscriptRecord[] = []; + for (const record of records) { + const uuid = typeof record.uuid === 'string' ? record.uuid : undefined; + if (uuid) { + if (seenUuids.has(uuid)) continue; + seenUuids.add(uuid); + } + normalized.push(record); + } + return normalized; +} + export function convertTranscript( sessionId: string, - records: readonly TranscriptRecord[], + rawRecords: readonly TranscriptRecord[], ): readonly StoredMessage[] { + const records = normalizeRecords(rawRecords); + // Prose already emitted for an assistant response, keyed by `message.id`. + const emittedProse = new Set(); const messages: StoredMessage[] = []; let turn: TurnAccumulator | undefined; let sequence = 0; @@ -302,12 +367,23 @@ export function convertTranscript( const closeTurn = (): void => { if (!turn) return; - // No terminal `turn_state` is emitted for a turn nothing corroborates. - // Measured across 1130 local transcripts, 13.9% of turns end this way — - // no assistant reply at all, or stopped at `tool_use` with the result - // never arriving. Those are genuinely unfinished: a killed process, a - // crash, or a session still open. Emitting `completed` for them would put - // a false terminal state on 1 in 7 imported turns. + // Every turn gets a terminal state, and which one depends on what the + // transcript actually says. + // + // Leaving one out is not the same as preserving "unfinished". Without a + // `turn_state`, `deriveTurnRecords` falls back to `inferLegacyTurnStatus`, + // which answers `completed` for any turn holding an assistant message + // (`session.ts:1250`) and marks it `inferred`. The Ledger then refuses + // that uncorroborated terminal and the repair path persists + // `failed / missing_terminal_event` — an internal-corruption verdict on a + // transcript that was merely cut short. Measured: 13.9% of turns across + // 1130 local transcripts end with no assistant reply or at a `tool_use` + // whose result never arrived. + // + // So an unfinished turn is recorded as what it is: a snapshot that ended + // mid-turn, with an `abortSource` naming the import rather than a user or + // a provider. `end_turn`, interrupt notices and API errors keep their own + // evidence and are unaffected. if (turn.aborted) { messages.push({ type: 'turn_state', @@ -338,6 +414,17 @@ export function convertTranscript( status: 'completed', partialOutputRetained: true, }); + } else { + messages.push({ + type: 'turn_state', + id: id('turn-state'), + turnId: turn.turnId, + ts: turn.lastTs, + status: 'aborted', + abortedAt: turn.lastTs, + abortSource: EXTERNAL_SNAPSHOT_ABORT_SOURCE, + partialOutputRetained: true, + }); } turn = undefined; }; @@ -413,7 +500,13 @@ export function convertTranscript( const stop = stringOf(message?.stop_reason); if (stop && TERMINAL_STOP_REASONS.has(stop)) turn.terminalStop = stop; - const thinking = thinkingText(message); + // One response's prose belongs to the response, not to each record that + // carries part of it. A fragment with no id is its own response. + const responseId = stringOf(message?.id); + const proseAlreadyEmitted = responseId !== undefined && emittedProse.has(responseId); + if (responseId !== undefined) emittedProse.add(responseId); + + const thinking = proseAlreadyEmitted ? '' : thinkingText(message); if (thinking) { messages.push({ type: 'assistant', @@ -426,7 +519,7 @@ export function convertTranscript( modelId, }); } - const text = claudeAssistantText(record); + const text = proseAlreadyEmitted ? undefined : claudeAssistantText(record); if (text) { messages.push({ type: 'assistant', From 82287ab5696a3229940759d8374453d936b5ddc2 Mon Sep 17 00:00:00 2001 From: Joob1n Date: Sat, 22 Aug 2026 01:03:24 +0800 Subject: [PATCH 4/5] test(runtime): pin the imported snapshot cutoff through Ledger materialization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit @Astro-Han asked for an end-to-end test through Ledger materialization, and I said in review that I had not added one. That was a decision made without checking: `runtime-ledger-repair.test.ts` already builds the stores, calls `materializeTranscriptLedger`, and asserts the resulting run — the harness I needed was sitting there. The cost I declined to pay was imagined. Two tests, and the pair is the point: - A turn carrying `aborted` with `abortSource: 'external_session_snapshot'` materializes to a `cancelled` run. The Ledger accepts it because the terminal is recorded rather than reconstructed. - The same turn with no terminal state at all materializes to `failed / missing_terminal_event`. The second is what the adapter used to produce. It now pins the reason the first exists, so the justification cannot quietly stop being true — and it turns "the Ledger repairs this to failed" from something I repeated out of a review comment into something that runs. Neither is observable from the adapter, which is exactly why the review asked for this seam: an adapter assertion can show which `turn_state` was emitted and nothing about what happens to it afterwards. Claude-Session: https://claude.ai/code/session_014ajaRxC4jydavY9nYUFj5J Generated-by: Claude Code (Claude Opus 5) --- .../__tests__/runtime-ledger-repair.test.ts | 142 ++++++++++++++++++ 1 file changed, 142 insertions(+) diff --git a/packages/runtime/src/__tests__/runtime-ledger-repair.test.ts b/packages/runtime/src/__tests__/runtime-ledger-repair.test.ts index d3903ed1fd..6e369d31af 100644 --- a/packages/runtime/src/__tests__/runtime-ledger-repair.test.ts +++ b/packages/runtime/src/__tests__/runtime-ledger-repair.test.ts @@ -192,3 +192,145 @@ test('repairs imported transcript turns into provider-neutral canonical history' await rm(root, { recursive: true, force: true }); } }); + +/** + * The Claude Code adapter records a turn the transcript stops inside as + * `aborted` with `abortSource: 'external_session_snapshot'`, rather than + * emitting no terminal state at all. + * + * The difference is only visible here. An adapter-level assertion can show + * which `turn_state` was emitted, but not what the Ledger does with it — and + * what it does is the whole reason the choice matters: an uncorroborated + * terminal is refused and the repair path writes `failed`, so a transcript + * that was merely cut short would import as internal corruption. + */ +test('an imported snapshot cutoff survives materialization as aborted', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-snapshot-cutoff-')); + const sessions = createSessionStore(root); + const runs = createSqliteAgentRunStore(root); + const runtimeEvents = createSqliteRuntimeStore(join(root, 'runtime.sqlite')); + let sequence = 0; + const newId = () => `cutoff-${++sequence}`; + + try { + const ts = Date.now() + 86_400_000; + const cutoffMessages: StoredMessage[] = [ + { type: 'user', id: 'c-user', turnId: 'turn-cut', ts, text: 'read the file' }, + { + type: 'assistant', + id: 'c-assistant', + turnId: 'turn-cut', + ts, + text: 'Reading it now.', + contentOrder: ['text'], + modelId: 'claude-opus-5', + }, + { + type: 'tool_call', + id: 'c-tool', + turnId: 'turn-cut', + ts, + toolName: 'Read', + args: { path: '/repo/a.ts' }, + }, + // No tool_result: the transcript ends between the call and its answer. + { + type: 'turn_state', + id: 'c-state', + turnId: 'turn-cut', + ts, + status: 'aborted', + abortedAt: ts, + abortSource: 'external_session_snapshot', + partialOutputRetained: true, + }, + ]; + const session = await sessions.createImportedSession( + { + cwd: '/repo', + llmConnectionSlug: 'anthropic', + model: 'claude-opus-5', + permissionMode: 'ask', + }, + cutoffMessages, + { adapterId: 'claude-code', sourceSessionId: 'cut-1' }, + ); + const repair = new RuntimeLedgerRepair({ + runStore: runs, + runtimeEventStore: runtimeEvents, + readMessages: (sessionId) => sessions.readMessages(sessionId), + appendMessage: (sessionId, message) => sessions.appendMessage(sessionId, message), + appendTurnState: async () => undefined, + newId, + now: () => 100, + }); + + await repair.materializeTranscriptLedger(session); + + const [run] = await runs.listSessionRuns(session.id); + assert.ok(run); + // `cancelled`, not `failed`: the Ledger accepted the recorded abort. Before + // the adapter emitted one, this same transcript materialized as + // `failed / missing_terminal_event`. + assert.equal(run.status, 'cancelled'); + assert.notEqual(run.failureClass, 'missing_terminal_event'); + } finally { + await runtimeEvents.close?.(); + await rm(root, { recursive: true, force: true }); + } +}); + +test('an imported turn with no terminal state is repaired to failed', async () => { + // The behaviour the adapter now avoids, pinned so the reason for emitting a + // cutoff cannot quietly stop being true. + const root = await mkdtemp(join(tmpdir(), 'maka-missing-terminal-')); + const sessions = createSessionStore(root); + const runs = createSqliteAgentRunStore(root); + const runtimeEvents = createSqliteRuntimeStore(join(root, 'runtime.sqlite')); + let sequence = 0; + const newId = () => `missing-${++sequence}`; + + try { + const ts = Date.now() + 86_400_000; + const session = await sessions.createImportedSession( + { + cwd: '/repo', + llmConnectionSlug: 'anthropic', + model: 'claude-opus-5', + permissionMode: 'ask', + }, + [ + { type: 'user', id: 'm-user', turnId: 'turn-missing', ts, text: 'read the file' }, + { + type: 'assistant', + id: 'm-assistant', + turnId: 'turn-missing', + ts, + text: 'Reading it now.', + contentOrder: ['text'], + modelId: 'claude-opus-5', + }, + ], + { adapterId: 'claude-code', sourceSessionId: 'missing-1' }, + ); + const repair = new RuntimeLedgerRepair({ + runStore: runs, + runtimeEventStore: runtimeEvents, + readMessages: (sessionId) => sessions.readMessages(sessionId), + appendMessage: (sessionId, message) => sessions.appendMessage(sessionId, message), + appendTurnState: async () => undefined, + newId, + now: () => 100, + }); + + await repair.materializeTranscriptLedger(session); + + const [run] = await runs.listSessionRuns(session.id); + assert.ok(run); + assert.equal(run.status, 'failed'); + assert.equal(run.failureClass, 'missing_terminal_event'); + } finally { + await runtimeEvents.close?.(); + await rm(root, { recursive: true, force: true }); + } +}); From 82e9ee8c386480cb270a557ad9c00ddad60158d6 Mon Sep 17 00:00:00 2001 From: Joob1n Date: Sat, 22 Aug 2026 08:21:49 +0800 Subject: [PATCH 5/5] fix(storage): resolve the Claude transcript graph before converting it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review was right that the converter read a lineage graph as a flat stream, and right that my `emittedProse` guard could drop real output. Measuring both turned up numbers larger than either of us assumed, and one shape neither of us had. **The guard dropped 4094 assistant text blocks across 392 of 1130 local transcripts.** A response is written as several records sharing `message.id`, and in 4093 of 14095 responses the visible text is not in the first of them — so emitting prose at the first fragment and suppressing it afterwards throws the reply away. My earlier check ("text appears at most once per `message.id`") was true and measured the wrong thing: the guard keys on the first fragment, not the first fragment with text. A response is now assembled from every fragment at the position of the first, thinking and text joined in fragment order, all of its calls before any of their results — they came from one API response, so a call written after a sibling's result did not follow it. No per-kind booleans: nothing is suppressed, so nothing needs a flag. **The graph branches for ordinary reasons, and mostly does.** Of 358 forked parents, 281 are one shape: a `tool_use` record whose two children are the next fragment of that same response and the result of the call it just made. Selecting "the active lineage" by walking parents back from the newest record calls all of that abandoned — on this corpus such a walk strands 6535 tool results and 18815 other records. I wrote that walk first; it is in the history of this branch and it was wrong. The one real ambiguity is 72 parents with two or more *user prompt* children: a rewind, where the user edited a prompt and resubmitted. ``` StreamVByte 是什么类型？我忘了 ← withdrawn StreamVByte 是什么方式？我忘了 ← asked ``` Resolution is now exactly that narrow — among sibling prompts the last written wins and takes the others' subtrees with it, 261 records across 21 transcripts. Everything else is kept, because nothing in the records says it was abandoned. **Compaction is not a branch and was being lost twice.** The boundary starts a new root; both sides are conversation that happened, and keeping only the newest would discard 24,695 records here. The `context_compacted` note was keyed on `isCompactSummary`, which belongs to the summary *user* record — consumed by the user branch and never reaching it — so the note was never emitted while `claudeUserAuthoredText` dropped the summary for being `isCompactSummary`. Both halves of the event, gone. It is now keyed on the `compact_boundary` record that states it, and held for the next turn when it arrives before one. 11 of 11 boundaries now survive. `logicalParentUuid` is deliberately not read as a parent: it equals `compactMetadata.preservedSegment.tailUuid`, a record written *after* the boundary, so following it points forward and closes a cycle through the summary. It stranded 2319 of 3392 records on one transcript before the walk was traced. Measured over 1130 local transcripts, before → after: | | before | after | | --- | ---: | ---: | | assistant text blocks dropped | 4094 | **0** | | assistant text preserved | 974,915 ch | **2,579,574 ch** | | tool_call / tool_result | 11970 / 11970 | **11993 / 11993** | | calls with no result | — | **0**, matching source | | compaction boundaries marked | 0 of 11 | **11 of 11** | Generated-by: Claude Opus 5 --- .../__tests__/runtime-ledger-repair.test.ts | 224 ++++++++++++++++- .../claude-code-session-adapter.test.ts | 43 ++-- .../claude-code-transcript-lineage.test.ts | 235 ++++++++++++++++++ .../src/claude-code-session-adapter.ts | 176 ++++++++----- .../src/claude-code-transcript-lineage.ts | 189 ++++++++++++++ 5 files changed, 786 insertions(+), 81 deletions(-) create mode 100644 packages/storage/src/__tests__/claude-code-transcript-lineage.test.ts create mode 100644 packages/storage/src/claude-code-transcript-lineage.ts diff --git a/packages/runtime/src/__tests__/runtime-ledger-repair.test.ts b/packages/runtime/src/__tests__/runtime-ledger-repair.test.ts index 6e369d31af..29fb7c8d41 100644 --- a/packages/runtime/src/__tests__/runtime-ledger-repair.test.ts +++ b/packages/runtime/src/__tests__/runtime-ledger-repair.test.ts @@ -1,5 +1,5 @@ import assert from 'node:assert/strict'; -import { mkdtemp, rm } from 'node:fs/promises'; +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { test } from 'node:test'; @@ -9,6 +9,7 @@ import { createSqliteRuntimeStore, createSessionStore, } from '@maka/storage'; +import { createExternalSessionAdapterRegistry } from '@maka/storage/external-sessions'; import { buildRuntimeEventModelReplayPlan } from '../model-history.js'; import { buildPriorRuntimeContext } from '../prior-run-context.js'; import { backfillRuntimeEventsFromStoredMessages } from '../runtime-event-backfill.js'; @@ -334,3 +335,224 @@ test('an imported turn with no terminal state is repaired to failed', async () = await rm(root, { recursive: true, force: true }); } }); + +test('a resolved Claude transcript replays as the conversation the user kept', async () => { + // The whole path, end to end: raw records → lineage resolution → conversion + // → Ledger materialization → the replay a continuation would be given. + // + // The transcript holds every shape the resolver exists for, together, the + // way a real one does: a rewound prompt whose branch was abandoned, a + // response split across records with its visible text in the LAST fragment, + // and two calls of one response separated by the first one's result. + const root = await mkdtemp(join(tmpdir(), 'maka-claude-lineage-')); + const sessions = createSessionStore(root); + const runs = createSqliteAgentRunStore(root); + const runtimeEvents = createSqliteRuntimeStore(join(root, 'runtime.sqlite')); + let sequence = 0; + const newId = () => `lineage-${++sequence}`; + const SOURCE_SESSION_ID = 'aaaaaaaa-0000-4000-8000-00000000e2e1'; + + try { + const stamp = (offset: number) => new Date(Date.now() + 86_400_000 + offset).toISOString(); + const records: Record[] = [ + { type: 'system', uuid: 'p0', parentUuid: null, timestamp: stamp(0), cwd: '/repo' }, + // The rewind, in the shape the corpus writes it: two prompts under one + // parent. The first was typed, answered, and then withdrawn by the edit. + { + type: 'user', + uuid: 'u1', + parentUuid: 'p0', + timestamp: stamp(500), + cwd: '/repo', + message: { role: 'user', content: 'summarise teh parser' }, + }, + { + type: 'assistant', + uuid: 'a_dead', + parentUuid: 'u1', + timestamp: stamp(1000), + cwd: '/repo', + message: { + role: 'assistant', + id: 'msg_dead', + model: 'claude-opus-5', + content: [{ type: 'text', text: 'ANSWER TO THE TYPO PROMPT' }], + stop_reason: 'end_turn', + }, + }, + { + type: 'user', + uuid: 'u2', + parentUuid: 'p0', + timestamp: stamp(2000), + cwd: '/repo', + message: { role: 'user', content: 'summarise the parser' }, + }, + // One response, three records. Thinking first, then a call, then — after + // that call's result — the visible text and a second call. + { + type: 'assistant', + uuid: 'a1', + parentUuid: 'u2', + timestamp: stamp(3000), + cwd: '/repo', + message: { + role: 'assistant', + id: 'msg_live', + model: 'claude-opus-5', + content: [{ type: 'thinking', thinking: 'Two files.' }], + stop_reason: 'tool_use', + }, + }, + { + type: 'assistant', + uuid: 'a2', + parentUuid: 'a1', + timestamp: stamp(4000), + cwd: '/repo', + message: { + role: 'assistant', + id: 'msg_live', + model: 'claude-opus-5', + content: [{ type: 'tool_use', id: 'toolu_a', name: 'Read', input: {} }], + stop_reason: 'tool_use', + }, + }, + { + type: 'user', + uuid: 'r1', + parentUuid: 'a2', + timestamp: stamp(5000), + cwd: '/repo', + message: { + role: 'user', + content: [{ type: 'tool_result', tool_use_id: 'toolu_a', content: 'parser.ts' }], + }, + }, + { + type: 'assistant', + uuid: 'a3', + parentUuid: 'r1', + timestamp: stamp(6000), + cwd: '/repo', + message: { + role: 'assistant', + id: 'msg_live', + model: 'claude-opus-5', + content: [ + { type: 'text', text: 'The parser is recursive descent.' }, + { type: 'tool_use', id: 'toolu_b', name: 'Read', input: {} }, + ], + stop_reason: 'end_turn', + }, + }, + { + type: 'user', + uuid: 'r2', + parentUuid: 'a3', + timestamp: stamp(7000), + cwd: '/repo', + message: { + role: 'user', + content: [{ type: 'tool_result', tool_use_id: 'toolu_b', content: 'lexer.ts' }], + }, + }, + ]; + + // Through the adapter registry, not a direct call to the converter: the + // seam an import actually crosses is `readSession`, and a test that + // skipped it would not notice the adapter refusing the transcript. + const claudeHome = join(root, 'claude-home'); + await mkdir(join(claudeHome, 'projects', '-repo'), { recursive: true }); + await writeFile( + join(claudeHome, 'projects', '-repo', `${SOURCE_SESSION_ID}.jsonl`), + `${records.map((record) => JSON.stringify(record)).join('\n')}\n`, + ); + const adapter = createExternalSessionAdapterRegistry({ + claudeCode: { claudeHome }, + }).require('claude-code'); + const messages = [...(await adapter.readSession(SOURCE_SESSION_ID)).messages]; + const session = await sessions.createImportedSession( + { + cwd: '/repo', + llmConnectionSlug: 'anthropic', + model: 'claude-opus-5', + permissionMode: 'ask', + }, + messages, + { adapterId: 'claude-code', sourceSessionId: SOURCE_SESSION_ID }, + ); + const repair = new RuntimeLedgerRepair({ + runStore: runs, + runtimeEventStore: runtimeEvents, + readMessages: (sessionId) => sessions.readMessages(sessionId), + appendMessage: (sessionId, message) => sessions.appendMessage(sessionId, message), + appendTurnState: async () => undefined, + newId, + now: () => 100, + }); + await repair.materializeTranscriptLedger(session); + + const [run] = await runs.listSessionRuns(session.id); + assert.ok(run); + const events = await runtimeEvents.readRuntimeEvents(session.id, run.runId); + const replay = buildRuntimeEventModelReplayPlan(events).items; + const text = replay + .filter((item) => item.kind === 'text') + .map((item) => (item as { role: string; content: string }).content); + + // No selected text is lost: the reply lives in the LAST fragment of its + // response, which the previous converter suppressed for not being first. + assert.ok( + text.some((content) => content.includes('The parser is recursive descent.')), + `reply missing from replay: ${JSON.stringify(text)}`, + ); + // No abandoned branch is imported: the rewound prompt and its answer are + // not context the continuation should be given. + assert.equal( + text.some((content) => content.includes('ANSWER TO THE TYPO PROMPT')), + false, + ); + assert.equal( + text.some((content) => content.includes('summarise teh parser')), + false, + ); + assert.ok(text.some((content) => content.includes('summarise the parser'))); + + // Every result is still paired with the correct call, and both calls of + // one response precede both results — they came from one API response, so + // neither was issued after the other's result came back. + const calls = messages.filter((m) => m.type === 'tool_call'); + const results = messages.filter((m) => m.type === 'tool_result'); + assert.deepEqual( + calls.map((m) => m.id), + ['toolu_a', 'toolu_b'], + ); + assert.deepEqual( + results.map((m) => (m as Extract).toolUseId), + ['toolu_a', 'toolu_b'], + ); + const shape = messages + .filter((m) => m.type === 'tool_call' || m.type === 'tool_result') + .map((m) => + m.type === 'tool_call' + ? `call:${m.id}` + : `result:${(m as Extract).toolUseId}`, + ); + assert.deepEqual(shape, [ + 'call:toolu_a', + 'call:toolu_b', + 'result:toolu_a', + 'result:toolu_b', + ]); + + // And the turn is terminal on its own evidence, not repaired into one. + assert.equal(run.status, 'completed'); + assert.notEqual(run.failureClass, 'missing_terminal_event'); + } finally { + runtimeEvents.close(); + runs.close?.(); + await sessions.close?.(); + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/packages/storage/src/__tests__/claude-code-session-adapter.test.ts b/packages/storage/src/__tests__/claude-code-session-adapter.test.ts index 43ae6b9720..86f0c10567 100644 --- a/packages/storage/src/__tests__/claude-code-session-adapter.test.ts +++ b/packages/storage/src/__tests__/claude-code-session-adapter.test.ts @@ -187,25 +187,27 @@ describe('ClaudeCodeSessionAdapter', () => { }); }); - test('one response split across records emits its prose once and keeps calls in place', async () => { - // The real shape, measured 374 times across 30 local transcripts: text and - // a first tool_use, then the result of that call, then a second tool_use - // carrying the same `message.id`. The second call was issued after the - // result, so folding it back to the first record's position would invert - // cause and effect. + test('one response is assembled from every fragment, calls before their results', async () => { + // The real shape, measured across 1130 local transcripts: a response is + // written as several records sharing `message.id`, the pieces separated by + // the results of calls the earlier pieces made. In 4093 of 14095 + // responses the visible text is NOT in the first fragment — so a guard + // that emits prose at the first record and suppresses it afterwards drops + // the reply. + // + // Both calls share one `message.id`, so they came from one API response + // and were issued together however the log interleaved them with the + // results arriving: they are one assistant step, and both precede both + // results. await withClaudeHome(async (home) => { const fragment = (blocks: Parameters[1]) => assistantFragment('msg_shared', blocks); - // Each fragment repeats the response's text, which is what a naive walk - // turns into three separate replies. await seed(home, 'aaaaaaaa-0000-4000-8000-000000000015', [ userRecord('read both files'), - fragment([{ type: 'text', text: 'Reading them now.' }]), - fragment([ - { type: 'text', text: 'Reading them now.' }, - { type: 'tool_use', id: 'toolu_1', name: 'Read', input: {} }, - ]), + fragment([{ type: 'thinking', thinking: 'Two files to read.' }]), + fragment([{ type: 'tool_use', id: 'toolu_1', name: 'Read', input: {} }]), toolResultRecord('toolu_1', 'first file'), + // The text arrives last, after a call and its result — the 4093 case. fragment([ { type: 'text', text: 'Reading them now.' }, { type: 'tool_use', id: 'toolu_2', name: 'Read', input: {} }, @@ -215,20 +217,25 @@ describe('ClaudeCodeSessionAdapter', () => { ]); const messages = await read(home, 'aaaaaaaa-0000-4000-8000-000000000015'); - // The split response contributes one visible reply, not three. - const texts = messages.filter( + const assistants = messages.filter( (m): m is Extract => m.type === 'assistant', ); + // The reply survives even though no fragment before it carried text, and + // the response is still one reply rather than one per fragment. assert.deepEqual( - texts.map((m) => m.text), + assistants.filter((m) => m.text.length > 0).map((m) => m.text), ['Reading them now.', 'Both read.'], ); + // The thinking from the first fragment belongs to the same response. + assert.deepEqual( + assistants.filter((m) => m.thinking).map((m) => m.thinking?.text), + ['Two files to read.'], + ); - // Both calls survive, and each still precedes its own result. const order = messages .filter((m) => m.type === 'tool_call' || m.type === 'tool_result') .map((m) => (m.type === 'tool_call' ? `call:${m.id}` : `result:${m.toolUseId}`)); - assert.deepEqual(order, ['call:toolu_1', 'result:toolu_1', 'call:toolu_2', 'result:toolu_2']); + assert.deepEqual(order, ['call:toolu_1', 'call:toolu_2', 'result:toolu_1', 'result:toolu_2']); }); }); diff --git a/packages/storage/src/__tests__/claude-code-transcript-lineage.test.ts b/packages/storage/src/__tests__/claude-code-transcript-lineage.test.ts new file mode 100644 index 0000000000..b969ba32e2 --- /dev/null +++ b/packages/storage/src/__tests__/claude-code-transcript-lineage.test.ts @@ -0,0 +1,235 @@ +import assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; +import { + isPromptRecord, + resolveTranscriptLineage, + transcriptParentUuid, + type TranscriptRecord, +} from '../claude-code-transcript-lineage.js'; + +function user(uuid: string, parentUuid: string | null, text: string): TranscriptRecord { + return { type: 'user', uuid, parentUuid, message: { role: 'user', content: text } }; +} + +function assistant( + uuid: string, + parentUuid: string | null, + messageId: string, + content: readonly Record[], +): TranscriptRecord { + return { + type: 'assistant', + uuid, + parentUuid, + message: { role: 'assistant', id: messageId, content }, + }; +} + +function toolResult(uuid: string, parentUuid: string, toolUseId: string): TranscriptRecord { + return { + type: 'user', + uuid, + parentUuid, + message: { role: 'user', content: [{ type: 'tool_result', tool_use_id: toolUseId }] }, + }; +} + +/** The boundary as Claude Code writes it: no parent, and a *forward* logical link. */ +function compactBoundary(uuid: string, preservedTailUuid: string): TranscriptRecord { + return { + type: 'system', + subtype: 'compact_boundary', + uuid, + parentUuid: null, + logicalParentUuid: preservedTailUuid, + compactMetadata: { trigger: 'manual', preservedSegment: { tailUuid: preservedTailUuid } }, + }; +} + +const uuids = (result: { records: readonly TranscriptRecord[] }): readonly unknown[] => + result.records.map((record) => record.uuid); + +describe('resolveTranscriptLineage', () => { + test('a withdrawn prompt and its answer are dropped, the edit kept', () => { + // The corpus shape, 72 times: the user typed a prompt, edited it, and + // resubmitted. Both are written under the same parent. Importing both + // presents a question the user withdrew, and its answer, as conversation. + const result = resolveTranscriptLineage([ + { type: 'system', uuid: 'p0', parentUuid: null }, + user('u1', 'p0', 'StreamVByte 是什么类型？我忘了'), + assistant('a1', 'u1', 'msg_1', [{ type: 'text', text: 'answer to the withdrawn one' }]), + user('u2', 'p0', 'StreamVByte 是什么方式？我忘了'), + assistant('a2', 'u2', 'msg_2', [{ type: 'text', text: 'answer to the one asked' }]), + ]); + assert.deepEqual(uuids(result), ['p0', 'u2', 'a2']); + assert.equal(result.withdrawnPrompts, 1); + assert.equal(result.abandoned, 2); + }); + + test('the whole withdrawn subtree goes, not just its head', () => { + const result = resolveTranscriptLineage([ + { type: 'system', uuid: 'p0', parentUuid: null }, + user('u1', 'p0', 'withdrawn'), + assistant('a1', 'u1', 'msg_1', [{ type: 'tool_use', id: 'call_dead', name: 'Read' }]), + toolResult('r1', 'a1', 'call_dead'), + assistant('a2', 'r1', 'msg_2', [{ type: 'text', text: 'abandoned' }]), + user('u2', 'p0', 'asked'), + assistant('a3', 'u2', 'msg_3', [{ type: 'text', text: 'kept' }]), + ]); + assert.deepEqual(uuids(result), ['p0', 'u2', 'a3']); + assert.equal(result.abandoned, 4); + }); + + test('three siblings leave only the last', () => { + // Measured 3 times in the corpus: a prompt edited twice. + const result = resolveTranscriptLineage([ + { type: 'system', uuid: 'p0', parentUuid: null }, + user('u1', 'p0', 'first try'), + user('u2', 'p0', 'second try'), + user('u3', 'p0', 'third try'), + assistant('a3', 'u3', 'msg_3', [{ type: 'text', text: 'kept' }]), + ]); + assert.deepEqual(uuids(result), ['p0', 'u3', 'a3']); + assert.equal(result.withdrawnPrompts, 2); + }); + + test('an ordinary tool fork survives whole', () => { + // 281 of 358 forks in the corpus are this: a `tool_use` record has two + // children — the next fragment of the same response, and the result of + // the call it just made. Nothing was abandoned, and a walk that picked + // one child would strand the other. + const result = resolveTranscriptLineage([ + user('u1', null, 'prompt'), + assistant('a1', 'u1', 'msg_1', [{ type: 'tool_use', id: 'call_1', name: 'Read' }]), + assistant('a2', 'a1', 'msg_1', [{ type: 'tool_use', id: 'call_2', name: 'Read' }]), + toolResult('r1', 'a1', 'call_1'), + toolResult('r2', 'a2', 'call_2'), + ]); + assert.deepEqual(uuids(result), ['u1', 'a1', 'a2', 'r1', 'r2']); + assert.equal(result.abandoned, 0); + }); + + test('two tool results under one parent are both kept', () => { + // Sibling `user` records are a rewind only when they are prompts. A tool + // result is the harness answering the model, not the user asking again. + const result = resolveTranscriptLineage([ + user('u1', null, 'prompt'), + assistant('a1', 'u1', 'msg_1', [ + { type: 'tool_use', id: 'call_1', name: 'Read' }, + { type: 'tool_use', id: 'call_2', name: 'Read' }, + ]), + toolResult('r1', 'a1', 'call_1'), + toolResult('r2', 'a1', 'call_2'), + ]); + assert.deepEqual(uuids(result), ['u1', 'a1', 'r1', 'r2']); + assert.equal(result.abandoned, 0); + assert.equal(result.withdrawnPrompts, 0); + }); + + test('an assistant fork with no prompt in it is left alone', () => { + const result = resolveTranscriptLineage([ + user('u1', null, 'prompt'), + assistant('a1', 'u1', 'msg_1', [{ type: 'text', text: 'one' }]), + assistant('a2', 'u1', 'msg_2', [{ type: 'text', text: 'two' }]), + ]); + assert.deepEqual(uuids(result), ['u1', 'a1', 'a2']); + assert.equal(result.abandoned, 0); + }); + + test('a tool result is not a prompt', () => { + assert.equal(isPromptRecord(user('u1', null, 'ask')), true); + assert.equal(isPromptRecord(toolResult('r1', 'a1', 'call_1')), false); + assert.equal(isPromptRecord(assistant('a1', 'u1', 'm', [])), false); + }); + + test('both sides of a compaction boundary survive', () => { + // The boundary starts a new root because the model's context restarted + // there. Both sides are conversation that happened. + const result = resolveTranscriptLineage([ + user('u1', null, 'before compaction'), + assistant('a1', 'u1', 'msg_1', [{ type: 'text', text: 'pre-boundary reply' }]), + compactBoundary('b1', 'a1'), + { type: 'user', uuid: 's1', parentUuid: 'b1', isCompactSummary: true, message: {} }, + user('u2', 's1', 'after compaction'), + assistant('a2', 'u2', 'msg_2', [{ type: 'text', text: 'post-boundary reply' }]), + ]); + assert.deepEqual(uuids(result), ['u1', 'a1', 'b1', 's1', 'u2', 'a2']); + assert.equal(result.abandoned, 0); + assert.equal(result.compactBoundaries, 1); + }); + + test("a boundary's logicalParentUuid is not followed as a parent", () => { + // It holds `preservedSegment.tailUuid` — a record written AFTER the + // boundary. Reading it as a parent points forward and closes a cycle + // through the summary. + assert.equal(transcriptParentUuid(compactBoundary('b1', 'later')), undefined); + const result = resolveTranscriptLineage([ + user('u1', null, 'before'), + assistant('a1', 'u1', 'msg_1', [{ type: 'text', text: 'pre' }]), + compactBoundary('b1', 'u2'), + { type: 'user', uuid: 's1', parentUuid: 'b1', isCompactSummary: true, message: {} }, + user('u2', 's1', 'after'), + ]); + assert.deepEqual(uuids(result), ['u1', 'a1', 'b1', 's1', 'u2']); + assert.equal(result.abandoned, 0); + }); + + test('a rewind before a compaction boundary does not disturb what follows', () => { + const result = resolveTranscriptLineage([ + { type: 'system', uuid: 'p0', parentUuid: null }, + user('u1', 'p0', 'withdrawn before'), + assistant('a_dead', 'u1', 'msg_dead', [{ type: 'text', text: 'abandoned pre-boundary' }]), + user('u1b', 'p0', 'asked before'), + assistant('a1', 'u1b', 'msg_1', [{ type: 'text', text: 'kept pre-boundary' }]), + compactBoundary('b1', 'a1'), + user('u2', 'b1', 'after'), + assistant('a2', 'u2', 'msg_2', [{ type: 'text', text: 'kept post-boundary' }]), + ]); + assert.deepEqual(uuids(result), ['p0', 'u1b', 'a1', 'b1', 'u2', 'a2']); + assert.equal(result.abandoned, 2); + }); + + test('a repeated uuid is kept once', () => { + const first = user('u1', null, 'prompt'); + const result = resolveTranscriptLineage([first, { ...first }, assistant('a1', 'u1', 'm', [])]); + assert.deepEqual(uuids(result), ['u1', 'a1']); + assert.equal(result.duplicates, 1); + }); + + test('a record the graph cannot place is kept', () => { + // Silence from the graph is not evidence of abandonment. Dropping history + // is the failure that cannot be undone once it is canonical. + const result = resolveTranscriptLineage([ + { type: 'user', message: { role: 'user', content: 'no uuid' } }, + user('u1', null, 'prompt'), + assistant('a1', 'u1', 'msg_1', [{ type: 'text', text: 'reply' }]), + ]); + assert.equal(result.records.length, 3); + assert.equal(result.abandoned, 0); + }); + + test('a sidechain record is passed through untouched', () => { + const result = resolveTranscriptLineage([ + user('u1', null, 'prompt'), + { type: 'assistant', uuid: 'sc1', parentUuid: 'nothing', isSidechain: true, message: {} }, + ]); + assert.equal(result.records.length, 2); + assert.equal(result.abandoned, 0); + }); + + test('a parent cycle terminates instead of hanging the import', () => { + // These links are written by another process; a cycle has to fail the + // walk, not spin in it. + const result = resolveTranscriptLineage([ + { type: 'user', uuid: 'x', parentUuid: 'y', message: { role: 'user', content: 'a' } }, + { type: 'user', uuid: 'y', parentUuid: 'x', message: { role: 'user', content: 'b' } }, + ]); + assert.equal(result.records.length, 2); + }); + + test('an empty transcript resolves to nothing rather than throwing', () => { + const result = resolveTranscriptLineage([]); + assert.deepEqual(result.records, []); + assert.equal(result.abandoned, 0); + }); +}); diff --git a/packages/storage/src/claude-code-session-adapter.ts b/packages/storage/src/claude-code-session-adapter.ts index 6725c5e71c..68dab79699 100644 --- a/packages/storage/src/claude-code-session-adapter.ts +++ b/packages/storage/src/claude-code-session-adapter.ts @@ -28,6 +28,10 @@ import type { ExternalSessionSummary, } from '@maka/core/external-session'; import type { StoredMessage } from '@maka/core/session'; +import { + resolveTranscriptLineage, + type TranscriptRecord, +} from './claude-code-transcript-lineage.js'; export const CLAUDE_CODE_SESSION_ADAPTER_ID = 'claude-code'; @@ -47,8 +51,6 @@ export interface ClaudeCodeSessionAdapterOptions { maxTranscriptBytes?: number; } -type TranscriptRecord = Record; - interface ParsedTranscript { readonly records: readonly TranscriptRecord[]; readonly cwd: string; @@ -312,50 +314,35 @@ interface TurnAccumulator { } /** - * One semantic event per source record, before any of it becomes a message. - * - * A transcript is an append log, not a list of distinct events. Two shapes - * break a naive one-line-one-message walk, and both occur in ordinary resume - * and recovery histories: - * - * - **A record can be written twice.** Replaying it emits the prompt, turn, or - * tool identity twice. Measured: 3 repeats across 1130 local transcripts — - * rare, and unrecoverable once persisted as canonical history. - * - **One assistant response is split across records sharing `message.id`, - * and the pieces can be separated by the `tool_result` of a call the earlier - * piece made.** Measured: 374 occurrences across 30 transcripts. Emitting - * each piece as its own assistant message turned one reply into four. - * - * Folding must not flatten the second shape into its first record's position: - * a later `tool_use` sharing that id was issued *after* the result in between, - * and moving it earlier would invert cause and effect. So the visible prose of - * a response is emitted once, at its first appearance, while each tool call - * stays where the log put it. Verified against the corpus: `text` and - * `thinking` each appear at most once per `message.id`, so emitting them once - * loses nothing. + * Which records are the conversation, decided before any of them becomes a + * message. See `claude-code-transcript-lineage` for what the transcript's own + * fields say about rewind branches, compaction boundaries, and fragmented + * responses; this file only converts what that returns. */ -function normalizeRecords(records: readonly TranscriptRecord[]): readonly TranscriptRecord[] { - const seenUuids = new Set(); - const normalized: TranscriptRecord[] = []; - for (const record of records) { - const uuid = typeof record.uuid === 'string' ? record.uuid : undefined; - if (uuid) { - if (seenUuids.has(uuid)) continue; - seenUuids.add(uuid); - } - normalized.push(record); - } - return normalized; -} - export function convertTranscript( sessionId: string, rawRecords: readonly TranscriptRecord[], ): readonly StoredMessage[] { - const records = normalizeRecords(rawRecords); - // Prose already emitted for an assistant response, keyed by `message.id`. - const emittedProse = new Set(); + const records = resolveTranscriptLineage(rawRecords).records; + // Every fragment of one assistant response, keyed by `message.id`. A + // response is emitted once, from all of its fragments, at the position of + // the first — so a later fragment's text is part of the reply rather than + // something the first fragment's absence of text can suppress. + const responseFragments = new Map(); + for (const record of records) { + if (record.type !== 'assistant') continue; + const responseId = stringOf(asMessageRecord(record)?.id); + if (responseId === undefined) continue; + const existing = responseFragments.get(responseId); + if (existing) existing.push(record); + else responseFragments.set(responseId, [record]); + } + const emittedResponses = new Set(); const messages: StoredMessage[] = []; + // A boundary can precede the first turn — a transcript that opens straight + // after a compaction. A system note needs a turn to hang from, so the fact + // waits for one rather than being dropped for arriving early. + let pendingCompactBoundaryTs: number | undefined; let turn: TurnAccumulator | undefined; let sequence = 0; const id = (kind: string): string => `claude-code:${sessionId}:${kind}:${sequence++}`; @@ -481,6 +468,16 @@ export function convertTranscript( // A human-authored user record opens a new turn. closeTurn(); turn = { turnId: nextTurnId(), lastTs: ts }; + if (pendingCompactBoundaryTs !== undefined) { + messages.push({ + type: 'system_note', + id: id('compact'), + turnId: turn.turnId, + ts: pendingCompactBoundaryTs, + kind: 'context_compacted', + }); + pendingCompactBoundaryTs = undefined; + } messages.push({ type: 'user', id: id('user'), turnId: turn.turnId, ts, text }); continue; } @@ -490,23 +487,53 @@ export function convertTranscript( // A transcript can open with an assistant record when the session was // resumed. Give it a turn rather than dropping the content. turn = { turnId: nextTurnId(), lastTs: ts }; + if (pendingCompactBoundaryTs !== undefined) { + messages.push({ + type: 'system_note', + id: id('compact'), + turnId: turn.turnId, + ts: pendingCompactBoundaryTs, + kind: 'context_compacted', + }); + pendingCompactBoundaryTs = undefined; + } } if (record.isApiErrorMessage === true) turn.failed = true; const message = asMessageRecord(record); + const responseId = stringOf(message?.id); + // A response is emitted once, at its first fragment, assembled from all + // of them. A later fragment reached here is that same response still + // being written — its content is already in what was emitted, and + // emitting again would repeat the reply. + if (responseId !== undefined) { + if (emittedResponses.has(responseId)) continue; + emittedResponses.add(responseId); + } + // A fragment with no id stands alone; it is the only fragment of itself. + const fragments = + (responseId === undefined ? undefined : responseFragments.get(responseId)) ?? [record]; + + // Status evidence is read from every fragment, not just the first: the + // `stop_reason` lands on whichever fragment the response finished on. + for (const fragment of fragments) { + if (fragment.isApiErrorMessage === true) turn.failed = true; + const stop = stringOf(asMessageRecord(fragment)?.stop_reason); + if (stop && TERMINAL_STOP_REASONS.has(stop)) turn.terminalStop = stop; + } + // The transcript names the model that produced each step. Carrying the // real value keeps an imported turn attributable; a placeholder would // put a model the user never ran onto their history. const modelId = stringOf(message?.model) ?? 'claude-code'; - const stop = stringOf(message?.stop_reason); - if (stop && TERMINAL_STOP_REASONS.has(stop)) turn.terminalStop = stop; - - // One response's prose belongs to the response, not to each record that - // carries part of it. A fragment with no id is its own response. - const responseId = stringOf(message?.id); - const proseAlreadyEmitted = responseId !== undefined && emittedProse.has(responseId); - if (responseId !== undefined) emittedProse.add(responseId); - const thinking = proseAlreadyEmitted ? '' : thinkingText(message); + // Concatenated in fragment order, which is the order the response was + // streamed. Joining rather than picking one: every delta is content the + // model produced, and choosing between them would be choosing which + // half of a reply to keep. + const thinking = fragments + .map((fragment) => thinkingText(asMessageRecord(fragment))) + .filter((part) => part.length > 0) + .join('\n\n'); if (thinking) { messages.push({ type: 'assistant', @@ -519,7 +546,10 @@ export function convertTranscript( modelId, }); } - const text = proseAlreadyEmitted ? undefined : claudeAssistantText(record); + const text = fragments + .map((fragment) => claudeAssistantText(fragment)) + .filter((part): part is string => part !== undefined && part.length > 0) + .join('\n\n'); if (text) { messages.push({ type: 'assistant', @@ -531,22 +561,44 @@ export function convertTranscript( modelId, }); } - for (const block of toolUseBlocks(message)) { - messages.push({ - type: 'tool_call', - // The id must equal the tool_use id so the result can match it. - id: stringOf(block.id) ?? id('tool-call'), - turnId: turn.turnId, - ts, - toolName: stringOf(block.name) ?? 'unknown', - args: block.input ?? {}, - }); + // Every call the response made, before any of their results. Calls + // sharing a `message.id` came from one API response, so they were + // issued together however the log interleaved them with the results + // arriving; a call written after its sibling's result did not follow it. + for (const fragment of fragments) { + for (const block of toolUseBlocks(asMessageRecord(fragment))) { + messages.push({ + type: 'tool_call', + // The id must equal the tool_use id so the result can match it. + id: stringOf(block.id) ?? id('tool-call'), + turnId: turn.turnId, + ts, + toolName: stringOf(block.name) ?? 'unknown', + args: block.input ?? {}, + }); + } } continue; } - if (record.isCompactSummary === true) { - if (!turn) continue; + // The compaction boundary, keyed on the record that states it. + // + // It used to be keyed on `isCompactSummary`, which belongs to the summary + // *user* record — and that record is consumed by the `user` branch above + // and never reaches here, so the note was never emitted. The import then + // carried the pre-boundary history flat with nothing saying a compaction + // had happened, while `claudeUserAuthoredText` dropped the summary itself + // for being `isCompactSummary`: both halves of the event lost at once. + // + // Pre-boundary records stay. They are the conversation that actually + // happened — 24,695 of them across the 5 compacted transcripts here — and + // the boundary marks where the model's context restarted, which is the + // part a reader cannot reconstruct from the messages themselves. + if (record.subtype === 'compact_boundary') { + if (!turn) { + pendingCompactBoundaryTs = ts; + continue; + } messages.push({ type: 'system_note', id: id('compact'), diff --git a/packages/storage/src/claude-code-transcript-lineage.ts b/packages/storage/src/claude-code-transcript-lineage.ts new file mode 100644 index 0000000000..b0f59d7eb6 --- /dev/null +++ b/packages/storage/src/claude-code-transcript-lineage.ts @@ -0,0 +1,189 @@ +/** + * Source resolution for a Claude Code transcript, ahead of any conversion. + * + * A `.jsonl` transcript is an append-only log of a `uuid` / `parentUuid` + * graph, and the graph is not a path. Reading it as one is the mistake this + * module exists to avoid — in both directions: + * + * **It branches for ordinary reasons.** Of 358 forked parents across 1130 + * local transcripts, 281 are one shape: an assistant record carrying a + * `tool_use` has two children, the next fragment of that same response and + * the `tool_result` of the call it just made. Nothing was abandoned; a + * response and a completion simply share a parent. Selecting "the active + * lineage" by walking parents back from the newest record treats all of that + * as dead — measured on this corpus, such a walk strands 6535 tool results + * and 18815 other records. + * + * **It branches for one real reason.** 72 forked parents have two or more + * *user prompt* children. Those are rewinds — the user edited a prompt and + * resubmitted — and the transcript keeps both. Importing both presents a + * question the user withdrew, and its answer, as conversation: + * + * ``` + * StreamVByte 是什么类型？我忘了 ← withdrawn + * StreamVByte 是什么方式？我忘了 ← asked + * ``` + * + * So resolution is exactly that narrow: among sibling prompts the last one + * written wins, and a withdrawn prompt takes its subtree with it. Every other + * branch is kept, because nothing in the records says it was abandoned. + * + * **Compaction is not a branch.** The boundary record carries + * `parentUuid: null` and starts a new root, because after a compaction the + * model's context no longer holds what came before. Both sides are + * conversation that happened and both are kept — keeping only the newest root + * would discard 24,695 records here. Its `logicalParentUuid` is *not* the + * backward link it resembles: it equals + * `compactMetadata.preservedSegment.tailUuid`, a record written after the + * boundary, so following it as a parent points forward and closes a cycle. + * + * Every rule above is read off fields the transcript states outright. Where + * the records are silent the record is kept: dropping history is the failure + * that cannot be undone once it is persisted as canonical. + */ + +export type TranscriptRecord = Record; + +export interface LineageResolution { + /** The selected records, in the order the file wrote them. */ + readonly records: readonly TranscriptRecord[]; + /** Records dropped as descending from a withdrawn prompt. */ + readonly abandoned: number; + /** Prompts withdrawn by a later sibling. */ + readonly withdrawnPrompts: number; + /** Records dropped as a repeat of a `uuid` already seen. */ + readonly duplicates: number; + /** `compact_boundary` records the file carries. */ + readonly compactBoundaries: number; +} + +function stringField(record: TranscriptRecord, key: string): string | undefined { + const value = record[key]; + return typeof value === 'string' && value.length > 0 ? value : undefined; +} + +/** + * The parent a record hangs from: `parentUuid`, and nothing else. + * + * `logicalParentUuid` is deliberately not consulted. On a compaction boundary + * it holds `preservedSegment.tailUuid` — a record written after the boundary, + * not before it — so reading it as a parent points forward and closes a cycle + * back through the summary. + */ +export function transcriptParentUuid(record: TranscriptRecord): string | undefined { + return stringField(record, 'parentUuid'); +} + +/** + * A record that is a human prompt rather than the harness speaking. + * + * Tool results are written as `user` records — the harness answering the + * model — so a response's completions are not prompts, and two of them + * sharing a parent is not a rewind. That distinction is the whole of the + * difference between the 281 ordinary forks and the 72 real ones. + */ +export function isPromptRecord(record: TranscriptRecord): boolean { + if (record.type !== 'user') return false; + const message = record.message; + if (typeof message !== 'object' || message === null) return true; + const content = (message as Record).content; + if (!Array.isArray(content)) return true; + return !content.some( + (block) => + typeof block === 'object' && + block !== null && + (block as Record).type === 'tool_result', + ); +} + +/** + * Drop repeats of a `uuid` already seen, keeping the first. + * + * A record written twice replays whatever identity it carries — a prompt, a + * turn boundary, a tool call. Measured: 3 repeats across 1130 local + * transcripts. Rare enough to be invisible in testing, permanent once it is + * persisted as canonical history. + */ +function deduplicate(records: readonly TranscriptRecord[]): { + readonly kept: readonly TranscriptRecord[]; + readonly duplicates: number; +} { + const seen = new Set(); + const kept: TranscriptRecord[] = []; + let duplicates = 0; + for (const record of records) { + const uuid = stringField(record, 'uuid'); + if (uuid !== undefined) { + if (seen.has(uuid)) { + duplicates += 1; + continue; + } + seen.add(uuid); + } + kept.push(record); + } + return { kept, duplicates }; +} + +const ROOT_KEY = ' root'; + +export function resolveTranscriptLineage( + rawRecords: readonly TranscriptRecord[], +): LineageResolution { + const { kept: records, duplicates } = deduplicate(rawRecords); + const compactBoundaries = records.filter((r) => r.subtype === 'compact_boundary').length; + + const main = records.filter((r) => r.isSidechain !== true && stringField(r, 'uuid') !== undefined); + if (main.length === 0) { + return { records, abandoned: 0, withdrawnPrompts: 0, duplicates, compactBoundaries }; + } + + const present = new Set(main.map((r) => stringField(r, 'uuid') as string)); + const childrenOf = new Map(); + for (const record of main) { + const parent = transcriptParentUuid(record); + // A parent outside this file is no parent: the record roots its own + // segment rather than being orphaned into nothing. + const key = parent !== undefined && present.has(parent) ? parent : ROOT_KEY; + const siblings = childrenOf.get(key); + if (siblings) siblings.push(record); + else childrenOf.set(key, [record]); + } + + // Among sibling prompts the last written is the one that was asked; the + // earlier ones were withdrawn by the edit that replaced them. + const withdrawn: TranscriptRecord[] = []; + for (const [, siblings] of childrenOf) { + const prompts = siblings.filter(isPromptRecord); + if (prompts.length < 2) continue; + withdrawn.push(...prompts.slice(0, -1)); + } + if (withdrawn.length === 0) { + return { records, abandoned: 0, withdrawnPrompts: 0, duplicates, compactBoundaries }; + } + + // A withdrawn prompt takes its subtree: the answer to a question that was + // never asked is not conversation either. + const dropped = new Set(); + const stack = [...withdrawn]; + while (stack.length > 0) { + const record = stack.pop() as TranscriptRecord; + const uuid = stringField(record, 'uuid') as string; + if (dropped.has(uuid)) continue; + dropped.add(uuid); + for (const child of childrenOf.get(uuid) ?? []) stack.push(child); + } + + const resolved = records.filter((record) => { + const uuid = stringField(record, 'uuid'); + return uuid === undefined || !dropped.has(uuid); + }); + + return { + records: resolved, + abandoned: dropped.size, + withdrawnPrompts: withdrawn.length, + duplicates, + compactBoundaries, + }; +}