From 110a87b4a69a9609c1eaf51b33b8348718aa6018 Mon Sep 17 00:00:00 2001 From: "Norman Nunley, Jr" Date: Wed, 13 May 2026 17:10:00 -0400 Subject: [PATCH 1/2] fix: re-ingest changed session files and invalidate stale journal entries --- src/db.ts | 10 +++++ src/index.ts | 7 ++- src/ingest.test.ts | 78 +++++++++++++++++++++++++++++++++- src/ingest.ts | 103 +++++++++++++++++++++++++++++++++++++++------ 4 files changed, 181 insertions(+), 17 deletions(-) diff --git a/src/db.ts b/src/db.ts index 59cf29c..f3d175e 100644 --- a/src/db.ts +++ b/src/db.ts @@ -79,6 +79,16 @@ export function initDb(dbPath: string): Database { } catch { // Column already exists — ignore } + try { + db.exec(`ALTER TABLE sessions ADD COLUMN source_size INTEGER`); + } catch { + // Column already exists — ignore + } + try { + db.exec(`ALTER TABLE sessions ADD COLUMN source_mtime TEXT`); + } catch { + // Column already exists — ignore + } _db = db; return db; diff --git a/src/index.ts b/src/index.ts index f917903..43d3c55 100755 --- a/src/index.ts +++ b/src/index.ts @@ -43,8 +43,13 @@ switch (command) { const result = ingestSessions(files, db, force); console.log( - `Ingested: ${result.ingested}, Skipped: ${result.skipped}, Errors: ${result.errors.length}` + `Ingested: ${result.ingested}, Re-ingested: ${result.reingested}, Skipped: ${result.skipped}, Errors: ${result.errors.length}` ); + if (result.invalidatedJournals > 0) { + console.log( + `Invalidated ${result.invalidatedJournals} stale journal entr${result.invalidatedJournals === 1 ? "y" : "ies"} — re-run \`summarize\` to regenerate.` + ); + } if (result.errors.length > 0) { for (const err of result.errors.slice(0, 10)) { console.error(` ${err}`); diff --git a/src/ingest.test.ts b/src/ingest.test.ts index c4f6ee9..07f7634 100644 --- a/src/ingest.test.ts +++ b/src/ingest.test.ts @@ -1,7 +1,7 @@ import { describe, test, expect, beforeEach, afterEach } from "bun:test"; import { scanSources, ingestSessions } from "./ingest"; import { initDb, closeDb } from "./db"; -import { mkdtempSync, rmSync, mkdirSync, writeFileSync, copyFileSync } from "fs"; +import { mkdtempSync, rmSync, mkdirSync, writeFileSync, copyFileSync, appendFileSync, utimesSync } from "fs"; import { join } from "path"; import { tmpdir } from "os"; @@ -112,7 +112,8 @@ describe("ingestSessions", () => { expect(first.errors.length).toBe(0); const second = ingestSessions([sessionFile], db, true); - expect(second.ingested).toBe(1); + expect(second.ingested).toBe(0); + expect(second.reingested).toBe(1); expect(second.skipped).toBe(0); expect(second.errors.length).toBe(0); @@ -170,4 +171,77 @@ describe("ingestSessions", () => { expect(session?.version).toBe("0.99.0-alpha.23"); expect(session?.message_count).toBe(2); }); + + test("re-ingests when source file has grown (claude/codex appends to existing session)", () => { + const fixturePath = join(import.meta.dir, "../tests/fixtures/test-session-1.jsonl"); + const projectDir = join(tempDir, "-Users-test-myapp"); + mkdirSync(projectDir, { recursive: true }); + const sessionFile = join(projectDir, "test-session-1.jsonl"); + copyFileSync(fixturePath, sessionFile); + + const first = ingestSessions([sessionFile], db); + expect(first.ingested).toBe(1); + const firstSnapshot = db.query("SELECT source_size, source_mtime FROM sessions").get() as { source_size: number; source_mtime: string }; + expect(firstSnapshot.source_size).toBeGreaterThan(0); + + // Append to simulate Claude/Codex continuing the session + const originalRaw = require("fs").readFileSync(fixturePath, "utf-8"); + appendFileSync(sessionFile, originalRaw); // doubles the content + + // Bump mtime to ensure detection even if the OS clock granularity coalesces the writes + const future = new Date(Date.now() + 60_000); + utimesSync(sessionFile, future, future); + + const second = ingestSessions([sessionFile], db); + expect(second.ingested).toBe(0); + expect(second.reingested).toBe(1); + expect(second.skipped).toBe(0); + + const secondSnapshot = db.query("SELECT source_size FROM sessions").get() as { source_size: number }; + expect(secondSnapshot.source_size).toBeGreaterThan(firstSnapshot.source_size); + }); + + test("skips unchanged files on repeated ingest (size and mtime both match)", () => { + const fixturePath = join(import.meta.dir, "../tests/fixtures/test-session-1.jsonl"); + const projectDir = join(tempDir, "-Users-test-myapp"); + mkdirSync(projectDir, { recursive: true }); + const sessionFile = join(projectDir, "test-session-1.jsonl"); + copyFileSync(fixturePath, sessionFile); + + ingestSessions([sessionFile], db); + const result = ingestSessions([sessionFile], db); + expect(result.ingested).toBe(0); + expect(result.reingested).toBe(0); + expect(result.skipped).toBe(1); + }); + + test("re-ingesting invalidates journal entries for the affected (date, project)", () => { + const fixturePath = join(import.meta.dir, "../tests/fixtures/test-session-1.jsonl"); + const projectDir = join(tempDir, "-Users-test-myapp"); + mkdirSync(projectDir, { recursive: true }); + const sessionFile = join(projectDir, "test-session-1.jsonl"); + copyFileSync(fixturePath, sessionFile); + + ingestSessions([sessionFile], db); + const session = db.query("SELECT id, project_id, date(started_at) as date FROM sessions").get() as { id: string; project_id: string; date: string }; + + // Plant a journal entry for that (date, project_id) + db.exec(` + INSERT INTO journal_entries (date, project_id, session_ids, headline, summary, generated_at, model_used) + VALUES ('${session.date}', '${session.project_id}', '["${session.id}"]', 'cached', 'cached summary', datetime('now'), 'test') + `); + const beforeCount = (db.query("SELECT COUNT(*) as n FROM journal_entries").get() as { n: number }).n; + expect(beforeCount).toBe(1); + + // Modify the file and re-ingest + appendFileSync(sessionFile, require("fs").readFileSync(fixturePath, "utf-8")); + utimesSync(sessionFile, new Date(Date.now() + 60_000), new Date(Date.now() + 60_000)); + + const result = ingestSessions([sessionFile], db); + expect(result.reingested).toBe(1); + expect(result.invalidatedJournals).toBe(1); + + const afterCount = (db.query("SELECT COUNT(*) as n FROM journal_entries").get() as { n: number }).n; + expect(afterCount).toBe(0); + }); }); diff --git a/src/ingest.ts b/src/ingest.ts index 0a310d3..8ca3add 100644 --- a/src/ingest.ts +++ b/src/ingest.ts @@ -56,17 +56,45 @@ export function scanSources( return files; } -/** Ingest session files into the database */ +type FileStatus = { + existingSessionId: string | null; + storedSize: number | null; + storedMtime: string | null; + recordedDate: string | null; + recordedProjectId: string | null; +}; + +/** Ingest session files into the database. + * Tracks each file's mtime+size so files that have grown (Claude/Codex append + * to their session JSONL during use) are detected and re-ingested. When a + * session is re-ingested, journal entries covering its (date, project) are + * deleted so the next summarize run regenerates them from updated content. */ export function ingestSessions( files: string[], db: Database, force = false -): { ingested: number; skipped: number; errors: string[] } { +): { + ingested: number; + reingested: number; + skipped: number; + invalidatedJournals: number; + errors: string[]; +} { let ingested = 0; + let reingested = 0; let skipped = 0; + let invalidatedJournals = 0; const errors: string[] = []; - const checkStmt = db.query("SELECT id FROM sessions WHERE source_path = ?"); + const checkStmt = db.query(` + SELECT + id as existingSessionId, + source_size as storedSize, + source_mtime as storedMtime, + date(started_at) as recordedDate, + project_id as recordedProjectId + FROM sessions WHERE source_path = ? + `); const checkSessionId = db.query("SELECT id FROM sessions WHERE id = ?"); const insertProject = db.prepare(` INSERT INTO projects (id, path, display_name, session_count) @@ -74,8 +102,8 @@ export function ingestSessions( ON CONFLICT(id) DO UPDATE SET path = excluded.path `); const insertSession = db.prepare(` - INSERT INTO sessions (id, parent_session_id, project_id, project_path, source_path, started_at, ended_at, git_branch, version, message_count, is_subagent, ingested_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now')) + INSERT INTO sessions (id, parent_session_id, project_id, project_path, source_path, started_at, ended_at, git_branch, version, message_count, is_subagent, source_size, source_mtime, ingested_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now')) `); const insertConvo = db.prepare(` INSERT INTO conversations (session_id, conversation_markdown, extracted_at) @@ -83,14 +111,34 @@ export function ingestSessions( `); const deleteConvo = db.prepare(`DELETE FROM conversations WHERE session_id = ?`); const deleteSession = db.prepare(`DELETE FROM sessions WHERE id = ?`); + const deleteJournal = db.prepare( + `DELETE FROM journal_entries WHERE date = ? AND project_id = ?` + ); for (const file of files) { - if (!force) { - const existing = checkStmt.get(file); - if (existing) { + let stat; + try { + stat = statSync(file); + } catch (err) { + errors.push(`${file}: stat failed: ${err}`); + continue; + } + const currentSize = stat.size; + const currentMtime = stat.mtime.toISOString(); + + const existing = checkStmt.get(file); + const isReingest = existing !== null && !force; + + if (existing && !force) { + // Skip unchanged files (both mtime and size match what we stored). + if ( + existing.storedSize === currentSize && + existing.storedMtime === currentMtime + ) { skipped++; continue; } + // Otherwise fall through and re-ingest. } try { @@ -101,8 +149,8 @@ export function ingestSessions( continue; } - // Skip if session ID already exists (e.g., same session in multiple project dirs) - if (!force) { + // Skip if a different file already owns this session ID. + if (!force && !existing) { const existingById = checkSessionId.get(session.sessionId); if (existingById) { skipped++; @@ -111,12 +159,33 @@ export function ingestSessions( } const projectId = session.projectName; + const newDate = session.startedAt.slice(0, 10); db.transaction(() => { - if (force) { + // For both --force and detected-change re-ingest: drop old rows first. + if (force || existing) { deleteConvo.run(session.sessionId); deleteSession.run(session.sessionId); } + + // Invalidate any journal entries this session contributed to. + // For a session whose date or project_id changed since last ingest, + // both the old and new (date, project_id) tuples must be invalidated. + const tuplesToInvalidate = new Set(); + if (existing?.recordedDate && existing?.recordedProjectId) { + tuplesToInvalidate.add( + `${existing.recordedDate}|${existing.recordedProjectId}` + ); + } + if (isReingest || force) { + tuplesToInvalidate.add(`${newDate}|${projectId}`); + } + for (const tuple of tuplesToInvalidate) { + const [date, pid] = tuple.split("|", 2) as [string, string]; + const result = deleteJournal.run(date, pid); + invalidatedJournals += result.changes; + } + insertProject.run( projectId, session.projectPath, @@ -134,12 +203,18 @@ export function ingestSessions( session.gitBranch, session.version, session.messageCount, - isSubagent + isSubagent, + currentSize, + currentMtime ); insertConvo.run(session.sessionId, session.toMarkdown()); })(); - ingested++; + if (isReingest || (force && existing)) { + reingested++; + } else { + ingested++; + } } catch (err) { errors.push(`${file}: ${err}`); } @@ -153,5 +228,5 @@ export function ingestSessions( session_count = (SELECT COUNT(*) FROM sessions WHERE sessions.project_id = projects.id) `); - return { ingested, skipped, errors }; + return { ingested, reingested, skipped, invalidatedJournals, errors }; } From 8a851170388d72c6f20ba619ac1d0b0a7c2f9f54 Mon Sep 17 00:00:00 2001 From: "Norman Nunley, Jr" Date: Wed, 13 May 2026 21:22:09 -0400 Subject: [PATCH 2/2] fix: keep cross-midnight sessions whole under start_at date MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously, sessions whose messages spanned midnight were split across two date buckets using per-message logical dates. In practice this created thin tail slivers on the next day — a few morning messages from a session that started the prior evening — that the LLM consistently rejected as "no engineering work" because the slivers lacked context. Skip-pattern analysis (2026-05-13, 8 recent skips) found that all skipped groups had tiny day-scoped content (<2KB; smallest 389 chars literal `/exit` boilerplate). The most recoverable false-negatives were these midnight-split slivers where substantive work happened on the start side of midnight but the tail got rejected in isolation. Fix: each session is now atomic and attributed to its `started_at` logical date (the date adjusted by `day_start_hour`). A session that begins late one night and continues past midnight contributes its full content to the start date. `splitConversationByDay` is kept as an exported helper for external tools that want per-message day attribution, but the orchestrator no longer uses it. Tests updated: the midnight-spanning suite now verifies atomic attribution (one group, all 6 messages on Feb 20) and that `filterDate("2026-02-21")` returns nothing for a session that started on Feb 20 even when some messages are timestamped Feb 21. --- src/summarize.test.ts | 51 +++++++++++++-------------- src/summarize.ts | 81 ++++++++++++++++++------------------------- 2 files changed, 58 insertions(+), 74 deletions(-) diff --git a/src/summarize.test.ts b/src/summarize.test.ts index 492dd90..7c3b796 100644 --- a/src/summarize.test.ts +++ b/src/summarize.test.ts @@ -216,43 +216,42 @@ describe("groupSessionsByDateAndProject - midnight spanning", () => { rmSync(tempDir, { recursive: true, force: true }); }); - test("splits a midnight-spanning session into two date groups", () => { + test("attributes the entire midnight-spanning session to its start date", () => { + // After the cross-midnight fix: sessions are atomic units attributed + // to their started_at logical date. The session begun at 22:00 on + // Feb 20 is wholly attributed to Feb 20, including the morning + // messages at 06:00/06:05 on Feb 21. const groups = groupSessionsByDateAndProject(db); - expect(groups.length).toBe(2); - - const sorted = groups.sort((a, b) => a.date.localeCompare(b.date)); - - // Feb 20: messages at 22:00, 22:10, 01:30, 01:35 (all logical date Feb 20) - expect(sorted[0]!.date).toBe("2026-02-20"); - expect(sorted[0]!.projectId).toBe("myapp"); - expect(sorted[0]!.sessionIds).toEqual(["s-midnight"]); - expect(sorted[0]!.conversations[0]).toContain("Late night refactor"); - expect(sorted[0]!.conversations[0]).toContain("Still at it"); - - // Feb 21: messages at 06:00, 06:05 (logical date Feb 21) - expect(sorted[1]!.date).toBe("2026-02-21"); - expect(sorted[1]!.projectId).toBe("myapp"); - expect(sorted[1]!.sessionIds).toEqual(["s-midnight"]); - expect(sorted[1]!.conversations[0]).toContain("Morning review"); + expect(groups.length).toBe(1); + expect(groups[0]!.date).toBe("2026-02-20"); + expect(groups[0]!.sessionIds).toEqual(["s-midnight"]); + // The full conversation (all 6 messages) should be present. + expect(groups[0]!.conversations[0]).toContain("Late night refactor"); + expect(groups[0]!.conversations[0]).toContain("Still at it"); + expect(groups[0]!.conversations[0]).toContain("Morning review"); }); test("filters out already-summarized date+project combos", () => { - // Insert a journal entry for Feb 20 + // Insert a journal entry for Feb 20 (the session's start date) db.exec(` INSERT INTO journal_entries (date, project_id, session_ids, headline, summary, topics, generated_at, model_used) VALUES ('2026-02-20', 'myapp', '["s-midnight"]', 'Test', 'Test summary', '[]', datetime('now'), 'test-model'); `); const groups = groupSessionsByDateAndProject(db); - // Only Feb 21 should remain - expect(groups.length).toBe(1); - expect(groups[0]!.date).toBe("2026-02-21"); + // Now nothing remains — the whole session was attributed to Feb 20 + // which is summarized. + expect(groups.length).toBe(0); }); - test("filterDate works with logical dates", () => { - const groups = groupSessionsByDateAndProject(db, "2026-02-21"); - expect(groups.length).toBe(1); - expect(groups[0]!.date).toBe("2026-02-21"); - expect(groups[0]!.conversations[0]).toContain("Morning review"); + test("filterDate matches start date", () => { + // Asking for Feb 21 should NOT return the session that started on Feb 20 + // even though some of its messages are timestamped Feb 21. + const groupsFeb21 = groupSessionsByDateAndProject(db, "2026-02-21"); + expect(groupsFeb21.length).toBe(0); + + const groupsFeb20 = groupSessionsByDateAndProject(db, "2026-02-20"); + expect(groupsFeb20.length).toBe(1); + expect(groupsFeb20[0]!.conversations[0]).toContain("Morning review"); }); }); diff --git a/src/summarize.ts b/src/summarize.ts index 38daa79..65f3d2b 100644 --- a/src/summarize.ts +++ b/src/summarize.ts @@ -125,8 +125,18 @@ export function splitConversationByDay( } /** Group unsummarized sessions by date and project. - * Splits conversations by logical day boundary so a session spanning midnight - * contributes messages to the correct date's journal entry. */ + * + * Each session is treated as an atomic unit and attributed to its + * `started_at` logical date. A session that begins late one night and + * continues past midnight contributes its full content to the start date, + * not split across days. + * + * Earlier versions split sessions at midnight using per-message timestamps, + * but in practice that produced thin tail slivers on the next day (a few + * morning messages) that the LLM consistently rejected as "no engineering + * work" — losing the connection to the substantive work that came earlier. + * Keeping sessions whole gives the journal entry for the start date access + * to the entire arc. */ export function groupSessionsByDateAndProject( db: Database, filterDate?: string, @@ -171,56 +181,31 @@ export function groupSessionsByDateAndProject( summarizedRows.map((r) => `${r.date}|${r.project_id}`) ); - // Group by (logical-date, project) + // Group by (logical-start-date, project). Each session is attributed + // atomically to its started_at logical date — we no longer split a + // session's content across days even if its messages span midnight. + // The session's started_at column is what `row.date` already holds (it's + // `date(s.started_at)` from the SELECT), so we only need to apply the + // dayStartHour cutoff for late-night starts. const groups = new Map(); for (const row of rows) { - const dayChunks = splitConversationByDay( - row.conversation_markdown, - dayStartHour - ); - - if (dayChunks) { - // New-format timestamps: split across logical dates - for (const [day, chunk] of dayChunks) { - const key = `${day}|${row.project_id}`; - if (!groups.has(key)) { - groups.set(key, { - date: day, - projectId: row.project_id, - projectName: row.display_name, - sessionIds: [], - conversations: [], - }); - } - const group = groups.get(key)!; - if (!group.sessionIds.includes(row.session_id)) { - group.sessionIds.push(row.session_id); - } - group.conversations.push(chunk); - } - } else { - // Old-format timestamps: fall back to session's started_at date - const fallbackDay = logicalDate( - row.date + " 12:00", - dayStartHour - ); - const key = `${fallbackDay}|${row.project_id}`; - if (!groups.has(key)) { - groups.set(key, { - date: fallbackDay, - projectId: row.project_id, - projectName: row.display_name, - sessionIds: [], - conversations: [], - }); - } - const group = groups.get(key)!; - if (!group.sessionIds.includes(row.session_id)) { - group.sessionIds.push(row.session_id); - } - group.conversations.push(row.conversation_markdown); + const sessionDate = logicalDate(row.date + " 12:00", dayStartHour); + const key = `${sessionDate}|${row.project_id}`; + if (!groups.has(key)) { + groups.set(key, { + date: sessionDate, + projectId: row.project_id, + projectName: row.display_name, + sessionIds: [], + conversations: [], + }); + } + const group = groups.get(key)!; + if (!group.sessionIds.includes(row.session_id)) { + group.sessionIds.push(row.session_id); } + group.conversations.push(row.conversation_markdown); } // Filter out already-summarized combos