From 1d652d823fd44c284e0ee41a7fde00640eb9ad1b Mon Sep 17 00:00:00 2001 From: tommy0103 Date: Wed, 8 Jul 2026 19:01:54 +0800 Subject: [PATCH] feat(persist): add shared record-stream persist layer (Phase 5b-2a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce scripts/persist.ts — the single, provider- and binding-agnostic layer that consumes an adapter's IndexRecord stream and writes rows into an injected SQLite handle (node:sqlite for skill/CLI, better-sqlite3 for the app). It is the only layer that touches the database. Write semantics are the canonical ones reconciled from the earlier drift: - messages upsert via ON CONFLICT (turn_duration_ms not in the column list, so it is never clobbered) - sessions merge with the existing row: started_at MIN, ended_at MAX, message_count reset-or-accumulate by resume state, fill-if-null for the rest; project_path is preserved and left to refreshSessionProjectPaths - message-turn-duration applies as a targeted UPDATE - delete-session cascades across all tables - the generator's return cursor is written back to index_state (mtime:lines → the two existing columns; no schema migration yet) Purely additive — buildIndex still uses the old indexJsonl path, so existing behavior is unchanged. Rewiring happens in 5b-2b. Adds tests/persist.test.mjs: all record kinds written, resume does not double-count message_count, fresh re-scan resets it, delete-session cascades. Full suite 115/115, lint + typecheck green. --- scripts/persist.ts | 112 +++++++++++++++++++++++++++++++++++++++++ tests/persist.test.mjs | 104 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 216 insertions(+) create mode 100644 scripts/persist.ts create mode 100644 tests/persist.test.mjs diff --git a/scripts/persist.ts b/scripts/persist.ts new file mode 100644 index 0000000..10f5e3a --- /dev/null +++ b/scripts/persist.ts @@ -0,0 +1,112 @@ +// Shared persist layer (see docs/adr/0001). +// +// Provider-agnostic and binding-agnostic: it consumes the IndexRecord stream +// from any adapter's parse() and writes rows into the injected database handle +// (node:sqlite for the skill/CLI, better-sqlite3 for the app — they share the +// prepare/run/get API). It is the ONLY layer that touches the database and the +// only place that knows the schema. Adapters stay pure. +// +// Write semantics are the canonical ones reconciled from the drift: messages +// upsert via ON CONFLICT; sessions merge with any existing row (started_at MIN, +// ended_at MAX, message_count reset-or-accumulate, fill-if-null for the rest); +// 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'; + +const minStr = (a: string | null, b: string | null) => (a == null ? b : b == null ? a : a < b ? a : b); +const maxStr = (a: string | null, b: string | null) => (a == null ? b : b == null ? a : a > b ? a : b); + +function statements(db: any) { + 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 (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + 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 (?,?,?,?,?)'), + 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 (?,?,?,?,?,?,?,?,?,?,?)'), + 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=?'), + }; +} + +// Cascade-delete every row belonging to a session/thread (guardian retraction). +function deleteSession(db: any, sessionId: string) { + db.prepare('DELETE FROM tool_results 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 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 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: 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) { + 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); + break; + case 'tool_call': + st.tc.run(r.id, r.message_uuid, r.session_id, r.name, 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); + break; + case 'summary': + st.sum.run(r.id, r.session_id, r.timestamp, r.source, r.content); + break; + case 'message-turn-duration': + st.turn.run(r.turn_duration_ms, r.uuid); + break; + case 'session': { + const prev = st.getSession.get(r.id); + st.ses.run( + r.id, + r.title ?? prev?.title ?? null, + r.project ?? prev?.project ?? null, + prev?.project_path ?? null, // authoritative project_path is set by refreshSessionProjectPaths + minStr(prev?.started_at ?? null, r.started_at), + maxStr(prev?.ended_at ?? null, r.ended_at), + r.git_branch ?? prev?.git_branch ?? null, + r.version ?? prev?.version ?? null, + resuming ? (prev?.message_count || 0) + r.message_count : r.message_count, + r.jsonl_path, + r.source, + ); + break; + } + case 'delete-session': + deleteSession(db, r.sessionId); + break; + default: + throw new Error(`persist: unhandled record kind ${(r as { kind: string }).kind}`); + } + }; + + let step = gen.next(); + while (!step.done) { write(step.value); step = gen.next(); } + const cursor = step.value; + + if (cursor != null) { + const [mtime, lines] = cursor.split(':'); + st.idx.run(unit.key, Number(mtime), Number(lines)); + } + return cursor; +} diff --git a/tests/persist.test.mjs b/tests/persist.test.mjs new file mode 100644 index 0000000..07dfd58 --- /dev/null +++ b/tests/persist.test.mjs @@ -0,0 +1,104 @@ +// Phase 5b-2a: unit-tests the shared persist layer in isolation (no buildIndex). +// Feeds claude.parse output into persist against an in-memory node:sqlite db and +// asserts rows + the drift-fixed session merge semantics. + +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createRequire } from 'node:module'; +import { mkdtempSync, writeFileSync, readFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +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 = readFileSync(new URL('../scripts/schema.sql', import.meta.url), 'utf8'); + +function fixtureUnit() { + const dir = mkdtempSync(join(tmpdir(), 'obelisk-persist-')); + const path = join(dir, 'sid-p.jsonl'); + const lines = [ + { type: 'ai-title', aiTitle: 'Persist Session' }, + { uuid: 'u1', type: 'user', timestamp: '2026-06-10T10:00:00Z', cwd: '/proj', message: { role: 'user', content: 'hi' } }, + { uuid: 'a1', type: 'assistant', timestamp: '2026-06-10T10:00:05Z', message: { role: 'assistant', model: 'm', content: [{ type: 'tool_use', id: 'tc1', name: 'Read', input: { file_path: '/f' } }] } }, + { type: 'system', subtype: 'turn_duration', parentUuid: 'a1', durationMs: 999 }, + { uuid: 'u2', type: 'user', timestamp: '2026-06-10T10:00:10Z', message: { role: 'user', content: [{ type: 'tool_result', tool_use_id: 'tc1', content: 'body', is_error: false }] } }, + { type: 'system', subtype: 'away_summary', uuid: 's1', timestamp: '2026-06-10T10:00:11Z', content: 'sum' }, + ]; + writeFileSync(path, lines.map(l => JSON.stringify(l)).join('\n') + '\n'); + return { key: path, sessionId: 'sid-p', project: 'quiet-zero' }; +} + +function freshDb() { + const db = new DatabaseSync(':memory:'); + db.exec(SCHEMA); + return db; +} + +test('persist writes all record kinds from one claude parse', () => { + const db = freshDb(); + const unit = fixtureUnit(); + + const cursor = persist(db, unit, parse(unit, null)); + + assert.equal(db.prepare('SELECT COUNT(*) c FROM messages').get().c, 3); + assert.equal(db.prepare('SELECT COUNT(*) c FROM tool_calls').get().c, 1); + assert.equal(db.prepare('SELECT COUNT(*) c FROM tool_results').get().c, 1); + assert.equal(db.prepare('SELECT COUNT(*) c FROM summaries').get().c, 1); + + const ses = db.prepare('SELECT * FROM sessions WHERE id=?').get('sid-p'); + assert.equal(ses.title, 'Persist Session'); + assert.equal(ses.message_count, 3); + assert.equal(ses.started_at, '2026-06-10T10:00:00Z'); + assert.equal(ses.ended_at, '2026-06-10T10:00:10Z'); + + // turn_duration applied via targeted UPDATE. + assert.equal(db.prepare('SELECT turn_duration_ms FROM messages WHERE uuid=?').get('a1').turn_duration_ms, 999); + + // Cursor persisted into index_state (mtime:lines → two columns). + const state = db.prepare('SELECT lines_processed FROM index_state WHERE jsonl_path=?').get(unit.key); + assert.equal(state.lines_processed, 6); + assert.equal(cursor.split(':')[1], '6'); +}); + +test('resuming from cursor does not double-count message_count', () => { + const db = freshDb(); + const unit = fixtureUnit(); + + const c1 = persist(db, unit, parse(unit, null)); + assert.equal(db.prepare('SELECT message_count m FROM sessions WHERE id=?').get('sid-p').m, 3); + + // Second run resumes at the stored cursor: parse skips all lines, yields an + // empty-chunk session (count 0); persist accumulates 3 + 0 = 3, not 6. + persist(db, unit, parse(unit, c1)); + assert.equal(db.prepare('SELECT message_count m FROM sessions WHERE id=?').get('sid-p').m, 3); + assert.equal(db.prepare('SELECT COUNT(*) c FROM messages').get().c, 3); +}); + +test('fresh full re-scan (no prior cursor) resets message_count instead of accumulating', () => { + const db = freshDb(); + const unit = fixtureUnit(); + + persist(db, unit, parse(unit, null)); + db.prepare('DELETE FROM index_state').run(); // simulate force / lost state + persist(db, unit, parse(unit, null)); + + assert.equal(db.prepare('SELECT message_count m FROM sessions WHERE id=?').get('sid-p').m, 3); + assert.equal(db.prepare('SELECT COUNT(*) c FROM messages').get().c, 3); +}); + +test('delete-session cascades across tables', () => { + const db = freshDb(); + const unit = fixtureUnit(); + persist(db, unit, parse(unit, null)); + + // Hand-roll a one-shot generator emitting a delete for the session. + function* del() { yield { kind: 'delete-session', sessionId: 'sid-p' }; return null; } + persist(db, { key: 'x', sessionId: 'sid-p' }, del()); + + 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); +});