mirror of
https://github.com/garrytan/gbrain.git
synced 2026-08-14 00:48:18 +00:00
Garry TanandClaude Opus 4.7
f56cc619ba
* fix(sync): infiniteGameExp + foxhoundinc 5-bug wave (#1422, #1433, #1434, #1309, #1436) Five real production bugs from infiniteGameExp (PostgreSQL onboarding) and foxhoundinc (dream-cycle reproduction), each silent-failure shape where gbrain told the user the operation succeeded when it didn't. * #1422 — `gbrain dream` swallowed connectEngine errors. Bind the caught error and surface `[dream] WARNING: could not connect to DB (...)` on stderr before falling through to filesystem-only phases. runDream(null) no-DB fallback preserved. * #1433 — `gbrain sync` deleted previously-indexed log.md / schema.md / index.md / README.md pages on every re-sync. Refactor isSyncable through private classifySync helper; expose unsyncableReason (companion returning the same tagged reason) and SYNC_SKIP_FILES named export. Cleanup loop guards on reason === 'metafile' before deleting. * #1434 — `gbrain sync` without --source on single-vault brains routed to source_id='default' (zero pages) and silently failed. Add resolver tier 5.5 'sole_non_default' AFTER brain_default (explicit user intent wins). Wire runSync + runImport to call resolveSourceWithTier unconditionally so the tier actually fires. Stderr nudge on tier hit; suppress with GBRAIN_NO_SOLE_NON_DEFAULT_NUDGE=1. * #1309 — overlapping ingest roots created duplicate pages. New BrainEngine.findDuplicatePage?(sourceId, {hash, frontmatterId}) with identity-based posture: SKIP when frontmatter.id matches (true external duplicate), WARN-ALWAYS on content_hash collision with different/missing fm.id, FAIL CLOSED on lookup error. Migration v95 adds partial index pages_dedup_idx (Postgres CONCURRENTLY, PGLite plain CREATE). * #1436 — MCP fuzzy get_page returned slug candidates from sources outside caller's scope. resolveSlugs signature extended with {sourceId?, sourceIds?} matching the sourceScopeOpts helper output; operations.ts threads it through. Both engines preserve unscoped back-compat for internal CLI callers. Plus a stable tiebreaker on searchVector ORDER BY (score DESC, page_id ASC, chunk_id ASC) in both engines. Caught while wiring the index above — basis-vector eval fixtures with tied scores depend on planner row order, which any new index on pages could flip. Pins eval-replay-gate ranking determinism against future index changes. Per codex review of the original plan: caught 6 load-bearing gaps that the engineering review missed (runSync bypass, #1436 misclassified as fixed, dedup fail-open, content-hash-alone too aggressive, soft-delete filter missing, tier-ordering contradiction). All folded in pre-merge. Tests: 65 new wave cases across 7 new files + 1 extended; all green. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * chore: bump version and changelog (v0.41.13.0) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2617 lines
110 KiB
TypeScript
2617 lines
110 KiB
TypeScript
import { existsSync, readFileSync, writeFileSync, statSync } from 'fs';
|
||
import { execFileSync } from 'child_process';
|
||
import { join, relative } from 'path';
|
||
import type { BrainEngine } from '../core/engine.ts';
|
||
import { importFile } from '../core/import-file.ts';
|
||
import { collectSyncableFiles } from './import.ts';
|
||
import { createInterface } from 'readline';
|
||
import {
|
||
buildSyncManifest,
|
||
isSyncable,
|
||
unsyncableReason,
|
||
resolveSlugForPath,
|
||
recordSyncFailures,
|
||
unacknowledgedSyncFailures,
|
||
acknowledgeSyncFailures,
|
||
formatCodeBreakdown,
|
||
} from '../core/sync.ts';
|
||
import { estimateTokens, CHUNKER_VERSION } from '../core/chunkers/code.ts';
|
||
import { EMBEDDING_MODEL, estimateEmbeddingCostUsd } from '../core/embedding.ts';
|
||
import { errorFor, serializeError } from '../core/errors.ts';
|
||
import type { SyncManifest } from '../core/sync.ts';
|
||
import { createProgress } from '../core/progress.ts';
|
||
import { getCliOptions, cliOptsToProgressOptions } from '../core/cli-options.ts';
|
||
import { loadConfig } from '../core/config.ts';
|
||
import {
|
||
autoConcurrency,
|
||
shouldRunParallel,
|
||
parseWorkers,
|
||
DEFAULT_PARALLEL_SOURCES,
|
||
} from '../core/sync-concurrency.ts';
|
||
import {
|
||
tryAcquireDbLock,
|
||
withRefreshingLock,
|
||
LockUnavailableError,
|
||
syncLockId,
|
||
SYNC_LOCK_ID,
|
||
} from '../core/db-lock.ts';
|
||
import {
|
||
withSourcePrefix,
|
||
slog,
|
||
serr,
|
||
} from '../core/console-prefix.ts';
|
||
import { loadStorageConfig } from '../core/storage-config.ts';
|
||
import { getDefaultSourcePath } from '../core/source-resolver.ts';
|
||
import { sortNewestFirst } from '../core/sort-newest-first.ts';
|
||
|
||
export interface SyncResult {
|
||
status: 'up_to_date' | 'synced' | 'first_sync' | 'dry_run' | 'blocked_by_failures';
|
||
fromCommit: string | null;
|
||
toCommit: string;
|
||
added: number;
|
||
modified: number;
|
||
deleted: number;
|
||
renamed: number;
|
||
chunksCreated: number;
|
||
/** Pages re-embedded during this sync's auto-embed step. 0 if --no-embed or skipped. */
|
||
embedded: number;
|
||
pagesAffected: string[];
|
||
failedFiles?: number; // count of parse failures (Bug 9)
|
||
}
|
||
|
||
/**
|
||
* v0.20.0 Cathedral II Layer 8 (D1) — walk each source's working tree and
|
||
* sum tokens for every syncable file. This is a conservative overestimate
|
||
* (full file content, not just the incremental diff) because `sync --all`
|
||
* on a source that hasn't been synced yet WILL embed every file in the
|
||
* working tree. For already-synced sources with only incremental changes,
|
||
* the overestimate is the ceiling, not the floor — users never get
|
||
* surprised by MORE cost than the preview claims. The false-high bias is
|
||
* intentional: a lower estimate that undersells the real bill would be
|
||
* worse than one that oversells.
|
||
*/
|
||
function estimateSyncAllCost(sources: Array<{ local_path: string | null; config: Record<string, unknown> }>): {
|
||
totalTokens: number;
|
||
totalFiles: number;
|
||
activeSources: number;
|
||
perSource: Array<{ path: string; tokens: number; files: number }>;
|
||
} {
|
||
let totalTokens = 0;
|
||
let totalFiles = 0;
|
||
let activeSources = 0;
|
||
const perSource: Array<{ path: string; tokens: number; files: number }> = [];
|
||
|
||
for (const src of sources) {
|
||
if (!src.local_path) continue;
|
||
const cfg = (src.config || {}) as { syncEnabled?: boolean; strategy?: 'markdown' | 'code' | 'auto' };
|
||
if (cfg.syncEnabled === false) continue;
|
||
activeSources++;
|
||
let sourceTokens = 0;
|
||
let sourceFiles = 0;
|
||
try {
|
||
// v0.31.2: cost preview routed through collectSyncableFiles
|
||
// (single hardened walker; see import.ts). Previously
|
||
// walkSyncableFiles used statSync (followed symlinks). New walker
|
||
// uses lstat + inode-cycle + max-depth so the preview matches
|
||
// what the real sync will actually walk.
|
||
const files = collectSyncableFiles(src.local_path, { strategy: cfg.strategy ?? 'markdown' });
|
||
for (const fullPath of files) {
|
||
try {
|
||
const stat = statSync(fullPath);
|
||
if (stat.size > 5_000_000) continue; // skip large binaries
|
||
const content = readFileSync(fullPath, 'utf-8');
|
||
sourceTokens += estimateTokens(content);
|
||
sourceFiles++;
|
||
} catch {
|
||
// Best-effort per file. Skip unreadable files silently;
|
||
// sync itself tolerates the same.
|
||
}
|
||
}
|
||
} catch {
|
||
// Best-effort: a source whose local_path is gone or unreadable just
|
||
// contributes 0. The sync itself would have failed anyway; no point
|
||
// blocking the preview on a pre-existing fault.
|
||
}
|
||
totalTokens += sourceTokens;
|
||
totalFiles += sourceFiles;
|
||
perSource.push({ path: src.local_path, tokens: sourceTokens, files: sourceFiles });
|
||
}
|
||
|
||
return { totalTokens, totalFiles, activeSources, perSource };
|
||
}
|
||
|
||
/** Interactive [y/N] prompt. Resolves false on non-y answers or EOF. */
|
||
async function promptYesNo(question: string): Promise<boolean> {
|
||
return new Promise((resolve) => {
|
||
const rl = createInterface({ input: process.stdin, output: process.stdout });
|
||
rl.question(question, (answer) => {
|
||
rl.close();
|
||
resolve(answer.trim().toLowerCase() === 'y' || answer.trim().toLowerCase() === 'yes');
|
||
});
|
||
rl.on('close', () => resolve(false));
|
||
});
|
||
}
|
||
|
||
export interface SyncOpts {
|
||
repoPath?: string;
|
||
dryRun?: boolean;
|
||
full?: boolean;
|
||
noPull?: boolean;
|
||
noEmbed?: boolean;
|
||
noExtract?: boolean;
|
||
/** Bug 9 — acknowledge + skip past current failure set (CLI --skip-failed). */
|
||
skipFailed?: boolean;
|
||
/** Bug 9 — re-attempt unacknowledged failures explicitly (CLI --retry-failed). */
|
||
retryFailed?: boolean;
|
||
/**
|
||
* v0.18.0 Step 5 — sync a specific named source. When set, sync reads
|
||
* local_path + last_commit from the sources table (not the global
|
||
* config.sync.* keys) and writes last_commit + last_sync_at back to
|
||
* the same row. Backward compat: when undefined, sync uses the
|
||
* pre-v0.17 global-config path unchanged.
|
||
*/
|
||
sourceId?: string;
|
||
/** Multi-repo: sync strategy override (markdown, code, auto). */
|
||
strategy?: 'markdown' | 'code' | 'auto';
|
||
/**
|
||
* Number of parallel workers for the import phase. When > 1, each worker
|
||
* gets its own small Postgres connection pool and files are dispatched via
|
||
* an atomic queue index (same pattern as `import --workers N`).
|
||
*
|
||
* Deletes and renames remain serial (order-dependent).
|
||
* Default: undefined → auto-concurrency picks (`src/core/sync-concurrency.ts`).
|
||
*
|
||
* v0.22.13 (PR #490 Q1): when this is explicitly set, the >50-file floor
|
||
* is bypassed — explicit user intent beats the auto-path safety net.
|
||
*/
|
||
concurrency?: number;
|
||
/**
|
||
* Internal: skip acquiring the gbrain-sync DB lock. Set by the cycle
|
||
* handler (cycle.ts) which already holds gbrain-cycle and therefore
|
||
* already serializes against other cycle runs. CLI sync, jobs handler,
|
||
* and any external caller leave this undefined so they take the lock.
|
||
*
|
||
* v0.22.13 (PR #490 CODEX-2). Not part of the public CLI surface.
|
||
*/
|
||
skipLock?: boolean;
|
||
/**
|
||
* Internal: override the DB lock id taken around the writer window.
|
||
* Not part of the public CLI surface — explicit escape hatch for
|
||
* callers that already know which lock id they want.
|
||
*
|
||
* Defaults to:
|
||
* - `gbrain-sync:<sourceId>` when `opts.sourceId` is set
|
||
* (multi-source / federated brains; the per-source invariant)
|
||
* - `gbrain-sync` (the legacy global lock) when `sourceId` is unset
|
||
* (single-default-source brains; preserves bit-for-bit behavior
|
||
* for installs that never set up multiple sources)
|
||
*
|
||
* Why source-id keyed by default (v0.40.3.0):
|
||
* PR #1314 originally only changed the lock id inside the parallel
|
||
* `sync --all` fan-out. That introduced a worse race than the global
|
||
* lock fixes — `gbrain sync --all` (per-source lock) running
|
||
* concurrently with `gbrain sync --source foo` (global lock) would
|
||
* both write source foo simultaneously. The fix: every source-scoped
|
||
* sync (CLI, --all fan-out, cycle, jobs handler) defaults to the
|
||
* per-source lock. Same source = same lock id, always.
|
||
*
|
||
* For the per-source path, `performSync` ALSO switches from bare
|
||
* `tryAcquireDbLock` to `withRefreshingLock` so long-running sources
|
||
* (the PR's whole motivation — media-corpus, 250K+ chunks) don't lose
|
||
* their lock at the 30-minute TTL mid-run. The legacy global-lock
|
||
* path keeps bare `tryAcquireDbLock` for back-compat (no caller
|
||
* depends on the global lock surviving past 30 minutes today).
|
||
*
|
||
* Total live Postgres connections per parallel `sync --all` wave:
|
||
* parallel × workers × 2 (per-file pool inside each worker)
|
||
* + parent pool. See DEFAULT_PARALLEL_SOURCES in sync-concurrency.ts.
|
||
*/
|
||
lockId?: string;
|
||
}
|
||
|
||
/**
|
||
* v0.32.7 CJK wave (codex post-merge F4): resolve a slug by `pages.source_path`
|
||
* first, falling back to `resolveSlugForPath(path)`.
|
||
*
|
||
* Frontmatter-fallback pages (emoji-only / Thai / Arabic / exotic-script
|
||
* filenames where `slugifyPath` returns empty and the slug came from the
|
||
* frontmatter) have a slug that ISN'T derivable from the path. Delete and
|
||
* rename operations that only know the path would otherwise orphan these
|
||
* pages by trying to delete the path-derived (wrong) slug.
|
||
*
|
||
* Returns the actual stored slug when source_path matches a row, or the
|
||
* path-derived slug when there's no match (normal-case path-derived pages).
|
||
*/
|
||
export async function resolveSlugByPathOrSourcePath(
|
||
engine: BrainEngine,
|
||
path: string,
|
||
sourceId?: string,
|
||
): Promise<string> {
|
||
try {
|
||
const rows = await engine.executeRaw<{ slug: string }>(
|
||
sourceId
|
||
? `SELECT slug FROM pages WHERE source_path = $1 AND source_id = $2 LIMIT 1`
|
||
: `SELECT slug FROM pages WHERE source_path = $1 LIMIT 1`,
|
||
sourceId ? [path, sourceId] : [path],
|
||
);
|
||
if (rows.length > 0 && rows[0].slug) return rows[0].slug;
|
||
} catch {
|
||
// Fall through — best-effort. Pre-migration brains or query errors
|
||
// shouldn't break delete/rename for path-derived pages.
|
||
}
|
||
return resolveSlugForPath(path);
|
||
}
|
||
|
||
/**
|
||
* git CLI helper.
|
||
*
|
||
* `configs` flags are emitted as `-c key=val` pairs BEFORE `-C repoPath` and
|
||
* BEFORE the subcommand. `core.quotepath=false` is always emitted first so CJK
|
||
* (and other non-ASCII) paths arrive as UTF-8 in `diff --name-status` and
|
||
* sibling commands. Callers that need additional git config should pass via
|
||
* the `configs` parameter; never inline `-c` into `args`.
|
||
*
|
||
* Exported for `test/sync.test.ts` invariant assertion only.
|
||
*/
|
||
export function buildGitInvocation(repoPath: string, args: string[], configs: string[] = []): string[] {
|
||
const cfg = ['core.quotepath=false', ...configs].flatMap(c => ['-c', c]);
|
||
return [...cfg, '-C', repoPath, ...args];
|
||
}
|
||
|
||
export function buildAutoEmbedArgs(slugs: string[], sourceId?: string): string[] {
|
||
return sourceId ? ['--source', sourceId, '--slugs', ...slugs] : ['--slugs', ...slugs];
|
||
}
|
||
|
||
/**
|
||
* Shell out to git with a generous maxBuffer.
|
||
*
|
||
* Node's default maxBuffer is 1 MiB. `git diff --name-status -M` on a
|
||
* 60–100K file repo easily exceeds that, causing an ENOBUFS crash that
|
||
* kills the sync process with no error message in the log.
|
||
*
|
||
* 100 MiB is generous but still bounded — a 100K-file diff with long
|
||
* paths tops out around 10–20 MiB in practice.
|
||
*/
|
||
function git(repoPath: string, args: string[], configs: string[] = []): string {
|
||
return execFileSync('git', buildGitInvocation(repoPath, args, configs), {
|
||
encoding: 'utf-8',
|
||
timeout: 30000,
|
||
maxBuffer: 100 * 1024 * 1024,
|
||
}).trim();
|
||
}
|
||
|
||
function hasOriginRemote(repoPath: string): boolean {
|
||
try {
|
||
execFileSync('git', buildGitInvocation(repoPath, ['remote', 'get-url', 'origin']), {
|
||
encoding: 'utf-8',
|
||
timeout: 30000,
|
||
stdio: ['ignore', 'ignore', 'ignore'],
|
||
});
|
||
return true;
|
||
} catch {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
function isDetachedHead(repoPath: string): boolean {
|
||
try {
|
||
git(repoPath, ['symbolic-ref', '--quiet', 'HEAD']);
|
||
return false;
|
||
} catch {
|
||
return true;
|
||
}
|
||
}
|
||
|
||
function unique<T>(items: T[]): T[] {
|
||
return [...new Set(items)];
|
||
}
|
||
|
||
function buildDetachedWorkingTreeManifest(repoPath: string): SyncManifest {
|
||
const manifest = buildSyncManifest(git(repoPath, ['diff', '--name-status', '-M', 'HEAD']));
|
||
const untracked = git(repoPath, ['ls-files', '--others', '--exclude-standard'])
|
||
.split('\n')
|
||
.filter(line => line.length > 0);
|
||
|
||
return {
|
||
added: unique([...manifest.added, ...untracked]),
|
||
modified: unique(manifest.modified),
|
||
deleted: unique(manifest.deleted),
|
||
renamed: manifest.renamed,
|
||
};
|
||
}
|
||
|
||
// v0.18.0 Step 5: source-scoped sync state helpers. When opts.sourceId
|
||
// is set, read/write the per-source row instead of the global config
|
||
// keys. These wrappers centralize the branch so every read/write site
|
||
// picks the right storage — future Step 5 work (failure-tracking per
|
||
// source) hooks here too.
|
||
async function readSyncAnchor(
|
||
engine: BrainEngine,
|
||
sourceId: string | undefined,
|
||
which: 'repo_path' | 'last_commit',
|
||
): Promise<string | null> {
|
||
if (sourceId) {
|
||
const col = which === 'repo_path' ? 'local_path' : 'last_commit';
|
||
const rows = await engine.executeRaw<Record<string, string | null>>(
|
||
`SELECT ${col} AS value FROM sources WHERE id = $1`,
|
||
[sourceId],
|
||
);
|
||
return rows[0]?.value ?? null;
|
||
}
|
||
return await engine.getConfig(`sync.${which}`);
|
||
}
|
||
|
||
async function writeSyncAnchor(
|
||
engine: BrainEngine,
|
||
sourceId: string | undefined,
|
||
which: 'repo_path' | 'last_commit',
|
||
value: string,
|
||
): Promise<void> {
|
||
if (sourceId) {
|
||
const col = which === 'repo_path' ? 'local_path' : 'last_commit';
|
||
// last_sync_at bookmarked on every last_commit advance.
|
||
if (which === 'last_commit') {
|
||
await engine.executeRaw(
|
||
`UPDATE sources SET last_commit = $1, last_sync_at = now() WHERE id = $2`,
|
||
[value, sourceId],
|
||
);
|
||
} else {
|
||
await engine.executeRaw(
|
||
`UPDATE sources SET ${col} = $1 WHERE id = $2`,
|
||
[value, sourceId],
|
||
);
|
||
}
|
||
return;
|
||
}
|
||
await engine.setConfig(`sync.${which}`, value);
|
||
}
|
||
|
||
/**
|
||
* v0.20.0 Cathedral II Layer 12 (SP-1 fix) — read/write the chunker version
|
||
* last used to sync a given source. When it mismatches CURRENT_CHUNKER_VERSION,
|
||
* `performSync` forces a full walk regardless of git HEAD equality. Without
|
||
* this gate, bumping CHUNKER_VERSION does NOTHING on an unchanged repo
|
||
* because sync short-circuits at `up_to_date` before reaching
|
||
* `importCodeFile`'s content_hash check.
|
||
*
|
||
* Per-source storage matches writeSyncAnchor's shape — sources.chunker_version
|
||
* TEXT column from the v27 migration. No global fallback: non-source syncs
|
||
* (pre-v0.17 brains with no sources table) never had CHUNKER_VERSION
|
||
* version-gating, so they keep the v0.19.0 behavior.
|
||
*/
|
||
async function readChunkerVersion(
|
||
engine: BrainEngine,
|
||
sourceId: string | undefined,
|
||
): Promise<string | null> {
|
||
if (!sourceId) return null;
|
||
const rows = await engine.executeRaw<{ chunker_version: string | null }>(
|
||
`SELECT chunker_version FROM sources WHERE id = $1`,
|
||
[sourceId],
|
||
);
|
||
return rows[0]?.chunker_version ?? null;
|
||
}
|
||
|
||
async function writeChunkerVersion(
|
||
engine: BrainEngine,
|
||
sourceId: string | undefined,
|
||
version: string,
|
||
): Promise<void> {
|
||
if (!sourceId) return;
|
||
await engine.executeRaw(
|
||
`UPDATE sources SET chunker_version = $1 WHERE id = $2`,
|
||
[version, sourceId],
|
||
);
|
||
}
|
||
|
||
/**
|
||
* v0.40 Federated Sync v2: `gbrain sync trigger --source <id> [--priority high|normal|low]`
|
||
*
|
||
* Push-trigger entry point. Wraps `queue.add('sync', ...)` with priority -10
|
||
* (above autopilot's 0) so push-triggered syncs preempt scheduled ones.
|
||
* Use cases: GitHub webhook handler (POST /webhooks/github), CLI nudge after
|
||
* a manual git pull, scripted dispatch from `gbrain sources federate`.
|
||
*
|
||
* Sets `auto_embed_backfill: true` so the extended sync handler (T6/T7)
|
||
* auto-enqueues an embed-backfill job after the sync settles.
|
||
*
|
||
* Output: prints `job_id=N` to stdout for shell composition. Errors exit 1.
|
||
*/
|
||
export async function runSyncTrigger(engine: BrainEngine, args: string[]): Promise<void> {
|
||
if (args.includes('--help') || args.includes('-h')) {
|
||
console.log(`Usage: gbrain sync trigger --source <id> [--priority high|normal|low]
|
||
|
||
Queue a push-triggered sync job for one source. Prints the resulting job id
|
||
on stdout. The autopilot worker picks it up and runs performSync against the
|
||
named source; if the sync added/modified pages, an embed-backfill job is
|
||
auto-enqueued (subject to D6 budget cap + D19 source-level cooldown).
|
||
|
||
Use cases:
|
||
- GitHub webhook → 'gbrain sync trigger --source <repo>'
|
||
- Manual nudge after 'git pull' inside a federated source
|
||
- Programmatic triggers from CI / shell automation
|
||
|
||
See also:
|
||
gbrain sources webhook set <id> Set up GitHub-signed push webhook
|
||
gbrain sources status Per-source sync + embed coverage
|
||
`);
|
||
return;
|
||
}
|
||
|
||
const sourceIdArg = args.find((a, i) => args[i - 1] === '--source') ?? null;
|
||
if (!sourceIdArg) {
|
||
console.error('Error: --source <id> is required');
|
||
console.error("Usage: gbrain sync trigger --source <id> [--priority high|normal|low]");
|
||
process.exit(2);
|
||
}
|
||
|
||
const priorityArg = args.find((a, i) => args[i - 1] === '--priority') ?? 'high';
|
||
const priorityMap: Record<string, number> = { high: -10, normal: 0, low: 5 };
|
||
const priority = priorityMap[priorityArg];
|
||
if (priority === undefined) {
|
||
console.error(`Invalid --priority value: "${priorityArg}". Must be high|normal|low.`);
|
||
process.exit(2);
|
||
}
|
||
|
||
// Verify source exists before submitting
|
||
const { fetchSource } = await import('../core/sources-load.ts');
|
||
const source = await fetchSource(engine, sourceIdArg);
|
||
if (!source) {
|
||
console.error(`Source "${sourceIdArg}" not found. List with: gbrain sources list`);
|
||
process.exit(1);
|
||
}
|
||
|
||
const { MinionQueue } = await import('../core/minions/queue.ts');
|
||
const queue = new MinionQueue(engine);
|
||
const job = await queue.add(
|
||
'sync',
|
||
{
|
||
sourceId: sourceIdArg,
|
||
repoPath: source.local_path,
|
||
auto_embed_backfill: true,
|
||
embed_reason: 'sync_trigger',
|
||
},
|
||
{
|
||
priority,
|
||
idempotency_key: `sync-trigger:${sourceIdArg}:${Math.floor(Date.now() / 30_000)}`,
|
||
maxWaiting: 1,
|
||
},
|
||
);
|
||
|
||
console.log(`job_id=${job.id}`);
|
||
}
|
||
|
||
export async function performSync(engine: BrainEngine, opts: SyncOpts): Promise<SyncResult> {
|
||
// v0.22.13 CODEX-2: cross-process writer lock prevents two concurrent
|
||
// syncs from racing on the same last_commit anchor (last writer wins,
|
||
// bookmark regresses, silent corruption).
|
||
//
|
||
// v0.40.5.0: per-source DB lock via `syncLockId(sourceId)`. Two sources
|
||
// (default + zion-brain) take distinct lock rows and don't serialize.
|
||
// SYNC_LOCK_ID is now a back-compat alias for syncLockId('default').
|
||
//
|
||
// v0.40.6.0 (D11 from PR #1314 review): pair the per-source lock with
|
||
// `withRefreshingLock` so long-running sources (media-corpus, 250K+
|
||
// chunks) don't lose their lock at the 30-minute TTL mid-run. Closes
|
||
// the bug class where a >30min sync could let a parallel acquire steal
|
||
// the lock and race on the final commit + bookmark write.
|
||
//
|
||
// skipLock is reserved for callers that already serialize via another
|
||
// mechanism (e.g. cycle.ts holds gbrain-cycle for the broader scope).
|
||
if (opts.skipLock) {
|
||
return await performSyncInner(engine, opts);
|
||
}
|
||
|
||
const lockKey = opts.lockId ?? syncLockId(opts.sourceId ?? 'default');
|
||
|
||
// When `opts.sourceId` is set OR `opts.lockId` is explicitly overridden,
|
||
// use the TTL-refreshing lock so long sources stay safe. The default
|
||
// path (no sourceId, no lockId) keeps the bare tryAcquireDbLock for
|
||
// bit-for-bit back-compat with single-default-source brains.
|
||
const usePerSourcePath = opts.lockId !== undefined || opts.sourceId !== undefined;
|
||
|
||
if (usePerSourcePath) {
|
||
try {
|
||
return await withRefreshingLock(engine, lockKey, () => performSyncInner(engine, opts));
|
||
} catch (err) {
|
||
if (err instanceof LockUnavailableError) {
|
||
throw new Error(await formatLockBusyMessage(engine, lockKey));
|
||
}
|
||
throw err;
|
||
}
|
||
}
|
||
|
||
// Legacy global-lock path (single-default-source brains).
|
||
const lockHandle = await tryAcquireDbLock(engine, lockKey);
|
||
if (!lockHandle) {
|
||
throw new Error(await formatLockBusyMessage(engine, lockKey));
|
||
}
|
||
try {
|
||
return await performSyncInner(engine, opts);
|
||
} finally {
|
||
try { await lockHandle.release(); } catch { /* best-effort release */ }
|
||
}
|
||
}
|
||
|
||
/**
|
||
* v0.41.6.0 D3: rich "Another sync is in progress" message that names the
|
||
* holder PID, hostname, age, and the right --break-lock invocation to
|
||
* recover. Falls back to the legacy message when inspectLock can't read
|
||
* the row (best-effort — the lock itself was still busy).
|
||
*/
|
||
async function formatLockBusyMessage(engine: BrainEngine, lockKey: string): Promise<string> {
|
||
const { inspectLock } = await import('../core/db-lock.ts');
|
||
let snap;
|
||
try { snap = await inspectLock(engine, lockKey); }
|
||
catch { snap = null; }
|
||
|
||
if (!snap) {
|
||
return (
|
||
`Another sync is in progress (lock ${lockKey} held). ` +
|
||
`Wait for it to finish, or run 'gbrain doctor' if it has been more than 30 minutes.`
|
||
);
|
||
}
|
||
|
||
const ageHuman = formatAgeHuman(snap.age_ms);
|
||
const breakHint = lockKey.startsWith('gbrain-sync:')
|
||
? `gbrain sync --break-lock --source ${lockKey.slice('gbrain-sync:'.length)}`
|
||
: `gbrain sync --break-lock`;
|
||
const ttlNote = snap.ttl_expired ? ' [TTL expired]' : '';
|
||
return (
|
||
`Another sync is in progress (lock ${lockKey} held by pid ${snap.holder_pid} on ${snap.holder_host}, ` +
|
||
`started ${ageHuman} ago${ttlNote}).\n` +
|
||
`If pid ${snap.holder_pid} is dead, re-run with --break-lock to clear it:\n` +
|
||
` ${breakHint}\n` +
|
||
`Or wait for the holder to finish.`
|
||
);
|
||
}
|
||
|
||
/**
|
||
* v0.41.6.0 D3: `gbrain sync --break-lock` / `--force-break-lock` worker.
|
||
* Returns the process exit code (0 = lock cleared or absent; 1 = refused).
|
||
*
|
||
* Safe path (`force=false`): refuses unless the holder is on this host
|
||
* AND either (a) TTL has expired (the lock is structurally available
|
||
* already) OR (b) the holder PID is dead AND the lock is older than 60s
|
||
* (the age guard defeats PID-reuse coincidence — Linux PID space wraps
|
||
* at 32768 so a 10-day-old lock with pid=12345 may be falsely
|
||
* refused-to-clear because an unrelated process now owns pid 12345; 60s
|
||
* is the codex F7-amended minimum age that makes coincidence unlikely).
|
||
*
|
||
* Force path (`force=true`): skips liveness check, deletes the row,
|
||
* warns loudly that the holder may still be writing.
|
||
*
|
||
* Both paths use the same atomic `DELETE ... RETURNING id` so a race
|
||
* with another break-lock or with TTL-eviction can't produce confusing
|
||
* post-conditions.
|
||
*/
|
||
async function runBreakLock(
|
||
engine: BrainEngine,
|
||
lockKey: string,
|
||
sourceId: string,
|
||
opts: { force: boolean; json: boolean },
|
||
): Promise<number> {
|
||
const { inspectLock, deleteLockRow } = await import('../core/db-lock.ts');
|
||
const { hostname } = await import('os');
|
||
const localHost = hostname();
|
||
let snap;
|
||
try { snap = await inspectLock(engine, lockKey); }
|
||
catch (e) {
|
||
const msg = e instanceof Error ? e.message : String(e);
|
||
if (opts.json) console.log(JSON.stringify({ status: 'error', error: msg, lock: lockKey }));
|
||
else console.error(`Failed to inspect lock ${lockKey}: ${msg}`);
|
||
return 1;
|
||
}
|
||
|
||
if (!snap) {
|
||
if (opts.json) console.log(JSON.stringify({ status: 'absent', lock: lockKey, source_id: sourceId }));
|
||
else console.log(`Lock ${lockKey} is not held (nothing to break).`);
|
||
return 0;
|
||
}
|
||
|
||
// Force path: skip all guards, atomic DELETE, warn.
|
||
if (opts.force) {
|
||
const { deleted } = await deleteLockRow(engine, lockKey, snap.holder_pid);
|
||
if (opts.json) {
|
||
console.log(JSON.stringify({
|
||
status: deleted ? 'force_broken' : 'race_already_cleared',
|
||
lock: lockKey, source_id: sourceId, snapshot: snap,
|
||
}));
|
||
} else if (deleted) {
|
||
console.log(`Force-broke lock ${lockKey} (was held by pid ${snap.holder_pid} on ${snap.holder_host}, age ${formatAgeHuman(snap.age_ms)}).`);
|
||
console.log('WARNING: the holder may still be writing. Verify with `gbrain doctor` before re-running.');
|
||
} else {
|
||
console.log(`Lock ${lockKey} was already cleared by another process between our check and DELETE (race-safe).`);
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
// Safe path: must be local host AND (TTL-expired OR (PID-dead AND age >= 60s)).
|
||
if (snap.holder_host !== localHost) {
|
||
if (opts.json) {
|
||
console.log(JSON.stringify({
|
||
status: 'refused',
|
||
reason: 'cross_host',
|
||
lock: lockKey, source_id: sourceId, snapshot: snap, local_host: localHost,
|
||
}));
|
||
} else {
|
||
console.error(`Lock ${lockKey} is held on a different host (${snap.holder_host}, this host is ${localHost}).`);
|
||
console.error('Cross-host PID liveness is unsound. To break anyway, use --force-break-lock');
|
||
console.error('(only safe when you KNOW the holder is dead — verify before forcing).');
|
||
}
|
||
return 1;
|
||
}
|
||
|
||
let safe = false;
|
||
let reason: string;
|
||
if (snap.ttl_expired) {
|
||
safe = true;
|
||
reason = 'ttl_expired';
|
||
} else {
|
||
// PID liveness check on local host. process.kill(pid, 0) throws ESRCH
|
||
// when the PID is dead. Combined with 60s age guard (per outside-voice F7).
|
||
let alive = true;
|
||
try { process.kill(snap.holder_pid, 0); }
|
||
catch { alive = false; }
|
||
const oldEnough = snap.age_ms >= 60_000;
|
||
if (!alive && oldEnough) {
|
||
safe = true;
|
||
reason = 'pid_dead_age_60s';
|
||
} else if (!alive && !oldEnough) {
|
||
reason = 'pid_dead_but_lock_too_young';
|
||
} else {
|
||
reason = 'pid_alive';
|
||
}
|
||
}
|
||
|
||
if (!safe) {
|
||
if (opts.json) {
|
||
console.log(JSON.stringify({
|
||
status: 'refused', reason, lock: lockKey, source_id: sourceId, snapshot: snap,
|
||
}));
|
||
} else {
|
||
console.error(`Refusing to break lock ${lockKey}: holder pid ${snap.holder_pid} appears alive on ${snap.holder_host} (age ${formatAgeHuman(snap.age_ms)}).`);
|
||
if (reason === 'pid_dead_but_lock_too_young') {
|
||
console.error('(PID is dead but the lock is younger than 60s — the PID may have been reused. Wait or use --force-break-lock if you are certain.)');
|
||
} else {
|
||
console.error('If the holder is wedged, kill it first then re-run --break-lock,');
|
||
console.error('OR use --force-break-lock to clear regardless (the holder may still write afterwards).');
|
||
}
|
||
}
|
||
return 1;
|
||
}
|
||
|
||
const { deleted } = await deleteLockRow(engine, lockKey, snap.holder_pid);
|
||
if (opts.json) {
|
||
console.log(JSON.stringify({
|
||
status: deleted ? 'broken' : 'race_already_cleared',
|
||
reason, lock: lockKey, source_id: sourceId, snapshot: snap,
|
||
}));
|
||
} else if (deleted) {
|
||
console.log(`Broke lock ${lockKey} (was held by pid ${snap.holder_pid} on ${snap.holder_host}, age ${formatAgeHuman(snap.age_ms)}; reason: ${reason}).`);
|
||
} else {
|
||
console.log(`Lock ${lockKey} was already cleared by another process between our check and DELETE (race-safe).`);
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
function formatAgeHuman(ms: number): string {
|
||
if (ms < 1000) return `${ms}ms`;
|
||
const s = Math.floor(ms / 1000);
|
||
if (s < 60) return `${s}s`;
|
||
const m = Math.floor(s / 60);
|
||
if (m < 60) return `${m}m${s % 60}s`;
|
||
const h = Math.floor(m / 60);
|
||
if (h < 24) return `${h}h${m % 60}m`;
|
||
const d = Math.floor(h / 24);
|
||
return `${d}d${h % 24}h`;
|
||
}
|
||
|
||
async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<SyncResult> {
|
||
// v0.41.8.0 (D9 / #1342): phase breadcrumbs. The #1342 reporter saw
|
||
// ZERO stderr output before their sync hang, which made the bug
|
||
// impossible to triage. Mirror the existing `[gbrain phase] sync.git_pull`
|
||
// pattern at the major phase boundaries so the next #1342-shaped
|
||
// report names WHICH phase spun. Doesn't fix #1342 but converts
|
||
// "hung with no output" into actionable diagnostic data.
|
||
serr(`[gbrain phase] sync.resolve_repo`);
|
||
// Resolve repo path
|
||
const repoPath = opts.repoPath || await readSyncAnchor(engine, opts.sourceId, 'repo_path');
|
||
if (!repoPath) {
|
||
const hint = opts.sourceId
|
||
? `Source "${opts.sourceId}" has no local_path. Run: gbrain sources add ${opts.sourceId} --path <path>`
|
||
: `No repo path specified. Use --repo or run gbrain init with --repo first.`;
|
||
throw new Error(hint);
|
||
}
|
||
|
||
serr(`[gbrain phase] sync.load_active_pack`);
|
||
// v0.39 T1.5: load active pack ONCE at sync entry; pass to every per-file
|
||
// importFile call below. Codex perf finding #7: per-file loadActivePack adds
|
||
// disk/YAML/hash overhead × thousands of files. Best-effort: pack load
|
||
// failure falls through to legacy inferType (parity preserved).
|
||
let syncActivePack: { page_types: ReadonlyArray<{ name: string; path_prefixes: ReadonlyArray<string> }> } | undefined;
|
||
try {
|
||
const { loadActivePack } = await import('../core/schema-pack/load-active.ts');
|
||
const { loadConfig } = await import('../core/config.ts');
|
||
const resolved = await loadActivePack({
|
||
cfg: loadConfig(),
|
||
remote: false, // sync is always a trusted CLI / autopilot caller
|
||
sourceId: opts.sourceId,
|
||
});
|
||
syncActivePack = { page_types: resolved.manifest.page_types };
|
||
} catch {
|
||
syncActivePack = undefined;
|
||
}
|
||
|
||
// v0.28: source-aware re-clone branch. When the source has a remote_url
|
||
// recorded (i.e. it was registered via `sources add --url`), the on-disk
|
||
// clone is auto-managed. validateRepoState classifies the on-disk state;
|
||
// we recover from missing/no-git/not-a-dir by re-cloning, refuse on
|
||
// url-drift or corruption with structured hints.
|
||
if (opts.sourceId) {
|
||
serr(`[gbrain phase] sync.validate_repo_state`);
|
||
const { validateRepoState } = await import('../core/git-remote.ts');
|
||
const { recloneIfMissing } = await import('../core/sources-ops.ts');
|
||
const cfgRows = await engine.executeRaw<{ config: unknown }>(
|
||
`SELECT config FROM sources WHERE id = $1`,
|
||
[opts.sourceId],
|
||
);
|
||
const cfg =
|
||
typeof cfgRows[0]?.config === 'string'
|
||
? (JSON.parse(cfgRows[0].config as string) as Record<string, unknown>)
|
||
: ((cfgRows[0]?.config ?? {}) as Record<string, unknown>);
|
||
const remoteUrl = typeof cfg.remote_url === 'string' ? cfg.remote_url : null;
|
||
if (remoteUrl) {
|
||
const state = validateRepoState(repoPath, remoteUrl);
|
||
switch (state) {
|
||
case 'healthy':
|
||
break;
|
||
case 'missing':
|
||
case 'no-git':
|
||
case 'not-a-dir':
|
||
serr(
|
||
`[gbrain] auto-recovery: re-cloning "${opts.sourceId}" (clone state: ${state}).`,
|
||
);
|
||
await recloneIfMissing(engine, opts.sourceId);
|
||
break;
|
||
case 'corrupted':
|
||
throw new Error(
|
||
`Source "${opts.sourceId}" clone at ${repoPath} is corrupted ` +
|
||
`(\`git remote get-url origin\` failed). Run: ` +
|
||
`gbrain sources remove ${opts.sourceId} --confirm-destructive && ` +
|
||
`gbrain sources add ${opts.sourceId} --url ${remoteUrl}`,
|
||
);
|
||
case 'url-drift':
|
||
throw new Error(
|
||
`Source "${opts.sourceId}" clone at ${repoPath} has a remote ` +
|
||
`that differs from config.remote_url=${remoteUrl}. ` +
|
||
`Re-clone with: gbrain sources rebase-clone ${opts.sourceId} ` +
|
||
`(if available, else: sources remove + sources add).`,
|
||
);
|
||
}
|
||
}
|
||
}
|
||
|
||
// Validate git repo
|
||
if (!existsSync(join(repoPath, '.git'))) {
|
||
throw new Error(`Not a git repository: ${repoPath}. GBrain sync requires a git-initialized repo.`);
|
||
}
|
||
|
||
serr(`[gbrain phase] sync.detect_head`);
|
||
// Detect detached HEAD up front so the working-tree fallback fires for both
|
||
// the default sync and `--no-pull` callers. Only the actual git pull is
|
||
// gated on opts.noPull.
|
||
const detachedHead = isDetachedHead(repoPath);
|
||
if (detachedHead && !opts.noPull) {
|
||
serr(`Detached HEAD on ${repoPath}; skipping git pull. Syncing from local working tree.`);
|
||
}
|
||
|
||
// Git pull (unless --no-pull). v0.28.1 codex finding (HIGH): the legacy
|
||
// git() helper at sync.ts:192 spawns git without GIT_SSRF_FLAGS, so
|
||
// every steady-state pull was bypassing the redirect/submodule/protocol
|
||
// hardening that cloneRepo applies. Route through pullRepo from
|
||
// git-remote.ts so the flag set is consistent across initial clone and
|
||
// ongoing pulls — single source of truth for the defensive flags.
|
||
const originRemotePresent = !opts.noPull && !detachedHead ? hasOriginRemote(repoPath) : false;
|
||
if (!opts.noPull && !detachedHead && !originRemotePresent) {
|
||
serr(`No origin remote on ${repoPath}; skipping git pull. Syncing from local working tree.`);
|
||
}
|
||
|
||
if (!opts.noPull && !detachedHead && originRemotePresent) {
|
||
const _t0 = Date.now();
|
||
serr(`[gbrain phase] sync.git_pull start`);
|
||
try {
|
||
const { pullRepo } = await import('../core/git-remote.ts');
|
||
pullRepo(repoPath);
|
||
serr(`[gbrain phase] sync.git_pull done ${Date.now() - _t0}ms`);
|
||
} catch (e: unknown) {
|
||
const msg = e instanceof Error ? e.message : String(e);
|
||
serr(`[gbrain phase] sync.git_pull error ${Date.now() - _t0}ms (${msg.slice(0, 80)})`);
|
||
if (msg.includes('non-fast-forward') || msg.includes('diverged')) {
|
||
serr(`Warning: git pull failed (remote diverged). Syncing from local state.`);
|
||
} else {
|
||
serr(`Warning: git pull failed: ${msg.slice(0, 100)}`);
|
||
}
|
||
}
|
||
}
|
||
|
||
// Get current HEAD
|
||
let headCommit: string;
|
||
try {
|
||
headCommit = git(repoPath, ['rev-parse', 'HEAD']);
|
||
} catch {
|
||
throw new Error(`No commits in repo ${repoPath}. Make at least one commit before syncing.`);
|
||
}
|
||
|
||
// Read sync state (source-scoped when sourceId is set, global otherwise)
|
||
const lastCommit = opts.full ? null : await readSyncAnchor(engine, opts.sourceId, 'last_commit');
|
||
|
||
// Ancestry validation: if lastCommit exists, verify it's still in history
|
||
if (lastCommit) {
|
||
try {
|
||
git(repoPath, ['cat-file', '-t', lastCommit]);
|
||
} catch {
|
||
serr(`Sync anchor commit ${lastCommit.slice(0, 8)} missing (force push?). Running full reimport.`);
|
||
return performFullSync(engine, repoPath, headCommit, opts);
|
||
}
|
||
|
||
// Verify ancestry
|
||
try {
|
||
git(repoPath, ['merge-base', '--is-ancestor', lastCommit, headCommit]);
|
||
} catch {
|
||
serr(`Sync anchor ${lastCommit.slice(0, 8)} is not an ancestor of HEAD. Running full reimport.`);
|
||
return performFullSync(engine, repoPath, headCommit, opts);
|
||
}
|
||
}
|
||
|
||
// First sync
|
||
if (!lastCommit) {
|
||
return performFullSync(engine, repoPath, headCommit, opts);
|
||
}
|
||
|
||
// v0.20.0 Cathedral II Layer 12 (codex SP-1 fix): before returning
|
||
// 'up_to_date' on git-HEAD equality, check the chunker version gate.
|
||
// If sources.chunker_version mismatches CURRENT_CHUNKER_VERSION, force
|
||
// a full re-walk so existing chunks get re-chunked under the new
|
||
// pipeline (qualified symbol names, parent scope, doc-comment column
|
||
// population, etc.). Without this, upgraded brains silently stay on
|
||
// the old chunks — the whole reason we bumped the version.
|
||
const storedVersion = await readChunkerVersion(engine, opts.sourceId);
|
||
const currentVersion = String(CHUNKER_VERSION);
|
||
const versionMismatch = storedVersion !== null && storedVersion !== currentVersion;
|
||
const versionNeverSet = storedVersion === null && opts.sourceId !== undefined;
|
||
const detachedWorkingTreeManifest = detachedHead ? buildDetachedWorkingTreeManifest(repoPath) : null;
|
||
const hasDetachedWorkingTreeChanges = detachedWorkingTreeManifest !== null &&
|
||
(detachedWorkingTreeManifest.added.length > 0 ||
|
||
detachedWorkingTreeManifest.modified.length > 0 ||
|
||
detachedWorkingTreeManifest.deleted.length > 0 ||
|
||
detachedWorkingTreeManifest.renamed.length > 0);
|
||
|
||
if (lastCommit === headCommit && !versionMismatch && !versionNeverSet && !hasDetachedWorkingTreeChanges) {
|
||
return {
|
||
status: 'up_to_date',
|
||
fromCommit: lastCommit,
|
||
toCommit: headCommit,
|
||
added: 0, modified: 0, deleted: 0, renamed: 0,
|
||
chunksCreated: 0,
|
||
embedded: 0,
|
||
pagesAffected: [],
|
||
};
|
||
}
|
||
|
||
if ((versionMismatch || versionNeverSet) && lastCommit === headCommit) {
|
||
slog(
|
||
`[sync] chunker_version gate: stored=${storedVersion ?? 'unset'}, current=${currentVersion}. ` +
|
||
`Forcing full re-chunk pass (git HEAD unchanged but pipeline version advanced).`,
|
||
);
|
||
const result = await performFullSync(engine, repoPath, headCommit, opts);
|
||
await writeChunkerVersion(engine, opts.sourceId, currentVersion);
|
||
return result;
|
||
}
|
||
|
||
// Diff using git diff (net result, not per-commit)
|
||
const diffOutput = git(repoPath, ['diff', '--name-status', '-M', `${lastCommit}..${headCommit}`]);
|
||
const manifest = buildSyncManifest(diffOutput);
|
||
if (detachedWorkingTreeManifest) {
|
||
manifest.added = unique([...manifest.added, ...detachedWorkingTreeManifest.added]);
|
||
manifest.modified = unique([...manifest.modified, ...detachedWorkingTreeManifest.modified]);
|
||
manifest.deleted = unique([...manifest.deleted, ...detachedWorkingTreeManifest.deleted]);
|
||
manifest.renamed = [...manifest.renamed, ...detachedWorkingTreeManifest.renamed];
|
||
}
|
||
|
||
// Filter to syncable files (strategy-aware)
|
||
const syncOpts = opts.strategy ? { strategy: opts.strategy } : undefined;
|
||
const filtered: SyncManifest = {
|
||
added: manifest.added.filter(p => isSyncable(p, syncOpts)),
|
||
modified: manifest.modified.filter(p => isSyncable(p, syncOpts)),
|
||
deleted: manifest.deleted.filter(p => isSyncable(p, syncOpts)),
|
||
renamed: manifest.renamed.filter(r => isSyncable(r.to, syncOpts)),
|
||
};
|
||
|
||
// Delete pages that became un-syncable (modified but filtered out).
|
||
// v0.20.0 Cathedral II SP-5: resolveSlugForPath picks the right slug shape
|
||
// (markdown vs code) based on the chunker's classifier, so a Rust file that
|
||
// became un-syncable (e.g., moved under `.gitignore` or filtered by
|
||
// strategy=markdown) deletes the actual code-slug page, not a ghost
|
||
// markdown-slug that never existed.
|
||
//
|
||
// v0.41.13 (#1433): the original cleanup loop deleted EVERY pre-existing
|
||
// page for unsyncable-modified paths, including `log.md`, `schema.md`,
|
||
// `index.md`, `README.md` — files that fail `isSyncable` precisely
|
||
// because they're metafiles by convention, not because the user
|
||
// "removed" them from the strategy. infiniteGameExp's domain `log.md`
|
||
// pages had been indexed by an older gbrain version (or via direct
|
||
// put_page) and were silently dropped on every subsequent sync. The
|
||
// fix uses `unsyncableReason` (factored from `isSyncable` so they
|
||
// cannot drift) to skip the delete when the reason is `'metafile'`.
|
||
//
|
||
// Honest scope: this guard only fixes the `manifest.modified` case.
|
||
// `manifest.deleted` is filtered upstream at sync.ts:757 via the same
|
||
// `isSyncable` call, so `rm log.md` followed by sync also doesn't
|
||
// delete the page. That's the same pre-fix behavior — removing the
|
||
// page requires `gbrain pages purge-deleted` or a direct MCP delete.
|
||
// Filed as v0.42+ follow-up for a `gbrain pages remove <slug>` surface.
|
||
const unsyncableModified = manifest.modified.filter(p => !isSyncable(p, syncOpts));
|
||
// v0.18.0+ multi-source: scope getPage + deletePage to opts.sourceId so
|
||
// unsyncable cleanup in source A doesn't accidentally sweep same-slug
|
||
// pages in sources B/C/D.
|
||
const pageOpts = opts.sourceId ? { sourceId: opts.sourceId } : undefined;
|
||
for (const path of unsyncableModified) {
|
||
// v0.41.13 #1433: never delete on metafile classification.
|
||
if (unsyncableReason(path, syncOpts) === 'metafile') continue;
|
||
const slug = await resolveSlugByPathOrSourcePath(engine, path, opts.sourceId);
|
||
try {
|
||
const existing = await engine.getPage(slug, pageOpts);
|
||
if (existing) {
|
||
await engine.deletePage(slug, pageOpts);
|
||
slog(` Deleted un-syncable page: ${slug}`);
|
||
}
|
||
} catch { /* ignore */ }
|
||
}
|
||
|
||
const totalChanges = filtered.added.length + filtered.modified.length +
|
||
filtered.deleted.length + filtered.renamed.length;
|
||
|
||
// Dry run
|
||
if (opts.dryRun) {
|
||
slog(`Sync dry run: ${lastCommit.slice(0, 8)}..${headCommit.slice(0, 8)}`);
|
||
if (filtered.added.length) slog(` Added: ${filtered.added.join(', ')}`);
|
||
if (filtered.modified.length) slog(` Modified: ${filtered.modified.join(', ')}`);
|
||
if (filtered.deleted.length) slog(` Deleted: ${filtered.deleted.join(', ')}`);
|
||
if (filtered.renamed.length) slog(` Renamed: ${filtered.renamed.map(r => `${r.from} -> ${r.to}`).join(', ')}`);
|
||
if (totalChanges === 0) slog(` No syncable changes.`);
|
||
return {
|
||
status: 'dry_run',
|
||
fromCommit: lastCommit,
|
||
toCommit: headCommit,
|
||
added: filtered.added.length,
|
||
modified: filtered.modified.length,
|
||
deleted: filtered.deleted.length,
|
||
renamed: filtered.renamed.length,
|
||
chunksCreated: 0,
|
||
embedded: 0,
|
||
pagesAffected: [],
|
||
};
|
||
}
|
||
|
||
if (totalChanges === 0) {
|
||
// Update sync state even with no syncable changes (git advanced)
|
||
await writeSyncAnchor(engine, opts.sourceId, 'last_commit', headCommit);
|
||
await engine.setConfig('sync.last_run', new Date().toISOString());
|
||
await writeChunkerVersion(engine, opts.sourceId, String(CHUNKER_VERSION));
|
||
return {
|
||
status: 'up_to_date',
|
||
fromCommit: lastCommit,
|
||
toCommit: headCommit,
|
||
added: 0, modified: 0, deleted: 0, renamed: 0,
|
||
chunksCreated: 0,
|
||
embedded: 0,
|
||
pagesAffected: [],
|
||
};
|
||
}
|
||
|
||
const noEmbed = opts.noEmbed || totalChanges > 100;
|
||
if (totalChanges > 100) {
|
||
slog(`Large sync (${totalChanges} files). Importing text, deferring embeddings.`);
|
||
}
|
||
|
||
const pagesAffected: string[] = [];
|
||
let chunksCreated = 0;
|
||
const start = Date.now();
|
||
|
||
// Per-file progress on stderr so agents see each step of a big sync.
|
||
// Phases: sync.deletes, sync.renames, sync.imports.
|
||
const progress = createProgress(cliOptsToProgressOptions(getCliOptions()));
|
||
|
||
// Process deletes first (prevents slug conflicts). SP-5: resolveSlugForPath
|
||
// dispatches to the right slug shape so code file deletes hit the real page.
|
||
// v0.18.0+ multi-source: scope deletePage so we only delete the source-A
|
||
// row, not every same-slug row across all sources.
|
||
const deleteOpts = opts.sourceId ? { sourceId: opts.sourceId } : undefined;
|
||
if (filtered.deleted.length > 0) {
|
||
progress.start('sync.deletes', filtered.deleted.length);
|
||
for (const path of filtered.deleted) {
|
||
const slug = await resolveSlugByPathOrSourcePath(engine, path, opts.sourceId);
|
||
await engine.deletePage(slug, deleteOpts);
|
||
pagesAffected.push(slug);
|
||
progress.tick(1, slug);
|
||
}
|
||
progress.finish();
|
||
}
|
||
|
||
// Process renames (updateSlug preserves page_id, chunks, embeddings).
|
||
// SP-5: both old and new slugs use resolveSlugForPath so a .ts → .ts
|
||
// rename (code→code), .md → .md (markdown→markdown), or cross-kind rename
|
||
// all resolve to the right slug shape for each side.
|
||
if (filtered.renamed.length > 0) {
|
||
progress.start('sync.renames', filtered.renamed.length);
|
||
// v0.18.0+ multi-source: scope updateSlug so the rename only touches the
|
||
// source-A row, not every same-slug row across sources (which would
|
||
// either sweep them all OR violate (source_id, slug) UNIQUE).
|
||
const renameOpts = opts.sourceId ? { sourceId: opts.sourceId } : undefined;
|
||
for (const { from, to } of filtered.renamed) {
|
||
const oldSlug = await resolveSlugByPathOrSourcePath(engine, from, opts.sourceId);
|
||
// The new path doesn't yet have a row, so resolve from path only.
|
||
const newSlug = resolveSlugForPath(to);
|
||
try {
|
||
await engine.updateSlug(oldSlug, newSlug, renameOpts);
|
||
} catch {
|
||
// Slug doesn't exist or collision, treat as add
|
||
}
|
||
// Reimport at new path (picks up content changes)
|
||
const filePath = join(repoPath, to);
|
||
if (existsSync(filePath)) {
|
||
const result = await importFile(engine, filePath, to, { noEmbed, sourceId: opts.sourceId, activePack: syncActivePack });
|
||
if (result.status === 'imported') chunksCreated += result.chunks;
|
||
}
|
||
pagesAffected.push(newSlug);
|
||
progress.tick(1, newSlug);
|
||
}
|
||
progress.finish();
|
||
}
|
||
|
||
// Process adds and modifies.
|
||
//
|
||
// NOTE: do NOT wrap this loop in engine.transaction(). importFromContent
|
||
// already opens its own inner transaction per file, and PGLite transactions
|
||
// are not reentrant — they acquire the same _runExclusiveTransaction mutex,
|
||
// so a nested call from inside a user callback queues forever on the mutex
|
||
// the outer transaction is still holding. Result: incremental sync hangs in
|
||
// ep_poll whenever the diff crosses the old > 10 threshold that used to
|
||
// trigger the outer wrap. Per-file atomicity is also the right granularity:
|
||
// one file's failure should not roll back the others' successful imports.
|
||
//
|
||
// v0.15.2: per-file progress on stderr via the shared reporter.
|
||
// Bug 9: per-file failures captured in `failedFiles` so the caller can
|
||
// gate `sync.last_commit` advancement and record recoverable errors.
|
||
const failedFiles: Array<{ path: string; error: string; line?: number }> = [];
|
||
const addsAndMods = [...filtered.added, ...filtered.modified];
|
||
|
||
// Sort newest-first so date-prefixed brain paths get embedded before older
|
||
// ones. See src/core/sort-newest-first.ts for the policy.
|
||
sortNewestFirst(addsAndMods);
|
||
|
||
// v0.22.13 (PR #490 Q5): one source of truth for the concurrency decision.
|
||
// engine.kind === 'pglite' → forced 1; explicit opts.concurrency wins;
|
||
// auto path returns DEFAULT_PARALLEL_WORKERS only when fileCount > 100.
|
||
const explicitConcurrency = opts.concurrency !== undefined;
|
||
const effectiveConcurrency = autoConcurrency(engine, addsAndMods.length, opts.concurrency);
|
||
const runParallel = shouldRunParallel(effectiveConcurrency, addsAndMods.length, explicitConcurrency);
|
||
|
||
if (addsAndMods.length > 0) {
|
||
progress.start('sync.imports', addsAndMods.length);
|
||
|
||
// Core import logic shared by serial and parallel paths.
|
||
// repoPath is validated non-null at the top of performSyncInner; narrow for TS.
|
||
const syncRepoPath = repoPath!;
|
||
async function importOnePath(eng: BrainEngine, path: string): Promise<void> {
|
||
const filePath = join(syncRepoPath, path);
|
||
if (!existsSync(filePath)) {
|
||
// CODEX-3 (v0.22.13): a file the diff said exists at headCommit but
|
||
// is gone from disk means the working tree has drifted (someone ran
|
||
// `git checkout` / `git reset` mid-sync, or the file was deleted
|
||
// post-diff). Record as a failure so last_commit does NOT advance —
|
||
// the silent-skip-then-advance pathology was the bug.
|
||
failedFiles.push({
|
||
path,
|
||
error: 'file vanished mid-sync (working tree drifted from headCommit)',
|
||
});
|
||
progress.tick(1, `skip:${path}`);
|
||
return;
|
||
}
|
||
try {
|
||
// v0.18.0+ multi-source: thread `opts.sourceId` so per-page tx writes
|
||
// (putPage / getTags / addTag / removeTag / deleteChunks / upsertChunks
|
||
// / addLink) target (sourceId, slug). Pre-fix the schema DEFAULT
|
||
// 'default' was applied even for non-default sources, fabricating
|
||
// duplicate rows that crashed bare-slug subqueries with Postgres 21000.
|
||
const result = await importFile(eng, filePath, path, { noEmbed, sourceId: opts.sourceId, activePack: syncActivePack });
|
||
if (result.status === 'imported') {
|
||
chunksCreated += result.chunks;
|
||
pagesAffected.push(result.slug);
|
||
} else if (result.status === 'skipped' && (result as any).error) {
|
||
failedFiles.push({ path, error: String((result as any).error) });
|
||
}
|
||
} catch (e: unknown) {
|
||
const msg = e instanceof Error ? e.message : String(e);
|
||
serr(` Warning: skipped ${path}: ${msg}`);
|
||
failedFiles.push({ path, error: msg });
|
||
}
|
||
progress.tick(1, path);
|
||
}
|
||
|
||
if (runParallel) {
|
||
// A1 (v0.22.13): use engine.kind discriminator instead of config?.engine
|
||
// string compare or constructor.name sniff. Q3: belt-and-suspenders fall
|
||
// back to serial when database_url is unset, so we never crash on a null
|
||
// assertion if config is missing.
|
||
const config = loadConfig();
|
||
if (engine.kind === 'pglite' || !config?.database_url) {
|
||
for (const path of addsAndMods) {
|
||
await importOnePath(engine, path);
|
||
}
|
||
} else {
|
||
const { PostgresEngine } = await import('../core/postgres-engine.ts');
|
||
const { resolvePoolSize } = await import('../core/db.ts');
|
||
const workerPoolSize = Math.min(2, resolvePoolSize(2));
|
||
const workerCount = Math.min(effectiveConcurrency, addsAndMods.length);
|
||
const databaseUrl = config.database_url;
|
||
|
||
// Q4 (v0.22.13): banner on stderr so stdout stays clean for --json.
|
||
serr(` Parallel sync: ${workerCount} workers for ${addsAndMods.length} files`);
|
||
|
||
const workerEngines: InstanceType<typeof PostgresEngine>[] = [];
|
||
try {
|
||
// Connect workers one-by-one rather than Promise.all so a partial
|
||
// failure leaves us with the connected ones in workerEngines for
|
||
// the finally-block cleanup. The original code lost track of
|
||
// already-connected engines on any one failure.
|
||
for (let i = 0; i < workerCount; i++) {
|
||
const eng = new PostgresEngine();
|
||
await eng.connect({ database_url: databaseUrl, poolSize: workerPoolSize });
|
||
workerEngines.push(eng);
|
||
}
|
||
|
||
// Atomic queue index — JS is single-threaded; the read-then-increment
|
||
// happens between awaits, so no lock is needed.
|
||
let queueIndex = 0;
|
||
await Promise.all(
|
||
workerEngines.map(async (eng) => {
|
||
while (true) {
|
||
const idx = queueIndex++;
|
||
if (idx >= addsAndMods.length) break;
|
||
await importOnePath(eng, addsAndMods[idx]);
|
||
}
|
||
}),
|
||
);
|
||
} finally {
|
||
// A2 (v0.22.13): try/finally guarantees connection cleanup even when
|
||
// the worker loop throws (partial connect failure, OOM, mid-import
|
||
// signal). Each disconnect is best-effort — one worker failing to
|
||
// disconnect must not strand the others.
|
||
await Promise.all(
|
||
workerEngines.map((e) =>
|
||
e.disconnect().catch((err: unknown) =>
|
||
serr(` worker disconnect failed: ${err instanceof Error ? err.message : String(err)}`),
|
||
),
|
||
),
|
||
);
|
||
}
|
||
}
|
||
} else {
|
||
// Serial path (small auto diffs or explicit --workers 1).
|
||
for (const path of addsAndMods) {
|
||
await importOnePath(engine, path);
|
||
}
|
||
}
|
||
|
||
progress.finish();
|
||
}
|
||
|
||
// CODEX-3 (v0.22.13): head-drift gate. If git HEAD moved during the import
|
||
// window (someone ran `git checkout` or `git pull` in another terminal /
|
||
// sibling Conductor workspace), the chunks we just imported reflect a
|
||
// different tree than `headCommit` claims. Refuse to advance last_commit
|
||
// so the next sync re-walks against the new HEAD. The lock from CODEX-2
|
||
// prevents *this* gbrain process from stepping on itself; this gate
|
||
// catches drift caused by external `git` commands the lock cannot see.
|
||
try {
|
||
const currentHead = git(repoPath, ['rev-parse', 'HEAD']);
|
||
if (currentHead !== headCommit) {
|
||
failedFiles.push({
|
||
path: '<head>',
|
||
error: `git HEAD drifted during sync: captured ${headCommit.slice(0, 8)}, now ${currentHead.slice(0, 8)}`,
|
||
});
|
||
}
|
||
} catch (e) {
|
||
// rev-parse failure is itself a drift signal (worktree disappeared).
|
||
failedFiles.push({
|
||
path: '<head>',
|
||
error: `git HEAD verification failed: ${e instanceof Error ? e.message : String(e)}`,
|
||
});
|
||
}
|
||
|
||
const elapsed = Date.now() - start;
|
||
|
||
// Bug 9 — gate the sync bookmark on success. If any per-file parse
|
||
// failed, record it to ~/.gbrain/sync-failures.jsonl and DO NOT advance
|
||
// sync.last_commit. The next sync re-walks the same diff and re-attempts
|
||
// the failed files. Escape hatches: --skip-failed acknowledges the
|
||
// current set, --retry-failed re-parses before running the normal sync.
|
||
if (failedFiles.length > 0) {
|
||
recordSyncFailures(failedFiles, headCommit);
|
||
// Emit structured summary grouped by error code so the operator
|
||
// can see *why* files failed, not just how many.
|
||
const codeBreakdown = formatCodeBreakdown(failedFiles);
|
||
if (!opts.skipFailed) {
|
||
serr(
|
||
`\nSync blocked: ${failedFiles.length} file(s) failed to parse:\n` +
|
||
`${codeBreakdown}\n\n` +
|
||
`Fix the YAML frontmatter in the files above and re-run, or use ` +
|
||
`'gbrain sync --skip-failed' to acknowledge and move on.`,
|
||
);
|
||
// Update last_run + repo_path (progress on infra) but NOT last_commit.
|
||
await engine.setConfig('sync.last_run', new Date().toISOString());
|
||
await writeSyncAnchor(engine, opts.sourceId, 'repo_path', repoPath);
|
||
return {
|
||
status: 'blocked_by_failures',
|
||
fromCommit: lastCommit,
|
||
toCommit: headCommit,
|
||
added: filtered.added.length,
|
||
modified: filtered.modified.length,
|
||
deleted: filtered.deleted.length,
|
||
renamed: filtered.renamed.length,
|
||
chunksCreated,
|
||
embedded: 0,
|
||
pagesAffected,
|
||
failedFiles: failedFiles.length,
|
||
};
|
||
}
|
||
// --skip-failed: acknowledge the now-recorded set and proceed.
|
||
const acked = acknowledgeSyncFailures();
|
||
if (acked.count > 0) {
|
||
serr(
|
||
` Acknowledged ${acked.count} failure(s) and advancing past them:\n` +
|
||
`${formatCodeBreakdown(acked.summary)}`,
|
||
);
|
||
}
|
||
}
|
||
|
||
// Update sync state AFTER all changes succeed (source-scoped when
|
||
// opts.sourceId is set, global config otherwise).
|
||
await writeSyncAnchor(engine, opts.sourceId, 'last_commit', headCommit);
|
||
await engine.setConfig('sync.last_run', new Date().toISOString());
|
||
await writeSyncAnchor(engine, opts.sourceId, 'repo_path', repoPath);
|
||
// v0.20.0 Cathedral II Layer 12: persist the chunker version we just
|
||
// finished with so the next sync's up_to_date gate respects it. Only
|
||
// source-scoped syncs track this (see readChunkerVersion for rationale).
|
||
await writeChunkerVersion(engine, opts.sourceId, String(CHUNKER_VERSION));
|
||
|
||
// Log ingest
|
||
await engine.logIngest({
|
||
source_type: 'git_sync',
|
||
source_ref: `${repoPath} @ ${headCommit.slice(0, 8)}`,
|
||
pages_updated: pagesAffected,
|
||
summary: `Sync: +${filtered.added.length} ~${filtered.modified.length} -${filtered.deleted.length} R${filtered.renamed.length}, ${chunksCreated} chunks, ${elapsed}ms`,
|
||
});
|
||
|
||
// Auto-extract links + timeline (always, extraction is cheap CPU).
|
||
// Thread opts.sourceId so the extract phase reconciles edges + timeline
|
||
// entries against the right source — pre-fix (Data R1 HIGH 1) this phase
|
||
// bypassed sourceId entirely and the bare-slug subquery in addTimelineEntry
|
||
// (Data R1 HIGH 2) crashed with 21000 in multi-source brains.
|
||
const extractOpts = opts.sourceId ? { sourceId: opts.sourceId } : undefined;
|
||
if (!opts.noExtract && pagesAffected.length > 0) {
|
||
try {
|
||
const { extractLinksForSlugs, extractTimelineForSlugs } = await import('./extract.ts');
|
||
const linksCreated = await extractLinksForSlugs(engine, repoPath, pagesAffected, extractOpts);
|
||
const timelineCreated = await extractTimelineForSlugs(engine, repoPath, pagesAffected, extractOpts);
|
||
if (linksCreated > 0 || timelineCreated > 0) {
|
||
slog(` Extracted: ${linksCreated} links, ${timelineCreated} timeline entries`);
|
||
}
|
||
} catch { /* extraction is best-effort */ }
|
||
}
|
||
|
||
// v0.31.2: facts extraction now routes through the shared
|
||
// src/core/facts/backstop.ts helper (PR1 commit 6). Sync uses
|
||
// queue mode (fire-and-forget) + 'high-only' filter so a 50-page
|
||
// sync doesn't block on N sequential Sonnet calls. The pre-fix
|
||
// inline loop is gone — it carried (a) a dead-code type filter
|
||
// ('conversation'/'transcript'/'therapy'/'call' aren't real
|
||
// PageTypes), (b) a divergent eligibility shape from put_page,
|
||
// and (c) raw extract→insert without dedup/supersede.
|
||
if (!opts.noExtract && pagesAffected.length > 0 && pagesAffected.length <= 50) {
|
||
const { runFactsBackstop } = await import('../core/facts/backstop.ts');
|
||
const factsSourceId = opts.sourceId ?? 'default';
|
||
for (const slug of pagesAffected) {
|
||
try {
|
||
// v0.40 D21: source-scoped getPage. Pre-v0.40 this called
|
||
// engine.getPage(slug) WITHOUT sourceId, then wrote facts under
|
||
// factsSourceId. On a federated brain with the same slug in two
|
||
// sources (e.g. people/garry-tan in default + zion-brain), this
|
||
// would attribute facts to the wrong source. Codex outside-voice
|
||
// catch on the v0.40 plan review.
|
||
const page = await engine.getPage(slug, { sourceId: factsSourceId });
|
||
if (!page) continue;
|
||
await runFactsBackstop(
|
||
{
|
||
slug,
|
||
type: page.type,
|
||
compiled_truth: page.compiled_truth ?? '',
|
||
frontmatter: page.frontmatter ?? {},
|
||
},
|
||
{
|
||
engine,
|
||
sourceId: factsSourceId,
|
||
sessionId: `sync:${slug}`,
|
||
source: 'sync:import',
|
||
mode: 'queue',
|
||
notabilityFilter: 'high-only',
|
||
},
|
||
);
|
||
} catch { /* per-page enqueue is best-effort */ }
|
||
}
|
||
}
|
||
|
||
// Auto-embed (skip for large syncs — embedding calls OpenAI).
|
||
// Thread sourceId so incremental source syncs embed the page row they just
|
||
// imported instead of falling back to the default source.
|
||
//
|
||
// v0.37 fix wave (Lane D.3 + CDX2-8): switched from `runEmbed` (which
|
||
// does its own process.exit) to `runEmbedCore` so sync can detect the
|
||
// dim-mismatch class and surface a stderr hint without killing the
|
||
// sync. Non-mismatch errors stay best-effort (rate limits, transient
|
||
// network) — those shouldn't break sync.
|
||
let embedded = 0;
|
||
if (!noEmbed && pagesAffected.length > 0 && pagesAffected.length <= 100) {
|
||
try {
|
||
const { runEmbedCore } = await import('./embed.ts');
|
||
const embedOpts = opts.sourceId
|
||
? { slugs: pagesAffected, sourceId: opts.sourceId }
|
||
: { slugs: pagesAffected };
|
||
await runEmbedCore(engine, embedOpts);
|
||
embedded = pagesAffected.length;
|
||
} catch (e: unknown) {
|
||
const { EmbeddingDimMismatchError } = await import('./embed.ts');
|
||
if (e instanceof EmbeddingDimMismatchError) {
|
||
serr('\n' + e.recipeMessage + '\n');
|
||
serr(`Tip: pass --no-embed to sync without embedding, then`);
|
||
serr(`run 'gbrain embed --stale' after fixing the schema.\n`);
|
||
}
|
||
// Other errors stay best-effort — rate limits, transient network.
|
||
}
|
||
} else if (noEmbed || totalChanges > 100) {
|
||
slog(`Text imported. Run 'gbrain embed --stale' to generate embeddings.`);
|
||
}
|
||
|
||
return {
|
||
status: 'synced',
|
||
fromCommit: lastCommit,
|
||
toCommit: headCommit,
|
||
added: filtered.added.length,
|
||
modified: filtered.modified.length,
|
||
deleted: filtered.deleted.length,
|
||
renamed: filtered.renamed.length,
|
||
chunksCreated,
|
||
embedded,
|
||
pagesAffected,
|
||
};
|
||
}
|
||
|
||
async function performFullSync(
|
||
engine: BrainEngine,
|
||
repoPath: string,
|
||
headCommit: string,
|
||
opts: SyncOpts,
|
||
): Promise<SyncResult> {
|
||
// Dry-run: walk the repo, count syncable files, return without writing.
|
||
// Fixes the silent-write-on-dry-run bug where performFullSync called
|
||
// runImport unconditionally regardless of opts.dryRun.
|
||
//
|
||
// v0.31.2 (codex C6): use the strategy-aware walker. Pre-fix this
|
||
// hardcoded `collectMarkdownFiles(repoPath)` and filtered with
|
||
// default-markdown `isSyncable(rel)`, so `gbrain sync --strategy
|
||
// code --dry-run` always reported zero files even when ~1500 code
|
||
// files were waiting.
|
||
if (opts.dryRun) {
|
||
const allFiles = collectSyncableFiles(repoPath, { strategy: opts.strategy ?? 'markdown' });
|
||
slog(
|
||
`Full-sync dry run (strategy=${opts.strategy ?? 'markdown'}): ` +
|
||
`${allFiles.length} file(s) would be imported ` +
|
||
`from ${repoPath} @ ${headCommit.slice(0, 8)}.`,
|
||
);
|
||
return {
|
||
status: 'dry_run',
|
||
fromCommit: null,
|
||
toCommit: headCommit,
|
||
added: allFiles.length,
|
||
modified: 0,
|
||
deleted: 0,
|
||
renamed: 0,
|
||
chunksCreated: 0,
|
||
embedded: 0,
|
||
pagesAffected: [],
|
||
};
|
||
}
|
||
|
||
// v0.22.13 (PR #490 A1 + Q5): full sync is always "large" by definition
|
||
// (entire working tree). Auto-concurrency fires unconditionally for Postgres;
|
||
// PGLite stays serial because its engine is single-connection. Routes the
|
||
// policy through autoConcurrency() so it stays consistent with incremental
|
||
// sync and the jobs handler.
|
||
const FULL_SYNC_LARGE_MARKER = Number.MAX_SAFE_INTEGER;
|
||
const fullConcurrency = autoConcurrency(engine, FULL_SYNC_LARGE_MARKER, opts.concurrency);
|
||
slog(`Running full import of ${repoPath}${fullConcurrency > 1 ? ` (${fullConcurrency} workers)` : ''}...`);
|
||
const { runImport } = await import('./import.ts');
|
||
const importArgs = [repoPath];
|
||
if (opts.noEmbed) importArgs.push('--no-embed');
|
||
if (fullConcurrency > 1) importArgs.push('--workers', String(fullConcurrency));
|
||
// v0.31.2: thread strategy through so code-strategy first sync
|
||
// actually enumerates code files (closes bug 1).
|
||
// v0.30.x: thread sourceId so performFullSync routes pages to the named
|
||
// source (incremental path already does this).
|
||
const _fullImportT0 = Date.now();
|
||
serr(`[gbrain phase] sync.fullsync.import start strategy=${opts.strategy ?? 'markdown'}`);
|
||
const result = await runImport(engine, importArgs, {
|
||
commit: headCommit,
|
||
strategy: opts.strategy,
|
||
sourceId: opts.sourceId,
|
||
});
|
||
serr(
|
||
`[gbrain phase] sync.fullsync.import done ${Date.now() - _fullImportT0}ms ` +
|
||
`imported=${result.imported} skipped=${result.skipped} errors=${result.errors}`,
|
||
);
|
||
|
||
// Bug 9 — gate the full-sync bookmark on success. runImport already
|
||
// writes its own sync.last_commit conditionally (import.ts), but
|
||
// performFullSync is called on first-sync + force-full paths where
|
||
// the sync module owns the last_commit write. Respect the same gate.
|
||
if (result.failures.length > 0) {
|
||
recordSyncFailures(result.failures, headCommit);
|
||
const codeBreakdown = formatCodeBreakdown(result.failures);
|
||
if (!opts.skipFailed) {
|
||
serr(
|
||
`\nFull sync blocked: ${result.failures.length} file(s) failed:\n` +
|
||
`${codeBreakdown}\n\n` +
|
||
`Fix the YAML in those files and re-run, or use '--skip-failed'.`,
|
||
);
|
||
await engine.setConfig('sync.last_run', new Date().toISOString());
|
||
await writeSyncAnchor(engine, opts.sourceId, 'repo_path', repoPath);
|
||
return {
|
||
status: 'blocked_by_failures',
|
||
fromCommit: null,
|
||
toCommit: headCommit,
|
||
added: 0, modified: 0, deleted: 0, renamed: 0,
|
||
chunksCreated: result.chunksCreated,
|
||
embedded: 0,
|
||
pagesAffected: [],
|
||
failedFiles: result.failures.length,
|
||
};
|
||
}
|
||
const acked = acknowledgeSyncFailures();
|
||
if (acked.count > 0) {
|
||
serr(
|
||
` Acknowledged ${acked.count} failure(s) and advancing past them:\n` +
|
||
`${formatCodeBreakdown(acked.summary)}`,
|
||
);
|
||
}
|
||
}
|
||
|
||
// Persist sync state so next sync is incremental (C1 fix: was missing).
|
||
// v0.18.0 Step 5: routed through writeSyncAnchor so --source pins it
|
||
// to the right sources row rather than the global config.
|
||
await writeSyncAnchor(engine, opts.sourceId, 'last_commit', headCommit);
|
||
await engine.setConfig('sync.last_run', new Date().toISOString());
|
||
await writeSyncAnchor(engine, opts.sourceId, 'repo_path', repoPath);
|
||
// v0.20.0 Cathedral II Layer 12: persist chunker version for the gate.
|
||
await writeChunkerVersion(engine, opts.sourceId, String(CHUNKER_VERSION));
|
||
|
||
// Full sync doesn't track pagesAffected, so fall back to embed --stale.
|
||
// v0.37 fix wave (Lane D.3 + CDX2-8): switched to runEmbedCore for the
|
||
// same reason as the incremental path — surface dim-mismatch via hint
|
||
// instead of silently swallowing or killing the process.
|
||
let embedded = 0;
|
||
if (!opts.noEmbed) {
|
||
try {
|
||
const { runEmbedCore } = await import('./embed.ts');
|
||
await runEmbedCore(engine, { stale: true });
|
||
embedded = result.imported;
|
||
} catch (e: unknown) {
|
||
const { EmbeddingDimMismatchError } = await import('./embed.ts');
|
||
if (e instanceof EmbeddingDimMismatchError) {
|
||
serr('\n' + e.recipeMessage + '\n');
|
||
serr(`Tip: pass --no-embed to sync without embedding, then`);
|
||
serr(`run 'gbrain embed --stale' after fixing the schema.\n`);
|
||
}
|
||
// Other errors stay best-effort.
|
||
}
|
||
}
|
||
|
||
return {
|
||
status: 'first_sync',
|
||
fromCommit: null,
|
||
toCommit: headCommit,
|
||
added: result.imported,
|
||
modified: 0,
|
||
deleted: 0,
|
||
renamed: 0,
|
||
chunksCreated: result.chunksCreated,
|
||
embedded,
|
||
pagesAffected: [],
|
||
};
|
||
}
|
||
|
||
export async function runSync(engine: BrainEngine, args: string[]) {
|
||
// v0.40 Federated Sync v2: `gbrain sync trigger` subcommand
|
||
// Routes to runSyncTrigger which queues a 'sync' minion job with
|
||
// auto_embed_backfill=true. Falls through to the normal sync path
|
||
// if 'trigger' isn't the first arg.
|
||
if (args[0] === 'trigger') {
|
||
return runSyncTrigger(engine, args.slice(1));
|
||
}
|
||
|
||
// v0.37 fix wave (Lane D.4 + CDX2-12): print usage when `--help`/`-h` is
|
||
// passed. Pre-fix this was unreachable because the dispatcher's generic
|
||
// CLI-only short-circuit fired first; sync is now in CLI_ONLY_SELF_HELP.
|
||
if (args.includes('--help') || args.includes('-h')) {
|
||
console.log(`Usage: gbrain sync [options]
|
||
|
||
Sync the brain repo's text content into the engine, then embed.
|
||
|
||
Options:
|
||
--no-embed Skip the embed step. Use this when the embed
|
||
provider is misconfigured or you want to defer
|
||
embedding (run 'gbrain embed --stale' later).
|
||
--workers N Run the import phase with N parallel workers
|
||
(alias: --concurrency). Default: 4 when the
|
||
diff is >100 files, else serial.
|
||
--source <id> Scope sync to a single source. Defaults to the
|
||
brain's default source.
|
||
--repo <path> Path to the brain repo. Defaults to the path
|
||
saved by 'gbrain init'.
|
||
--full Force a full re-sync (rare; usually incremental).
|
||
--dry-run Show what would be synced without writing.
|
||
--skip-failed Acknowledge previously-recorded sync failures so
|
||
the bookmark can advance past unparseable files.
|
||
--retry-failed Re-attempt previously-failed files; clear on success.
|
||
--watch Re-sync continuously on an interval.
|
||
--interval N Watch-mode interval in seconds (default 60).
|
||
--no-pull Skip 'git pull' before the sync (useful for tests).
|
||
--all Sync every registered source instead of just the
|
||
default (multi-source brains).
|
||
--parallel N (with --all) Run up to N sources concurrently.
|
||
Default: min(sourceCount, --workers, 4). Each
|
||
source takes its own per-source DB lock
|
||
(gbrain-sync:<source_id>) so independent sources
|
||
sync without contending. Total live Postgres
|
||
connections per wave ≈ parallel × workers × 2
|
||
(per-file pool) + parent pool. Pass --parallel 1
|
||
to force serial.
|
||
--json Emit a structured JSON envelope on stdout
|
||
({schema_version: 1, sources, parallel,
|
||
ok_count, error_count}). Human banners route to
|
||
stderr so '--json | jq' parses cleanly.
|
||
Exit codes: 0 = all sources ok, 1 = any error,
|
||
2 = cost-prompt-not-confirmed.
|
||
--yes Accept any interactive prompts (CI / non-TTY).
|
||
|
||
See also:
|
||
gbrain embed --stale Re-embed all stale chunks (post --no-embed).
|
||
gbrain doctor Diagnose dim mismatches and other sync issues.
|
||
`);
|
||
return;
|
||
}
|
||
|
||
const repoPath = args.find((a, i) => args[i - 1] === '--repo') || undefined;
|
||
const watch = args.includes('--watch');
|
||
const intervalStr = args.find((a, i) => args[i - 1] === '--interval');
|
||
const interval = intervalStr ? parseInt(intervalStr, 10) : 60;
|
||
const dryRun = args.includes('--dry-run');
|
||
const full = args.includes('--full');
|
||
const noPull = args.includes('--no-pull');
|
||
const noEmbed = args.includes('--no-embed');
|
||
const skipFailed = args.includes('--skip-failed');
|
||
const retryFailed = args.includes('--retry-failed');
|
||
const syncAll = args.includes('--all');
|
||
const jsonOut = args.includes('--json');
|
||
const yesFlag = args.includes('--yes');
|
||
// v0.41.6.0 D3: lock-recovery flags. --break-lock (safe) verifies the
|
||
// holder is local-host + (TTL-expired OR PID-dead+60s-old) before
|
||
// deleting the row. --force-break-lock skips the liveness check. Both
|
||
// are refused when combined with --all (per-source invocation required;
|
||
// v0.40 lock keys are gbrain-sync:<sourceId>).
|
||
const breakLock = args.includes('--break-lock');
|
||
const forceBreakLock = args.includes('--force-break-lock');
|
||
|
||
// v0.41.6.0 D3: handle --break-lock / --force-break-lock BEFORE the
|
||
// sync would otherwise contend on the lock. Resolves the active source,
|
||
// inspects the lock, decides break/refuse, exits without running sync.
|
||
if (breakLock || forceBreakLock) {
|
||
if (syncAll) {
|
||
console.error('--break-lock / --force-break-lock cannot be combined with --all.');
|
||
console.error('Per-source lock keys (gbrain-sync:<sourceId>) require per-source invocation.');
|
||
console.error('');
|
||
console.error('To recover stale locks across all sources, shell-loop:');
|
||
console.error(` for src in $(gbrain sources list --json | jq -r '.[].id'); do gbrain sync --break-lock --source "$src"; done`);
|
||
process.exit(1);
|
||
}
|
||
const sourceArg = args.find((a, i) => args[i - 1] === '--source');
|
||
const sourceId = sourceArg ?? 'default';
|
||
const lockKey = `gbrain-sync:${sourceId}`;
|
||
const exit = await runBreakLock(engine, lockKey, sourceId, { force: forceBreakLock, json: jsonOut });
|
||
process.exit(exit);
|
||
}
|
||
|
||
// v0.41.6.0 D1: preflight embedding credentials BEFORE the import phase
|
||
// so a missing OPENAI_API_KEY exits with one clean line instead of
|
||
// writing N identical entries to sync-failures.jsonl. Skipped when
|
||
// --no-embed (the canonical opt-out) or --dry-run (no provider calls
|
||
// happen in dry-run anyway).
|
||
if (!noEmbed && !dryRun) {
|
||
const { validateEmbeddingCreds, EmbeddingCredentialError } = await import('../core/embed-preflight.ts');
|
||
try {
|
||
validateEmbeddingCreds();
|
||
} catch (e) {
|
||
if (e instanceof EmbeddingCredentialError) {
|
||
if (jsonOut) {
|
||
console.log(JSON.stringify({ status: 'embedding_credentials_missing', diagnosis: e.diagnosis }));
|
||
} else {
|
||
console.error('');
|
||
console.error(e.userMessage);
|
||
console.error('');
|
||
}
|
||
process.exit(1);
|
||
}
|
||
throw e;
|
||
}
|
||
}
|
||
// v0.40 D4+D18: parallel `sync --all` by default; --serial opts back to v1.
|
||
// --no-auto-embed skips the per-source embed-backfill auto-enqueue.
|
||
// --max-sources N caps fan-out (default min(sources.length, 8)).
|
||
const serialFlag = args.includes('--serial');
|
||
const noAutoEmbed = args.includes('--no-auto-embed');
|
||
const maxSourcesStr = args.find((a, i) => args[i - 1] === '--max-sources');
|
||
const maxSources = maxSourcesStr ? parseInt(maxSourcesStr, 10) : undefined;
|
||
if (maxSourcesStr && (!Number.isFinite(maxSources!) || maxSources! < 1)) {
|
||
console.error(`Invalid --max-sources value: "${maxSourcesStr}". Must be a positive integer.`);
|
||
process.exit(1);
|
||
}
|
||
const strategyArg = args.find((a, i) => args[i - 1] === '--strategy') as SyncOpts['strategy'] | undefined;
|
||
const concurrencyStr = args.find((a, i) => args[i - 1] === '--concurrency' || args[i - 1] === '--workers');
|
||
const parallelStr = args.find((a, i) => args[i - 1] === '--parallel');
|
||
// v0.22.13 (PR #490 Q2): parseWorkers throws on '0', '-3', 'foo', '1.5' instead
|
||
// of silently falling through to auto-concurrency or NaN. Loud failure beats
|
||
// a 4-worker spawn from a typo. v0.40.3.0: same validation applies to --parallel.
|
||
let concurrency: number | undefined;
|
||
let parallelOverride: number | undefined;
|
||
try {
|
||
concurrency = parseWorkers(concurrencyStr);
|
||
} catch (e) {
|
||
console.error(e instanceof Error ? e.message : String(e));
|
||
process.exit(1);
|
||
}
|
||
try {
|
||
parallelOverride = parseWorkers(parallelStr);
|
||
} catch (e) {
|
||
console.error(e instanceof Error ? e.message : String(e));
|
||
process.exit(1);
|
||
}
|
||
|
||
// --skip-failed: acknowledge pre-existing unacked failures BEFORE the sync
|
||
// runs, not only ones the current run produces. Without this, the common
|
||
// recovery flow — fix the YAML, re-run sync, then run --skip-failed to
|
||
// clear the log — fails to clear anything: when there are no NEW failures
|
||
// (because the files are now fixed), the inner ack path in performSync is
|
||
// never reached, and "Already up to date." leaves the log untouched. Both
|
||
// doctor and printSyncResult instruct users to run --skip-failed in
|
||
// exactly this case, so the flag has to handle stale entries up-front.
|
||
if (skipFailed) {
|
||
const stale = unacknowledgedSyncFailures();
|
||
if (stale.length > 0) {
|
||
const acked = acknowledgeSyncFailures();
|
||
console.log(`Acknowledged ${acked.count} pre-existing failure(s).`);
|
||
}
|
||
}
|
||
|
||
// v0.18.0 Step 5: --source resolves to a sources(id) row. Falls back
|
||
// to pre-v0.17 global config (sync.repo_path + sync.last_commit) when
|
||
// no flag, no env, no dotfile is present.
|
||
//
|
||
// v0.41.13 (#1434): always call the resolver, not just when explicit/env
|
||
// is set. Pre-fix, `gbrain sync` without --source skipped resolution and
|
||
// left sourceId undefined — which the engine treated as the seeded
|
||
// 'default' source. Users with a single non-default registered source
|
||
// (studiovault, etc.) silently routed every write to a source holding
|
||
// 0 pages, then createVersion threw on the slug lookup.
|
||
//
|
||
// The resolver's new `sole_non_default` tier (5.5) routes those
|
||
// single-source brains to the right place automatically; the nudge
|
||
// surfaces the auto-route to stderr so the user knows what happened
|
||
// and can pass --source to override if needed.
|
||
const explicitSource = args.find((a, i) => args[i - 1] === '--source') || null;
|
||
const { resolveSourceWithTier, formatSoleNonDefaultNudge } = await import('../core/source-resolver.ts');
|
||
const resolved = await resolveSourceWithTier(engine, explicitSource);
|
||
const sourceId: string = resolved.source_id;
|
||
if (resolved.tier === 'sole_non_default') {
|
||
const nudge = formatSoleNonDefaultNudge(sourceId);
|
||
if (nudge) process.stderr.write(nudge + '\n');
|
||
}
|
||
|
||
// v0.19.0 — `sync --all` iterates all registered sources with a
|
||
// local_path. Sources are the canonical v0.18.0 abstraction: per-source
|
||
// last_commit, last_sync_at, config.federated flags. Per-source
|
||
// bookmarks live in the sources table (not ~/.gbrain/config.json),
|
||
// which is why this path replaced Garry's OpenClaw `multi-repo.ts` shim.
|
||
//
|
||
// Only sources with a non-null local_path participate. A GitHub-only
|
||
// source (no checkout) has nothing for `sync` to pull. Sources with
|
||
// syncEnabled=false in config.jsonb are skipped too.
|
||
if (syncAll) {
|
||
const sources = await engine.executeRaw<{ id: string; name: string; local_path: string | null; config: Record<string, unknown> }>(
|
||
`SELECT id, name, local_path, config FROM sources WHERE local_path IS NOT NULL`,
|
||
);
|
||
if (!sources || sources.length === 0) {
|
||
console.log('No sources with local_path configured. Use `gbrain sources add <id> --path <path>` first.');
|
||
return;
|
||
}
|
||
|
||
// v0.20.0 Cathedral II Layer 8 D1 — cost preview + ConfirmationRequired
|
||
// gate. Before kicking off a multi-source sync that may embed tens of
|
||
// thousands of chunks (real money), walk the sync-diff set(s), sum
|
||
// tokens, compute USD estimate, and gate:
|
||
// - TTY + !json + !yes → interactive [y/N] prompt
|
||
// - non-TTY OR --json OR piped → emit ConfirmationRequired envelope,
|
||
// exit 2 (reserve 1 for runtime errors)
|
||
// - --yes → skip prompt entirely
|
||
// - --dry-run → preview + exit 0
|
||
// Skipped entirely when --no-embed is set (user already opted out of
|
||
// the cost and will run `embed --stale` later).
|
||
if (!noEmbed) {
|
||
const preview = estimateSyncAllCost(sources);
|
||
const costUsd = estimateEmbeddingCostUsd(preview.totalTokens);
|
||
const previewMsg =
|
||
`sync --all preview: ${preview.totalFiles} files across ${preview.activeSources} source(s), ` +
|
||
`~${preview.totalTokens.toLocaleString()} tokens, est. $${costUsd.toFixed(2)} on ${EMBEDDING_MODEL}.`;
|
||
|
||
if (dryRun) {
|
||
if (jsonOut) {
|
||
console.log(JSON.stringify({ status: 'dry_run', preview, costUsd, model: EMBEDDING_MODEL }));
|
||
} else {
|
||
console.log(previewMsg);
|
||
console.log('--dry-run: exit without syncing.');
|
||
}
|
||
return;
|
||
}
|
||
|
||
if (!yesFlag) {
|
||
const isTTY = Boolean(process.stdout.isTTY) && Boolean(process.stdin.isTTY);
|
||
if (!isTTY || jsonOut) {
|
||
// Agent-facing path: emit structured envelope, exit 2.
|
||
const envelope = serializeError(errorFor({
|
||
class: 'ConfirmationRequired',
|
||
code: 'cost_preview_requires_yes',
|
||
message: previewMsg,
|
||
hint: 'Pass --yes to proceed, or --dry-run to see the preview and exit 0.',
|
||
}));
|
||
console.log(JSON.stringify({ error: envelope, preview, costUsd, model: EMBEDDING_MODEL }));
|
||
process.exit(2);
|
||
}
|
||
// Interactive TTY path: prompt [y/N].
|
||
console.log(previewMsg);
|
||
const answer = await promptYesNo('Proceed? [y/N] ');
|
||
if (!answer) {
|
||
console.log('Cancelled.');
|
||
return;
|
||
}
|
||
}
|
||
}
|
||
|
||
// v0.40.5.0 Federated Sync v2 (master) + v0.40.6.0 layering (this branch):
|
||
// master added parallel fan-out via pMapAllSettled, embed-backfill auto-
|
||
// submit, --serial / --max-sources / --no-auto-embed flags, and feature-
|
||
// flagged the whole thing behind sync.federated_v2. This branch layers
|
||
// additive improvements on top:
|
||
// - humanSink swap so `--json` keeps stdout clean (D4)
|
||
// - --skip-failed / --retry-failed reject under parallel>1 (D15 — the
|
||
// sync-failures.jsonl is brain-global, parallel acks race)
|
||
// - connection-budget stderr warning at parallel × workers × 2 > 16 (D10)
|
||
// - withSourcePrefix wrap inside runOne so slog/serr lines from
|
||
// performSync get the [<source-id>] prefix under parallel mode (D6)
|
||
// - stable JSON envelope {schema_version:1, sources, ...} when --json
|
||
const { isFederatedV2Enabled } = await import('../core/feature-flags.ts');
|
||
const v2Enabled = await isFederatedV2Enabled(engine);
|
||
const activeSources = sources.filter((s) => {
|
||
const cfg = (s.config || {}) as { syncEnabled?: boolean };
|
||
return cfg.syncEnabled !== false;
|
||
});
|
||
const disabledCount = sources.length - activeSources.length;
|
||
const humanSink: NodeJS.WriteStream = jsonOut ? process.stderr : process.stdout;
|
||
const writeHuman = (line: string) => humanSink.write(line + '\n');
|
||
|
||
if (disabledCount > 0) {
|
||
writeHuman(`Skipping ${disabledCount} disabled source(s).`);
|
||
}
|
||
|
||
if (activeSources.length === 0) {
|
||
if (jsonOut) {
|
||
console.log(JSON.stringify({
|
||
schema_version: 1,
|
||
sources: [],
|
||
parallel: 0,
|
||
ok_count: 0,
|
||
error_count: 0,
|
||
}));
|
||
}
|
||
return;
|
||
}
|
||
|
||
// Per-source result accumulator for the optional --json envelope.
|
||
type PerSourceResult = {
|
||
sourceId: string;
|
||
sourceName: string;
|
||
status: 'ok' | 'error';
|
||
result?: SyncResult;
|
||
error?: string;
|
||
};
|
||
const perSourceResults: PerSourceResult[] = [];
|
||
|
||
const runOne = async (src: typeof sources[number]): Promise<SyncResult> => {
|
||
const cfg = (src.config || {}) as { strategy?: 'markdown' | 'code' | 'auto' };
|
||
// D18: parallel path defers embed; auto-enqueue embed-backfill after.
|
||
const effectiveNoEmbed = v2Enabled && !serialFlag && !noEmbed ? true : noEmbed;
|
||
const repoOpts: SyncOpts = {
|
||
repoPath: src.local_path!,
|
||
dryRun, full, noPull,
|
||
noEmbed: effectiveNoEmbed,
|
||
skipFailed, retryFailed,
|
||
sourceId: src.id,
|
||
strategy: cfg.strategy,
|
||
concurrency,
|
||
};
|
||
// v0.40.6.0 (D6): wrap performSync in withSourcePrefix so every slog /
|
||
// serr line emitted from inside the sync code path gets prefixed with
|
||
// `[<source-id>] `. Under master's pMapAllSettled fan-out, this is
|
||
// what makes `grep '\[media-corpus\]'` against parallel output work.
|
||
const result = await withSourcePrefix(src.id, () => performSync(engine, repoOpts));
|
||
if (result.status !== 'dry_run' && result.status !== 'blocked_by_failures') {
|
||
manageGitignore(src.local_path!, engine.kind);
|
||
}
|
||
// D18: auto-enqueue embed-backfill per source (unless opted out)
|
||
if (
|
||
v2Enabled &&
|
||
!noAutoEmbed &&
|
||
!dryRun &&
|
||
result.status !== 'dry_run' &&
|
||
result.status !== 'up_to_date'
|
||
) {
|
||
try {
|
||
const { submitEmbedBackfill } = await import('../core/embed-backfill-submit.ts');
|
||
const sub = await submitEmbedBackfill(engine, src.id, { reason: 'sync_all' });
|
||
if (sub.status === 'submitted') {
|
||
writeHuman(` → embed-backfill job ${sub.jobId} queued for ${src.name}`);
|
||
} else if (sub.status === 'cooldown') {
|
||
writeHuman(` → embed-backfill skipped (cooldown) for ${src.name}`);
|
||
} else if (sub.status === 'spend_capped') {
|
||
writeHuman(` → embed-backfill skipped (24h spend cap $${sub.spendCapUsd}) for ${src.name}`);
|
||
}
|
||
} catch (e) {
|
||
process.stderr.write(` → embed-backfill submission failed for ${src.name}: ${e instanceof Error ? e.message : String(e)}\n`);
|
||
}
|
||
}
|
||
return result;
|
||
};
|
||
|
||
const parallelEligible =
|
||
v2Enabled && !serialFlag && engine.kind !== 'pglite' && activeSources.length > 1;
|
||
|
||
// v0.40.6.0 (D15): refuse --skip-failed / --retry-failed when running
|
||
// parallel. sync-failures.jsonl is brain-global; parallel acks race.
|
||
if (parallelEligible && (skipFailed || retryFailed)) {
|
||
const flag = skipFailed ? '--skip-failed' : '--retry-failed';
|
||
console.error(
|
||
`Error: ${flag} is not supported under parallel sync.\n` +
|
||
` (the sync-failures log is brain-global and parallel acks race).\n` +
|
||
` Re-run with --serial for the recovery flow:\n` +
|
||
` gbrain sync --all --serial ${flag}`,
|
||
);
|
||
process.exit(1);
|
||
}
|
||
|
||
// Effective parallelism — surfaced in the --json envelope so consumers
|
||
// know how the run was actually dispatched. 1 in the serial fallback,
|
||
// capped at min(sourceCount, --max-sources, 8) in the parallel path.
|
||
const effectiveParallel = parallelEligible
|
||
? Math.min(activeSources.length, maxSources ?? 8)
|
||
: 1;
|
||
|
||
if (parallelEligible) {
|
||
const { pMapAllSettled } = await import('../core/parallel.ts');
|
||
const cap = effectiveParallel;
|
||
|
||
// v0.40.6.0 (D10): connection-budget stderr warning. Each per-file
|
||
// worker opens its own PostgresEngine with poolSize=2, so the real
|
||
// live-connection ceiling is `cap × workers × 2` per wave plus the
|
||
// parent pool. The original PR understated by 2× — fix the math.
|
||
const effectiveWorkers = concurrency ?? 4;
|
||
const budget = cap * effectiveWorkers * 2;
|
||
if (budget > 16) {
|
||
process.stderr.write(
|
||
`[sync --all] Connection budget: parallel=${cap} × workers=${effectiveWorkers} × 2 ` +
|
||
`(per-file pool) = ${budget} concurrent connections per fan-out wave (+ parent pool). ` +
|
||
`Check pgbouncer/Postgres max_connections (SELECT count(*) FROM pg_stat_activity); ` +
|
||
`raise the cap or lower --max-sources/--workers if you see "too many clients" errors.\n`,
|
||
);
|
||
}
|
||
|
||
writeHuman(`\nParallel sync: ${activeSources.length} sources, ${cap} concurrent workers.\n`);
|
||
const results = await pMapAllSettled(activeSources, cap, async (src) => {
|
||
const r = await runOne(src);
|
||
return { name: src.name, result: r };
|
||
});
|
||
// Print per-source aggregate at the end. humanSink so --json stays clean.
|
||
writeHuman('\n--- sync --all aggregate ---');
|
||
for (let i = 0; i < results.length; i++) {
|
||
const r = results[i];
|
||
const src = activeSources[i];
|
||
if (r.status === 'fulfilled') {
|
||
writeHuman(` ✓ ${src.name}: ${r.value.result.status} (added=${r.value.result.added}, modified=${r.value.result.modified}, deleted=${r.value.result.deleted})`);
|
||
perSourceResults.push({
|
||
sourceId: src.id,
|
||
sourceName: src.name,
|
||
status: 'ok',
|
||
result: r.value.result,
|
||
});
|
||
} else {
|
||
const msg = r.reason instanceof Error ? r.reason.message : String(r.reason);
|
||
process.stderr.write(` ✗ ${src.name}: ${msg}\n`);
|
||
perSourceResults.push({
|
||
sourceId: src.id,
|
||
sourceName: src.name,
|
||
status: 'error',
|
||
error: msg,
|
||
});
|
||
}
|
||
}
|
||
} else {
|
||
for (const src of activeSources) {
|
||
writeHuman(`\n--- Syncing source: ${src.name} ---`);
|
||
try {
|
||
const result = await runOne(src);
|
||
printSyncResult(result, humanSink);
|
||
perSourceResults.push({
|
||
sourceId: src.id,
|
||
sourceName: src.name,
|
||
status: 'ok',
|
||
result,
|
||
});
|
||
} catch (e: unknown) {
|
||
const msg = e instanceof Error ? e.message : String(e);
|
||
process.stderr.write(`Error syncing ${src.name}: ${msg}\n`);
|
||
perSourceResults.push({
|
||
sourceId: src.id,
|
||
sourceName: src.name,
|
||
status: 'error',
|
||
error: msg,
|
||
});
|
||
}
|
||
}
|
||
}
|
||
|
||
const okCount = perSourceResults.filter((r) => r.status === 'ok').length;
|
||
const errCount = perSourceResults.filter((r) => r.status === 'error').length;
|
||
|
||
if (jsonOut) {
|
||
// Sort by source_id at emit time so the envelope is deterministic
|
||
// even though completion order is not (pMapAllSettled semantics).
|
||
const sortedSources = perSourceResults
|
||
.slice()
|
||
.sort((a, b) => a.sourceId.localeCompare(b.sourceId))
|
||
.map((r) => ({
|
||
source_id: r.sourceId,
|
||
name: r.sourceName,
|
||
status: r.status,
|
||
...(r.result ? {
|
||
sync_status: r.result.status,
|
||
added: r.result.added,
|
||
modified: r.result.modified,
|
||
deleted: r.result.deleted,
|
||
chunks_created: r.result.chunksCreated,
|
||
embedded: r.result.embedded,
|
||
} : {}),
|
||
...(r.error ? { error: r.error } : {}),
|
||
}));
|
||
console.log(JSON.stringify({
|
||
schema_version: 1,
|
||
sources: sortedSources,
|
||
parallel: effectiveParallel,
|
||
ok_count: okCount,
|
||
error_count: errCount,
|
||
}));
|
||
}
|
||
|
||
if (errCount > 0) process.exit(1);
|
||
return;
|
||
}
|
||
|
||
const opts: SyncOpts = { repoPath, dryRun, full, noPull, noEmbed, skipFailed, retryFailed, sourceId, strategy: strategyArg, concurrency };
|
||
|
||
// Bug 9 — --retry-failed: before running normal sync, clear acknowledgment
|
||
// flags so the sync picks them up as fresh work. The actual re-attempt
|
||
// happens inside the regular incremental/full loop because once the commit
|
||
// pointer is behind the failures, the diff naturally revisits them.
|
||
if (retryFailed) {
|
||
const failures = unacknowledgedSyncFailures();
|
||
if (failures.length === 0) {
|
||
console.log('No unacknowledged sync failures to retry.');
|
||
} else {
|
||
console.log(`Retrying ${failures.length} previously-failed file(s)...`);
|
||
// Don't acknowledge them yet — they must succeed to clear.
|
||
}
|
||
}
|
||
|
||
if (!watch) {
|
||
const result = await performSync(engine, opts);
|
||
printSyncResult(result);
|
||
// Issue #2 + eng-review pass-2 finding #1 + Codex P1: manage .gitignore ONLY
|
||
// on successful sync. Skip on dry-run (don't mutate disk in preview mode)
|
||
// and blocked_by_failures (sync state is inconsistent — defer .gitignore
|
||
// until next clean run). Resolve the effective repo path so the wire-up
|
||
// fires in the common case where the user runs `gbrain sync` without
|
||
// passing --repo every time.
|
||
if (result.status !== 'dry_run' && result.status !== 'blocked_by_failures') {
|
||
const effectiveRepoPath = opts.repoPath ?? (await getDefaultSourcePath(engine));
|
||
if (effectiveRepoPath) {
|
||
manageGitignore(effectiveRepoPath, engine.kind);
|
||
}
|
||
}
|
||
return;
|
||
}
|
||
|
||
// Watch mode
|
||
let consecutiveErrors = 0;
|
||
console.log(`Watching for changes every ${interval}s... (Ctrl+C to stop)`);
|
||
|
||
while (true) {
|
||
try {
|
||
const result = await performSync(engine, { ...opts, full: false });
|
||
consecutiveErrors = 0;
|
||
if (result.status === 'synced') {
|
||
const ts = new Date().toISOString().slice(11, 19);
|
||
console.log(`[${ts}] Synced: +${result.added} ~${result.modified} -${result.deleted} R${result.renamed}`);
|
||
}
|
||
// Same gate as non-watch: only manage .gitignore on successful sync.
|
||
// Same repo-resolution path so watch mode catches the implicit-resolved case.
|
||
if (result.status !== 'dry_run' && result.status !== 'blocked_by_failures') {
|
||
const effectiveRepoPath = opts.repoPath ?? (await getDefaultSourcePath(engine));
|
||
if (effectiveRepoPath) {
|
||
manageGitignore(effectiveRepoPath, engine.kind);
|
||
}
|
||
}
|
||
} catch (e: unknown) {
|
||
consecutiveErrors++;
|
||
const msg = e instanceof Error ? e.message : String(e);
|
||
console.error(`[${new Date().toISOString().slice(11, 19)}] Sync error (${consecutiveErrors}/5): ${msg}`);
|
||
if (consecutiveErrors >= 5) {
|
||
console.error(`5 consecutive sync failures. Stopping watch.`);
|
||
process.exit(1);
|
||
}
|
||
}
|
||
await new Promise(r => setTimeout(r, interval * 1000));
|
||
}
|
||
}
|
||
|
||
/**
|
||
* v0.40.3.0 — resolve effective per-source concurrency for `sync --all`.
|
||
*
|
||
* Inputs:
|
||
* - sourceCount: number of `syncEnabled !== false` sources to walk
|
||
* - explicitParallel: user's `--parallel N` value (post `parseWorkers`),
|
||
* `undefined` when the flag was not provided
|
||
* - workers: user's `--workers N` value, used as a soft cap when no
|
||
* explicit `--parallel` is given. The per-source budget can't exceed
|
||
* the per-file worker count because each per-file worker opens its
|
||
* own PostgresEngine pool — see DEFAULT_PARALLEL_SOURCES in
|
||
* `src/core/sync-concurrency.ts` for the connection-math story.
|
||
* - engineKind: PGLite is single-connection → always returns 1.
|
||
*
|
||
* Rules:
|
||
* - PGLite → always 1
|
||
* - sourceCount <= 0 → 1 (divide-by-zero guard)
|
||
* - explicit `--parallel` → wins, clamped to `[1, sourceCount]`
|
||
* - auto path → `min(sourceCount, workers ?? DEFAULT_PARALLEL_SOURCES)`
|
||
*
|
||
* Returns >= 1. Single-source brains return 1 (no point fanning out).
|
||
*/
|
||
export function resolveParallelism(input: {
|
||
sourceCount: number;
|
||
explicitParallel?: number;
|
||
workers?: number;
|
||
engineKind: 'pglite' | 'postgres';
|
||
}): number {
|
||
if (input.engineKind === 'pglite') return 1;
|
||
if (input.sourceCount <= 0) return 1;
|
||
if (input.explicitParallel !== undefined) {
|
||
return Math.max(1, Math.min(input.explicitParallel, input.sourceCount));
|
||
}
|
||
const ceiling = input.workers && input.workers > 0
|
||
? Math.min(input.workers, DEFAULT_PARALLEL_SOURCES)
|
||
: DEFAULT_PARALLEL_SOURCES;
|
||
return Math.max(1, Math.min(input.sourceCount, ceiling));
|
||
}
|
||
|
||
/**
|
||
* v0.40.3.0 — per-source sync wrapper for the `--all` worker pool.
|
||
*
|
||
* Three responsibilities:
|
||
* 1. Build the per-source `SyncOpts` from the shared CLI flags.
|
||
* 2. Wrap the call in `withSourcePrefix(src.id, ...)` so every
|
||
* `slog`/`serr` line emitted from inside `performSync` (and its
|
||
* callees) gets prefixed with `[<source-id>] ` for greppable
|
||
* parallel output (D6 + D12 + D13).
|
||
* 3. Pre-render the start banner into the returned `log` string so
|
||
* the worker pool flushes it (and any subsequent `printSyncResult`)
|
||
* via the human sink — which `--json` routes to stderr to keep
|
||
* stdout JSON-only (D4).
|
||
*
|
||
* Note: source.name is shown in the start banner (one-shot, easy to
|
||
* escape) but the prefix on every line uses source.id (slug-validated;
|
||
* no newline-injection risk per D13).
|
||
*
|
||
* The per-source DB lock invariant (D8) fires inside `performSync` —
|
||
* since `repoOpts.sourceId` is set, the per-source lock is the default.
|
||
* `withRefreshingLock` (D11) handles TTL renewal automatically for
|
||
* sources that exceed 30 minutes.
|
||
*/
|
||
export async function syncOneSource(
|
||
engine: BrainEngine,
|
||
src: { id: string; name: string; local_path: string | null; config: Record<string, unknown> },
|
||
shared: {
|
||
dryRun: boolean;
|
||
full: boolean;
|
||
noPull: boolean;
|
||
noEmbed: boolean;
|
||
skipFailed: boolean;
|
||
retryFailed: boolean;
|
||
concurrency: number | undefined;
|
||
},
|
||
): Promise<{ result: SyncResult; log: string }> {
|
||
const cfg = (src.config || {}) as { strategy?: 'markdown' | 'code' | 'auto' };
|
||
const log = `\n--- Syncing source: ${src.name} ---\n`;
|
||
const repoOpts: SyncOpts = {
|
||
repoPath: src.local_path!,
|
||
dryRun: shared.dryRun,
|
||
full: shared.full,
|
||
noPull: shared.noPull,
|
||
noEmbed: shared.noEmbed,
|
||
skipFailed: shared.skipFailed,
|
||
retryFailed: shared.retryFailed,
|
||
sourceId: src.id,
|
||
strategy: cfg.strategy,
|
||
concurrency: shared.concurrency,
|
||
// lockId defaults to `gbrain-sync:${src.id}` via the invariant in
|
||
// performSync (no explicit override needed — sourceId triggers it).
|
||
};
|
||
const result = await withSourcePrefix(src.id, () => performSync(engine, repoOpts));
|
||
return { result, log };
|
||
}
|
||
|
||
/**
|
||
* v0.40.3.0 — read-only per-source dashboard for `gbrain sources status`.
|
||
*
|
||
* Aggregates from existing tables (no schema changes):
|
||
* - sources: last_commit, last_sync_at, archived, config.syncEnabled
|
||
* (filtered: archived=false, local_path IS NOT NULL)
|
||
* - pages: per-source page count (excluding soft-deleted)
|
||
* - content_chunks: per-source total + count of unembedded chunks for
|
||
* the ACTIVE embedding column (resolved via the
|
||
* registry — see `src/core/search/embedding-column.ts`).
|
||
* Voyage / multimodal / non-default-column brains
|
||
* see counts against the column they actually use.
|
||
* - sync-failures.jsonl: unacknowledged failures (brain-global; the
|
||
* JSONL log isn't per-source. v0.40.4 TODO source-scopes it.)
|
||
*
|
||
* Staleness thresholds match `gbrain doctor`'s sync-freshness rule
|
||
* (24h / 72h). Sources that have NEVER synced (last_sync_at IS NULL)
|
||
* report `staleness_hours: null` so callers can disambiguate "first run
|
||
* pending" from "32h since last successful sync".
|
||
*
|
||
* Errors propagate. Pre-v0.40.3.0 the dashboard swallowed all DB errors
|
||
* and reported zero counts, which lied at exactly the moment it mattered
|
||
* (Q2 sub-fix from Codex review). The dashboard is read-only — a thrown
|
||
* error surfaces the real problem (DB down, permission denied, statement
|
||
* timeout) instead of misleading the operator with a "0 chunks" report.
|
||
*/
|
||
export interface SyncStatusReportSource {
|
||
source_id: string;
|
||
name: string;
|
||
local_path: string | null;
|
||
sync_enabled: boolean;
|
||
last_sync_at: string | null;
|
||
staleness_hours: number | null;
|
||
staleness_class: 'fresh' | 'stale' | 'severe' | 'unknown';
|
||
last_commit: string | null;
|
||
pages: number;
|
||
chunks_total: number;
|
||
chunks_unembedded: number;
|
||
embedding_coverage_pct: number;
|
||
}
|
||
|
||
export interface SyncStatusReport {
|
||
schema_version: 1;
|
||
generated_at: string;
|
||
sources: SyncStatusReportSource[];
|
||
unacknowledged_failures: number;
|
||
/** The embedding column counts were computed against. Useful for
|
||
* operators verifying their Voyage / multimodal setup is reported
|
||
* correctly. */
|
||
embedding_column: string;
|
||
}
|
||
|
||
export async function buildSyncStatusReport(
|
||
engine: BrainEngine,
|
||
sources: Array<{ id: string; name: string; local_path: string | null; config: Record<string, unknown> }>,
|
||
): Promise<SyncStatusReport> {
|
||
// Resolve the active embedding column via the registry. Brains pointed
|
||
// at Voyage / multimodal / any non-default column get accurate counts
|
||
// for the column they actually use (D16 → A, Codex P2 #10).
|
||
const { resolveEmbeddingColumn, quoteIdentifier } = await import('../core/search/embedding-column.ts');
|
||
// loadConfig() returns null when ~/.gbrain/config.json is missing.
|
||
// resolveEmbeddingColumn handles missing fields via its own
|
||
// gateway-fallback chain, so a minimal stub satisfies the call shape.
|
||
const cfg = loadConfig() ?? ({ engine: engine.kind } as Parameters<typeof resolveEmbeddingColumn>[1]);
|
||
const resolved = resolveEmbeddingColumn(undefined, cfg);
|
||
const embeddingColIdent = quoteIdentifier(resolved.name);
|
||
|
||
const sourceIds = sources.map((s) => s.id);
|
||
type SourceRow = {
|
||
id: string;
|
||
last_commit: string | null;
|
||
last_sync_at: string | Date | null;
|
||
};
|
||
type CountRow = {
|
||
source_id: string;
|
||
pages: string | number;
|
||
chunks_total: string | number;
|
||
chunks_unembedded: string | number;
|
||
};
|
||
|
||
// Pull last_commit + last_sync_at fresh (caller may have called us
|
||
// with stale rows). Empty source list → skip the round-trip.
|
||
const sourceRows = sourceIds.length === 0
|
||
? []
|
||
: await engine.executeRaw<SourceRow>(
|
||
`SELECT id, last_commit, last_sync_at FROM sources WHERE id = ANY($1::text[])`,
|
||
[sourceIds],
|
||
);
|
||
const sourceMap = new Map<string, SourceRow>();
|
||
for (const r of sourceRows) sourceMap.set(r.id, r);
|
||
|
||
// Per-source page + chunk + unembedded-chunk counts in a single
|
||
// round-trip. Canonical SQL (verified against
|
||
// src/commands/doctor.ts:2740): content_chunks joined on page_id
|
||
// (NOT page_slug — Codex P0 #1), filtered for non-soft-deleted pages
|
||
// (NOT NULL — soft-delete shipped v0.26.5), unembedded counted
|
||
// against the resolved active embedding column (D16 → A).
|
||
//
|
||
// No try/catch swallow — a thrown error means DB down / permission
|
||
// denied / statement timeout (NOT a schema variant). Surfacing the
|
||
// real error is better than a misleading "0 chunks" report (Q2).
|
||
let countRows: CountRow[] = [];
|
||
if (sourceIds.length > 0) {
|
||
countRows = await engine.executeRaw<CountRow>(
|
||
`WITH s AS (
|
||
SELECT unnest($1::text[]) AS source_id
|
||
)
|
||
SELECT
|
||
s.source_id,
|
||
COALESCE(p.pages, 0) AS pages,
|
||
COALESCE(c.chunks_total, 0) AS chunks_total,
|
||
COALESCE(c.chunks_unembedded, 0) AS chunks_unembedded
|
||
FROM s
|
||
LEFT JOIN (
|
||
SELECT source_id, COUNT(*) AS pages
|
||
FROM pages
|
||
WHERE deleted_at IS NULL
|
||
GROUP BY source_id
|
||
) p ON p.source_id = s.source_id
|
||
LEFT JOIN (
|
||
SELECT pg.source_id,
|
||
COUNT(*) AS chunks_total,
|
||
COUNT(*) FILTER (WHERE cc.${embeddingColIdent} IS NULL) AS chunks_unembedded
|
||
FROM content_chunks cc
|
||
JOIN pages pg ON pg.id = cc.page_id
|
||
WHERE pg.deleted_at IS NULL
|
||
GROUP BY pg.source_id
|
||
) c ON c.source_id = s.source_id`,
|
||
[sourceIds],
|
||
);
|
||
}
|
||
const countMap = new Map<string, { pages: number; chunks_total: number; chunks_unembedded: number }>();
|
||
for (const r of countRows) {
|
||
countMap.set(r.source_id, {
|
||
pages: Number(r.pages) || 0,
|
||
chunks_total: Number(r.chunks_total) || 0,
|
||
chunks_unembedded: Number(r.chunks_unembedded) || 0,
|
||
});
|
||
}
|
||
|
||
const now = Date.now();
|
||
const out: SyncStatusReportSource[] = sources.map((src) => {
|
||
const cfgEntry = (src.config || {}) as { syncEnabled?: boolean };
|
||
const row = sourceMap.get(src.id) || { id: src.id, last_commit: null, last_sync_at: null };
|
||
const counts = countMap.get(src.id) || { pages: 0, chunks_total: 0, chunks_unembedded: 0 };
|
||
const lastSyncMs = row.last_sync_at
|
||
? (row.last_sync_at instanceof Date ? row.last_sync_at.getTime() : Date.parse(row.last_sync_at))
|
||
: null;
|
||
const stalenessHours = lastSyncMs !== null && Number.isFinite(lastSyncMs)
|
||
? (now - lastSyncMs) / 3_600_000
|
||
: null;
|
||
let stalenessClass: 'fresh' | 'stale' | 'severe' | 'unknown' = 'unknown';
|
||
if (stalenessHours !== null) {
|
||
if (stalenessHours < 24) stalenessClass = 'fresh';
|
||
else if (stalenessHours < 72) stalenessClass = 'stale';
|
||
else stalenessClass = 'severe';
|
||
}
|
||
const embeddingCoveragePct = counts.chunks_total === 0
|
||
? 100
|
||
: Math.round(((counts.chunks_total - counts.chunks_unembedded) / counts.chunks_total) * 1000) / 10;
|
||
const lastSyncIso = row.last_sync_at
|
||
? (row.last_sync_at instanceof Date ? row.last_sync_at.toISOString() : row.last_sync_at)
|
||
: null;
|
||
return {
|
||
source_id: src.id,
|
||
name: src.name,
|
||
local_path: src.local_path,
|
||
sync_enabled: cfgEntry.syncEnabled !== false,
|
||
last_sync_at: lastSyncIso,
|
||
staleness_hours: stalenessHours === null ? null : Math.round(stalenessHours * 10) / 10,
|
||
staleness_class: stalenessClass,
|
||
last_commit: row.last_commit,
|
||
pages: counts.pages,
|
||
chunks_total: counts.chunks_total,
|
||
chunks_unembedded: counts.chunks_unembedded,
|
||
embedding_coverage_pct: embeddingCoveragePct,
|
||
};
|
||
});
|
||
|
||
// Unacknowledged sync failures — brain-global (the JSONL log isn't
|
||
// per-source). v0.40.4 TODO will source-scope this. Best-effort:
|
||
// missing file / parse error returns 0, doesn't throw the dashboard.
|
||
let unackedCount = 0;
|
||
try {
|
||
unackedCount = unacknowledgedSyncFailures().length;
|
||
} catch {
|
||
unackedCount = 0;
|
||
}
|
||
|
||
return {
|
||
schema_version: 1,
|
||
generated_at: new Date().toISOString(),
|
||
sources: out,
|
||
unacknowledged_failures: unackedCount,
|
||
embedding_column: resolved.name,
|
||
};
|
||
}
|
||
|
||
/**
|
||
* v0.40.3.0 — render a `SyncStatusReport` as a human-readable table.
|
||
*
|
||
* `sink` defaults to `process.stdout` so the bare `gbrain sources status`
|
||
* invocation writes its table to stdout. `--json` callers don't use
|
||
* this — they emit `JSON.stringify(report)` to stdout directly.
|
||
*/
|
||
export function printSyncStatusReport(
|
||
report: SyncStatusReport,
|
||
sink: NodeJS.WriteStream = process.stdout,
|
||
): void {
|
||
const write = (line: string) => sink.write(line + '\n');
|
||
write(`\nSync status — generated ${report.generated_at}`);
|
||
write(`Embedding column: ${report.embedding_column}\n`);
|
||
if (report.sources.length === 0) {
|
||
write(' (no sources registered)');
|
||
return;
|
||
}
|
||
const headers = ['SOURCE', 'STATE', 'STALENESS', 'PAGES', 'EMBEDDED', 'LAST SYNC'];
|
||
const rows = report.sources.map((s) => {
|
||
const stale = s.staleness_hours === null
|
||
? 'never'
|
||
: `${s.staleness_hours.toFixed(1)}h`;
|
||
const stateBits: string[] = [];
|
||
if (!s.sync_enabled) stateBits.push('disabled');
|
||
stateBits.push(s.staleness_class);
|
||
return [
|
||
s.name,
|
||
stateBits.join(','),
|
||
stale,
|
||
String(s.pages),
|
||
`${s.embedding_coverage_pct}%`,
|
||
s.last_sync_at ?? '(never)',
|
||
];
|
||
});
|
||
const widths = headers.map((h, i) =>
|
||
Math.max(h.length, ...rows.map((r) => r[i].length)),
|
||
);
|
||
// Numeric columns (PAGES at index 3, EMBEDDED at index 4, STALENESS
|
||
// at index 2) right-pad-left so digits align cleanly. Text columns
|
||
// left-pad-right per the existing `sources list` convention.
|
||
const NUMERIC_COLS = new Set([2, 3, 4]);
|
||
const fmt = (cells: string[]) =>
|
||
cells.map((c, i) => (NUMERIC_COLS.has(i) ? c.padStart(widths[i]) : c.padEnd(widths[i]))).join(' ');
|
||
write(fmt(headers));
|
||
write(fmt(widths.map((w) => '-'.repeat(w))));
|
||
for (const r of rows) write(fmt(r));
|
||
write(`\nUnacknowledged sync failures (brain-wide): ${report.unacknowledged_failures}`);
|
||
const severe = report.sources.filter((s) => s.staleness_class === 'severe').length;
|
||
if (severe > 0) {
|
||
write(`WARNING: ${severe} source(s) are SEVERELY stale (>72h). Run \`gbrain sync --all\` to refresh.`);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Auto-manage .gitignore entries for db_only directories.
|
||
*
|
||
* Caller invokes ONLY on successful sync — this function trusts that the
|
||
* sync's data state is consistent. See `runSync` for the gating logic.
|
||
*
|
||
* Idempotent: re-running adds no duplicate entries. The managed block has
|
||
* a stable comment header so it's grep-able and editable.
|
||
*
|
||
* Skipped (with actionable warning) when:
|
||
* - GBRAIN_NO_GITIGNORE=1 — D23 escape hatch for shared-repo setups
|
||
* - The repo is a git submodule (`.git` is a file not a directory) —
|
||
* D49 lock; submodule .gitignore changes don't survive parent updates
|
||
*
|
||
* On PGLite (D4): emits a once-per-process soft-warn explaining that
|
||
* tiering has limited effect — but still manages the .gitignore so the
|
||
* config-present user gets the gitignore housekeeping.
|
||
*
|
||
* Failures (write permission denied, EROFS, etc.) are caught, warned, and
|
||
* swallowed (D9 lock). Sync's primary job is moving data; .gitignore
|
||
* management is a side effect — don't kill the main job for the side effect.
|
||
*/
|
||
let _pgliteTierWarned = false;
|
||
export function __resetPGLiteTierWarn(): void {
|
||
_pgliteTierWarned = false;
|
||
}
|
||
|
||
export function manageGitignore(
|
||
repoPath: string,
|
||
engineKind?: 'pglite' | 'postgres',
|
||
): void {
|
||
if (process.env.GBRAIN_NO_GITIGNORE === '1') {
|
||
return;
|
||
}
|
||
|
||
// Submodule + worktree detection (closes #889 misclassification).
|
||
// Both submodules and worktrees use `.git` as a FILE (not a directory), so
|
||
// statSync.isFile() doesn't discriminate. Discriminator is the gitdir path
|
||
// segment:
|
||
// - submodule: gitdir contains `/modules/<name>` (skip — managed by parent)
|
||
// - worktree: gitdir contains `/worktrees/<name>` (MANAGE — first-class repo)
|
||
// Both contracts are documented Git internal layouts and stable across all 4
|
||
// {relative, absolute} × {modules, worktrees} combinations, including the
|
||
// absorbed-submodule case from `git submodule absorbgitdirs`.
|
||
// Malformed `.git` file (no `gitdir:` prefix, unreadable) → MANAGE (fail-closed
|
||
// toward managing, preserving the pre-#889 catch{} behavior).
|
||
const dotGit = join(repoPath, '.git');
|
||
if (existsSync(dotGit)) {
|
||
try {
|
||
if (statSync(dotGit).isFile()) {
|
||
const content = readFileSync(dotGit, 'utf-8');
|
||
const match = content.match(/gitdir:\s*(.+)/);
|
||
const gitdir = match ? match[1].trim() : '';
|
||
if (gitdir.includes('/modules/')) {
|
||
console.warn(
|
||
`Note: skipping .gitignore management — ${repoPath} is a git submodule. ` +
|
||
`Add db_only directories to your parent repo's .gitignore manually.`,
|
||
);
|
||
return;
|
||
}
|
||
// Worktree (gitdir contains /worktrees/) OR malformed .git falls through
|
||
// to the existing manage path. Worktrees are first-class repos — they
|
||
// need .gitignore management too. Malformed → MANAGE preserves the
|
||
// pre-#889 fail-closed-toward-managing catch behavior.
|
||
}
|
||
} catch {
|
||
// proceed; can't tell, default to managing
|
||
}
|
||
}
|
||
|
||
let storageConfig;
|
||
try {
|
||
storageConfig = loadStorageConfig(repoPath);
|
||
} catch (error) {
|
||
// StorageConfigError (overlap) or read error — surface, don't manage.
|
||
console.warn(
|
||
`Skipped .gitignore update: ${error instanceof Error ? error.message : String(error)}`,
|
||
);
|
||
return;
|
||
}
|
||
if (!storageConfig || storageConfig.db_only.length === 0) {
|
||
return;
|
||
}
|
||
|
||
// D4 soft-warn: storage tiering has limited effect on PGLite, but the
|
||
// .gitignore housekeeping still helps. Warn once per process; proceed.
|
||
if (engineKind === 'pglite' && !_pgliteTierWarned) {
|
||
_pgliteTierWarned = true;
|
||
console.warn(
|
||
`Note: storage tiering has limited effect on PGLite — pages live in your ` +
|
||
`local database file regardless of tier. Managing .gitignore anyway.`,
|
||
);
|
||
}
|
||
|
||
const gitignorePath = join(repoPath, '.gitignore');
|
||
let gitignoreContent = '';
|
||
|
||
if (existsSync(gitignorePath)) {
|
||
try {
|
||
gitignoreContent = readFileSync(gitignorePath, 'utf-8');
|
||
} catch (error) {
|
||
console.warn(
|
||
`Could not read ${gitignorePath} (${error instanceof Error ? error.message : String(error)}) — ` +
|
||
`skipping .gitignore update. Add db_only directories manually.`,
|
||
);
|
||
return;
|
||
}
|
||
}
|
||
|
||
const existingLines = new Set(gitignoreContent.split('\n').map((line) => line.trim()));
|
||
const linesToAdd: string[] = [];
|
||
|
||
for (const dir of storageConfig.db_only) {
|
||
if (!existingLines.has(dir) && !existingLines.has(`/${dir}`)) {
|
||
linesToAdd.push(dir);
|
||
}
|
||
}
|
||
|
||
if (linesToAdd.length === 0) return;
|
||
|
||
if (gitignoreContent && !gitignoreContent.endsWith('\n')) {
|
||
gitignoreContent += '\n';
|
||
}
|
||
gitignoreContent += '\n# Auto-managed by gbrain (db_only directories)\n';
|
||
gitignoreContent += linesToAdd.join('\n') + '\n';
|
||
|
||
try {
|
||
writeFileSync(gitignorePath, gitignoreContent);
|
||
} catch (error) {
|
||
console.warn(
|
||
`Could not update ${gitignorePath} (${error instanceof Error ? error.message : String(error)}) — ` +
|
||
`please add db_only directories manually:\n ${linesToAdd.join('\n ')}`,
|
||
);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Render a SyncResult to a Writable sink.
|
||
*
|
||
* `sink` defaults to `process.stdout` so existing single-source callers
|
||
* see identical output. The `--all` parallel path passes `process.stderr`
|
||
* when `--json` is set, so banners stay off stdout and the JSON envelope
|
||
* pipes cleanly through `jq` (D4).
|
||
*/
|
||
function printSyncResult(result: SyncResult, sink: NodeJS.WriteStream = process.stdout) {
|
||
const write = (line: string) => sink.write(line + '\n');
|
||
switch (result.status) {
|
||
case 'up_to_date':
|
||
write('Already up to date.');
|
||
break;
|
||
case 'synced':
|
||
write(`Synced ${result.fromCommit?.slice(0, 8)}..${result.toCommit.slice(0, 8)}:`);
|
||
write(` +${result.added} added, ~${result.modified} modified, -${result.deleted} deleted, R${result.renamed} renamed`);
|
||
write(` ${result.chunksCreated} chunks created${result.embedded > 0 ? `, ${result.embedded} pages embedded` : ''}`);
|
||
break;
|
||
case 'first_sync':
|
||
write(`First sync complete. Checkpoint: ${result.toCommit.slice(0, 8)}`);
|
||
write(` ${result.added} file(s) imported, ${result.chunksCreated} chunks${result.embedded > 0 ? `, ${result.embedded} pages embedded` : ''}`);
|
||
break;
|
||
case 'dry_run':
|
||
break; // already printed in performSync
|
||
case 'blocked_by_failures':
|
||
write(`Sync BLOCKED at ${result.toCommit.slice(0, 8)}: ${result.failedFiles ?? 0} file(s) failed to parse.`);
|
||
write(` See ~/.gbrain/sync-failures.jsonl for details, or run 'gbrain doctor'.`);
|
||
write(` Fix the files then re-run 'gbrain sync', or 'gbrain sync --skip-failed' to move on.`);
|
||
break;
|
||
}
|
||
}
|