Spaces:
Paused
Paused
File size: 4,051 Bytes
fb4d8fe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | import type { DatabaseSync } from "node:sqlite";
import type { SessionFileEntry } from "./session-files.js";
import { createSubsystemLogger } from "../logging/subsystem.js";
import {
buildSessionEntry,
listSessionFilesForAgent,
sessionPathForFile,
} from "./session-files.js";
const log = createSubsystemLogger("memory");
type ProgressState = {
completed: number;
total: number;
label?: string;
report: (update: { completed: number; total: number; label?: string }) => void;
};
export async function syncSessionFiles(params: {
agentId: string;
db: DatabaseSync;
needsFullReindex: boolean;
progress?: ProgressState;
batchEnabled: boolean;
concurrency: number;
runWithConcurrency: <T>(tasks: Array<() => Promise<T>>, concurrency: number) => Promise<T[]>;
indexFile: (entry: SessionFileEntry) => Promise<void>;
vectorTable: string;
ftsTable: string;
ftsEnabled: boolean;
ftsAvailable: boolean;
model: string;
dirtyFiles: Set<string>;
}) {
const files = await listSessionFilesForAgent(params.agentId);
const activePaths = new Set(files.map((file) => sessionPathForFile(file)));
const indexAll = params.needsFullReindex || params.dirtyFiles.size === 0;
log.debug("memory sync: indexing session files", {
files: files.length,
indexAll,
dirtyFiles: params.dirtyFiles.size,
batch: params.batchEnabled,
concurrency: params.concurrency,
});
if (params.progress) {
params.progress.total += files.length;
params.progress.report({
completed: params.progress.completed,
total: params.progress.total,
label: params.batchEnabled ? "Indexing session files (batch)..." : "Indexing session files…",
});
}
const tasks = files.map((absPath) => async () => {
if (!indexAll && !params.dirtyFiles.has(absPath)) {
if (params.progress) {
params.progress.completed += 1;
params.progress.report({
completed: params.progress.completed,
total: params.progress.total,
});
}
return;
}
const entry = await buildSessionEntry(absPath);
if (!entry) {
if (params.progress) {
params.progress.completed += 1;
params.progress.report({
completed: params.progress.completed,
total: params.progress.total,
});
}
return;
}
const record = params.db
.prepare(`SELECT hash FROM files WHERE path = ? AND source = ?`)
.get(entry.path, "sessions") as { hash: string } | undefined;
if (!params.needsFullReindex && record?.hash === entry.hash) {
if (params.progress) {
params.progress.completed += 1;
params.progress.report({
completed: params.progress.completed,
total: params.progress.total,
});
}
return;
}
await params.indexFile(entry);
if (params.progress) {
params.progress.completed += 1;
params.progress.report({
completed: params.progress.completed,
total: params.progress.total,
});
}
});
await params.runWithConcurrency(tasks, params.concurrency);
const staleRows = params.db
.prepare(`SELECT path FROM files WHERE source = ?`)
.all("sessions") as Array<{ path: string }>;
for (const stale of staleRows) {
if (activePaths.has(stale.path)) {
continue;
}
params.db
.prepare(`DELETE FROM files WHERE path = ? AND source = ?`)
.run(stale.path, "sessions");
try {
params.db
.prepare(
`DELETE FROM ${params.vectorTable} WHERE id IN (SELECT id FROM chunks WHERE path = ? AND source = ?)`,
)
.run(stale.path, "sessions");
} catch {}
params.db
.prepare(`DELETE FROM chunks WHERE path = ? AND source = ?`)
.run(stale.path, "sessions");
if (params.ftsEnabled && params.ftsAvailable) {
try {
params.db
.prepare(`DELETE FROM ${params.ftsTable} WHERE path = ? AND source = ? AND model = ?`)
.run(stale.path, "sessions", params.model);
} catch {}
}
}
}
|