diff --git a/dist/index-cli.js b/dist/index-cli.js index 950a4c7d..7975f979 100644 --- a/dist/index-cli.js +++ b/dist/index-cli.js @@ -66,11 +66,12 @@ async function main() { console.log(`Orphaned entries: ${issues.orphaned.length}`); console.log(`Outdated files: ${issues.outdated.length}`); console.log(`Corrupted files: ${issues.corrupted.length}`); + console.log(`Not in index: ${issues.unindexed.length}`); if (issues.missing.length > 0) { console.log('\nMissing summaries:'); issues.missing.forEach(m => console.log(` ${m.path}`)); } - if (issues.missing.length + issues.orphaned.length + issues.outdated.length + issues.corrupted.length > 0) { + if (issues.missing.length + issues.orphaned.length + issues.outdated.length + issues.corrupted.length + issues.unindexed.length > 0) { console.log('\nRun with --repair to fix these issues.'); process.exit(1); } diff --git a/dist/sync.js b/dist/sync.js index 0629761d..1b5dd10b 100644 --- a/dist/sync.js +++ b/dist/sync.js @@ -11,6 +11,7 @@ const EXCLUSION_MARKERS = [ SUMMARIZER_CONTEXT_MARKER, ]; const MARKER_SCAN_CHUNK_BYTES = 1 << 20; // 1 MiB +const RENAME_RETRIES = 4; /** * Stream and scan for any exclusion marker, carrying an overlap between * chunks so a marker split across a boundary is still found. A single @@ -136,10 +137,29 @@ function copyIfNewer(src, dest) { return false; // Dest is current, skip } } - // Atomic copy: temp file + rename + // Atomic copy: temp file + rename. On Windows the rename throws (EPERM/EBUSY) while + // another process has the destination open; retry, then fall back to a direct copy, + // and always remove the temp file so a failure never leaves a full copy behind. const tempDest = dest + '.tmp.' + process.pid; - fs.copyFileSync(src, tempDest); - fs.renameSync(tempDest, dest); // Atomic on same filesystem + try { + fs.copyFileSync(src, tempDest); + for (let attempt = 0;; attempt++) { + try { + fs.renameSync(tempDest, dest); // Atomic on same filesystem + break; + } + catch (error) { + if (attempt >= RENAME_RETRIES) { + fs.copyFileSync(src, dest); + break; + } + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 100 * (attempt + 1)); + } + } + } + finally { + fs.rmSync(tempDest, { force: true }); + } // Preserve source mtime: harnesses without per-message timestamps (Cursor // agent transcripts) fall back to file mtime. Round up to the next whole // millisecond — utimes can't always represent the source's sub-millisecond diff --git a/dist/verify.d.ts b/dist/verify.d.ts index 9558c879..2770ad92 100644 --- a/dist/verify.d.ts +++ b/dist/verify.d.ts @@ -16,6 +16,10 @@ export interface VerificationResult { path: string; error: string; }>; + /** Archived conversations with a real summary but no rows in the index (e.g. after the database was rebuilt). */ + unindexed: Array<{ + path: string; + }>; } export declare function verifyIndex(): Promise; export declare function repairIndex(issues: VerificationResult): Promise; diff --git a/dist/verify.js b/dist/verify.js index dd4535d2..ab5f1448 100644 --- a/dist/verify.js +++ b/dist/verify.js @@ -9,7 +9,8 @@ export async function verifyIndex() { missing: [], orphaned: [], outdated: [], - corrupted: [] + corrupted: [], + unindexed: [] }; const archiveDir = getArchiveDir(); // Track all files we find @@ -49,13 +50,18 @@ export async function verifyIndex() { result.missing.push({ path: conversationPath, reason: 'No summary file' }); continue; } - if (isErroredSentinel(fs.readFileSync(summaryPath, 'utf-8'))) { + const summaryContent = fs.readFileSync(summaryPath, 'utf-8'); + if (isErroredSentinel(summaryContent)) { result.missing.push({ path: conversationPath, reason: 'Previous summarization failed (error sentinel)' }); continue; } - // Check if file is outdated (modified after last_indexed) + // Check if file is outdated (modified after last_indexed), or never indexed at all. + // An empty summary is the permanent-skip sentinel (#91): nothing to index there. const lastIndexed = getFileLastIndexed(db, conversationPath); - if (lastIndexed !== null) { + if (lastIndexed === null && summaryContent.length > 0) { + result.unindexed.push({ path: conversationPath }); + } + else if (lastIndexed !== null) { const fileStat = fs.statSync(conversationPath); if (fileStat.mtimeMs > lastIndexed) { result.outdated.push({ @@ -105,12 +111,14 @@ export async function repairIndex(issues) { console.log(`Removing orphaned entry: ${orphan.uuid}`); deleteExchange(db, orphan.uuid); } - // Re-index missing and outdated conversations + // Re-index missing and outdated conversations, and index the unindexed ones. An unindexed + // conversation already has its summary, so it is not summarized again. const toReindex = [ - ...issues.missing.map(m => m.path), - ...issues.outdated.map(o => o.path) + ...issues.missing.map(m => ({ conversationPath: m.path, summarize: true })), + ...issues.outdated.map(o => ({ conversationPath: o.path, summarize: true })), + ...(issues.unindexed ?? []).map(u => ({ conversationPath: u.path, summarize: false })) ]; - for (const conversationPath of toReindex) { + for (const { conversationPath, summarize } of toReindex) { console.log(`Re-indexing: ${conversationPath}`); try { // Extract project name from path @@ -124,10 +132,12 @@ export async function repairIndex(issues) { continue; } // Generate/update summary - const summaryPath = conversationPath.replace('.jsonl', '-summary.txt'); - const summary = await summarizeConversation(exchanges); - fs.writeFileSync(summaryPath, summary, 'utf-8'); - console.log(` Created summary: ${summary.split(/\s+/).length} words`); + if (summarize) { + const summaryPath = conversationPath.replace('.jsonl', '-summary.txt'); + const summary = await summarizeConversation(exchanges); + fs.writeFileSync(summaryPath, summary, 'utf-8'); + console.log(` Created summary: ${summary.split(/\s+/).length} words`); + } // Index exchanges for (const exchange of exchanges) { const toolNames = exchange.toolCalls?.map(tc => tc.toolName); diff --git a/src/index-cli.ts b/src/index-cli.ts index 272ecdba..d917a9f3 100644 --- a/src/index-cli.ts +++ b/src/index-cli.ts @@ -74,13 +74,14 @@ async function main() { console.log(`Orphaned entries: ${issues.orphaned.length}`); console.log(`Outdated files: ${issues.outdated.length}`); console.log(`Corrupted files: ${issues.corrupted.length}`); + console.log(`Not in index: ${issues.unindexed.length}`); if (issues.missing.length > 0) { console.log('\nMissing summaries:'); issues.missing.forEach(m => console.log(` ${m.path}`)); } - if (issues.missing.length + issues.orphaned.length + issues.outdated.length + issues.corrupted.length > 0) { + if (issues.missing.length + issues.orphaned.length + issues.outdated.length + issues.corrupted.length + issues.unindexed.length > 0) { console.log('\nRun with --repair to fix these issues.'); process.exit(1); } else { diff --git a/src/sync.ts b/src/sync.ts index d9ee6810..c4626f48 100644 --- a/src/sync.ts +++ b/src/sync.ts @@ -13,6 +13,7 @@ const EXCLUSION_MARKERS = [ ]; const MARKER_SCAN_CHUNK_BYTES = 1 << 20; // 1 MiB +const RENAME_RETRIES = 4; /** * Stream and scan for any exclusion marker, carrying an overlap between @@ -148,10 +149,27 @@ function copyIfNewer(src: string, dest: string): boolean { } } - // Atomic copy: temp file + rename + // Atomic copy: temp file + rename. On Windows the rename throws (EPERM/EBUSY) while + // another process has the destination open; retry, then fall back to a direct copy, + // and always remove the temp file so a failure never leaves a full copy behind. const tempDest = dest + '.tmp.' + process.pid; - fs.copyFileSync(src, tempDest); - fs.renameSync(tempDest, dest); // Atomic on same filesystem + try { + fs.copyFileSync(src, tempDest); + for (let attempt = 0; ; attempt++) { + try { + fs.renameSync(tempDest, dest); // Atomic on same filesystem + break; + } catch (error) { + if (attempt >= RENAME_RETRIES) { + fs.copyFileSync(src, dest); + break; + } + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 100 * (attempt + 1)); + } + } + } finally { + fs.rmSync(tempDest, { force: true }); + } // Preserve source mtime: harnesses without per-message timestamps (Cursor // agent transcripts) fall back to file mtime. Round up to the next whole diff --git a/src/verify.ts b/src/verify.ts index 0dd2c5db..d4ded175 100644 --- a/src/verify.ts +++ b/src/verify.ts @@ -10,6 +10,8 @@ export interface VerificationResult { orphaned: Array<{ uuid: string; path: string }>; outdated: Array<{ path: string; fileTime: number; dbTime: number }>; corrupted: Array<{ path: string; error: string }>; + /** Archived conversations with a real summary but no rows in the index (e.g. after the database was rebuilt). */ + unindexed: Array<{ path: string }>; } export async function verifyIndex(): Promise { @@ -17,7 +19,8 @@ export async function verifyIndex(): Promise { missing: [], orphaned: [], outdated: [], - corrupted: [] + corrupted: [], + unindexed: [] }; const archiveDir = getArchiveDir(); @@ -70,14 +73,18 @@ export async function verifyIndex(): Promise { result.missing.push({ path: conversationPath, reason: 'No summary file' }); continue; } - if (isErroredSentinel(fs.readFileSync(summaryPath, 'utf-8'))) { + const summaryContent = fs.readFileSync(summaryPath, 'utf-8'); + if (isErroredSentinel(summaryContent)) { result.missing.push({ path: conversationPath, reason: 'Previous summarization failed (error sentinel)' }); continue; } - // Check if file is outdated (modified after last_indexed) + // Check if file is outdated (modified after last_indexed), or never indexed at all. + // An empty summary is the permanent-skip sentinel (#91): nothing to index there. const lastIndexed = getFileLastIndexed(db, conversationPath); - if (lastIndexed !== null) { + if (lastIndexed === null && summaryContent.length > 0) { + result.unindexed.push({ path: conversationPath }); + } else if (lastIndexed !== null) { const fileStat = fs.statSync(conversationPath); if (fileStat.mtimeMs > lastIndexed) { result.outdated.push({ @@ -136,13 +143,15 @@ export async function repairIndex(issues: VerificationResult): Promise { deleteExchange(db, orphan.uuid); } - // Re-index missing and outdated conversations + // Re-index missing and outdated conversations, and index the unindexed ones. An unindexed + // conversation already has its summary, so it is not summarized again. const toReindex = [ - ...issues.missing.map(m => m.path), - ...issues.outdated.map(o => o.path) + ...issues.missing.map(m => ({ conversationPath: m.path, summarize: true })), + ...issues.outdated.map(o => ({ conversationPath: o.path, summarize: true })), + ...(issues.unindexed ?? []).map(u => ({ conversationPath: u.path, summarize: false })) ]; - for (const conversationPath of toReindex) { + for (const { conversationPath, summarize } of toReindex) { console.log(`Re-indexing: ${conversationPath}`); try { // Extract project name from path @@ -159,10 +168,12 @@ export async function repairIndex(issues: VerificationResult): Promise { } // Generate/update summary - const summaryPath = conversationPath.replace('.jsonl', '-summary.txt'); - const summary = await summarizeConversation(exchanges); - fs.writeFileSync(summaryPath, summary, 'utf-8'); - console.log(` Created summary: ${summary.split(/\s+/).length} words`); + if (summarize) { + const summaryPath = conversationPath.replace('.jsonl', '-summary.txt'); + const summary = await summarizeConversation(exchanges); + fs.writeFileSync(summaryPath, summary, 'utf-8'); + console.log(` Created summary: ${summary.split(/\s+/).length} words`); + } // Index exchanges for (const exchange of exchanges) { diff --git a/test/tmp-cleanup-and-unindexed.test.ts b/test/tmp-cleanup-and-unindexed.test.ts new file mode 100644 index 00000000..92ddabf9 --- /dev/null +++ b/test/tmp-cleanup-and-unindexed.test.ts @@ -0,0 +1,110 @@ +import { describe, it, expect, beforeEach, afterEach, afterAll, vi } from 'vitest'; +import fs from 'fs'; +import path from 'path'; +import os from 'os'; +import { syncConversations } from '../src/sync.js'; +import { verifyIndex, repairIndex } from '../src/verify.js'; +import { summarizeConversation } from '../src/summarizer.js'; +import { suppressConsole } from './test-utils.js'; + +vi.mock('../src/embeddings.js', () => ({ + initEmbeddings: vi.fn(async () => {}), + generateExchangeEmbedding: vi.fn(async () => new Array(384).fill(0.1)) +})); + +vi.mock('../src/summarizer.js', () => ({ + summarizeConversation: vi.fn(async () => 'Mock summary') +})); + +const restoreConsole = suppressConsole(); + +function line(obj: object): string { + return JSON.stringify(obj); +} + +function toolUseConversation(extraTurns = 0): string { + const lines = [ + line({ type: 'user', message: { role: 'user', content: 'List the files' }, timestamp: '2024-01-01T00:00:00Z' }), + line({ + type: 'assistant', + message: { + role: 'assistant', + content: [ + { type: 'text', text: 'Listing.' }, + { type: 'tool_use', id: 'toolu_01', name: 'Bash', input: { command: 'ls' } }, + { type: 'tool_use', name: 'Read', input: { file_path: 'a.txt' } } // no id: hashed fallback + ] + }, + timestamp: '2024-01-01T00:00:01Z' + }) + ]; + for (let i = 0; i < extraTurns; i++) { + lines.push(line({ type: 'user', message: { role: 'user', content: `more ${i}` }, timestamp: `2024-01-01T00:01:0${i}Z` })); + lines.push(line({ type: 'assistant', message: { role: 'assistant', content: `done ${i}` }, timestamp: `2024-01-01T00:02:0${i}Z` })); + } + return lines.join('\n') + '\n'; +} + +describe('no orphan .tmp copies, unindexed conversations', () => { + let testDir: string; + let archiveDir: string; + + beforeEach(() => { + testDir = fs.mkdtempSync(path.join(os.tmpdir(), 'episodic-memory-fixes-test-')); + archiveDir = path.join(testDir, 'archive'); + fs.mkdirSync(archiveDir, { recursive: true }); + fs.mkdirSync(path.join(testDir, 'projects'), { recursive: true }); + process.env.TEST_PROJECTS_DIR = path.join(testDir, 'projects'); + process.env.TEST_ARCHIVE_DIR = archiveDir; + process.env.TEST_DB_PATH = path.join(testDir, 'db.sqlite'); + vi.mocked(summarizeConversation).mockClear(); + }); + + afterEach(() => { + vi.restoreAllMocks(); + delete process.env.TEST_PROJECTS_DIR; + delete process.env.TEST_ARCHIVE_DIR; + delete process.env.TEST_DB_PATH; + fs.rmSync(testDir, { recursive: true, force: true }); + }); + + it('leaves no .tmp copy behind when the rename onto the archive copy fails', async () => { + const source = path.join(testDir, 'source', 'project'); + fs.mkdirSync(source, { recursive: true }); + fs.writeFileSync(path.join(source, 'conv.jsonl'), toolUseConversation()); + const rename = vi.spyOn(fs, 'renameSync').mockImplementation(() => { + throw Object.assign(new Error('EPERM: operation not permitted, rename'), { code: 'EPERM' }); + }); + + const result = await syncConversations(path.join(testDir, 'source'), archiveDir, { skipIndex: true, skipSummaries: true }); + + expect(rename).toHaveBeenCalled(); + expect(result.copied).toBe(1); + const files = fs.readdirSync(path.join(archiveDir, 'project')); + expect(files.filter(f => /\.tmp\.\d+$/.test(f))).toEqual([]); + expect(fs.readFileSync(path.join(archiveDir, 'project', 'conv.jsonl'), 'utf-8')).toBe(toolUseConversation()); + }); + + it('reports an archived conversation with a summary but no rows as unindexed, and repair indexes it without summarizing', async () => { + const project = path.join(archiveDir, 'project'); + fs.mkdirSync(project, { recursive: true }); + const conv = path.join(project, 'conv.jsonl'); + fs.writeFileSync(conv, toolUseConversation()); + fs.writeFileSync(conv.replace('.jsonl', '-summary.txt'), 'An existing summary'); + // An empty summary is the permanent-skip sentinel: never reported as unindexed. + fs.writeFileSync(path.join(project, 'empty.jsonl'), line({ type: 'summary', summary: 'x' }) + '\n'); + fs.writeFileSync(path.join(project, 'empty-summary.txt'), ''); + + const before = await verifyIndex(); + expect(before.unindexed.map(u => u.path)).toEqual([conv]); + expect(before.missing).toEqual([]); + + await repairIndex(before); + + expect(summarizeConversation).not.toHaveBeenCalled(); + expect(fs.readFileSync(conv.replace('.jsonl', '-summary.txt'), 'utf-8')).toBe('An existing summary'); + const after = await verifyIndex(); + expect(after.unindexed).toEqual([]); + }); + afterAll(() => restoreConsole()); +});