From 0eeb328c18073b21f8ac96cbb8331c107fa35c14 Mon Sep 17 00:00:00 2001 From: cat0825 Date: Thu, 6 Aug 2026 00:20:27 +0800 Subject: [PATCH 1/7] fix(storage): import legacy JSONL session transcripts into SQLite (#2260) After the JSONL->SQLite cutover (#1994, #2029), sessions created before the switch stayed on disk as sessions//session.jsonl but never appeared in the UI: the new storage layer only reads SQLite and there was no migration path (issue #2260). Add a one-time importer (importLegacySessionsOnce) that scans the legacy sessions directory, decodes each schemaVersion:1 transcript with the pre-#1994 compatibility rules (backend remapping, missing-field defaults), creates the session under its original id via the idempotent createStableSession path, appends the decoded messages, and restores the original lifecycle timestamps and flags. Design: - Idempotency key is the session id itself (probeStableSessionCreate), so re-runs and concurrent first launches converge without duplicates. - Per-file atomicity: a transcript imports fully or is skipped and reported; corrupt records are never laundered into the authoritative store. Failures never block startup or other files. - Legacy files are retained as migration evidence. - Wired into createSessionStore: list/listCatalogPage/listHeaders await the lazy import so upgraded installs see their pre-cutover sessions. --- .../corrupt-line-session.jsonl | 5 + .../no-turn-state-session.jsonl | 3 + .../legacy-sessions/normal-session.jsonl | 7 + .../sparse-header-session.jsonl | 4 + .../__tests__/legacy-session-import.test.ts | 157 ++++++++ packages/storage/src/index.ts | 2 + packages/storage/src/legacy-session-import.ts | 371 ++++++++++++++++++ packages/storage/src/session-store.ts | 27 ++ 8 files changed, 576 insertions(+) create mode 100644 packages/storage/src/__tests__/fixtures/legacy-sessions/corrupt-line-session.jsonl create mode 100644 packages/storage/src/__tests__/fixtures/legacy-sessions/no-turn-state-session.jsonl create mode 100644 packages/storage/src/__tests__/fixtures/legacy-sessions/normal-session.jsonl create mode 100644 packages/storage/src/__tests__/fixtures/legacy-sessions/sparse-header-session.jsonl create mode 100644 packages/storage/src/__tests__/legacy-session-import.test.ts create mode 100644 packages/storage/src/legacy-session-import.ts diff --git a/packages/storage/src/__tests__/fixtures/legacy-sessions/corrupt-line-session.jsonl b/packages/storage/src/__tests__/fixtures/legacy-sessions/corrupt-line-session.jsonl new file mode 100644 index 0000000000..62a4bd37a1 --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/legacy-sessions/corrupt-line-session.jsonl @@ -0,0 +1,5 @@ +{"id":"legacy-corrupt-0001","workspaceRoot":"/Users/fixture/Library/Application Support/Maka/workspaces/default","cwd":"/Users/fixture/work/corrupt-project","createdAt":1783921000000,"lastUsedAt":1783921010000,"name":"Corrupt Session","isFlagged":false,"labels":[],"isArchived":false,"status":"active","statusUpdatedAt":1783921010000,"hasUnread":false,"backend":"ai-sdk","llmConnectionSlug":"openai-compatible","connectionLocked":true,"model":"demo-model","permissionMode":"ask","schemaVersion":1,"lastMessageAt":1783921010000} +{"type":"user","id":"msg-user-corrupt-1","turnId":"turn-corrupt-1","ts":1783921005000,"text":"this file is corrupt"} +{"type":"turn_state","id":"msg-turnstate-corrupt","turnId":"turn-corrupt-1","ts":1783921005000,"status":"running","partialOutputRetained":false} +{this line is not valid json +{"type":"turn_state","id":"msg-turnstate-corrupt-done","turnId":"turn-corrupt-1","ts":1783921010000,"status":"completed","partialOutputRetained":false} diff --git a/packages/storage/src/__tests__/fixtures/legacy-sessions/no-turn-state-session.jsonl b/packages/storage/src/__tests__/fixtures/legacy-sessions/no-turn-state-session.jsonl new file mode 100644 index 0000000000..5f0bc36598 --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/legacy-sessions/no-turn-state-session.jsonl @@ -0,0 +1,3 @@ +{"id":"legacy-noturnstate-1","workspaceRoot":"/Users/fixture/Library/Application Support/Maka/workspaces/default","cwd":"/Users/fixture/work/no-turnstate-project","createdAt":1783922000000,"lastUsedAt":1783922010000,"name":"No Turn State","isFlagged":false,"labels":[],"isArchived":false,"status":"active","statusUpdatedAt":1783922010000,"hasUnread":false,"backend":"ai-sdk","llmConnectionSlug":"openai-compatible","connectionLocked":true,"model":"demo-model","permissionMode":"ask","schemaVersion":1,"lastMessageAt":1783922010000} +{"type":"user","id":"msg-user-noturn-1","turnId":"turn-noturn-1","ts":1783922005000,"text":"hello"} +{"type":"assistant","id":"msg-assistant-noturn-1","turnId":"turn-noturn-1","ts":1783922009000,"text":"Hello back","modelId":"demo-model"} diff --git a/packages/storage/src/__tests__/fixtures/legacy-sessions/normal-session.jsonl b/packages/storage/src/__tests__/fixtures/legacy-sessions/normal-session.jsonl new file mode 100644 index 0000000000..f8c8570604 --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/legacy-sessions/normal-session.jsonl @@ -0,0 +1,7 @@ +{"id":"legacy-a1b2c3d4e5f6","workspaceRoot":"/Users/fixture/Library/Application Support/Maka/workspaces/default","cwd":"/Users/fixture/work/demo-project","createdAt":1783929889564,"lastUsedAt":1783929891400,"name":"New Chat","isFlagged":false,"labels":[],"isArchived":false,"status":"active","statusUpdatedAt":1783929891400,"hasUnread":false,"backend":"ai-sdk","llmConnectionSlug":"openai-compatible","connectionLocked":true,"model":"demo-model","permissionMode":"ask","schemaVersion":1,"lastMessageAt":1783929891400} +{"type":"user","id":"msg-user-0001","turnId":"turn-0001","ts":1783929889572,"text":"hi"} +{"type":"turn_state","id":"msg-turnstate-running","turnId":"turn-0001","ts":1783929889572,"status":"running","partialOutputRetained":false} +{"type":"assistant","id":"msg-assistant-0001","turnId":"turn-0001","ts":1783929891000,"text":"Hello! How can I help you today?","modelId":"demo-model"} +{"type":"token_usage","id":"msg-tokenusage-0001","turnId":"turn-0001","ts":1783929891300,"input":5,"output":12,"total":17,"cacheHitInput":0,"cacheMissInput":5,"costUsd":0.0001,"rawFinishReason":"stop","runtimeSteps":1} +{"type":"turn_state","id":"msg-turnstate-completed","turnId":"turn-0001","ts":1783929891400,"status":"completed","partialOutputRetained":false} +{"type":"system_note","id":"msg-systemnote-0001","turnId":"turn-0001","ts":1783929891400,"kind":"session_resume"} diff --git a/packages/storage/src/__tests__/fixtures/legacy-sessions/sparse-header-session.jsonl b/packages/storage/src/__tests__/fixtures/legacy-sessions/sparse-header-session.jsonl new file mode 100644 index 0000000000..90bb126b7d --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/legacy-sessions/sparse-header-session.jsonl @@ -0,0 +1,4 @@ +{"id":"legacy-sparse-0001","workspaceRoot":"/Users/fixture/Library/Application Support/Maka/workspaces/default","cwd":"/Users/fixture/work/sparse-project","createdAt":1783920000000,"lastUsedAt":1783920010000,"name":"Sparse Session","isFlagged":false,"isArchived":false,"hasUnread":false,"backend":"claude","llmConnectionSlug":"anthropic-compatible","connectionLocked":true,"model":"","schemaVersion":1} +{"type":"user","id":"msg-user-sparse-1","turnId":"turn-sparse-1","ts":1783920005000,"text":"hello from a sparse session"} +{"type":"turn_state","id":"msg-turnstate-sparse","turnId":"turn-sparse-1","ts":1783920005000,"status":"running","partialOutputRetained":false} +{"type":"turn_state","id":"msg-turnstate-sparse-done","turnId":"turn-sparse-1","ts":1783920010000,"status":"completed","partialOutputRetained":false} diff --git a/packages/storage/src/__tests__/legacy-session-import.test.ts b/packages/storage/src/__tests__/legacy-session-import.test.ts new file mode 100644 index 0000000000..08801947b2 --- /dev/null +++ b/packages/storage/src/__tests__/legacy-session-import.test.ts @@ -0,0 +1,157 @@ +import assert from 'node:assert/strict'; +import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { fileURLToPath } from 'node:url'; +import { importLegacySessionsOnce } from '../legacy-session-import.js'; +import { createSessionStore } from '../session-store.js'; + +function fixturePath(name: string): string { + // tsc emits only .ts sources to dist, so fixtures stay in src and are read + // from the compiled test via a relative path back up the tree. + return fileURLToPath( + new URL(`../../src/__tests__/fixtures/legacy-sessions/${name}`, import.meta.url), + ); +} + +async function seedLegacySession( + workspace: string, + sessionId: string, + fixtureName: string, +): Promise { + const content = await readFile(fixturePath(fixtureName), 'utf8'); + const dir = join(workspace, 'sessions', sessionId); + await mkdir(dir, { recursive: true }); + await writeFile(join(dir, 'session.jsonl'), content); +} + +async function withWorkspace( + run: (context: { + sessions: ReturnType; + workspace: string; + }) => Promise, +): Promise { + const root = await mkdtemp(join(tmpdir(), 'maka-legacy-import-')); + const workspace = join(root, 'workspace'); + const sessions = createSessionStore(workspace); + try { + await run({ sessions, workspace }); + } finally { + await sessions.close?.(); + await rm(root, { recursive: true, force: true }); + } +} + +test('imports a complete legacy session with messages in order and restored timestamps', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-a1b2c3d4e5f6', 'normal-session.jsonl'); + + const result = await importLegacySessionsOnce(sessions, workspace); + assert.deepEqual(result, { + imported: 1, + skipped: 0, + failed: 0, + failures: [], + }); + + const header = await sessions.readHeaderSnapshot('legacy-a1b2c3d4e5f6'); + // The importer restores the legacy lifecycle facts that createStableSession + // would otherwise stamp with `Date.now()`. + assert.equal(header.createdAt, 1783929889564); + assert.equal(header.lastUsedAt, 1783929891400); + assert.equal(header.lastMessageAt, 1783929891400); + assert.equal(header.connectionLocked, true); + assert.equal(header.name, 'New Chat'); + assert.equal(header.cwd, '/Users/fixture/work/demo-project'); + + const messages = await sessions.readMessages('legacy-a1b2c3d4e5f6'); + assert.deepEqual( + messages.map((message) => message.type), + ['user', 'turn_state', 'assistant', 'token_usage', 'turn_state', 'system_note'], + ); + assert.equal(messages[0]?.type, 'user'); + assert.equal((messages[0] as { text?: string }).text, 'hi'); + assert.equal(messages[2]?.type, 'assistant'); + assert.equal((messages[2] as { text?: string }).text, 'Hello! How can I help you today?'); + + // The session appears in the list through the public surface too, which + // also exercises the factory wiring (list awaits the lazy import). + const summaries = await sessions.list(); + assert.ok(summaries.some((summary) => summary.id === 'legacy-a1b2c3d4e5f6')); + }); +}); + +test('is idempotent: a second run skips the already-imported session without duplicating data', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-a1b2c3d4e5f6', 'normal-session.jsonl'); + + const first = await importLegacySessionsOnce(sessions, workspace); + assert.equal(first.imported, 1); + + const second = await importLegacySessionsOnce(sessions, workspace); + assert.deepEqual(second, { imported: 0, skipped: 1, failed: 0, failures: [] }); + + const messages = await sessions.readMessages('legacy-a1b2c3d4e5f6'); + assert.equal(messages.length, 6, 'no duplicate messages after a second import'); + }); +}); + +test('a corrupt transcript is skipped as failed while other sessions still import', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-a1b2c3d4e5f6', 'normal-session.jsonl'); + await seedLegacySession(workspace, 'legacy-corrupt-0001', 'corrupt-line-session.jsonl'); + + const result = await importLegacySessionsOnce(sessions, workspace); + assert.equal(result.imported, 1); + assert.equal(result.failed, 1); + assert.equal(result.failures.length, 1); + assert.equal(result.failures[0]?.sessionId, 'legacy-corrupt-0001'); + assert.match(result.failures[0]?.error ?? '', /corrupt JSONL record/); + + // The corrupt session was not partially created. + await assert.rejects( + sessions.readHeaderSnapshot('legacy-corrupt-0001'), + /Session metadata not found/, + ); + const good = await sessions.readMessages('legacy-a1b2c3d4e5f6'); + assert.equal(good.length, 6); + }); +}); + +test('a sparse legacy header is completed with compatibility defaults', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-sparse-0001', 'sparse-header-session.jsonl'); + + const result = await importLegacySessionsOnce(sessions, workspace); + assert.equal(result.imported, 1); + + const header = await sessions.readHeaderSnapshot('legacy-sparse-0001'); + // Missing optional fields fall back to the pre-#1994 defaults. + assert.equal(header.permissionMode, 'ask'); + assert.equal(header.collaborationMode, 'agent'); + assert.equal(header.orchestrationMode, 'default'); + // The legacy `claude` backend is remapped, and an empty model to `default`. + assert.equal(header.backend, 'ai-sdk'); + assert.equal(header.model, 'default'); + // A non-default name is treated as a manually set title. + assert.equal(header.titleIsManual, true); + assert.equal(header.name, 'Sparse Session'); + }); +}); + +test('a legacy turn without turn_state records normalizes to completed on read', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-noturnstate-1', 'no-turn-state-session.jsonl'); + + const result = await importLegacySessionsOnce(sessions, workspace); + assert.equal(result.imported, 1); + + const turns = await sessions.listTurns('legacy-noturnstate-1'); + assert.ok(turns.length >= 1); + // With no recorded turn_state, core infers the turn as completed from the + // presence of an assistant message. + assert.equal(turns[0]?.status, 'completed'); + assert.equal(turns[0]?.statusSource, 'inferred'); + }); +}); diff --git a/packages/storage/src/index.ts b/packages/storage/src/index.ts index f67c54972c..94be1d5815 100644 --- a/packages/storage/src/index.ts +++ b/packages/storage/src/index.ts @@ -9,6 +9,8 @@ export { normalizeSessionHeader, projectSessionCatalogMessages, } from './session-store.js'; +export { importLegacySessionsOnce } from './legacy-session-import.js'; +export type { LegacySessionImportResult } from './legacy-session-import.js'; export type { CreateStableSessionRequest, CreateStableSessionResult, diff --git a/packages/storage/src/legacy-session-import.ts b/packages/storage/src/legacy-session-import.ts new file mode 100644 index 0000000000..e4a6f7f830 --- /dev/null +++ b/packages/storage/src/legacy-session-import.ts @@ -0,0 +1,371 @@ +import { createHash } from 'node:crypto'; +import type { Dirent } from 'node:fs'; +import { readdir, readFile } from 'node:fs/promises'; +import { join } from 'node:path'; +import { + DEFAULT_SESSION_NAME, + decodeStoredMessageForRecovery, + isCollaborationMode, + isOrchestrationMode, + isPermissionMode, + isSessionBlockedReason, + isSessionStatus, + type CreateSessionInput, + type SessionHeader, + type StoredMessage, +} from '@maka/core'; +import type { SessionAuthorityStore } from './session-store.js'; + +/** + * One-time importer for the legacy file-backed session transcripts that the + * JSONL→SQLite cutover (#1994) left behind. + * + * Before #1994, every session lived at `sessions//session.jsonl`: + * line 1 is a `schemaVersion: 1` header record, every following line is a + * message record. #1994 made SQLite the sole operational authority and dropped + * the JSONL session tree (#2029) without an import path, so sessions created + * before the cutover stay on disk but never appear in the UI (issue #2260). + * + * Design: + * - Runs once per session: idempotency key is the session id itself. We probe + * with `probeStableSessionCreate` and skip any id already present in SQLite, + * so re-running (e.g. after a partial failure or a second launch) is safe + * and does not duplicate data. + * - Per-file atomicity: a file either imports completely (header + all + * messages) or is skipped and counted as failed. We never write a partial + * session, and we never silently degrade a corrupt record into a synthetic + * note in the new authoritative store. + * - Best-effort: a malformed or unreadable file is reported in the result and + * never blocks startup or the import of other sessions. + * - Legacy files are retained after import, honoring the repository policy + * that legacy stores are kept as migration evidence. + * + * The compatibility defaults mirror the old `decodeSessionHeader` (kept public + * for one-way importers before #1994 deleted it): missing `permissionMode` + * defaults to `ask`, `collaborationMode` to `agent`, `orchestrationMode` to + * `default`, model to `default`, and `claude`/`pi` backends are remapped. + * Final header validation happens inside `createStableSession` (the metadata + * store runs `normalizeSessionHeader`), so this module does not re-implement + * it. + */ + +const LEGACY_SESSIONS_DIR = 'sessions'; +const LEGACY_TRANSCRIPT_FILE = 'session.jsonl'; + +/** + * Stable request fingerprint for legacy imports. `createStableSession` + * requires a `sha256:` fingerprint; using a constant derived from this + * module's identity makes every import of the same session id the "same + * request", so concurrent first-launch processes converge on one winner. + */ +const LEGACY_IMPORT_FINGERPRINT = `sha256:${createHash('sha256') + .update('maka-legacy-session-import') + .digest('hex')}`; + +export interface LegacySessionImportResult { + imported: number; + skipped: number; + failed: number; + failures: Array<{ sessionId: string; error: string }>; +} + +/** + * Import all legacy `sessions//session.jsonl` transcripts under + * `workspaceRoot` into the SQLite-backed session store. + * + * The store must be open (a `createSessionStore(workspaceRoot)` instance is + * ready immediately). Idempotent; safe to call on every launch. + */ +export async function importLegacySessionsOnce( + store: SessionAuthorityStore, + workspaceRoot: string, +): Promise { + const result: LegacySessionImportResult = { imported: 0, skipped: 0, failed: 0, failures: [] }; + const sessionsDir = join(workspaceRoot, LEGACY_SESSIONS_DIR); + + let entries: Dirent[]; + try { + entries = await readdir(sessionsDir, { withFileTypes: true }); + } catch (error) { + // No legacy sessions directory is the normal case for installs that + // never predate the cutover; it is not a failure. + if (isNodeErrorWithCode(error, 'ENOENT')) return result; + throw error; + } + + const sessionDirs = entries + .filter((entry) => entry.isDirectory()) + .map((entry) => entry.name) + .sort(); + + for (const sessionId of sessionDirs) { + const transcriptPath = join(sessionsDir, sessionId, LEGACY_TRANSCRIPT_FILE); + try { + const outcome = await importLegacySessionFile(store, sessionId, transcriptPath); + if (outcome === 'imported') { + result.imported += 1; + } else { + result.skipped += 1; + } + } catch (error) { + result.failed += 1; + result.failures.push({ + sessionId, + error: error instanceof Error ? error.message : String(error), + }); + } + } + + return result; +} + +async function importLegacySessionFile( + store: SessionAuthorityStore, + sessionId: string, + transcriptPath: string, +): Promise<'imported' | 'skipped'> { + const { header, messages } = await readLegacyTranscript(transcriptPath, sessionId); + + // Idempotency: any session id already known to SQLite — whether imported by + // an earlier run or created by the user — is left untouched. + const probe = await store.probeStableSessionCreate(sessionId, LEGACY_IMPORT_FINGERPRINT); + if (probe.kind !== 'absent') return 'skipped'; + + const input = toCreateSessionInput(header); + const created = await store.createStableSession({ + sessionId, + requestFingerprint: LEGACY_IMPORT_FINGERPRINT, + input, + }); + // A concurrent process may have won the race between probe and create; + // both `existing` and `conflict` mean the id is taken, so skip. + if (created.kind !== 'created') return 'skipped'; + + await store.appendMessages(sessionId, messages); + + // `createStableSession` stamps now-based timestamps and default flags; the + // legacy header carries the real lifecycle facts, so restore them. + await store.updateHeader(sessionId, legacyHeaderPatch(header, messages)); + + return 'imported'; +} + +async function readLegacyTranscript( + transcriptPath: string, + sessionId: string, +): Promise<{ header: SessionHeader; messages: StoredMessage[] }> { + const text = await readFile(transcriptPath, 'utf8'); + const lines = text.split('\n').filter((line) => line.trim().length > 0); + if (lines.length === 0) { + throw new Error(`Legacy session ${sessionId} is empty`); + } + + const header = decodeLegacySessionHeader(JSON.parse(lines[0]!), sessionId); + const messages: StoredMessage[] = []; + for (let index = 1; index < lines.length; index += 1) { + const line = lines[index]!; + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch (error) { + throw new Error( + `Legacy session ${sessionId} has a corrupt JSONL record at line ${index + 1}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + // Strict decode: a malformed record fails the whole file rather than + // being laundered into the new authoritative store. + messages.push(decodeStoredMessageForRecovery(parsed)); + } + return { header, messages }; +} + +/** + * The legacy header was a loose JSON object (some fields optional, old + * backends named differently). This mirrors the pre-#1994 `decodeSessionHeader` + * compatibility rules; the final strict shape is enforced by the metadata + * store's `normalizeSessionHeader` on create. + */ +export function decodeLegacySessionHeader(value: unknown, sessionId: string): SessionHeader { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new Error(`Invalid legacy session header for session ${sessionId}: expected an object`); + } + const header = value as LegacyStoredSessionHeader; + const permissionMode = isPermissionMode(header.permissionMode) ? header.permissionMode : 'ask'; + const collaborationMode = isCollaborationMode(header.collaborationMode) + ? header.collaborationMode + : 'agent'; + const orchestrationMode = isOrchestrationMode(header.orchestrationMode) + ? header.orchestrationMode + : 'default'; + const model = + typeof header.model === 'string' && header.model.length > 0 ? header.model : 'default'; + const status = resolveLegacyStatus(header); + const blockedReason = + status === 'blocked' && isSessionBlockedReason(header.blockedReason) + ? header.blockedReason + : undefined; + const statusUpdatedAt = + header.statusUpdatedAt ?? + header.archivedAt ?? + header.lastMessageAt ?? + header.lastUsedAt ?? + header.createdAt; + const titleIsManual = + typeof header.titleIsManual === 'boolean' + ? header.titleIsManual + : normalizeLegacySessionName(header.name) !== DEFAULT_SESSION_NAME; + + const backend = legacyBackend(header.backend); + + return { + id: sessionId, + workspaceRoot: header.workspaceRoot, + cwd: header.cwd, + ...(header.projectId !== undefined ? { projectId: header.projectId } : {}), + createdAt: header.createdAt, + lastUsedAt: header.lastUsedAt, + ...(header.lastMessageAt !== undefined ? { lastMessageAt: header.lastMessageAt } : {}), + name: header.name, + titleIsManual, + isFlagged: header.isFlagged, + labels: (header.labels ?? []) as string[], + isArchived: header.isArchived, + ...(header.archivedAt !== undefined ? { archivedAt: header.archivedAt } : {}), + status, + ...(blockedReason !== undefined ? { blockedReason } : {}), + statusUpdatedAt, + ...(header.parentSessionId !== undefined ? { parentSessionId: header.parentSessionId } : {}), + ...(header.branchOfTurnId !== undefined ? { branchOfTurnId: header.branchOfTurnId } : {}), + ...(header.revisionRootSessionId !== undefined + ? { revisionRootSessionId: header.revisionRootSessionId } + : {}), + ...(header.revisionParentSessionId !== undefined + ? { revisionParentSessionId: header.revisionParentSessionId } + : {}), + ...(header.revisionOfTurnId !== undefined ? { revisionOfTurnId: header.revisionOfTurnId } : {}), + ...(header.revisionIndex !== undefined ? { revisionIndex: header.revisionIndex } : {}), + ...(header.revisionState !== undefined ? { revisionState: header.revisionState } : {}), + hasUnread: header.hasUnread, + backend, + llmConnectionSlug: header.llmConnectionSlug, + connectionLocked: header.connectionLocked, + model, + permissionMode, + collaborationMode, + orchestrationMode, + schemaVersion: 1, + }; +} + +function legacyBackend(backend: unknown): SessionHeader['backend'] { + if (backend === 'ai-sdk' || backend === 'claude') return 'ai-sdk'; + if (backend === 'pi-agent' || backend === 'pi') return 'pi-agent'; + return 'fake'; +} + +function resolveLegacyStatus(header: LegacyStoredSessionHeader): SessionHeader['status'] { + if (header.isArchived) return 'archived'; + if (isSessionStatus(header.status) && header.status !== 'archived') return header.status; + return 'active'; +} + +function normalizeLegacySessionName(name: string): string { + return name === 'New Session' ? DEFAULT_SESSION_NAME : name; +} + +function toCreateSessionInput(header: SessionHeader): CreateSessionInput { + return { + cwd: header.cwd, + ...(header.projectId !== undefined ? { projectId: header.projectId } : {}), + name: header.name, + backend: header.backend, + llmConnectionSlug: header.llmConnectionSlug, + model: header.model, + permissionMode: header.permissionMode, + collaborationMode: header.collaborationMode, + orchestrationMode: header.orchestrationMode, + ...(header.status !== undefined ? { status: header.status } : {}), + ...(header.blockedReason !== undefined ? { blockedReason: header.blockedReason } : {}), + labels: header.labels, + ...(header.parentSessionId !== undefined ? { parentSessionId: header.parentSessionId } : {}), + ...(header.branchOfTurnId !== undefined ? { branchOfTurnId: header.branchOfTurnId } : {}), + ...(header.revisionRootSessionId !== undefined + ? { revisionRootSessionId: header.revisionRootSessionId } + : {}), + ...(header.revisionParentSessionId !== undefined + ? { revisionParentSessionId: header.revisionParentSessionId } + : {}), + ...(header.revisionOfTurnId !== undefined ? { revisionOfTurnId: header.revisionOfTurnId } : {}), + ...(header.revisionIndex !== undefined ? { revisionIndex: header.revisionIndex } : {}), + ...(header.revisionState !== undefined ? { revisionState: header.revisionState } : {}), + }; +} + +/** + * Fields `buildSessionHeader` cannot express (it stamps `Date.now()` and + * default flags) but the legacy header carries. `updateHeader` re-validates + * through `normalizeSessionHeader`, so the values must be canonically shaped. + */ +function legacyHeaderPatch( + header: SessionHeader, + messages: readonly StoredMessage[], +): Partial { + const hasUserMessage = messages.some((message) => message.type === 'user'); + return { + createdAt: header.createdAt, + lastUsedAt: header.lastUsedAt, + ...(header.lastMessageAt !== undefined ? { lastMessageAt: header.lastMessageAt } : {}), + statusUpdatedAt: header.statusUpdatedAt, + titleIsManual: header.titleIsManual, + isFlagged: header.isFlagged, + isArchived: header.isArchived, + ...(header.archivedAt !== undefined ? { archivedAt: header.archivedAt } : {}), + hasUnread: header.hasUnread, + connectionLocked: header.connectionLocked || hasUserMessage, + }; +} + +/** The loose legacy header shape accepted by `decodeLegacySessionHeader`. */ +type LegacyStoredSessionHeader = { + workspaceRoot: string; + cwd: string; + projectId?: string | null; + createdAt: number; + lastUsedAt: number; + lastMessageAt?: number; + name: string; + titleIsManual?: unknown; + isFlagged: boolean; + labels?: unknown; + isArchived: boolean; + archivedAt?: number; + status?: unknown; + blockedReason?: unknown; + statusUpdatedAt?: number; + parentSessionId?: string; + branchOfTurnId?: string; + revisionRootSessionId?: string; + revisionParentSessionId?: string; + revisionOfTurnId?: string; + revisionIndex?: number; + revisionState?: 'preparing' | 'committed'; + hasUnread: boolean; + backend: unknown; + llmConnectionSlug: string; + connectionLocked: boolean; + model?: unknown; + permissionMode?: unknown; + collaborationMode?: unknown; + orchestrationMode?: unknown; +}; + +function isNodeErrorWithCode(error: unknown, code: string): boolean { + return ( + typeof error === 'object' && + error !== null && + 'code' in error && + (error as { code?: unknown }).code === code + ); +} diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index cd6e7eaf1b..b427ce3f22 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -13,6 +13,7 @@ import { type VersionedSessionIdentity, } from './sqlite-session-metadata-store.js'; import { isDiscardableConversationCopy } from './session-conversation-copy.js'; +import { importLegacySessionsOnce } from './legacy-session-import.js'; import { acquireOperationalStateDatabase, OPERATIONAL_STATE_DATABASE_NAME, @@ -272,6 +273,7 @@ export function createSessionStoreWithTestDependencies( class SqliteSessionStore implements SessionAuthorityStore { private readonly metadata: SqliteSessionMetadataStore; private readonly workspaceRoot: string; + private legacyImportPromise: Promise | null = null; private closePromise: Promise | null = null; constructor(workspaceRoot: string, _dependencies: SessionAuthorityStoreTestDependencies) { @@ -283,6 +285,28 @@ class SqliteSessionStore implements SessionAuthorityStore { ); } + /** + * One-time legacy JSONL session import, awaited before any read that feeds + * the session list so upgraded installs see their pre-cutover sessions. + * + * The import itself is best-effort (per-file errors are reported, never + * thrown), so this promise never rejects; it exists only to serialize the + * first list against the import. + */ + private ensureLegacyImported(): Promise { + this.legacyImportPromise ??= importLegacySessionsOnce(this, this.workspaceRoot).then( + (result) => { + if (result.imported > 0 || result.failed > 0) { + console.error( + `[storage] legacy session import: ${result.imported} imported, ${result.skipped} skipped, ${result.failed} failed`, + result.failures.length > 0 ? result.failures : undefined, + ); + } + }, + ); + return this.legacyImportPromise; + } + async create( input: CreateSessionInput, initialBoundary?: ExecutionBoundary, @@ -454,6 +478,7 @@ class SqliteSessionStore implements SessionAuthorityStore { async list(filter?: SessionListFilter): Promise { await this.ensureReady(); + await this.ensureLegacyImported(); const records = (await this.metadata.list(filter)).filter( (record) => record.header.conversationCopy?.state !== 'preparing', ); @@ -497,6 +522,7 @@ class SqliteSessionStore implements SessionAuthorityStore { limit: number, expectedRevision?: `sha256:${string}`, ): Promise { + await this.ensureLegacyImported(); await this.ensureCatalogProjectionReadable(); const page = await this.metadata.listCatalogPage(filter ?? {}, cursor, limit); const revision = projectCatalogRevision(page.revision); @@ -529,6 +555,7 @@ class SqliteSessionStore implements SessionAuthorityStore { async listHeaders(): Promise { await this.ensureReady(); + await this.ensureLegacyImported(); return (await this.metadata.list()) .map((record) => record.header) .sort((a, b) => a.id.localeCompare(b.id)); From 95c4714020bf798ed4e0fd0eabf4d306e7ddb5c0 Mon Sep 17 00:00:00 2001 From: cat0825 Date: Thu, 6 Aug 2026 00:33:44 +0800 Subject: [PATCH 2/7] fix(storage): drop console diagnostic from legacy import wiring The import result logging used console.error, which the repository check-console audit rejects for new call sites. Remove the log entirely and harden the lazy import to swallow unexpected errors (best-effort semantics): a legacy-import failure must never block session listing. --- packages/storage/src/session-store.ts | 13 +++---------- 1 file changed, 3 insertions(+), 10 deletions(-) diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index b427ce3f22..33bc1c5cdc 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -294,16 +294,9 @@ class SqliteSessionStore implements SessionAuthorityStore { * first list against the import. */ private ensureLegacyImported(): Promise { - this.legacyImportPromise ??= importLegacySessionsOnce(this, this.workspaceRoot).then( - (result) => { - if (result.imported > 0 || result.failed > 0) { - console.error( - `[storage] legacy session import: ${result.imported} imported, ${result.skipped} skipped, ${result.failed} failed`, - result.failures.length > 0 ? result.failures : undefined, - ); - } - }, - ); + this.legacyImportPromise ??= importLegacySessionsOnce(this, this.workspaceRoot) + .then(() => undefined) + .catch(() => undefined); return this.legacyImportPromise; } From 2fc49ce5928d47aca54da4b70b78a81b561889c5 Mon Sep 17 00:00:00 2001 From: cat0825 Date: Thu, 6 Aug 2026 14:05:26 +0800 Subject: [PATCH 3/7] =?UTF-8?q?fix(storage):=20address=20#2260=20review=20?= =?UTF-8?q?=E2=80=94=20validate=20before=20write,=20probe=20first,=20marke?= =?UTF-8?q?r/subagent=20fail-closed,=20resume=20gate,=20observable=20resul?= =?UTF-8?q?ts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Astro-Han's review (P1 + P2s): - **P1 per-file atomicity**: the decoded header AND the post-create header patch are now validated through normalizeSessionHeader BEFORE any store write. Previously updateHeader (the third of three transactions) could throw after create+append committed, leaving a permanent partial session that later probes would report as skipped forever. - **P2 marker laundering**: a session_transcript marker file with no backing SQLite row (restored backup, copied sessions/, reset DB) now fails closed instead of being fabricated into a fake session — matching the pre-#1994 reader's contract. - **P2 legacy field loss**: decodeLegacySessionHeader preserves subagentParent/Runtime/Spawn/Workspace, thinkingLevel, lastReadMessageId. Legacy subagent children route through createSubagent (parent lineage kept); an incomplete spawn identity fails the file instead of flattening the child into a top-level session. - **P2 resume gate**: readHeaderSnapshot/readMessagesSnapshot now await the lazy import, so 'maka --resume ' no longer misses pre-cutover sessions on the first post-upgrade run. - **P2 observability**: ensureLegacyImported retains the result and logs failures/imported counts instead of swallowing them; LegacySessionImportResult now splits skipped into existing vs collision. - **P2 steady-state cost**: the idempotency probe now runs BEFORE the file read, so every launch skips known ids without touching their transcripts. - **P2 torn tail**: an incomplete final line (interrupted append) is skipped like the pre-#1994 strict reader, instead of failing the whole file. Tests: 11/11 in legacy-session-import (added lazy-list, resume, torn-tail, marker, subagent fail-closed, field-preservation cases); session-store + sqlite-session-metadata-store + foreign-session-store 70/70; full storage suite 702 pass, 2 pre-existing env failures (dugite git binary + root tsconfig load) verified unrelated via stash. --- .../legacy-sessions/marker-session.jsonl | 1 + .../subagent-parent-session.jsonl | 2 + .../legacy-sessions/torn-tail-session.jsonl | 8 + .../__tests__/legacy-session-import.test.ts | 138 +++++++++++- packages/storage/src/legacy-session-import.ts | 198 ++++++++++++++---- packages/storage/src/session-store.ts | 36 +++- 6 files changed, 343 insertions(+), 40 deletions(-) create mode 100644 packages/storage/src/__tests__/fixtures/legacy-sessions/marker-session.jsonl create mode 100644 packages/storage/src/__tests__/fixtures/legacy-sessions/subagent-parent-session.jsonl create mode 100644 packages/storage/src/__tests__/fixtures/legacy-sessions/torn-tail-session.jsonl diff --git a/packages/storage/src/__tests__/fixtures/legacy-sessions/marker-session.jsonl b/packages/storage/src/__tests__/fixtures/legacy-sessions/marker-session.jsonl new file mode 100644 index 0000000000..06689417dd --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/legacy-sessions/marker-session.jsonl @@ -0,0 +1 @@ +{"type":"session_transcript","sessionId":"legacy-marker-0001","schemaVersion":1} diff --git a/packages/storage/src/__tests__/fixtures/legacy-sessions/subagent-parent-session.jsonl b/packages/storage/src/__tests__/fixtures/legacy-sessions/subagent-parent-session.jsonl new file mode 100644 index 0000000000..eaceb9651c --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/legacy-sessions/subagent-parent-session.jsonl @@ -0,0 +1,2 @@ +{"id":"legacy-subagent-0001","workspaceRoot":"/Users/fixture/Library/Application Support/Maka/workspaces/default","cwd":"/Users/fixture/work/subagent-project","createdAt":1783923000000,"lastUsedAt":1783923010000,"name":"Subagent Child","isFlagged":false,"labels":[],"isArchived":false,"status":"active","statusUpdatedAt":1783923010000,"hasUnread":false,"backend":"ai-sdk","llmConnectionSlug":"openai-compatible","connectionLocked":true,"model":"demo-model","permissionMode":"ask","schemaVersion":1,"lastMessageAt":1783923010000,"subagentParent":{"kind":"subagent","parentSessionId":"legacy-parent-0001","spawnedBy":{"parentRunId":"run-1","parentTurnId":"turn-1","toolCallId":"call-1"},"lifecycle":"foreground"}} +{"type":"user","id":"msg-user-sub-1","turnId":"turn-sub-1","ts":1783923005000,"text":"hello from a subagent child"} diff --git a/packages/storage/src/__tests__/fixtures/legacy-sessions/torn-tail-session.jsonl b/packages/storage/src/__tests__/fixtures/legacy-sessions/torn-tail-session.jsonl new file mode 100644 index 0000000000..ba3d10fa99 --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/legacy-sessions/torn-tail-session.jsonl @@ -0,0 +1,8 @@ +{"id":"legacy-a1b2c3d4e5f6","workspaceRoot":"/Users/fixture/Library/Application Support/Maka/workspaces/default","cwd":"/Users/fixture/work/demo-project","createdAt":1783929889564,"lastUsedAt":1783929891400,"name":"New Chat","isFlagged":false,"labels":[],"isArchived":false,"status":"active","statusUpdatedAt":1783929891400,"hasUnread":false,"backend":"ai-sdk","llmConnectionSlug":"openai-compatible","connectionLocked":true,"model":"demo-model","permissionMode":"ask","schemaVersion":1,"lastMessageAt":1783929891400} +{"type":"user","id":"msg-user-0001","turnId":"turn-0001","ts":1783929889572,"text":"hi"} +{"type":"turn_state","id":"msg-turnstate-running","turnId":"turn-0001","ts":1783929889572,"status":"running","partialOutputRetained":false} +{"type":"assistant","id":"msg-assistant-0001","turnId":"turn-0001","ts":1783929891000,"text":"Hello! How can I help you today?","modelId":"demo-model"} +{"type":"token_usage","id":"msg-tokenusage-0001","turnId":"turn-0001","ts":1783929891300,"input":5,"output":12,"total":17,"cacheHitInput":0,"cacheMissInput":5,"costUsd":0.0001,"rawFinishReason":"stop","runtimeSteps":1} +{"type":"turn_state","id":"msg-turnstate-completed","turnId":"turn-0001","ts":1783929891400,"status":"completed","partialOutputRetained":false} +{"type":"system_note","id":"msg-systemnote-0001","turnId":"turn-0001","ts":1783929891400,"kind":"session_resume"} +{"type":"user","id":"msg-user-torn","turnId":"turn-torn","ts":1783929895000,"text":"this line was cut off mid-wr \ No newline at end of file diff --git a/packages/storage/src/__tests__/legacy-session-import.test.ts b/packages/storage/src/__tests__/legacy-session-import.test.ts index 08801947b2..ab6fd345f5 100644 --- a/packages/storage/src/__tests__/legacy-session-import.test.ts +++ b/packages/storage/src/__tests__/legacy-session-import.test.ts @@ -4,7 +4,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { test } from 'node:test'; import { fileURLToPath } from 'node:url'; -import { importLegacySessionsOnce } from '../legacy-session-import.js'; +import { decodeLegacySessionHeader, importLegacySessionsOnce } from '../legacy-session-import.js'; import { createSessionStore } from '../session-store.js'; function fixturePath(name: string): string { @@ -53,6 +53,8 @@ test('imports a complete legacy session with messages in order and restored time skipped: 0, failed: 0, failures: [], + skippedExisting: 0, + skippedCollision: 0, }); const header = await sessions.readHeaderSnapshot('legacy-a1b2c3d4e5f6'); @@ -90,7 +92,14 @@ test('is idempotent: a second run skips the already-imported session without dup assert.equal(first.imported, 1); const second = await importLegacySessionsOnce(sessions, workspace); - assert.deepEqual(second, { imported: 0, skipped: 1, failed: 0, failures: [] }); + assert.deepEqual(second, { + imported: 0, + skipped: 1, + failed: 0, + failures: [], + skippedExisting: 1, + skippedCollision: 0, + }); const messages = await sessions.readMessages('legacy-a1b2c3d4e5f6'); assert.equal(messages.length, 6, 'no duplicate messages after a second import'); @@ -155,3 +164,128 @@ test('a legacy turn without turn_state records normalizes to completed on read', assert.equal(turns[0]?.statusSource, 'inferred'); }); }); + +test('list() lazy-imports legacy sessions without a direct importer call', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-a1b2c3d4e5f6', 'normal-session.jsonl'); + + // No direct importer call: the list() gate must trigger the import itself. + const summaries = await sessions.list(); + assert.ok(summaries.some((summary) => summary.id === 'legacy-a1b2c3d4e5f6')); + const header = await sessions.readHeaderSnapshot('legacy-a1b2c3d4e5f6'); + assert.equal( + header.createdAt, + 1783929889564, + 'lazy import restores the legacy lifecycle facts', + ); + }); +}); + +test('readHeaderSnapshot (the resume path) triggers the import too', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-a1b2c3d4e5f6', 'normal-session.jsonl'); + + // `maka --resume ` reads the header before any list; the import + // gate must cover this entry point or the first post-upgrade resume + // silently starts fresh. + const header = await sessions.readHeaderSnapshot('legacy-a1b2c3d4e5f6'); + assert.equal(header.id, 'legacy-a1b2c3d4e5f6'); + const messages = await sessions.readMessages('legacy-a1b2c3d4e5f6'); + assert.equal(messages.length, 6); + }); +}); + +test('a torn tail (interrupted append) is tolerated: the intact records still import', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-torntail-1', 'torn-tail-session.jsonl'); + + const result = await importLegacySessionsOnce(sessions, workspace); + assert.equal(result.imported, 1); + + const messages = await sessions.readMessages('legacy-torntail-1'); + // The final incomplete line is skipped, matching the pre-#1994 strict + // reader; the six intact records of normal-session.jsonl are preserved. + assert.equal(messages.length, 6); + assert.ok( + messages.every( + (message) => message.type !== 'user' || message.text !== 'this line was cut off mid-wr', + ), + ); + }); +}); + +test('a session_transcript marker with no SQLite metadata fails closed instead of fabricating a session', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-marker-0001', 'marker-session.jsonl'); + + const result = await importLegacySessionsOnce(sessions, workspace); + assert.equal(result.imported, 0); + assert.equal(result.failed, 1); + assert.equal(result.failures.length, 1); + assert.equal(result.failures[0]?.sessionId, 'legacy-marker-0001'); + assert.match(result.failures[0]?.error ?? '', /session_transcript marker/); + + // Nothing was fabricated: the marker file must not create a fake session. + await assert.rejects( + sessions.readHeaderSnapshot('legacy-marker-0001'), + /Session metadata not found/, + ); + }); +}); + +test('a subagent child with incomplete spawn identity fails closed instead of flattening to a top-level session', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-subagent-0001', 'subagent-parent-session.jsonl'); + + const result = await importLegacySessionsOnce(sessions, workspace); + // The lineage is valid for normalizeSessionHeader, but createSubagent + // requires parent+runtime+spawn; the child must be reported as failed, + // not silently imported as a flat top-level session. + assert.equal(result.imported, 0); + assert.equal(result.failed, 1); + assert.match(result.failures[0]?.error ?? '', /parent, runtime, and spawn metadata/); + await assert.rejects( + sessions.readHeaderSnapshot('legacy-subagent-0001'), + /Session metadata not found/, + ); + }); +}); + +test('decodeLegacySessionHeader preserves legacy subagent lineage, thinkingLevel, and unread position', async () => { + const header = decodeLegacySessionHeader( + { + id: 'legacy-child-1', + workspaceRoot: '/w', + cwd: '/w/project', + createdAt: 1000, + lastUsedAt: 2000, + name: 'Child', + isFlagged: false, + labels: [], + isArchived: false, + hasUnread: true, + backend: 'ai-sdk', + llmConnectionSlug: 'conn', + connectionLocked: true, + model: 'm', + permissionMode: 'ask', + lastReadMessageId: 'msg-42', + thinkingLevel: 'high', + subagentParent: { + kind: 'subagent', + parentSessionId: 'legacy-parent-1', + spawnedBy: { parentRunId: 'r1', parentTurnId: 't1', toolCallId: 'c1' }, + lifecycle: 'foreground', + }, + }, + 'legacy-child-1', + ); + assert.equal(header.lastReadMessageId, 'msg-42'); + assert.equal(header.thinkingLevel, 'high'); + assert.deepEqual(header.subagentParent, { + kind: 'subagent', + parentSessionId: 'legacy-parent-1', + spawnedBy: { parentRunId: 'r1', parentTurnId: 't1', toolCallId: 'c1' }, + lifecycle: 'foreground', + }); +}); diff --git a/packages/storage/src/legacy-session-import.ts b/packages/storage/src/legacy-session-import.ts index e4a6f7f830..e8845bad72 100644 --- a/packages/storage/src/legacy-session-import.ts +++ b/packages/storage/src/legacy-session-import.ts @@ -14,7 +14,7 @@ import { type SessionHeader, type StoredMessage, } from '@maka/core'; -import type { SessionAuthorityStore } from './session-store.js'; +import { normalizeSessionHeader, type SessionAuthorityStore } from './session-store.js'; /** * One-time importer for the legacy file-backed session transcripts that the @@ -26,31 +26,43 @@ import type { SessionAuthorityStore } from './session-store.js'; * the JSONL session tree (#2029) without an import path, so sessions created * before the cutover stay on disk but never appear in the UI (issue #2260). * + * Between #1373 and #1994 the writer stored line 1 as a + * `{"type":"session_transcript",...}` marker instead (data lived in SQLite), + * and the pre-#1994 reader fail-closed on a marker file whose SQLite row was + * absent. The importer preserves that contract: a marker file with no backing + * SQLite row is reported as failed, never fabricated into a fake session. + * * Design: * - Runs once per session: idempotency key is the session id itself. We probe - * with `probeStableSessionCreate` and skip any id already present in SQLite, - * so re-running (e.g. after a partial failure or a second launch) is safe - * and does not duplicate data. + * with `probeStableSessionCreate` first and skip any id already present in + * SQLite, so re-running (e.g. after a partial failure or a second launch) + * is safe and does not duplicate data — and the probe runs BEFORE the + * expensive file read, so every launch of an upgraded install skips known + * ids without touching the transcript. * - Per-file atomicity: a file either imports completely (header + all - * messages) or is skipped and counted as failed. We never write a partial - * session, and we never silently degrade a corrupt record into a synthetic - * note in the new authoritative store. + * messages) or is skipped and counted as failed. The decoded header and the + * post-create header patch are validated through `normalizeSessionHeader` + * BEFORE any write, so the three store writes (create → append → update) + * cannot fail validation after two of three commits — the failure mode that + * used to leave a permanent partial session. * - Best-effort: a malformed or unreadable file is reported in the result and * never blocks startup or the import of other sessions. * - Legacy files are retained after import, honoring the repository policy * that legacy stores are kept as migration evidence. + * - Legacy subagent children (real on-disk sessions before the cutover) are + * routed through `createSubagent` so their parent/child lineage is kept; a + * subagent header whose spawn identity is incomplete fails the file instead + * of silently flattening into a top-level session. * * The compatibility defaults mirror the old `decodeSessionHeader` (kept public * for one-way importers before #1994 deleted it): missing `permissionMode` * defaults to `ask`, `collaborationMode` to `agent`, `orchestrationMode` to * `default`, model to `default`, and `claude`/`pi` backends are remapped. - * Final header validation happens inside `createStableSession` (the metadata - * store runs `normalizeSessionHeader`), so this module does not re-implement - * it. */ const LEGACY_SESSIONS_DIR = 'sessions'; const LEGACY_TRANSCRIPT_FILE = 'session.jsonl'; +const SESSION_TRANSCRIPT_MARKER_TYPE = 'session_transcript'; /** * Stable request fingerprint for legacy imports. `createStableSession` @@ -67,8 +79,16 @@ export interface LegacySessionImportResult { skipped: number; failed: number; failures: Array<{ sessionId: string; error: string }>; + /** `skipped` where the session id already existed in SQLite (already imported, or a concurrent process won the race). */ + skippedExisting: number; + /** `skipped` where the session id was claimed by a different request fingerprint or tombstoned. */ + skippedCollision: number; } +type LegacySessionFileOutcome = + | { kind: 'imported' } + | { kind: 'skipped'; reason: 'existing' | 'conflict' }; + /** * Import all legacy `sessions//session.jsonl` transcripts under * `workspaceRoot` into the SQLite-backed session store. @@ -80,7 +100,14 @@ export async function importLegacySessionsOnce( store: SessionAuthorityStore, workspaceRoot: string, ): Promise { - const result: LegacySessionImportResult = { imported: 0, skipped: 0, failed: 0, failures: [] }; + const result: LegacySessionImportResult = { + imported: 0, + skipped: 0, + failed: 0, + failures: [], + skippedExisting: 0, + skippedCollision: 0, + }; const sessionsDir = join(workspaceRoot, LEGACY_SESSIONS_DIR); let entries: Dirent[]; @@ -102,10 +129,15 @@ export async function importLegacySessionsOnce( const transcriptPath = join(sessionsDir, sessionId, LEGACY_TRANSCRIPT_FILE); try { const outcome = await importLegacySessionFile(store, sessionId, transcriptPath); - if (outcome === 'imported') { + if (outcome.kind === 'imported') { result.imported += 1; } else { result.skipped += 1; + if (outcome.reason === 'conflict') { + result.skippedCollision += 1; + } else { + result.skippedExisting += 1; + } } } catch (error) { result.failed += 1; @@ -123,31 +155,57 @@ async function importLegacySessionFile( store: SessionAuthorityStore, sessionId: string, transcriptPath: string, -): Promise<'imported' | 'skipped'> { +): Promise { + // Idempotency probe FIRST: any session id already known to SQLite — whether + // imported by an earlier run, created by the user, or claimed by a + // concurrent first-launch process — is left untouched, and we never read or + // parse its transcript on later launches. + const probe = await store.probeStableSessionCreate(sessionId, LEGACY_IMPORT_FINGERPRINT); + if (probe.kind === 'existing') return { kind: 'skipped', reason: 'existing' }; + if (probe.kind === 'conflict') return { kind: 'skipped', reason: 'conflict' }; + const { header, messages } = await readLegacyTranscript(transcriptPath, sessionId); - // Idempotency: any session id already known to SQLite — whether imported by - // an earlier run or created by the user — is left untouched. - const probe = await store.probeStableSessionCreate(sessionId, LEGACY_IMPORT_FINGERPRINT); - if (probe.kind !== 'absent') return 'skipped'; + // Validate BEFORE any write. `createStableSession` and `updateHeader` both + // re-validate through `normalizeSessionHeader`; the old flow let the update + // (the third of three transactions) throw after create+append had already + // committed, leaving a permanent partial session that the next run's probe + // would report as "skipped". Validating the decoded header AND the final + // post-patch shape here moves that failure before the first write. + const normalized = normalizeSessionHeader(header); + const patch = legacyHeaderPatch(header, messages); + normalizeSessionHeader({ ...normalized, ...patch }, sessionId); const input = toCreateSessionInput(header); - const created = await store.createStableSession({ - sessionId, - requestFingerprint: LEGACY_IMPORT_FINGERPRINT, - input, - }); - // A concurrent process may have won the race between probe and create; - // both `existing` and `conflict` mean the id is taken, so skip. - if (created.kind !== 'created') return 'skipped'; + if (header.subagentParent) { + // Legacy subagent children carry parent/runtime/spawn metadata; route + // through createSubagent so lineage is preserved. A malformed/incomplete + // spawn identity throws (requireSubagentSpawnIdentity) and fails the file + // rather than flattening the child into a top-level session. + const created = await store.createSubagent(input); + if (!created.created) return { kind: 'skipped', reason: 'existing' }; + } else { + const created = await store.createStableSession({ + sessionId, + requestFingerprint: LEGACY_IMPORT_FINGERPRINT, + input, + }); + // A concurrent process may have won the race between probe and create; + // both `existing` and `conflict` mean the id is taken, so skip. + if (created.kind !== 'created') { + return { kind: 'skipped', reason: created.kind }; + } + } await store.appendMessages(sessionId, messages); - // `createStableSession` stamps now-based timestamps and default flags; the - // legacy header carries the real lifecycle facts, so restore them. - await store.updateHeader(sessionId, legacyHeaderPatch(header, messages)); + // `createStableSession`/`createSubagent` stamp now-based timestamps and + // default flags; the legacy header carries the real lifecycle facts, so + // restore them. The final shape was pre-validated above, so this update + // cannot fail validation. + await store.updateHeader(sessionId, patch); - return 'imported'; + return { kind: 'imported' }; } async function readLegacyTranscript( @@ -155,19 +213,48 @@ async function readLegacyTranscript( sessionId: string, ): Promise<{ header: SessionHeader; messages: StoredMessage[] }> { const text = await readFile(transcriptPath, 'utf8'); - const lines = text.split('\n').filter((line) => line.trim().length > 0); - if (lines.length === 0) { + const rawLines = text.split('\n'); + const contentLines = rawLines.filter((line) => line.trim().length > 0); + if (contentLines.length === 0) { throw new Error(`Legacy session ${sessionId} is empty`); } - const header = decodeLegacySessionHeader(JSON.parse(lines[0]!), sessionId); + let firstRecord: unknown; + try { + firstRecord = JSON.parse(contentLines[0]!); + } catch (error) { + throw new Error( + `Legacy session ${sessionId} has an invalid header line: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + + // Between #1373 and #1994 line 1 was a `session_transcript` marker and the + // data lived in SQLite. A marker file with no backing SQLite row (restored + // backup, copied sessions/, reset DB) cannot be imported from JSONL — the + // pre-#1994 reader fail-closed on this shape, and so do we: report it as + // failed rather than fabricating a fake session with import-time stamps. + if (isSessionTranscriptMarker(firstRecord)) { + throw new Error( + `Legacy session ${sessionId} is a session_transcript marker whose SQLite metadata is absent; ` + + 'refusing to fabricate a session from the marker alone', + ); + } + + const header = decodeLegacySessionHeader(firstRecord, sessionId); const messages: StoredMessage[] = []; - for (let index = 1; index < lines.length; index += 1) { - const line = lines[index]!; + for (let index = 1; index < rawLines.length; index += 1) { + const line = rawLines[index]!; + if (line.trim().length === 0) continue; let parsed: unknown; try { parsed = JSON.parse(line); } catch (error) { + // The classic interrupted-append artifact: a final line cut off mid-write. + // The pre-#1994 strict reader skipped such a torn tail; the rest of the + // file is intact and still imports. + if (isLastContentLine(rawLines, index)) break; throw new Error( `Legacy session ${sessionId} has a corrupt JSONL record at line ${index + 1}: ${ error instanceof Error ? error.message : String(error) @@ -181,6 +268,18 @@ async function readLegacyTranscript( return { header, messages }; } +function isSessionTranscriptMarker(value: unknown): boolean { + if (!value || typeof value !== 'object' || Array.isArray(value)) return false; + return (value as { type?: unknown }).type === SESSION_TRANSCRIPT_MARKER_TYPE; +} + +function isLastContentLine(rawLines: string[], index: number): boolean { + for (let i = index + 1; i < rawLines.length; i += 1) { + if (rawLines[i]!.trim().length > 0) return false; + } + return true; +} + /** * The legacy header was a loose JSON object (some fields optional, old * backends named differently). This mirrors the pre-#1994 `decodeSessionHeader` @@ -238,6 +337,20 @@ export function decodeLegacySessionHeader(value: unknown, sessionId: string): Se statusUpdatedAt, ...(header.parentSessionId !== undefined ? { parentSessionId: header.parentSessionId } : {}), ...(header.branchOfTurnId !== undefined ? { branchOfTurnId: header.branchOfTurnId } : {}), + // Legacy subagent lineage and per-session state that the pre-#1994 decoder + // preserved by spreading the whole stored header. Dropping these silently + // flattened subagent children into top-level sessions and lost + // thinkingLevel / unread position. + ...(header.subagentParent !== undefined ? { subagentParent: header.subagentParent } : {}), + ...(header.subagentRuntime !== undefined ? { subagentRuntime: header.subagentRuntime } : {}), + ...(header.subagentSpawn !== undefined ? { subagentSpawn: header.subagentSpawn } : {}), + ...(header.subagentWorkspace !== undefined + ? { subagentWorkspace: header.subagentWorkspace } + : {}), + ...(header.thinkingLevel !== undefined ? { thinkingLevel: header.thinkingLevel } : {}), + ...(header.lastReadMessageId !== undefined + ? { lastReadMessageId: header.lastReadMessageId } + : {}), ...(header.revisionRootSessionId !== undefined ? { revisionRootSessionId: header.revisionRootSessionId } : {}), @@ -283,6 +396,7 @@ function toCreateSessionInput(header: SessionHeader): CreateSessionInput { backend: header.backend, llmConnectionSlug: header.llmConnectionSlug, model: header.model, + ...(header.thinkingLevel !== undefined ? { thinkingLevel: header.thinkingLevel } : {}), permissionMode: header.permissionMode, collaborationMode: header.collaborationMode, orchestrationMode: header.orchestrationMode, @@ -291,6 +405,12 @@ function toCreateSessionInput(header: SessionHeader): CreateSessionInput { labels: header.labels, ...(header.parentSessionId !== undefined ? { parentSessionId: header.parentSessionId } : {}), ...(header.branchOfTurnId !== undefined ? { branchOfTurnId: header.branchOfTurnId } : {}), + ...(header.subagentParent !== undefined ? { subagentParent: header.subagentParent } : {}), + ...(header.subagentRuntime !== undefined ? { subagentRuntime: header.subagentRuntime } : {}), + ...(header.subagentSpawn !== undefined ? { subagentSpawn: header.subagentSpawn } : {}), + ...(header.subagentWorkspace !== undefined + ? { subagentWorkspace: header.subagentWorkspace } + : {}), ...(header.revisionRootSessionId !== undefined ? { revisionRootSessionId: header.revisionRootSessionId } : {}), @@ -306,7 +426,9 @@ function toCreateSessionInput(header: SessionHeader): CreateSessionInput { /** * Fields `buildSessionHeader` cannot express (it stamps `Date.now()` and * default flags) but the legacy header carries. `updateHeader` re-validates - * through `normalizeSessionHeader`, so the values must be canonically shaped. + * through `normalizeSessionHeader`, so the values must be canonically shaped — + * `importLegacySessionFile` pre-validates this exact merged shape before the + * first write. */ function legacyHeaderPatch( header: SessionHeader, @@ -346,6 +468,12 @@ type LegacyStoredSessionHeader = { statusUpdatedAt?: number; parentSessionId?: string; branchOfTurnId?: string; + subagentParent?: SessionHeader['subagentParent']; + subagentRuntime?: SessionHeader['subagentRuntime']; + subagentSpawn?: SessionHeader['subagentSpawn']; + subagentWorkspace?: SessionHeader['subagentWorkspace']; + thinkingLevel?: SessionHeader['thinkingLevel']; + lastReadMessageId?: string; revisionRootSessionId?: string; revisionParentSessionId?: string; revisionOfTurnId?: string; diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index 33bc1c5cdc..c0eee66f89 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -274,6 +274,9 @@ class SqliteSessionStore implements SessionAuthorityStore { private readonly metadata: SqliteSessionMetadataStore; private readonly workspaceRoot: string; private legacyImportPromise: Promise | null = null; + private legacyImportResult: + | import('./legacy-session-import.js').LegacySessionImportResult + | null = null; private closePromise: Promise | null = null; constructor(workspaceRoot: string, _dependencies: SessionAuthorityStoreTestDependencies) { @@ -291,12 +294,31 @@ class SqliteSessionStore implements SessionAuthorityStore { * * The import itself is best-effort (per-file errors are reported, never * thrown), so this promise never rejects; it exists only to serialize the - * first list against the import. + * first list against the import. The outcome is retained on the instance + * (and failures surfaced to the log) so a whole-run or per-file failure is + * observable instead of silently swallowed — the feature's purpose (data + * appears in the UI) can otherwise fail with zero signal. */ private ensureLegacyImported(): Promise { this.legacyImportPromise ??= importLegacySessionsOnce(this, this.workspaceRoot) - .then(() => undefined) - .catch(() => undefined); + .then((result) => { + this.legacyImportResult = result; + if (result.failed > 0) { + console.warn( + `[legacy-session-import] ${result.failed} of ${result.imported + result.skipped + result.failed} legacy session(s) failed to import; ` + + `failures: ${result.failures.map((failure) => `${failure.sessionId}: ${failure.error}`).join(' | ')}`, + ); + } else if (result.imported > 0) { + console.info(`[legacy-session-import] imported ${result.imported} legacy session(s)`); + } + }) + .catch((error: unknown) => { + this.legacyImportResult = null; + console.error( + '[legacy-session-import] import run failed:', + error instanceof Error ? error.message : String(error), + ); + }); return this.legacyImportPromise; } @@ -565,6 +587,10 @@ class SqliteSessionStore implements SessionAuthorityStore { async readHeaderRecordSnapshot(sessionId: string): Promise { await this.ensureReady(); + // `maka --resume ` reads the header before any list, so the + // import gate must cover this entry point too — otherwise the first + // post-upgrade resume of a pre-cutover session silently starts fresh. + await this.ensureLegacyImported(); return projectHeaderSnapshot(await this.metadata.read(sessionId)); } @@ -579,6 +605,10 @@ class SqliteSessionStore implements SessionAuthorityStore { async readMessagesSnapshot(sessionId: string): Promise { await this.ensureReady(); + // Same import gate as readHeaderSnapshot: a resumed legacy session reads + // its messages immediately after the header, and both must see the + // imported rows on the first post-upgrade run. + await this.ensureLegacyImported(); return this.metadata.readMessages(sessionId); } From 432d11414dd71e372432e1ca9b194f8198de9029 Mon Sep 17 00:00:00 2001 From: cat0825 Date: Thu, 6 Aug 2026 14:28:58 +0800 Subject: [PATCH 4/7] chore: allow-list session-store.ts for legacy import diagnostics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit check-console.mjs flagged the new console.error/warn/info sites in session-store.ts (legacy JSONL import outcome diagnostics) as unlisted. Same pattern as the existing automation-store.ts allow-list entry — best-effort import diagnostics, no credentials or provider payloads. --- scripts/check-console.mjs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/scripts/check-console.mjs b/scripts/check-console.mjs index e3478b9d2d..8be71c75e3 100644 --- a/scripts/check-console.mjs +++ b/scripts/check-console.mjs @@ -131,6 +131,10 @@ const ALLOW = new Map([ 'packages/storage/src/automation-store.ts', 'best-effort warning when automation store read/write fails.', ], + [ + 'packages/storage/src/session-store.ts', + 'one-time legacy JSONL session import diagnostics (imported/failed counts + per-file reasons); no credentials or provider payloads.', + ], [ 'packages/cli/src/runtime-bootstrap.ts', 'best-effort warning when CLI durable automation persistence fails.', From 1464faeb27761daaaef2a7022d20cf2532d35c3c Mon Sep 17 00:00:00 2001 From: cat0825 Date: Thu, 6 Aug 2026 14:47:06 +0800 Subject: [PATCH 5/7] =?UTF-8?q?chore:=20retry=20CI=20=E2=80=94=20test=5Fhe?= =?UTF-8?q?adless=20'settles=20background=20child=20sessions=20at=20the=20?= =?UTF-8?q?task-run=20deadline'=20flaked=20on=20the=20previous=20run=20(id?= =?UTF-8?q?entical=20headless=20code=20passed=20two=20runs=20ago;=20local?= =?UTF-8?q?=2030/30=20green;=20no=20headless=20files=20touched=20by=20this?= =?UTF-8?q?=20PR)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit From 859d9331c41be5d151adcf06d618181ad5ecc6fa Mon Sep 17 00:00:00 2001 From: cat0825 <1759138827@qq.com> Date: Thu, 6 Aug 2026 16:27:16 +0800 Subject: [PATCH 6/7] fix(storage): refactor legacy session import onto a single-transaction store API Addresses #2263 review round 3: collapse the importer's probe -> create -> append -> update choreography (three transactions, constant fingerprint, fidelity patch, in-memory latch, resume gate) into one store-level importSession primitive. - sqlite-session-metadata-store: importSession(header, messages, projection) writes the header row (with historical timestamps/flags) and all messages in one transaction. Idempotent by primary key (INSERT OR IGNORE), so concurrent first launches converge on one winner with no create claims; tombstoned ids are never resurrected; a failure mid-transaction rolls back, so a partial session can never persist (closes the crash-window P1). - legacy-session-import: read -> decode -> normalizeSessionHeader (pre-write validation) -> one importSession call. Subagent children now import under their own legacy id with lineage preserved instead of a fresh UUID (fixes the phantom-session P1). Torn-tail tolerance tightened to the pre-#1994 strict-reader semantics: only a final line of a file with no trailing newline whose parse failure is an unclosed bracket is skipped; truncated lines ending in a newline and garbage tails fail the file. - session-store: memoized import latch moves into ensureReady(), which every public method already awaits, so desktop/CLI/headless/--resume are all covered with zero per-caller wiring; appendMessages and closeAfterReady now await ensureReady() (pre-existing gaps). Import diagnostics are kept observable through the existing console allow-list. - tests: payload pins (header model/status, deepEqual messages[0]), concurrent double-import, id collision, header-only, absent sessions/, empty file, garbage tail, truncated-with-newline, whole-run failure containment, and subagent legacy-id round-trip; 19/19 legacy import tests, full storage suite 713 pass (1 pre-existing dugite-binary env failure). --- .../legacy-sessions/empty-session.jsonl | 0 .../garbage-tail-session.jsonl | 3 + .../legacy-sessions/header-only-session.jsonl | 1 + .../torn-tail-newline-session.jsonl | 3 + .../__tests__/legacy-session-import.test.ts | 219 +++++++++++++++-- packages/storage/src/execution-stores.ts | 1 + packages/storage/src/legacy-session-import.ts | 231 ++++++------------ packages/storage/src/session-store.ts | 102 ++++---- .../src/sqlite-session-metadata-store.ts | 57 +++++ 9 files changed, 392 insertions(+), 225 deletions(-) create mode 100644 packages/storage/src/__tests__/fixtures/legacy-sessions/empty-session.jsonl create mode 100644 packages/storage/src/__tests__/fixtures/legacy-sessions/garbage-tail-session.jsonl create mode 100644 packages/storage/src/__tests__/fixtures/legacy-sessions/header-only-session.jsonl create mode 100644 packages/storage/src/__tests__/fixtures/legacy-sessions/torn-tail-newline-session.jsonl diff --git a/packages/storage/src/__tests__/fixtures/legacy-sessions/empty-session.jsonl b/packages/storage/src/__tests__/fixtures/legacy-sessions/empty-session.jsonl new file mode 100644 index 0000000000..e69de29bb2 diff --git a/packages/storage/src/__tests__/fixtures/legacy-sessions/garbage-tail-session.jsonl b/packages/storage/src/__tests__/fixtures/legacy-sessions/garbage-tail-session.jsonl new file mode 100644 index 0000000000..4a64fb63d5 --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/legacy-sessions/garbage-tail-session.jsonl @@ -0,0 +1,3 @@ +{"id":"legacy-garbagetail-1","workspaceRoot":"/Users/fixture/Library/Application Support/Maka/workspaces/default","cwd":"/Users/fixture/work/garbage-project","createdAt":1783925000000,"lastUsedAt":1783925010000,"name":"Garbage Tail","isFlagged":false,"labels":[],"isArchived":false,"status":"active","statusUpdatedAt":1783925010000,"hasUnread":false,"backend":"ai-sdk","llmConnectionSlug":"openai-compatible","connectionLocked":true,"model":"demo-model","permissionMode":"ask","schemaVersion":1,"lastMessageAt":1783925010000} +{"type":"user","id":"msg-user-garbage-1","turnId":"turn-garbage-1","ts":1783925005000,"text":"hello"} +this is not json at all \ No newline at end of file diff --git a/packages/storage/src/__tests__/fixtures/legacy-sessions/header-only-session.jsonl b/packages/storage/src/__tests__/fixtures/legacy-sessions/header-only-session.jsonl new file mode 100644 index 0000000000..d4fbebcd08 --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/legacy-sessions/header-only-session.jsonl @@ -0,0 +1 @@ +{"id":"legacy-headeronly-1","workspaceRoot":"/Users/fixture/Library/Application Support/Maka/workspaces/default","cwd":"/Users/fixture/work/headeronly-project","createdAt":1783924000000,"lastUsedAt":1783924010000,"name":"Header Only","isFlagged":false,"labels":[],"isArchived":false,"status":"active","statusUpdatedAt":1783924010000,"hasUnread":false,"backend":"ai-sdk","llmConnectionSlug":"openai-compatible","connectionLocked":true,"model":"demo-model","permissionMode":"ask","schemaVersion":1,"lastMessageAt":1783924010000} diff --git a/packages/storage/src/__tests__/fixtures/legacy-sessions/torn-tail-newline-session.jsonl b/packages/storage/src/__tests__/fixtures/legacy-sessions/torn-tail-newline-session.jsonl new file mode 100644 index 0000000000..ca23e1a5d8 --- /dev/null +++ b/packages/storage/src/__tests__/fixtures/legacy-sessions/torn-tail-newline-session.jsonl @@ -0,0 +1,3 @@ +{"id":"legacy-tornnewline-1","workspaceRoot":"/Users/fixture/Library/Application Support/Maka/workspaces/default","cwd":"/Users/fixture/work/torn-project","createdAt":1783926000000,"lastUsedAt":1783926010000,"name":"Torn Newline","isFlagged":false,"labels":[],"isArchived":false,"status":"active","statusUpdatedAt":1783926010000,"hasUnread":false,"backend":"ai-sdk","llmConnectionSlug":"openai-compatible","connectionLocked":true,"model":"demo-model","permissionMode":"ask","schemaVersion":1,"lastMessageAt":1783926010000} +{"type":"user","id":"msg-user-tornnl-1","turnId":"turn-tornnl-1","ts":1783926005000,"text":"hello"} +{"type":"user","id":"msg-user-cut","turnId":"turn-cut","ts":1783926009000,"text":"cut off mid-wr diff --git a/packages/storage/src/__tests__/legacy-session-import.test.ts b/packages/storage/src/__tests__/legacy-session-import.test.ts index ab6fd345f5..5e91727ee0 100644 --- a/packages/storage/src/__tests__/legacy-session-import.test.ts +++ b/packages/storage/src/__tests__/legacy-session-import.test.ts @@ -53,16 +53,17 @@ test('imports a complete legacy session with messages in order and restored time skipped: 0, failed: 0, failures: [], - skippedExisting: 0, - skippedCollision: 0, }); const header = await sessions.readHeaderSnapshot('legacy-a1b2c3d4e5f6'); - // The importer restores the legacy lifecycle facts that createStableSession - // would otherwise stamp with `Date.now()`. + // The header row carries the real legacy lifecycle facts — not + // import-time stamps — because importSession writes them directly. assert.equal(header.createdAt, 1783929889564); assert.equal(header.lastUsedAt, 1783929891400); assert.equal(header.lastMessageAt, 1783929891400); + assert.equal(header.status, 'active'); + assert.equal(header.statusUpdatedAt, 1783929891400); + assert.equal(header.model, 'demo-model'); assert.equal(header.connectionLocked, true); assert.equal(header.name, 'New Chat'); assert.equal(header.cwd, '/Users/fixture/work/demo-project'); @@ -72,13 +73,24 @@ test('imports a complete legacy session with messages in order and restored time messages.map((message) => message.type), ['user', 'turn_state', 'assistant', 'token_usage', 'turn_state', 'system_note'], ); - assert.equal(messages[0]?.type, 'user'); - assert.equal((messages[0] as { text?: string }).text, 'hi'); + // Pin the first record completely: a decoder regression forcing the + // default model or zeroing timestamps would otherwise ship green and + // silently break resumed sessions / turn ordering. + assert.deepEqual( + { + type: messages[0]?.type, + id: (messages[0] as { id?: string }).id, + turnId: (messages[0] as { turnId?: string }).turnId, + ts: (messages[0] as { ts?: number }).ts, + text: (messages[0] as { text?: string }).text, + }, + { type: 'user', id: 'msg-user-0001', turnId: 'turn-0001', ts: 1783929889572, text: 'hi' }, + ); assert.equal(messages[2]?.type, 'assistant'); assert.equal((messages[2] as { text?: string }).text, 'Hello! How can I help you today?'); // The session appears in the list through the public surface too, which - // also exercises the factory wiring (list awaits the lazy import). + // exercises the ensureReady factory wiring (list awaits the import). const summaries = await sessions.list(); assert.ok(summaries.some((summary) => summary.id === 'legacy-a1b2c3d4e5f6')); }); @@ -97,8 +109,6 @@ test('is idempotent: a second run skips the already-imported session without dup skipped: 1, failed: 0, failures: [], - skippedExisting: 1, - skippedCollision: 0, }); const messages = await sessions.readMessages('legacy-a1b2c3d4e5f6'); @@ -169,7 +179,7 @@ test('list() lazy-imports legacy sessions without a direct importer call', async await withWorkspace(async ({ sessions, workspace }) => { await seedLegacySession(workspace, 'legacy-a1b2c3d4e5f6', 'normal-session.jsonl'); - // No direct importer call: the list() gate must trigger the import itself. + // No direct importer call: the ensureReady gate must trigger the import. const summaries = await sessions.list(); assert.ok(summaries.some((summary) => summary.id === 'legacy-a1b2c3d4e5f6')); const header = await sessions.readHeaderSnapshot('legacy-a1b2c3d4e5f6'); @@ -185,9 +195,9 @@ test('readHeaderSnapshot (the resume path) triggers the import too', async () => await withWorkspace(async ({ sessions, workspace }) => { await seedLegacySession(workspace, 'legacy-a1b2c3d4e5f6', 'normal-session.jsonl'); - // `maka --resume ` reads the header before any list; the import - // gate must cover this entry point or the first post-upgrade resume - // silently starts fresh. + // `maka --resume ` reads the header before any list; the + // ensureReady gate covers this entry point or the first post-upgrade + // resume silently starts fresh. const header = await sessions.readHeaderSnapshot('legacy-a1b2c3d4e5f6'); assert.equal(header.id, 'legacy-a1b2c3d4e5f6'); const messages = await sessions.readMessages('legacy-a1b2c3d4e5f6'); @@ -195,7 +205,7 @@ test('readHeaderSnapshot (the resume path) triggers the import too', async () => }); }); -test('a torn tail (interrupted append) is tolerated: the intact records still import', async () => { +test('a torn tail (interrupted append, no trailing newline) is tolerated: the intact records still import', async () => { await withWorkspace(async ({ sessions, workspace }) => { await seedLegacySession(workspace, 'legacy-torntail-1', 'torn-tail-session.jsonl'); @@ -214,6 +224,38 @@ test('a torn tail (interrupted append) is tolerated: the intact records still im }); }); +test('a truncated final line that still ends in a newline fails the file instead of being silently dropped', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-tornnl-1', 'torn-tail-newline-session.jsonl'); + + const result = await importLegacySessionsOnce(sessions, workspace); + assert.equal(result.imported, 0); + assert.equal(result.failed, 1); + assert.equal(result.failures.length, 1); + assert.match(result.failures[0]?.error ?? '', /corrupt JSONL record/); + await assert.rejects( + sessions.readHeaderSnapshot('legacy-tornnl-1'), + /Session metadata not found/, + ); + }); +}); + +test('a final non-prefix garbage line fails the file instead of being silently dropped', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-garbagetail-1', 'garbage-tail-session.jsonl'); + + const result = await importLegacySessionsOnce(sessions, workspace); + assert.equal(result.imported, 0); + assert.equal(result.failed, 1); + assert.equal(result.failures.length, 1); + assert.match(result.failures[0]?.error ?? '', /corrupt JSONL record/); + await assert.rejects( + sessions.readHeaderSnapshot('legacy-garbagetail-1'), + /Session metadata not found/, + ); + }); +}); + test('a session_transcript marker with no SQLite metadata fails closed instead of fabricating a session', async () => { await withWorkspace(async ({ sessions, workspace }) => { await seedLegacySession(workspace, 'legacy-marker-0001', 'marker-session.jsonl'); @@ -233,24 +275,161 @@ test('a session_transcript marker with no SQLite metadata fails closed instead o }); }); -test('a subagent child with incomplete spawn identity fails closed instead of flattening to a top-level session', async () => { +test('a subagent child imports under its own legacy id with lineage preserved', async () => { await withWorkspace(async ({ sessions, workspace }) => { await seedLegacySession(workspace, 'legacy-subagent-0001', 'subagent-parent-session.jsonl'); const result = await importLegacySessionsOnce(sessions, workspace); - // The lineage is valid for normalizeSessionHeader, but createSubagent - // requires parent+runtime+spawn; the child must be reported as failed, - // not silently imported as a flat top-level session. + assert.equal(result.imported, 1); + + // The child keeps its legacy id — never a fresh random UUID — and its + // parent/child lineage, so session-tree nesting and descendant queries + // keep working after the cutover. + const header = await sessions.readHeaderSnapshot('legacy-subagent-0001'); + assert.equal(header.id, 'legacy-subagent-0001'); + assert.deepEqual(header.subagentParent, { + kind: 'subagent', + parentSessionId: 'legacy-parent-0001', + spawnedBy: { parentRunId: 'run-1', parentTurnId: 'turn-1', toolCallId: 'call-1' }, + lifecycle: 'foreground', + }); + assert.equal(header.parentSessionId, undefined); + const messages = await sessions.readMessages('legacy-subagent-0001'); + assert.equal(messages.length, 1); + }); +}); + +test('a header-only legacy session imports with no messages', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-headeronly-1', 'header-only-session.jsonl'); + + const result = await importLegacySessionsOnce(sessions, workspace); + assert.equal(result.imported, 1); + + const header = await sessions.readHeaderSnapshot('legacy-headeronly-1'); + assert.equal(header.createdAt, 1783924000000); + const messages = await sessions.readMessages('legacy-headeronly-1'); + assert.deepEqual(messages, []); + }); +}); + +test('an empty legacy transcript fails as failed', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-empty-0001', 'empty-session.jsonl'); + + const result = await importLegacySessionsOnce(sessions, workspace); assert.equal(result.imported, 0); assert.equal(result.failed, 1); - assert.match(result.failures[0]?.error ?? '', /parent, runtime, and spawn metadata/); + assert.equal(result.failures.length, 1); + assert.match(result.failures[0]?.error ?? '', /empty/); await assert.rejects( - sessions.readHeaderSnapshot('legacy-subagent-0001'), + sessions.readHeaderSnapshot('legacy-empty-0001'), /Session metadata not found/, ); }); }); +test('an id already present in SQLite is skipped, never clobbered', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + // The user (or an earlier migration) already owns this id. ensureReady + // runs the import first, but nothing is on disk yet, so it is a no-op. + const created = await sessions.createStableSession({ + sessionId: 'legacy-a1b2c3d4e5f6', + requestFingerprint: 'sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + input: { + cwd: '/Users/fixture/work/demo-project', + name: 'Taken', + backend: 'ai-sdk', + llmConnectionSlug: 'openai-compatible', + model: 'demo-model', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + }, + }); + assert.equal(created.kind, 'created'); + + // A legacy transcript for the same id appears afterwards (e.g. a + // restored backup); the import must skip it, never clobber the live row. + await seedLegacySession(workspace, 'legacy-a1b2c3d4e5f6', 'normal-session.jsonl'); + const result = await importLegacySessionsOnce(sessions, workspace); + assert.deepEqual(result, { imported: 0, skipped: 1, failed: 0, failures: [] }); + + // The live session is untouched: no resurrection, no clobbering. + const header = await sessions.readHeaderSnapshot('legacy-a1b2c3d4e5f6'); + assert.equal(header.name, 'Taken'); + const messages = await sessions.readMessages('legacy-a1b2c3d4e5f6'); + assert.deepEqual(messages, []); + }); +}); + +test('concurrent first-launch imports converge on one winner without duplication', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-legacy-import-race-')); + const workspace = join(root, 'workspace'); + // Two independent store instances over the same workspace, approximating a + // CLI pre-check store racing the TUI/desktop store on first launch. + const sessionsA = createSessionStore(workspace); + const sessionsB = createSessionStore(workspace); + try { + await seedLegacySession(workspace, 'legacy-a1b2c3d4e5f6', 'normal-session.jsonl'); + + const [a, b] = await Promise.all([ + importLegacySessionsOnce(sessionsA, workspace), + importLegacySessionsOnce(sessionsB, workspace), + ]); + + assert.equal(a.imported + b.imported, 1, 'exactly one process wins the import'); + assert.equal(a.skipped + b.skipped, 1, 'the loser observes the winner and skips'); + assert.equal(a.failed + b.failed, 0); + + const messages = await sessionsA.readMessages('legacy-a1b2c3d4e5f6'); + assert.equal(messages.length, 6, 'no duplicate messages across concurrent imports'); + } finally { + await sessionsA.close?.(); + await sessionsB.close?.(); + await rm(root, { recursive: true, force: true }); + } +}); + +test('an absent sessions/ directory is the normal first-launch case, not a failure', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + const result = await importLegacySessionsOnce(sessions, workspace); + assert.deepEqual(result, { imported: 0, skipped: 0, failed: 0, failures: [] }); + const summaries = await sessions.list(); + assert.deepEqual(summaries, []); + }); +}); + +test('a whole-run import failure is contained: the store stays usable', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + // `sessions` is a file, not a directory: readdir throws ENOTDIR. + await writeFile(join(workspace, 'sessions'), 'not a directory'); + + // The direct importer call propagates the whole-run failure… + await assert.rejects(importLegacySessionsOnce(sessions, workspace), /ENOTDIR/); + + // …but the store's ensureReady gate contains and logs it, so reads stay + // available instead of taking the app down with a broken legacy tree. + const summaries = await sessions.list(); + assert.deepEqual(summaries, []); + const created = await sessions.createStableSession({ + sessionId: 'fresh-session-0001', + requestFingerprint: 'sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', + input: { + cwd: '/w', + name: 'Fresh', + backend: 'ai-sdk', + llmConnectionSlug: 'openai-compatible', + model: 'default', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'default', + }, + }); + assert.equal(created.kind, 'created'); + }); +}); + test('decodeLegacySessionHeader preserves legacy subagent lineage, thinkingLevel, and unread position', async () => { const header = decodeLegacySessionHeader( { diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index 8bcb7d9e5f..9ff01efd56 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -315,6 +315,7 @@ async function createExecutionStoresForWrite sessionStore.createStableSession(request, initialBoundary)), discardStableConversationCopy: (sessionId, requestFingerprint) => run(() => sessionStore.discardStableConversationCopy(sessionId, requestFingerprint)), + importSession: (header, messages) => run(() => sessionStore.importSession(header, messages)), createSubagent: (input, initialBoundary) => run(() => sessionStore.createSubagent(input, initialBoundary)), createAgentGraphOperator: (input, request, expectedRevision, initialBoundary) => diff --git a/packages/storage/src/legacy-session-import.ts b/packages/storage/src/legacy-session-import.ts index e8845bad72..233e927c72 100644 --- a/packages/storage/src/legacy-session-import.ts +++ b/packages/storage/src/legacy-session-import.ts @@ -1,4 +1,3 @@ -import { createHash } from 'node:crypto'; import type { Dirent } from 'node:fs'; import { readdir, readFile } from 'node:fs/promises'; import { join } from 'node:path'; @@ -10,7 +9,6 @@ import { isPermissionMode, isSessionBlockedReason, isSessionStatus, - type CreateSessionInput, type SessionHeader, type StoredMessage, } from '@maka/core'; @@ -33,26 +31,28 @@ import { normalizeSessionHeader, type SessionAuthorityStore } from './session-st * SQLite row is reported as failed, never fabricated into a fake session. * * Design: - * - Runs once per session: idempotency key is the session id itself. We probe - * with `probeStableSessionCreate` first and skip any id already present in - * SQLite, so re-running (e.g. after a partial failure or a second launch) - * is safe and does not duplicate data — and the probe runs BEFORE the - * expensive file read, so every launch of an upgraded install skips known - * ids without touching the transcript. - * - Per-file atomicity: a file either imports completely (header + all - * messages) or is skipped and counted as failed. The decoded header and the - * post-create header patch are validated through `normalizeSessionHeader` - * BEFORE any write, so the three store writes (create → append → update) - * cannot fail validation after two of three commits — the failure mode that - * used to leave a permanent partial session. + * - Each file imports through one store-level `importSession` call: the + * header row (with its historical timestamps and flags) and all messages + * land in a single SQLite transaction. A file either imports completely or + * is counted as failed with nothing persisted — a partial session can never + * survive, even across process death or a failpoint mid-write. + * - Idempotency is the session id's primary key: `importSession` is + * `INSERT OR IGNORE`, so re-runs (and concurrent first launches, e.g. a CLI + * pre-check store racing the TUI/desktop store) converge on one winner with + * no create claims, fingerprints, or probes — the losing process sees + * `existing` and skips without reading the transcript. + * - The decoded header is validated through `normalizeSessionHeader` BEFORE + * the write, so a malformed header fails the file without persisting + * anything. * - Best-effort: a malformed or unreadable file is reported in the result and - * never blocks startup or the import of other sessions. + * never blocks startup or the import of other sessions. A whole-run failure + * (e.g. an unreadable `sessions/` directory) propagates to the caller; the + * store's `ensureReady` latch contains it and logs it. * - Legacy files are retained after import, honoring the repository policy * that legacy stores are kept as migration evidence. * - Legacy subagent children (real on-disk sessions before the cutover) are - * routed through `createSubagent` so their parent/child lineage is kept; a - * subagent header whose spawn identity is incomplete fails the file instead - * of silently flattening into a top-level session. + * imported under their own legacy id with parent/runtime/spawn lineage + * preserved, so session-tree nesting and descendant queries keep working. * * The compatibility defaults mirror the old `decodeSessionHeader` (kept public * for one-way importers before #1994 deleted it): missing `permissionMode` @@ -64,37 +64,23 @@ const LEGACY_SESSIONS_DIR = 'sessions'; const LEGACY_TRANSCRIPT_FILE = 'session.jsonl'; const SESSION_TRANSCRIPT_MARKER_TYPE = 'session_transcript'; -/** - * Stable request fingerprint for legacy imports. `createStableSession` - * requires a `sha256:` fingerprint; using a constant derived from this - * module's identity makes every import of the same session id the "same - * request", so concurrent first-launch processes converge on one winner. - */ -const LEGACY_IMPORT_FINGERPRINT = `sha256:${createHash('sha256') - .update('maka-legacy-session-import') - .digest('hex')}`; - export interface LegacySessionImportResult { imported: number; skipped: number; failed: number; failures: Array<{ sessionId: string; error: string }>; - /** `skipped` where the session id already existed in SQLite (already imported, or a concurrent process won the race). */ - skippedExisting: number; - /** `skipped` where the session id was claimed by a different request fingerprint or tombstoned. */ - skippedCollision: number; } -type LegacySessionFileOutcome = - | { kind: 'imported' } - | { kind: 'skipped'; reason: 'existing' | 'conflict' }; +type LegacySessionFileOutcome = 'imported' | 'skipped'; /** * Import all legacy `sessions//session.jsonl` transcripts under * `workspaceRoot` into the SQLite-backed session store. * * The store must be open (a `createSessionStore(workspaceRoot)` instance is - * ready immediately). Idempotent; safe to call on every launch. + * ready immediately). Idempotent; safe to call on every launch. A whole-run + * failure (missing directory is the normal case and is not a failure) throws; + * per-file failures are reported in the result and never throw. */ export async function importLegacySessionsOnce( store: SessionAuthorityStore, @@ -105,8 +91,6 @@ export async function importLegacySessionsOnce( skipped: 0, failed: 0, failures: [], - skippedExisting: 0, - skippedCollision: 0, }; const sessionsDir = join(workspaceRoot, LEGACY_SESSIONS_DIR); @@ -129,15 +113,10 @@ export async function importLegacySessionsOnce( const transcriptPath = join(sessionsDir, sessionId, LEGACY_TRANSCRIPT_FILE); try { const outcome = await importLegacySessionFile(store, sessionId, transcriptPath); - if (outcome.kind === 'imported') { + if (outcome === 'imported') { result.imported += 1; } else { result.skipped += 1; - if (outcome.reason === 'conflict') { - result.skippedCollision += 1; - } else { - result.skippedExisting += 1; - } } } catch (error) { result.failed += 1; @@ -156,56 +135,18 @@ async function importLegacySessionFile( sessionId: string, transcriptPath: string, ): Promise { - // Idempotency probe FIRST: any session id already known to SQLite — whether - // imported by an earlier run, created by the user, or claimed by a - // concurrent first-launch process — is left untouched, and we never read or - // parse its transcript on later launches. - const probe = await store.probeStableSessionCreate(sessionId, LEGACY_IMPORT_FINGERPRINT); - if (probe.kind === 'existing') return { kind: 'skipped', reason: 'existing' }; - if (probe.kind === 'conflict') return { kind: 'skipped', reason: 'conflict' }; - const { header, messages } = await readLegacyTranscript(transcriptPath, sessionId); - // Validate BEFORE any write. `createStableSession` and `updateHeader` both - // re-validate through `normalizeSessionHeader`; the old flow let the update - // (the third of three transactions) throw after create+append had already - // committed, leaving a permanent partial session that the next run's probe - // would report as "skipped". Validating the decoded header AND the final - // post-patch shape here moves that failure before the first write. - const normalized = normalizeSessionHeader(header); - const patch = legacyHeaderPatch(header, messages); - normalizeSessionHeader({ ...normalized, ...patch }, sessionId); - - const input = toCreateSessionInput(header); - if (header.subagentParent) { - // Legacy subagent children carry parent/runtime/spawn metadata; route - // through createSubagent so lineage is preserved. A malformed/incomplete - // spawn identity throws (requireSubagentSpawnIdentity) and fails the file - // rather than flattening the child into a top-level session. - const created = await store.createSubagent(input); - if (!created.created) return { kind: 'skipped', reason: 'existing' }; - } else { - const created = await store.createStableSession({ - sessionId, - requestFingerprint: LEGACY_IMPORT_FINGERPRINT, - input, - }); - // A concurrent process may have won the race between probe and create; - // both `existing` and `conflict` mean the id is taken, so skip. - if (created.kind !== 'created') { - return { kind: 'skipped', reason: created.kind }; - } - } + // Validate the fully-decoded header BEFORE any write. The single import + // transaction below is atomic, so a malformed header must fail here — never + // after a partial commit. + const normalized = normalizeSessionHeader(header, sessionId); - await store.appendMessages(sessionId, messages); - - // `createStableSession`/`createSubagent` stamp now-based timestamps and - // default flags; the legacy header carries the real lifecycle facts, so - // restore them. The final shape was pre-validated above, so this update - // cannot fail validation. - await store.updateHeader(sessionId, patch); - - return { kind: 'imported' }; + // One transaction: header row (with historical timestamps/flags) + all + // messages in order. Idempotent by primary key, so an id already in SQLite + // (earlier run, user-created, concurrent winner) is skipped, never clobbered. + const outcome = await store.importSession(normalized, messages); + return outcome === 'imported' ? 'imported' : 'skipped'; } async function readLegacyTranscript( @@ -213,6 +154,7 @@ async function readLegacyTranscript( sessionId: string, ): Promise<{ header: SessionHeader; messages: StoredMessage[] }> { const text = await readFile(transcriptPath, 'utf8'); + const endsWithNewline = text.endsWith('\n'); const rawLines = text.split('\n'); const contentLines = rawLines.filter((line) => line.trim().length > 0); if (contentLines.length === 0) { @@ -251,10 +193,14 @@ async function readLegacyTranscript( try { parsed = JSON.parse(line); } catch (error) { - // The classic interrupted-append artifact: a final line cut off mid-write. - // The pre-#1994 strict reader skipped such a torn tail; the rest of the - // file is intact and still imports. - if (isLastContentLine(rawLines, index)) break; + // The classic interrupted-append artifact is a final line cut off + // mid-write: the file does not end with a newline and the truncated + // record never closed its outer bracket. The pre-#1994 strict reader + // skipped exactly that shape (`!endsWithNewline && lastLine && + // incomplete-prefix`) and failed on anything else; so do we — a corrupt + // line that ends in a newline, or a garbage tail, fails the whole file + // rather than silently dropping a record. + if (isTornTail(rawLines, index, endsWithNewline)) break; throw new Error( `Legacy session ${sessionId} has a corrupt JSONL record at line ${index + 1}: ${ error instanceof Error ? error.message : String(error) @@ -273,18 +219,40 @@ function isSessionTranscriptMarker(value: unknown): boolean { return (value as { type?: unknown }).type === SESSION_TRANSCRIPT_MARKER_TYPE; } -function isLastContentLine(rawLines: string[], index: number): boolean { - for (let i = index + 1; i < rawLines.length; i += 1) { - if (rawLines[i]!.trim().length > 0) return false; +/** + * True only for the torn-tail shape the pre-#1994 strict reader tolerated: the + * final line of a file that does not end with a newline, whose parse failure + * is explained by an unclosed JSON bracket (i.e. the record was truncated + * mid-write rather than being garbage). Any other unparseable line — including + * a truncated line that still ends in a newline — is a real corruption. + */ +function isTornTail(rawLines: string[], index: number, endsWithNewline: boolean): boolean { + if (endsWithNewline) return false; + if (index !== rawLines.length - 1) return false; + return isIncompleteJsonPrefix(rawLines[index]!); +} + +function isIncompleteJsonPrefix(line: string): boolean { + let depth = 0; + let inString = false; + let escaped = false; + for (const char of line) { + if (inString) { + if (escaped) escaped = false; + else if (char === '\\') escaped = true; + else if (char === '"') inString = false; + } else if (char === '"') inString = true; + else if (char === '{' || char === '[') depth += 1; + else if (char === '}' || char === ']') depth -= 1; } - return true; + return depth > 0; } /** * The legacy header was a loose JSON object (some fields optional, old * backends named differently). This mirrors the pre-#1994 `decodeSessionHeader` - * compatibility rules; the final strict shape is enforced by the metadata - * store's `normalizeSessionHeader` on create. + * compatibility rules; the final strict shape is enforced by + * `normalizeSessionHeader` before any write. */ export function decodeLegacySessionHeader(value: unknown, sessionId: string): SessionHeader { if (!value || typeof value !== 'object' || Array.isArray(value)) { @@ -388,67 +356,6 @@ function normalizeLegacySessionName(name: string): string { return name === 'New Session' ? DEFAULT_SESSION_NAME : name; } -function toCreateSessionInput(header: SessionHeader): CreateSessionInput { - return { - cwd: header.cwd, - ...(header.projectId !== undefined ? { projectId: header.projectId } : {}), - name: header.name, - backend: header.backend, - llmConnectionSlug: header.llmConnectionSlug, - model: header.model, - ...(header.thinkingLevel !== undefined ? { thinkingLevel: header.thinkingLevel } : {}), - permissionMode: header.permissionMode, - collaborationMode: header.collaborationMode, - orchestrationMode: header.orchestrationMode, - ...(header.status !== undefined ? { status: header.status } : {}), - ...(header.blockedReason !== undefined ? { blockedReason: header.blockedReason } : {}), - labels: header.labels, - ...(header.parentSessionId !== undefined ? { parentSessionId: header.parentSessionId } : {}), - ...(header.branchOfTurnId !== undefined ? { branchOfTurnId: header.branchOfTurnId } : {}), - ...(header.subagentParent !== undefined ? { subagentParent: header.subagentParent } : {}), - ...(header.subagentRuntime !== undefined ? { subagentRuntime: header.subagentRuntime } : {}), - ...(header.subagentSpawn !== undefined ? { subagentSpawn: header.subagentSpawn } : {}), - ...(header.subagentWorkspace !== undefined - ? { subagentWorkspace: header.subagentWorkspace } - : {}), - ...(header.revisionRootSessionId !== undefined - ? { revisionRootSessionId: header.revisionRootSessionId } - : {}), - ...(header.revisionParentSessionId !== undefined - ? { revisionParentSessionId: header.revisionParentSessionId } - : {}), - ...(header.revisionOfTurnId !== undefined ? { revisionOfTurnId: header.revisionOfTurnId } : {}), - ...(header.revisionIndex !== undefined ? { revisionIndex: header.revisionIndex } : {}), - ...(header.revisionState !== undefined ? { revisionState: header.revisionState } : {}), - }; -} - -/** - * Fields `buildSessionHeader` cannot express (it stamps `Date.now()` and - * default flags) but the legacy header carries. `updateHeader` re-validates - * through `normalizeSessionHeader`, so the values must be canonically shaped — - * `importLegacySessionFile` pre-validates this exact merged shape before the - * first write. - */ -function legacyHeaderPatch( - header: SessionHeader, - messages: readonly StoredMessage[], -): Partial { - const hasUserMessage = messages.some((message) => message.type === 'user'); - return { - createdAt: header.createdAt, - lastUsedAt: header.lastUsedAt, - ...(header.lastMessageAt !== undefined ? { lastMessageAt: header.lastMessageAt } : {}), - statusUpdatedAt: header.statusUpdatedAt, - titleIsManual: header.titleIsManual, - isFlagged: header.isFlagged, - isArchived: header.isArchived, - ...(header.archivedAt !== undefined ? { archivedAt: header.archivedAt } : {}), - hasUnread: header.hasUnread, - connectionLocked: header.connectionLocked || hasUserMessage, - }; -} - /** The loose legacy header shape accepted by `decodeLegacySessionHeader`. */ type LegacyStoredSessionHeader = { workspaceRoot: string; diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index c0eee66f89..bbf9292029 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -222,6 +222,15 @@ export interface SessionAuthorityStore extends SessionStore { initialBoundary?: ExecutionBoundary, ): Promise; discardStableConversationCopy(sessionId: string, requestFingerprint: string): Promise; + /** + * Insert a session with its historical facts atomically (header + messages + * in one transaction, idempotent by session id). Used by the one-time + * legacy JSONL importer; not part of the normal session lifecycle. + */ + importSession( + header: SessionHeader, + messages: readonly StoredMessage[], + ): Promise<'imported' | 'existing'>; listCatalogPage( filter: SessionListFilter | undefined, cursor: SessionCatalogPageCursor | undefined, @@ -273,10 +282,7 @@ export function createSessionStoreWithTestDependencies( class SqliteSessionStore implements SessionAuthorityStore { private readonly metadata: SqliteSessionMetadataStore; private readonly workspaceRoot: string; - private legacyImportPromise: Promise | null = null; - private legacyImportResult: - | import('./legacy-session-import.js').LegacySessionImportResult - | null = null; + private readyPromise: Promise | null = null; private closePromise: Promise | null = null; constructor(workspaceRoot: string, _dependencies: SessionAuthorityStoreTestDependencies) { @@ -289,37 +295,41 @@ class SqliteSessionStore implements SessionAuthorityStore { } /** - * One-time legacy JSONL session import, awaited before any read that feeds - * the session list so upgraded installs see their pre-cutover sessions. + * One-time legacy JSONL session import, awaited by every public method so + * upgraded installs see their pre-cutover sessions from any entry point — + * desktop boot, CLI, headless, and `maka --resume ` all reach a + * read/write method before touching session data, and each awaits this + * latch (same shape as `importLegacyCatalogOnce` in project-catalog.ts). * - * The import itself is best-effort (per-file errors are reported, never - * thrown), so this promise never rejects; it exists only to serialize the - * first list against the import. The outcome is retained on the instance - * (and failures surfaced to the log) so a whole-run or per-file failure is + * The import itself is best-effort: per-file errors are reported in the + * result and never thrown, and a whole-run failure (e.g. an unreadable + * sessions/ directory) is logged and dropped rather than taking the read + * path down with it. The outcome is surfaced to the log so a failure is * observable instead of silently swallowed — the feature's purpose (data * appears in the UI) can otherwise fail with zero signal. */ - private ensureLegacyImported(): Promise { - this.legacyImportPromise ??= importLegacySessionsOnce(this, this.workspaceRoot) - .then((result) => { - this.legacyImportResult = result; - if (result.failed > 0) { - console.warn( - `[legacy-session-import] ${result.failed} of ${result.imported + result.skipped + result.failed} legacy session(s) failed to import; ` + - `failures: ${result.failures.map((failure) => `${failure.sessionId}: ${failure.error}`).join(' | ')}`, - ); - } else if (result.imported > 0) { - console.info(`[legacy-session-import] imported ${result.imported} legacy session(s)`); - } - }) - .catch((error: unknown) => { - this.legacyImportResult = null; - console.error( - '[legacy-session-import] import run failed:', - error instanceof Error ? error.message : String(error), + private ensureReady(): Promise { + this.readyPromise ??= this.importLegacySessionsOnce(); + return this.readyPromise; + } + + private async importLegacySessionsOnce(): Promise { + try { + const result = await importLegacySessionsOnce(this, this.workspaceRoot); + if (result.failed > 0) { + console.warn( + `[legacy-session-import] ${result.failed} of ${result.imported + result.skipped + result.failed} legacy session(s) failed to import; ` + + `failures: ${result.failures.map((failure) => `${failure.sessionId}: ${failure.error}`).join(' | ')}`, ); - }); - return this.legacyImportPromise; + } else if (result.imported > 0) { + console.info(`[legacy-session-import] imported ${result.imported} legacy session(s)`); + } + } catch (error) { + console.error( + '[legacy-session-import] import run failed:', + error instanceof Error ? error.message : String(error), + ); + } } async create( @@ -408,6 +418,18 @@ class SqliteSessionStore implements SessionAuthorityStore { return this.metadata.discardStableSessionCreate(sessionId, requestFingerprint); } + async importSession( + header: SessionHeader, + messages: readonly StoredMessage[], + ): Promise<'imported' | 'existing'> { + // Deliberately not awaited against ensureReady: the importer drives the + // migration, so gating its own write primitive on the same latch would + // self-deadlock. The store is already ready by construction here (the + // metadata store is created in the constructor and importSession only + // touches it). + return this.metadata.importSession(header, messages, projectSessionCatalogMessages(messages)); + } + async createSubagent( input: CreateSessionInput, initialBoundary?: ExecutionBoundary, @@ -493,7 +515,6 @@ class SqliteSessionStore implements SessionAuthorityStore { async list(filter?: SessionListFilter): Promise { await this.ensureReady(); - await this.ensureLegacyImported(); const records = (await this.metadata.list(filter)).filter( (record) => record.header.conversationCopy?.state !== 'preparing', ); @@ -537,7 +558,6 @@ class SqliteSessionStore implements SessionAuthorityStore { limit: number, expectedRevision?: `sha256:${string}`, ): Promise { - await this.ensureLegacyImported(); await this.ensureCatalogProjectionReadable(); const page = await this.metadata.listCatalogPage(filter ?? {}, cursor, limit); const revision = projectCatalogRevision(page.revision); @@ -570,7 +590,6 @@ class SqliteSessionStore implements SessionAuthorityStore { async listHeaders(): Promise { await this.ensureReady(); - await this.ensureLegacyImported(); return (await this.metadata.list()) .map((record) => record.header) .sort((a, b) => a.id.localeCompare(b.id)); @@ -587,10 +606,9 @@ class SqliteSessionStore implements SessionAuthorityStore { async readHeaderRecordSnapshot(sessionId: string): Promise { await this.ensureReady(); - // `maka --resume ` reads the header before any list, so the - // import gate must cover this entry point too — otherwise the first - // post-upgrade resume of a pre-cutover session silently starts fresh. - await this.ensureLegacyImported(); + // `maka --resume ` reads the header before any list; the + // import runs in ensureReady, so the first post-upgrade resume of a + // pre-cutover session sees its imported rows. return projectHeaderSnapshot(await this.metadata.read(sessionId)); } @@ -605,10 +623,6 @@ class SqliteSessionStore implements SessionAuthorityStore { async readMessagesSnapshot(sessionId: string): Promise { await this.ensureReady(); - // Same import gate as readHeaderSnapshot: a resumed legacy session reads - // its messages immediately after the header, and both must see the - // imported rows on the first post-upgrade run. - await this.ensureLegacyImported(); return this.metadata.readMessages(sessionId); } @@ -643,6 +657,7 @@ class SqliteSessionStore implements SessionAuthorityStore { async appendMessages(sessionId: string, messages: StoredMessage[]): Promise { if (messages.length === 0) return; + await this.ensureReady(); await this.metadata.appendMessages( sessionId, messages, @@ -816,6 +831,9 @@ class SqliteSessionStore implements SessionAuthorityStore { } private async closeAfterReady(): Promise { + // Ensure the one-time import has settled before closing the database so + // a concurrent close cannot race an in-flight migration. + await this.ensureReady(); this.metadata.close(); } @@ -829,8 +847,6 @@ class SqliteSessionStore implements SessionAuthorityStore { return this.updateHeader(header.id, { connectionLocked: true }); } - private async ensureReady(): Promise {} - private async ensureCatalogProjectionReadable(): Promise { await this.ensureReady(); } diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index fa05e88d35..a4a9acf430 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -1254,6 +1254,63 @@ export class SqliteSessionMetadataStore { return this.readCatalogRevisionSync(); } + /** + * Import a session with its historical facts in a single SQLite + * transaction: the header row is written with the given (historical) + * timestamps and flags, and every message is appended in order. + * + * Idempotent by primary key: if the session id already exists — imported + * by an earlier run, created by the user, or written by a concurrent + * first-launch process — nothing is written and `'existing'` is returned. + * Tombstoned ids are never resurrected. Concurrent first launches converge + * on one winner for free: SQLite serializes the transaction and the loser + * observes the winner's row, so no create claims or fingerprints are + * needed. A failure anywhere inside the transaction (e.g. a failpoint) + * rolls back the whole import, so a partial session can never persist. + */ + async importSession( + header: SessionHeader, + messages: readonly StoredMessage[], + projection: SessionCatalogMessageProjection, + ): Promise<'imported' | 'existing'> { + this.assertOpen(); + const normalized = normalizeSessionHeader(header); + assertSafeSessionId(normalized.id); + assertCatalogMessageProjection(projection); + // Canonicalize every record exactly like appendMessages: round-trip + // through JSON so the stored form matches what the recovery path reads. + const encoded = messages.map((message) => { + const json = JSON.stringify(message); + const canonical = decodeStoredMessageForRecovery(JSON.parse(json) as unknown); + return { message: canonical, json }; + }); + return this.transaction(() => { + if (this.hasTombstone(normalized.id)) return 'existing'; + const inserted = this.tryInsertHeader(normalized, 1, normalized.createdAt, true); + if (!inserted) return 'existing'; + if (encoded.length > 0) { + const insert = this.db.prepare(` + INSERT INTO session_messages( + session_id, sequence, message_id, message_type, message_ts, record_json + ) VALUES (?, ?, ?, ?, ?, ?) + `); + for (let sequence = 0; sequence < encoded.length; sequence += 1) { + const entry = encoded[sequence]!; + insert.run( + normalized.id, + sequence, + entry.message.id, + entry.message.type, + entry.message.ts, + entry.json, + ); + } + this.updateCatalogProjectionSync(normalized.id, projection, false); + } + return 'imported'; + }); + } + async appendMessages( sessionId: string, messages: readonly StoredMessage[], From ec1cb46ddb0dd2f75d0493492217fbd5efb8a5d5 Mon Sep 17 00:00:00 2001 From: cat0825 <1759138827@qq.com> Date: Thu, 6 Aug 2026 16:33:10 +0800 Subject: [PATCH 7/7] fix(storage): probe legacy session ids before reading transcripts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restores the probe-before-read steady-state cost from review round 2 on the single-transaction design: the importer now asks the store whether a session id already exists (live or tombstoned) before opening or parsing its file, so every launch of an upgraded install pays a directory listing plus per-id SQLite existence checks. importSession remains the idempotency authority — a race between the probe and the write still converges on one winner via the primary key. Adds hasSession to the store surface and a test that corrupts the on-disk transcript between runs to pin that a skipped id is never re-read. --- .../__tests__/legacy-session-import.test.ts | 18 ++++++++++++++++++ packages/storage/src/execution-stores.ts | 1 + packages/storage/src/legacy-session-import.ts | 17 +++++++++++++++-- packages/storage/src/session-store.ts | 11 +++++++++++ .../src/sqlite-session-metadata-store.ts | 14 ++++++++++++++ 5 files changed, 59 insertions(+), 2 deletions(-) diff --git a/packages/storage/src/__tests__/legacy-session-import.test.ts b/packages/storage/src/__tests__/legacy-session-import.test.ts index 5e91727ee0..a328ccb77a 100644 --- a/packages/storage/src/__tests__/legacy-session-import.test.ts +++ b/packages/storage/src/__tests__/legacy-session-import.test.ts @@ -116,6 +116,24 @@ test('is idempotent: a second run skips the already-imported session without dup }); }); +test('a second run skips without re-reading the transcript', async () => { + await withWorkspace(async ({ sessions, workspace }) => { + await seedLegacySession(workspace, 'legacy-a1b2c3d4e5f6', 'normal-session.jsonl'); + + const first = await importLegacySessionsOnce(sessions, workspace); + assert.equal(first.imported, 1); + + // Corrupt the transcript on disk: a re-read would now fail the file, so + // a skipped result pins that the probe runs BEFORE the file read. + await writeFile( + join(workspace, 'sessions', 'legacy-a1b2c3d4e5f6', 'session.jsonl'), + '{ this is not valid json anymore', + ); + const second = await importLegacySessionsOnce(sessions, workspace); + assert.deepEqual(second, { imported: 0, skipped: 1, failed: 0, failures: [] }); + }); +}); + test('a corrupt transcript is skipped as failed while other sessions still import', async () => { await withWorkspace(async ({ sessions, workspace }) => { await seedLegacySession(workspace, 'legacy-a1b2c3d4e5f6', 'normal-session.jsonl'); diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index 9ff01efd56..97a68ea4d9 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -316,6 +316,7 @@ async function createExecutionStoresForWrite run(() => sessionStore.discardStableConversationCopy(sessionId, requestFingerprint)), importSession: (header, messages) => run(() => sessionStore.importSession(header, messages)), + hasSession: (sessionId) => run(() => sessionStore.hasSession(sessionId)), createSubagent: (input, initialBoundary) => run(() => sessionStore.createSubagent(input, initialBoundary)), createAgentGraphOperator: (input, request, expectedRevision, initialBoundary) => diff --git a/packages/storage/src/legacy-session-import.ts b/packages/storage/src/legacy-session-import.ts index 233e927c72..a2aab76a89 100644 --- a/packages/storage/src/legacy-session-import.ts +++ b/packages/storage/src/legacy-session-import.ts @@ -39,8 +39,10 @@ import { normalizeSessionHeader, type SessionAuthorityStore } from './session-st * - Idempotency is the session id's primary key: `importSession` is * `INSERT OR IGNORE`, so re-runs (and concurrent first launches, e.g. a CLI * pre-check store racing the TUI/desktop store) converge on one winner with - * no create claims, fingerprints, or probes — the losing process sees - * `existing` and skips without reading the transcript. + * no create claims or fingerprints. A cheap `hasSession` probe runs BEFORE + * the file read, so every launch of an upgraded install pays a directory + * listing plus per-id existence checks and never re-reads an imported + * transcript. * - The decoded header is validated through `normalizeSessionHeader` BEFORE * the write, so a malformed header fails the file without persisting * anything. @@ -112,6 +114,17 @@ export async function importLegacySessionsOnce( for (const sessionId of sessionDirs) { const transcriptPath = join(sessionsDir, sessionId, LEGACY_TRANSCRIPT_FILE); try { + // Idempotency probe BEFORE the file read: an id already known to + // SQLite — imported by an earlier run, created by the user, or claimed + // by a concurrent first-launch process — is skipped without opening or + // parsing its transcript, so every launch of an upgraded install pays + // only a directory listing plus per-id existence checks. importSession + // remains the authority on idempotency (a race between this probe and + // the write converges on one winner via the primary key). + if (await store.hasSession(sessionId)) { + result.skipped += 1; + continue; + } const outcome = await importLegacySessionFile(store, sessionId, transcriptPath); if (outcome === 'imported') { result.imported += 1; diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index bbf9292029..d21eeaaa3d 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -231,6 +231,11 @@ export interface SessionAuthorityStore extends SessionStore { header: SessionHeader, messages: readonly StoredMessage[], ): Promise<'imported' | 'existing'>; + /** + * Cheap existence probe used by the legacy importer to skip ids already in + * SQLite before reading their transcripts. + */ + hasSession(sessionId: string): Promise; listCatalogPage( filter: SessionListFilter | undefined, cursor: SessionCatalogPageCursor | undefined, @@ -430,6 +435,12 @@ class SqliteSessionStore implements SessionAuthorityStore { return this.metadata.importSession(header, messages, projectSessionCatalogMessages(messages)); } + async hasSession(sessionId: string): Promise { + // Same rationale as importSession: no ensureReady gate — the importer + // drives the migration and must not await its own latch. + return this.metadata.hasSession(sessionId); + } + async createSubagent( input: CreateSessionInput, initialBoundary?: ExecutionBoundary, diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index a4a9acf430..7e66d890d6 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -1311,6 +1311,20 @@ export class SqliteSessionMetadataStore { }); } + /** + * Cheap existence probe used by the legacy importer before reading a + * transcript: an id already present in SQLite (live or tombstoned) is + * skipped without opening or parsing its file. Read-only; safe on every + * launch. + */ + async hasSession(sessionId: string): Promise { + this.assertOpen(); + assertSafeSessionId(sessionId); + return this.readTransaction( + () => this.readRecordSync(sessionId) !== undefined || this.hasTombstone(sessionId), + ); + } + async appendMessages( sessionId: string, messages: readonly StoredMessage[],