feat(persist): add shared record-stream persist layer (Phase 5b-2a)

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.
2026-07-08 19:01:54 +08:00
parent 2b30d9596d
commit 1d652d823f
2 changed files with 216 additions and 0 deletions
@@ -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<IndexRecord, Cursor>): 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;
+}
@@ -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);
+});