From 1a7be39c7b02d6e7349d6bc582f4ab14745b0076 Mon Sep 17 00:00:00 2001 From: luoye520ww <100058663+luoye520ww@users.noreply.github.com> Date: Tue, 14 Jul 2026 05:04:55 +0800 Subject: [PATCH 1/2] feat(storage): define thread store doctor contract --- kun/src/contracts/index.ts | 1 + kun/src/contracts/thread-store-diagnostics.ts | 48 +++++++++++++++++++ kun/tests/thread-store-diagnostics.test.ts | 47 ++++++++++++++++++ 3 files changed, 96 insertions(+) create mode 100644 kun/src/contracts/thread-store-diagnostics.ts create mode 100644 kun/tests/thread-store-diagnostics.test.ts diff --git a/kun/src/contracts/index.ts b/kun/src/contracts/index.ts index 60524d76a..8f57b8f53 100644 --- a/kun/src/contracts/index.ts +++ b/kun/src/contracts/index.ts @@ -16,3 +16,4 @@ export * from './tool-output-limits.js' export * from './memory.js' export * from './model-endpoint-format.js' export * from './extension-providers.js' +export * from './thread-store-diagnostics.js' diff --git a/kun/src/contracts/thread-store-diagnostics.ts b/kun/src/contracts/thread-store-diagnostics.ts new file mode 100644 index 000000000..6edf38aea --- /dev/null +++ b/kun/src/contracts/thread-store-diagnostics.ts @@ -0,0 +1,48 @@ +import { z } from 'zod' + +/** + * Result of checking one persisted thread without changing any files. + * + * The doctor deliberately reports each storage surface independently. A + * missing SQLite index is recoverable from JSONL, while invalid event data is + * not something a repair pass may silently reconstruct. + */ +export const ThreadStoreArtifactStatus = z.enum([ + 'ok', + 'missing', + 'invalid', + 'truncated', + 'mismatch' +]) +export type ThreadStoreArtifactStatus = z.infer + +export const ThreadStoreDiagnosticSeverity = z.enum(['warning', 'error']) +export type ThreadStoreDiagnosticSeverity = z.infer + +export const ThreadStoreDiagnosticIssue = z.object({ + // Diagnostics must remain safe to render and export even when a parser + // reports data copied from a damaged log line. + code: z.string().min(1).max(128), + message: z.string().min(1).max(1024), + severity: ThreadStoreDiagnosticSeverity +}).strict() +export type ThreadStoreDiagnosticIssue = z.infer + +export const ThreadStoreDiagnostic = z.object({ + threadId: z.string().min(1).max(256), + metadata: ThreadStoreArtifactStatus, + events: ThreadStoreArtifactStatus, + sqliteIndex: ThreadStoreArtifactStatus, + attachments: ThreadStoreArtifactStatus, + recoverable: z.boolean(), + issues: z.array(ThreadStoreDiagnosticIssue).max(64), + checkedAt: z.string().datetime({ offset: true }) +}).strict() +export type ThreadStoreDiagnostic = z.infer + +export const ThreadStoreDiagnosticReport = z.object({ + schemaVersion: z.literal(1), + checkedAt: z.string().datetime({ offset: true }), + threads: z.array(ThreadStoreDiagnostic) +}).strict() +export type ThreadStoreDiagnosticReport = z.infer diff --git a/kun/tests/thread-store-diagnostics.test.ts b/kun/tests/thread-store-diagnostics.test.ts new file mode 100644 index 000000000..2d2c7b53e --- /dev/null +++ b/kun/tests/thread-store-diagnostics.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from 'vitest' +import { + ThreadStoreDiagnostic, + ThreadStoreDiagnosticReport +} from '../src/contracts/thread-store-diagnostics.js' + +const diagnostic = { + threadId: 'thr_demo', + metadata: 'ok' as const, + events: 'truncated' as const, + sqliteIndex: 'missing' as const, + attachments: 'ok' as const, + recoverable: true, + issues: [ + { + code: 'truncated_events', + message: 'The event log ends with an incomplete JSONL record.', + severity: 'warning' as const + } + ], + checkedAt: '2026-07-14T12:00:00.000Z' +} + +describe('ThreadStoreDiagnostic contract', () => { + it('accepts a bounded per-thread diagnostic', () => { + expect(ThreadStoreDiagnostic.parse(diagnostic)).toEqual(diagnostic) + }) + + it('accepts an empty report and preserves the schema version', () => { + const report = { + schemaVersion: 1 as const, + checkedAt: '2026-07-14T12:00:00.000Z', + threads: [] + } + + expect(ThreadStoreDiagnosticReport.parse(report)).toEqual(report) + }) + + it('rejects unknown artifact states and non-ISO timestamps', () => { + expect(() => ThreadStoreDiagnostic.parse({ ...diagnostic, events: 'corrupt' })).toThrow() + expect(() => ThreadStoreDiagnostic.parse({ ...diagnostic, checkedAt: 'yesterday' })).toThrow() + }) + + it('rejects extra fields so the diagnostic contract stays stable', () => { + expect(() => ThreadStoreDiagnostic.parse({ ...diagnostic, absolutePath: 'C:\\secret' })).toThrow() + }) +}) From 889c6972d671a31a58cf8ff4db918495ad3b418d Mon Sep 17 00:00:00 2001 From: luoye520ww <100058663+luoye520ww@users.noreply.github.com> Date: Tue, 14 Jul 2026 05:33:22 +0800 Subject: [PATCH 2/2] feat(storage): add read-only thread store doctor --- kun/src/services/index.ts | 1 + kun/src/services/thread-store-doctor.ts | 414 ++++++++++++++++++++++++ kun/tests/thread-store-doctor.test.ts | 159 +++++++++ 3 files changed, 574 insertions(+) create mode 100644 kun/src/services/thread-store-doctor.ts create mode 100644 kun/tests/thread-store-doctor.test.ts diff --git a/kun/src/services/index.ts b/kun/src/services/index.ts index ec8e07591..08da17934 100644 --- a/kun/src/services/index.ts +++ b/kun/src/services/index.ts @@ -13,3 +13,4 @@ export * from './extension-host-broker.js' export * from './extension-secret-reveal-consent.js' export * from './extension-configuration-service.js' export * from './extension-view-session-service.js' +export * from './thread-store-doctor.js' diff --git a/kun/src/services/thread-store-doctor.ts b/kun/src/services/thread-store-doctor.ts new file mode 100644 index 000000000..630535ebe --- /dev/null +++ b/kun/src/services/thread-store-doctor.ts @@ -0,0 +1,414 @@ +import { readFile, readdir, stat } from 'node:fs/promises' +import { join, resolve } from 'node:path' +import { AttachmentMetadata } from '../contracts/attachments.js' +import { RuntimeEvent } from '../contracts/events.js' +import { + ThreadStoreDiagnostic, + ThreadStoreDiagnosticReport, + type ThreadStoreArtifactStatus, + type ThreadStoreDiagnosticIssue +} from '../contracts/thread-store-diagnostics.js' +import { ThreadSchema, type ThreadRecord } from '../contracts/threads.js' +import { isSafeThreadId } from '../contracts/thread-id.js' + +const MAX_ARTIFACT_BYTES = 16 * 1024 * 1024 +const MAX_JSONL_RECORDS = 100_000 + +export type ThreadStoreDoctorOptions = { + dataDir: string + sqlitePath?: string + attachmentRootDir?: string + nowIso?: () => string +} + +type JsonlInspection = { + status: ThreadStoreArtifactStatus + records: unknown[] + issues: ThreadStoreDiagnosticIssue[] +} + +type AttachmentIndex = { + valid: Map + invalidIds: Set +} + +/** Read-only health scan for the JSONL/SQLite/attachment thread store. */ +export async function scanThreadStore( + options: ThreadStoreDoctorOptions +): Promise { + const checkedAt = options.nowIso?.() ?? new Date().toISOString() + const threadsRoot = resolve(options.dataDir, 'threads') + const attachmentIndex = await loadAttachmentIndex(options.attachmentRootDir) + const sqlite = await openReadonlyIndex(options.sqlitePath ?? join(options.dataDir, 'index.sqlite3')) + const diagnostics: ThreadStoreDiagnostic[] = [] + + try { + const entries = await readdir(threadsRoot, { withFileTypes: true }) + const threadIds = entries + .filter((entry) => entry.isDirectory() && isSafeThreadId(entry.name)) + .map((entry) => entry.name) + .sort() + + for (const threadId of threadIds) { + diagnostics.push(await scanThread({ + threadId, + threadsRoot, + attachmentIndex, + sqlite, + checkedAt + })) + } + } catch (error) { + if (!isMissing(error)) throw error + } finally { + sqlite.index?.close() + } + + return ThreadStoreDiagnosticReport.parse({ + schemaVersion: 1, + checkedAt, + threads: diagnostics + }) +} + +async function scanThread(input: { + threadId: string + threadsRoot: string + attachmentIndex: AttachmentIndex + sqlite: ReadonlyIndexState + checkedAt: string +}): Promise { + const threadRoot = join(input.threadsRoot, input.threadId) + const metadata = await inspectMetadata(threadRoot, input.threadId) + const events = await inspectEvents(join(threadRoot, 'events.jsonl'), input.threadId) + const sqliteIndex = await inspectSqliteIndex(input.sqlite, input.threadId, threadRoot) + const attachments = inspectAttachments(metadata.records, input.attachmentIndex) + const issues = [...metadata.issues, ...events.issues, ...sqliteIndex.issues, ...attachments.issues].slice(0, 64) + + return ThreadStoreDiagnostic.parse({ + threadId: input.threadId, + metadata: metadata.status, + events: events.status, + sqliteIndex: sqliteIndex.status, + attachments: attachments.status, + recoverable: isRecoverable(metadata.status, events.status, attachments.status), + issues, + checkedAt: input.checkedAt + }) +} + +async function inspectMetadata(threadRoot: string, threadId: string): Promise { + const metadataPath = join(threadRoot, 'metadata.jsonl') + const metadata = await inspectJsonl(metadataPath) + if (metadata.status !== 'missing') { + const issues = [...metadata.issues] + let validRecord = false + for (const entry of metadata.records) { + if (!isRecord(entry) || entry.kind !== 'thread_metadata') continue + const parsed = ThreadSchema.safeParse(entry.thread) + if (parsed.success && parsed.data.id === threadId) validRecord = true + } + if (!validRecord) { + issues.push(issue('invalid_metadata', 'No valid thread metadata record was found.', 'error')) + return { status: 'invalid', records: [], issues } + } + return { ...metadata, issues } + } + + const legacyPath = join(threadRoot, 'thread.json') + try { + const raw = await readBoundedFile(legacyPath) + const parsed = ThreadSchema.safeParse(JSON.parse(raw)) + if (!parsed.success || parsed.data.id !== threadId) { + return { + status: 'invalid', + records: [], + issues: [issue('invalid_metadata', 'The legacy thread metadata is invalid.', 'error')] + } + } + return { status: 'ok', records: [parsed.data], issues: [] } + } catch (error) { + if (isMissing(error)) { + return { + status: 'missing', + records: [], + issues: [issue('missing_metadata', 'No thread metadata file was found.', 'error')] + } + } + return { + status: 'invalid', + records: [], + issues: [issue('invalid_metadata', 'The thread metadata could not be read.', 'error')] + } + } +} + +async function inspectEvents(path: string, threadId: string): Promise { + const inspection = await inspectJsonl(path) + const issues = [...inspection.issues] + let validEvents = 0 + for (const entry of inspection.records) { + const parsed = RuntimeEvent.safeParse(entry) + if (parsed.success && parsed.data.threadId === threadId) validEvents += 1 + } + if (inspection.status === 'ok' && inspection.records.length > 0 && validEvents === 0) { + issues.push(issue('invalid_events', 'No valid events for this thread were found.', 'error')) + return { status: 'invalid', records: [], issues } + } + return { ...inspection, issues } +} + +async function inspectJsonl(path: string): Promise { + let raw: string + try { + raw = await readBoundedFile(path) + } catch (error) { + if (isMissing(error)) return { status: 'missing', records: [], issues: [] } + return { + status: 'invalid', + records: [], + issues: [issue('unreadable_artifact', 'A persisted JSONL artifact could not be read.', 'error')] + } + } + if (!raw.trim()) return { status: 'missing', records: [], issues: [] } + + const records: unknown[] = [] + let malformed = false + let malformedFinal = false + let malformedNonFinal = false + const lines = raw.split('\n') + let nonEmptyIndex = -1 + for (let index = 0; index < lines.length; index += 1) { + const line = lines[index].trim() + if (!line) continue + nonEmptyIndex = index + try { + records.push(JSON.parse(line)) + } catch { + malformed = true + const isFinal = index === lines.length - 1 || lines.slice(index + 1).every((rest) => !rest.trim()) + malformedFinal ||= isFinal + malformedNonFinal ||= !isFinal + } + if (records.length > MAX_JSONL_RECORDS) { + return { + status: 'invalid', + records: [], + issues: [issue('artifact_too_large', 'A JSONL artifact contains too many records.', 'error')] + } + } + } + if (nonEmptyIndex < 0) return { status: 'missing', records: [], issues: [] } + if (!malformed) return { status: 'ok', records, issues: [] } + if (!malformedNonFinal && malformedFinal) { + return { + status: 'truncated', + records, + issues: [issue('truncated_artifact', 'The JSONL artifact ends with an incomplete record.', 'warning')] + } + } + return { + status: 'invalid', + records: [], + issues: [issue('invalid_artifact', 'The JSONL artifact contains malformed records.', 'error')] + } +} + +function inspectAttachments(records: unknown[], index: AttachmentIndex): { + status: ThreadStoreArtifactStatus + issues: ThreadStoreDiagnosticIssue[] +} { + const thread = records + .map((entry) => { + if (isRecord(entry) && entry.kind === 'thread_metadata') return ThreadSchema.safeParse(entry.thread) + return ThreadSchema.safeParse(entry) + }) + .find((result): result is { success: true; data: ThreadRecord } => result.success)?.data + if (!thread) return { status: 'ok', issues: [] } + + const ids = new Set() + for (const turn of thread.turns) { + for (const id of turn.attachmentIds ?? []) ids.add(id) + for (const item of turn.items) { + if ('attachmentIds' in item) { + for (const id of item.attachmentIds ?? []) ids.add(id) + } + } + } + if (ids.size === 0) return { status: 'ok', issues: [] } + + const issues: ThreadStoreDiagnosticIssue[] = [] + let status: ThreadStoreArtifactStatus = 'ok' + for (const id of ids) { + const attachment = index.valid.get(id) + if (!attachment) { + const nextStatus = index.invalidIds.has(id) ? 'invalid' : 'missing' + status = worseStatus(status, nextStatus) + issues.push(issue(nextStatus === 'invalid' ? 'invalid_attachment' : 'missing_attachment', 'A referenced attachment is unavailable.', 'error')) + continue + } + if (!attachment.contentExists || !attachment.metadata.threadIds.includes(thread.id)) { + status = worseStatus(status, 'mismatch') + issues.push(issue('attachment_scope_mismatch', 'A referenced attachment is missing content or thread scope.', 'error')) + } + } + return { status, issues } +} + +function worseStatus( + current: ThreadStoreArtifactStatus, + next: ThreadStoreArtifactStatus +): ThreadStoreArtifactStatus { + const rank: Record = { + ok: 0, + truncated: 1, + missing: 2, + mismatch: 3, + invalid: 4 + } + return rank[next] > rank[current] ? next : current +} + +function isRecoverable( + metadata: ThreadStoreArtifactStatus, + events: ThreadStoreArtifactStatus, + attachments: ThreadStoreArtifactStatus +): boolean { + return metadata === 'ok' + && events !== 'invalid' + && attachments === 'ok' +} + +function issue( + code: string, + message: string, + severity: ThreadStoreDiagnosticIssue['severity'] +): ThreadStoreDiagnosticIssue { + return { code, message, severity } +} + +function isRecord(value: unknown): value is Record { + return Boolean(value) && typeof value === 'object' && !Array.isArray(value) +} + +function isMissing(error: unknown): boolean { + return Boolean(error && typeof error === 'object' && 'code' in error && error.code === 'ENOENT') +} + +async function readBoundedFile(path: string): Promise { + const fileStat = await stat(path) + if (fileStat.size > MAX_ARTIFACT_BYTES) throw new Error('artifact_too_large') + return readFile(path, 'utf8') +} + +type ReadonlyIndex = { + getThread: (threadId: string) => ReadonlyIndexRow | undefined + close: () => void +} + +type ReadonlyIndexRow = { + metadata_path?: string + messages_path?: string + events_path?: string +} + +type ReadonlyIndexState = { + status: 'ok' | 'missing' | 'invalid' + index: ReadonlyIndex | null +} + +async function openReadonlyIndex(path: string): Promise { + let db: { close: () => void } | null = null + try { + await stat(path) + const sqlite = await import('better-sqlite3') + const Database = sqlite.default + const handle = new Database(path, { readonly: true, fileMustExist: true }) + db = handle + const statement = handle.prepare('SELECT metadata_path, messages_path, events_path FROM threads WHERE id = ?') + return { + status: 'ok', + index: { + getThread: (threadId) => statement.get(threadId) as ReadonlyIndexRow | undefined, + close: () => handle.close() + } + } + } catch (error) { + db?.close() + return { status: isMissing(error) ? 'missing' : 'invalid', index: null } + } +} + +async function inspectSqliteIndex( + sqlite: ReadonlyIndexState, + threadId: string, + threadRoot: string +): Promise<{ status: ThreadStoreArtifactStatus; issues: ThreadStoreDiagnosticIssue[] }> { + if (sqlite.status === 'missing') { + return { + status: 'missing', + issues: [issue('missing_sqlite_index', 'The SQLite index is unavailable.', 'warning')] + } + } + if (sqlite.status === 'invalid' || !sqlite.index) { + return { + status: 'invalid', + issues: [issue('invalid_sqlite_index', 'The SQLite index could not be opened.', 'error')] + } + } + try { + const row = sqlite.index.getThread(threadId) + if (!row) { + return { + status: 'mismatch', + issues: [issue('sqlite_index_mismatch', 'The SQLite index has no row for this thread.', 'warning')] + } + } + const expected = { + metadata_path: join(threadRoot, 'metadata.jsonl'), + messages_path: join(threadRoot, 'messages.jsonl'), + events_path: join(threadRoot, 'events.jsonl') + } + if (Object.entries(expected).some(([key, value]) => resolve(String(row[key as keyof typeof row] ?? '')) !== resolve(value))) { + return { + status: 'mismatch', + issues: [issue('sqlite_index_mismatch', 'The SQLite index paths do not match the thread files.', 'warning')] + } + } + return { status: 'ok', issues: [] } + } catch { + return { + status: 'invalid', + issues: [issue('invalid_sqlite_index', 'The SQLite index could not be queried.', 'error')] + } + } +} + +async function loadAttachmentIndex(rootDir: string | undefined): Promise { + const valid = new Map() + const invalidIds = new Set() + if (!rootDir) return { valid, invalidIds } + const entries = await readdir(rootDir, { withFileTypes: true }).catch(() => []) + for (const entry of entries) { + if (!entry.isFile() || !entry.name.endsWith('.json')) continue + const id = entry.name.slice(0, -'.json'.length) + try { + const metadata = AttachmentMetadata.parse(JSON.parse(await readBoundedFile(join(rootDir, entry.name)))) + valid.set(id, { + metadata, + contentExists: await fileExists(join(rootDir, `${id}.bin`)) + }) + } catch { + invalidIds.add(id) + } + } + return { valid, invalidIds } +} + +async function fileExists(path: string): Promise { + try { + return (await stat(path)).isFile() + } catch { + return false + } +} diff --git a/kun/tests/thread-store-doctor.test.ts b/kun/tests/thread-store-doctor.test.ts new file mode 100644 index 000000000..25dfdee16 --- /dev/null +++ b/kun/tests/thread-store-doctor.test.ts @@ -0,0 +1,159 @@ +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it } from 'vitest' +import { createThreadRecord } from '../src/domain/thread.js' +import { createTurnRecord } from '../src/domain/turn.js' +import { scanThreadStore } from '../src/services/thread-store-doctor.js' + +const roots: string[] = [] + +describe('scanThreadStore', () => { + afterEach(async () => { + await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))) + }) + + it('reports healthy JSONL and does not mutate the store', async () => { + const root = await makeRoot() + const thread = createThreadRecord({ + id: 'thr_healthy', + title: 'Healthy', + workspace: root, + model: 'deepseek-chat', + createdAt: '2026-01-01T00:00:00.000Z' + }) + const threadRoot = join(root, 'threads', thread.id) + await mkdir(threadRoot, { recursive: true }) + const metadataPath = join(threadRoot, 'metadata.jsonl') + const eventsPath = join(threadRoot, 'events.jsonl') + await writeFile(metadataPath, `${JSON.stringify({ kind: 'thread_metadata', version: 1, timestamp: thread.createdAt, thread })}\n`) + await writeFile(eventsPath, `${JSON.stringify({ + kind: 'heartbeat', seq: 1, timestamp: thread.createdAt, threadId: thread.id + })}\n`) + const before = await Promise.all([metadataPath, eventsPath].map(async (path) => ({ path, text: await readFileText(path) }))) + + const report = await scanThreadStore({ dataDir: root, nowIso: () => '2026-07-14T00:00:00.000Z' }) + expect(report.threads[0]).toMatchObject({ + threadId: thread.id, + metadata: 'ok', + events: 'ok', + sqliteIndex: 'missing', + attachments: 'ok', + recoverable: true + }) + expect(await Promise.all(before.map(async ({ path }) => readFileText(path)))).toEqual(before.map(({ text }) => text)) + }) + + it('classifies an incomplete final JSONL record as truncated', async () => { + const root = await makeRoot() + const thread = createThreadRecord({ + id: 'thr_truncated', + title: 'Truncated', + workspace: root, + model: 'deepseek-chat', + createdAt: '2026-01-01T00:00:00.000Z' + }) + const threadRoot = join(root, 'threads', thread.id) + await mkdir(threadRoot, { recursive: true }) + await writeFile(join(threadRoot, 'metadata.jsonl'), `${JSON.stringify({ kind: 'thread_metadata', version: 1, timestamp: thread.createdAt, thread })}\n`) + await writeFile(join(threadRoot, 'events.jsonl'), '{"kind":"heartbeat"') + + const report = await scanThreadStore({ dataDir: root }) + expect(report.threads[0]).toMatchObject({ events: 'truncated', recoverable: true }) + }) + + it('fails closed for invalid metadata', async () => { + const root = await makeRoot() + const threadRoot = join(root, 'threads', 'thr_invalid') + await mkdir(threadRoot, { recursive: true }) + await writeFile(join(threadRoot, 'metadata.jsonl'), '{"kind":"thread_metadata","thread":{"id":"thr_invalid"}}\n') + + const report = await scanThreadStore({ dataDir: root }) + expect(report.threads[0]).toMatchObject({ metadata: 'invalid', recoverable: false }) + expect(report.threads[0]?.issues.some((item) => item.code === 'invalid_metadata')).toBe(true) + }) + + it('checks a real read-only SQLite index and attachment content', async () => { + const root = await makeRoot() + const thread = createThreadRecord({ + id: 'thr_indexed', + title: 'Indexed', + workspace: root, + model: 'deepseek-chat', + createdAt: '2026-01-01T00:00:00.000Z' + }) + const attachmentId = 'att_0123456789abcdef01234567' + const threadWithAttachment = { + ...thread, + turns: [createTurnRecord({ id: 'turn_1', threadId: thread.id, prompt: 'attached', attachmentIds: [attachmentId] })] + } + const threadRoot = join(root, 'threads', thread.id) + const attachmentRoot = join(root, 'attachments') + await mkdir(threadRoot, { recursive: true }) + await mkdir(attachmentRoot, { recursive: true }) + await writeFile(join(threadRoot, 'metadata.jsonl'), `${JSON.stringify({ kind: 'thread_metadata', version: 1, timestamp: thread.createdAt, thread: threadWithAttachment })}\n`) + await writeFile(join(threadRoot, 'events.jsonl'), `${JSON.stringify({ + kind: 'heartbeat', seq: 1, timestamp: thread.createdAt, threadId: thread.id + })}\n`) + await writeFile(join(attachmentRoot, `${attachmentId}.json`), JSON.stringify({ + id: attachmentId, + name: 'missing.bin', + kind: 'document', + mimeType: 'text/plain', + byteSize: 3, + hash: 'a'.repeat(64), + threadIds: [thread.id], + workspaces: [], + createdAt: thread.createdAt, + updatedAt: thread.createdAt + })) + + const sqlitePath = join(root, 'index.sqlite3') + const sqlite = await import('better-sqlite3') + const db = new sqlite.default(sqlitePath) + db.exec('CREATE TABLE threads (id TEXT PRIMARY KEY, metadata_path TEXT, messages_path TEXT, events_path TEXT)') + db.prepare('INSERT INTO threads VALUES (?, ?, ?, ?)').run( + thread.id, + join(threadRoot, 'metadata.jsonl'), + join(threadRoot, 'messages.jsonl'), + join(threadRoot, 'events.jsonl') + ) + db.close() + + const report = await scanThreadStore({ dataDir: root, attachmentRootDir: attachmentRoot }) + expect(report.threads[0]).toMatchObject({ + sqliteIndex: 'ok', + attachments: 'mismatch', + recoverable: false + }) + }) + + it('does not treat an interior malformed JSONL record as truncation', async () => { + const root = await makeRoot() + const thread = createThreadRecord({ + id: 'thr_interior_bad', + title: 'Interior bad', + workspace: root, + model: 'deepseek-chat', + createdAt: '2026-01-01T00:00:00.000Z' + }) + const threadRoot = join(root, 'threads', thread.id) + await mkdir(threadRoot, { recursive: true }) + await writeFile(join(threadRoot, 'metadata.jsonl'), `${JSON.stringify({ kind: 'thread_metadata', version: 1, timestamp: thread.createdAt, thread })}\n`) + await writeFile(join(threadRoot, 'events.jsonl'), '{bad}\n{"kind":"heartbeat","seq":1,"timestamp":"2026-01-01T00:00:00.000Z","threadId":"thr_interior_bad"}\n') + + const report = await scanThreadStore({ dataDir: root }) + expect(report.threads[0]?.events).toBe('invalid') + }) +}) + +async function makeRoot(): Promise { + const root = await mkdtemp(join(tmpdir(), 'kun-doctor-')) + roots.push(root) + return root +} + +async function readFileText(path: string): Promise { + const { readFile } = await import('node:fs/promises') + return readFile(path, 'utf8') +}