From f79f1b3e3b037731c984f905c8cbcb8bf4668773 Mon Sep 17 00:00:00 2001 From: tommy0103 Date: Tue, 21 Jul 2026 00:58:34 +0800 Subject: [PATCH] refactor: add canonical transcript assembly seam --- CONTEXT.md | 20 +- README.md | 6 +- app/src/main/index.ts | 40 +- app/src/main/indexer.ts | 101 +-- app/src/renderer/src/data.js | 16 +- app/src/shared/session-detail-assembly.mjs | 176 +----- app/src/shared/session-detail-types.ts | 118 +--- app/tests/electron-session-reader-state.mjs | 6 +- app/tests/electron-session-virtualization.mjs | 4 +- .../adr/0001-parse-core-and-persist-layers.md | 10 +- ...anonical-transcript-session-detail-seam.md | 40 ++ packages/core/package.json | 2 + packages/core/src/db.ts | 32 +- packages/core/src/indexer.ts | 84 +-- packages/core/src/parsing.ts | 30 +- packages/core/src/persist.ts | 48 +- packages/core/src/providers/claude.ts | 193 +++++- packages/core/src/providers/codex.ts | 81 ++- packages/core/src/providers/kimi.ts | 83 ++- packages/core/src/providers/types.ts | 35 +- packages/core/src/schema-migrations.ts | 34 + packages/core/src/schema.sql | 6 +- packages/core/src/session-detail.ts | 597 ++++++++++++++++++ tests/app-indexer.test.mjs | 16 +- tests/app-kimi-index.test.mjs | 47 +- tests/app-main-settings.test.mjs | 1 + tests/app-provider-indexer.test.mjs | 2 +- tests/app-rollback-guard.test.mjs | 8 +- tests/claude-parse.test.mjs | 76 ++- tests/codex-parse.test.mjs | 30 +- tests/codex-replay-tool-identity.test.mjs | 69 ++ tests/kimi-parse.test.mjs | 59 +- tests/persist.test.mjs | 26 + tests/provider-registry.test.mjs | 1 + tests/provider-schema-stability.test.mjs | 4 +- tests/provider-session-detail.test.mjs | 212 +++++++ tests/runtime.test.mjs | 6 +- tests/session-detail-assembly.test.mjs | 86 ++- tests/session-live-reload.test.mjs | 6 +- 39 files changed, 1741 insertions(+), 670 deletions(-) create mode 100644 docs/adr/0007-canonical-transcript-session-detail-seam.md create mode 100644 packages/core/src/schema-migrations.ts create mode 100644 packages/core/src/session-detail.ts create mode 100644 tests/codex-replay-tool-identity.test.mjs create mode 100644 tests/provider-session-detail.test.mjs diff --git a/CONTEXT.md b/CONTEXT.md index e3e9476..3509517 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -40,13 +40,23 @@ node:fs/path/os — deliberately node:sqlite-free so the compiled providers can consumed by the app (whose Electron runtime has no `node:sqlite`). _Avoid_: parse core, parser, ingest -**Record**: -One normalized row destined for the index (session, message, tool call, tool -result, summary, subagent, workflow, …), emitted by a provider adapter before any -persistence happens. +**Transcript record**: +One provider-normalized fact (session, message, tool call, tool result, summary, +subagent, workflow, …) in the canonical transcript language. Provider adapters +resolve source-specific deduplication, visibility, and identity before emitting +it. The persist layer serializes transcript records; the session detail +assembler can consume the same stream directly. +_Avoid_: database row, raw event + +**Session detail assembler**: +The provider-independent module that projects canonical transcript records into +the timeline shape used by the app. It may consume records directly from a +provider adapter's fresh full parse or after a SQLite round-trip. Provider deltas +require prior state and are handled by the snapshot/patch seam instead. It never +branches on provider and never infers provider semantics from message text. **Persist layer**: -The single shared, provider- and binding-agnostic writer that consumes records +The single shared, provider- and binding-agnostic writer that consumes transcript records from any adapter and writes them into an injected SQLite handle inside a transaction. The binding is injected — `node:sqlite` (CLI) or `better-sqlite3` (app) — so there is one persist implementation, not one per diff --git a/README.md b/README.md index 5ba9937..31aa25f 100644 --- a/README.md +++ b/README.md @@ -200,9 +200,11 @@ Full-text search via FTS5 covers all layers. packages/core/ # @obelisk/core npm workspace (TypeScript + ESM) ├── src/ │ ├── providers/ -│ │ ├── types.ts # Provider + IndexRecord contract +│ │ ├── types.ts # Provider + TranscriptRecord contract │ │ ├── claude.ts # Claude Code adapter (line-incremental) -│ │ └── codex.ts # Codex adapter (full-reparse) +│ │ ├── codex.ts # Codex adapter (full-reparse) +│ │ └── kimi.ts # Kimi Code adapter (session projection) +│ ├── session-detail.ts # Provider-independent transcript projection │ ├── persist.ts # Binding-agnostic record writer (upsert/merge) │ ├── tx.ts # Write transaction + connection config │ ├── write-coordinator.ts # Bounded retry policy diff --git a/app/src/main/index.ts b/app/src/main/index.ts index f651b6c..dab7385 100644 --- a/app/src/main/index.ts +++ b/app/src/main/index.ts @@ -10,6 +10,7 @@ import { createIndexerService } from './indexer-service.ts'; import { createWorkerBuildIndex } from './indexer-worker-client.ts'; import { buildRecapExportQuery } from './recap-capture-query.ts'; import { acquireWriterLease, writerLockPathFor } from '../../../packages/core/src/writer-lease.ts'; +import { migrateCoreSchemaColumns } from '../../../packages/core/src/schema-migrations.ts'; import { createBuiltinProviderRegistry } from '../../../packages/core/src/providers/builtins.ts'; import { buildSourceCatalog, @@ -32,7 +33,7 @@ import type { SessionWorkflowRow, } from '../shared/session-detail-types.ts'; import { createSessionPatch } from '../shared/session-patch.mjs'; -import { assembleSessionMessages } from '../shared/session-detail-assembly.mjs'; +import { assembleSessionDetail } from '../shared/session-detail-assembly.mjs'; const __dirname = path.dirname(fileURLToPath(import.meta.url)); @@ -160,35 +161,12 @@ function resolveSchemaPath() { return candidates.find(p => fs.existsSync(p)); } -function ensureColumn(db, table, column, definition) { - const columns = db.prepare(`PRAGMA table_info(${table})`).all().map(c => c.name); - if (!columns.includes(column)) db.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`); -} - -function tableExists(db, table) { - return Boolean(db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name=?").get(table)); -} - -function migrateExistingColumns(db) { - if (tableExists(db, 'sessions')) ensureColumn(db, 'sessions', 'source', "TEXT DEFAULT 'claude'"); - if (tableExists(db, 'messages')) { - ensureColumn(db, 'messages', 'content_type', 'TEXT'); - ensureColumn(db, 'messages', 'is_meta', 'INTEGER DEFAULT 0'); - ensureColumn(db, 'messages', 'source', "TEXT DEFAULT 'claude'"); - } - if (tableExists(db, 'memories')) { - ensureColumn(db, 'memories', 'anchors', 'TEXT'); - ensureColumn(db, 'memories', 'deleted_at', 'TEXT'); - ensureColumn(db, 'memories', 'deleted_reason', 'TEXT'); - } -} - function migrateDb(db) { if (typeof db.exec !== 'function' || typeof db.prepare !== 'function') return; - migrateExistingColumns(db); + migrateCoreSchemaColumns(db); const schemaPath = resolveSchemaPath(); if (schemaPath) db.exec(fs.readFileSync(schemaPath, 'utf8')); - migrateExistingColumns(db); + migrateCoreSchemaColumns(db); } function closeDb() { @@ -417,7 +395,7 @@ function querySessionMessages(sessionId: string): SessionMessageRow[] { return db.prepare(` SELECT m.uuid, m.session_id, m.type, m.parent_uuid, m.timestamp, m.role, m.text, m.model, m.is_sidechain, m.agent_id, m.input_tokens, m.output_tokens, m.cwd, m.skill, m.turn_duration_ms, - m.content_type, m.is_meta, m.source + m.content_type, m.is_meta, m.visibility, m.source FROM messages m WHERE m.session_id = ? AND m.agent_id IS NULL ORDER BY m.timestamp, m.uuid `).all(sessionId) as SessionMessageRow[]; } @@ -464,9 +442,11 @@ function querySessionSnapshot(sessionId: string): SessionDetailAssemblyInput { function querySessionDisplaySnapshot(sessionId: string): SessionPatchSnapshot { const snapshot = querySessionSnapshot(sessionId); + const detail = assembleSessionDetail(snapshot); return { - messages: assembleSessionMessages(snapshot), - workflows: snapshot.workflows, + messages: detail.messages, + workflows: detail.workflows, + summaries: detail.summaries, }; } @@ -544,7 +524,7 @@ ipcMain.handle('db:getSubagentMessages', (_, agentId) => { return db.prepare(` SELECT m.uuid, m.session_id, m.type, m.parent_uuid, m.timestamp, m.role, m.text, m.model, m.is_sidechain, m.agent_id, m.input_tokens, m.output_tokens, m.cwd, m.skill, m.turn_duration_ms, - m.content_type, m.is_meta, m.source + m.content_type, m.is_meta, m.visibility, m.source FROM messages m WHERE m.agent_id = ? ORDER BY m.timestamp, m.uuid `).all(agentId); }); diff --git a/app/src/main/indexer.ts b/app/src/main/indexer.ts index 9dbb269..85859f2 100644 --- a/app/src/main/indexer.ts +++ b/app/src/main/indexer.ts @@ -11,23 +11,18 @@ import { writeProviderIndexMarkers, } from '../../../packages/core/src/provider-indexing.ts'; import { runWriteTransaction, configureConnection, betterSqliteTransactionAdapter } from '../../../packages/core/src/tx.ts'; +import { migrateCoreSchemaColumns } from '../../../packages/core/src/schema-migrations.ts'; import { acquireWriterLease, writerLockPathFor } from '../../../packages/core/src/writer-lease.ts'; import { runRetryableWriteTransaction, isBeginBusyFailure, hasUnusableTransaction } from '../../../packages/core/src/write-coordinator.ts'; import { inferProjectPath, - isDir, - readLines, - codexDbId, } from '../../../packages/core/src/parsing.ts'; const __dirname = path.dirname(fileURLToPath(import.meta.url)); const DEFAULT_CLAUDE_DIR = path.join(os.homedir(), '.claude'); -const DEFAULT_CODEX_DIR = path.join(os.homedir(), '.codex'); const DEFAULT_OBELISK_DIR = path.join(os.homedir(), '.obelisk'); const DEFAULT_DB_PATH = path.join(DEFAULT_OBELISK_DIR, 'obelisk.sqlite'); -const DEFAULT_PROJECTS_DIR = path.join(DEFAULT_CLAUDE_DIR, 'projects'); -const DEFAULT_HISTORY_PATH = path.join(DEFAULT_CLAUDE_DIR, 'history.jsonl'); function resolveSchemaPath() { const candidates = [ @@ -42,7 +37,7 @@ function resolveSchemaPath() { function installSchema(db, schemaPath = resolveSchemaPath()) { db.exec(fs.readFileSync(schemaPath, 'utf8')); - migrateDb(db); + migrateCoreSchemaColumns(db); } function openIndexDb({ dbPath = DEFAULT_DB_PATH, schemaPath = resolveSchemaPath(), DatabaseImpl = Database }: { dbPath?: string; schemaPath?: string; DatabaseImpl?: new (dbPath: string) => any } = {}) { @@ -53,21 +48,6 @@ function openIndexDb({ dbPath = DEFAULT_DB_PATH, schemaPath = resolveSchemaPath( return db; } -function ensureColumn(db, table, column, definition) { - const columns = db.prepare(`PRAGMA table_info(${table})`).all().map(c => c.name); - if (!columns.includes(column)) db.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`); -} - -function migrateDb(db) { - ensureColumn(db, 'sessions', 'source', "TEXT DEFAULT 'claude'"); - ensureColumn(db, 'messages', 'content_type', 'TEXT'); - ensureColumn(db, 'messages', 'is_meta', 'INTEGER DEFAULT 0'); - ensureColumn(db, 'messages', 'source', "TEXT DEFAULT 'claude'"); - ensureColumn(db, 'memories', 'anchors', 'TEXT'); - ensureColumn(db, 'memories', 'deleted_at', 'TEXT'); - ensureColumn(db, 'memories', 'deleted_reason', 'TEXT'); -} - function copyMemoriesFromDb(db, sourceDbPath) { if (!sourceDbPath || !fs.existsSync(sourceDbPath)) return false; db.prepare('ATTACH DATABASE ? AS previous_obelisk').run(sourceDbPath); @@ -126,23 +106,6 @@ function sessionIdFromChangedPath(projectsDir, changedPath) { return null; } -function indexCodexSessionIndex(db, { codexDir = DEFAULT_CODEX_DIR } = {}) { - const indexPath = path.join(codexDir, 'session_index.jsonl'); - if (!fs.existsSync(indexPath)) return; - readLines(indexPath, (line) => { - let item; - try { - item = JSON.parse(line); - } catch (error) { - console.warn(`Warning: malformed Codex session index line: ${(error as Error).message}`); - return; - } - if (!item.id || !item.thread_name) return; - db.prepare('UPDATE sessions SET title=COALESCE(title, ?), ended_at=COALESCE(ended_at, ?) WHERE id=? AND source=?') - .run(item.thread_name, item.updated_at || null, codexDbId(item.id), 'codex'); - }); -} - function refreshSessionProjectPaths(db) { const sessions = db.prepare('SELECT id, project FROM sessions').all(); const cwdStmt = db.prepare(` @@ -158,61 +121,6 @@ function refreshSessionProjectPaths(db) { } } -function indexWorkflows(db, { projectsDir = DEFAULT_PROJECTS_DIR } = {}) { - if (!fs.existsSync(projectsDir)) return; - let projects; - try { projects = fs.readdirSync(projectsDir); } catch { return; } - for (const proj of projects) { - const pp = path.join(projectsDir, proj); - if (!isDir(pp)) continue; - let entries; - try { entries = fs.readdirSync(pp); } catch { continue; } - for (const sd of entries) { - const wd = path.join(pp, sd, 'workflows'); - if (!isDir(wd)) continue; - let wfFiles; - try { wfFiles = fs.readdirSync(wd); } catch { continue; } - for (const f of wfFiles) { - if (!f.endsWith('.json')) continue; - let wf; - try { - wf = JSON.parse(fs.readFileSync(path.join(wd, f), 'utf8')); - } catch (error) { - console.warn(`Warning: failed to read workflow ${f}: ${(error as Error).message}`); - continue; - } - if (!wf.runId) continue; - const ac = db.prepare('SELECT COUNT(*) as c FROM workflow_agents WHERE run_id=?').get(wf.runId); - db.prepare('INSERT OR REPLACE INTO workflows (run_id,session_id,task_id,script,result_json,timestamp,agent_count,duration_ms,total_tokens,status,workflow_name) VALUES(?,?,?,?,?,?,?,?,?,?,?)').run( - wf.runId, sd, wf.taskId||null, wf.script||null, - wf.result ? JSON.stringify(wf.result) : null, wf.timestamp||null, ac?.c||0, - wf.durationMs||null, wf.totalTokens||null, wf.status||null, wf.workflowName||null); - const progress = wf.workflowProgress || []; - for (const item of progress) { - if (item.type !== 'workflow_agent' || !item.agentId) continue; - db.prepare('UPDATE workflow_agents SET phase=?, label=?, model=?, state=?, duration_ms=?, tokens=?, tool_calls=? WHERE agent_id=?').run( - item.phaseTitle||null, item.label||null, item.model||null, item.state||null, - item.durationMs||null, item.tokens||null, item.toolCalls||null, 'agent-' + item.agentId); - } - } - } - } -} - -function indexHistory(db, { historyPath = DEFAULT_HISTORY_PATH } = {}) { - if (!fs.existsSync(historyPath)) return; - readLines(historyPath, (line) => { - let item; - try { - item = JSON.parse(line); - } catch (error) { - console.warn(`Warning: malformed history line: ${(error as Error).message}`); - return; - } - if (item.sessionId && item.title) db.prepare('UPDATE sessions SET title=? WHERE id=? AND title IS NULL').run(item.title, item.sessionId); - }); -} - function rebuildFts(db) { db.exec("INSERT INTO messages_fts(messages_fts) VALUES('rebuild')"); db.exec("INSERT INTO memories_fts(memories_fts) VALUES('rebuild')"); @@ -286,7 +194,6 @@ interface BuildIndexOptions { claudeDir?: string; codexDir?: string; projectsDir?: string; - historyPath?: string; dbPath?: string; schemaPath?: string; DatabaseImpl?: new (dbPath: string) => any; @@ -339,7 +246,6 @@ function buildIndex({ claudeDir = DEFAULT_CLAUDE_DIR, codexDir = path.join(path.dirname(claudeDir), '.codex'), projectsDir = path.join(claudeDir, 'projects'), - historyPath = path.join(claudeDir, 'history.jsonl'), dbPath = DEFAULT_DB_PATH, schemaPath = resolveSchemaPath(), DatabaseImpl = Database, @@ -489,10 +395,7 @@ function buildIndex({ // index would otherwise be left inconsistent). try { runRetryableWriteTransaction(txDb, () => { - indexWorkflows(db, { projectsDir }); refreshSessionProjectPaths(db); - indexHistory(db, { historyPath }); - indexCodexSessionIndex(db, { codexDir }); if (messageFtsTriggersDropped) installSchema(db, schemaPath); ftsRebuilt = ensureFtsReady(db, { force }); writeIndexMarker(db, '__last_build__'); diff --git a/app/src/renderer/src/data.js b/app/src/renderer/src/data.js index 964a2ef..f7a73b7 100644 --- a/app/src/renderer/src/data.js +++ b/app/src/renderer/src/data.js @@ -7,7 +7,7 @@ import { applySessionPatch, createSessionPatchCursor, } from '../../shared/session-patch.mjs'; -import { assembleSessionMessages } from '../../shared/session-detail-assembly.mjs'; +import { assembleSessionDetail } from '../../shared/session-detail-assembly.mjs'; const sessionMessageSnapshots = new Map(); const MAX_SESSION_MESSAGE_SNAPSHOTS = 3; @@ -25,6 +25,7 @@ function sessionMetadata(session) { const metadata = { ...session }; delete metadata.messages; delete metadata.workflow; + delete metadata.summaries; return markRaw(metadata); } @@ -96,9 +97,11 @@ export async function loadSessionDetail(sessionId) { window.obelisk.getSessionWorkflows(sessionId), window.obelisk.getSessionSummaries(sessionId), ]); + const detail = assembleSessionDetail({ messages, toolCalls, toolResults, subagents, workflows, summaries }); const snapshot = { - messages: assembleSessionMessages({ messages, toolCalls, toolResults, subagents, workflows }), - workflows, + messages: detail.messages, + workflows: detail.workflows, + summaries: detail.summaries, }; const metadata = sessionMetadata(state.sessions.find(candidate => candidate.id === sessionId)); rememberSessionMessageSnapshot(sessionId, { @@ -154,13 +157,14 @@ export function getCachedSessionDetail(sessionId) { }); } -function commitSessionDetail(sessionId, { messages, workflows = [] }, { updateStore, metadata = null }) { +function commitSessionDetail(sessionId, { messages, workflows = [], summaries = [] }, { updateStore, metadata = null }) { const session = state.sessions.find(candidate => candidate.id === sessionId); const assembled = { ...(session || {}), ...(metadata || {}), id: sessionId, messages: markRaw(messages), + summaries: markRaw(summaries), }; if (workflows.length > 0) assembled.workflow = workflows[0]; @@ -181,13 +185,13 @@ export async function loadSubagentDetail(agentId) { window.obelisk.getSubagentToolCalls(agentId), window.obelisk.getSubagentToolResults(agentId), ]); - return assembleSessionMessages({ + return assembleSessionDetail({ messages, toolCalls, toolResults, subagents: [], workflows: [], - }); + }).messages; } const TEXT_LIMIT = 10000; diff --git a/app/src/shared/session-detail-assembly.mjs b/app/src/shared/session-detail-assembly.mjs index 9b11b47..2e0b380 100644 --- a/app/src/shared/session-detail-assembly.mjs +++ b/app/src/shared/session-detail-assembly.mjs @@ -1,173 +1,9 @@ -// @ts-check +import { assembleSessionDetail } from '../../../packages/core/src/session-detail.ts'; -/** @typedef {import('./session-detail-types.ts').AssembledMessage} AssembledMessage */ -/** @typedef {import('./session-detail-types.ts').AssembledToolCall} AssembledToolCall */ -/** @typedef {import('./session-detail-types.ts').SessionDetailAssemblyInput} SessionDetailAssemblyInput */ -/** @typedef {import('./session-detail-types.ts').SessionSubagentRow} SessionSubagentRow */ -/** @typedef {import('./session-detail-types.ts').SessionToolResultRow} SessionToolResultRow */ +export { assembleSessionDetail }; -/** - * @param {SessionDetailAssemblyInput} input - * @returns {AssembledMessage[]} - */ -export function assembleSessionMessages({ messages, toolCalls, toolResults, subagents, workflows }) { - const resultsByCallId = /** @type {Map} */ (new Map()); - for (const result of toolResults || []) resultsByCallId.set(result.tool_use_id, result); - - const subagentsByCallId = /** @type {Map} */ (new Map()); - for (const subagent of subagents || []) { - if (subagent.parent_tool_use_id) subagentsByCallId.set(subagent.parent_tool_use_id, subagent); - } - - const callsByMessageUuid = /** @type {Map} */ (new Map()); - for (const toolCall of toolCalls || []) { - const call = /** @type {AssembledToolCall} */ ({ - id: toolCall.id, - name: toolCall.name, - input_json: toolCall.input_json, - result: resultsByCallId.get(toolCall.id) || null, - }); - const subagent = subagentsByCallId.get(toolCall.id); - if (subagent) { - call.subagent = { - agent_id: subagent.agent_id, - agent_type: subagent.agent_type, - description: subagent.description, - }; - } - const messageUuid = toolCall.message_uuid; - const calls = callsByMessageUuid.get(messageUuid) || []; - calls.push(call); - callsByMessageUuid.set(messageUuid, calls); - } - - for (const workflow of workflows || []) { - for (const calls of callsByMessageUuid.values()) { - for (const call of calls) { - if (call.name !== 'Workflow' || call.workflow) continue; - const resultText = call.result?.content || ''; - if (!resultText.includes(workflow.run_id) && !resultText.includes(workflow.workflow_name || '___none___')) continue; - call.workflow = { - run_id: workflow.run_id, - workflow_name: workflow.workflow_name, - status: workflow.status, - duration_ms: workflow.duration_ms, - total_tokens: workflow.total_tokens, - agent_count: workflow.agent_count, - agents: (workflow.agents || []).map(agent => ({ - agent_id: agent.agent_id, - phase: agent.phase, - label: agent.label, - state: agent.state, - tokens: agent.tokens, - duration_ms: agent.duration_ms, - })), - }; - } - } - } - - const metaPattern = /^\s*<(task-notification|command-name|local-command|system-reminder)/; - const rawAssembled = (messages || []).map(message => { - const assembled = /** @type {AssembledMessage} */ ({ - uuid: message.uuid, - type: message.type || message.role, - timestamp: message.timestamp, - text: message.text, - content_type: message.content_type || null, - is_meta: message.is_meta || (message.text && metaPattern.test(message.text) ? 1 : 0), - }); - const calls = callsByMessageUuid.get(message.uuid); - if (calls?.length) assembled.tool_calls = calls; - return assembled; - }); - - const assembledMessages = /** @type {AssembledMessage[]} */ ([]); - for (let index = 0; index < rawAssembled.length; index++) { - const message = rawAssembled[index]; - if (message.content_type === 'tool_result') continue; - - if (message.type === 'assistant' && message.content_type === 'thinking') { - const thinkingParts = [message.text || '']; - let nextIndex = index + 1; - while ( - nextIndex < rawAssembled.length - && rawAssembled[nextIndex].type === 'assistant' - && rawAssembled[nextIndex].content_type === 'thinking' - ) { - thinkingParts.push(rawAssembled[nextIndex].text || ''); - nextIndex++; - } - if ( - nextIndex < rawAssembled.length - && rawAssembled[nextIndex].type === 'assistant' - && rawAssembled[nextIndex].content_type !== 'thinking' - ) { - rawAssembled[nextIndex]._thinking = thinkingParts.join('\n\n'); - index = nextIndex - 1; - continue; - } - assembledMessages.push({ ...message, text: thinkingParts.join('\n\n'), content_type: 'thinking' }); - index = nextIndex - 1; - continue; - } - - if (message.type === 'assistant' && message.content_type === 'tool_use') { - const merged = /** @type {AssembledMessage} */ ({ - ...message, - tool_calls: [...(message.tool_calls || [])], - }); - const mergedCalls = merged.tool_calls || []; - if (message._thinking) merged._thinking = message._thinking; - const skillOnly = mergedCalls.length === 1 && mergedCalls[0].name === 'Skill'; - let nextIndex = index + 1; - while (nextIndex < rawAssembled.length) { - const next = rawAssembled[nextIndex]; - if (next.content_type === 'tool_result') { - nextIndex++; - continue; - } - if (next.is_meta && next.text && next.text.includes('Base directory for this skill')) { - merged._skillMd = next.text; - nextIndex++; - continue; - } - if (!skillOnly && next.type === 'assistant' && next.content_type === 'tool_use') { - if (next.tool_calls) mergedCalls.push(...next.tool_calls); - if (next.text && !merged.text) merged.text = next.text; - nextIndex++; - continue; - } - break; - } - assembledMessages.push(merged); - index = nextIndex - 1; - continue; - } - - const output = /** @type {AssembledMessage} */ ({ ...message }); - if (message._thinking) output._thinking = message._thinking; - if (message.type === 'assistant' && message.content_type !== 'tool_use' && message.content_type !== 'thinking') { - if (!output.tool_calls) output.tool_calls = []; - let nextIndex = index + 1; - while (nextIndex < rawAssembled.length) { - const next = rawAssembled[nextIndex]; - if (next.content_type === 'tool_result') { - nextIndex++; - continue; - } - if (next.type === 'assistant' && next.content_type === 'tool_use') { - if (next.tool_calls) output.tool_calls.push(...next.tool_calls); - nextIndex++; - continue; - } - break; - } - if (!output.tool_calls.length) delete output.tool_calls; - index = nextIndex - 1; - } - assembledMessages.push(output); - } - - return assembledMessages; +// Compatibility for local/generated app tooling; production callers use the +// single Core assembleSessionDetail seam. +export function assembleSessionMessages(input) { + return assembleSessionDetail(input).messages; } diff --git a/app/src/shared/session-detail-types.ts b/app/src/shared/session-detail-types.ts index 00073e4..98fb280 100644 --- a/app/src/shared/session-detail-types.ts +++ b/app/src/shared/session-detail-types.ts @@ -1,105 +1,13 @@ -export interface SessionMessageRow { - [key: string]: unknown; - uuid: string; - type?: string | null; - role?: string | null; - timestamp?: string | null; - text?: string | null; - content_type?: string | null; - is_meta?: number | boolean | null; -} - -export interface SessionToolResultRow { - [key: string]: unknown; - tool_use_id: string; - content?: string | null; -} - -export interface SessionToolCallRow { - [key: string]: unknown; - id: string; - message_uuid: string; - name: string; - input_json?: string | null; -} - -export interface SessionSubagentRow { - [key: string]: unknown; - agent_id: string; - parent_tool_use_id?: string | null; - agent_type?: string | null; - description?: string | null; -} - -export interface SessionWorkflowAgentRow { - [key: string]: unknown; - agent_id: string; - phase?: string | null; - label?: string | null; - state?: string | null; - tokens?: number | null; - duration_ms?: number | null; -} - -export interface SessionWorkflowRow { - [key: string]: unknown; - run_id: string; - workflow_name?: string | null; - status?: string | null; - duration_ms?: number | null; - total_tokens?: number | null; - agent_count?: number | null; - agents?: SessionWorkflowAgentRow[] | null; -} - -export interface SessionSummaryRow { - [key: string]: unknown; - id: string | number; -} - -export interface SessionDetailAssemblyInput { - messages?: SessionMessageRow[]; - toolCalls?: SessionToolCallRow[]; - toolResults?: SessionToolResultRow[]; - subagents?: SessionSubagentRow[]; - workflows?: SessionWorkflowRow[]; - summaries?: SessionSummaryRow[]; -} - -export interface AssembledToolCall { - [key: string]: unknown; - id: string; - name: string; - input_json?: string | null; - result: SessionToolResultRow | null; - subagent?: { - agent_id: string; - agent_type?: string | null; - description?: string | null; - }; - workflow?: { - run_id: string; - workflow_name?: string | null; - status?: string | null; - duration_ms?: number | null; - total_tokens?: number | null; - agent_count?: number | null; - agents: Array<{ - agent_id: string; - phase?: string | null; - label?: string | null; - state?: string | null; - tokens?: number | null; - duration_ms?: number | null; - }>; - }; -} - -export interface AssembledMessage extends SessionMessageRow { - type?: string | null; - content_type?: string | null; - is_meta?: number | boolean | null; - tool_calls?: AssembledToolCall[]; - _thinking?: string; - _skillMd?: string; -} +export type { + AssembledMessage, + AssembledToolCall, + SessionDetailRows as SessionDetailAssemblyInput, + SessionDetailSessionRow, + SessionMessageRow, + SessionSubagentRow, + SessionSummaryRow, + SessionToolCallRow, + SessionToolResultRow, + SessionWorkflowAgentRow, + SessionWorkflowRow, +} from '../../../packages/core/src/session-detail.ts'; diff --git a/app/tests/electron-session-reader-state.mjs b/app/tests/electron-session-reader-state.mjs index b11c240..d060efc 100644 --- a/app/tests/electron-session-reader-state.mjs +++ b/app/tests/electron-session-reader-state.mjs @@ -3,7 +3,7 @@ import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; import { setTimeout as delay } from 'node:timers/promises'; import { createSessionPatch } from '../src/shared/session-patch.mjs'; -import { assembleSessionMessages } from '../src/shared/session-detail-assembly.mjs'; +import { assembleSessionDetail } from '../src/shared/session-detail-assembly.mjs'; const here = dirname(fileURLToPath(import.meta.url)); const appRoot = join(here, '..'); @@ -106,13 +106,13 @@ function registerHandlers() { ipcMain.handle('db:getSessionPatch', (_event, sessionId, cursor) => { const fixture = fixtures[sessionId]; const patch = createSessionPatch({ - messages: assembleSessionMessages({ + messages: assembleSessionDetail({ messages: fixture.messages, toolCalls: fixture.toolCalls, toolResults: fixture.toolResults, subagents: [], workflows: [], - }), + }).messages, workflows: [], }, cursor); return { ...patch, session: summary(sessionId) }; diff --git a/app/tests/electron-session-virtualization.mjs b/app/tests/electron-session-virtualization.mjs index 6e880f3..41c0148 100644 --- a/app/tests/electron-session-virtualization.mjs +++ b/app/tests/electron-session-virtualization.mjs @@ -5,7 +5,7 @@ import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; import { setTimeout as delay } from 'node:timers/promises'; import { createSessionPatch } from '../src/shared/session-patch.mjs'; -import { assembleSessionMessages } from '../src/shared/session-detail-assembly.mjs'; +import { assembleSessionDetail } from '../src/shared/session-detail-assembly.mjs'; const here = dirname(fileURLToPath(import.meta.url)); const appRoot = join(here, '..'); @@ -530,7 +530,7 @@ function registerHandlers() { nextPatchDelayMs = 0; if (delayMs > 0) await delay(delayMs); const patch = createSessionPatch({ - messages: assembleSessionMessages({ messages, toolCalls, toolResults, subagents: [], workflows: [] }), + messages: assembleSessionDetail({ messages, toolCalls, toolResults, subagents: [], workflows: [] }).messages, workflows: [], }, cursor); ipcReads.patchMessageRows.push(patch.changes.messages.length); diff --git a/docs/adr/0001-parse-core-and-persist-layers.md b/docs/adr/0001-parse-core-and-persist-layers.md index 4a27b61..5e82870 100644 --- a/docs/adr/0001-parse-core-and-persist-layers.md +++ b/docs/adr/0001-parse-core-and-persist-layers.md @@ -33,7 +33,9 @@ binding-agnostic and does not need a per-binding implementation. `IndexUnit` when correctness requires whole-unit semantics — for example, Codex duplicate reconciliation or Kimi `context.undo` / `context.clear` replay. Each adapter maps its own resume/change semantics onto the existing - `mtime` and `lines_processed` cursor pair in `index_state`. + `mtime` and `lines_processed` cursor pair in `index_state`. The emitted + `TranscriptRecord` stream is also the input to provider-independent session + detail assembly; see ADR-0007. - **Persist axis — one shared orchestration.** A single provider-agnostic, binding-agnostic layer consumes records from any adapter and writes them: incremental `index_state` bookkeeping, FTS maintenance, and the canonical @@ -56,9 +58,9 @@ main process migrates to ESM (ADR-0003) to import the shared core. The real work is disentangling the currently interleaved parse-and-write inside `indexJsonl` / `indexCodexJsonl` into (pure adapter parse) + (shared persist). -The SQLite schema and normalized `IndexRecord` union are the stable center of -the design. Provider-only concepts are either projected lossily into that -language or ignored; they do not add provider columns or tables. The registry, +The normalized `TranscriptRecord` union is the stable center of the design, and +SQLite is one serialization adapter for it. Provider-only concepts are either +projected lossily into that language or ignored. The registry, not provider switches, drives both indexers, watcher roots, persisted source roots, source catalog/UI labels and colors, and raw-record routing. Adding Pi therefore changes the Pi adapter, its registration, and its conformance tests; diff --git a/docs/adr/0007-canonical-transcript-session-detail-seam.md b/docs/adr/0007-canonical-transcript-session-detail-seam.md new file mode 100644 index 0000000..54b496a --- /dev/null +++ b/docs/adr/0007-canonical-transcript-session-detail-seam.md @@ -0,0 +1,40 @@ +# Canonical transcript records are the session-detail seam + +**Context.** Provider adapters originally emitted database-shaped records, while +the desktop app reconstructed presentation semantics after querying SQLite. +Although that reconstruction had no explicit provider switch, it still inferred +metadata from raw message text. As more providers are added, those heuristics +would make provider semantics leak into a shared presentation module and allow +the direct parse path to drift from the persisted path. + +**Decision.** Every provider adapter emits a canonical `TranscriptRecord` +stream. The adapter owns all source-specific interpretation: duplicate raw +events, stable identities, tool relationships, message classification, and +visibility. `visibility` is separate from `is_meta`: hidden transport context is +omitted from session detail, while visible system evidence can remain a metadata +card. Presentation-sensitive concepts are explicit canonical fields: tool calls +carry a presentation class, Skill instructions carry a content type, and +workflows carry their parent tool-call identity. + +The Core `assembleSessionDetail(input)` module is the only session-detail seam. +It accepts either a provider's complete transcript stream from a fresh parse +(`cursor = null`) or table-shaped rows after a persistence round-trip. A delta +parse cannot produce a complete detail snapshot without prior state, so the +assembler rejects a `SessionRecord` whose `countMode` is `delta`; incremental UI +updates use the existing snapshot/patch seam. Its internal row adapter restores +the canonical record language before assembly. The implementation may sort, +group thinking, and attach tool results, subagents, and workflows, but it never +checks the provider and never parses message text to recover provider semantics. +Tool names are likewise display data, not assembly control flow. + +The persist layer only serializes transcript records and cursor state. SQLite is +not the source of transcript semantics, and a persistence round-trip must not +change the assembled result. + +**Consequences.** A new provider is complete only when its canonical transcript +can pass directly through `assembleSessionDetail`. Provider conformance tests +cover that seam, while persistence tests verify that canonical classification +survives a database round-trip. Codex-owned normalization now classifies hidden +context envelopes and structurally removes image wrappers before duplicate +reconciliation. Adding another provider does not add branches to the app's +session-detail code. diff --git a/packages/core/package.json b/packages/core/package.json index 2becca4..f3bb2e8 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -11,6 +11,7 @@ "./parsing": "./dist/parsing.js", "./persist": "./dist/persist.js", "./provider-indexing": "./dist/provider-indexing.js", + "./session-detail": "./dist/session-detail.js", "./providers/claude": "./dist/providers/claude.js", "./providers/codex": "./dist/providers/codex.js", "./providers/kimi": "./dist/providers/kimi.js", @@ -18,6 +19,7 @@ "./providers/builtins": "./dist/providers/builtins.js", "./providers/types": "./dist/providers/types.js", "./query": "./dist/query.js", + "./schema-migrations": "./dist/schema-migrations.js", "./sqlite-types": "./dist/sqlite-types.js", "./tx": "./dist/tx.js", "./write-coordinator": "./dist/write-coordinator.js", diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index 5f6ad25..425e8dc 100644 --- a/packages/core/src/db.ts +++ b/packages/core/src/db.ts @@ -2,6 +2,7 @@ import { createRequire } from 'node:module'; import { CLAUDE_DIR, CODEX_DIR, TEXT_LIMIT, trunc, truncJson, extractText, extractContentType, extractMessageIsMeta, filePath, isDir, readLines } from './parsing.ts'; import { configureConnection } from './tx.ts'; +import { migrateCoreSchemaColumns } from './schema-migrations.ts'; import type { NodeSqliteDb, SqliteDb } from './sqlite-types.ts'; const require = createRequire(import.meta.url); const fs = require('node:fs'); @@ -26,9 +27,9 @@ function openDb(): NodeSqliteDb { fs.mkdirSync(path.dirname(DB_PATH), { recursive: true }); const db = new DatabaseSync(DB_PATH); configureConnection(db, { busyTimeoutMs: 250 }); - migrateExistingColumns(db); + migrateCoreSchemaColumns(db); db.exec(SCHEMA); - migrateDb(db); + migrateCoreSchemaColumns(db); return db; } @@ -44,33 +45,6 @@ function openWriterLeaseDb(lockPath: string): NodeSqliteDb { return new DatabaseSync(lockPath); } -function ensureColumn(db: SqliteDb, table: string, column: string, definition: string): void { - const columns = db.prepare(`PRAGMA table_info(${table})`).all().map(c => c.name); - if (!columns.includes(column)) db.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`); -} - -function tableExists(db: SqliteDb, table: string): boolean { - return Boolean(db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name=?").get(table)); -} - -function migrateExistingColumns(db: SqliteDb): void { - if (tableExists(db, 'sessions')) ensureColumn(db, 'sessions', 'source', "TEXT DEFAULT 'claude'"); - if (tableExists(db, 'messages')) { - ensureColumn(db, 'messages', 'content_type', 'TEXT'); - ensureColumn(db, 'messages', 'is_meta', 'INTEGER DEFAULT 0'); - ensureColumn(db, 'messages', 'source', "TEXT DEFAULT 'claude'"); - } - if (tableExists(db, 'memories')) { - ensureColumn(db, 'memories', 'anchors', 'TEXT'); - ensureColumn(db, 'memories', 'deleted_at', 'TEXT'); - ensureColumn(db, 'memories', 'deleted_reason', 'TEXT'); - } -} - -function migrateDb(db: SqliteDb): void { - migrateExistingColumns(db); -} - function rebuildMemoryFts(db: SqliteDb): void { db.exec("INSERT INTO memories_fts(memories_fts) VALUES('rebuild')"); } diff --git a/packages/core/src/indexer.ts b/packages/core/src/indexer.ts index ca87be4..83663e0 100644 --- a/packages/core/src/indexer.ts +++ b/packages/core/src/indexer.ts @@ -1,9 +1,6 @@ // Passive-pull indexing orchestration for the Core package. import { DB_PATH, openDb, openReadDb, openWriterLeaseDb, rebuildMemoryFts } from './db.ts'; -import { - CLAUDE_DIR, CODEX_DIR, PROJECTS_DIR, fs, path, isDir, readLines, - inferProjectPath, codexDbId, -} from './parsing.ts'; +import { fs, inferProjectPath } from './parsing.ts'; import { createProviderIndexPlan, indexProviderPlan, @@ -15,10 +12,6 @@ import { runRetryableWriteTransaction, isBeginBusyFailure, hasUnusableTransactio import { createBuiltinProviderRegistry } from './providers/builtins.ts'; import type { NodeSqliteDb, SqliteRow } from './sqlite-types.ts'; -const HISTORY_PATH = path.join(CLAUDE_DIR, 'history.jsonl'); - -type JsonRecord = Record; - interface SkippedFile { path: string; error: string; @@ -35,23 +28,6 @@ function errorMessage(error: unknown): string { } -function indexCodexSessionIndex(db: NodeSqliteDb): void { - const indexPath = path.join(CODEX_DIR, 'session_index.jsonl'); - if (!fs.existsSync(indexPath)) return; - readLines(indexPath, (line) => { - let item: JsonRecord; - try { - item = JSON.parse(line); - } catch (e) { - process.stderr.write(`Warning: malformed Codex session index line: ${errorMessage(e)}\n`); - return; - } - if (!item.id || !item.thread_name) return; - db.prepare('UPDATE sessions SET title=COALESCE(title, ?), ended_at=COALESCE(ended_at, ?) WHERE id=? AND source=?') - .run(item.thread_name, item.updated_at || null, codexDbId(item.id), 'codex'); - }); -} - function refreshSessionProjectPaths(db: NodeSqliteDb): void { const sessions = db.prepare('SELECT id, project FROM sessions').all(); const cwdStmt = db.prepare(` @@ -68,61 +44,6 @@ function refreshSessionProjectPaths(db: NodeSqliteDb): void { } } -function indexWorkflows(db: NodeSqliteDb): void { - if (!fs.existsSync(PROJECTS_DIR)) return; - let projects; - try { projects = fs.readdirSync(PROJECTS_DIR); } catch { return; } - for (const proj of projects) { - const pp = path.join(PROJECTS_DIR, proj); - if (!isDir(pp)) continue; - let entries; - try { entries = fs.readdirSync(pp); } catch { continue; } - for (const sd of entries) { - const wd = path.join(pp, sd, 'workflows'); - if (!isDir(wd)) continue; - let wfFiles; - try { wfFiles = fs.readdirSync(wd); } catch { continue; } - for (const f of wfFiles) { - if (!f.endsWith('.json')) continue; - let wf: JsonRecord; - try { - wf = JSON.parse(fs.readFileSync(path.join(wd, f), 'utf8')); - } catch (e) { - process.stderr.write(`Warning: failed to read workflow ${f}: ${errorMessage(e)}\n`); - continue; - } - if (!wf.runId) continue; - const ac = db.prepare('SELECT COUNT(*) as c FROM workflow_agents WHERE run_id=?').get(wf.runId); - db.prepare('INSERT OR REPLACE INTO workflows (run_id,session_id,task_id,script,result_json,timestamp,agent_count,duration_ms,total_tokens,status,workflow_name) VALUES(?,?,?,?,?,?,?,?,?,?,?)').run( - wf.runId, sd, wf.taskId||null, wf.script||null, - wf.result ? JSON.stringify(wf.result) : null, wf.timestamp||null, ac?.c||0, - wf.durationMs||null, wf.totalTokens||null, wf.status||null, wf.workflowName||null); - const progress = wf.workflowProgress || []; - for (const item of progress) { - if (item.type !== 'workflow_agent' || !item.agentId) continue; - db.prepare('UPDATE workflow_agents SET phase=?, label=?, model=?, state=?, duration_ms=?, tokens=?, tool_calls=? WHERE agent_id=?').run( - item.phaseTitle||null, item.label||null, item.model||null, item.state||null, - item.durationMs||null, item.tokens||null, item.toolCalls||null, 'agent-' + item.agentId); - } - } - } - } -} - -function indexHistory(db: NodeSqliteDb): void { - if (!fs.existsSync(HISTORY_PATH)) return; - readLines(HISTORY_PATH, (line) => { - let item: JsonRecord; - try { - item = JSON.parse(line); - } catch (e) { - process.stderr.write(`Warning: malformed history line: ${errorMessage(e)}\n`); - return; - } - if (item.sessionId && item.title) db.prepare('UPDATE sessions SET title=? WHERE id=? AND title IS NULL').run(item.title, item.sessionId); - }); -} - const BUILD_DEBOUNCE_MS = 30000; const APP_HEARTBEAT_FRESH_MS = 60000; @@ -221,10 +142,7 @@ function buildIndex({ force = false }: { force?: boolean } = {}) { // the build (a half-finalized index would be inconsistent). try { runRetryableWriteTransaction(txDb, () => { - indexWorkflows(db); refreshSessionProjectPaths(db); - indexHistory(db); - indexCodexSessionIndex(db); db.exec("INSERT INTO messages_fts(messages_fts) VALUES('rebuild')"); rebuildMemoryFts(db); db.prepare("INSERT OR REPLACE INTO index_state (jsonl_path, mtime, lines_processed) VALUES ('__last_build__', ?, 0)").run(Date.now()); diff --git a/packages/core/src/parsing.ts b/packages/core/src/parsing.ts index 7ecc9e3..b8b34d2 100644 --- a/packages/core/src/parsing.ts +++ b/packages/core/src/parsing.ts @@ -85,6 +85,7 @@ function extractContentType(content: JsonValue): string { } const COMMAND_ENVELOPE_RE = /^\s*([^<]+<\/command-name>|<(?:task-notification|system-reminder)\b|' + && image?.type === 'input_image' + && close?.type === 'input_text' + && typeof close.text === 'string' + && close.text.trim() === '' + ) { + index += 2; + continue; + } if (typeof block?.text === 'string') parts.push(block.text); } return parts.length ? parts.join('\n') : null; @@ -340,7 +360,7 @@ function codexToolOutput(payload: JsonRecord): string | null { export { fs, path, os, CLAUDE_DIR, CODEX_DIR, PROJECTS_DIR, CODEX_SESSIONS_DIR, TEXT_LIMIT, - trunc, truncJson, extractText, extractContentType, extractMessageIsMeta, filePath, isDir, readLines, + trunc, truncJson, extractText, extractContentType, extractMessageIsMeta, isSkillInstructions, filePath, isDir, readLines, legacyProjectPathFromSlug, normalizeObservedCwd, projectSlugFromPath, inferProjectPath, discoverJsonlFiles, discoverCodexJsonlFiles, codexDbId, codexRawId, codexLineUuid, codexCallId, codexParentThreadId, codexIsGuardianThread, diff --git a/packages/core/src/persist.ts b/packages/core/src/persist.ts index fb6a392..44c9fb2 100644 --- a/packages/core/src/persist.ts +++ b/packages/core/src/persist.ts @@ -1,6 +1,6 @@ // Shared Core persist layer (see docs/adr/0001). // -// Provider-agnostic and binding-agnostic: it consumes the IndexRecord stream +// Provider-agnostic and binding-agnostic: it consumes the TranscriptRecord stream // from any adapter's parse() and writes rows into the injected database handle // (node:sqlite for the CLI, better-sqlite3 for the app — they share the // prepare/run/get API). It is the ONLY layer that touches the database and the @@ -12,7 +12,7 @@ // turn-duration is a targeted UPDATE; delete-session cascades. The generator's // return value is the new cursor, persisted verbatim into index_state. -import type { Cursor, IndexRecord, IndexUnit } from './providers/types.ts'; +import type { Cursor, TranscriptRecord, IndexUnit } from './providers/types.ts'; import type { SqliteDb } from './sqlite-types.ts'; const minStr = (a: string | null, b: string | null) => (a == null ? b : b == null ? a : a < b ? a : b); @@ -21,16 +21,17 @@ const maxStr = (a: string | null, b: string | null) => (a == null ? b : b == nul function statements(db: SqliteDb) { return { msg: db.prepare(` - INSERT INTO messages (uuid,session_id,type,parent_uuid,timestamp,role,text,content_type,is_meta,model,is_sidechain,agent_id,input_tokens,output_tokens,cwd,skill,source) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + INSERT INTO messages (uuid,session_id,type,parent_uuid,timestamp,role,text,content_type,is_meta,visibility,model,is_sidechain,agent_id,input_tokens,output_tokens,cwd,skill,source) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(uuid) DO UPDATE SET session_id=excluded.session_id, type=excluded.type, parent_uuid=excluded.parent_uuid, timestamp=excluded.timestamp, role=excluded.role, text=excluded.text, - content_type=excluded.content_type, is_meta=excluded.is_meta, model=excluded.model, + content_type=excluded.content_type, is_meta=excluded.is_meta, + visibility=excluded.visibility, model=excluded.model, is_sidechain=excluded.is_sidechain, agent_id=excluded.agent_id, input_tokens=excluded.input_tokens, output_tokens=excluded.output_tokens, cwd=excluded.cwd, skill=excluded.skill, source=excluded.source`), - tc: db.prepare('INSERT OR REPLACE INTO tool_calls (id,message_uuid,session_id,name,input_json,file_path) VALUES (?,?,?,?,?,?)'), + tc: db.prepare('INSERT OR REPLACE INTO tool_calls (id,message_uuid,session_id,name,presentation,input_json,file_path) VALUES (?,?,?,?,?,?,?)'), tr: db.prepare('INSERT OR REPLACE INTO tool_results (tool_use_id,message_uuid,session_id,content,file_path,is_error) VALUES (?,?,?,?,?,?)'), sum: db.prepare('INSERT OR REPLACE INTO summaries (id,session_id,timestamp,source,content) VALUES (?,?,?,?,?)'), ses: db.prepare('INSERT OR REPLACE INTO sessions (id,title,project,project_path,started_at,ended_at,git_branch,version,message_count,jsonl_path,source) VALUES (?,?,?,?,?,?,?,?,?,?,?)'), @@ -44,6 +45,25 @@ function statements(db: SqliteDb) { description=COALESCE(excluded.description, subagents.description), duration_ms=COALESCE(excluded.duration_ms, subagents.duration_ms), total_tokens=COALESCE(excluded.total_tokens, subagents.total_tokens)`), + wf: db.prepare(` + INSERT OR REPLACE INTO workflows + (run_id,session_id,parent_tool_use_id,task_id,script,result_json,timestamp,agent_count,duration_ms,total_tokens,status,workflow_name) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?)`), + wa: db.prepare(` + INSERT INTO workflow_agents + (agent_id,run_id,session_id,agent_type,description,phase,label,model,state,duration_ms,tokens,tool_calls) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(agent_id) DO UPDATE SET + run_id=excluded.run_id, session_id=excluded.session_id, + agent_type=COALESCE(excluded.agent_type, workflow_agents.agent_type), + description=COALESCE(excluded.description, workflow_agents.description), + phase=COALESCE(excluded.phase, workflow_agents.phase), + label=COALESCE(excluded.label, workflow_agents.label), + model=COALESCE(excluded.model, workflow_agents.model), + state=COALESCE(excluded.state, workflow_agents.state), + duration_ms=COALESCE(excluded.duration_ms, workflow_agents.duration_ms), + tokens=COALESCE(excluded.tokens, workflow_agents.tokens), + tool_calls=COALESCE(excluded.tool_calls, workflow_agents.tool_calls)`), turn: db.prepare('UPDATE messages SET turn_duration_ms=? WHERE uuid=?'), idx: db.prepare('INSERT OR REPLACE INTO index_state (jsonl_path,mtime,lines_processed) VALUES (?,?,?)'), getSession: db.prepare('SELECT * FROM sessions WHERE id=?'), @@ -56,22 +76,24 @@ function deleteSession(db: SqliteDb, sessionId: string) { db.prepare('DELETE FROM tool_calls WHERE session_id=? OR message_uuid IN (SELECT uuid FROM messages WHERE session_id=? OR agent_id=?)').run(sessionId, sessionId, sessionId); db.prepare('DELETE FROM messages WHERE session_id=? OR agent_id=?').run(sessionId, sessionId); db.prepare('DELETE FROM subagents WHERE agent_id=? OR session_id=?').run(sessionId, sessionId); + db.prepare('DELETE FROM workflow_agents WHERE session_id=?').run(sessionId); + db.prepare('DELETE FROM workflows WHERE session_id=?').run(sessionId); db.prepare('DELETE FROM summaries WHERE session_id=?').run(sessionId); db.prepare('DELETE FROM sessions WHERE id=?').run(sessionId); } // Consume one unit's record stream into the database and return the new cursor // (also written to index_state). `db` is any SQLite handle sharing prepare/run. -export function persist(db: SqliteDb, unit: IndexUnit, gen: Generator): Cursor { +export function persist(db: SqliteDb, unit: IndexUnit, gen: Generator): Cursor { const st = statements(db); - const write = (r: IndexRecord) => { + const write = (r: TranscriptRecord) => { switch (r.kind) { case 'message': - st.msg.run(r.uuid, r.session_id, r.type, r.parent_uuid, r.timestamp, r.role, r.text, r.content_type, r.is_meta, r.model, r.is_sidechain, r.agent_id, r.input_tokens, r.output_tokens, r.cwd, r.skill, r.source); + st.msg.run(r.uuid, r.session_id, r.type, r.parent_uuid, r.timestamp, r.role, r.text, r.content_type, r.is_meta, r.visibility, r.model, r.is_sidechain, r.agent_id, r.input_tokens, r.output_tokens, r.cwd, r.skill, r.source); break; case 'tool_call': - st.tc.run(r.id, r.message_uuid, r.session_id, r.name, r.input_json, r.file_path); + st.tc.run(r.id, r.message_uuid, r.session_id, r.name, r.presentation, r.input_json, r.file_path); break; case 'tool_result': st.tr.run(r.tool_use_id, r.message_uuid, r.session_id, r.content, r.file_path, r.is_error); @@ -82,6 +104,12 @@ export function persist(db: SqliteDb, unit: IndexUnit, gen: Generator): number | null { const fields = [ @@ -58,9 +63,25 @@ function totalInputTokens(usage: Record): number | null { function discoverAt(rootDir: string, ctx: DiscoverContext): IndexUnit[] { const projectsDir = join(rootDir, 'projects'); + const historyPath = normalize(join(rootDir, 'history.jsonl')); + const historyTitles = new Map(); + if (fs.existsSync(historyPath)) { + readLines(historyPath, (line: string) => { + try { + const item = JSON.parse(line); + if (item?.sessionId && item?.title) historyTitles.set(item.sessionId, item.title); + } catch { /* malformed history entry */ } + }); + } const changedTranscriptPaths = new Set(); + const changedWorkflowPaths = new Set(); const forcedPaths = new Set(); + let historyChanged = false; for (const changedPath of ctx.changedPaths ?? []) { + const rootRelative = isAbsolute(changedPath) + ? normalize(changedPath) + : normalize(join(rootDir, changedPath)); + if (rootRelative === historyPath) historyChanged = true; const absolute = isAbsolute(changedPath) ? normalize(changedPath) : normalize(join(projectsDir, changedPath)); @@ -72,13 +93,16 @@ function discoverAt(rootDir: string, ctx: DiscoverContext): IndexUnit[] { forcedPaths.add(transcript); } else if (absolute.toLowerCase().endsWith('.jsonl')) { changedTranscriptPaths.add(absolute); + } else if (absolute.toLowerCase().endsWith('.json')) { + changedWorkflowPaths.add(absolute); } } - return discoverJsonlFiles(projectsDir).filter((file) => { + const transcriptUnits = discoverJsonlFiles(projectsDir).filter((file) => { const normalizedPath = normalize(file.path); - if (ctx.changedPaths !== undefined && !changedTranscriptPaths.has(normalizedPath)) return false; + if (ctx.changedPaths !== undefined && !historyChanged && !changedTranscriptPaths.has(normalizedPath)) return false; const cursor = ctx.lastCursor(file.path); - return forcedPaths.has(normalizedPath) + return historyChanged + || forcedPaths.has(normalizedPath) || cursor === null || Number(cursor.split(':')[0]) < fs.statSync(file.path).mtimeMs; }).map((f: any) => ({ @@ -87,25 +111,155 @@ function discoverAt(rootDir: string, ctx: DiscoverContext): IndexUnit[] { project: f.project, isSubagent: f.isSubagent, agentId: f.agentId, - meta: f.workflowRunId ? { workflowRunId: f.workflowRunId } : undefined, + meta: { + ...(f.workflowRunId ? { workflowRunId: f.workflowRunId } : {}), + ...(historyTitles.has(f.sessionId) ? { historyTitle: historyTitles.get(f.sessionId) } : {}), + }, })); + + const workflowUnits: IndexUnit[] = []; + if (!fs.existsSync(projectsDir)) return transcriptUnits; + let projects: string[]; + try { projects = fs.readdirSync(projectsDir); } catch { return transcriptUnits; } + for (const project of projects) { + const projectPath = join(projectsDir, project); + if (!isDir(projectPath)) continue; + let sessionIds: string[]; + try { sessionIds = fs.readdirSync(projectPath); } catch { continue; } + for (const sessionId of sessionIds) { + const workflowDir = join(projectPath, sessionId, 'workflows'); + if (!isDir(workflowDir)) continue; + const mainTranscriptPath = join(projectPath, `${sessionId}.jsonl`); + let files: string[]; + try { files = fs.readdirSync(workflowDir); } catch { continue; } + for (const file of files) { + if (!file.endsWith('.json')) continue; + const workflowPath = join(workflowDir, file); + const normalizedPath = normalize(workflowPath); + const relationshipChanged = changedTranscriptPaths.has(normalize(mainTranscriptPath)); + if ( + ctx.changedPaths !== undefined + && !changedWorkflowPaths.has(normalizedPath) + && !relationshipChanged + ) continue; + const mtime = fs.statSync(workflowPath).mtimeMs; + const cursor = ctx.lastCursor(workflowPath); + if (!relationshipChanged && cursor !== null && Number(cursor.split(':')[0]) >= mtime) continue; + workflowUnits.push({ + key: workflowPath, + sessionId, + project, + meta: { kind: 'workflow', mainTranscriptPath } satisfies ClaudeWorkflowUnitMeta, + }); + } + } + } + return [...transcriptUnits, ...workflowUnits]; } export function discover(ctx: DiscoverContext): IndexUnit[] { return discoverAt(join(homedir(), '.claude'), ctx); } -export function* parse(unit: IndexUnit, cursor: Cursor): Generator { +function toolResultText(content: unknown): string { + if (typeof content === 'string') return content; + if (!Array.isArray(content)) return ''; + return content.map((part) => typeof part?.text === 'string' ? part.text : '').join('\n'); +} + +function workflowParentToolUseId( + transcriptPath: string, + runId: string, + workflowName: string | null, +): string | null { + if (!fs.existsSync(transcriptPath)) return null; + const workflowToolIds = new Set(); + let parentToolUseId: string | null = null; + readLines(transcriptPath, (line: string) => { + let record: any; + try { record = JSON.parse(line); } catch { return; } + const content = record?.message?.content; + if (!Array.isArray(content)) return; + if (record.type === 'assistant') { + for (const block of content) { + if (block?.type === 'tool_use' && block?.name === 'Workflow' && typeof block.id === 'string') { + workflowToolIds.add(block.id); + } + } + return; + } + if (record.type !== 'user') return; + for (const block of content) { + if (block?.type !== 'tool_result' || !workflowToolIds.has(block.tool_use_id)) continue; + const text = toolResultText(block.content); + if (!text.includes(runId) && !(workflowName && text.includes(workflowName))) continue; + parentToolUseId = block.tool_use_id; + return false; + } + }); + return parentToolUseId; +} + +function* parseWorkflow(unit: IndexUnit): Generator { + const mtime = fs.statSync(unit.key).mtimeMs; + const outCursor = `${mtime}:1`; + let workflow: any; + try { workflow = JSON.parse(fs.readFileSync(unit.key, 'utf8')); } catch { return outCursor; } + if (!workflow?.runId) return outCursor; + const meta = unit.meta as ClaudeWorkflowUnitMeta; + const progress = Array.isArray(workflow.workflowProgress) ? workflow.workflowProgress : []; + const agents = progress.filter((item: any) => item?.type === 'workflow_agent' && item.agentId); + yield { + kind: 'workflow', + run_id: workflow.runId, + session_id: unit.sessionId, + parent_tool_use_id: workflowParentToolUseId( + meta.mainTranscriptPath, + workflow.runId, + workflow.workflowName || null, + ), + task_id: workflow.taskId || null, + script: workflow.script || null, + result_json: workflow.result ? JSON.stringify(workflow.result) : null, + timestamp: workflow.timestamp || null, + agent_count: agents.length, + duration_ms: workflow.durationMs || null, + total_tokens: workflow.totalTokens || null, + status: workflow.status || null, + workflow_name: workflow.workflowName || null, + }; + for (const item of agents) { + yield { + kind: 'workflow_agent', + agent_id: `agent-${item.agentId}`, + run_id: workflow.runId, + session_id: unit.sessionId, + phase: item.phaseTitle || null, + label: item.label || null, + model: item.model || null, + state: item.state || null, + duration_ms: item.durationMs || null, + tokens: item.tokens || null, + tool_calls: item.toolCalls || null, + }; + } + return outCursor; +} + +export function* parse(unit: IndexUnit, cursor: Cursor): Generator { + if ((unit.meta as ClaudeWorkflowUnitMeta | undefined)?.kind === 'workflow') { + return yield* parseWorkflow(unit); + } const skip = cursorToSkip(cursor); const mtime = fs.statSync(unit.key).mtimeMs; const isSubagent = unit.isSubagent === true; - const records: IndexRecord[] = []; + const records: TranscriptRecord[] = []; const sm = { started_at: null as string | null, ended_at: null as string | null, git_branch: null as string | null, version: null as string | null, - title: null as string | null, + title: ((unit.meta as { historyTitle?: string } | undefined)?.historyTitle ?? null) as string | null, n: 0, }; const subagentStats = { @@ -149,15 +303,17 @@ export function* parse(unit: IndexUnit, cursor: Cursor): Generator [join(configuredRoot, 'projects')], + indexVersionMarker: CLAUDE_CANONICAL_TRANSCRIPT_MARKER, + watchRoots: (configuredRoot) => [ + join(configuredRoot, 'projects'), + join(configuredRoot, 'history.jsonl'), + ], discover: (ctx) => discoverAt(rootDir, ctx), parse, raw: rawClaude, diff --git a/packages/core/src/providers/codex.ts b/packages/core/src/providers/codex.ts index afc4c12..7111be2 100644 --- a/packages/core/src/providers/codex.ts +++ b/packages/core/src/providers/codex.ts @@ -21,13 +21,14 @@ import { codexIsGuardianThread, codexAgentNickname, codexAgentRole, codexUsage, codexEventText, codexMessagePayloadText, codexVisibleMessageKey, codexToolInput, codexToolOutput, + extractMessageIsMeta, isSkillInstructions, readCodexGuardianThreadInfo, } from '../parsing.ts'; import type { Cursor, DiscoverContext, - IndexRecord, + TranscriptRecord, IndexUnit, MessageRecord, ProviderAdapter, @@ -36,11 +37,40 @@ import type { } from './types.ts'; export const name = 'codex'; +const CODEX_CANONICAL_TRANSCRIPT_MARKER = '__codex_canonical_transcript_v2__'; + +const HIDDEN_CONTEXT_ENVELOPE_RE = /^\s*<(environment_context|codex_internal_context)\b[^>]*>[\s\S]*<\/\1>\s*$/; + +function messageVisibility(role: string, text: string | null): 'visible' | 'hidden' { + return role === 'user' && typeof text === 'string' && HIDDEN_CONTEXT_ENVELOPE_RE.test(text) + ? 'hidden' + : 'visible'; +} function discoverAt(rootDir: string, ctx: DiscoverContext): IndexUnit[] { const sessionsDir = join(rootDir, 'sessions'); + const sessionIndexPath = normalize(join(rootDir, 'session_index.jsonl')); + const sessionIndex = new Map(); + if (fs.existsSync(sessionIndexPath)) { + readLines(sessionIndexPath, (line: string) => { + try { + const item = JSON.parse(line); + if (item?.id && item?.thread_name) { + sessionIndex.set(codexRawId(item.id) as string, { + title: item.thread_name, + updatedAt: item.updated_at || null, + }); + } + } catch { /* malformed session-index entry */ } + }); + } const changedFiles = new Set(); + let sessionIndexChanged = false; for (const changedPath of ctx.changedPaths ?? []) { + const rootRelative = isAbsolute(changedPath) + ? normalize(changedPath) + : normalize(join(rootDir, changedPath)); + if (rootRelative === sessionIndexPath) sessionIndexChanged = true; const absolute = isAbsolute(changedPath) ? normalize(changedPath) : normalize(join(sessionsDir, changedPath)); @@ -49,10 +79,10 @@ function discoverAt(rootDir: string, ctx: DiscoverContext): IndexUnit[] { if (absolute.toLowerCase().endsWith('.jsonl')) changedFiles.add(absolute); } return discoverCodexJsonlFiles(sessionsDir).flatMap((file) => { - if (ctx.changedPaths !== undefined && !changedFiles.has(normalize(file.path))) return []; + if (ctx.changedPaths !== undefined && !sessionIndexChanged && !changedFiles.has(normalize(file.path))) return []; const cursor = ctx.lastCursor(file.path); const guardian = readCodexGuardianThreadInfo(file.path); - if (cursor !== null && Number(cursor.split(':')[0]) >= fs.statSync(file.path).mtimeMs && guardian === null) { + if (!sessionIndexChanged && cursor !== null && Number(cursor.split(':')[0]) >= fs.statSync(file.path).mtimeMs && guardian === null) { return []; } let meta: any = null; @@ -67,10 +97,16 @@ function discoverAt(rootDir: string, ctx: DiscoverContext): IndexUnit[] { }); const rawId = meta ? codexRawId(meta.id) : null; const parentId = meta ? codexParentThreadId(meta) : null; + const indexed = rawId ? sessionIndex.get(rawId) : undefined; return [{ key: file.path, sessionId: guardian === null ? codexDbId(parentId || rawId) ?? '' : '', - meta: { source: 'codex', guardian: guardian !== null }, + meta: { + source: 'codex', + guardian: guardian !== null, + indexedTitle: indexed?.title, + indexedUpdatedAt: indexed?.updatedAt, + }, }]; }); } @@ -79,7 +115,7 @@ export function discover(ctx: DiscoverContext): IndexUnit[] { return discoverAt(join(homedir(), '.codex'), ctx); } -export function* parse(unit: IndexUnit, _cursor: Cursor): Generator { +export function* parse(unit: IndexUnit, _cursor: Cursor): Generator { const mtime = fs.statSync(unit.key).mtimeMs; const records: { lineNum: number; obj: any }[] = []; let lineNum = 0; @@ -106,14 +142,19 @@ export function* parse(unit: IndexUnit, _cursor: Cursor): Generator codexLineUuid(threadRawId, n) as string; - const out: IndexRecord[] = []; + const out: TranscriptRecord[] = []; const msgByUuid = new Map(); + const indexedMeta = unit.meta as { indexedTitle?: string; indexedUpdatedAt?: string | null } | undefined; + const initialTimestamp = (meta.timestamp || metaRecord.obj.timestamp || null) as string | null; + const indexedUpdatedAt = indexedMeta?.indexedUpdatedAt ?? null; const sm = { - started_at: (meta.timestamp || metaRecord.obj.timestamp || null) as string | null, - ended_at: (meta.timestamp || metaRecord.obj.timestamp || null) as string | null, + started_at: initialTimestamp, + ended_at: indexedUpdatedAt && (!initialTimestamp || indexedUpdatedAt > initialTimestamp) + ? indexedUpdatedAt + : initialTimestamp, git_branch: (meta.git?.branch || null) as string | null, version: (meta.cli_version || null) as string | null, - title: null as string | null, + title: indexedMeta?.indexedTitle ?? null, n: 0, lastMessageUuid: null as string | null, lastTextAssistantUuid: null as string | null, @@ -135,10 +176,14 @@ export function* parse(unit: IndexUnit, _cursor: Cursor): Generator { + const visibility = messageVisibility(role, text); + const skillInstructions = role === 'user' && isSkillInstructions(text); const rec: MessageRecord = { kind: 'message', uuid, session_id: sessionId, type, parent_uuid: sm.lastMessageUuid, - timestamp: timestamp || null, role, text: trunc(text), content_type: contentType, - is_meta: isMeta, model: currentModel, is_sidechain: isSidechain, agent_id: agentId, + timestamp: timestamp || null, role, text: trunc(text), + content_type: skillInstructions ? 'skill_instructions' : contentType, + is_meta: visibility === 'hidden' || skillInstructions ? 1 : (isMeta || extractMessageIsMeta({}, text)), visibility, + model: currentModel, is_sidechain: isSidechain, agent_id: agentId, input_tokens: null, output_tokens: null, cwd: currentCwd, skill: null, source: 'codex', }; out.push(rec); @@ -191,13 +236,13 @@ export function* parse(unit: IndexUnit, _cursor: Cursor): Generator part?.text).filter((part: unknown) => typeof part === 'string'); - messageText = parts.length > 0 ? parts.join('\n') : null; + messageText = codexMessagePayloadText(payload); } } catch { /* malformed source line */ } } @@ -323,6 +367,7 @@ export function createCodexProvider({ rootDir = join(homedir(), '.codex') }: { r return { name, descriptor: { id: name, name: 'Codex', vendor: 'OpenAI', defaultRoot: rootDir, color: '#10a37f' }, + indexVersionMarker: CODEX_CANONICAL_TRANSCRIPT_MARKER, watchRoots: (configuredRoot) => [join(configuredRoot, 'sessions'), join(configuredRoot, 'session_index.jsonl')], discover: (ctx) => discoverAt(rootDir, ctx), parse, diff --git a/packages/core/src/providers/kimi.ts b/packages/core/src/providers/kimi.ts index 77dc5d7..839bf4e 100644 --- a/packages/core/src/providers/kimi.ts +++ b/packages/core/src/providers/kimi.ts @@ -11,7 +11,7 @@ import { filePath, projectSlugFromPath, trunc, truncJson } from '../parsing.ts'; import type { Cursor, DiscoverContext, - IndexRecord, + TranscriptRecord, IndexUnit, MessageRecord, ProviderAdapter, @@ -50,12 +50,13 @@ interface ProjectedSession { readonly toolResults: ToolResultRecord[]; readonly summaries: SummaryRecord[]; readonly subagents: SubagentRecord[]; - readonly durations: IndexRecord[]; + readonly durations: TranscriptRecord[]; readonly mainMessageCount: number; readonly mainWirePath: string; } const SOURCE = 'kimi'; +export const KIMI_CANONICAL_TRANSCRIPT_MARKER = '__kimi_canonical_transcript_v2__'; function defaultKimiRoot(): string { return process.env['KIMI_CODE_HOME'] ?? join(homedir(), '.kimi-code'); @@ -203,6 +204,46 @@ function isRealUserMessage(message: JsonRecord): boolean { && origin.trigger === 'user-slash'; } +function slashCommandText(command: string, args: unknown): string { + const trimmedArgs = typeof args === 'string' ? args.trim() : ''; + return trimmedArgs.length > 0 ? `${command} ${trimmedArgs}` : command; +} + +function userSlashCommandText(message: JsonRecord): string | null { + const origin = message.origin as JsonRecord | undefined; + if (message.role === 'user' && origin?.trigger === 'user-slash') { + if (origin.kind === 'skill_activation' && typeof origin.skillName === 'string') { + return slashCommandText(`/${origin.skillName}`, origin.skillArgs); + } + if ( + origin.kind === 'plugin_command' + && typeof origin.pluginId === 'string' + && typeof origin.commandName === 'string' + ) { + return slashCommandText(`/${origin.pluginId}:${origin.commandName}`, origin.commandArgs); + } + } + return null; +} + +function projectedMessageText(message: JsonRecord): string | null { + const slashCommand = userSlashCommandText(message); + return slashCommand === null ? messageText(message.content) : trunc(slashCommand); +} + +function isMetaMessage(message: JsonRecord): boolean { + const origin = message.origin as JsonRecord | undefined; + if (origin === undefined || origin.kind === 'user') return false; + return !isRealUserMessage(message); +} + +function canonicalMessageContentType(message: JsonRecord): string { + const origin = message.origin as JsonRecord | undefined; + return origin?.kind === 'skill_activation' && !isRealUserMessage(message) + ? 'skill_instructions' + : messageContentType(message.content); +} + function projectSession(meta: KimiSessionUnitMeta, sessionId: string, state: JsonRecord): ProjectedSession { const cwd = typeof state.cwd === 'string' ? state.cwd @@ -213,7 +254,7 @@ function projectSession(meta: KimiSessionUnitMeta, sessionId: string, state: Jso const toolCalls: ToolCallRecord[] = []; const toolResults: ToolResultRecord[] = []; const summaries: SummaryRecord[] = []; - const durations: IndexRecord[] = []; + const durations: TranscriptRecord[] = []; const childParentCalls = new Map(); let mainMessageCount = 0; @@ -314,9 +355,10 @@ function projectSession(meta: KimiSessionUnitMeta, sessionId: string, state: Jso parent_uuid: previousUuid, timestamp, role: source.role, - text: messageText(source.content), - content_type: messageContentType(source.content), - is_meta: origin !== undefined && origin.kind !== 'user' ? 1 : 0, + text: projectedMessageText(source), + content_type: canonicalMessageContentType(source), + is_meta: isMetaMessage(source) ? 1 : 0, + visibility: 'visible', model, is_sidechain: wire.main ? 0 : 1, agent_id: agentDbId, @@ -344,6 +386,7 @@ function projectSession(meta: KimiSessionUnitMeta, sessionId: string, state: Jso message_uuid: messageUuid, session_id: sessionId, name, + presentation: name === 'Skill' ? 'skill' : 'default', input_json: truncJson(args) ?? '{}', file_path: filePath(name, args as JsonRecord | undefined), }); @@ -405,6 +448,7 @@ function projectSession(meta: KimiSessionUnitMeta, sessionId: string, state: Jso text: partText(part), content_type: typeof part.type === 'string' ? part.type : 'unknown', is_meta: 0, + visibility: 'visible', model, is_sidechain: wire.main ? 0 : 1, agent_id: agentDbId, @@ -421,7 +465,8 @@ function projectSession(meta: KimiSessionUnitMeta, sessionId: string, state: Jso const toolId = namespacedToolId(sessionId, wire.agentId, event.toolCallId); pushMessage({ kind: 'message', uuid, session_id: sessionId, type: 'assistant', parent_uuid: previousUuid, - timestamp, role: 'assistant', text: null, content_type: 'tool_use', is_meta: 0, model, + timestamp, role: 'assistant', text: null, content_type: 'tool_use', is_meta: 0, + visibility: 'visible', model, is_sidechain: wire.main ? 0 : 1, agent_id: agentDbId, input_tokens: null, output_tokens: null, cwd, skill: null, source: SOURCE, }, event.stepUuid); @@ -431,6 +476,7 @@ function projectSession(meta: KimiSessionUnitMeta, sessionId: string, state: Jso message_uuid: uuid, session_id: sessionId, name: String(event.name ?? 'tool'), + presentation: event.name === 'Skill' ? 'skill' : 'default', input_json: truncJson(event.args ?? {}) ?? '{}', file_path: filePath(String(event.name ?? 'tool'), event.args as JsonRecord | undefined), }); @@ -552,13 +598,19 @@ function rawFromWire(path: string, messageUuid: string): RawRecord | null { try { const record = JSON.parse(line) as JsonRecord; if (record.type === 'context.append_message') { - const content = (record.message as JsonRecord | undefined)?.content; - const parts = contentParts(content).map((part) => { - if (part.type === 'text' && typeof part.text === 'string') return part.text; - if (part.type === 'thinking' && typeof part.thinking === 'string') return part.thinking; - return null; - }).filter((part): part is string => part !== null); - projectedText = parts.length > 0 ? parts.join('\n') : null; + const message = record.message as JsonRecord | undefined; + if (message !== undefined) { + const slashCommand = userSlashCommandText(message); + if (slashCommand !== null) projectedText = slashCommand; + else { + const parts = contentParts(message.content).map((part) => { + if (part.type === 'text' && typeof part.text === 'string') return part.text; + if (part.type === 'thinking' && typeof part.thinking === 'string') return part.thinking; + return null; + }).filter((part): part is string => part !== null); + projectedText = parts.length > 0 ? parts.join('\n') : null; + } + } } else if (record.type === 'context.append_loop_event') { const part = (record.event as JsonRecord | undefined)?.part as JsonRecord | undefined; if (part?.type === 'text' && typeof part.text === 'string') projectedText = part.text; @@ -580,6 +632,7 @@ export function createKimiProvider({ rootDir = defaultKimiRoot() }: { rootDir?: return { name, descriptor: { id: name, name: 'Kimi Code', vendor: 'Moonshot AI', defaultRoot: rootDir, color: '#6d6afc' }, + indexVersionMarker: KIMI_CANONICAL_TRANSCRIPT_MARKER, watchRoots: (configuredRoot) => [join(configuredRoot, 'sessions'), join(configuredRoot, 'session_index.jsonl')], discover(ctx: DiscoverContext): IndexUnit[] { const units: IndexUnit[] = []; @@ -604,7 +657,7 @@ export function createKimiProvider({ rootDir = defaultKimiRoot() }: { rootDir?: } return units; }, - *parse(unit: IndexUnit, _cursor: Cursor): Generator { + *parse(unit: IndexUnit, _cursor: Cursor): Generator { const meta = unit.meta as KimiSessionUnitMeta; const before = cursorFor(meta.statePath, meta.wireFiles); const state = readState(meta.statePath); diff --git a/packages/core/src/providers/types.ts b/packages/core/src/providers/types.ts index d6eb539..96f42b0 100644 --- a/packages/core/src/providers/types.ts +++ b/packages/core/src/providers/types.ts @@ -6,12 +6,13 @@ // NOT assumed to be a single JSONL file — an adapter may read a SQLite store, // a directory tree, etc. So discovery, change-detection, and resume cursoring // are all adapter-owned and format-specific. -// - Persist axis: one shared, provider- and binding-agnostic orchestration -// that consumes the records and writes them (index_state, FTS, upsert). +// - Consumer axis: shared, provider-agnostic modules consume the canonical +// transcript for persistence and session-detail presentation. // -// This file defines only the shapes crossing that boundary. Record fields mirror -// the columns in packages/core/src/schema.sql; keep them in sync. Types only — no runtime -// code — so consumers must import with `import type`. +// This file defines only the shapes crossing that seam. Many fields mirror the +// SQLite schema, but the database is a serialization adapter rather than the +// source of transcript semantics. Types only — no runtime code — so consumers +// must import with `import type`. // Opaque per-unit resume/watermark token. The orchestration stores it verbatim // (in index_state) and hands it back on the next run; ONLY the adapter that @@ -46,12 +47,10 @@ export interface DiscoverContext { changedPaths?: string[]; } -/** Discriminated union of everything an adapter's parse can emit. Each record - * kind maps to one schema table (see packages/core/src/schema.sql); `delete-session` is - * the exception — a retraction op, not a table. Sources without a table - * (history.jsonl, codex session_index.jsonl) are not records: adapters fold them - * into the SessionRecord they already emit. */ -export type IndexRecord = +/** Canonical language emitted by every provider adapter. Persist serializes it; + * session-detail assembly consumes it directly. Most record kinds map to one + * schema table; update/retraction records encode canonical state transitions. */ +export type TranscriptRecord = | SessionRecord | MessageRecord | ToolCallRecord @@ -63,6 +62,8 @@ export type IndexRecord = | MessageTurnDurationRecord | DeleteSessionRecord; +export type MessageVisibility = 'visible' | 'hidden'; + export interface MessageRecord { kind: 'message'; uuid: string; @@ -74,6 +75,8 @@ export interface MessageRecord { text: string | null; content_type: string | null; is_meta: 0 | 1; + /** Provider-normalized display eligibility. Assemblers never infer this from text. */ + visibility: MessageVisibility; model: string | null; is_sidechain: 0 | 1; agent_id: string | null; @@ -91,6 +94,7 @@ export interface ToolCallRecord { message_uuid: string; session_id: string; name: string; + presentation: 'default' | 'skill'; input_json: string; file_path: string | null; } @@ -128,17 +132,18 @@ export interface SubagentRecord { total_tokens?: number | null; } -// A workflow run. `agent_count` is intentionally absent: it is a derived -// aggregate (COUNT of workflow_agents for this run) that persist computes, since -// the agents may be indexed on different runs than the workflow metadata. +// A workflow run. `agent_count` is optional presentation metadata; persist still +// computes the authoritative aggregate because agents may arrive on other runs. export interface WorkflowRecord { kind: 'workflow'; run_id: string; session_id: string; + parent_tool_use_id?: string | null; task_id: string | null; script: string | null; result_json: string | null; timestamp: string | null; + agent_count: number; duration_ms: number | null; total_tokens: number | null; status: string | null; @@ -223,7 +228,7 @@ export interface Provider { /** Discover units needing (re)indexing, using stored cursors to detect change. */ discover(ctx: DiscoverContext): IndexUnit[]; /** Yield records for one unit resuming from `cursor`; return the new cursor. */ - parse(unit: IndexUnit, cursor: Cursor): Generator; + parse(unit: IndexUnit, cursor: Cursor): Generator; } /** Serializable source metadata consumed by settings and renderer surfaces. */ diff --git a/packages/core/src/schema-migrations.ts b/packages/core/src/schema-migrations.ts new file mode 100644 index 0000000..44673d8 --- /dev/null +++ b/packages/core/src/schema-migrations.ts @@ -0,0 +1,34 @@ +import type { SqliteDb } from './sqlite-types.ts'; + +const COLUMN_MIGRATIONS = [ + ['sessions', 'source', "TEXT DEFAULT 'claude'"], + ['messages', 'content_type', 'TEXT'], + ['messages', 'is_meta', 'INTEGER DEFAULT 0'], + ['messages', 'visibility', "TEXT DEFAULT 'visible'"], + ['messages', 'source', "TEXT DEFAULT 'claude'"], + ['tool_calls', 'presentation', "TEXT DEFAULT 'default'"], + ['workflows', 'parent_tool_use_id', 'TEXT'], + ['memories', 'anchors', 'TEXT'], + ['memories', 'deleted_at', 'TEXT'], + ['memories', 'deleted_reason', 'TEXT'], +] as const; + +function tableExists(db: SqliteDb, table: string): boolean { + return Boolean(db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name=?").get(table)); +} + +/** Binding-agnostic additive migrations shared by the CLI and desktop app. */ +export function migrateCoreSchemaColumns(db: SqliteDb): void { + const columnsByTable = new Map>(); + for (const [table, column, definition] of COLUMN_MIGRATIONS) { + if (!tableExists(db, table)) continue; + let columns = columnsByTable.get(table); + if (!columns) { + columns = new Set(db.prepare(`PRAGMA table_info(${table})`).all().map((row) => String(row.name))); + columnsByTable.set(table, columns); + } + if (columns.has(column)) continue; + db.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`); + columns.add(column); + } +} diff --git a/packages/core/src/schema.sql b/packages/core/src/schema.sql index f7a9f0a..4f377e4 100644 --- a/packages/core/src/schema.sql +++ b/packages/core/src/schema.sql @@ -6,14 +6,14 @@ CREATE TABLE IF NOT EXISTS sessions ( CREATE TABLE IF NOT EXISTS messages ( uuid TEXT PRIMARY KEY, session_id TEXT, type TEXT, parent_uuid TEXT, timestamp TEXT, role TEXT, text TEXT, content_type TEXT, - is_meta INTEGER DEFAULT 0, model TEXT, + is_meta INTEGER DEFAULT 0, visibility TEXT DEFAULT 'visible', model TEXT, is_sidechain INTEGER DEFAULT 0, agent_id TEXT, input_tokens INTEGER, output_tokens INTEGER, cwd TEXT, skill TEXT, turn_duration_ms INTEGER, source TEXT DEFAULT 'claude'); CREATE TABLE IF NOT EXISTS tool_calls ( id TEXT PRIMARY KEY, message_uuid TEXT, session_id TEXT, - name TEXT, input_json TEXT, file_path TEXT); + name TEXT, presentation TEXT DEFAULT 'default', input_json TEXT, file_path TEXT); CREATE TABLE IF NOT EXISTS tool_results ( tool_use_id TEXT PRIMARY KEY, message_uuid TEXT, session_id TEXT, content TEXT, file_path TEXT, is_error INTEGER DEFAULT 0); @@ -21,7 +21,7 @@ CREATE TABLE IF NOT EXISTS subagents ( agent_id TEXT PRIMARY KEY, session_id TEXT, parent_tool_use_id TEXT, agent_type TEXT, description TEXT, duration_ms INTEGER, total_tokens INTEGER); CREATE TABLE IF NOT EXISTS workflows ( - run_id TEXT PRIMARY KEY, session_id TEXT, task_id TEXT, + run_id TEXT PRIMARY KEY, session_id TEXT, parent_tool_use_id TEXT, task_id TEXT, script TEXT, result_json TEXT, timestamp TEXT, agent_count INTEGER DEFAULT 0, duration_ms INTEGER, total_tokens INTEGER, status TEXT, workflow_name TEXT); CREATE TABLE IF NOT EXISTS workflow_agents ( diff --git a/packages/core/src/session-detail.ts b/packages/core/src/session-detail.ts new file mode 100644 index 0000000..c4b13f4 --- /dev/null +++ b/packages/core/src/session-detail.ts @@ -0,0 +1,597 @@ +import type { + TranscriptRecord, + MessageRecord, + SessionRecord, + SummaryRecord, + ToolCallRecord, + ToolResultRecord, + WorkflowAgentRecord, + WorkflowRecord, +} from './providers/types.ts'; + +type WithoutKind = Omit; + +export interface SessionDetailMessage { + [key: string]: unknown; + uuid: string; + type: string | null; + timestamp: string | null; + text: string | null; + content_type: string | null; + is_meta: 0 | 1; + turn_duration_ms: number | null; +} + +export type SessionDetailToolResult = WithoutKind; + +export interface SessionDetailWorkflowAgent { + agent_id: string; + phase: string | null; + label: string | null; + state: string | null; + tokens: number | null; + duration_ms: number | null; +} + +export interface SessionDetailWorkflow { + [key: string]: unknown; + run_id: string; + parent_tool_use_id: string | null; + workflow_name: string | null; + status: string | null; + duration_ms: number | null; + total_tokens: number | null; + agent_count: number | null; + agents: SessionDetailWorkflowAgent[]; +} + +export interface AssembledToolCall { + id: string; + name: string; + presentation: 'default' | 'skill'; + input_json: string | null; + result: SessionDetailToolResult | null; + subagent?: { + agent_id: string; + agent_type: string | null; + description: string | null; + }; + workflow?: SessionDetailWorkflow; +} + +export interface AssembledMessage extends SessionDetailMessage { + [key: string]: unknown; + tool_calls?: AssembledToolCall[]; + _thinking?: string; + _skillMd?: string; +} + +export interface SessionDetailSnapshot { + session: SessionDetailSession | null; + messages: AssembledMessage[]; + workflows: SessionDetailWorkflow[]; + summaries: SessionDetailSummary[]; +} + +export type SessionDetailSummary = WithoutKind & Record; + +export type SessionDetailSession = Omit, 'countMode'>; + +export interface SessionDetailSessionRow { + [key: string]: unknown; + id: string; + title?: string | null; + project?: string | null; + started_at?: string | null; + ended_at?: string | null; + git_branch?: string | null; + version?: string | null; + message_count?: number | null; + jsonl_path?: string | null; + source?: string | null; +} + +export interface SessionMessageRow { + [key: string]: unknown; + uuid: string; + session_id?: string | null; + type?: string | null; + role?: string | null; + parent_uuid?: string | null; + timestamp?: string | null; + text?: string | null; + content_type?: string | null; + is_meta?: number | boolean | null; + visibility?: string | null; +} + +export interface SessionToolResultRow { + [key: string]: unknown; + tool_use_id: string; + message_uuid?: string | null; + session_id?: string | null; + content?: string | null; +} + +export interface SessionToolCallRow { + [key: string]: unknown; + id: string; + message_uuid: string; + session_id?: string | null; + name: string; + presentation?: string | null; + input_json?: string | null; +} + +export interface SessionSubagentRow { + [key: string]: unknown; + agent_id: string; + session_id?: string | null; + parent_tool_use_id?: string | null; + agent_type?: string | null; + description?: string | null; +} + +export interface SessionWorkflowAgentRow { + [key: string]: unknown; + agent_id: string; + run_id?: string | null; + session_id?: string | null; + phase?: string | null; + label?: string | null; + state?: string | null; + tokens?: number | null; + duration_ms?: number | null; +} + +export interface SessionWorkflowRow { + [key: string]: unknown; + run_id: string; + session_id?: string | null; + parent_tool_use_id?: string | null; + workflow_name?: string | null; + status?: string | null; + duration_ms?: number | null; + total_tokens?: number | null; + agent_count?: number | null; + agents?: SessionWorkflowAgentRow[] | null; +} + +export interface SessionSummaryRow { + [key: string]: unknown; + id: string | number; + session_id?: string | null; +} + +export interface SessionDetailRows { + session?: SessionDetailSessionRow | null; + messages?: SessionMessageRow[]; + toolCalls?: SessionToolCallRow[]; + toolResults?: SessionToolResultRow[]; + subagents?: SessionSubagentRow[]; + workflows?: SessionWorkflowRow[]; + summaries?: SessionSummaryRow[]; +} + +function withoutKind(record: T): WithoutKind { + const { kind: _kind, ...value } = record; + return value; +} + +function assembleMessages( + messages: SessionDetailMessage[], + toolCalls: ToolCallRecord[], + toolResults: ToolResultRecord[], + subagents: Extract[], + workflows: SessionDetailWorkflow[], +): AssembledMessage[] { + const resultsByCallId = new Map(); + for (const result of toolResults) resultsByCallId.set(result.tool_use_id, withoutKind(result)); + + const subagentsByCallId = new Map>(); + for (const subagent of subagents) { + if (subagent.parent_tool_use_id) subagentsByCallId.set(subagent.parent_tool_use_id, subagent); + } + + const callsByMessageUuid = new Map(); + const workflowsByCallId = new Map( + workflows + .filter((workflow) => workflow.parent_tool_use_id) + .map((workflow) => [workflow.parent_tool_use_id as string, workflow]), + ); + for (const toolCall of toolCalls) { + const call: AssembledToolCall = { + id: toolCall.id, + name: toolCall.name, + presentation: toolCall.presentation, + input_json: toolCall.input_json, + result: resultsByCallId.get(toolCall.id) ?? null, + }; + const subagent = subagentsByCallId.get(toolCall.id); + if (subagent) { + call.subagent = { + agent_id: subagent.agent_id, + agent_type: subagent.agent_type ?? null, + description: subagent.description ?? null, + }; + } + const workflow = workflowsByCallId.get(toolCall.id); + if (workflow) call.workflow = workflow; + const calls = callsByMessageUuid.get(toolCall.message_uuid) ?? []; + calls.push(call); + callsByMessageUuid.set(toolCall.message_uuid, calls); + } + + const raw = messages.map((message): AssembledMessage => { + const assembled: AssembledMessage = { ...message }; + const calls = callsByMessageUuid.get(message.uuid); + if (calls?.length) assembled.tool_calls = calls; + return assembled; + }); + + const output: AssembledMessage[] = []; + for (let index = 0; index < raw.length; index++) { + const message = raw[index]; + if (message.content_type === 'tool_result') continue; + + if (message.type === 'assistant' && message.content_type === 'thinking') { + const thinkingParts = [message.text ?? '']; + let nextIndex = index + 1; + while ( + nextIndex < raw.length + && raw[nextIndex].type === 'assistant' + && raw[nextIndex].content_type === 'thinking' + ) { + thinkingParts.push(raw[nextIndex].text ?? ''); + nextIndex++; + } + if ( + nextIndex < raw.length + && raw[nextIndex].type === 'assistant' + && raw[nextIndex].content_type !== 'thinking' + ) { + raw[nextIndex]._thinking = thinkingParts.join('\n\n'); + index = nextIndex - 1; + continue; + } + output.push({ ...message, text: thinkingParts.join('\n\n'), content_type: 'thinking' }); + index = nextIndex - 1; + continue; + } + + if (message.type === 'assistant' && message.content_type === 'tool_use') { + const merged: AssembledMessage = { + ...message, + tool_calls: [...(message.tool_calls ?? [])], + }; + const mergedCalls = merged.tool_calls ?? []; + if (message._thinking) merged._thinking = message._thinking; + const skillOnly = mergedCalls.length === 1 && mergedCalls[0].presentation === 'skill'; + let nextIndex = index + 1; + while (nextIndex < raw.length) { + const next = raw[nextIndex]; + if (next.content_type === 'tool_result') { + nextIndex++; + continue; + } + if (next.content_type === 'skill_instructions' && next.text) { + merged._skillMd = next.text; + nextIndex++; + continue; + } + if (!skillOnly && next.type === 'assistant' && next.content_type === 'tool_use') { + if (next.tool_calls) mergedCalls.push(...next.tool_calls); + if (next.text && !merged.text) merged.text = next.text; + nextIndex++; + continue; + } + break; + } + output.push(merged); + index = nextIndex - 1; + continue; + } + + const assembled: AssembledMessage = { ...message }; + if (message._thinking) assembled._thinking = message._thinking; + if (message.type === 'assistant' + && message.content_type !== 'tool_use' + && message.content_type !== 'thinking') { + if (!assembled.tool_calls) assembled.tool_calls = []; + let nextIndex = index + 1; + while (nextIndex < raw.length) { + const next = raw[nextIndex]; + if (next.content_type === 'tool_result') { + nextIndex++; + continue; + } + if (next.type === 'assistant' && next.content_type === 'tool_use') { + if (next.tool_calls) assembled.tool_calls.push(...next.tool_calls); + nextIndex++; + continue; + } + break; + } + if (!assembled.tool_calls.length) delete assembled.tool_calls; + index = nextIndex - 1; + } + output.push(assembled); + } + + return output; +} + +/** + * Project one canonical provider record stream into the app's session detail. + * Provider-specific wire semantics must already be resolved before this seam. + */ +function assembleTranscriptRecords(records: Iterable): SessionDetailSnapshot { + let session: SessionDetailSession | null = null; + const messages: SessionDetailMessage[] = []; + const messagesByUuid = new Map(); + const mainMessageUuids = new Set(); + const toolCalls: ToolCallRecord[] = []; + const toolResults: ToolResultRecord[] = []; + const subagents: Extract[] = []; + const workflows: WorkflowRecord[] = []; + const workflowAgentsById = new Map(); + const summaries: SessionDetailSummary[] = []; + + for (const record of records) { + switch (record.kind) { + case 'session': + if (record.countMode === 'delta') { + throw new Error('Direct session detail assembly requires a fresh full parse (cursor = null), not a provider delta'); + } + session = { + id: record.id, + title: record.title, + project: record.project, + started_at: record.started_at, + ended_at: record.ended_at, + git_branch: record.git_branch, + version: record.version, + message_count: record.message_count, + jsonl_path: record.jsonl_path, + source: record.source, + }; + break; + case 'message': { + if (record.visibility === 'hidden') break; + const message: SessionDetailMessage = { + uuid: record.uuid, + type: record.type || record.role, + timestamp: record.timestamp, + text: record.text, + content_type: record.content_type, + is_meta: record.is_meta, + turn_duration_ms: typeof (record as MessageRecord & { turn_duration_ms?: unknown }).turn_duration_ms === 'number' + ? (record as MessageRecord & { turn_duration_ms: number }).turn_duration_ms + : null, + }; + messages.push(message); + messagesByUuid.set(message.uuid, message); + if (record.agent_id === null) mainMessageUuids.add(message.uuid); + break; + } + case 'tool_call': + toolCalls.push(record); + break; + case 'tool_result': + toolResults.push(record); + break; + case 'subagent': + subagents.push(record); + break; + case 'workflow': + workflows.push(record); + break; + case 'workflow_agent': + workflowAgentsById.set(record.agent_id, { + ...(workflowAgentsById.get(record.agent_id) ?? record), + ...Object.fromEntries( + Object.entries(record).filter(([, value]) => value !== null && value !== undefined), + ), + } as WorkflowAgentRecord); + break; + case 'summary': + summaries.push(withoutKind(record)); + break; + case 'message-turn-duration': { + const message = messagesByUuid.get(record.uuid); + if (message) message.turn_duration_ms = record.turn_duration_ms; + break; + } + case 'delete-session': + break; + } + } + + const assembledWorkflows: SessionDetailWorkflow[] = workflows.map((workflow) => { + const agents = [...workflowAgentsById.values()] + .filter((agent) => agent.run_id === workflow.run_id) + .map((agent) => ({ + agent_id: agent.agent_id, + phase: agent.phase ?? null, + label: agent.label ?? null, + state: agent.state ?? null, + tokens: agent.tokens ?? null, + duration_ms: agent.duration_ms ?? null, + })); + return { + run_id: workflow.run_id, + parent_tool_use_id: workflow.parent_tool_use_id ?? null, + workflow_name: workflow.workflow_name, + status: workflow.status, + duration_ms: workflow.duration_ms, + total_tokens: workflow.total_tokens, + agent_count: workflow.agent_count ?? agents.length, + agents, + }; + }); + + const detailMessages = session === null + ? messages + : messages.filter((message) => mainMessageUuids.has(message.uuid)); + detailMessages.sort((left, right) => { + const leftTimestamp = left.timestamp ?? ''; + const rightTimestamp = right.timestamp ?? ''; + if (leftTimestamp !== rightTimestamp) return leftTimestamp < rightTimestamp ? -1 : 1; + return left.uuid < right.uuid ? -1 : left.uuid > right.uuid ? 1 : 0; + }); + + return { + session, + messages: assembleMessages(detailMessages, toolCalls, toolResults, subagents, assembledWorkflows), + workflows: assembledWorkflows, + summaries, + }; +} + +/** Adapt persisted rows back into the same canonical record language providers emit. */ +function sessionDetailRecordsFromRows(input: SessionDetailRows): TranscriptRecord[] { + const records: TranscriptRecord[] = []; + if (input.session) { + const session = input.session; + records.push({ + kind: 'session', + id: session.id, + title: typeof session.title === 'string' ? session.title : null, + project: typeof session.project === 'string' ? session.project : null, + started_at: typeof session.started_at === 'string' ? session.started_at : null, + ended_at: typeof session.ended_at === 'string' ? session.ended_at : null, + git_branch: typeof session.git_branch === 'string' ? session.git_branch : null, + version: typeof session.version === 'string' ? session.version : null, + message_count: typeof session.message_count === 'number' ? session.message_count : 0, + countMode: 'total', + jsonl_path: typeof session.jsonl_path === 'string' ? session.jsonl_path : '', + source: typeof session.source === 'string' ? session.source : '', + }); + } + for (const message of input.messages ?? []) { + records.push({ + ...message, + kind: 'message', + uuid: message.uuid, + session_id: typeof message.session_id === 'string' ? message.session_id : '', + type: typeof message.type === 'string' ? message.type : typeof message.role === 'string' ? message.role : '', + parent_uuid: typeof message.parent_uuid === 'string' ? message.parent_uuid : null, + timestamp: typeof message.timestamp === 'string' ? message.timestamp : null, + role: typeof message.role === 'string' ? message.role : null, + text: typeof message.text === 'string' ? message.text : null, + content_type: typeof message.content_type === 'string' ? message.content_type : null, + is_meta: message.is_meta ? 1 : 0, + visibility: message.visibility === 'hidden' ? 'hidden' : 'visible', + model: typeof message.model === 'string' ? message.model : null, + is_sidechain: message.is_sidechain ? 1 : 0, + agent_id: typeof message.agent_id === 'string' ? message.agent_id : null, + input_tokens: typeof message.input_tokens === 'number' ? message.input_tokens : null, + output_tokens: typeof message.output_tokens === 'number' ? message.output_tokens : null, + cwd: typeof message.cwd === 'string' ? message.cwd : null, + skill: typeof message.skill === 'string' ? message.skill : null, + source: typeof message.source === 'string' ? message.source : '', + }); + } + for (const toolCall of input.toolCalls ?? []) { + records.push({ + ...toolCall, + kind: 'tool_call', + id: toolCall.id, + message_uuid: toolCall.message_uuid, + session_id: typeof toolCall.session_id === 'string' ? toolCall.session_id : '', + name: toolCall.name, + presentation: toolCall.presentation === 'skill' ? 'skill' : 'default', + input_json: typeof toolCall.input_json === 'string' ? toolCall.input_json : '', + file_path: typeof toolCall.file_path === 'string' ? toolCall.file_path : null, + }); + } + for (const result of input.toolResults ?? []) { + records.push({ + ...result, + kind: 'tool_result', + tool_use_id: result.tool_use_id, + message_uuid: typeof result.message_uuid === 'string' ? result.message_uuid : '', + session_id: typeof result.session_id === 'string' ? result.session_id : '', + content: typeof result.content === 'string' ? result.content : '', + file_path: typeof result.file_path === 'string' ? result.file_path : null, + is_error: result.is_error ? 1 : 0, + }); + } + for (const subagent of input.subagents ?? []) { + records.push({ + ...subagent, + kind: 'subagent', + agent_id: subagent.agent_id, + session_id: typeof subagent.session_id === 'string' ? subagent.session_id : '', + parent_tool_use_id: typeof subagent.parent_tool_use_id === 'string' ? subagent.parent_tool_use_id : null, + agent_type: typeof subagent.agent_type === 'string' ? subagent.agent_type : null, + description: typeof subagent.description === 'string' ? subagent.description : null, + duration_ms: typeof subagent.duration_ms === 'number' ? subagent.duration_ms : null, + total_tokens: typeof subagent.total_tokens === 'number' ? subagent.total_tokens : null, + }); + } + for (const workflow of input.workflows ?? []) { + records.push({ + ...workflow, + kind: 'workflow', + run_id: workflow.run_id, + session_id: typeof workflow.session_id === 'string' ? workflow.session_id : '', + parent_tool_use_id: typeof workflow.parent_tool_use_id === 'string' ? workflow.parent_tool_use_id : null, + task_id: typeof workflow.task_id === 'string' ? workflow.task_id : null, + script: typeof workflow.script === 'string' ? workflow.script : null, + result_json: typeof workflow.result_json === 'string' ? workflow.result_json : null, + timestamp: typeof workflow.timestamp === 'string' ? workflow.timestamp : null, + agent_count: typeof workflow.agent_count === 'number' ? workflow.agent_count : 0, + duration_ms: typeof workflow.duration_ms === 'number' ? workflow.duration_ms : null, + total_tokens: typeof workflow.total_tokens === 'number' ? workflow.total_tokens : null, + status: typeof workflow.status === 'string' ? workflow.status : null, + workflow_name: typeof workflow.workflow_name === 'string' ? workflow.workflow_name : null, + }); + for (const agent of workflow.agents ?? []) { + records.push({ + ...agent, + kind: 'workflow_agent', + agent_id: agent.agent_id, + run_id: typeof agent.run_id === 'string' ? agent.run_id : workflow.run_id, + session_id: typeof agent.session_id === 'string' + ? agent.session_id + : typeof workflow.session_id === 'string' ? workflow.session_id : '', + ...(typeof agent.agent_type === 'string' ? { agent_type: agent.agent_type } : {}), + ...(typeof agent.description === 'string' ? { description: agent.description } : {}), + ...(typeof agent.phase === 'string' ? { phase: agent.phase } : {}), + ...(typeof agent.label === 'string' ? { label: agent.label } : {}), + ...(typeof agent.model === 'string' ? { model: agent.model } : {}), + ...(typeof agent.state === 'string' ? { state: agent.state } : {}), + ...(typeof agent.duration_ms === 'number' ? { duration_ms: agent.duration_ms } : {}), + ...(typeof agent.tokens === 'number' ? { tokens: agent.tokens } : {}), + ...(typeof agent.tool_calls === 'number' ? { tool_calls: agent.tool_calls } : {}), + }); + } + } + for (const summary of input.summaries ?? []) { + records.push({ + ...summary, + kind: 'summary', + id: String(summary.id), + session_id: typeof summary.session_id === 'string' ? summary.session_id : '', + timestamp: typeof summary.timestamp === 'string' ? summary.timestamp : null, + source: typeof summary.source === 'string' ? summary.source : '', + content: typeof summary.content === 'string' ? summary.content : '', + }); + } + return records; +} + +/** + * Assemble detail from either a provider's complete canonical transcript stream + * (a fresh parse with cursor = null) or the same records after a persistence + * round-trip. This is the only presentation seam. + */ +export function assembleSessionDetail( + input: Iterable | SessionDetailRows, +): SessionDetailSnapshot { + const records = Symbol.iterator in input + ? input as Iterable + : sessionDetailRecordsFromRows(input as SessionDetailRows); + return assembleTranscriptRecords(records); +} diff --git a/tests/app-indexer.test.mjs b/tests/app-indexer.test.mjs index 6c34896..c1e5ab9 100644 --- a/tests/app-indexer.test.mjs +++ b/tests/app-indexer.test.mjs @@ -7,7 +7,7 @@ import { join } from 'node:path'; const require = createRequire(import.meta.url); import { buildIndex } from '../app/src/main/indexer.ts'; -import { CLAUDE_INPUT_TOKEN_SEMANTICS_MARKER } from '../packages/core/src/providers/claude.ts'; +import { CLAUDE_CANONICAL_TRANSCRIPT_MARKER } from '../packages/core/src/providers/claude.ts'; const { DatabaseSync } = require('node:sqlite'); class TestDatabase { @@ -123,7 +123,7 @@ test('app indexer refreshes unchanged Claude usage when input token semantics ch const stale = new TestDatabase(dbPath); stale.prepare('UPDATE messages SET input_tokens = 10 WHERE uuid = ?').run('msg-token-semantics-1'); stale.prepare('DELETE FROM index_state WHERE jsonl_path = ?') - .run(CLAUDE_INPUT_TOKEN_SEMANTICS_MARKER); + .run(CLAUDE_CANONICAL_TRANSCRIPT_MARKER); stale.close(); buildIndex({ @@ -140,7 +140,7 @@ test('app indexer refreshes unchanged Claude usage when input token semantics ch ); assert.ok( refreshed.prepare('SELECT jsonl_path FROM index_state WHERE jsonl_path = ?') - .get(CLAUDE_INPUT_TOKEN_SEMANTICS_MARKER), + .get(CLAUDE_CANONICAL_TRANSCRIPT_MARKER), ); refreshed.close(); }); @@ -437,11 +437,12 @@ test('app indexer loads Codex root sessions into the shared schema', () => { ]); assert.equal(messages[1].turn_duration_ms, 4321); - const tool = db.prepare('SELECT * FROM tool_calls WHERE id=?').get('codex:call_codex_1'); + const toolId = `codex:${codexId}:call_codex_1`; + const tool = db.prepare('SELECT * FROM tool_calls WHERE id=?').get(toolId); assert.equal(tool.session_id, `codex:${codexId}`); assert.equal(tool.name, 'exec_command'); assert.equal(tool.message_uuid, `codex:${codexId}:000004`); - const toolResult = db.prepare('SELECT message_uuid, content FROM tool_results WHERE tool_use_id=?').get('codex:call_codex_1'); + const toolResult = db.prepare('SELECT message_uuid, content FROM tool_results WHERE tool_use_id=?').get(toolId); assert.equal(toolResult.message_uuid, `codex:${codexId}:000004`); assert.equal(toolResult.content, '/tmp/obelisk-app'); db.close(); @@ -771,7 +772,8 @@ test('app indexer maps Codex subagent threads onto parent sessions', () => { assert.equal(db.prepare('SELECT COUNT(*) AS c FROM sessions WHERE id=?').get(`codex:${childId}`).c, 0); const subagent = db.prepare('SELECT * FROM subagents WHERE agent_id=?').get(`codex:${childId}`); assert.equal(subagent.session_id, `codex:${parentId}`); - assert.equal(subagent.parent_tool_use_id, 'codex:call_spawn_1'); + const spawnToolId = `codex:${parentId}:call_spawn_1`; + assert.equal(subagent.parent_tool_use_id, spawnToolId); assert.equal(subagent.agent_type, 'worker'); assert.equal(subagent.description, 'Plato'); @@ -779,7 +781,7 @@ test('app indexer maps Codex subagent threads onto parent sessions', () => { assert.equal(spawnMessage.session_id, `codex:${parentId}`); assert.equal(spawnMessage.content_type, 'tool_use'); assert.equal(spawnMessage.source, 'codex'); - assert.equal(db.prepare('SELECT name, message_uuid FROM tool_calls WHERE id=?').get('codex:call_spawn_1').message_uuid, spawnMessage.uuid); + assert.equal(db.prepare('SELECT name, message_uuid FROM tool_calls WHERE id=?').get(spawnToolId).message_uuid, spawnMessage.uuid); const childMessages = db.prepare('SELECT session_id, agent_id, is_sidechain, source, text FROM messages WHERE agent_id=? ORDER BY timestamp, uuid').all(`codex:${childId}`); assert.deepEqual(childMessages.map(m => [m.session_id, m.agent_id, m.is_sidechain, m.source, m.text]), [ diff --git a/tests/app-kimi-index.test.mjs b/tests/app-kimi-index.test.mjs index 247a2a6..e331dbe 100644 --- a/tests/app-kimi-index.test.mjs +++ b/tests/app-kimi-index.test.mjs @@ -6,6 +6,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { buildIndex } from '../app/src/main/indexer.ts'; +import { createKimiProvider } from '../packages/core/src/providers/kimi.ts'; const require = createRequire(import.meta.url); const { DatabaseSync } = require('node:sqlite'); @@ -21,7 +22,7 @@ class TestDatabase { close() { return this.db.close(); } } -function writeSession(kimiDir) { +function writeSession(kimiDir, { userSlash = false } = {}) { const sessionDir = join(kimiDir, 'sessions', 'workspace-1', 'session-index-1'); const mainDir = join(sessionDir, 'agents', 'main'); mkdirSync(mainDir, { recursive: true }); @@ -35,7 +36,15 @@ function writeSession(kimiDir) { const wirePath = join(mainDir, 'wire.jsonl'); const records = [ { type: 'metadata', protocol_version: '1.5', created_at: 1753005600000 }, - { type: 'context.append_message', time: 1753005601000, message: { role: 'user', content: [{ type: 'text', text: 'kimi index needle' }], toolCalls: [], origin: { kind: 'user' } } }, + { type: 'context.append_message', time: 1753005601000, message: userSlash + ? { + role: 'user', content: 'Expanded skill instructions.', toolCalls: [], + origin: { + kind: 'skill_activation', trigger: 'user-slash', skillName: 'obelisk', + skillArgs: 'find prior decisions', + }, + } + : { role: 'user', content: [{ type: 'text', text: 'kimi index needle' }], toolCalls: [], origin: { kind: 'user' } } }, ]; writeFileSync(wirePath, records.map((record) => JSON.stringify(record)).join('\n') + '\n'); return { sessionDir, wirePath, records }; @@ -123,3 +132,37 @@ test('Kimi undo and clear replace the indexed session instead of leaving stale r assert.equal(db.prepare('SELECT message_count FROM sessions WHERE id=?').get('kimi:session-index-1').message_count, 0); db.close(); }); + +test('Kimi prompt semantics marker replays unchanged sessions once', () => { + const home = mkdtempSync(join(tmpdir(), 'obelisk-kimi-prompt-marker-')); + const claudeDir = join(home, '.claude'); + const codexDir = join(home, '.codex'); + const kimiDir = join(home, '.kimi-code'); + const dbPath = join(home, '.obelisk', 'obelisk.sqlite'); + writeSession(kimiDir, { userSlash: true }); + const options = { + claudeDir, + codexDir, + providerRoots: { kimi: kimiDir }, + dbPath, + DatabaseImpl: TestDatabase, + }; + + buildIndex(options); + let db = new TestDatabase(dbPath); + const marker = createKimiProvider({ rootDir: kimiDir }).indexVersionMarker; + assert.equal(typeof marker, 'string'); + db.prepare("UPDATE messages SET text='stale expanded instructions', is_meta=1 WHERE source='kimi'").run(); + db.prepare('DELETE FROM index_state WHERE jsonl_path=?').run(marker); + db.close(); + + const replay = buildIndex(options); + assert.deepEqual(replay.affectedSessionIds, ['kimi:session-index-1']); + db = new TestDatabase(dbPath); + assert.deepEqual( + { ...db.prepare("SELECT text,is_meta FROM messages WHERE source='kimi'").get() }, + { text: '/obelisk find prior decisions', is_meta: 0 }, + ); + assert.equal(db.prepare('SELECT COUNT(*) AS c FROM index_state WHERE jsonl_path=?').get(marker).c, 1); + db.close(); +}); diff --git a/tests/app-main-settings.test.mjs b/tests/app-main-settings.test.mjs index cdc9338..edd7732 100644 --- a/tests/app-main-settings.test.mjs +++ b/tests/app-main-settings.test.mjs @@ -258,6 +258,7 @@ test('main process watches every root declared by the built-in provider registry assert.equal(serviceOptions.length, 1); assert.deepEqual(serviceOptions[0].watchDirs, [ join(claudeDir, 'projects'), + join(claudeDir, 'history.jsonl'), join(codexDir, 'sessions'), join(codexDir, 'session_index.jsonl'), join(home, '.kimi-code', 'sessions'), diff --git a/tests/app-provider-indexer.test.mjs b/tests/app-provider-indexer.test.mjs index b9ea16c..dcce5a8 100644 --- a/tests/app-provider-indexer.test.mjs +++ b/tests/app-provider-indexer.test.mjs @@ -41,7 +41,7 @@ test('app indexer persists every provider through one registry-driven loop', () yield { kind: 'message', uuid: 'alpha:message', session_id: unit.sessionId, type: 'user', parent_uuid: null, timestamp: '2026-07-20T10:00:00.000Z', role: 'user', - text: 'registry tracer bullet', content_type: 'text', is_meta: 0, model: null, + text: 'registry tracer bullet', content_type: 'text', is_meta: 0, visibility: 'visible', model: null, is_sidechain: 0, agent_id: null, input_tokens: null, output_tokens: null, cwd: '/tmp/alpha', skill: null, source: 'alpha', }; diff --git a/tests/app-rollback-guard.test.mjs b/tests/app-rollback-guard.test.mjs index b276eae..fcd3d71 100644 --- a/tests/app-rollback-guard.test.mjs +++ b/tests/app-rollback-guard.test.mjs @@ -223,13 +223,7 @@ test('BEGIN contention during finalize defers the build', () => { test('a finalize database error is propagated instead of swallowed as malformed input', () => { const { home, dbPath, projectsDir } = twoFileHome('hello alpha', 'hello beta'); - const workflowDir = join(projectsDir, '-tmp-proj', 'alpha', 'workflows'); - mkdirSync(workflowDir, { recursive: true }); - writeFileSync(join(workflowDir, 'run.json'), JSON.stringify({ - runId: 'workflow-1', - workflowName: 'POISON WORKFLOW', - })); - const Db = makeDbClass(args => args.some(arg => typeof arg === 'string' && arg.includes('POISON WORKFLOW'))); + const Db = makeDbClass(args => args.some(arg => arg === '__last_build__')); assert.throws(() => buildIndex({ force: false, diff --git a/tests/claude-parse.test.mjs b/tests/claude-parse.test.mjs index ee53ba9..061c39e 100644 --- a/tests/claude-parse.test.mjs +++ b/tests/claude-parse.test.mjs @@ -4,11 +4,16 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; -import { mkdtempSync, writeFileSync, statSync } from 'node:fs'; +import { mkdirSync, mkdtempSync, readFileSync, writeFileSync, statSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; -import { parse } from '../packages/core/src/providers/claude.ts'; +import { createClaudeProvider, parse } from '../packages/core/src/providers/claude.ts'; +import { assembleSessionDetail } from '../packages/core/src/session-detail.ts'; +import { persist } from '../packages/core/src/persist.ts'; + +const SCHEMA = readFileSync(new URL('../packages/core/src/schema.sql', import.meta.url), 'utf8'); function writeFixture() { const dir = mkdtempSync(join(tmpdir(), 'obelisk-claude-parse-')); @@ -69,6 +74,10 @@ test('claude parse() yields the expected record stream for a main session', () = assert.equal(sessions[0].ended_at, '2026-06-10T10:00:10Z'); assert.equal(sessions[0].git_branch, 'main'); + const detail = assembleSessionDetail(values); + assert.deepEqual(detail.messages.map((message) => message.text), ['hi', 'ok']); + assert.equal(detail.messages[1].tool_calls[0].result.content, 'file body'); + // Cursor encodes mtime:lines (6 lines consumed). assert.equal(ret, `${statSync(path).mtimeMs}:6`); }); @@ -90,3 +99,66 @@ test('claude parse() resumes from a cursor, skipping already-indexed lines', () assert.deepEqual(values.filter(r => r.kind !== 'session'), []); assert.equal(values.find(r => r.kind === 'session').message_count, 0); }); + +test('claude provider emits workflow artifacts with an explicit canonical tool edge', () => { + const root = mkdtempSync(join(tmpdir(), 'obelisk-claude-workflow-')); + const projectDir = join(root, 'projects', '-proj'); + const workflowDir = join(projectDir, 'sid-workflow', 'workflows'); + const workflowAgentDir = join(projectDir, 'sid-workflow', 'subagents', 'workflows', 'run-workflow'); + mkdirSync(workflowDir, { recursive: true }); + mkdirSync(workflowAgentDir, { recursive: true }); + writeFileSync(join(projectDir, 'sid-workflow.jsonl'), [ + { + uuid: 'assistant-workflow', type: 'assistant', timestamp: '2026-06-10T10:00:00Z', + message: { role: 'assistant', content: [{ type: 'tool_use', id: 'workflow-tool', name: 'Workflow', input: {} }] }, + }, + { + uuid: 'workflow-result', type: 'user', timestamp: '2026-06-10T10:00:01Z', + message: { role: 'user', content: [{ type: 'tool_result', tool_use_id: 'workflow-tool', content: 'run-workflow complete' }] }, + }, + ].map(line => JSON.stringify(line)).join('\n') + '\n'); + writeFileSync(join(workflowDir, 'run-workflow.json'), JSON.stringify({ + runId: 'run-workflow', + workflowName: 'Review', + status: 'complete', + workflowProgress: [{ type: 'workflow_agent', agentId: '7', phaseTitle: 'review', label: 'Reviewer' }], + })); + writeFileSync(join(workflowAgentDir, 'agent-7.jsonl'), `${JSON.stringify({ + uuid: 'workflow-agent-message', type: 'user', timestamp: '2026-06-10T10:00:00Z', + message: { role: 'user', content: 'review it' }, + })}\n`); + writeFileSync(join(workflowAgentDir, 'agent-7.meta.json'), JSON.stringify({ + agentType: 'reviewer', description: 'Review the implementation', + })); + writeFileSync(join(root, 'history.jsonl'), `${JSON.stringify({ + sessionId: 'sid-workflow', title: 'History-owned title', + })}\n`); + const provider = createClaudeProvider({ rootDir: root }); + const units = provider.discover({ lastCursor: () => null }); + const records = units.flatMap(unit => drain(provider.parse(unit, null)).values); + + const workflow = records.find(record => record.kind === 'workflow'); + assert.equal(workflow.parent_tool_use_id, 'workflow-tool'); + const detail = assembleSessionDetail(records); + assert.equal(detail.session.title, 'History-owned title'); + assert.equal(detail.messages[0].tool_calls[0].workflow.run_id, 'run-workflow'); + assert.equal(detail.workflows[0].agents[0].label, 'Reviewer'); + assert.equal(detail.workflows[0].agents.length, 1); + + const db = new DatabaseSync(':memory:'); + db.exec(SCHEMA); + for (const unit of units) persist(db, unit, provider.parse(unit, null)); + const workflows = db.prepare('SELECT * FROM workflows').all(); + for (const row of workflows) row.agents = db.prepare('SELECT * FROM workflow_agents WHERE run_id=?').all(row.run_id); + const persistedDetail = assembleSessionDetail({ + session: db.prepare('SELECT * FROM sessions').get(), + messages: db.prepare('SELECT * FROM messages ORDER BY timestamp, uuid').all(), + toolCalls: db.prepare('SELECT * FROM tool_calls').all(), + toolResults: db.prepare('SELECT * FROM tool_results').all(), + subagents: db.prepare('SELECT * FROM subagents').all(), + workflows, + summaries: db.prepare('SELECT * FROM summaries').all(), + }); + assert.deepEqual(persistedDetail, detail); + db.close(); +}); diff --git a/tests/codex-parse.test.mjs b/tests/codex-parse.test.mjs index e7208eb..a7b9526 100644 --- a/tests/codex-parse.test.mjs +++ b/tests/codex-parse.test.mjs @@ -5,11 +5,11 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; -import { mkdtempSync, writeFileSync } from 'node:fs'; +import { mkdirSync, mkdtempSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { parse } from '../packages/core/src/providers/codex.ts'; +import { createCodexProvider, parse } from '../packages/core/src/providers/codex.ts'; function writeFixture(lines) { const dir = mkdtempSync(join(tmpdir(), 'obelisk-codex-parse-')); @@ -56,9 +56,9 @@ test('codex parse() yields a deduped, tool-aware record stream with a total sess assert.equal(textAssistant.output_tokens, 50); // Tool call + result. - assert.deepEqual(byKind('tool_call').map(t => ({ id: t.id, name: t.name })), [{ id: 'codex:call_1', name: 'shell' }]); + assert.deepEqual(byKind('tool_call').map(t => ({ id: t.id, name: t.name })), [{ id: `codex:${META.id}:call_1`, name: 'shell' }]); assert.equal(byKind('tool_result').length, 1); - assert.equal(byKind('tool_result')[0].tool_use_id, 'codex:call_1'); + assert.equal(byKind('tool_result')[0].tool_use_id, `codex:${META.id}:call_1`); // task_complete → turn duration on the text-assistant message. assert.deepEqual(byKind('message-turn-duration').map(d => d.turn_duration_ms), [1500]); @@ -84,3 +84,25 @@ test('codex parse() retracts a guardian thread via delete-session and emits noth assert.equal(values[0].kind, 'delete-session'); assert.match(values[0].sessionId, /^codex:/); }); + +test('codex provider folds session_index metadata into its canonical session record', () => { + const root = mkdtempSync(join(tmpdir(), 'obelisk-codex-index-meta-')); + const sessionsDir = join(root, 'sessions', '2026', '06', '10'); + mkdirSync(sessionsDir, { recursive: true }); + const path = join(sessionsDir, `rollout-${META.id}.jsonl`); + writeFileSync(path, `${JSON.stringify({ + type: 'session_meta', timestamp: '2026-06-10T10:00:00Z', payload: META, + })}\n`); + const indexPath = join(root, 'session_index.jsonl'); + writeFileSync(indexPath, `${JSON.stringify({ + id: META.id, thread_name: 'Indexed title', updated_at: '2026-06-10T11:00:00Z', + })}\n`); + const provider = createCodexProvider({ rootDir: root }); + const units = provider.discover({ lastCursor: () => '9999999999999:1', changedPaths: [indexPath] }); + + assert.equal(units.length, 1); + const { values } = drain(provider.parse(units[0], null)); + const session = values.find(record => record.kind === 'session'); + assert.equal(session.title, 'Indexed title'); + assert.equal(session.ended_at, '2026-06-10T11:00:00Z'); +}); diff --git a/tests/codex-replay-tool-identity.test.mjs b/tests/codex-replay-tool-identity.test.mjs new file mode 100644 index 0000000..2db41b8 --- /dev/null +++ b/tests/codex-replay-tool-identity.test.mjs @@ -0,0 +1,69 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { DatabaseSync } from 'node:sqlite'; +import { mkdtempSync, readFileSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { assembleSessionDetail } from '../app/src/shared/session-detail-assembly.mjs'; +import { persist } from '../packages/core/src/persist.ts'; +import { parse } from '../packages/core/src/providers/codex.ts'; + +const SCHEMA = readFileSync(new URL('../packages/core/src/schema.sql', import.meta.url), 'utf8'); +const PARENT_ID = '019ed000-0000-7000-8000-000000000101'; +const REPLAY_ID = '019ed000-0000-7000-8000-000000000102'; + +function writeRollout(path, meta, source) { + writeFileSync(path, [ + { timestamp: '2026-06-15T10:00:00Z', type: 'session_meta', payload: meta }, + { + timestamp: '2026-06-15T10:00:01Z', + type: 'response_item', + payload: { type: 'custom_tool_call', call_id: 'call_shared', name: 'exec', input: source }, + }, + { + timestamp: '2026-06-15T10:00:02Z', + type: 'response_item', + payload: { type: 'custom_tool_call_output', call_id: 'call_shared', output: 'done' }, + }, + ].map(line => JSON.stringify(line)).join('\n') + '\n'); +} + +test('a replayed Codex call cannot steal the visible message tool association', () => { + const dir = mkdtempSync(join(tmpdir(), 'obelisk-codex-replay-')); + const parentPath = join(dir, 'parent.jsonl'); + const replayPath = join(dir, 'replay.jsonl'); + writeRollout(parentPath, { + id: PARENT_ID, + timestamp: '2026-06-15T10:00:00Z', + cwd: '/proj', + }, 'text("parent")'); + writeRollout(replayPath, { + id: REPLAY_ID, + forked_from_id: PARENT_ID, + timestamp: '2026-06-15T10:00:00Z', + cwd: '/proj', + }, 'text("replay")'); + + const db = new DatabaseSync(':memory:'); + db.exec(SCHEMA); + for (const path of [parentPath, replayPath]) { + const unit = { key: path, sessionId: '', meta: { source: 'codex' } }; + persist(db, unit, parse(unit, null)); + } + + const sessionId = `codex:${PARENT_ID}`; + const messages = db.prepare( + 'SELECT * FROM messages WHERE session_id=? AND agent_id IS NULL ORDER BY timestamp, uuid', + ).all(sessionId); + const toolCalls = db.prepare('SELECT * FROM tool_calls WHERE session_id=?').all(sessionId); + const toolResults = db.prepare('SELECT * FROM tool_results WHERE session_id=?').all(sessionId); + const assembled = assembleSessionDetail({ messages, toolCalls, toolResults, subagents: [], workflows: [] }).messages; + + assert.equal(messages.length, 1, 'the replay remains outside the visible session timeline'); + assert.equal(toolCalls.length, 2, 'each rollout owns an independently addressable tool call'); + assert.equal(toolResults.length, 2, 'each rollout owns an independently addressable tool result'); + assert.deepEqual(assembled[0].tool_calls?.map(call => call.input_json), ['"text(\\"parent\\")"']); + assert.equal(assembled[0].tool_calls?.[0].result?.content, 'done'); + db.close(); +}); diff --git a/tests/kimi-parse.test.mjs b/tests/kimi-parse.test.mjs index c589bdf..3082f7c 100644 --- a/tests/kimi-parse.test.mjs +++ b/tests/kimi-parse.test.mjs @@ -6,6 +6,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { createKimiProvider } from '../packages/core/src/providers/kimi.ts'; +import { assembleSessionDetail } from '../packages/core/src/session-detail.ts'; function drain(gen) { const values = []; @@ -71,7 +72,7 @@ test('kimi provider discovers a changed session directory and returns a stable c assert.deepEqual(unchanged, []); }); -test('kimi provider folds main and subagent wire logs into the existing record language', () => { +test('kimi provider folds main and subagent wire logs into the canonical transcript language', () => { const { root } = writeKimiFixture(); const provider = createKimiProvider({ rootDir: root }); const unit = provider.discover({ lastCursor: () => null })[0]; @@ -83,7 +84,7 @@ test('kimi provider folds main and subagent wire logs into the existing record l : record); assert.equal( createHash('sha256').update(JSON.stringify(goldenRecords)).digest('hex'), - 'ce3c70798bbc50e438605d86eafb28482630ee38dc2baa41a695975c84646822', + '09d76616919435a46a3395194349e696e0d8f6717a24b018515e1f3867ec347a', 'complete yielded record sequence changed', ); @@ -125,6 +126,10 @@ test('kimi provider folds main and subagent wire logs into the existing record l assert.deepEqual(byKind('subagent').map((record) => [record.agent_id, record.parent_tool_use_id, record.agent_type]), [ ['kimi:session-native-1:agent-7', 'kimi:session-native-1:main:call-1', 'explore'], ]); + + const detail = assembleSessionDetail(values); + assert.equal(detail.messages.some((message) => message.text === 'child prompt'), false); + assert.equal(detail.messages.flatMap((message) => message.tool_calls ?? [])[0].result.content.includes('file body'), true); }); test('kimi provider ignores a torn final wire line until it is completed', () => { @@ -193,6 +198,56 @@ test('kimi provider scopes changed-path discovery to one session and bypasses an assert.deepEqual(units.map(unit => unit.key), [firstDir]); }); +test('kimi provider presents user-slash activations as real user prompts', () => { + const root = mkdtempSync(join(tmpdir(), 'obelisk-kimi-user-slash-')); + const sessionDir = join(root, 'sessions', 'workspace-1', 'session-user-slash-1'); + const mainDir = join(sessionDir, 'agents', 'main'); + mkdirSync(mainDir, { recursive: true }); + writeFileSync(join(sessionDir, 'state.json'), JSON.stringify({ workDir: '/tmp/user-slash' })); + const records = [ + { type: 'metadata', protocol_version: '1.5', created_at: 1 }, + { type: 'context.append_message', time: 2, message: { + role: 'user', content: 'User activated the skill and loaded its full instructions.', toolCalls: [], + origin: { + kind: 'skill_activation', trigger: 'user-slash', skillName: 'obelisk', + skillArgs: ' synthesize my history ', + }, + } }, + { type: 'context.append_message', time: 3, message: { + role: 'user', content: 'Expanded plugin command implementation.', toolCalls: [], + origin: { + kind: 'plugin_command', trigger: 'user-slash', pluginId: 'demo', + commandName: 'ship', commandArgs: ' --fast ', + }, + } }, + { type: 'context.append_message', time: 4, message: { + role: 'user', content: 'Model-triggered skill instructions.', toolCalls: [], + origin: { kind: 'skill_activation', trigger: 'model-tool', skillName: 'review' }, + } }, + ]; + writeFileSync(join(mainDir, 'wire.jsonl'), records.map(record => JSON.stringify(record)).join('\n') + '\n'); + const provider = createKimiProvider({ rootDir: root }); + const unit = provider.discover({ lastCursor: () => null })[0]; + + const { values } = drain(provider.parse(unit, null)); + + const messages = values.filter(record => record.kind === 'message'); + assert.deepEqual(messages.map(record => ({ + text: record.text, + is_meta: record.is_meta, + })), [ + { text: '/obelisk synthesize my history', is_meta: 0 }, + { text: '/demo:ship --fast', is_meta: 0 }, + { text: 'Model-triggered skill instructions.', is_meta: 1 }, + ]); + assert.equal(provider.raw({ + source: 'kimi', + messageUuid: messages[0].uuid, + session: { jsonl_path: join(mainDir, 'wire.jsonl') }, + agentId: null, + }).messageText, '/obelisk synthesize my history'); +}); + test('kimi provider maps protocol-1.0 embedded tool calls and results', () => { const root = mkdtempSync(join(tmpdir(), 'obelisk-kimi-legacy-tools-')); const sessionDir = join(root, 'sessions', 'workspace-1', 'session-tools-1'); diff --git a/tests/persist.test.mjs b/tests/persist.test.mjs index 6c529cc..4171fe2 100644 --- a/tests/persist.test.mjs +++ b/tests/persist.test.mjs @@ -92,10 +92,34 @@ test('fresh full re-scan (no prior cursor) resets message_count instead of accum assert.equal(db.prepare('SELECT COUNT(*) c FROM messages').get().c, 3); }); +test('persist round-trips canonical workflow records and their tool relationship', () => { + const db = freshDb(); + function* records() { + yield { + kind: 'workflow', run_id: 'run-1', session_id: 'sid-p', parent_tool_use_id: 'tool-1', + task_id: 'task-1', script: 'review', result_json: '{}', timestamp: '2026-06-10T10:00:00Z', + agent_count: 1, duration_ms: 10, total_tokens: 20, status: 'complete', workflow_name: 'Review', + }; + yield { + kind: 'workflow_agent', agent_id: 'agent-1', run_id: 'run-1', session_id: 'sid-p', + agent_type: 'reviewer', phase: 'review', label: 'Reviewer', state: 'complete', tokens: 20, + }; + return null; + } + + persist(db, { key: 'workflow', sessionId: 'sid-p' }, records()); + + assert.equal(db.prepare('SELECT parent_tool_use_id FROM workflows WHERE run_id=?').get('run-1').parent_tool_use_id, 'tool-1'); + assert.equal(db.prepare('SELECT phase FROM workflow_agents WHERE agent_id=?').get('agent-1').phase, 'review'); + db.close(); +}); + test('delete-session cascades across tables', () => { const db = freshDb(); const unit = fixtureUnit(); persist(db, unit, parse(unit, null)); + db.prepare('INSERT INTO workflows (run_id,session_id) VALUES (?,?)').run('run-delete', 'sid-p'); + db.prepare('INSERT INTO workflow_agents (agent_id,run_id,session_id) VALUES (?,?,?)').run('agent-delete', 'run-delete', 'sid-p'); // Hand-roll a one-shot generator emitting a delete for the session. function* del() { yield { kind: 'delete-session', sessionId: 'sid-p' }; return null; } @@ -104,4 +128,6 @@ test('delete-session cascades across tables', () => { assert.equal(db.prepare('SELECT COUNT(*) c FROM sessions WHERE id=?').get('sid-p').c, 0); assert.equal(db.prepare('SELECT COUNT(*) c FROM messages WHERE session_id=?').get('sid-p').c, 0); assert.equal(db.prepare('SELECT COUNT(*) c FROM tool_calls WHERE session_id=?').get('sid-p').c, 0); + assert.equal(db.prepare('SELECT COUNT(*) c FROM workflows WHERE session_id=?').get('sid-p').c, 0); + assert.equal(db.prepare('SELECT COUNT(*) c FROM workflow_agents WHERE session_id=?').get('sid-p').c, 0); }); diff --git a/tests/provider-registry.test.mjs b/tests/provider-registry.test.mjs index 936d3e9..8363c90 100644 --- a/tests/provider-registry.test.mjs +++ b/tests/provider-registry.test.mjs @@ -70,6 +70,7 @@ test('built-in provider registry exposes every source without caller-side branch ]); assert.deepEqual(registry.watchRoots(), [ '/sources/claude/projects', + '/sources/claude/history.jsonl', '/sources/codex/sessions', '/sources/codex/session_index.jsonl', '/sources/kimi/sessions', diff --git a/tests/provider-schema-stability.test.mjs b/tests/provider-schema-stability.test.mjs index d0f0620..56a08b0 100644 --- a/tests/provider-schema-stability.test.mjs +++ b/tests/provider-schema-stability.test.mjs @@ -3,10 +3,10 @@ import assert from 'node:assert/strict'; import { createHash } from 'node:crypto'; import { readFileSync } from 'node:fs'; -test('provider adapters do not change the frozen SQLite schema', () => { +test('canonical transcript persistence schema changes only by explicit decision', () => { const schema = readFileSync(new URL('../packages/core/src/schema.sql', import.meta.url)); assert.equal( createHash('sha256').update(schema).digest('hex'), - '3e0615ed2db0d7338561df4d51c4240395714c191aa69567ffcdb70efec49826', + 'ef5d0eea6f91c50e78ca5e28ecdc7b3ed5db83db59200642cc25866158f9d307', ); }); diff --git a/tests/provider-session-detail.test.mjs b/tests/provider-session-detail.test.mjs new file mode 100644 index 0000000..ef124f8 --- /dev/null +++ b/tests/provider-session-detail.test.mjs @@ -0,0 +1,212 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, readFileSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; + +import { assembleSessionDetail } from '../packages/core/src/session-detail.ts'; +import { persist } from '../packages/core/src/persist.ts'; +import { parse as parseCodex } from '../packages/core/src/providers/codex.ts'; + +const SCHEMA = readFileSync(new URL('../packages/core/src/schema.sql', import.meta.url), 'utf8'); + +function writeCodexFixture(lines) { + const dir = mkdtempSync(join(tmpdir(), 'obelisk-provider-detail-')); + const path = join(dir, 'rollout.jsonl'); + writeFileSync(path, `${lines.map(line => JSON.stringify(line)).join('\n')}\n`); + return path; +} + +test('a provider record stream assembles directly into session detail', () => { + const threadId = '019e8951-3e7d-7343-a3e3-05bff48a317d'; + const path = writeCodexFixture([ + { + type: 'session_meta', + timestamp: '2026-06-10T10:00:00Z', + payload: { id: threadId, cwd: '/proj', timestamp: '2026-06-10T10:00:00Z' }, + }, + { + type: 'event_msg', + timestamp: '2026-06-10T10:00:01Z', + payload: { type: 'user_message', message: 'inspect the repository' }, + }, + { + type: 'event_msg', + timestamp: '2026-06-10T10:00:02Z', + payload: { type: 'agent_message', message: 'I will inspect it.' }, + }, + { + type: 'response_item', + timestamp: '2026-06-10T10:00:03Z', + payload: { type: 'function_call', call_id: 'call_1', name: 'shell', arguments: '{"cmd":"ls"}' }, + }, + { + type: 'response_item', + timestamp: '2026-06-10T10:00:04Z', + payload: { type: 'function_call_output', call_id: 'call_1', output: 'package.json' }, + }, + ]); + + const records = [...parseCodex({ key: path, sessionId: '' }, null)]; + const detail = assembleSessionDetail(records); + + assert.deepEqual(detail.messages.map(message => message.text), [ + 'inspect the repository', + 'I will inspect it.', + ]); + assert.equal(detail.messages[1].tool_calls?.[0].name, 'shell'); + assert.equal(detail.messages[1].tool_calls?.[0].result?.content, 'package.json'); + + const db = new DatabaseSync(':memory:'); + db.exec(SCHEMA); + persist(db, { key: path, sessionId: '' }, parseCodex({ key: path, sessionId: '' }, null)); + const persistedDetail = assembleSessionDetail({ + session: db.prepare('SELECT * FROM sessions').get(), + messages: db.prepare('SELECT * FROM messages ORDER BY timestamp, uuid').all(), + toolCalls: db.prepare('SELECT * FROM tool_calls').all(), + toolResults: db.prepare('SELECT * FROM tool_results').all(), + }); + assert.deepEqual(persistedDetail, detail); + db.close(); +}); + +test('provider-classified hidden context never reaches session detail', () => { + const threadId = '019e8951-3e7d-7343-a3e3-05bff48a317e'; + const path = writeCodexFixture([ + { + type: 'session_meta', + timestamp: '2026-06-10T10:00:00Z', + payload: { id: threadId, cwd: '/proj', timestamp: '2026-06-10T10:00:00Z' }, + }, + { + type: 'response_item', + timestamp: '2026-06-10T10:00:01Z', + payload: { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: '\n /proj\n' }], + }, + }, + { + type: 'response_item', + timestamp: '2026-06-10T10:00:02Z', + payload: { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: '\nsecret state\n' }], + }, + }, + { + type: 'event_msg', + timestamp: '2026-06-10T10:00:03Z', + payload: { type: 'user_message', message: 'show the actual request' }, + }, + ]); + + const records = [...parseCodex({ key: path, sessionId: '' }, null)]; + const detail = assembleSessionDetail(records); + + assert.deepEqual(detail.messages.map(message => message.text), ['show the actual request']); + assert.equal( + records.filter(record => record.kind === 'message' && record.visibility === 'hidden').length, + 2, + ); +}); + +test('provider normalization removes only structural image wrappers before deduplication', () => { + const threadId = '019e8951-3e7d-7343-a3e3-05bff48a317f'; + const path = writeCodexFixture([ + { + type: 'session_meta', + timestamp: '2026-06-10T10:00:00Z', + payload: { id: threadId, cwd: '/proj', timestamp: '2026-06-10T10:00:00Z' }, + }, + { + type: 'event_msg', + timestamp: '2026-06-10T10:00:01Z', + payload: { type: 'user_message', message: 'look at this screenshot' }, + }, + { + type: 'response_item', + timestamp: '2026-06-10T10:00:01Z', + payload: { + type: 'message', + role: 'user', + content: [ + { type: 'input_text', text: 'look at this screenshot' }, + { type: 'input_text', text: '' }, + { type: 'input_image', image_url: 'data:image/png;base64,AAAA' }, + { type: 'input_text', text: '' }, + ], + }, + }, + ]); + + const records = [...parseCodex({ key: path, sessionId: '' }, null)]; + const detail = assembleSessionDetail(records); + + assert.deepEqual(detail.messages.map(message => message.text), ['look at this screenshot']); +}); + +test('canonical visibility survives persistence before row-based assembly', () => { + const threadId = '019e8951-3e7d-7343-a3e3-05bff48a3180'; + const path = writeCodexFixture([ + { + type: 'session_meta', + timestamp: '2026-06-10T10:00:00Z', + payload: { id: threadId, cwd: '/proj', timestamp: '2026-06-10T10:00:00Z' }, + }, + { + type: 'response_item', + timestamp: '2026-06-10T10:00:01Z', + payload: { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'hidden' }], + }, + }, + { + type: 'event_msg', + timestamp: '2026-06-10T10:00:02Z', + payload: { type: 'user_message', message: 'visible request' }, + }, + ]); + const db = new DatabaseSync(':memory:'); + db.exec(SCHEMA); + + persist(db, { key: path, sessionId: '' }, parseCodex({ key: path, sessionId: '' }, null)); + const messages = db.prepare('SELECT * FROM messages ORDER BY timestamp, uuid').all(); + const assembled = assembleSessionDetail({ messages }).messages; + + assert.equal(messages[0].visibility, 'hidden'); + assert.deepEqual(assembled.map(message => message.text), ['visible request']); + db.close(); +}); + +test('provider normalization classifies Skill instructions before assembly', () => { + const threadId = '019e8951-3e7d-7343-a3e3-05bff48a3181'; + const path = writeCodexFixture([ + { + type: 'session_meta', + timestamp: '2026-06-10T10:00:00Z', + payload: { id: threadId, cwd: '/proj', timestamp: '2026-06-10T10:00:00Z' }, + }, + { + type: 'response_item', + timestamp: '2026-06-10T10:00:01Z', + payload: { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'Base directory for this skill: /tmp/skill\n# Instructions' }], + }, + }, + ]); + + const records = [...parseCodex({ key: path, sessionId: '' }, null)]; + const message = records.find(record => record.kind === 'message'); + + assert.equal(message.content_type, 'skill_instructions'); + assert.equal(message.is_meta, 1); + assert.equal(message.visibility, 'visible'); +}); diff --git a/tests/runtime.test.mjs b/tests/runtime.test.mjs index 021fe27..508359e 100644 --- a/tests/runtime.test.mjs +++ b/tests/runtime.test.mjs @@ -173,8 +173,8 @@ test('runtime indexes Codex root sessions into the shared query helpers', () => })), developerReplay: search('developer replay', { source: 'codex', limit: 5 }).length, rawHasEventLine: raw(${JSON.stringify(`codex:${codexId}:000002`)}, { limit: 1000 })?.text.includes('codex user asks for runtime indexing') || false, - tool: sql('SELECT id, message_uuid, session_id, name FROM tool_calls WHERE id=?', 'codex:call_codex_1')[0], - toolResult: sql('SELECT tool_use_id, message_uuid, session_id, content FROM tool_results WHERE tool_use_id=?', 'codex:call_codex_1')[0], + tool: sql('SELECT id, message_uuid, session_id, name FROM tool_calls WHERE id=?', ${JSON.stringify(`codex:${codexId}:call_codex_1`)})[0], + toolResult: sql('SELECT tool_use_id, message_uuid, session_id, content FROM tool_results WHERE tool_use_id=?', ${JSON.stringify(`codex:${codexId}:call_codex_1`)})[0], overviewSources: overview({ limit: 5 }).totals.sources }; `); @@ -440,7 +440,7 @@ test('runtime maps Codex child threads onto subagents', () => { assert.deepEqual(payload.subagents, [{ agent_id: `codex:${childId}`, session_id: `codex:${parentId}`, - parent_tool_use_id: 'codex:call_spawn_1', + parent_tool_use_id: `codex:${parentId}:call_spawn_1`, agent_type: 'worker', description: 'Plato', messageCount: 2, diff --git a/tests/session-detail-assembly.test.mjs b/tests/session-detail-assembly.test.mjs index 4e7519b..08c6b09 100644 --- a/tests/session-detail-assembly.test.mjs +++ b/tests/session-detail-assembly.test.mjs @@ -1,21 +1,33 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; -import { assembleSessionMessages } from '../app/src/shared/session-detail-assembly.mjs'; +import { assembleSessionDetail } from '../app/src/shared/session-detail-assembly.mjs'; test('session assembly preserves thinking and attaches tool result and subagent evidence', () => { const messages = [ - { uuid: 'thinking-1', type: 'assistant', content_type: 'thinking', text: 'reasoning' }, - { uuid: 'answer-1', type: 'assistant', content_type: 'text', text: 'answer' }, - { uuid: 'tool-1', type: 'assistant', content_type: 'tool_use', text: '' }, - { uuid: 'result-1', type: 'user', content_type: 'tool_result', text: '' }, + { + uuid: 'thinking-1', timestamp: '2026-06-10T10:00:00Z', + type: 'assistant', content_type: 'thinking', text: 'reasoning', + }, + { + uuid: 'answer-1', timestamp: '2026-06-10T10:00:01Z', + type: 'assistant', content_type: 'text', text: 'answer', + }, + { + uuid: 'tool-1', timestamp: '2026-06-10T10:00:02Z', + type: 'assistant', content_type: 'tool_use', text: '', + }, + { + uuid: 'result-1', timestamp: '2026-06-10T10:00:03Z', + type: 'user', content_type: 'tool_result', text: '', + }, ]; - const assembled = assembleSessionMessages({ + const assembled = assembleSessionDetail({ messages, toolCalls: [{ id: 'call-1', message_uuid: 'tool-1', name: 'Agent', input_json: '{"description":"inspect"}' }], toolResults: [{ tool_use_id: 'call-1', message_uuid: 'result-1', content: 'done', is_error: 0 }], subagents: [{ agent_id: 'agent-1', parent_tool_use_id: 'call-1', agent_type: 'reviewer', description: 'inspect' }], workflows: [], - }); + }).messages; assert.equal(assembled.length, 1); assert.equal(assembled[0].uuid, 'answer-1'); @@ -25,34 +37,76 @@ test('session assembly preserves thinking and attaches tool result and subagent }); test('session assembly keeps Skill evidence standalone and embeds matching workflow agents', () => { - const assembled = assembleSessionMessages({ + const assembled = assembleSessionDetail({ messages: [ { uuid: 'skill-1', type: 'assistant', content_type: 'tool_use', text: '' }, - { uuid: 'skill-md', type: 'user', content_type: 'text', is_meta: 1, text: 'Base directory for this skill\n# Skill' }, + { uuid: 'skill-md', type: 'user', content_type: 'skill_instructions', is_meta: 1, text: '# Skill instructions' }, { uuid: 'workflow-1', type: 'assistant', content_type: 'tool_use', text: '' }, ], toolCalls: [ - { id: 'call-skill', message_uuid: 'skill-1', name: 'Skill', input_json: '{"skill":"obelisk"}' }, - { id: 'call-workflow', message_uuid: 'workflow-1', name: 'Workflow', input_json: '{}' }, + { id: 'call-skill', message_uuid: 'skill-1', name: 'Skill', presentation: 'skill', input_json: '{"skill":"obelisk"}' }, + { id: 'call-workflow', message_uuid: 'workflow-1', name: 'Workflow', presentation: 'default', input_json: '{}' }, ], - toolResults: [{ tool_use_id: 'call-workflow', content: 'run-1 complete', is_error: 0 }], + toolResults: [{ tool_use_id: 'call-workflow', content: 'complete', is_error: 0 }], subagents: [], workflows: [{ run_id: 'run-1', + parent_tool_use_id: 'call-workflow', workflow_name: 'review', status: 'complete', agents: [{ agent_id: 'agent-1', phase: 'review', label: 'Reviewer', state: 'complete' }], }], - }); + }).messages; - assert.equal(assembled[0]._skillMd, 'Base directory for this skill\n# Skill'); + assert.equal(assembled[0]._skillMd, '# Skill instructions'); assert.equal(assembled[1].tool_calls[0].workflow.run_id, 'run-1'); assert.deepEqual(assembled[1].tool_calls[0].workflow.agents, [{ agent_id: 'agent-1', phase: 'review', label: 'Reviewer', state: 'complete', - tokens: undefined, - duration_ms: undefined, + tokens: null, + duration_ms: null, }]); }); + +test('session assembly trusts canonical classification instead of parsing provider text', () => { + const detail = assembleSessionDetail({ + messages: [{ + uuid: 'provider-owned-classification', + type: 'user', + content_type: 'text', + is_meta: 0, + text: 'text alone does not define presentation semantics', + }], + }); + + assert.equal(detail.messages[0].is_meta, 0); +}); + +test('canonical ordering is stable across provider and SQLite iteration order', () => { + const detail = assembleSessionDetail([ + { + kind: 'message', uuid: 'b', session_id: 'session', type: 'user', parent_uuid: null, + timestamp: '2026-06-10T10:00:00Z', role: 'user', text: 'second', content_type: 'text', + is_meta: 0, visibility: 'visible', model: null, is_sidechain: 0, agent_id: null, + input_tokens: null, output_tokens: null, cwd: null, skill: null, source: 'test', + }, + { + kind: 'message', uuid: 'a', session_id: 'session', type: 'user', parent_uuid: null, + timestamp: '2026-06-10T10:00:00Z', role: 'user', text: 'first', content_type: 'text', + is_meta: 0, visibility: 'visible', model: null, is_sidechain: 0, agent_id: null, + input_tokens: null, output_tokens: null, cwd: null, skill: null, source: 'test', + }, + ]); + + assert.deepEqual(detail.messages.map(message => message.uuid), ['a', 'b']); +}); + +test('direct session assembly rejects an incomplete provider delta', () => { + assert.throws(() => assembleSessionDetail([{ + kind: 'session', id: 'session', title: null, project: null, + started_at: null, ended_at: null, git_branch: null, version: null, + message_count: 1, countMode: 'delta', jsonl_path: '/session.jsonl', source: 'test', + }]), /fresh full parse/); +}); diff --git a/tests/session-live-reload.test.mjs b/tests/session-live-reload.test.mjs index 4aec3b6..074cdad 100644 --- a/tests/session-live-reload.test.mjs +++ b/tests/session-live-reload.test.mjs @@ -9,7 +9,7 @@ import { loadSessionDetail, materializeSessionDetailPatch, } from '../app/src/renderer/src/data.js'; -import { assembleSessionMessages } from '../app/src/shared/session-detail-assembly.mjs'; +import { assembleSessionDetail } from '../app/src/shared/session-detail-assembly.mjs'; import { createSessionPatch } from '../app/src/shared/session-patch.mjs'; test('live updates coalesce while scrolling and load only the latest after scroll end', async () => { @@ -123,13 +123,13 @@ test('a skipped live patch does not advance the visible patch baseline', async t getSessionWorkflows: async () => [], getSessionSummaries: async () => [], getSessionPatch: async (_id, cursor) => { - const snapshotAtCall = { messages: assembleSessionMessages({ + const snapshotAtCall = { messages: assembleSessionDetail({ messages: rows, toolCalls: [], toolResults: [], subagents: [], workflows: [], - }), workflows: [] }; + }).messages, workflows: [] }; patchCalls++; if (patchCalls === 1) { firstPatchStarted();