From 0598c29aad48afd044b30852e9d72cc5b71516b6 Mon Sep 17 00:00:00 2001 From: tommy0103 Date: Wed, 8 Jul 2026 20:43:54 +0800 Subject: [PATCH] feat(providers): migrate codex indexing to adapter + persist, remove legacy indexers (Phase 5c) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Complete the skill-side provider migration: codex now goes through a pure adapter and the shared persist layer, and the two original monolithic indexers are gone. New: - scripts/providers/codex.ts — pure codex adapter. Full-reparse (buffers the whole file) because the event_msg↔response_item dedup needs whole-file, bidirectional knowledge; emits SessionRecord with countMode 'total'. Handles guardian threads (→ delete-session), agent spawns/tool calls (→ tool_call/subagent), token_count (patched onto the message record) and task_complete (→ message-turn-duration). Contract: - SessionRecord.countMode ('total' | 'delta') tells persist whether to replace or accumulate message_count — claude is line-incremental (delta), codex full-reparse (total). SubagentRecord non-key fields are optional; persist merges them column-wise with COALESCE. MessageTurnDurationRecord.turn_duration_ms is nullable. Orchestration: - buildIndex's codex branch parses via the adapter and writes via persist. An unchanged file is skipped but still swept for stale guardian rows (routed through persist as a delete-session), preserving prior behavior. Cleanup: - Remove the now-unused indexJsonl, indexCodexJsonl, deleteCodexThreadRows and upsertCodexSubagent — their semantics now live in the adapters + persist. indexer.mjs drops from ~840 to 428 lines. Codex pure helpers stay exported for codex.ts and the guardian sweep (physical move deferred to the app-side reorg). - Migrate the upsert drift test off indexJsonl to the claude.parse + persist path, keeping the rowid-stability and count-replace regression guards. Tests: tests/codex-parse.test.mjs (record-stream golden: dedup, tools, token patch, turn-duration, guardian→delete) and tests/codex-index.test.mjs (full buildIndex path: fresh build + incremental full-reparse, total-count replace, no duplicates). Verified equivalent on the real ~/.obelisk index: codex messages 82476 and subagents 522 identical before/after, zero guardian leakage; real incremental confirmed (touch a codex file → reparsed idempotently, unchanged files skipped). lint + typecheck clean, 119/119. --- scripts/indexer.mjs | 458 ++-------------------------- scripts/persist.ts | 22 +- scripts/providers/claude.ts | 3 +- scripts/providers/codex.ts | 220 +++++++++++++ scripts/providers/types.ts | 22 +- tests/codex-index.test.mjs | 78 +++++ tests/codex-parse.test.mjs | 86 ++++++ tests/indexer-upsert-drift.test.mjs | 54 ++-- 8 files changed, 474 insertions(+), 469 deletions(-) create mode 100644 scripts/providers/codex.ts create mode 100644 tests/codex-index.test.mjs create mode 100644 tests/codex-parse.test.mjs diff --git a/scripts/indexer.mjs b/scripts/indexer.mjs index cfbbdb2..909aac1 100644 --- a/scripts/indexer.mjs +++ b/scripts/indexer.mjs @@ -1,6 +1,7 @@ -import { CLAUDE_DIR, CODEX_DIR, openDb, rebuildMemoryFts, trunc, truncJson, extractText, extractContentType, extractMessageIsMeta, filePath, isDir, readLines, fs, path } from './db.mjs'; +import { CLAUDE_DIR, CODEX_DIR, openDb, rebuildMemoryFts, isDir, readLines, fs, path } from './db.mjs'; import { persist } from './persist.ts'; import { parse as claudeParse } from './providers/claude.ts'; +import { parse as codexParse } from './providers/codex.ts'; const PROJECTS_DIR = path.join(CLAUDE_DIR, 'projects'); const HISTORY_PATH = path.join(CLAUDE_DIR, 'history.jsonl'); @@ -103,116 +104,6 @@ function needsReindex(db, fp) { return mt > row.mtime ? { needed: true, skip: row.lines_processed } : { needed: false, skip: 0 }; } -function indexJsonl(db, fi) { - const { needed, skip } = needsReindex(db, fi.path); - if (!needed) return; - const mt = fs.statSync(fi.path).mtimeMs; - - const ins = { - 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 (?,?,?,?,?,?,?,?,?,?,?)'), - 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 (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) - 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, - 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 (?,?,?,?,?,?)'), - 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 (?,?,?,?,?)'), - idx: db.prepare('INSERT OR REPLACE INTO index_state (jsonl_path,mtime,lines_processed) VALUES (?,?,?)'), - }; - - const existing = !fi.isSubagent ? db.prepare('SELECT * FROM sessions WHERE id = ?').get(fi.sessionId) : null; - const sm = { - started_at: existing?.started_at || null, - ended_at: existing?.ended_at || null, - git_branch: existing?.git_branch || null, - version: existing?.version || null, - title: existing?.title || null, - n: skip > 0 ? (existing?.message_count || 0) : 0, - cwds: [], - }; - - let lineNum = 0; - readLines(fi.path, (line) => { - lineNum++; - if (lineNum <= skip) return; - let obj; - try { obj = JSON.parse(line); } catch { return; } - const sid = fi.sessionId; - const ts = obj.timestamp || null; - - if (obj.type === 'ai-title' && obj.aiTitle) { sm.title = obj.aiTitle; return; } - if (obj.type === 'system' && obj.subtype === 'away_summary' && obj.content) { - ins.sum.run(obj.uuid || `${sid}-away-${ts}`, sid, ts, 'away_summary', obj.content); - return; - } - if (obj.type === 'system' && obj.subtype === 'turn_duration' && obj.parentUuid && obj.durationMs) { - db.prepare('UPDATE messages SET turn_duration_ms=? WHERE uuid=?').run(obj.durationMs, obj.parentUuid); - return; - } - if (obj.type !== 'user' && obj.type !== 'assistant') return; - - if (ts && (!sm.started_at || ts < sm.started_at)) sm.started_at = ts; - if (ts && (!sm.ended_at || ts > sm.ended_at)) sm.ended_at = ts; - if (obj.gitBranch) sm.git_branch = obj.gitBranch; - if (obj.version) sm.version = obj.version; - sm.n++; - if (!fi.isSubagent && obj.cwd) sm.cwds.push(obj.cwd); - - const msg = obj.message || {}; - const text = extractText(msg.content); - const contentType = extractContentType(msg.content); - const isMeta = extractMessageIsMeta(obj, text); - const usage = msg.usage || {}; - const aid = fi.isSubagent ? fi.agentId : (obj.agentId || null); - - if (obj.uuid) { - ins.msg.run(obj.uuid, sid, obj.type, obj.parentUuid || null, ts, - msg.role || obj.type, text, contentType, isMeta, msg.model || null, - obj.isSidechain ? 1 : 0, aid, usage.input_tokens || null, usage.output_tokens || null, - obj.cwd || null, obj.attributionSkill || null, 'claude'); - } - - if (obj.type === 'assistant' && Array.isArray(msg.content)) { - for (const b of msg.content) { - if (b.type === 'tool_use' && b.id) - ins.tc.run(b.id, obj.uuid, sid, b.name, truncJson(b.input || {}), filePath(b.name, b.input)); - } - } - - if (obj.type === 'user' && Array.isArray(msg.content)) { - for (const b of msg.content) { - if (b.type !== 'tool_result' || !b.tool_use_id) continue; - const rt = typeof b.content === 'string' ? b.content - : Array.isArray(b.content) ? b.content.map(c => c.text || '').join('\n') : ''; - ins.tr.run(b.tool_use_id, obj.uuid, sid, trunc(rt), obj.toolUseResult?.filePath || null, b.is_error ? 1 : 0); - } - } - }); - - if (!fi.isSubagent) { - const pp = inferProjectPath(fi.project, sm.cwds); - ins.ses.run(fi.sessionId, sm.title, fi.project, pp, sm.started_at, sm.ended_at, sm.git_branch, sm.version, sm.n, fi.path, 'claude'); - } - ins.idx.run(fi.path, mt, lineNum); -} - function codexDbId(id) { if (!id) return null; const raw = String(id).replace(/^codex:/, ''); @@ -247,25 +138,6 @@ function codexIsGuardianThread(meta, records = []) { return records.some(({ obj }) => obj?.payload?.model === 'codex-auto-review' || obj?.model === 'codex-auto-review'); } -function deleteCodexThreadRows(db, threadRawId) { - const threadId = codexDbId(threadRawId); - if (!threadId) return; - db.prepare(` - DELETE FROM tool_results - WHERE session_id = ? - OR message_uuid IN (SELECT uuid FROM messages WHERE session_id = ? OR agent_id = ?) - `).run(threadId, threadId, threadId); - db.prepare(` - DELETE FROM tool_calls - WHERE session_id = ? - OR message_uuid IN (SELECT uuid FROM messages WHERE session_id = ? OR agent_id = ?) - `).run(threadId, threadId, threadId); - db.prepare('DELETE FROM messages WHERE session_id = ? OR agent_id = ?').run(threadId, threadId); - db.prepare('DELETE FROM subagents WHERE agent_id = ? OR session_id = ?').run(threadId, threadId); - db.prepare('DELETE FROM summaries WHERE session_id = ?').run(threadId); - db.prepare('DELETE FROM sessions WHERE id = ?').run(threadId); -} - function readCodexGuardianThreadInfo(filePath) { const records = []; let metaRecord = null; @@ -356,299 +228,6 @@ function codexToolOutput(payload) { return null; } -function upsertCodexSubagent(db, { - agentId, - sessionId, - parentToolUseId = null, - agentType = null, - description = null, - durationMs = null, - totalTokens = null, -} = {}) { - if (!agentId || !sessionId) return; - db.prepare(` - INSERT INTO subagents (agent_id,session_id,parent_tool_use_id,agent_type,description,duration_ms,total_tokens) - VALUES (?,?,?,?,?,?,?) - ON CONFLICT(agent_id) DO UPDATE SET - session_id=excluded.session_id, - parent_tool_use_id=COALESCE(excluded.parent_tool_use_id, subagents.parent_tool_use_id), - agent_type=COALESCE(excluded.agent_type, subagents.agent_type), - description=COALESCE(excluded.description, subagents.description), - duration_ms=COALESCE(excluded.duration_ms, subagents.duration_ms), - total_tokens=COALESCE(excluded.total_tokens, subagents.total_tokens) - `).run(agentId, sessionId, parentToolUseId, agentType, description, durationMs, totalTokens); -} - -function indexCodexJsonl(db, fi) { - const state = needsReindex(db, fi.path); - if (!state.needed) { - const guardian = readCodexGuardianThreadInfo(fi.path); - if (guardian) deleteCodexThreadRows(db, guardian.threadRawId); - return; - } - const mt = fs.statSync(fi.path).mtimeMs; - const records = []; - let lineNum = 0; - readLines(fi.path, (line) => { - lineNum++; - try { - records.push({ lineNum, obj: JSON.parse(line) }); - } catch {} - }); - - const metaRecord = records.find(r => r.obj?.type === 'session_meta' && r.obj.payload?.id); - if (!metaRecord) { - db.prepare('INSERT OR REPLACE INTO index_state (jsonl_path,mtime,lines_processed) VALUES (?,?,?)').run(fi.path, mt, lineNum); - return; - } - - const meta = metaRecord.obj.payload; - const threadRawId = codexRawId(meta.id); - if (codexIsGuardianThread(meta, records)) { - deleteCodexThreadRows(db, threadRawId); - db.prepare('INSERT OR REPLACE INTO index_state (jsonl_path,mtime,lines_processed) VALUES (?,?,?)').run(fi.path, mt, lineNum); - return; - } - const parentRawId = codexParentThreadId(meta); - const sessionId = codexDbId(parentRawId || threadRawId); - const agentId = parentRawId ? codexDbId(threadRawId) : null; - const isSidechain = agentId ? 1 : 0; - const projectPath = normalizeObservedCwd(meta.cwd); - const project = projectSlugFromPath(projectPath); - const sm = { - started_at: meta.timestamp || metaRecord.obj.timestamp || null, - ended_at: meta.timestamp || metaRecord.obj.timestamp || null, - git_branch: meta.git?.branch || null, - version: meta.cli_version || null, - title: null, - n: 0, - cwds: projectPath ? [projectPath] : [], - lastMessageUuid: null, - lastTextAssistantUuid: null, - totalInputTokens: 0, - totalOutputTokens: 0, - }; - - const ins = { - 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 (?,?,?,?,?,?,?,?,?,?,?)'), - 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 (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) - 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, - 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 (?,?,?,?,?,?)'), - tr: db.prepare('INSERT OR REPLACE INTO tool_results (tool_use_id,message_uuid,session_id,content,file_path,is_error) VALUES (?,?,?,?,?,?)'), - idx: db.prepare('INSERT OR REPLACE INTO index_state (jsonl_path,mtime,lines_processed) VALUES (?,?,?)'), - dur: db.prepare('UPDATE messages SET turn_duration_ms=? WHERE uuid=?'), - usage: db.prepare('UPDATE messages SET input_tokens=?, output_tokens=? WHERE uuid=?'), - }; - - let currentCwd = projectPath; - let currentModel = null; - const eventMessageKeys = new Set(); - const callMessageUuids = new Map(); - - for (const { obj } of records) { - if (obj?.type !== 'event_msg') continue; - const payload = obj.payload || {}; - if (payload.type !== 'user_message' && payload.type !== 'agent_message') continue; - const text = codexEventText(payload); - if (text === null) continue; - eventMessageKeys.add(codexVisibleMessageKey(payload.type === 'user_message' ? 'user' : 'assistant', text)); - } - - const updateBounds = (ts) => { - if (!ts) return; - if (!sm.started_at || ts < sm.started_at) sm.started_at = ts; - if (!sm.ended_at || ts > sm.ended_at) sm.ended_at = ts; - }; - - const insertMessage = ({ uuid, type, role, text = null, contentType = 'text', timestamp, isMeta = 0 }) => { - ins.msg.run( - uuid, - sessionId, - type, - sm.lastMessageUuid, - timestamp || null, - role, - trunc(text), - contentType, - isMeta, - currentModel, - isSidechain, - agentId, - null, - null, - currentCwd, - null, - 'codex', - ); - sm.lastMessageUuid = uuid; - if (!agentId) sm.n++; - if (type === 'assistant' && contentType === 'text') sm.lastTextAssistantUuid = uuid; - updateBounds(timestamp); - return uuid; - }; - - for (const { lineNum: currentLine, obj } of records) { - const ts = obj.timestamp || null; - if (obj.type === 'session_meta') { - if (obj.payload?.cwd) { - currentCwd = normalizeObservedCwd(obj.payload.cwd) || currentCwd; - if (currentCwd) sm.cwds.push(currentCwd); - } - if (obj.payload?.git?.branch) sm.git_branch = obj.payload.git.branch; - if (obj.payload?.cli_version) sm.version = obj.payload.cli_version; - updateBounds(obj.payload?.timestamp || ts); - continue; - } - if (obj.type === 'turn_context') { - currentCwd = normalizeObservedCwd(obj.payload?.cwd) || currentCwd; - currentModel = obj.payload?.model || currentModel; - if (currentCwd) sm.cwds.push(currentCwd); - updateBounds(ts); - continue; - } - if (obj.type === 'event_msg') { - const payload = obj.payload || {}; - if (payload.type === 'user_message' || payload.type === 'agent_message' || payload.type === 'agent_reasoning') { - const text = codexEventText(payload); - if (text === null) continue; - const isReasoning = payload.type === 'agent_reasoning'; - insertMessage({ - uuid: codexLineUuid(threadRawId, currentLine), - type: payload.type === 'user_message' ? 'user' : 'assistant', - role: payload.type === 'user_message' ? 'user' : 'assistant', - text, - contentType: isReasoning ? 'thinking' : 'text', - timestamp: ts, - }); - continue; - } - if (payload.type === 'collab_agent_spawn_end' && payload.call_id && payload.new_thread_id) { - const uuid = insertMessage({ - uuid: codexLineUuid(threadRawId, currentLine), - type: 'assistant', - role: 'assistant', - text: null, - contentType: 'tool_use', - timestamp: ts, - }); - const toolId = codexCallId(payload.call_id); - const description = payload.new_agent_nickname || payload.new_agent_role || 'Agent'; - const input = { - description, - subagent_type: payload.new_agent_role || 'Agent', - prompt: payload.prompt || '', - new_thread_id: payload.new_thread_id, - model: payload.model || null, - reasoning_effort: payload.reasoning_effort || null, - }; - ins.tc.run(toolId, uuid, sessionId, 'Agent', truncJson(input), null); - callMessageUuids.set(toolId, uuid); - upsertCodexSubagent(db, { - agentId: codexDbId(payload.new_thread_id), - sessionId, - parentToolUseId: toolId, - agentType: payload.new_agent_role || null, - description, - }); - continue; - } - if (payload.type === 'task_complete') { - if (sm.lastTextAssistantUuid && payload.duration_ms !== undefined) { - ins.dur.run(payload.duration_ms || null, sm.lastTextAssistantUuid); - } - updateBounds(ts); - continue; - } - if (payload.type === 'token_count') { - const usage = codexUsage(payload); - if (usage.inputTokens !== null) sm.totalInputTokens = usage.inputTokens; - if (usage.outputTokens !== null) sm.totalOutputTokens = usage.outputTokens; - if (sm.lastTextAssistantUuid && (usage.inputTokens !== null || usage.outputTokens !== null)) { - ins.usage.run(usage.inputTokens, usage.outputTokens, sm.lastTextAssistantUuid); - } - continue; - } - if (payload.type === 'thread_name_updated' && payload.thread_name) { - sm.title = payload.thread_name; - } - continue; - } - if (obj.type !== 'response_item') continue; - const payload = obj.payload || {}; - if (payload.type === 'message' && payload.role !== 'developer') { - const text = codexMessagePayloadText(payload); - const role = payload.role || 'assistant'; - if (text !== null && !eventMessageKeys.has(codexVisibleMessageKey(role, text))) { - insertMessage({ - uuid: codexLineUuid(threadRawId, currentLine), - type: role === 'user' ? 'user' : 'assistant', - role, - text, - contentType: 'text', - timestamp: ts, - }); - } - continue; - } - if (['function_call', 'custom_tool_call', 'tool_search_call', 'web_search_call'].includes(payload.type) && payload.call_id) { - const uuid = insertMessage({ - uuid: codexLineUuid(threadRawId, currentLine), - type: 'assistant', - role: 'assistant', - text: null, - contentType: 'tool_use', - timestamp: ts, - }); - const name = payload.name || payload.tool || payload.type.replace(/_call$/, ''); - const toolId = codexCallId(payload.call_id); - ins.tc.run(toolId, uuid, sessionId, name, truncJson(codexToolInput(payload)), null); - callMessageUuids.set(toolId, uuid); - continue; - } - if (['function_call_output', 'custom_tool_call_output', 'tool_search_output'].includes(payload.type) && payload.call_id) { - const toolId = codexCallId(payload.call_id); - ins.tr.run(toolId, callMessageUuids.get(toolId) || null, sessionId, trunc(codexToolOutput(payload) || ''), null, payload.is_error ? 1 : 0); - } - } - - if (agentId) { - const tokenTotal = (sm.totalInputTokens || 0) + (sm.totalOutputTokens || 0); - const started = sm.started_at ? new Date(sm.started_at).getTime() : null; - const ended = sm.ended_at ? new Date(sm.ended_at).getTime() : null; - upsertCodexSubagent(db, { - agentId, - sessionId, - agentType: codexAgentRole(meta), - description: codexAgentNickname(meta), - durationMs: started && ended ? ended - started : null, - totalTokens: tokenTotal || null, - }); - } else { - const pp = inferProjectPath(project, sm.cwds); - ins.ses.run(sessionId, sm.title, project, pp, sm.started_at, sm.ended_at, sm.git_branch, sm.version, sm.n, fi.path, 'codex'); - } - ins.idx.run(fi.path, mt, lineNum); -} - function indexCodexSessionIndex(db) { const indexPath = path.join(CODEX_DIR, 'session_index.jsonl'); if (!fs.existsSync(indexPath)) return; @@ -763,6 +342,13 @@ function shouldSkipBuild(db, { now = Date.now() } = {}) { return { skip: false }; } +// A one-shot record stream that retracts a session, for routing guardian sweeps +// through persist (the single db writer) instead of deleting rows directly. +function* guardianDelete(sessionId) { + yield { kind: 'delete-session', sessionId }; + return null; +} + function buildIndex({ force = false } = {}) { const db = openDb(); if (!force) { @@ -782,7 +368,20 @@ function buildIndex({ force = false } = {}) { db.exec('BEGIN'); try { if (f.source === 'codex') { - indexCodexJsonl(db, f); + // Codex goes through the pure adapter + shared persist (docs/adr/0001), + // full-reparse (countMode 'total') when the file changed. An unchanged + // file is not reparsed, but is still swept for stale guardian rows: a + // guardian/auto-review thread must never linger in the index, even if it + // was indexed before guardian detection removed it. + const { needed } = needsReindex(db, f.path); + if (needed) { + persist(db, { key: f.path, sessionId: '' }, codexParse({ key: f.path, sessionId: '' }, null)); + } else { + const guardian = readCodexGuardianThreadInfo(f.path); + if (guardian) { + persist(db, { key: f.path, sessionId: '' }, guardianDelete(codexDbId(guardian.threadRawId))); + } + } } else { // Claude transcripts now go through the pure adapter + shared persist // (docs/adr/0001). needsReindex keeps the "skip unchanged file" fast path; @@ -817,4 +416,13 @@ function buildIndex({ force = false } = {}) { db.close(); } -export { buildIndex, indexJsonl, discoverJsonlFiles, inferProjectPath, refreshSessionProjectPaths, shouldSkipBuild }; +export { + buildIndex, discoverJsonlFiles, inferProjectPath, refreshSessionProjectPaths, shouldSkipBuild, + // Pure codex helpers consumed by providers/codex.ts (temporary export; they + // move into codex.ts once indexCodexJsonl is removed in the 5c cleanup). + discoverCodexJsonlFiles, normalizeObservedCwd, projectSlugFromPath, + codexRawId, codexDbId, codexCallId, codexLineUuid, codexParentThreadId, + codexIsGuardianThread, codexAgentNickname, codexAgentRole, codexUsage, + codexEventText, codexMessagePayloadText, codexVisibleMessageKey, + codexToolInput, codexToolOutput, +}; diff --git a/scripts/persist.ts b/scripts/persist.ts index 10f5e3a..0815f4e 100644 --- a/scripts/persist.ts +++ b/scripts/persist.ts @@ -33,10 +33,19 @@ function statements(db: any) { 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 (?,?,?,?,?,?,?,?,?,?,?)'), + sub: db.prepare(` + INSERT INTO subagents (agent_id,session_id,parent_tool_use_id,agent_type,description,duration_ms,total_tokens) + VALUES (?,?,?,?,?,?,?) + ON CONFLICT(agent_id) DO UPDATE SET + session_id=excluded.session_id, + parent_tool_use_id=COALESCE(excluded.parent_tool_use_id, subagents.parent_tool_use_id), + agent_type=COALESCE(excluded.agent_type, subagents.agent_type), + description=COALESCE(excluded.description, subagents.description), + duration_ms=COALESCE(excluded.duration_ms, subagents.duration_ms), + total_tokens=COALESCE(excluded.total_tokens, subagents.total_tokens)`), 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=?'), - getState: db.prepare('SELECT lines_processed FROM index_state WHERE jsonl_path=?'), }; } @@ -54,9 +63,6 @@ function deleteSession(db: any, sessionId: string) { // (also written to index_state). `db` is any SQLite handle sharing prepare/run. export function persist(db: any, unit: IndexUnit, gen: Generator): Cursor { const st = statements(db); - // A prior lines_processed>0 means this parse resumed, so message_count must - // accumulate onto the existing row rather than reset (matches indexJsonl). - const resuming = ((st.getState.get(unit.key)?.lines_processed as number) || 0) > 0; const write = (r: IndexRecord) => { switch (r.kind) { @@ -72,11 +78,17 @@ export function persist(db: any, unit: IndexUnit, gen: Generator 0 ? 'delta' : 'total', + jsonl_path: unit.key, source: 'claude', }); } diff --git a/scripts/providers/codex.ts b/scripts/providers/codex.ts new file mode 100644 index 0000000..8025bee --- /dev/null +++ b/scripts/providers/codex.ts @@ -0,0 +1,220 @@ +// Codex provider adapter (see docs/adr/0001). +// +// Pure: discovers Codex rollout files and parses one into a record stream. It +// never touches the Obelisk database. Unlike claude, codex is a FULL-REPARSE +// adapter: it buffers every line and re-emits every record on each run, because +// the event_msg ↔ response_item dedup needs whole-file (bidirectional) knowledge +// (the matching pair sits ±1 line apart but in either order). Hence the session +// record uses countMode 'total' (persist replaces the count, never accumulates). +// The per-line logic mirrors the original indexCodexJsonl. + +import { createRequire } from 'node:module'; +const require = createRequire(import.meta.url); +const fs = require('node:fs'); + +import { trunc, truncJson, readLines } from '../db.mjs'; +import { + discoverCodexJsonlFiles, normalizeObservedCwd, projectSlugFromPath, + codexRawId, codexDbId, codexCallId, codexLineUuid, codexParentThreadId, + codexIsGuardianThread, codexAgentNickname, codexAgentRole, codexUsage, + codexEventText, codexMessagePayloadText, codexVisibleMessageKey, + codexToolInput, codexToolOutput, +} from '../indexer.mjs'; + +import type { Cursor, DiscoverContext, IndexRecord, IndexUnit, MessageRecord, Provider } from './types.ts'; + +export const name = 'codex'; + +export function discover(_ctx: DiscoverContext): IndexUnit[] { + return discoverCodexJsonlFiles().map((f: any) => ({ key: f.path, sessionId: '', meta: { source: 'codex' } })); +} + +export function* parse(unit: IndexUnit, _cursor: Cursor): Generator { + const mtime = fs.statSync(unit.key).mtimeMs; + const records: { lineNum: number; obj: any }[] = []; + let lineNum = 0; + readLines(unit.key, (line: string) => { + lineNum++; + try { records.push({ lineNum, obj: JSON.parse(line) }); } catch { /* skip malformed */ } + }); + const outCursor = `${mtime}:${lineNum}`; + + const metaRecord = records.find(r => r.obj?.type === 'session_meta' && r.obj.payload?.id); + if (!metaRecord) return outCursor; + + const meta = metaRecord.obj.payload; + const threadRawId = codexRawId(meta.id) as string; + if (codexIsGuardianThread(meta, records)) { + yield { kind: 'delete-session', sessionId: codexDbId(threadRawId) as string }; + return outCursor; + } + + const parentRawId = codexParentThreadId(meta); + const sessionId = codexDbId(parentRawId || threadRawId) as string; + const agentId = (parentRawId ? codexDbId(threadRawId) : null) as string | null; + const isSidechain: 0 | 1 = agentId ? 1 : 0; + const project = projectSlugFromPath(normalizeObservedCwd(meta.cwd)); + const lineUuid = (n: number): string => codexLineUuid(threadRawId, n) as string; + + const out: IndexRecord[] = []; + const msgByUuid = new Map(); + const sm = { + started_at: (meta.timestamp || metaRecord.obj.timestamp || null) as string | null, + ended_at: (meta.timestamp || metaRecord.obj.timestamp || null) as string | null, + git_branch: (meta.git?.branch || null) as string | null, + version: (meta.cli_version || null) as string | null, + title: null as string | null, + n: 0, + lastMessageUuid: null as string | null, + lastTextAssistantUuid: null as string | null, + totalInputTokens: 0, + totalOutputTokens: 0, + }; + + let currentCwd = normalizeObservedCwd(meta.cwd); + let currentModel: string | null = null; + const eventMessageKeys = new Set(); + const callMessageUuids = new Map(); + + const updateBounds = (ts: string | null) => { + if (!ts) return; + if (!sm.started_at || ts < sm.started_at) sm.started_at = ts; + if (!sm.ended_at || ts > sm.ended_at) sm.ended_at = ts; + }; + + const insertMessage = ({ uuid, type, role, text = null, contentType = 'text', timestamp, isMeta = 0 }: { + uuid: string; type: string; role: string; text?: string | null; contentType?: string; timestamp: string | null; isMeta?: 0 | 1; + }) => { + 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, + input_tokens: null, output_tokens: null, cwd: currentCwd, skill: null, source: 'codex', + }; + out.push(rec); + msgByUuid.set(uuid, rec); + sm.lastMessageUuid = uuid; + if (!agentId) sm.n++; + if (type === 'assistant' && contentType === 'text') sm.lastTextAssistantUuid = uuid; + updateBounds(timestamp); + return uuid; + }; + + // First pass: collect visible event_msg keys so duplicate response_items drop. + for (const { obj } of records) { + if (obj?.type !== 'event_msg') continue; + const payload = obj.payload || {}; + if (payload.type !== 'user_message' && payload.type !== 'agent_message') continue; + const text = codexEventText(payload); + if (text === null) continue; + eventMessageKeys.add(codexVisibleMessageKey(payload.type === 'user_message' ? 'user' : 'assistant', text)); + } + + for (const { lineNum: currentLine, obj } of records) { + const ts = obj.timestamp || null; + if (obj.type === 'session_meta') { + if (obj.payload?.cwd) currentCwd = normalizeObservedCwd(obj.payload.cwd) || currentCwd; + if (obj.payload?.git?.branch) sm.git_branch = obj.payload.git.branch; + if (obj.payload?.cli_version) sm.version = obj.payload.cli_version; + updateBounds(obj.payload?.timestamp || ts); + continue; + } + if (obj.type === 'turn_context') { + currentCwd = normalizeObservedCwd(obj.payload?.cwd) || currentCwd; + currentModel = obj.payload?.model || currentModel; + updateBounds(ts); + continue; + } + if (obj.type === 'event_msg') { + const payload = obj.payload || {}; + if (payload.type === 'user_message' || payload.type === 'agent_message' || payload.type === 'agent_reasoning') { + const text = codexEventText(payload); + if (text === null) continue; + const isReasoning = payload.type === 'agent_reasoning'; + insertMessage({ + uuid: lineUuid(currentLine), + type: payload.type === 'user_message' ? 'user' : 'assistant', + role: payload.type === 'user_message' ? 'user' : 'assistant', + text, contentType: isReasoning ? 'thinking' : 'text', timestamp: ts, + }); + continue; + } + if (payload.type === 'collab_agent_spawn_end' && payload.call_id && payload.new_thread_id) { + const uuid = insertMessage({ uuid: lineUuid(currentLine), type: 'assistant', role: 'assistant', text: null, contentType: 'tool_use', timestamp: ts }); + const toolId = codexCallId(payload.call_id) as string; + const description = payload.new_agent_nickname || payload.new_agent_role || 'Agent'; + const input = { + description, subagent_type: payload.new_agent_role || 'Agent', prompt: payload.prompt || '', + new_thread_id: payload.new_thread_id, model: payload.model || null, reasoning_effort: payload.reasoning_effort || null, + }; + out.push({ kind: 'tool_call', id: toolId, message_uuid: uuid, session_id: sessionId, name: 'Agent', input_json: truncJson(input) as string, file_path: null }); + callMessageUuids.set(toolId, uuid); + out.push({ kind: 'subagent', agent_id: codexDbId(payload.new_thread_id) as string, session_id: sessionId, parent_tool_use_id: toolId, agent_type: payload.new_agent_role || null, description }); + continue; + } + if (payload.type === 'task_complete') { + if (sm.lastTextAssistantUuid && payload.duration_ms !== undefined) { + out.push({ kind: 'message-turn-duration', uuid: sm.lastTextAssistantUuid, turn_duration_ms: payload.duration_ms || null }); + } + updateBounds(ts); + continue; + } + if (payload.type === 'token_count') { + const usage = codexUsage(payload); + if (usage.inputTokens !== null) sm.totalInputTokens = usage.inputTokens; + if (usage.outputTokens !== null) sm.totalOutputTokens = usage.outputTokens; + if (sm.lastTextAssistantUuid && (usage.inputTokens !== null || usage.outputTokens !== null)) { + const rec = msgByUuid.get(sm.lastTextAssistantUuid); + if (rec) { rec.input_tokens = usage.inputTokens; rec.output_tokens = usage.outputTokens; } + } + continue; + } + if (payload.type === 'thread_name_updated' && payload.thread_name) sm.title = payload.thread_name; + continue; + } + if (obj.type !== 'response_item') continue; + const payload = obj.payload || {}; + if (payload.type === 'message' && payload.role !== 'developer') { + const text = codexMessagePayloadText(payload); + const role = payload.role || 'assistant'; + if (text !== null && !eventMessageKeys.has(codexVisibleMessageKey(role, text))) { + insertMessage({ uuid: lineUuid(currentLine), type: role === 'user' ? 'user' : 'assistant', role, text, contentType: 'text', timestamp: ts }); + } + continue; + } + if (['function_call', 'custom_tool_call', 'tool_search_call', 'web_search_call'].includes(payload.type) && payload.call_id) { + const uuid = insertMessage({ uuid: lineUuid(currentLine), type: 'assistant', role: 'assistant', text: null, contentType: 'tool_use', timestamp: ts }); + const name = payload.name || payload.tool || payload.type.replace(/_call$/, ''); + const toolId = codexCallId(payload.call_id) as string; + out.push({ kind: 'tool_call', id: toolId, message_uuid: uuid, session_id: sessionId, name, input_json: truncJson(codexToolInput(payload)) as string, file_path: null }); + callMessageUuids.set(toolId, uuid); + continue; + } + if (['function_call_output', 'custom_tool_call_output', 'tool_search_output'].includes(payload.type) && payload.call_id) { + const toolId = codexCallId(payload.call_id) as string; + out.push({ kind: 'tool_result', tool_use_id: toolId, message_uuid: callMessageUuids.get(toolId) || '', session_id: sessionId, content: trunc(codexToolOutput(payload) || ''), file_path: null, is_error: payload.is_error ? 1 : 0 }); + } + } + + if (agentId) { + const started = sm.started_at ? new Date(sm.started_at).getTime() : null; + const ended = sm.ended_at ? new Date(sm.ended_at).getTime() : null; + const tokenTotal = (sm.totalInputTokens || 0) + (sm.totalOutputTokens || 0); + out.push({ + kind: 'subagent', agent_id: agentId, session_id: sessionId, + agent_type: codexAgentRole(meta), description: codexAgentNickname(meta), + duration_ms: started && ended ? ended - started : null, total_tokens: tokenTotal || null, + }); + } else { + out.push({ + kind: 'session', id: sessionId, title: sm.title, project, + started_at: sm.started_at, ended_at: sm.ended_at, git_branch: sm.git_branch, version: sm.version, + message_count: sm.n, countMode: 'total', jsonl_path: unit.key, source: 'codex', + }); + } + + yield* out; + return outCursor; +} + +export const codexProvider: Provider = { name, discover, parse }; diff --git a/scripts/providers/types.ts b/scripts/providers/types.ts index 713bd6d..ba6d0df 100644 --- a/scripts/providers/types.ts +++ b/scripts/providers/types.ts @@ -113,15 +113,18 @@ export interface SummaryRecord { content: string; } +// One codex subagent. Like workflow_agent, a row can be contributed by more than +// one point in the parse (the spawn event vs the agent's own thread), so non-key +// fields are optional and persist merges them column-wise with COALESCE. export interface SubagentRecord { kind: 'subagent'; agent_id: string; session_id: string; - parent_tool_use_id: string | null; - agent_type: string | null; - description: string | null; - duration_ms: number | null; - total_tokens: number | null; + parent_tool_use_id?: string | null; + agent_type?: string | null; + description?: string | null; + duration_ms?: number | null; + total_tokens?: number | null; } // A workflow run. `agent_count` is intentionally absent: it is a derived @@ -170,7 +173,7 @@ export interface WorkflowAgentRecord { export interface MessageTurnDurationRecord { kind: 'message-turn-duration'; uuid: string; - turn_duration_ms: number; + turn_duration_ms: number | null; } // Retraction op (not a table). The adapter emits this when a previously-indexed @@ -188,6 +191,12 @@ export interface DeleteSessionRecord { // fill-if-null (COALESCE) so those never clobber a value already present. // project_path is NOT set here — the orchestration's global pass derives it from // persisted message cwds (refreshSessionProjectPaths). +// +// countMode tells persist how to treat message_count, because providers differ: +// a line-incremental adapter (claude) yields only new messages ('delta', persist +// accumulates onto the existing row); a full-reparse adapter (codex) yields every +// message each run ('total', persist replaces). A 'delta' parse from an empty +// cursor is equivalent to 'total'. export interface SessionRecord { kind: 'session'; id: string; @@ -198,6 +207,7 @@ export interface SessionRecord { git_branch: string | null; version: string | null; message_count: number; + countMode: 'total' | 'delta'; jsonl_path: string; source: string; } diff --git a/tests/codex-index.test.mjs b/tests/codex-index.test.mjs new file mode 100644 index 0000000..97a39fd --- /dev/null +++ b/tests/codex-index.test.mjs @@ -0,0 +1,78 @@ +// Phase 5c: exercises the full codex buildIndex path (discover → codex.parse → +// persist) for both a fresh full build and an incremental rebuild after append. +// Codex is full-reparse with countMode 'total', so growth must REPLACE the count +// (not accumulate) and upsert messages (no duplicates). + +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createRequire } from 'node:module'; +import { mkdtempSync, mkdirSync, writeFileSync, appendFileSync, utimesSync, statSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { dirname, join, resolve } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { spawnSync } from 'node:child_process'; + +const repoRoot = resolve(dirname(fileURLToPath(import.meta.url)), '..'); +const require = createRequire(import.meta.url); +const { DatabaseSync } = require('node:sqlite'); + +function runRuntime(args, home) { + return spawnSync(process.execPath, ['scripts/runtime.mjs', ...args], { + cwd: repoRoot, env: { ...process.env, HOME: home }, encoding: 'utf8', + }); +} + +const ID = '019ed000-0000-7000-8000-000000000001'; + +function metaLine() { + return JSON.stringify({ type: 'session_meta', timestamp: '2026-06-15T10:00:00Z', payload: { id: ID, timestamp: '2026-06-15T10:00:00Z', cwd: '/tmp/cdx', cli_version: '1.0' } }); +} +function evt(type, message, ts) { + return JSON.stringify({ type: 'event_msg', timestamp: ts, payload: { type, message } }); +} + +function clearDebounce(home) { + const db = new DatabaseSync(join(home, '.obelisk', 'obelisk.sqlite')); + db.prepare("DELETE FROM index_state WHERE jsonl_path='__last_build__'").run(); + db.close(); +} + +function codexCounts(home) { + writeFileSync(join(home, 'q.mjs'), `return { + sessions: sql("SELECT COUNT(*) c FROM sessions WHERE source='codex'")[0].c, + mc: sql("SELECT message_count FROM sessions WHERE source='codex'")[0]?.message_count ?? null, + msgs: sql("SELECT COUNT(*) c FROM messages WHERE source='codex'")[0].c, + hits: search('followup', { source: 'codex', limit: 5 }).length, + };`); + const r = runRuntime(['--query', join(home, 'q.mjs')], home); + assert.equal(r.status, 0, r.stderr || r.stdout); + return JSON.parse(r.stdout); +} + +test('codex full build then incremental rebuild replaces the total count without duplicates', () => { + const home = mkdtempSync(join(tmpdir(), 'obelisk-codex-idx-')); + const dir = join(home, '.codex', 'sessions', '2026', '06', '15'); + mkdirSync(dir, { recursive: true }); + const jsonl = join(dir, `rollout-2026-06-15T10-00-00-${ID}.jsonl`); + + // Full build: one user + one agent message. + writeFileSync(jsonl, [metaLine(), evt('user_message', 'codex hello', '2026-06-15T10:00:01Z'), evt('agent_message', 'codex reply', '2026-06-15T10:00:02Z')].join('\n') + '\n'); + assert.equal(runRuntime(['--build'], home).status, 0); + + let c = codexCounts(home); + assert.equal(c.sessions, 1, 'one codex session indexed'); + assert.equal(c.mc, 2, 'two messages counted'); + assert.equal(c.msgs, 2); + + // Append a third message; bump mtime; incremental rebuild (full-reparse). + appendFileSync(jsonl, evt('user_message', 'codex followup', '2026-06-15T10:01:00Z') + '\n'); + const t = statSync(jsonl).mtimeMs / 1000 + 10; + utimesSync(jsonl, t, t); + clearDebounce(home); + + c = codexCounts(home); + // 'total' replace: 3, not 5 (2+3) and not a stale 2. + assert.equal(c.mc, 3, 'message_count replaced with the new total'); + assert.equal(c.msgs, 3, 'exactly three messages, upserted (no duplicates)'); + assert.equal(c.hits, 1, 'the appended message is searchable'); +}); diff --git a/tests/codex-parse.test.mjs b/tests/codex-parse.test.mjs new file mode 100644 index 0000000..2a2ce91 --- /dev/null +++ b/tests/codex-parse.test.mjs @@ -0,0 +1,86 @@ +// Phase 5c-2 golden test: pins the codex adapter's parse() record stream. +// Binding-independent (no database). Covers the event_msg↔response_item dedup, +// tool call/result, token patching, turn-duration, the 'total' session count, +// and guardian-thread → delete-session. + +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { parse } from '../scripts/providers/codex.ts'; + +function writeFixture(lines) { + const dir = mkdtempSync(join(tmpdir(), 'obelisk-codex-parse-')); + const path = join(dir, 'rollout.jsonl'); + writeFileSync(path, lines.map(l => JSON.stringify(l)).join('\n') + '\n'); + return path; +} + +function drain(gen) { + const values = []; + let step = gen.next(); + while (!step.done) { values.push(step.value); step = gen.next(); } + return { values, ret: step.value }; +} + +const META = { id: '019e8951-3e7d-7343-a3e3-05bff48a317d', cwd: '/proj', git: { branch: 'main' }, cli_version: '1.2', timestamp: '2026-06-10T10:00:00Z' }; + +test('codex parse() yields a deduped, tool-aware record stream with a total session', () => { + const path = writeFixture([ + { type: 'session_meta', timestamp: '2026-06-10T10:00:00Z', payload: META }, + { type: 'event_msg', timestamp: '2026-06-10T10:00:01Z', payload: { type: 'user_message', message: 'hello codex' } }, + { type: 'event_msg', timestamp: '2026-06-10T10:00:02Z', payload: { type: 'agent_message', message: 'hi there' } }, + // Duplicate of the agent_message above — must be deduped (dropped). + { type: 'response_item', timestamp: '2026-06-10T10:00:02Z', payload: { type: 'message', role: 'assistant', content: [{ type: 'output_text', text: 'hi there' }] } }, + { 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: 'file listing' } }, + { type: 'event_msg', payload: { type: 'token_count', info: { last_token_usage: { input_tokens: 100, output_tokens: 50 } } } }, + { type: 'event_msg', timestamp: '2026-06-10T10:00:05Z', payload: { type: 'task_complete', duration_ms: 1500 } }, + ]); + + const { values } = drain(parse({ key: path, sessionId: '' }, null)); + const byKind = k => values.filter(r => r.kind === k); + + // Three messages: user, assistant text, assistant tool_use. The duplicate + // response_item 'hi there' was deduped. + const msgs = byKind('message'); + assert.equal(msgs.length, 3); + assert.equal(msgs.filter(m => m.text === 'hi there').length, 1, 'agent_message deduped against response_item'); + assert.equal(msgs.every(m => m.source === 'codex'), true); + + // token_count patched the last text-assistant message's tokens. + const textAssistant = msgs.find(m => m.role === 'assistant' && m.content_type === 'text'); + assert.equal(textAssistant.input_tokens, 100); + 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.equal(byKind('tool_result').length, 1); + assert.equal(byKind('tool_result')[0].tool_use_id, 'codex:call_1'); + + // task_complete → turn duration on the text-assistant message. + assert.deepEqual(byKind('message-turn-duration').map(d => d.turn_duration_ms), [1500]); + + // One session record, full-reparse semantics. + const sessions = byKind('session'); + assert.equal(sessions.length, 1); + assert.equal(sessions[0].source, 'codex'); + assert.equal(sessions[0].countMode, 'total'); + assert.equal(sessions[0].message_count, 3); + assert.equal(sessions[0].git_branch, 'main'); +}); + +test('codex parse() retracts a guardian thread via delete-session and emits nothing else', () => { + const path = writeFixture([ + { type: 'session_meta', timestamp: '2026-06-10T10:00:00Z', payload: { ...META, source: { subagent: { other: 'guardian' } } } }, + { type: 'event_msg', timestamp: '2026-06-10T10:00:01Z', payload: { type: 'user_message', message: 'ignored' } }, + ]); + + const { values } = drain(parse({ key: path, sessionId: '' }, null)); + + assert.equal(values.length, 1); + assert.equal(values[0].kind, 'delete-session'); + assert.match(values[0].sessionId, /^codex:/); +}); diff --git a/tests/indexer-upsert-drift.test.mjs b/tests/indexer-upsert-drift.test.mjs index a214314..dc31b58 100644 --- a/tests/indexer-upsert-drift.test.mjs +++ b/tests/indexer-upsert-drift.test.mjs @@ -1,64 +1,54 @@ -// Regression test for the indexer silent-drift fix. -// -// scripts/indexer.mjs and app/indexer.js had diverged in indexJsonl's message -// write: scripts used INSERT OR REPLACE (churns rowid → FTS churn) and always -// carried the previous message_count forward (inflating it on a full re-scan), -// while app used ON CONFLICT DO UPDATE and reset the count when skip===0. app's -// semantics are canonical; this pins them so the two cannot drift again and so -// the Phase 5 provider-adapter merge inherits one known-correct behavior. +// Regression test for the message write semantics (formerly the indexJsonl +// INSERT-OR-REPLACE vs ON-CONFLICT drift; now enforced through the shared +// persist layer). Re-indexing a claude session must upsert messages (stable +// rowid, no FTS churn) and, because claude parses fresh from an empty cursor +// (countMode 'total'), must replace message_count rather than accumulate. import { test } from 'node:test'; import assert from 'node:assert/strict'; import { createRequire } from 'node:module'; -import { mkdtempSync, writeFileSync } from 'node:fs'; +import { mkdtempSync, writeFileSync, readFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { indexJsonl } from '../scripts/indexer.mjs'; +import { parse } from '../scripts/providers/claude.ts'; +import { persist } from '../scripts/persist.ts'; const require = createRequire(import.meta.url); const { DatabaseSync } = require('node:sqlite'); -const SCHEMA = require('node:fs').readFileSync(new URL('../scripts/schema.sql', import.meta.url), 'utf8'); +const SCHEMA = readFileSync(new URL('../scripts/schema.sql', import.meta.url), 'utf8'); -function writeSessionJsonl() { +function fixtureUnit() { const dir = mkdtempSync(join(tmpdir(), 'obelisk-drift-')); - const jsonlPath = join(dir, 'sid-drift.jsonl'); + const jsonlPath = join(dir, 'sess.jsonl'); const lines = [ { uuid: 'u-1', type: 'user', timestamp: '2026-06-10T10:00:00Z', cwd: '/tmp/proj', message: { role: 'user', content: 'first question' } }, { uuid: 'a-1', type: 'assistant', timestamp: '2026-06-10T10:00:05Z', message: { role: 'assistant', model: 'claude-opus', content: 'first answer' } }, { uuid: 'u-2', type: 'user', timestamp: '2026-06-10T10:00:10Z', cwd: '/tmp/proj', message: { role: 'user', content: 'second question' } }, ]; writeFileSync(jsonlPath, lines.map(l => JSON.stringify(l)).join('\n') + '\n'); - return jsonlPath; + return { key: jsonlPath, sessionId: 'sid-drift', project: 'quiet-zero' }; } -test('re-indexing a session upserts messages (stable rowid) and does not inflate message_count', () => { +test('re-indexing upserts messages (stable rowid) and replaces message_count', () => { const db = new DatabaseSync(':memory:'); db.exec(SCHEMA); - const fi = { path: writeSessionJsonl(), sessionId: 'sid-drift', project: 'quiet-zero' }; - - indexJsonl(db, fi); + const unit = fixtureUnit(); + persist(db, unit, parse(unit, null)); const countAfterFirst = db.prepare('SELECT message_count FROM sessions WHERE id=?').get('sid-drift').message_count; const rowidAfterFirst = db.prepare('SELECT rowid FROM messages WHERE uuid=?').get('u-1').rowid; - const totalMessages = db.prepare('SELECT COUNT(*) AS c FROM messages').get().c; - assert.equal(countAfterFirst, 3, 'three user/assistant messages counted'); - assert.equal(totalMessages, 3); + assert.equal(countAfterFirst, 3); + assert.equal(db.prepare('SELECT COUNT(*) AS c FROM messages').get().c, 3); - // Simulate a fresh full re-scan (force / lost index_state): skip resets to 0. - db.prepare('DELETE FROM index_state').run(); - indexJsonl(db, fi); + // Re-index the same session from scratch (fresh parse → countMode 'total'). + persist(db, unit, parse(unit, null)); const countAfterSecond = db.prepare('SELECT message_count FROM sessions WHERE id=?').get('sid-drift').message_count; const rowidAfterSecond = db.prepare('SELECT rowid FROM messages WHERE uuid=?').get('u-1').rowid; - const totalAfterSecond = db.prepare('SELECT COUNT(*) AS c FROM messages').get().c; - - // message_count is reset+recounted, not accumulated (would be 6 under the old bug). - assert.equal(countAfterSecond, 3, 'message_count must not inflate on re-scan'); - // No duplicate rows. - assert.equal(totalAfterSecond, 3); - // Upsert preserves rowid; INSERT OR REPLACE would have churned it. - assert.equal(rowidAfterSecond, rowidAfterFirst, 'upsert must preserve message rowid (no REPLACE churn)'); + assert.equal(countAfterSecond, 3, 'message_count is replaced, not accumulated (would be 6 under the old bug)'); + assert.equal(db.prepare('SELECT COUNT(*) AS c FROM messages').get().c, 3, 'no duplicate rows'); + assert.equal(rowidAfterSecond, rowidAfterFirst, 'upsert preserves rowid (INSERT OR REPLACE would churn it)'); db.close(); });