diff --git a/dist/indexer.js b/dist/indexer.js index 8afa1d35..9f8ab1d3 100644 --- a/dist/indexer.js +++ b/dist/indexer.js @@ -7,6 +7,7 @@ import { summarizeConversation } from './summarizer.js'; import { getArchiveDir, getExcludedProjects, getConversationSourceDirs, findJsonlFiles, statIfExists } from './paths.js'; import { formatErrorSentinel, shouldQueueForSummary } from './summary-sentinel.js'; import { getMaxMessageBytes, isOversizeExchange } from './message-size.js'; +import { copyIfNewer } from './sync.js'; // Set max output tokens for Claude SDK (used by summarizer) process.env.CLAUDE_CODE_MAX_OUTPUT_TOKENS = '20000'; // Increase max listeners for concurrent API calls @@ -184,13 +185,11 @@ export async function indexSession(sessionId, concurrency = 1, noSummaries = fal const projectArchive = path.join(ARCHIVE_DIR, project); fs.mkdirSync(projectArchive, { recursive: true }); const archivePath = path.join(projectArchive, file); + const maxIndexedLine = db.prepare('SELECT COALESCE(MAX(line_end), 0) AS maxLine FROM exchanges WHERE archive_path = ?').get(archivePath).maxLine; // Archive + parse — source may vanish mid-run (Claude Code cleanup). let exchanges; try { - if (!fs.existsSync(archivePath)) { - fs.mkdirSync(path.dirname(archivePath), { recursive: true }); - fs.copyFileSync(sourcePath, archivePath); - } + copyIfNewer(sourcePath, archivePath); exchanges = await parseConversation(sourcePath, project, archivePath); } catch (error) { @@ -199,6 +198,7 @@ export async function indexSession(sessionId, concurrency = 1, noSummaries = fal break; } if (exchanges.length > 0) { + const newExchanges = exchanges.filter(exchange => exchange.lineStart > maxIndexedLine); // Generate summary (unless --no-summaries) const summaryPath = archivePath.replace('.jsonl', '-summary.txt'); if (!noSummaries && shouldQueueForSummary(summaryPath)) { @@ -220,7 +220,7 @@ export async function indexSession(sessionId, concurrency = 1, noSummaries = fal // Index const maxMessageBytes = getMaxMessageBytes(); let oversizeSkipped = 0; - for (const exchange of exchanges) { + for (const exchange of newExchanges) { // Skip oversize single messages BEFORE embedding — a foreign // summarizer's pasted transcript is noise, and embedding it is the // expensive waste (#139). @@ -235,7 +235,7 @@ export async function indexSession(sessionId, concurrency = 1, noSummaries = fal if (oversizeSkipped > 0) { console.log(` Skipped ${oversizeSkipped} oversize exchange(s) (> ${maxMessageBytes} bytes; set EPISODIC_MEMORY_MAX_MESSAGE_BYTES to change) — likely embedded-transcript payloads (#139)`); } - console.log(`✅ Indexed session ${sessionId}: ${exchanges.length} exchanges`); + console.log(`✅ Indexed session ${sessionId}: ${newExchanges.length} new exchanges`); } db.close(); break; diff --git a/dist/sync.d.ts b/dist/sync.d.ts index 658e98ad..5ff63eee 100644 --- a/dist/sync.d.ts +++ b/dist/sync.d.ts @@ -39,5 +39,6 @@ export interface SyncOptions { * Claude quota and can stall on a permission prompt. */ export declare function buildSyncOptionsFromEnv(env: NodeJS.ProcessEnv): SyncOptions; +export declare function copyIfNewer(src: string, dest: string): boolean; export declare function extractSessionIdFromPath(filePath: string): string | null; export declare function syncConversations(sourceDir: string, destDir: string, options?: SyncOptions): Promise; diff --git a/dist/sync.js b/dist/sync.js index 0629761d..fbf82e92 100644 --- a/dist/sync.js +++ b/dist/sync.js @@ -122,7 +122,7 @@ function hasConversationContent(filePath) { export function buildSyncOptionsFromEnv(env) { return { skipSummaries: env.EPISODIC_MEMORY_SKIP_SUMMARIES === '1' }; } -function copyIfNewer(src, dest) { +export function copyIfNewer(src, dest) { // Ensure destination directory exists const destDir = path.dirname(dest); if (!fs.existsSync(destDir)) { @@ -132,7 +132,7 @@ function copyIfNewer(src, dest) { if (fs.existsSync(dest)) { const srcStat = fs.statSync(src); const destStat = fs.statSync(dest); - if (destStat.mtimeMs >= srcStat.mtimeMs) { + if (destStat.mtimeMs >= srcStat.mtimeMs && destStat.size === srcStat.size) { return false; // Dest is current, skip } } diff --git a/src/indexer.ts b/src/indexer.ts index 173ef679..26d22e32 100644 --- a/src/indexer.ts +++ b/src/indexer.ts @@ -9,6 +9,7 @@ import { ConversationExchange } from './types.js'; import { getArchiveDir, getExcludedProjects, getConversationSourceDirs, findJsonlFiles, statIfExists } from './paths.js'; import { formatErrorSentinel, shouldQueueForSummary } from './summary-sentinel.js'; import { getMaxMessageBytes, isOversizeExchange } from './message-size.js'; +import { copyIfNewer } from './sync.js'; // Set max output tokens for Claude SDK (used by summarizer) process.env.CLAUDE_CODE_MAX_OUTPUT_TOKENS = '20000'; @@ -242,14 +243,14 @@ export async function indexSession(sessionId: string, concurrency: number = 1, n fs.mkdirSync(projectArchive, { recursive: true }); const archivePath = path.join(projectArchive, file); + const maxIndexedLine = (db.prepare( + 'SELECT COALESCE(MAX(line_end), 0) AS maxLine FROM exchanges WHERE archive_path = ?' + ).get(archivePath) as { maxLine: number }).maxLine; // Archive + parse — source may vanish mid-run (Claude Code cleanup). let exchanges; try { - if (!fs.existsSync(archivePath)) { - fs.mkdirSync(path.dirname(archivePath), { recursive: true }); - fs.copyFileSync(sourcePath, archivePath); - } + copyIfNewer(sourcePath, archivePath); exchanges = await parseConversation(sourcePath, project, archivePath); } catch (error) { console.log(`Skipped ${file} (read failed: ${error instanceof Error ? error.message : error})`); @@ -258,6 +259,7 @@ export async function indexSession(sessionId: string, concurrency: number = 1, n } if (exchanges.length > 0) { + const newExchanges = exchanges.filter(exchange => exchange.lineStart > maxIndexedLine); // Generate summary (unless --no-summaries) const summaryPath = archivePath.replace('.jsonl', '-summary.txt'); if (!noSummaries && shouldQueueForSummary(summaryPath)) { @@ -276,7 +278,7 @@ export async function indexSession(sessionId: string, concurrency: number = 1, n // Index const maxMessageBytes = getMaxMessageBytes(); let oversizeSkipped = 0; - for (const exchange of exchanges) { + for (const exchange of newExchanges) { // Skip oversize single messages BEFORE embedding — a foreign // summarizer's pasted transcript is noise, and embedding it is the // expensive waste (#139). @@ -297,7 +299,7 @@ export async function indexSession(sessionId: string, concurrency: number = 1, n console.log(` Skipped ${oversizeSkipped} oversize exchange(s) (> ${maxMessageBytes} bytes; set EPISODIC_MEMORY_MAX_MESSAGE_BYTES to change) — likely embedded-transcript payloads (#139)`); } - console.log(`✅ Indexed session ${sessionId}: ${exchanges.length} exchanges`); + console.log(`✅ Indexed session ${sessionId}: ${newExchanges.length} new exchanges`); } db.close(); diff --git a/src/sync.ts b/src/sync.ts index d9ee6810..5ae654b2 100644 --- a/src/sync.ts +++ b/src/sync.ts @@ -132,7 +132,7 @@ export function buildSyncOptionsFromEnv(env: NodeJS.ProcessEnv): SyncOptions { return { skipSummaries: env.EPISODIC_MEMORY_SKIP_SUMMARIES === '1' }; } -function copyIfNewer(src: string, dest: string): boolean { +export function copyIfNewer(src: string, dest: string): boolean { // Ensure destination directory exists const destDir = path.dirname(dest); if (!fs.existsSync(destDir)) { @@ -143,7 +143,7 @@ function copyIfNewer(src: string, dest: string): boolean { if (fs.existsSync(dest)) { const srcStat = fs.statSync(src); const destStat = fs.statSync(dest); - if (destStat.mtimeMs >= srcStat.mtimeMs) { + if (destStat.mtimeMs >= srcStat.mtimeMs && destStat.size === srcStat.size) { return false; // Dest is current, skip } } diff --git a/test/incremental-indexing.test.ts b/test/incremental-indexing.test.ts index 797f5b07..7cf06153 100644 --- a/test/incremental-indexing.test.ts +++ b/test/incremental-indexing.test.ts @@ -1,9 +1,10 @@ import { describe, it, expect, beforeEach, afterEach } from 'vitest'; -import { mkdtempSync, mkdirSync, writeFileSync, appendFileSync, rmSync } from 'fs'; +import { mkdtempSync, mkdirSync, writeFileSync, appendFileSync, readFileSync, rmSync, statSync, utimesSync } from 'fs'; import { join } from 'path'; import { tmpdir } from 'os'; import Database from 'better-sqlite3'; -import { indexUnprocessed } from '../src/indexer.js'; +import { indexSession, indexUnprocessed } from '../src/indexer.js'; +import { getArchiveDir } from '../src/paths.js'; import { suppressConsole } from './test-utils.js'; /** @@ -105,4 +106,33 @@ describe('indexer: incremental indexing', () => { await indexUnprocessed(1, true); expect(countExchanges()).toBe(5); }); + + it('does not replace indexed rows when indexing a resumed session', async () => { + const projectDir = join(projectsDir, 'project-a'); + mkdirSync(projectDir, { recursive: true }); + const transcriptPath = join(projectDir, 'session-1.jsonl'); + writeFileSync(transcriptPath, makeExchangeLines(1, 'session-1'), 'utf-8'); + + await indexSession('session-1', 1, true); + const before = new Database(dbPath); + before.prepare('UPDATE exchanges SET last_indexed = ? WHERE line_start = 1').run(123); + before.close(); + + const archivePath = join(getArchiveDir(), 'project-a', 'session-1.jsonl'); + const archiveMtime = statSync(archivePath).mtime; + appendFileSync(transcriptPath, makeExchangeLines(2, 'session-1'), 'utf-8'); + // Appends can share a filesystem timestamp with the archived prefix. + utimesSync(transcriptPath, archiveMtime, archiveMtime); + await indexSession('session-1', 1, true); + + const after = new Database(dbPath); + const rows = after.prepare('SELECT line_start AS lineStart, last_indexed AS lastIndexed FROM exchanges ORDER BY line_start') + .all() as Array<{ lineStart: number; lastIndexed: number }>; + after.close(); + expect(rows).toHaveLength(2); + expect(rows[0]).toEqual({ lineStart: 1, lastIndexed: 123 }); + expect(rows[1].lineStart).toBe(3); + expect(rows[1].lastIndexed).toBeGreaterThan(123); + expect(readFileSync(archivePath, 'utf-8')).toBe(readFileSync(transcriptPath, 'utf-8')); + }); });