fix(indexer): keep session indexing incremental on append

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;

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<SyncResult>;

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
         }
     }

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();

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
     }
   }

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'));
+  });
 });