Compare commits

...
Author SHA1 Message Date
root c21180c972 perf: incremental extract — only process slugs that sync touched
The autopilot-cycle runs every 5 min. Its extract phase was doing a full
filesystem walk of ALL markdown files (54K+) — twice (links + timeline).
On a brain this size, extract alone exceeded the 600s job timeout,
producing zero useful writes.

Fix: sync already returns pagesAffected (the slugs it added/modified).
Pipe that list through to extract. When provided, extract reads ONLY
those files instead of walking the entire brain directory.

- Add ExtractOpts.slugs for targeted extraction
- Add extractForSlugs() — single-pass links + timeline for specific slugs
- cycle.ts: capture sync's pagesAffected, pass to runPhaseExtract
- If sync didn't run or failed, extract falls back to full walk (safe)
- If pagesAffected is empty (nothing changed), extract returns instantly

Expected improvement: 54K file reads → ~10-50 per cycle. The full walk
is still available via CLI `gbrain extract` and on first-run.
2026-04-25 00:46:35 +00:00
2 changed files with 168 additions and 5 deletions
+134
View File
@@ -295,6 +295,13 @@ export interface ExtractOpts {
dryRun?: boolean;
/** Emit JSON (progress to stderr, result to stdout) instead of human text. */
jsonMode?: boolean;
/**
* Incremental mode: only extract from these specific slugs.
* When provided, skips the full directory walk and reads only the
* files corresponding to these slugs. Massive perf win on large brains.
* Pass undefined or omit for a full walk (CLI / first-run path).
*/
slugs?: string[];
}
/**
@@ -315,6 +322,21 @@ export async function runExtractCore(engine: BrainEngine, opts: ExtractOpts): Pr
const jsonMode = !!opts.jsonMode;
const result: ExtractResult = { links_created: 0, timeline_entries_created: 0, pages_processed: 0 };
// Incremental path: if specific slugs provided, only extract from those files.
// This is the cycle path — sync tells us what changed, we only re-extract those.
if (opts.slugs !== undefined) {
if (opts.slugs.length === 0) {
// Nothing changed — skip entirely.
return result;
}
const r = await extractForSlugs(engine, opts.dir, opts.slugs, opts.mode, dryRun, jsonMode);
result.links_created = r.links_created;
result.timeline_entries_created = r.timeline_created;
result.pages_processed = r.pages;
return result;
}
// Full walk path: CLI `gbrain extract` or first-run.
if (opts.mode === 'links' || opts.mode === 'all') {
const r = await extractLinksFromDir(engine, opts.dir, dryRun, jsonMode);
result.links_created = r.created;
@@ -411,6 +433,118 @@ export async function runExtract(engine: BrainEngine, args: string[]) {
}
}
/**
* Incremental extract: process only the specified slugs.
*
* Instead of walking 54K+ files, reads only the files that sync says changed.
* Still needs the full slug set for link resolution (resolveSlug needs to know
* all valid targets), but that's a single readdir, not 54K readFileSync calls.
*
* Combines links + timeline extraction in a single pass over each file —
* the full-walk path reads every file TWICE (once for links, once for timeline).
*/
async function extractForSlugs(
engine: BrainEngine,
brainDir: string,
slugs: string[],
mode: 'links' | 'timeline' | 'all',
dryRun: boolean,
jsonMode: boolean,
): Promise<{ links_created: number; timeline_created: number; pages: number }> {
// Build the full slug set for link resolution (fast: just readdir, no file reads)
const allFiles = walkMarkdownFiles(brainDir);
const allSlugs = new Set(allFiles.map(f => f.relPath.replace('.md', '')));
const doLinks = mode === 'links' || mode === 'all';
const doTimeline = mode === 'timeline' || mode === 'all';
const progress = createProgress(cliOptsToProgressOptions(getCliOptions()));
progress.start('extract.incremental', slugs.length);
let linksCreated = 0;
let timelineCreated = 0;
let pagesProcessed = 0;
const linkBatch: LinkBatchInput[] = [];
const timelineBatch: TimelineBatchInput[] = [];
async function flushLinks() {
if (linkBatch.length === 0) return;
try {
linksCreated += await engine.addLinksBatch(linkBatch);
} catch (e) {
const msg = e instanceof Error ? e.message : String(e);
if (!jsonMode) console.error(` link batch error (${linkBatch.length} rows lost): ${msg}`);
} finally {
linkBatch.length = 0;
}
}
async function flushTimeline() {
if (timelineBatch.length === 0) return;
try {
timelineCreated += await engine.addTimelineEntriesBatch(timelineBatch);
} catch (e) {
const msg = e instanceof Error ? e.message : String(e);
if (!jsonMode) console.error(` timeline batch error (${timelineBatch.length} rows lost): ${msg}`);
} finally {
timelineBatch.length = 0;
}
}
for (const slug of slugs) {
const relPath = slug + '.md';
const fullPath = join(brainDir, relPath);
try {
if (!existsSync(fullPath)) continue; // deleted file — sync already handled removal
const content = readFileSync(fullPath, 'utf-8');
// Links
if (doLinks) {
const links = await extractLinksFromFile(content, relPath, allSlugs);
for (const link of links) {
if (dryRun) {
if (!jsonMode) console.log(` ${link.from_slug}${link.to_slug} (${link.link_type})`);
linksCreated++;
} else {
linkBatch.push(link);
if (linkBatch.length >= BATCH_SIZE) await flushLinks();
}
}
}
// Timeline
if (doTimeline) {
const entries = extractTimelineFromContent(content, slug);
for (const entry of entries) {
if (dryRun) {
if (!jsonMode) console.log(` ${entry.slug}: ${entry.date}${entry.summary}`);
timelineCreated++;
} else {
timelineBatch.push({ slug: entry.slug, date: entry.date, source: entry.source, summary: entry.summary, detail: entry.detail });
if (timelineBatch.length >= BATCH_SIZE) await flushTimeline();
}
}
}
pagesProcessed++;
} catch { /* skip unreadable */ }
progress.tick(1);
}
await flushLinks();
await flushTimeline();
progress.finish();
if (!jsonMode) {
const label = dryRun ? '(dry run) would create' : 'created';
console.log(`Incremental extract: ${label} ${linksCreated} link(s), ${timelineCreated} timeline entries from ${pagesProcessed}/${slugs.length} page(s)`);
}
return { links_created: linksCreated, timeline_created: timelineCreated, pages: pagesProcessed };
}
async function extractLinksFromDir(
engine: BrainEngine, brainDir: string, dryRun: boolean, jsonMode: boolean,
): Promise<{ created: number; pages: number }> {
+34 -5
View File
@@ -416,12 +416,18 @@ async function runPhaseBacklinks(brainDir: string, dryRun: boolean): Promise<Pha
}
}
/** Extended sync result that also carries the changed slug list for downstream phases. */
interface SyncPhaseResult extends PhaseResult {
/** Slugs that sync added or modified. Used by extract for incremental processing. */
pagesAffected?: string[];
}
async function runPhaseSync(
engine: BrainEngine,
brainDir: string,
dryRun: boolean,
pull: boolean,
): Promise<PhaseResult> {
): Promise<SyncPhaseResult> {
try {
const { performSync } = await import('../commands/sync.ts');
const result = await performSync(engine, {
@@ -448,6 +454,7 @@ async function runPhaseSync(
syncStatus: result.status,
dryRun,
},
pagesAffected: result.pagesAffected,
};
} catch (e) {
return {
@@ -465,6 +472,7 @@ async function runPhaseExtract(
engine: BrainEngine,
brainDir: string,
dryRun: boolean,
changedSlugs?: string[],
): Promise<PhaseResult> {
try {
const { runExtractCore } = await import('../commands/extract.ts');
@@ -480,15 +488,29 @@ async function runPhaseExtract(
details: { dryRun: true, reason: 'no_dry_run_support' },
};
}
const result = await runExtractCore(engine, { mode: 'all', dir: brainDir });
// Incremental path: if sync told us which slugs changed, only extract those.
// On a 54K-page brain this turns a 10-minute full walk into a sub-second pass.
const result = await runExtractCore(engine, {
mode: 'all',
dir: brainDir,
slugs: changedSlugs, // undefined = full walk (first run / manual)
});
const linksCreated = result?.links_created ?? 0;
const timelineCreated = result?.timeline_entries_created ?? 0;
const incremental = changedSlugs !== undefined;
return {
phase: 'extract',
status: 'ok',
duration_ms: 0,
summary: `${linksCreated} link(s), ${timelineCreated} timeline entries`,
details: { linksCreated, timelineCreated, pages_processed: result?.pages_processed ?? 0 },
summary: incremental
? `${linksCreated} link(s), ${timelineCreated} timeline entries (incremental: ${changedSlugs.length} slugs)`
: `${linksCreated} link(s), ${timelineCreated} timeline entries`,
details: {
linksCreated, timelineCreated,
pages_processed: result?.pages_processed ?? 0,
incremental,
...(incremental ? { slugs_targeted: changedSlugs.length } : {}),
},
};
} catch (e) {
return {
@@ -663,6 +685,8 @@ export async function runCycle(
}
// ── Phase 3: sync ───────────────────────────────────────────
// Track which slugs sync touched so extract can run incrementally.
let syncPagesAffected: string[] | undefined;
if (phases.includes('sync')) {
if (!engine) {
phaseResults.push({
@@ -676,6 +700,8 @@ export async function runCycle(
progress.start('cycle.sync');
const { result, duration_ms } = await timePhase(() => runPhaseSync(engine, opts.brainDir, dryRun, pull));
result.duration_ms = duration_ms;
// Capture changed slugs for incremental extract.
syncPagesAffected = (result as SyncPhaseResult).pagesAffected;
phaseResults.push(result);
progress.finish();
}
@@ -693,8 +719,11 @@ export async function runCycle(
details: { reason: 'no_database' },
});
} else {
// Pass changed slugs from sync for incremental extract.
// If sync didn't run (phases exclude it) or failed, syncPagesAffected
// is undefined → extract falls back to full walk (safe default).
progress.start('cycle.extract');
const { result, duration_ms } = await timePhase(() => runPhaseExtract(engine, opts.brainDir, dryRun));
const { result, duration_ms } = await timePhase(() => runPhaseExtract(engine, opts.brainDir, dryRun, syncPagesAffected));
result.duration_ms = duration_ms;
phaseResults.push(result);
progress.finish();