Compare commits

..
Author SHA1 Message Date
Garry TanandClaude Fable 5 c3e9daa18f test(sync): pin the config-fingerprint gate wiring end-to-end (#2156)
The primitives (compute/read/write) were pinned but nothing exercised the
actual performSync gate: glob drift with git HEAD unchanged must break out
of the 'Already up to date' early return, force a full re-walk that imports
newly-included files, and settle back to up_to_date once re-stamped.
Deleting the configMismatch term from the gate now fails a committed test.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-22 11:10:48 -07:00
f28613178a feat(sources): persist + honor --include / --exclude globs across sync and lint (#2156)
Takeover of PR #2157, rebased onto current master.

- gbrain sources add --include/--exclude (repeatable) persist into
  sources.config.include_globs / exclude_globs; every subsequent sync
  of the source honors them.
- gbrain sync gains --include (allow-list counterpart to the existing
  #753/#774 --exclude), merged with the source row's persisted globs.
  Include/exclude are matched scope-relative in both the incremental
  and full-sync paths (import.ts gains the include filter); exclusion
  stays conservative (never deletes previously-imported pages).
- gbrain lint gains --include/--exclude and auto-lifts the persisted
  source globs when the lint target matches a source's local_path.
- Migration v125 (renumbered from the PR's v117/v120): a
  sources.config_fingerprint column caches a SHA-256 of the
  walk-affecting config fields (strategy + globs) so changing globs on
  an already-synced source forces a full re-walk instead of "Already
  up to date" (git HEAD unchanged). NULL = never stamped; first
  post-upgrade sync stamps quietly.

Deviations from #2157 while rebasing: globs are NOT threaded through
isSyncable/unsyncableReason in the incremental path — the unsyncable
cleanup loop deletes pages for non-metafile classifications, which
would violate the documented conservative #1433 posture; instead
include mirrors master's existing scope-relative excluded() helper.
CLI --exclude/--include now merge (union) with persisted config globs
rather than being replaced by them.

Fixes #2156
Takeover of #2157

Co-authored-by: brettdavies <brettdavies@users.noreply.github.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-21 14:36:26 -07:00
30 changed files with 1544 additions and 648 deletions
+38
View File
@@ -115,6 +115,7 @@ Full subcommand reference:
```
gbrain sources add <id> --path <p> [--name <n>] [--federated|--no-federated] [--force]
[--include <glob>...] [--exclude <glob>...]
Register a source. id: [a-z0-9](?:[a-z0-9-]{0,30}[a-z0-9])?
--path must be a git repo (or a subdirectory of one) — see
"The git requirement for --path sources" below. --force
@@ -131,6 +132,43 @@ gbrain sources federate <id>
gbrain sources unfederate <id>
```
## Filtering what gets synced (--include / --exclude)
`--include` and `--exclude` on `gbrain sources add` accept repeatable glob
patterns and are honored by every subsequent sync AND lint of the source.
Common Obsidian vault setups need to exclude authoring scaffolding so it
doesn't pollute search:
```bash
# Skip Templates/, Drafts/, and the smart-env sidecar; everything else syncs.
gbrain sources add vault \
--path ~/Documents/vault --federated \
--exclude 'Templates/**' \
--exclude 'Drafts/**' \
--exclude '.smart-env/**'
# Or: only sync the people/ and companies/ subtrees of a CRM vault.
gbrain sources add crm \
--path ~/Documents/crm --no-federated \
--include 'people/**' \
--include 'companies/**'
```
Both persist into `sources.config.include_globs` / `exclude_globs` arrays.
The filter runs `include` first, then `exclude`, so a path inside
`people/**` is still rejected if it also matches `exclude_globs`. Globs use
the same matcher as the rest of gbrain's sync classifier (`matchesAnyGlob`
in `src/core/sync.ts`) and are matched against the source-root-relative
path. Exclusion is conservative: it never deletes previously-imported pages.
`gbrain sync --include <glob> --exclude <glob>` and
`gbrain lint <dir> --include <glob> --exclude <glob>` take the same
repeatable flags for one-off scope changes; for lint the persisted source
globs are auto-applied when the lint target matches a source's `local_path`.
Changing the persisted globs on an existing source triggers a full re-walk
on the next sync (the source's config fingerprint invalidates the
"already up to date" gate).
## The git requirement for --path sources
Every `--path` source must be a git repository (or live inside one — a
+11
View File
@@ -53,6 +53,13 @@ export async function runImport(
strategy?: SyncStrategy;
sourceId?: string;
managedBookmark?: boolean;
/**
* #2156: allow-list glob patterns — only dir-relative paths matching at
* least one pattern are imported. Applied BEFORE `exclude`. Threaded by
* performFullSync from `gbrain sync --include` / the source row's
* persisted `config.include_globs`.
*/
include?: string[];
/**
* #753/#774: glob patterns to exclude from the import (same semantics as
* `isSyncable`'s `exclude` — matched against the dir-relative path).
@@ -215,6 +222,10 @@ export async function runImport(
);
const fileTypeLabel = strategy === 'code' ? 'code'
: strategy === 'auto' ? 'syncable' : 'markdown';
// #2156: apply --include allow-list globs first (threaded by performFullSync).
if (opts.include && opts.include.length > 0) {
allFiles = allFiles.filter(abs => matchesAnyGlob(relative(dir, abs), opts.include));
}
// #753/#774: apply --exclude glob patterns (threaded by performFullSync).
if (opts.exclude && opts.exclude.length > 0) {
const beforeExclude = allFiles.length;
-2
View File
@@ -2059,8 +2059,6 @@ export async function registerBuiltinHandlers(
sourceId,
windowSeconds,
brainDir: repoPath,
// #2750: worker cancel/timeout/lock-loss propagates into the drain.
abortSignal: job.signal,
});
} catch (e) {
if (e instanceof LockUnavailableError) {
+172 -12
View File
@@ -17,7 +17,7 @@
*/
import { readFileSync, writeFileSync, readdirSync, statSync, lstatSync, existsSync } from 'fs';
import { join, relative } from 'path';
import { join, relative, resolve } from 'path';
import { isAborted } from '../core/abort-check.ts';
import { parseMarkdown, type ParseValidationCode } from '../core/markdown.ts';
import {
@@ -26,7 +26,9 @@ import {
DEFAULT_BYTES_WARN,
} from '../core/content-sanity.ts';
import { loadOperatorLiterals } from '../core/content-sanity-literals.ts';
import { loadConfig, loadConfigWithEngine, gbrainPath } from '../core/config.ts';
import { loadConfig, loadConfigWithEngine, toEngineConfig, gbrainPath } from '../core/config.ts';
import { matchesAnyGlob } from '../core/sync.ts';
import { parseGlobList } from './sync.ts';
import type { BrainEngine } from '../core/engine.ts';
export interface LintIssue {
@@ -378,21 +380,89 @@ async function resolveLintContentSanity(
};
}
/** Collect markdown files from a directory */
function collectPages(dir: string): string[] {
/** Collect markdown files from a directory.
*
* When `opts.include` or `opts.exclude` are set, each candidate `.md` path's
* POSIX-style relative path (relative to `dir`) is matched against the same
* glob semantics sync uses (`matchesAnyGlob`). `include` allow-lists;
* `exclude` deny-lists. Empty or undefined arrays leave the filter
* unengaged. Symmetric with `isSyncable` in `src/core/sync.ts` so a
* source-config `exclude_globs` honored by `gbrain sync` is also honored
* by `gbrain lint` against the same dir.
*/
function collectPages(
dir: string,
opts: { include?: string[]; exclude?: string[] } = {},
): string[] {
const { include, exclude } = opts;
const haveInclude = !!(include && include.length > 0);
const haveExclude = !!(exclude && exclude.length > 0);
const pages: string[] = [];
function walk(d: string) {
for (const entry of readdirSync(d)) {
if (entry.startsWith('.') || entry.startsWith('_')) continue;
const full = join(d, entry);
if (lstatSync(full).isDirectory()) walk(full);
else if (entry.endsWith('.md')) pages.push(full);
else if (entry.endsWith('.md')) {
if (haveInclude || haveExclude) {
// Match against the path RELATIVE to `dir` (the source root),
// normalized to POSIX separators by matchesAnyGlob. A
// source-config glob like `Resources/veriff/**` is anchored at
// the source root; matching against the absolute path would
// require the user to anchor on their `$HOME` or repo prefix,
// which is brittle.
const rel = relative(dir, full);
if (haveInclude && !matchesAnyGlob(rel, include)) continue;
if (haveExclude && matchesAnyGlob(rel, exclude)) continue;
}
pages.push(full);
}
}
}
walk(dir);
return pages.sort();
}
/** Look up the source row whose `local_path` resolves to the same absolute
* directory as `target`, and return its persisted `include_globs` /
* `exclude_globs` as parsed string arrays. Returns an empty object when no
* matching source exists, when the row has no globs configured, or when the
* lookup throws (best-effort — auto-resolution must never break standalone
* lint on brains without a sources table).
*
* Mirrors how `syncOneSource` lifts the same fields off `src.config` before
* threading them into `SyncOpts.include` / `SyncOpts.exclude`.
*/
async function resolveSourceGlobsForTarget(
engine: BrainEngine,
target: string,
): Promise<{ include?: string[]; exclude?: string[] }> {
try {
const absTarget = resolve(target);
const rows = await engine.executeRaw<{ config: unknown }>(
`SELECT config FROM sources
WHERE archived IS NOT TRUE
AND local_path IS NOT NULL
AND local_path = $1
LIMIT 1`,
[absTarget],
);
if (rows.length === 0) return {};
const cfg = (rows[0].config && typeof rows[0].config === 'object')
? rows[0].config as Record<string, unknown>
: {};
return {
include: parseGlobList(cfg.include_globs),
exclude: parseGlobList(cfg.exclude_globs),
};
} catch {
// Engine not connected, sources table missing on a fresh brain, RLS
// denial in an unusual scope — all best-effort. Lint proceeds without
// filtering rather than fail-closed.
return {};
}
}
export interface LintOpts {
target: string;
fix?: boolean;
@@ -414,6 +484,22 @@ export interface LintOpts {
* yields + checks this every 200 pages.
*/
signal?: AbortSignal;
/**
* Glob filters threaded into the file walker. When set, paths relative to
* `target` are matched against the patterns using the same semantics as
* `gbrain sync` (`matchesAnyGlob` in `src/core/sync.ts`). `include`
* allow-lists; `exclude` deny-lists; both unset == no filter.
*
* When BOTH are unset AND `engine` is provided, `runLintCore` attempts to
* auto-resolve them from the `sources` row whose `local_path` matches
* `target` — symmetric with `syncOneSource`, so a user who has run
* `gbrain sources add --exclude 'Resources/veriff/**'` sees the same
* exclusion applied to `gbrain lint <same-dir>` and to the cycle.lint
* phase without restating it on every invocation. Explicit caller-supplied
* arrays always win over the source-row lift.
*/
include?: string[];
exclude?: string[];
}
export interface LintResult {
@@ -440,7 +526,21 @@ export async function runLintCore(opts: LintOpts): Promise<LintResult> {
}
const isSingleFile = statSync(opts.target).isFile();
const pages = isSingleFile ? [opts.target] : collectPages(opts.target);
// Resolve glob filters. Explicit caller-supplied include/exclude win;
// otherwise lift from `sources.config.{include,exclude}_globs` when an
// engine is available and the target matches a known source's local_path.
// Single-file lints skip the resolve entirely — globs are a directory
// walk concern.
let include = opts.include;
let exclude = opts.exclude;
const haveExplicit = (include && include.length > 0) || (exclude && exclude.length > 0);
if (!isSingleFile && !haveExplicit && opts.engine) {
const resolved = await resolveSourceGlobsForTarget(opts.engine, opts.target);
include = resolved.include;
exclude = resolved.exclude;
}
const pages = isSingleFile ? [opts.target] : collectPages(opts.target, { include, exclude });
// Resolve content-sanity config once for this lint run (D1: lift DB
// config when reachable). Caller can pre-pass via opts.contentSanity
@@ -491,14 +591,27 @@ export async function runLintCore(opts: LintOpts): Promise<LintResult> {
}
export async function runLint(args: string[]) {
const target = args.find(a => !a.startsWith('--'));
const target = args.find(a => !a.startsWith('--') && !args[args.indexOf(a) - 1]?.match(/^--(include|exclude)$/));
const doFix = args.includes('--fix');
const dryRun = args.includes('--dry-run');
// Parse repeatable `--include <glob>` and `--exclude <glob>` flags.
// Symmetric with `gbrain sources add --include / --exclude` from PR #2157;
// explicit flags here override the source-config lift performed below for
// dir-mode lints.
const cliInclude: string[] = [];
const cliExclude: string[] = [];
for (let i = 0; i < args.length; i++) {
if (args[i] === '--include' && i + 1 < args.length) cliInclude.push(args[++i]);
else if (args[i] === '--exclude' && i + 1 < args.length) cliExclude.push(args[++i]);
}
if (!target) {
console.error('Usage: gbrain lint <dir|file.md> [--fix] [--dry-run]');
console.error(' --fix Auto-fix fixable issues (LLM preambles, code fences)');
console.error(' --dry-run Preview fixes without writing');
console.error('Usage: gbrain lint <dir|file.md> [--fix] [--dry-run] [--include <glob>]... [--exclude <glob>]...');
console.error(' --fix Auto-fix fixable issues (LLM preambles, code fences)');
console.error(' --dry-run Preview fixes without writing');
console.error(' --include <glob> Repeatable; only lint paths matching at least one pattern');
console.error(' --exclude <glob> Repeatable; skip paths matching any pattern (applied after --include)');
process.exit(1);
}
@@ -510,7 +623,44 @@ export async function runLint(args: string[]) {
// Single file or directory — print human detail as we go, then rely on
// Core for the aggregate numbers at the end.
const isSingleFile = statSync(target).isFile();
const pages = isSingleFile ? [target] : collectPages(target);
// Resolve glob filters for directory lints. Explicit CLI flags win;
// otherwise lift from `sources.config.{include,exclude}_globs` matching
// `target`. Connect a transient engine for the lookup only when (a) no
// explicit flags were passed AND (b) file/env config suggests an engine is
// available — mirrors the connect-disconnect pattern in
// `resolveLintContentSanity` (issue #1678: standalone CLI never shares the
// db.ts singleton, so create + dispose here is safe).
let runInclude: string[] | undefined = cliInclude.length > 0 ? cliInclude : undefined;
let runExclude: string[] | undefined = cliExclude.length > 0 ? cliExclude : undefined;
if (!isSingleFile && runInclude === undefined && runExclude === undefined) {
const base = loadConfig();
if (base?.database_url || base?.database_path) {
try {
const { createEngine } = await import('../core/engine-factory.ts');
const { connectWithRetry } = await import('../core/db.ts');
const engineCfg = toEngineConfig(base);
const engine = await createEngine(engineCfg);
try {
// Use the same connect path the rest of the CLI uses
// (`connectEngine` in cli.ts). `engine.connect({})` with empty
// opts drops the URL — confirmed by direct probe. `noRetry: true`
// keeps the standalone lint snappy (no retry tax when the brain
// happens to be unreachable; auto-resolve degrades to no-filter).
await connectWithRetry(engine, engineCfg, { noRetry: true });
const lifted = await resolveSourceGlobsForTarget(engine, target);
runInclude = lifted.include;
runExclude = lifted.exclude;
} finally {
await engine.disconnect().catch(() => { /* best-effort */ });
}
} catch {
// best-effort; fall through to no-filter
}
}
}
const pages = isSingleFile ? [target] : collectPages(target, { include: runInclude, exclude: runExclude });
// Progress on stderr. Stdout keeps the per-issue human output it always had.
const { createProgress } = await import('../core/progress.ts');
@@ -557,7 +707,17 @@ export async function runLint(args: string[]) {
// produces canonical numbers for the summary line).
// Pass contentSanity through so runLintCore skips its own resolve
// (we already resolved once for the human-detail loop above).
const result = await runLintCore({ target, fix: doFix, dryRun, contentSanity });
// Pass include/exclude so the aggregate scope matches the human-detail
// walk above — otherwise the summary line reports the unfiltered count
// even though the per-page details were already filtered.
const result = await runLintCore({
target,
fix: doFix,
dryRun,
contentSanity,
include: runInclude,
exclude: runExclude,
});
console.log(`\n${result.pages_scanned} pages scanned. ${result.total_issues} issue(s) in ${result.pages_with_issues} page(s).`);
if (doFix) {
console.log(`${dryRun ? '(dry run) ' : ''}${result.total_fixed} auto-fixed.`);
+34 -1
View File
@@ -122,7 +122,8 @@ async function runAdd(engine: BrainEngine, args: string[]): Promise<void> {
if (!id) {
console.error(
'Usage: gbrain sources add <id> [--path <path> | --url <https-url>] ' +
'[--name <display>] [--federated|--no-federated] [--clone-dir <path>] [--force]',
'[--name <display>] [--federated|--no-federated] [--clone-dir <path>] [--force] ' +
'[--include <glob>...] [--exclude <glob>...]',
);
process.exit(2);
}
@@ -135,6 +136,12 @@ async function runAdd(engine: BrainEngine, args: string[]): Promise<void> {
let patFile: string | undefined;
let noHarden = false;
let force = false;
// Repeatable. `--include 'people/**' --include 'companies/**'` accumulates.
// Persisted into sources.config.include_globs / .exclude_globs and read at
// sync time by commands/sync.ts so `Templates/`, `.smart-env/`, `Drafts/`
// and other vault scaffolding can be skipped without renaming directories.
const includeGlobs: string[] = [];
const excludeGlobs: string[] = [];
for (let i = 1; i < args.length; i++) {
const a = args[i];
@@ -147,6 +154,24 @@ async function runAdd(engine: BrainEngine, args: string[]): Promise<void> {
if (a === '--pat-file') { patFile = args[++i]; continue; }
if (a === '--no-harden') { noHarden = true; continue; }
if (a === '--force') { force = true; continue; }
if (a === '--include') {
const v = args[++i];
if (!v || v.startsWith('--')) {
console.error('Error: --include requires a glob argument (e.g. --include "people/**")');
process.exit(2);
}
includeGlobs.push(v);
continue;
}
if (a === '--exclude') {
const v = args[++i];
if (!v || v.startsWith('--')) {
console.error('Error: --exclude requires a glob argument (e.g. --exclude "Templates/**")');
process.exit(2);
}
excludeGlobs.push(v);
continue;
}
console.error(`Unknown flag: ${a}`);
process.exit(2);
}
@@ -167,6 +192,8 @@ async function runAdd(engine: BrainEngine, args: string[]): Promise<void> {
federated,
cloneDir,
force,
includeGlobs: includeGlobs.length > 0 ? includeGlobs : undefined,
excludeGlobs: excludeGlobs.length > 0 ? excludeGlobs : undefined,
});
// Topology A discovery: if the just-added source carries a brain-resident
@@ -190,6 +217,12 @@ async function runAdd(engine: BrainEngine, args: string[]): Promise<void> {
console.log(
` federated: ${fed}${fed ? ' — appears in cross-source default search' : ' — only searched when explicitly named via --source'}`,
);
if (includeGlobs.length > 0) {
console.log(` include globs: ${includeGlobs.join(', ')}`);
}
if (excludeGlobs.length > 0) {
console.log(` exclude globs: ${excludeGlobs.join(', ')}`);
}
// v0.42.44 — auto-harden managed clones for git durability the moment a brain
// repo is added with a PAT. Best-effort: NEVER fail `add` if hardening fails.
+256 -18
View File
@@ -1,6 +1,7 @@
import { existsSync, readFileSync, writeFileSync, statSync, realpathSync } from 'fs';
import { execFileSync } from 'child_process';
import { join, relative } from 'path';
import { createHash } from 'crypto';
import type { BrainEngine } from '../core/engine.ts';
import { DELETE_BATCH_SIZE } from '../core/engine-constants.ts';
import { importFile } from '../core/import-file.ts';
@@ -756,12 +757,23 @@ export interface SyncOpts {
* are rejected before any git op runs.
*/
srcSubpath?: string;
/**
* #2156 glob patterns files must match to be synced (allow-list).
* Populated from the source row's persisted `config.include_globs`
* (set via `gbrain sources add --include <glob>`) or the repeatable
* `--include` CLI flag. Matched against the scope-relative path, same
* anchoring as `exclude`. `exclude` is applied after `include`: a path
* matching an include pattern is still rejected if it also matches an
* exclude pattern. Empty arrays are the same as undefined (no filter).
*/
include?: string[];
/**
* #753/#774 glob patterns for files to exclude from sync (repeatable
* `--exclude` on the CLI). Matched against the scope-relative path in both
* the full-sync and incremental paths. Excluded files are never imported;
* exclusion does NOT delete previously-imported pages (conservative,
* matching the #1433 metafile posture).
* `--exclude` on the CLI; #2156: also populated from the source row's
* persisted `config.exclude_globs`). Matched against the scope-relative
* path in both the full-sync and incremental paths. Excluded files are
* never imported; exclusion does NOT delete previously-imported pages
* (conservative, matching the #1433 metafile posture).
*/
exclude?: string[];
/**
@@ -1153,6 +1165,29 @@ function unique<T>(items: T[]): T[] {
// `src/core/sync-delta.ts` (re-imported below) so the inline cost estimator
// prices detached sources through the same code the executor imports them with.
/**
* Defensive parse for the JSONB-loaded `config.include_globs` / `config.exclude_globs`
* arrays read off the sources row. The column is a free-form JSONB and could
* contain anything coerce to a string-only array, drop empties, and return
* undefined when the result has no useful entries so the caller can decide
* not to engage glob-filtering at all.
*/
export function parseGlobList(value: unknown): string[] | undefined {
if (!Array.isArray(value)) return undefined;
const globs = value.filter((v): v is string => typeof v === 'string' && v.length > 0);
return globs.length > 0 ? globs : undefined;
}
/**
* Union of CLI-supplied glob patterns (one-off, this invocation) and the
* source row's persisted config globs (every sync). Deduped; undefined when
* neither side has entries so `SyncOpts` stays unset and no filter engages.
*/
export function mergeGlobs(cli: string[], persisted: string[] | undefined): string[] | undefined {
const merged = [...new Set([...cli, ...(persisted ?? [])])];
return merged.length > 0 ? merged : undefined;
}
// 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
@@ -1315,6 +1350,125 @@ async function writeChunkerVersion(
);
}
/**
* #2157 follow-on: detect when sources.config has shifted in a way that
* affects which paths the walker will include this run. The "Already up
* to date" gate at performSync's git-HEAD equality check honored chunker
* version match but ignored config drift a user who changes
* `sources.config.exclude_globs` mid-life got "Already up to date" on
* the next sync because git HEAD was unchanged, with no observable
* effect until `gbrain sync --full`.
*
* Fingerprint covers exactly the walk-affecting fields that flow from
* `sources.config` into `SyncOpts` at the syncOneSource call site:
* `strategy`, `include_globs`, `exclude_globs`. CLI-supplied --include
* / --exclude overrides do NOT participate they are one-off scope
* changes, not source state, and shouldn't invalidate the row's
* checkpoint. (A user running `gbrain sync --exclude X` on a row whose
* stored config has no X is intentionally narrowing this one pass; on
* the next no-flags sync, the row config governs again.)
*
* Array order is normalized (alphabetical, post-defensive-parse) so
* `["a/**", "b/**"]` and `["b/**", "a/**"]` fingerprint identically.
* `parseGlobList` shares the same defensive coercion as the call site
* that builds SyncOpts, so hand-edited or pre-normalization rows
* fingerprint to the same shape the walker actually sees.
*/
export function computeSourceConfigFingerprint(rawConfig: unknown): string {
const cfg = (rawConfig || {}) as {
strategy?: unknown;
include_globs?: unknown;
exclude_globs?: unknown;
};
const canonical = JSON.stringify({
strategy: typeof cfg.strategy === 'string' ? cfg.strategy : null,
include_globs: (parseGlobList(cfg.include_globs) ?? []).slice().sort(),
exclude_globs: (parseGlobList(cfg.exclude_globs) ?? []).slice().sort(),
});
return createHash('sha256').update(canonical).digest('hex');
}
/**
* Read the per-source fingerprint stamp. NULL on pre-migration rows or
* sources that have never been synced the gate treats NULL as
* "fingerprint unknown" and skips the invalidation check so first-time
* post-upgrade syncs don't spuriously force-full.
*/
export async function readConfigFingerprint(
engine: BrainEngine,
sourceId: string | undefined,
): Promise<string | null> {
if (!sourceId) return null;
const rows = await engine.executeRaw<{ config_fingerprint: string | null }>(
`SELECT config_fingerprint FROM sources WHERE id = $1`,
[sourceId],
);
return rows[0]?.config_fingerprint ?? null;
}
export async function writeConfigFingerprint(
engine: BrainEngine,
sourceId: string | undefined,
fingerprint: string,
): Promise<void> {
if (!sourceId) return;
await engine.executeRaw(
`UPDATE sources SET config_fingerprint = $1 WHERE id = $2`,
[fingerprint, sourceId],
);
}
/**
* Read the raw `sources.config` value for the named source. Returns an
* empty object for missing or never-configured rows. The reader is
* defensive about legacy double-encoded JSONB rows (`{"federated":true}`
* stored as a JSON string scalar, the #2339 class) the engine's
* `r.config` may arrive as either a string or an object, and both are
* normalized to an object before the fingerprint computation walks the
* keys.
*/
async function readSourceConfig(
engine: BrainEngine,
sourceId: string | undefined,
): Promise<unknown> {
if (!sourceId) return {};
const rows = await engine.executeRaw<{ config: unknown }>(
`SELECT config FROM sources WHERE id = $1`,
[sourceId],
);
const raw = rows[0]?.config;
if (raw === null || raw === undefined) return {};
if (typeof raw === 'string') {
try { return JSON.parse(raw); } catch { return {}; }
}
return raw;
}
/**
* Read-hash-stamp wrapper for sync-completion sites outside the gate's
* scope (e.g. `performFullSync`'s `advanceFull` closure, which doesn't
* see `performSync`'s cached `currentConfigFp` because it's a separate
* function). Reads the row's current config and stamps a fresh
* fingerprint.
*
* Race note: if `sources.config` was mutated between the gate's read
* and this stamp, the freshly-read value wins. The walker still used
* the gate-time effective globs (already captured into `opts.include` /
* `opts.exclude` upstream), so the stamp can drift from what was
* actually walked. In practice mid-sync mutations are rare and the
* NEXT sync will re-evaluate against the latest config anyway, so the
* minor staleness is acceptable and avoids threading the gate-time
* fingerprint through every helper signature.
*/
async function stampSourceConfigFingerprint(
engine: BrainEngine,
sourceId: string | undefined,
): Promise<void> {
if (!sourceId) return;
const cfg = await readSourceConfig(engine, sourceId);
await writeConfigFingerprint(engine, sourceId, computeSourceConfigFingerprint(cfg));
}
/**
* v0.40 Federated Sync v2: `gbrain sync trigger --source <id> [--priority high|normal|low]`
*
@@ -2163,7 +2317,25 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
detachedWorkingTreeManifest.deleted.length > 0 ||
detachedWorkingTreeManifest.renamed.length > 0);
if (lastCommit === headCommit && !versionMismatch && !versionNeverSet && !hasDetachedWorkingTreeChanges) {
// #2157 follow-on: parallel gate for sources.config drift. Without
// this, changing `sources.config.exclude_globs` (or include_globs /
// strategy) on a synced source had no observable effect on the next
// sync because git HEAD was unchanged — the "Already up to date"
// branch below returned without re-walking. Mismatch path mirrors the
// chunker_version gate exactly so both kinds of drift route through
// the same `performFullSync` recovery.
//
// NULL stored fingerprint is "never stamped" (pre-v125 brain OR fresh
// source whose first sync hasn't completed yet). Treated as
// pass-through in the up-to-date check — first post-upgrade sync
// stamps the column quietly so subsequent passes have a baseline.
const storedConfigFp = await readConfigFingerprint(engine, opts.sourceId);
const currentSourceConfig = await readSourceConfig(engine, opts.sourceId);
const currentConfigFp = computeSourceConfigFingerprint(currentSourceConfig);
const configMismatch = storedConfigFp !== null && storedConfigFp !== currentConfigFp;
const configNeverStamped = storedConfigFp === null && opts.sourceId !== undefined;
if (lastCommit === headCommit && !versionMismatch && !versionNeverSet && !hasDetachedWorkingTreeChanges && !configMismatch) {
// v0.42.52.0 (PR #22xx): bump last_sync_at as a heartbeat on every successful
// 0-changes sync. D4 invariant ("never advance last_commit on partial") is
// preserved: last_sync_at is a monitoring signal (doctor sync_freshness
@@ -2176,6 +2348,14 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
[opts.sourceId],
);
}
// First post-upgrade sync on a pre-v125 brain lands here with
// configNeverStamped=true; stamp the fingerprint so the gate has a
// baseline for the NEXT pass. A spurious re-walk on the upgrade
// pass would surprise users; quietly establishing the baseline does
// not.
if (configNeverStamped) {
await writeConfigFingerprint(engine, opts.sourceId, currentConfigFp);
}
return {
status: 'up_to_date',
fromCommit: lastCommit,
@@ -2187,13 +2367,21 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
};
}
if ((versionMismatch || versionNeverSet) && lastCommit === headCommit) {
if ((versionMismatch || versionNeverSet || configMismatch) && lastCommit === headCommit) {
const reasons: string[] = [];
if (versionMismatch || versionNeverSet) {
reasons.push(`chunker_version=${storedVersion ?? 'unset'}${currentVersion}`);
}
if (configMismatch) {
reasons.push(`config_fingerprint=${storedConfigFp?.slice(0, 8)}${currentConfigFp.slice(0, 8)}`);
}
slog(
`[sync] chunker_version gate: stored=${storedVersion ?? 'unset'}, current=${currentVersion}. ` +
`Forcing full re-chunk pass (git HEAD unchanged but pipeline version advanced).`,
`[sync] full re-walk forced (${reasons.join(', ')}): ` +
`git HEAD unchanged but a walk-affecting setting advanced.`,
);
const result = await performFullSync(engine, fullSyncRoots, headCommit, opts);
await writeChunkerVersion(engine, opts.sourceId, currentVersion);
await writeConfigFingerprint(engine, opts.sourceId, currentConfigFp);
return result;
}
@@ -2237,8 +2425,16 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
scoped && p.startsWith(syncScopeRelPath + '/') ? p.slice(syncScopeRelPath.length + 1) : p;
const excluded = (p: string): boolean =>
opts.exclude !== undefined && opts.exclude.length > 0 && matchesAnyGlob(scopeRel(p), opts.exclude);
// #2156: include globs are an allow-list, same scope-relative anchoring as
// exclude. Populated from the source row's persisted config.include_globs
// (or CLI --include). Deliberately NOT threaded into syncOpts/isSyncable:
// the unsyncable-cleanup loop below deletes pages for non-metafile
// classifications, and glob filtering must stay conservative (never delete
// previously-imported pages — the documented #1433 posture for --exclude).
const included = (p: string): boolean =>
opts.include === undefined || opts.include.length === 0 || matchesAnyGlob(scopeRel(p), opts.include);
// Filter to syncable files (strategy-aware + scope-aware + exclude-aware)
// Filter to syncable files (strategy-aware + scope-aware + glob-aware)
const syncOpts = opts.strategy ? { strategy: opts.strategy } : undefined;
// #1970 (F-C): a rename whose DESTINATION is unsyncable drops out of BOTH
// `renamed` (only `r.to` is kept below) AND `deleted` (git emits it as `R`,
@@ -2252,13 +2448,13 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
!(inScope(r.to) && isSyncable(r.to, syncOpts)))
.map(r => r.from);
const filtered: SyncManifest = {
added: manifest.added.filter(p => inScope(p) && !excluded(p) && isSyncable(p, syncOpts)),
modified: manifest.modified.filter(p => inScope(p) && !excluded(p) && isSyncable(p, syncOpts)),
added: manifest.added.filter(p => inScope(p) && included(p) && !excluded(p) && isSyncable(p, syncOpts)),
modified: manifest.modified.filter(p => inScope(p) && included(p) && !excluded(p) && isSyncable(p, syncOpts)),
deleted: unique([
...manifest.deleted.filter(p => inScope(p) && isSyncable(p, syncOpts)),
...renamedToUnsyncable,
]),
renamed: manifest.renamed.filter(r => inScope(r.to) && !excluded(r.to) && isSyncable(r.to, syncOpts)),
renamed: manifest.renamed.filter(r => inScope(r.to) && included(r.to) && !excluded(r.to) && isSyncable(r.to, syncOpts)),
};
// NAV-4: warn when --exclude filtered out every candidate change — almost
@@ -2355,6 +2551,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
await writeSyncAnchor(engine, opts.sourceId, 'last_commit', pin, commitTimeMs(gitContextRoot, pin));
await engine.setConfig('sync.last_run', new Date().toISOString());
await writeChunkerVersion(engine, opts.sourceId, String(CHUNKER_VERSION));
await writeConfigFingerprint(engine, opts.sourceId, currentConfigFp);
await clearOpCheckpoint(engine, ckpt.paths);
await clearOpCheckpoint(engine, ckpt.target);
return {
@@ -3179,6 +3376,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
await engine.setConfig('sync.last_run', new Date().toISOString());
await writeSyncAnchor(engine, opts.sourceId, 'repo_path', anchorPath);
await writeChunkerVersion(engine, opts.sourceId, String(CHUNKER_VERSION));
await writeConfigFingerprint(engine, opts.sourceId, currentConfigFp);
await clearOpCheckpoint(engine, ckpt.paths);
await clearOpCheckpoint(engine, ckpt.target);
};
@@ -3430,6 +3628,9 @@ async function performFullSync(
// files were waiting.
if (opts.dryRun) {
let allFiles = collectSyncableFiles(syncScopeRoot, { strategy: opts.strategy ?? 'markdown' });
if (opts.include && opts.include.length > 0) {
allFiles = allFiles.filter(abs => matchesAnyGlob(relative(syncScopeRoot, abs), opts.include));
}
if (opts.exclude && opts.exclude.length > 0) {
allFiles = allFiles.filter(abs => !matchesAnyGlob(relative(syncScopeRoot, abs), opts.exclude));
}
@@ -3475,6 +3676,7 @@ async function performFullSync(
commit: headCommit,
strategy: opts.strategy,
sourceId: opts.sourceId,
include: opts.include,
exclude: opts.exclude,
slugRoot,
// issue #1939: performFullSync owns the failure ledger + bookmark via the
@@ -3504,6 +3706,7 @@ async function performFullSync(
await engine.setConfig('sync.last_run', new Date().toISOString());
await writeSyncAnchor(engine, opts.sourceId, 'repo_path', anchorPath);
await writeChunkerVersion(engine, opts.sourceId, String(CHUNKER_VERSION));
await stampSourceConfigFingerprint(engine, opts.sourceId);
};
const fullGate = await applySyncFailureGate({
@@ -3967,8 +4170,12 @@ Options:
run at the repo root; imports are scoped to the subdir
and slugs stay root-relative (wiki/page1). Passing the
subdirectory directly as --repo also works.
--include <glob> Only sync files matching at least one glob (repeatable;
matched against the scope-relative path). Merged with
the source's persisted config.include_globs.
--exclude <glob> Exclude files matching the glob from sync (repeatable;
matched against the scope-relative path).
matched against the scope-relative path; applied after
--include). Merged with config.exclude_globs.
--dry-run Show what would be synced without writing.
--skip-failed Acknowledge previously-recorded sync failures so
the bookmark can advance past unparseable files.
@@ -4129,14 +4336,17 @@ See also:
}
const strategyArg = args.find((a, i) => args[i - 1] === '--strategy') as SyncOpts['strategy'] | undefined;
// #753/#774: monorepo subdir-source flags. --exclude is repeatable.
// #2156: --include is the allow-list counterpart, same repeatable shape.
const srcSubpath = args.find((a, i) => args[i - 1] === '--src-subpath') || undefined;
const excludePatterns: string[] = [];
const includePatterns: string[] = [];
for (let i = 0; i < args.length; i++) {
if (args[i] === '--exclude' && i + 1 < args.length) excludePatterns.push(args[i + 1]);
if (args[i] === '--include' && i + 1 < args.length) includePatterns.push(args[i + 1]);
}
if (syncAll && (srcSubpath || excludePatterns.length > 0)) {
if (syncAll && (srcSubpath || excludePatterns.length > 0 || includePatterns.length > 0)) {
console.error(
`--src-subpath/--exclude scope a single sync invocation; they cannot be combined with --all. ` +
`--src-subpath/--include/--exclude scope a single sync invocation; they cannot be combined with --all. ` +
`For --all runs, register the subdirectory as the source's local_path instead ` +
`(gbrain sources add <id> --path <repo>/<subdir>).`,
);
@@ -4337,7 +4547,11 @@ See also:
const onAllSigint = () => { try { allInterrupt.abort(new Error('SIGINT')); } catch { /* */ } };
const runOne = async (src: typeof sources[number]): Promise<SyncResult> => {
const cfg = (src.config || {}) as { strategy?: 'markdown' | 'code' | 'auto' };
const cfg = (src.config || {}) as {
strategy?: 'markdown' | 'code' | 'auto';
include_globs?: unknown;
exclude_globs?: unknown;
};
// D18: parallel path defers embed; auto-enqueue embed-backfill after.
// v0.42.42.0 (#2139): `autoDeferEmbeds` (the inline gate tripped in a
// non-TTY session) ALSO forces deferral — global by design (the gate's
@@ -4375,6 +4589,8 @@ See also:
skipFailed, retryFailed, noSchemaPack,
sourceId: src.id,
strategy: cfg.strategy,
include: parseGlobList(cfg.include_globs),
exclude: parseGlobList(cfg.exclude_globs),
concurrency,
signal: composeAbortSignals(allInterrupt.signal, controller?.signal),
};
@@ -4586,11 +4802,27 @@ See also:
// lock released by its own finally) instead of a hard cut.
const singleSourceInterrupt = new AbortController();
const onSingleSourceSigint = () => { try { singleSourceInterrupt.abort(new Error('SIGINT')); } catch { /* */ } };
// Read persisted include/exclude globs from the source row, mirroring the
// --all fan-out's `runOne` closure above. Best-effort: a fetch failure
// falls through to "no glob filters", preserving pre-existing behavior.
// sourceId is always set here (resolveSourceWithTier ran above), so this
// path never silently runs without source-config awareness.
let sourceCfg: { include_globs?: unknown; exclude_globs?: unknown } = {};
try {
const { fetchSource } = await import('../core/sources-load.ts');
const src = await fetchSource(engine, sourceId);
if (src?.config && typeof src.config === 'object') {
sourceCfg = src.config as { include_globs?: unknown; exclude_globs?: unknown };
}
} catch { /* fall through to no filters */ }
const opts: SyncOpts = {
repoPath, dryRun, full, noPull, noEmbed, noExtract, skipFailed, retryFailed, noSchemaPack, sourceId,
strategy: strategyArg, concurrency,
srcSubpath,
exclude: excludePatterns.length > 0 ? excludePatterns : undefined,
// #2156: union of the repeatable CLI flags (one-off, this invocation
// only) and the source row's persisted config globs (every sync).
include: mergeGlobs(includePatterns, parseGlobList(sourceCfg.include_globs)),
exclude: mergeGlobs(excludePatterns, parseGlobList(sourceCfg.exclude_globs)),
signal: composeAbortSignals(singleSourceInterrupt.signal, singleSourceController?.signal),
};
@@ -4818,7 +5050,11 @@ export async function syncOneSource(
noExtract?: boolean;
},
): Promise<{ result: SyncResult; log: string }> {
const cfg = (src.config || {}) as { strategy?: 'markdown' | 'code' | 'auto' };
const cfg = (src.config || {}) as {
strategy?: 'markdown' | 'code' | 'auto';
include_globs?: unknown;
exclude_globs?: unknown;
};
const log = `\n--- Syncing source: ${src.name} ---\n`;
const repoOpts: SyncOpts = {
repoPath: src.local_path!,
@@ -4832,6 +5068,8 @@ export async function syncOneSource(
noSchemaPack: shared.noSchemaPack,
sourceId: src.id,
strategy: cfg.strategy,
include: parseGlobList(cfg.include_globs),
exclude: parseGlobList(cfg.exclude_globs),
concurrency: shared.concurrency,
// lockId defaults to `gbrain-sync:${src.id}` via the invariant in
// performSync (no explicit override needed — sourceId triggers it).
+16 -77
View File
@@ -23,10 +23,6 @@
*/
import type { BrainEngine } from '../engine.ts';
import { anySignal } from '../abort-check.ts';
/** Fresh cleanup budget for the lock release after the window signal fires. */
const LOCK_RELEASE_GRACE_MS = 5_000;
export interface ExtractAtomsDrainDeps {
/**
@@ -34,13 +30,13 @@ export interface ExtractAtomsDrainDeps {
* via `withRefreshingLock`. MUST throw when the lock is held by another
* process (e.g. `LockUnavailableError`) — the drain lets that propagate so
* the caller can report `cycle_already_running` and exit, matching the
* routine cycle's skip contract. The signal bounds lock acquisition too.
* routine cycle's skip contract.
*/
withLock: <T>(work: () => Promise<T>, signal: AbortSignal) => Promise<T>;
/** Process one batch. The signal fires at the drain wallclock deadline. */
runBatch: (signal: AbortSignal) => Promise<{ extracted: number; skipped: number }>;
withLock: <T>(work: () => Promise<T>) => Promise<T>;
/** Process one bounded batch (rediscovers eligibility). Returns counts. */
runBatch: () => Promise<{ extracted: number; skipped: number }>;
/** Count remaining eligible-but-unextracted pages, or null on query error. */
countRemaining: (signal: AbortSignal) => Promise<number | null>;
countRemaining: () => Promise<number | null>;
/** Injectable clock. Production: Date.now. */
now: () => number;
/** Optional progress sink (one line per batch). */
@@ -52,8 +48,6 @@ export interface ExtractAtomsDrainOpts {
windowMs: number;
/** Hard cap on batches (belt-and-suspenders against a 0-progress loop). Default 1000. */
maxBatches?: number;
/** External caller cancellation (worker timeout / shutdown). */
abortSignal?: AbortSignal;
}
export interface ExtractAtomsDrainResult {
@@ -74,79 +68,35 @@ export async function runExtractAtomsDrain(
opts: ExtractAtomsDrainOpts,
): Promise<ExtractAtomsDrainResult> {
const maxBatches = opts.maxBatches ?? 1000;
const deadline = deps.now() + opts.windowMs;
// #2750: the window used to be checked only BETWEEN batches, so one slow
// batch (sequential LLM calls) or a hung lock/count/write overran it without
// bound (observed window=120s → 282.5s). A real-time deadline signal now
// cancels (Postgres) or abandons (PGLite, cooperative) whatever is in
// flight; the injected clock still drives loop-boundary checks so the pure
// loop stays unit-testable.
const signal = anySignal(
AbortSignal.timeout(Math.max(1, opts.windowMs)),
opts.abortSignal,
);
const result: ExtractAtomsDrainResult = await deps.withLock(async () => {
return deps.withLock(async () => {
const deadline = deps.now() + opts.windowMs;
let extracted = 0;
let skipped = 0;
let batches = 0;
let stopped: ExtractAtomsDrainResult['stopped'] = 'window';
while (deps.now() < deadline && !signal.aborted) {
while (deps.now() < deadline) {
if (batches >= maxBatches) { stopped = 'max_batches'; break; }
let before: number | null;
try {
before = await deps.countRemaining(signal);
} catch (err) {
if (signal.aborted) break;
throw err;
}
const before = await deps.countRemaining();
if (before === 0) { stopped = 'drained'; break; }
// The backlog count consumed the same wallclock budget — re-check so a
// slow count can't hand the batch a window that already expired.
if (deps.now() >= deadline || signal.aborted) break;
let r: { extracted: number; skipped: number };
try {
r = await deps.runBatch(signal);
} catch (err) {
if (signal.aborted) break;
throw err;
}
const r = await deps.runBatch();
extracted += r.extracted;
skipped += r.skipped;
batches++;
deps.onBatch?.({ batch: batches, extracted: r.extracted, remaining: before });
// A deadline abort inside the batch can surface as zero progress;
// window exhaustion wins over the generic no_progress label.
if (deps.now() >= deadline || signal.aborted) break;
// Stop if a batch made zero forward progress — extraction is failing or
// everything left is ineligible (e.g. all skipped). Prevents a hot loop
// that spends budget without draining.
if (r.extracted === 0 && r.skipped === 0) { stopped = 'no_progress'; break; }
}
// After the window elapsed, don't spend more unbounded time on a final
// count — report remaining as unknown instead of overrunning further.
const windowElapsed = signal.aborted || deps.now() >= deadline;
let remaining: number | null = null;
if (!windowElapsed) {
try {
remaining = await deps.countRemaining(signal);
} catch (err) {
if (!signal.aborted) throw err;
}
}
const remaining = await deps.countRemaining();
if (remaining === 0) stopped = 'drained';
return { phase: 'extract_atoms', status: 'ok', extracted, skipped, remaining, batches, stopped };
}, signal);
// Internal window expiry is a normal partial result. An EXTERNAL abort
// (worker cancel/timeout/shutdown) must reject so Minion records the abort.
if (opts.abortSignal?.aborted) throw opts.abortSignal.reason;
return result;
});
}
// ─── Shared wiring helper (v0.42.x #1685 DECISION 5A) ──────────────────────
@@ -184,8 +134,6 @@ export interface DrainForSourceOpts {
maxBatches?: number;
/** Optional per-batch progress sink (stderr line in dream; job progress in the handler). */
onBatch?: ExtractAtomsDrainDeps['onBatch'];
/** Worker cancellation / shutdown signal (Minion `job.signal`). */
abortSignal?: AbortSignal;
}
export async function runExtractAtomsDrainForSource(
@@ -201,17 +149,12 @@ export async function runExtractAtomsDrainForSource(
return runExtractAtomsDrain(
{
withLock: (work, signal) => withRefreshingLock(engine, lockId, work, {
ttlMinutes: 5,
signal,
releaseTimeoutMs: LOCK_RELEASE_GRACE_MS,
}),
runBatch: async (signal) => {
withLock: (work) => withRefreshingLock(engine, lockId, work, { ttlMinutes: 5 }),
runBatch: async () => {
const r = await runPhaseExtractAtoms(engine, {
sourceId: extractionSourceId,
dryRun: false,
brainDir: opts.brainDir,
abortSignal: signal,
});
const d = (r.details ?? {}) as Record<string, unknown>;
return {
@@ -219,14 +162,10 @@ export async function runExtractAtomsDrainForSource(
skipped: Number(d.duplicates_skipped ?? 0),
};
},
countRemaining: (signal) => countExtractAtomsBacklog(engine, extractionSourceId, signal),
countRemaining: () => countExtractAtomsBacklog(engine, extractionSourceId),
now: Date.now,
onBatch: opts.onBatch,
},
{
windowMs: opts.windowSeconds * 1000,
maxBatches: opts.maxBatches,
abortSignal: opts.abortSignal,
},
{ windowMs: opts.windowSeconds * 1000, maxBatches: opts.maxBatches },
);
}
+17 -68
View File
@@ -58,10 +58,6 @@ import { createHash } from 'crypto';
import { slugifySegment } from '../sync.ts';
const DEFAULT_BUDGET_USD = 0.3;
// #2750: fresh wallclock budget for the receipt/rollup bookkeeping writes when
// the caller's deadline already fired — committed atoms must not lose their
// cost/receipt trail, but the writes can't be unbounded either.
const BOOKKEEPING_GRACE_MS = 5_000;
// v0.42+ TODO: read atom_type enum from active pack manifest at runtime.
const ATOM_TYPES = [
@@ -159,13 +155,6 @@ export interface ExtractAtomsOpts {
* `heartbeat()` on the passed reporter.
*/
progress?: ProgressReporter;
/**
* #2750: caller deadline/cancellation. Forwarded to every gateway call and
* DB query/write so the drain window bounds real lifetime, plus a
* cooperative between-item check (the PGLite path, where query abort only
* abandons the waiter).
*/
abortSignal?: AbortSignal;
}
interface ExtractedAtom {
@@ -223,7 +212,6 @@ export async function discoverExtractablePages(
engine: BrainEngine,
sourceId: string,
affectedSlugs?: string[],
abortSignal?: AbortSignal,
): Promise<DiscoveredPage[]> {
const hasFilter = Array.isArray(affectedSlugs) && affectedSlugs.length > 0;
const sql = `
@@ -263,16 +251,13 @@ export async function discoverExtractablePages(
slug: string;
compiled_truth: string;
content_hash: string;
}>(sql, params, { signal: abortSignal });
}>(sql, params);
return rows.map((r) => ({
slug: r.slug,
content: r.compiled_truth,
contentHash: r.content_hash,
}));
} catch (err) {
// A deadline abort is not a fail-soft condition — propagate so the
// caller stops instead of proceeding with an empty page list.
if (abortSignal?.aborted) throw err;
const msg = err instanceof Error ? err.message : String(err);
console.error(`[extract_atoms] page-discovery query failed: ${msg}`);
return []; // fail-soft: transcript path still proceeds
@@ -297,7 +282,6 @@ export async function discoverExtractablePages(
export async function countExtractAtomsBacklog(
engine: BrainEngine,
sourceId?: string,
abortSignal?: AbortSignal,
): Promise<number | null> {
try {
// Two modes: scoped (the phase's per-source `remaining`) vs brain-wide
@@ -337,10 +321,9 @@ export async function countExtractAtomsBacklog(
const params = scoped
? [sourceId, extractableTypes, MIN_PAGE_CHARS_FOR_EXTRACTION]
: [extractableTypes, MIN_PAGE_CHARS_FOR_EXTRACTION];
const rows = await engine.executeRaw<{ cnt: string | number }>(sql, params, { signal: abortSignal });
const rows = await engine.executeRaw<{ cnt: string | number }>(sql, params);
return Number(rows[0]?.cnt ?? 0);
} catch (err) {
if (abortSignal?.aborted) throw err;
const msg = err instanceof Error ? err.message : String(err);
console.error(`[extract_atoms] backlog count failed: ${msg}`);
return null;
@@ -367,7 +350,6 @@ export async function atomsExistingForHashes(
engine: BrainEngine,
sourceId: string,
contentHash16s: string[],
abortSignal?: AbortSignal,
): Promise<Set<string>> {
if (contentHash16s.length === 0) return new Set();
try {
@@ -379,11 +361,9 @@ export async function atomsExistingForHashes(
AND deleted_at IS NULL
AND frontmatter->>'source_hash' = ANY($2::text[])`,
[sourceId, contentHash16s],
{ signal: abortSignal },
);
return new Set(rows.map(r => r.h));
} catch (err) {
if (abortSignal?.aborted) throw err;
const msg = err instanceof Error ? err.message : String(err);
console.error(`[extract_atoms] batch idempotency check failed (assuming none extracted): ${msg}`);
return new Set();
@@ -404,7 +384,6 @@ export async function runPhaseExtractAtoms(
): Promise<PhaseResult> {
const sourceId = opts.sourceId ?? 'default';
const chat = opts._chat ?? gatewayChat;
if (opts.abortSignal?.aborted) throw opts.abortSignal.reason;
// 1a. Get transcripts (test seam OR production discovery).
// v0.41.2.1: config loader switched to loadConfigWithEngine() so the
@@ -446,7 +425,7 @@ export async function runPhaseExtractAtoms(
if (opts._pages !== undefined) {
pages = opts._pages;
} else {
pages = await discoverExtractablePages(engine, sourceId, opts.affectedSlugs, opts.abortSignal);
pages = await discoverExtractablePages(engine, sourceId, opts.affectedSlugs);
}
// 2. Apply transcript-side source-hash idempotency in ONE batch query
@@ -458,7 +437,7 @@ export async function runPhaseExtractAtoms(
// Surface a heartbeat before the batch query so even an instant
// short-circuit shows a sign of life (closes Issue 2 silent-phase pain).
opts.progress?.heartbeat(`checking existing atoms for ${allHashes16.length} transcripts`);
const existingHashes = await atomsExistingForHashes(engine, sourceId, allHashes16, opts.abortSignal);
const existingHashes = await atomsExistingForHashes(engine, sourceId, allHashes16);
for (const t of transcripts) {
if (existingHashes.has(t.contentHash.slice(0, 16))) {
duplicatesSkipped++;
@@ -522,7 +501,6 @@ export async function runPhaseExtractAtoms(
const failures: Array<{ source: string; error: string }> = [];
let estimatedSpendUsd = 0;
const budgetCap = DEFAULT_BUDGET_USD;
let deadlineAborted = false;
// v0.41.19.0 (T3): throttled yield helper. Fires `opts.yieldDuringPhase`
// every 30s. Cycle.ts threads `buildYieldDuringPhase(lock, outer)` so
@@ -548,12 +526,6 @@ export async function runPhaseExtractAtoms(
}
for (const item of work) {
// #2750: cooperative between-item abort. Works on every engine — this is
// the primary bound on PGLite, where query abort only abandons the waiter.
if (opts.abortSignal?.aborted) {
deadlineAborted = true;
break;
}
await maybeYield();
if (estimatedSpendUsd >= budgetCap) {
if (item.kind === 'transcript') transcriptsSkipped++;
@@ -572,22 +544,16 @@ export async function runPhaseExtractAtoms(
},
],
maxTokens: 2000,
abortSignal: opts.abortSignal,
});
// Rough cost estimate — Haiku at ~$0.80/M input + $4/M output.
// A completed gateway call is billable even if the deadline fires
// immediately afterward, so record usage BEFORE the abort check.
estimatedSpendUsd +=
(result.usage.input_tokens * 0.8 + result.usage.output_tokens * 4.0) / 1_000_000;
if (opts.abortSignal?.aborted) {
deadlineAborted = true;
break;
}
// Post-await yield: closes the "long LLM call past TTL" hazard
// codex flagged. The 30s throttle inside maybeYield bounds the
// actual refresh rate so this is cheap when calls are fast.
await maybeYield();
// Rough cost estimate — Haiku at ~$0.80/M input + $4/M output
estimatedSpendUsd +=
(result.usage.input_tokens * 0.8 + result.usage.output_tokens * 4.0) / 1_000_000;
const atoms = parseAtomsResponse(result.text);
if (atoms.length === 0) {
if (item.kind === 'transcript') transcriptsProcessed++;
@@ -626,7 +592,7 @@ export async function runPhaseExtractAtoms(
},
timeline: '',
},
{ sourceId, signal: opts.abortSignal },
{ sourceId },
);
totalAtomsExtracted++;
}
@@ -639,11 +605,6 @@ export async function runPhaseExtractAtoms(
// Reporter rate-limits to ~1 line/sec; safe to tick every iter.
opts.progress?.tick(1, `${totalAtomsExtracted} atoms / ${duplicatesSkipped} skipped`);
} catch (err) {
// A deadline abort is a partial result, not a per-item failure.
if (opts.abortSignal?.aborted) {
deadlineAborted = true;
break;
}
failures.push({
source: originLabel,
error: err instanceof Error ? err.message : String(err),
@@ -654,12 +615,6 @@ export async function runPhaseExtractAtoms(
// v0.42 Wave B2: write extract receipt + rollup row when the phase
// actually extracted atoms. Both are best-effort per F-OUT-19 —
// audit-trail / search-visibility surfaces don't block the phase result.
//
// #2750: bookkeeping runs on a FRESH short grace signal, never the caller's
// work deadline — the deadline may have already fired (partial run) and
// committed atoms must not lose their receipt/cost trail; but the writes
// stay bounded so the overrun is capped at the grace window.
const bookkeepingSignal = opts.dryRun ? undefined : AbortSignal.timeout(BOOKKEEPING_GRACE_MS);
if (!opts.dryRun && totalAtomsExtracted > 0) {
const runId = `atoms-${Date.now().toString(36)}-${sourceId.slice(0, 4)}`;
try {
@@ -674,24 +629,19 @@ export async function runPhaseExtractAtoms(
summary:
`Extracted ${totalAtomsExtracted} atoms from ` +
`${transcriptsProcessed} transcripts + ${pagesProcessed} pages.`,
}, { signal: bookkeepingSignal });
});
} catch (err) {
console.error(`[extract_atoms] receipt write failed: ${(err as Error).message}`);
}
}
if (!opts.dryRun) {
try {
await upsertExtractRollup(engine, {
kind: 'atoms',
source_id: sourceId,
cost_delta: estimatedSpendUsd,
// A deadline-truncated run is not a completed round.
round_completed_delta: failures.length === 0 && !deadlineAborted ? 1 : 0,
halt_delta: failures.length > 0 ? 1 : 0,
}, { signal: bookkeepingSignal });
} catch (err) {
console.error(`[extract_atoms] rollup write failed: ${(err as Error).message}`);
}
await upsertExtractRollup(engine, {
kind: 'atoms',
source_id: sourceId,
cost_delta: estimatedSpendUsd,
round_completed_delta: failures.length === 0 ? 1 : 0,
halt_delta: failures.length > 0 ? 1 : 0,
});
}
return {
@@ -720,7 +670,6 @@ export async function runPhaseExtractAtoms(
budget_usd: budgetCap,
source_id: sourceId,
dry_run: opts.dryRun ?? false,
deadline_aborted: deadlineAborted,
},
};
}
+22 -50
View File
@@ -26,8 +26,7 @@ import type { BrainEngine } from './engine.ts';
export interface DbLockHandle {
id: string;
/** Optional signal bounds the release DELETE (deadline-bound callers). */
release: (signal?: AbortSignal) => Promise<void>;
release: () => Promise<void>;
refresh: () => Promise<void>;
}
@@ -174,7 +173,6 @@ export async function tryAcquireDbLock(
engine: BrainEngine,
lockId: string,
ttlMinutes: number = DEFAULT_TTL_MINUTES,
opts: { signal?: AbortSignal } = {},
): Promise<DbLockHandle | null> {
const pid = process.pid;
const host = hostname();
@@ -207,26 +205,20 @@ export async function tryAcquireDbLock(
// `gbrain sync --break-lock --max-age <s>` uses last_refreshed_at (not
// acquired_at) to identify wedged-but-alive holders without stealing
// healthy long-running holders that are actively refreshing.
// #2750: routed through executeRaw so a deadline-bound caller's signal
// can cancel a hung acquire (pool exhaustion). Cancellation is
// transactional; in the rare ambiguous-commit case the row's TTL is the
// backstop (drain locks use a short 5-minute TTL).
const rows = await engine.executeRaw<{ id: string }>(
`INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at)
VALUES ($1, $2, $3, NOW(), NOW() + $4::interval, NOW())
ON CONFLICT (id) DO UPDATE
SET holder_pid = $2,
holder_host = $3,
acquired_at = NOW(),
ttl_expires_at = NOW() + $4::interval,
last_refreshed_at = NOW()
WHERE gbrain_cycle_locks.ttl_expires_at < NOW()
AND (gbrain_cycle_locks.last_refreshed_at IS NULL
OR gbrain_cycle_locks.last_refreshed_at < NOW() - $5 * INTERVAL '1 second')
RETURNING id`,
[lockId, pid, host, ttl, stealGraceSeconds],
{ signal: opts.signal },
);
const rows: Array<{ id: string }> = await sql`
INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at)
VALUES (${lockId}, ${pid}, ${host}, NOW(), NOW() + ${ttl}::interval, NOW())
ON CONFLICT (id) DO UPDATE
SET holder_pid = ${pid},
holder_host = ${host},
acquired_at = NOW(),
ttl_expires_at = NOW() + ${ttl}::interval,
last_refreshed_at = NOW()
WHERE gbrain_cycle_locks.ttl_expires_at < NOW()
AND (gbrain_cycle_locks.last_refreshed_at IS NULL
OR gbrain_cycle_locks.last_refreshed_at < NOW() - ${stealGraceSeconds} * INTERVAL '1 second')
RETURNING id
`;
if (rows.length === 0) return null;
const deregister = registerCleanup(`db-lock:${lockId}`, async () => {
await sql`
@@ -249,17 +241,12 @@ export async function tryAcquireDbLock(
[ttl, lockId, pid],
);
},
release: async (signal?: AbortSignal) => {
release: async () => {
deregister();
// Direct session pool (same rationale as refresh, #1794) + optional
// signal so a deadline-bound caller's release can't hang forever on
// an exhausted pooler. TTL is the backstop if the DELETE is cancelled.
await engine.executeRawDirect(
`DELETE FROM gbrain_cycle_locks
WHERE id = $1 AND holder_pid = $2`,
[lockId, pid],
{ signal },
);
await sql`
DELETE FROM gbrain_cycle_locks
WHERE id = ${lockId} AND holder_pid = ${pid}
`;
},
};
}
@@ -316,11 +303,6 @@ export async function tryAcquireDbLock(
const first = await acquireOnce();
if (first) return first;
// #2750: deadline-bound callers prefer an honest busy result over the
// best-effort same-host takeover below, whose inspect/delete/retry calls
// are not signal-bounded. The initial upsert already reclaims expired locks.
if (opts.signal) return null;
// v0.42 (#1780 Gap 3): the lock is held and its TTL hasn't expired (the
// upsert's ON CONFLICT ... WHERE ttl_expires_at < NOW() returned no row).
// If the holder is on THIS host, provably dead, and past the grace window,
@@ -814,10 +796,6 @@ export interface WithRefreshingLockOpts {
ttlMinutes?: number;
/** Heartbeat-fail threshold in ms — abort if SELECT 1 takes longer. Default 30000. */
heartbeatTimeoutMs?: number;
/** #2750: bound lock acquisition with the caller's deadline signal. */
signal?: AbortSignal;
/** Fresh cleanup budget for the release DELETE when `signal` is set. Default 5000. */
releaseTimeoutMs?: number;
}
/**
@@ -837,7 +815,7 @@ export async function withRefreshingLock<T>(
// Refresh 6x per TTL window so a missed tick doesn't expire the lock.
const refreshIntervalMs = Math.max(15000, (ttlMinutes * 60 * 1000) / 6);
const handle = await tryAcquireDbLock(engine, lockId, ttlMinutes, { signal: opts.signal });
const handle = await tryAcquireDbLock(engine, lockId, ttlMinutes);
if (!handle) throw new LockUnavailableError(lockId);
let healthOk = true;
@@ -876,13 +854,7 @@ export async function withRefreshingLock<T>(
return await work();
} finally {
clearInterval(interval);
// #2750: when the caller is deadline-bound, its work signal may already
// have fired — release on a FRESH short grace signal so cleanup neither
// inherits the spent deadline nor hangs unbounded. TTL is the backstop.
const releaseSignal = opts.signal
? AbortSignal.timeout(opts.releaseTimeoutMs ?? 5_000)
: undefined;
try { await handle.release(releaseSignal); } catch { /* idempotent; TTL backstop */ }
try { await handle.release(); } catch { /* idempotent */ }
if (!healthOk) {
// Surface that the heartbeat detected backend trouble — caller can
// log to the connection-events audit if desired.
+1 -9
View File
@@ -696,16 +696,8 @@ export interface BrainEngine {
* is included in the INSERT column list so ON CONFLICT (source_id, slug)
* DO UPDATE actually targets the intended row instead of fabricating a
* duplicate at (default, slug). Multi-source brains MUST pass sourceId.
*
* `opts.signal` (#2750): optional cancellation for deadline-bound writers.
* Postgres cancels the in-flight statement; PGLite pre-checks only (query
* cancellation is not possible in-process — cooperative abort between calls).
*/
putPage(
slug: string,
page: PageInput,
opts?: { sourceId?: string; signal?: AbortSignal },
): Promise<Page>;
putPage(slug: string, page: PageInput, opts?: { sourceId?: string }): Promise<Page>;
/**
* v0.41.13 (#1309) — identity-based dedup pre-check for the import pipeline.
*
+1 -2
View File
@@ -187,7 +187,6 @@ function buildReceiptFrontmatter(input: ExtractReceiptInput): Record<string, unk
export async function writeReceipt(
engine: BrainEngine,
input: ExtractReceiptInput,
opts?: { signal?: AbortSignal },
): Promise<{ slug: string; page: Page }> {
const slug = receiptSlug(input);
const title = `${input.kind}${input.round}${input.source_id}`;
@@ -202,7 +201,7 @@ export async function writeReceipt(
compiled_truth,
frontmatter,
},
{ sourceId: input.source_id, signal: opts?.signal },
{ sourceId: input.source_id },
);
return { slug, page };
-4
View File
@@ -70,7 +70,6 @@ function today(): string {
export async function upsertExtractRollup(
engine: BrainEngine,
input: RollupUpsertInput,
opts?: { signal?: AbortSignal },
): Promise<{ ok: boolean; error?: string }> {
const day = input.day ?? today();
const cost = input.cost_delta ?? 0;
@@ -97,12 +96,9 @@ export async function upsertExtractRollup(
rollup_write_failures = extract_rollup_7d.rollup_write_failures + EXCLUDED.rollup_write_failures,
updated_at = now()`,
[input.kind, input.source_id, day, cost, halts, evalFails, evalPasses, completed, failures],
{ signal: opts?.signal },
);
return { ok: true };
} catch (err) {
// Signal-bounded callers get the abort surfaced, not a swallowed `ok:false`.
if (opts?.signal?.aborted) throw err;
const msg = (err as Error).message || String(err);
// Don't spam: log once per process per (kind, day) error class.
rollupErrorLogOnce(input.kind, day, msg);
+26
View File
@@ -5671,6 +5671,32 @@ export const MIGRATIONS: Migration[] = [
`);
},
},
{
version: 125,
name: 'sources_config_fingerprint',
// #2157 follow-on: the "Already up to date" gate at sync.ts honors
// git-HEAD equality + chunker-version match but ignored source-config
// drift. A user who runs `gbrain sources add default --exclude
// 'Templates/**'` AFTER an initial sync got "Already up to date" on
// the next pass because git HEAD was unchanged — the new exclusion
// never reached the walk until `gbrain sync --full`.
//
// This column caches a SHA-256 fingerprint of the walk-affecting
// fields in `sources.config` (strategy + include_globs +
// exclude_globs); mismatches trigger a full re-walk via the same code
// path as a chunker_version bump.
//
// NULL on pre-migration rows is treated as "not yet stamped" by
// readConfigFingerprint, so the FIRST sync after upgrade is normal
// (no spurious force-full just because the column was added).
//
// Keep in sync with src/schema.sql and src/core/schema-embedded.ts.
idempotent: true,
sql: `
ALTER TABLE sources
ADD COLUMN IF NOT EXISTS config_fingerprint TEXT;
`,
},
];
export const LATEST_VERSION = MIGRATIONS.length > 0
+1 -9
View File
@@ -1003,15 +1003,7 @@ export class PGLiteEngine implements BrainEngine {
return { slug: r.slug, id: Number(r.id) };
}
async putPage(
slug: string,
page: PageInput,
opts?: { sourceId?: string; signal?: AbortSignal },
): Promise<Page> {
// #2750: PGLite is in-process WASM — no query cancellation. Pre-check so
// an already-fired deadline skips the write; abort is cooperative
// between calls (same posture as executeRaw's documented gap).
if (opts?.signal?.aborted) throw new DOMException('aborted', 'AbortError');
async putPage(slug: string, page: PageInput, opts?: { sourceId?: string }): Promise<Page> {
slug = validateSlug(slug);
const hash = page.content_hash || contentHash(page);
const frontmatter = page.frontmatter || {};
+3 -55
View File
@@ -72,32 +72,6 @@ function escapeSqlStringLiteral(value: string): string {
return value.replace(/'/g, "''");
}
/**
* #2750: race a promise against an AbortSignal, detaching the listener once
* settled (long-lived drain signals are reused across many calls, so a bare
* Promise.race would leak one listener per call). The abandoned promise keeps
* running; used only for pool-acquisition waits where that is harmless.
*/
function waitForSignal<T>(work: Promise<T>, signal?: AbortSignal): Promise<T> {
if (!signal) return work;
if (signal.aborted) return Promise.reject(new DOMException('aborted', 'AbortError'));
return new Promise<T>((resolve, reject) => {
let settled = false;
const finish = (fn: () => void) => {
if (settled) return;
settled = true;
signal.removeEventListener('abort', onAbort);
fn();
};
const onAbort = () => finish(() => reject(new DOMException('aborted', 'AbortError')));
signal.addEventListener('abort', onAbort, { once: true });
work.then(
(value) => finish(() => resolve(value)),
(err) => finish(() => reject(err)),
);
});
}
export function getPostgresSchema(
dims: number = DEFAULT_EMBEDDING_DIMENSIONS,
model: string = DEFAULT_EMBEDDING_MODEL,
@@ -1087,12 +1061,7 @@ export class PostgresEngine implements BrainEngine {
});
}
async putPage(
slug: string,
page: PageInput,
opts?: { sourceId?: string; signal?: AbortSignal },
): Promise<Page> {
if (opts?.signal?.aborted) throw new DOMException('aborted', 'AbortError');
async putPage(slug: string, page: PageInput, opts?: { sourceId?: string }): Promise<Page> {
slug = validateSlug(slug);
const sql = this.sql;
const hash = page.content_hash || contentHash(page);
@@ -1127,7 +1096,7 @@ export class PostgresEngine implements BrainEngine {
const sourceUri = page.source_uri ?? null;
const ingestedVia = page.ingested_via ?? null;
const ingestedAt = (sourceKind || sourceUri || ingestedVia) ? new Date() : null;
const pending = sql`
const rows = await sql`
INSERT INTO pages (source_id, slug, type, page_kind, title, compiled_truth, timeline, frontmatter, content_hash, updated_at, effective_date, effective_date_source, import_filename, chunker_version, source_path, source_kind, source_uri, ingested_via, ingested_at)
VALUES (${sourceId}, ${slug}, ${page.type}, ${pageKind}, ${page.title}, ${page.compiled_truth}, ${page.timeline || ''}, ${sql.json(frontmatter as Parameters<typeof sql.json>[0])}, ${hash}, now(), ${effectiveDate}, ${effectiveDateSource}, ${importFilename}, COALESCE(${chunkerVersion}::smallint, ${MARKDOWN_CHUNKER_VERSION}), ${sourcePath}, ${sourceKind}, ${sourceUri}, ${ingestedVia}, ${ingestedAt})
ON CONFLICT (source_id, slug) DO UPDATE SET
@@ -1150,22 +1119,6 @@ export class PostgresEngine implements BrainEngine {
ingested_at = COALESCE(EXCLUDED.ingested_at, pages.ingested_at)
RETURNING id, source_id, slug, type, title, compiled_truth, timeline, frontmatter, content_hash, created_at, updated_at, effective_date, effective_date_source, import_filename, source_kind, source_uri, ingested_via, ingested_at
`;
// #2750: cancel the in-flight statement when the caller's deadline fires,
// same .cancel() wiring as runUnsafe (postgres.js pending queries).
if (opts?.signal) {
const signal = opts.signal;
const onAbort = () => {
try { (pending as unknown as { cancel?: () => void }).cancel?.(); } catch { /* best-effort */ }
};
signal.addEventListener('abort', onAbort, { once: true });
try {
const rows = await pending;
return rowToPage(rows[0]);
} finally {
signal.removeEventListener('abort', onAbort);
}
}
const rows = await pending;
return rowToPage(rows[0]);
}
@@ -5854,16 +5807,11 @@ export class PostgresEngine implements BrainEngine {
params?: unknown[],
opts?: { signal?: AbortSignal },
): Promise<T[]> {
// #2750: an already-fired signal short-circuits BEFORE any pool routing,
// and the direct-pool acquisition itself is signal-bounded — under pooler
// exhaustion `ddl()` can stall indefinitely, which used to make even a
// "bounded" lock release hang past its caller's deadline.
if (opts?.signal?.aborted) throw new DOMException('aborted', 'AbortError');
// Inside an open transaction, _sql is the reserved tx connection (set via
// defineProperty in transaction()); never reroute off it.
const inTransaction = this._sql !== null && this.connectionManager?.peekReadPool() !== this._sql;
const conn = (!inTransaction && this.connectionManager?.isDualPoolActive())
? await waitForSignal(this.connectionManager.ddl(), opts?.signal)
? await this.connectionManager.ddl()
: this.sql;
return this.runUnsafe<T>(conn, sql, params, opts);
}
+7
View File
@@ -39,6 +39,13 @@ CREATE TABLE IF NOT EXISTS sources (
-- bypassing the git-HEAD up_to_date early-return so CHUNKER_VERSION bumps
-- actually trigger re-chunking on upgrade.
chunker_version TEXT,
-- #2157 follow-on: SHA-256 fingerprint of the walk-affecting fields in
-- \`config\` (strategy + include_globs + exclude_globs). Mismatch forces a
-- full re-walk via the same code path as chunker_version, so a user who
-- changes \`sources.config.exclude_globs\` mid-life doesn't get "Already up
-- to date" on the next sync. NULL on pre-migration rows is treated as
-- "not yet stamped" and skips the gate (preserves first-run semantics).
config_fingerprint TEXT,
-- v0.26.5: soft-delete + recovery window. \`archive\` flips archived=true and
-- sets archive_expires_at = now() + 72h. The autopilot purge phase
-- hard-deletes rows where archive_expires_at <= now(). Promoted from a
+25
View File
@@ -155,6 +155,19 @@ export interface AddSourceOpts {
* runs). Does NOT auto-`git init` anything — see `addSource` docstring.
*/
force?: boolean;
/**
* Glob filters persisted into `sources.config.include_globs` /
* `sources.config.exclude_globs`. Read at sync time by
* `commands/sync.ts:syncOneSource` and the single-source path, threaded
* into `isSyncable` / `unsyncableReason` (their `SyncableOptions` shape
* has carried this contract since v0.41.13).
*
* Empty / unspecified arrays are not persisted at all (no `[]` written
* to the JSONB), which keeps the row identical to today for sources
* that don't use filtering.
*/
includeGlobs?: string[];
excludeGlobs?: string[];
}
export interface RemoveSourceOpts {
@@ -429,6 +442,12 @@ export async function addSource(
if (opts.federated !== null && opts.federated !== undefined) {
config.federated = opts.federated;
}
if (opts.includeGlobs && opts.includeGlobs.length > 0) {
config.include_globs = opts.includeGlobs;
}
if (opts.excludeGlobs && opts.excludeGlobs.length > 0) {
config.exclude_globs = opts.excludeGlobs;
}
const displayName = opts.name ?? opts.id;
try {
@@ -508,6 +527,12 @@ export async function addSource(
if (opts.federated !== null && opts.federated !== undefined) {
config.federated = opts.federated;
}
if (opts.includeGlobs && opts.includeGlobs.length > 0) {
config.include_globs = opts.includeGlobs;
}
if (opts.excludeGlobs && opts.excludeGlobs.length > 0) {
config.exclude_globs = opts.excludeGlobs;
}
const displayName = opts.name ?? opts.id;
await engine.executeRaw(
`INSERT INTO sources (id, name, local_path, config)
+7
View File
@@ -219,6 +219,13 @@ function globToRegex(pattern: string): RegExp {
return new RegExp(regex);
}
/**
* Test a normalized POSIX-style path against an array of glob patterns. Returns
* true if any pattern matches. Empty / undefined `patterns` returns false (no
* filter engaged). Exported so non-sync surfaces (lint walker, future ingest
* variants) can apply the same glob semantics as `isSyncable` without
* re-declaring `globToRegex`.
*/
export function matchesAnyGlob(path: string, patterns?: string[]): boolean {
if (!patterns || patterns.length === 0) return false;
const normalized = path.replace(/\\/g, '/');
+7
View File
@@ -35,6 +35,13 @@ CREATE TABLE IF NOT EXISTS sources (
-- bypassing the git-HEAD up_to_date early-return so CHUNKER_VERSION bumps
-- actually trigger re-chunking on upgrade.
chunker_version TEXT,
-- #2157 follow-on: SHA-256 fingerprint of the walk-affecting fields in
-- `config` (strategy + include_globs + exclude_globs). Mismatch forces a
-- full re-walk via the same code path as chunker_version, so a user who
-- changes `sources.config.exclude_globs` mid-life doesn't get "Already up
-- to date" on the next sync. NULL on pre-migration rows is treated as
-- "not yet stamped" and skips the gate (preserves first-run semantics).
config_fingerprint TEXT,
-- v0.26.5: soft-delete + recovery window. `archive` flips archived=true and
-- sets archive_expires_at = now() + 72h. The autopilot purge phase
-- hard-deletes rows where archive_expires_at <= now(). Promoted from a
@@ -17,7 +17,6 @@ import { runPhaseExtractAtoms, parseAtomsResponse } from '../../src/core/cycle/e
import { runPhaseSynthesizeConcepts } from '../../src/core/cycle/synthesize-concepts.ts';
import { resetPgliteState } from '../helpers/reset-pglite.ts';
import type { ChatResult, ChatOpts } from '../../src/core/ai/gateway.ts';
import type { BrainEngine } from '../../src/core/engine.ts';
let engine: PGLiteEngine;
@@ -178,180 +177,6 @@ describe('v0.41 T5: runPhaseExtractAtoms via stubbed chat', () => {
expect((result.details?.failures as unknown[]).length).toBe(1);
});
// ── #2750: caller deadline bounds the phase ────────────────────────────
test('caller deadline aborts a hung chat before processing the next item', async () => {
let calls = 0;
const chat = async (opts: ChatOpts) => {
calls++;
return await new Promise<never>((_resolve, reject) => {
const signal = opts.abortSignal;
if (!signal) return reject(new Error('missing abort signal'));
if (signal.aborted) return reject(signal.reason);
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
});
};
const started = Date.now();
const result = await runPhaseExtractAtoms(engine, {
_transcripts: [
{ filePath: '/hung.txt', content: 'a', contentHash: 'hung-a' },
{ filePath: '/never.txt', content: 'b', contentHash: 'hung-b' },
],
_pages: [],
_chat: chat as typeof import('../../src/core/ai/gateway.ts').chat,
abortSignal: AbortSignal.timeout(25),
});
expect(Date.now() - started).toBeLessThan(2_000);
expect(calls).toBe(1);
expect(result.status).toBe('ok');
expect(result.details?.deadline_aborted).toBe(true);
expect(result.details?.atoms_extracted).toBe(0);
expect(result.details?.failures).toEqual([]);
});
test('billable chat usage is counted when the deadline fires as the response resolves', async () => {
const controller = new AbortController();
const chat = async (opts: ChatOpts): Promise<ChatResult> => {
controller.abort(new DOMException('deadline', 'TimeoutError'));
return stubChat(`[{"title":"late","atom_type":"insight","body":"b"}]`, {
input_tokens: 1_000,
output_tokens: 500,
})(opts);
};
const result = await runPhaseExtractAtoms(engine, {
_transcripts: [{ filePath: '/late.txt', content: 'a', contentHash: 'late' }],
_pages: [],
_chat: chat,
abortSignal: controller.signal,
});
expect(result.details?.deadline_aborted).toBe(true);
expect(Number(result.details?.estimated_spend_usd)).toBeGreaterThan(0);
expect(result.details?.atoms_extracted).toBe(0);
});
test('deadline after partial progress still writes receipt and incomplete rollup', async () => {
const controller = new AbortController();
let calls = 0;
let notifySecondChat!: () => void;
const secondChatStarted = new Promise<void>((resolve) => { notifySecondChat = resolve; });
const chat = async (opts: ChatOpts): Promise<ChatResult> => {
calls++;
if (calls === 1) {
return stubChat(`[{"title":"committed","atom_type":"insight","body":"b"}]`)(opts);
}
notifySecondChat();
return await new Promise<never>((_resolve, reject) => {
const signal = opts.abortSignal;
if (!signal) return reject(new Error('missing abort signal'));
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
});
};
const pending = runPhaseExtractAtoms(engine, {
_transcripts: [
{ filePath: '/committed.txt', content: 'a', contentHash: 'committed-a' },
{ filePath: '/hung.txt', content: 'b', contentHash: 'hung-b' },
],
_pages: [],
_chat: chat,
abortSignal: controller.signal,
});
await secondChatStarted;
controller.abort(new DOMException('deadline', 'TimeoutError'));
const result = await pending;
const atoms = await engine.executeRaw<{ n: number }>(
`SELECT COUNT(*)::int AS n FROM pages WHERE type = 'atom'`,
);
const receipts = await engine.executeRaw<{ n: number }>(
`SELECT COUNT(*)::int AS n FROM pages WHERE type = 'extract_receipt'`,
);
const rollups = await engine.executeRaw<{
cost_usd: string | number;
round_completed_count: string | number;
}>(
`SELECT cost_usd, round_completed_count
FROM extract_rollup_7d
WHERE kind = 'atoms' AND source_id = 'default'`,
);
expect(result.details?.deadline_aborted).toBe(true);
expect(atoms[0].n).toBe(1);
expect(receipts[0].n).toBe(1);
expect(Number(rollups[0].cost_usd)).toBeGreaterThan(0);
expect(Number(rollups[0].round_completed_count)).toBe(0);
});
test('bookkeeping runs on a fresh grace signal, not the fired work deadline', async () => {
const controller = new AbortController();
let putCalls = 0;
let receiptSignal: AbortSignal | undefined;
let rollupSignal: AbortSignal | undefined;
const signalAwareEngine = {
executeRaw: async (sql: string, _params?: unknown[], opts?: { signal?: AbortSignal }) => {
if (sql.includes('INSERT INTO extract_rollup_7d')) rollupSignal = opts?.signal;
return [];
},
putPage: async (_slug: string, _page: unknown, opts?: { signal?: AbortSignal }) => {
putCalls++;
if (putCalls === 1) {
// Atom write in flight; the work deadline fires before bookkeeping.
controller.abort(new DOMException('work deadline', 'TimeoutError'));
} else {
receiptSignal = opts?.signal;
}
return {};
},
} as unknown as BrainEngine;
const result = await runPhaseExtractAtoms(signalAwareEngine, {
_transcripts: [{ filePath: '/one.txt', content: 'a', contentHash: 'one' }],
_pages: [],
_chat: stubChat(`[{"title":"one","atom_type":"insight","body":"b"}]`),
abortSignal: controller.signal,
});
expect(result.details?.atoms_extracted).toBe(1);
expect(putCalls).toBe(2); // atom write + receipt write
expect(receiptSignal).toBeDefined();
expect(receiptSignal).not.toBe(controller.signal);
expect(receiptSignal?.aborted).toBe(false);
expect(rollupSignal).toBe(receiptSignal);
});
test('caller deadline cancels a hung atom write and stops the phase', async () => {
const controller = new AbortController();
let notifyWriteStarted!: () => void;
const writeStarted = new Promise<void>((resolve) => { notifyWriteStarted = resolve; });
let writeCalls = 0;
const signalAwareEngine = {
executeRaw: async () => [],
putPage: async (_slug: string, _page: unknown, opts?: { signal?: AbortSignal }) => {
writeCalls++;
notifyWriteStarted();
return await new Promise<never>((_resolve, reject) => {
const signal = opts?.signal;
if (!signal) return reject(new Error('missing abort signal'));
if (signal.aborted) return reject(signal.reason);
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
});
},
} as unknown as BrainEngine;
const pending = runPhaseExtractAtoms(signalAwareEngine, {
_transcripts: [{ filePath: '/hung-write.txt', content: 'a', contentHash: 'hung-write' }],
_pages: [],
_chat: stubChat(`[{"title":"hung write","atom_type":"insight","body":"b"}]`),
abortSignal: controller.signal,
});
await writeStarted;
controller.abort(new DOMException('deadline', 'TimeoutError'));
const result = await pending;
expect(writeCalls).toBe(1);
expect(result.details?.deadline_aborted).toBe(true);
expect(result.details?.atoms_extracted).toBe(0);
});
// v0.41.2.1 regression case (D9 #14 wording): with _pages:[] and same
// _transcripts, all PRE-EXISTING PhaseResult.details fields match
// pre-fix values byte-for-byte. The new fields (pages_processed,
+9 -131
View File
@@ -43,135 +43,26 @@ describe('runExtractAtomsDrain (issue #1678)', () => {
expect(batches).toBe(3);
});
it('stops at the wallclock window; remaining is unknown (no post-window count)', async () => {
// Each batch consumes 60ms of the 100ms window: two batches fit, the
// third boundary check sees 120 ≥ 100 and stops. #2750: after the window
// elapses the final countRemaining is SKIPPED (it would overrun the
// window), so remaining reports null.
let now = 0;
it('stops at the wallclock window with remaining > 0', async () => {
// SYNC stepping clock: now() #1 sets deadline (0+100=100); the while-check
// then sees 50, 50 (two batches), then 999999 → past deadline → stop.
const times = [0, 50, 50, 999_999];
let ti = 0;
const now = () => times[Math.min(ti++, times.length - 1)];
const result = await runExtractAtomsDrain(
{
withLock: passThroughLock,
countRemaining: async () => 5, // never drains
runBatch: async () => {
now += 60;
return { extracted: 1, skipped: 0 };
},
now: () => now,
runBatch: async () => ({ extracted: 1, skipped: 0 }),
now,
},
{ windowMs: 100 },
);
expect(result.stopped).toBe('window');
expect(result.remaining).toBeNull();
expect(result.remaining).toBe(5);
expect(result.batches).toBe(2);
});
it('passes one drain-level deadline signal into count and batch', async () => {
const seen: AbortSignal[] = [];
const controller = new AbortController();
let now = 0;
const result = await runExtractAtomsDrain(
{
withLock: passThroughLock,
countRemaining: async (signal) => {
seen.push(signal);
return 5;
},
runBatch: async (signal) => {
seen.push(signal);
now = 100;
return { extracted: 1, skipped: 0 };
},
now: () => now,
},
{ windowMs: 100, abortSignal: controller.signal },
);
expect(result.stopped).toBe('window');
expect(result.batches).toBe(1);
expect(seen.length).toBe(2);
expect(seen[0]).toBe(seen[1]);
// Combined (timeout + external) signal, not the raw external one.
expect(seen[0]).not.toBe(controller.signal);
});
it('aborts a hung backlog count at the window deadline and releases the lock', async () => {
let released = false;
const result = await runExtractAtomsDrain(
{
withLock: async (work) => {
try { return await work(); }
finally { released = true; }
},
// Hangs until the drain's real-time deadline signal fires (10ms).
countRemaining: (signal) => new Promise((_resolve, reject) => {
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
}),
runBatch: async () => ({ extracted: 0, skipped: 0 }),
now: () => 0, // injected clock never advances — the SIGNAL must save us
},
{ windowMs: 10 },
);
expect(result.stopped).toBe('window');
expect(result.remaining).toBeNull();
expect(released).toBe(true);
});
it('rethrows external cancellation after releasing the lock', async () => {
const controller = new AbortController();
let released = false;
const pending = runExtractAtomsDrain(
{
withLock: async (work) => {
try { return await work(); }
finally { released = true; }
},
countRemaining: (signal) => new Promise((_resolve, reject) => {
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
}),
runBatch: async () => ({ extracted: 0, skipped: 0 }),
now: () => 0,
},
{ windowMs: 1_000_000, abortSignal: controller.signal },
);
controller.abort(new DOMException('worker timeout', 'AbortError'));
await expect(pending).rejects.toThrow('worker timeout');
expect(released).toBe(true);
});
it('classifies a deadline-exhausted zero-progress batch as window, not no_progress', async () => {
let now = 0;
const result = await runExtractAtomsDrain(
{
withLock: passThroughLock,
countRemaining: async () => 5,
runBatch: async () => {
now = 100; // batch consumed the whole window and returned nothing
return { extracted: 0, skipped: 0 };
},
now: () => now,
},
{ windowMs: 100 },
);
expect(result.stopped).toBe('window');
expect(result.batches).toBe(1);
});
it('bounds a hung lock acquisition with the drain deadline signal', async () => {
const started = Date.now();
await expect(runExtractAtomsDrain(
{
withLock: (_work, signal) => new Promise((_resolve, reject) => {
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
}),
countRemaining: async () => 1,
runBatch: async () => ({ extracted: 0, skipped: 0 }),
now: Date.now,
},
{ windowMs: 10 },
)).rejects.toThrow();
expect(Date.now() - started).toBeLessThan(1_000);
});
it('stops on a zero-progress batch (no hot loop)', async () => {
let batches = 0;
const result = await runExtractAtomsDrain(
@@ -242,17 +133,4 @@ describe('shared wiring helper holds the cycle lock (5A)', () => {
expect(src).toContain('cycleLockIdFor(opts.sourceId)');
expect(src).toContain('withRefreshingLock(engine, lockId');
});
// #2750: the deadline signal must reach the phase, the backlog count, AND
// the lock wrapper — and the transcript path (brainDir) must stay wired
// exactly as the routine callers expect (PR #2752 takeover reverted its
// unsanctioned transcript-suppression scope change).
it('threads the drain deadline signal through phase, count, and lock', () => {
const jobsSrc = readFileSync(join(import.meta.dir, '../src/commands/jobs.ts'), 'utf8');
expect(src).toContain('abortSignal: signal');
expect(src).toContain('countExtractAtomsBacklog(engine, extractionSourceId, signal)');
expect(src).toContain('brainDir: opts.brainDir');
expect(src).not.toContain('_transcripts');
expect(jobsSrc).toContain('abortSignal: job.signal');
});
});
+6 -9
View File
@@ -24,18 +24,15 @@ import type { BrainEngine } from '../../src/core/engine.ts';
// Mock engine: healthCheck() calls engine.executeRaw; return empty rows so
// the query path exercises without needing Postgres.
//
// #1849: start() acquires the queue-scoped DB singleton lock via
// tryAcquireDbLock. #2750 routed the acquire upsert through engine.executeRaw
// (signal-boundable) and release through engine.executeRawDirect, so the
// stub returns a single row from the lock upsert (length 1 → acquired) and
// empty rows everywhere else. Each spawned runner is a fresh process, so
// there's no cross-test lock state to clean up.
// #1849: start() now acquires the queue-scoped DB singleton lock via
// tryAcquireDbLock, which uses the postgres `sql` tagged-template escape hatch.
// The stub returns a single row from every call so acquire succeeds (length 1
// → acquired) and refresh/release are no-ops. Each spawned runner is a fresh
// process, so there's no cross-test lock state to clean up.
const sqlStub = (..._args: unknown[]) => Promise.resolve([{ id: 'supervisor-lock' }]);
const mockEngine: Partial<BrainEngine> = {
kind: 'postgres' as const,
executeRaw: async (query: string) =>
query.includes('gbrain_cycle_locks') ? [{ id: 'supervisor-lock' }] : [],
executeRawDirect: async () => [],
executeRaw: async () => [],
sql: sqlStub,
} as unknown as BrainEngine;
+153
View File
@@ -0,0 +1,153 @@
/**
* `gbrain lint` source-glob filter walker integration.
*
* PR #2157 (commit cf9a3b18, `feat/sync-source-glob-filters`) wired
* `sources.config.include_globs` / `exclude_globs` into `gbrain sync` so a
* user could exclude `Resources/veriff/**` and have every subsequent sync
* honor it. The lint command walked the same source dirs blind and emitted
* findings against paths the user had already declared out of scope a
* half-finished feature.
*
* This patch extends the same persisted glob contract to lint:
* - `gbrain lint` gains `--include / --exclude` flags (parallel to sync).
* - `runLintCore` lifts `sources.config.{include,exclude}_globs` for any
* target whose absolute path matches a source row's `local_path`, so the
* cycle.lint phase + Minion lint handlers honor the same filter without
* restating it.
* - The walker in `collectPages` applies the filter using the SAME
* `matchesAnyGlob` helper sync uses, anchored at the target dir (so a
* persisted `Resources/veriff/**` glob written against the source root
* works without rewriting it as an absolute path).
*
* These tests pin the walker contract. The engine-side lift
* (`resolveSourceGlobsForTarget`) is best-effort by design (returns `{}` on
* any error) and is exercised by the dream-cycle lint phase end-to-end.
*/
import { describe, test, expect, beforeAll, afterAll } from 'bun:test';
import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from 'fs';
import { join } from 'path';
import { tmpdir } from 'os';
// runLintCore is the library entry — the same surface the cycle.lint phase
// and Minion handlers call. Exercising it covers the walker via its real
// callsite; testing `collectPages` directly would skip the wiring.
import { runLintCore } from '../src/commands/lint.ts';
// A self-contained content-sanity stub so the test never touches a real
// engine / config file. Empty operator-literal list keeps the content-sanity
// pass silent so the only findings come from the structural rules
// (no-frontmatter etc.).
const STUB_CS = {
fail_on_throw: false,
warn_on_throw: false,
bytes_warn: 1024 * 1024,
operator_literals: [],
};
describe('runLintCore — source-glob walker filter', () => {
let root: string;
beforeAll(() => {
root = mkdtempSync(join(tmpdir(), 'gbrain-lint-globs-'));
// Three subtrees with mixed structured / archive-style content.
// All pages have `# Title` headers but no frontmatter so each one
// emits at least one `no-frontmatter` issue under the default rule set.
mkdirSync(join(root, 'Notes'), { recursive: true });
mkdirSync(join(root, 'Resources', 'veriff'), { recursive: true });
mkdirSync(join(root, 'Resources', 'prior-art', 'archive-v1'), { recursive: true });
writeFileSync(join(root, 'Notes', 'a.md'), '# A\nbody\n');
writeFileSync(join(root, 'Notes', 'b.md'), '# B\nbody\n');
writeFileSync(join(root, 'Resources', 'veriff', 'spec-1.md'), '# Veriff spec 1\nbody\n');
writeFileSync(join(root, 'Resources', 'veriff', 'spec-2.md'), '# Veriff spec 2\nbody\n');
writeFileSync(join(root, 'Resources', 'prior-art', 'archive-v1', 'old.md'), '# Old\nbody\n');
});
afterAll(() => {
rmSync(root, { recursive: true, force: true });
});
test('no filter — walks every .md (regression guard for default behavior)', async () => {
const result = await runLintCore({
target: root,
contentSanity: STUB_CS,
});
expect(result.pages_scanned).toBe(5);
expect(result.pages_with_issues).toBeGreaterThan(0);
});
test('exclude glob skips matching paths (Resources/veriff/** off-limits)', async () => {
const result = await runLintCore({
target: root,
contentSanity: STUB_CS,
exclude: ['Resources/veriff/**'],
});
// 5 total minus 2 veriff specs = 3 pages walked.
expect(result.pages_scanned).toBe(3);
});
test('exclude with multiple patterns is union (veriff + prior-art both skipped)', async () => {
const result = await runLintCore({
target: root,
contentSanity: STUB_CS,
exclude: ['Resources/veriff/**', 'Resources/prior-art/**'],
});
// 5 total minus 3 (2 veriff + 1 archive-v1) = 2 pages walked.
expect(result.pages_scanned).toBe(2);
});
test('include glob narrows the walk to matching paths only', async () => {
const result = await runLintCore({
target: root,
contentSanity: STUB_CS,
include: ['Notes/**'],
});
expect(result.pages_scanned).toBe(2);
});
test('exclude runs AFTER include (same precedence as `gbrain sync`)', async () => {
const result = await runLintCore({
target: root,
contentSanity: STUB_CS,
include: ['**/*.md'],
exclude: ['Resources/**'],
});
// include lets everything through; exclude drops the 3 Resources/* files.
expect(result.pages_scanned).toBe(2);
});
test('empty include / exclude arrays do NOT engage the filter', async () => {
// Symmetric with `parseGlobList` returning undefined for empty input —
// an empty include would otherwise classify every path as a miss and
// silently zero out the lint scope. Pin the guard at the walker level.
const result = await runLintCore({
target: root,
contentSanity: STUB_CS,
include: [],
exclude: [],
});
expect(result.pages_scanned).toBe(5);
});
test('exclude semantics match sync — `**` matches across path segments', async () => {
const result = await runLintCore({
target: root,
contentSanity: STUB_CS,
exclude: ['**/spec-*.md'],
});
// Both Veriff specs match the deep glob; Notes + archive-v1 survive.
expect(result.pages_scanned).toBe(3);
});
test('single-file target bypasses the filter (file mode is not a walk)', async () => {
// A user lints one .md explicitly: filters are a directory-walk concern,
// so the file is processed even if its name would match an exclude.
const result = await runLintCore({
target: join(root, 'Resources', 'veriff', 'spec-1.md'),
contentSanity: STUB_CS,
exclude: ['Resources/veriff/**'],
});
expect(result.pages_scanned).toBe(1);
});
});
+1 -26
View File
@@ -98,39 +98,14 @@ describe('PostgresEngine.executeRawDirect — routing decision (PR #1816)', () =
});
test('already-aborted signal short-circuits with AbortError before routing the query', async () => {
let unsafeCalls = 0;
let ddlCalls = 0;
const readConn: FakeSql = { unsafe: async () => { unsafeCalls++; return []; } };
const readConn = fakeSql('read');
const directConn = fakeSql('direct');
const engine = makeEngine({ dualPoolActive: true, readConn, directConn });
const e = engine as unknown as { connectionManager: { ddl: () => Promise<FakeSql> } };
e.connectionManager.ddl = async () => { ddlCalls++; return directConn; };
const ac = new AbortController();
ac.abort();
await expect(
engine.executeRawDirect('UPDATE minion_jobs SET x=1', [], { signal: ac.signal }),
).rejects.toThrow(/abort/i);
// #2750: short-circuits BEFORE pool routing — no ddl(), no unsafe().
expect(ddlCalls).toBe(0);
expect(unsafeCalls).toBe(0);
});
test('#2750: signal bounds a stalled direct-pool acquisition before unsafe starts', async () => {
let unsafeCalls = 0;
const readConn: FakeSql = { unsafe: async () => { unsafeCalls++; return []; } };
const directConn = fakeSql('direct');
const engine = makeEngine({ dualPoolActive: true, readConn, directConn });
const e = engine as unknown as { connectionManager: { ddl: () => Promise<FakeSql> } };
e.connectionManager.ddl = () => new Promise<FakeSql>(() => {}); // pooler exhausted: never resolves
const started = Date.now();
await expect(engine.executeRawDirect(
'DELETE FROM gbrain_cycle_locks',
[],
{ signal: AbortSignal.timeout(10) },
)).rejects.toThrow(/abort/i);
expect(Date.now() - started).toBeLessThan(1_000);
expect(unsafeCalls).toBe(0);
});
});
+6
View File
@@ -694,6 +694,12 @@ const COLUMN_EXEMPTIONS = new Set<string>([
'minion_jobs.quiet_hours',
'minion_jobs.stagger_key',
'sources.chunker_version',
// #2157 follow-on (migration v125). TEXT column read by performSync's
// `Already up to date` gate; not referenced by any CREATE INDEX. Same
// upgrade-path coverage as sources.chunker_version above: fresh installs
// get it via the CREATE TABLE in src/schema.sql + schema-embedded.ts;
// pre-existing brains get it via the idempotent ALTER TABLE in v125.
'sources.config_fingerprint',
'access_tokens.permissions',
'takes.resolved_quality',
'pages.emotional_weight_recomputed_at',
+121
View File
@@ -159,6 +159,127 @@ describe('sources add', () => {
await expect(runSources(engine, ['add', 'plans', '--path', '/tmp/gstack/plans']))
.rejects.toThrow(/overlaps with existing source "gstack"/);
});
// Glob filters — TODO #3 from the brettdavies fork recon. Pre-fix, the
// `SyncableOptions` shape in `src/core/sync.ts` had been carrying
// `include` / `exclude` since v0.41.13, but commands/sync.ts:1454 never
// populated them and `sources add` had no flag to persist them — so users
// had no way to tell gbrain to skip `Templates/` in an Obsidian vault.
test('--exclude persists glob into sources.config.exclude_globs', async () => {
const { engine, calls } = makeStub({
'SELECT id, name, local_path, last_commit, last_sync_at, config, created_at': [{
id: 'vault',
name: 'vault',
local_path: '/tmp/vault',
last_commit: null,
last_sync_at: null,
config: '{"exclude_globs":["Templates/**"]}',
created_at: new Date(),
}],
});
await runSources(engine, ['add', 'vault', '--path', '/tmp/vault', '--exclude', 'Templates/**']);
const insert = calls.find(c => c.sql.includes('INSERT INTO sources'));
expect(insert!.params[3]).toBe('{"exclude_globs":["Templates/**"]}');
});
test('--include persists glob into sources.config.include_globs', async () => {
const { engine, calls } = makeStub({
'SELECT id, name, local_path, last_commit, last_sync_at, config, created_at': [{
id: 'wiki',
name: 'wiki',
local_path: '/tmp/wiki',
last_commit: null,
last_sync_at: null,
config: '{"include_globs":["people/**"]}',
created_at: new Date(),
}],
});
await runSources(engine, ['add', 'wiki', '--path', '/tmp/wiki', '--include', 'people/**']);
const insert = calls.find(c => c.sql.includes('INSERT INTO sources'));
expect(insert!.params[3]).toBe('{"include_globs":["people/**"]}');
});
test('--exclude is repeatable; preserves order', async () => {
const { engine, calls } = makeStub({
'SELECT id, name, local_path, last_commit, last_sync_at, config, created_at': [{
id: 'vault',
name: 'vault',
local_path: '/tmp/vault',
last_commit: null,
last_sync_at: null,
config: '{}',
created_at: new Date(),
}],
});
await runSources(engine, [
'add', 'vault', '--path', '/tmp/vault',
'--exclude', 'Templates/**',
'--exclude', '.smart-env/**',
'--exclude', 'Drafts/**',
]);
const insert = calls.find(c => c.sql.includes('INSERT INTO sources'));
expect(insert!.params[3]).toBe('{"exclude_globs":["Templates/**",".smart-env/**","Drafts/**"]}');
});
test('--include and --exclude compose in one command (federated source with both filter axes)', async () => {
const { engine, calls } = makeStub({
'SELECT id, name, local_path, last_commit, last_sync_at, config, created_at': [{
id: 'vault',
name: 'vault',
local_path: '/tmp/vault',
last_commit: null,
last_sync_at: null,
config: '{"federated":true,"include_globs":["people/**"],"exclude_globs":["Templates/**"]}',
created_at: new Date(),
}],
});
await runSources(engine, [
'add', 'vault', '--path', '/tmp/vault', '--federated',
'--include', 'people/**',
'--exclude', 'Templates/**',
]);
const insert = calls.find(c => c.sql.includes('INSERT INTO sources'));
expect(insert!.params[3]).toBe(
'{"federated":true,"include_globs":["people/**"],"exclude_globs":["Templates/**"]}',
);
});
test('omitted glob flags leave config untouched (no [] entries persisted)', async () => {
// Regression guard: empty glob arrays must NOT be written. Otherwise a
// brain that never opts into filtering grows {"include_globs": [],
// "exclude_globs": []} cruft in every source row, and the parseGlobList
// path would return undefined anyway (the cruft is purely noise).
const { engine, calls } = makeStub({
'SELECT id, name, local_path, last_commit, last_sync_at, config, created_at': [{
id: 'gstack',
name: 'gstack',
local_path: '/tmp/gstack',
last_commit: null,
last_sync_at: null,
config: '{}',
created_at: new Date(),
}],
});
await runSources(engine, ['add', 'gstack', '--path', '/tmp/gstack']);
const insert = calls.find(c => c.sql.includes('INSERT INTO sources'));
expect(insert!.params[3]).toBe('{}');
});
test('--exclude requires a glob argument', async () => {
const { engine } = makeStub();
const code = await withExitCapture(() => runSources(engine, [
'add', 'vault', '--path', '/tmp/vault', '--exclude',
]));
expect(code).toBe(2);
});
test('--include rejects a flag-like value (--include --path looks like a typo)', async () => {
const { engine } = makeStub();
const code = await withExitCapture(() => runSources(engine, [
'add', 'vault', '--path', '/tmp/vault', '--include', '--federated',
]));
expect(code).toBe(2);
});
});
// ── add — #2707 git-repo validation (CLI wiring) ───────────────
+95
View File
@@ -0,0 +1,95 @@
/**
* #2157 follow-on (migration v125) end-to-end gate wiring.
*
* `test/sync-config-fingerprint.test.ts` pins the persistence + comparison
* primitives (compute/read/write). This file pins the WIRING inside
* `performSync`: with git HEAD unchanged, a drift in the walk-affecting
* `sources.config` fields must break out of the "Already up to date" early
* return and force a full re-walk and the re-stamped fingerprint must
* settle the gate back to `up_to_date` on the following pass. Deleting the
* `configMismatch` term from the gate condition fails this test; none of the
* primitive tests would catch that.
*/
import { test, expect, beforeAll, afterAll } from 'bun:test';
import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from 'fs';
import { join } from 'path';
import { tmpdir } from 'os';
import { execFileSync } from 'child_process';
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
import { performSync } from '../src/commands/sync.ts';
let engine: PGLiteEngine;
let repoPath: string;
function git(cwd: string, ...args: string[]) {
execFileSync('git', args, { cwd, stdio: 'pipe' });
}
beforeAll(async () => {
engine = new PGLiteEngine();
await engine.connect({});
await engine.initSchema();
repoPath = mkdtempSync(join(tmpdir(), 'gbrain-fp-gate-'));
mkdirSync(join(repoPath, 'wiki'));
mkdirSync(join(repoPath, 'memory'));
writeFileSync(join(repoPath, 'wiki', 'page1.md'), '# Page 1\n\nbody\n');
writeFileSync(join(repoPath, 'memory', 'note1.md'), '# Note 1\n\nbody\n');
git(repoPath, 'init');
git(repoPath, 'add', '-A');
git(repoPath, '-c', 'user.email=t@example.com', '-c', 'user.name=t', 'commit', '-m', 'init');
await engine.executeRaw(
`INSERT INTO sources (id, name, local_path, config) VALUES ($1, $2, $3, $4::text::jsonb)`,
['vault', 'vault', repoPath, JSON.stringify({ include_globs: ['wiki/**'] })],
);
}, 60_000);
afterAll(async () => {
await engine?.disconnect();
rmSync(repoPath, { recursive: true, force: true });
});
test('config-glob drift with unchanged git HEAD forces a re-walk, then settles', async () => {
// First sync: row config include_globs = ['wiki/**'], caller threads it
// (as syncOneSource / the single-source CLI path do). memory/* skipped.
const first = await performSync(engine, {
repoPath, sourceId: 'vault', include: ['wiki/**'],
noPull: true, noEmbed: true, full: true,
});
expect(first.status).toBe('first_sync');
expect(await engine.getPage('wiki/page1')).not.toBeNull();
expect(await engine.getPage('memory/note1')).toBeNull();
// No drift, HEAD unchanged: gate stays quiet.
const second = await performSync(engine, {
repoPath, sourceId: 'vault', include: ['wiki/**'],
noPull: true, noEmbed: true,
});
expect(second.status).toBe('up_to_date');
// User widens the persisted globs (what `gbrain sources add --include`
// writes). Git HEAD has NOT moved.
await engine.executeRaw(
`UPDATE sources SET config = $1::text::jsonb WHERE id = $2`,
[JSON.stringify({ include_globs: ['wiki/**', 'memory/**'] }), 'vault'],
);
// Pre-fix this returned `up_to_date` (HEAD unchanged) and memory/note1
// stayed missing until a manual `--full`. The fingerprint gate must force
// the full re-walk instead.
const third = await performSync(engine, {
repoPath, sourceId: 'vault', include: ['wiki/**', 'memory/**'],
noPull: true, noEmbed: true,
});
expect(third.status).not.toBe('up_to_date');
expect(await engine.getPage('memory/note1')).not.toBeNull();
// Re-stamped fingerprint matches the current row: gate settles.
const fourth = await performSync(engine, {
repoPath, sourceId: 'vault', include: ['wiki/**', 'memory/**'],
noPull: true, noEmbed: true,
});
expect(fourth.status).toBe('up_to_date');
});
+234
View File
@@ -0,0 +1,234 @@
/**
* #2157 follow-on (migration v125 `sources.config_fingerprint`).
*
* The "Already up to date" gate at performSync's git-HEAD equality check
* honored chunker_version match but ignored `sources.config` drift.
* Changing `sources.config.exclude_globs` (or include_globs / strategy)
* had no observable effect on the next sync because git HEAD was
* unchanged the gate returned early and the new walk scope never
* applied. This file exercises the persistence shape + drift detection
* end-to-end on PGLite, including:
*
* - Migration v125 actually adds the column (regression guard against
* a future re-numbering or accidental deletion).
* - read/write round-trips preserve the value.
* - The fingerprint differs across the three walk-affecting fields
* and is order-insensitive on the array fields.
* - NULL fingerprint on pre-v125 rows treats as "not stamped" so a
* first post-upgrade sync doesn't spuriously force-full.
* - A toggle-and-revert leaves the stored fingerprint matching the
* current row, so the gate stays quiet.
*
* The wired-up gate behavior (force-full triggered on mismatch) is
* exercised by the existing sync end-to-end tests; here we pin the
* persistence + comparison primitives the gate depends on.
*/
import { describe, test, expect, beforeAll, afterAll } from 'bun:test';
import {
computeSourceConfigFingerprint,
readConfigFingerprint,
writeConfigFingerprint,
} from '../src/commands/sync.ts';
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
let engine: PGLiteEngine;
beforeAll(async () => {
engine = new PGLiteEngine();
await engine.connect({});
await engine.initSchema();
}, 60_000);
afterAll(async () => {
await engine?.disconnect();
});
/** Insert a fresh source row with the given config. Returns the id. */
async function makeSource(
id: string,
config: Record<string, unknown> = {},
): Promise<string> {
await engine.executeRaw(
`INSERT INTO sources (id, name, local_path, config) VALUES ($1, $2, $3, $4::text::jsonb)`,
[id, id, `/tmp/${id}`, JSON.stringify(config)],
);
return id;
}
describe('migration v125 — sources.config_fingerprint column', () => {
test('column exists on the sources table', async () => {
const rows = await engine.executeRaw<{ column_name: string }>(
`SELECT column_name FROM information_schema.columns
WHERE table_name = 'sources' AND column_name = 'config_fingerprint'`,
);
expect(rows).toHaveLength(1);
});
test('column is nullable (preserves pre-migration row semantics)', async () => {
const rows = await engine.executeRaw<{ is_nullable: string }>(
`SELECT is_nullable FROM information_schema.columns
WHERE table_name = 'sources' AND column_name = 'config_fingerprint'`,
);
expect(rows[0]?.is_nullable).toBe('YES');
});
});
describe('readConfigFingerprint / writeConfigFingerprint — persistence round-trip', () => {
test('round-trip: write then read returns the same value', async () => {
const id = await makeSource('rt-basic', { exclude_globs: ['Templates/**'] });
const fp = computeSourceConfigFingerprint({ exclude_globs: ['Templates/**'] });
await writeConfigFingerprint(engine, id, fp);
const got = await readConfigFingerprint(engine, id);
expect(got).toBe(fp);
});
test('NULL on never-stamped row (pre-v125 semantics)', async () => {
const id = await makeSource('rt-never');
const got = await readConfigFingerprint(engine, id);
expect(got).toBeNull();
});
test('undefined sourceId returns null (legacy non-source-scoped sync)', async () => {
const got = await readConfigFingerprint(engine, undefined);
expect(got).toBeNull();
});
test('write with undefined sourceId is a no-op (does not throw)', async () => {
// The legacy global-sync code path hits this branch; the guard must
// be silent rather than fail the sync run.
await writeConfigFingerprint(engine, undefined, 'deadbeef'.repeat(8));
// No assertion beyond "did not throw"; the function returns void.
});
test('overwrite: a second write replaces the prior fingerprint', async () => {
const id = await makeSource('rt-overwrite');
await writeConfigFingerprint(engine, id, 'a'.repeat(64));
await writeConfigFingerprint(engine, id, 'b'.repeat(64));
const got = await readConfigFingerprint(engine, id);
expect(got).toBe('b'.repeat(64));
});
});
describe('end-to-end drift simulation — the gate semantics this column enables', () => {
test('first stamp matches computed fingerprint of the row config', async () => {
const cfg = { exclude_globs: ['Templates/**', 'Photos/**'], strategy: 'markdown' };
const id = await makeSource('e2e-first-stamp', cfg);
const computed = computeSourceConfigFingerprint(cfg);
await writeConfigFingerprint(engine, id, computed);
expect(await readConfigFingerprint(engine, id)).toBe(computed);
});
test('exclude_globs mutation makes stored != current (drift detected)', async () => {
const before = { exclude_globs: ['Templates/**'] };
const after = { exclude_globs: ['Templates/**', 'Photos/**'] };
const id = await makeSource('e2e-exclude-drift', before);
const beforeFp = computeSourceConfigFingerprint(before);
await writeConfigFingerprint(engine, id, beforeFp);
// Simulate the user mutating sources.config via `gbrain sources add
// --exclude`. The gate's next read of (stored, computed-from-current)
// detects the drift and forces a re-walk.
await engine.executeRaw(
`UPDATE sources SET config = $1::text::jsonb WHERE id = $2`,
[JSON.stringify(after), id],
);
const afterFp = computeSourceConfigFingerprint(after);
const stored = await readConfigFingerprint(engine, id);
expect(stored).toBe(beforeFp);
expect(stored).not.toBe(afterFp);
});
test('toggle-and-revert: add then remove same pattern leaves stored matching current', async () => {
const original = { exclude_globs: ['Templates/**'] };
const id = await makeSource('e2e-toggle', original);
const originalFp = computeSourceConfigFingerprint(original);
await writeConfigFingerprint(engine, id, originalFp);
// Add a pattern (drift) then remove it (revert).
await engine.executeRaw(
`UPDATE sources SET config = $1::text::jsonb WHERE id = $2`,
[JSON.stringify({ exclude_globs: ['Templates/**', 'Photos/**'] }), id],
);
await engine.executeRaw(
`UPDATE sources SET config = $1::text::jsonb WHERE id = $2`,
[JSON.stringify(original), id],
);
const revertedFp = computeSourceConfigFingerprint(original);
expect(revertedFp).toBe(originalFp);
expect(await readConfigFingerprint(engine, id)).toBe(originalFp);
// ⇒ Gate compares storedFp (==originalFp) to currentFp (==originalFp): no drift, no force-full.
});
test('include_globs drift detected independently', async () => {
const before = { include_globs: ['people/**'] };
const after = { include_globs: ['people/**', 'companies/**'] };
const id = await makeSource('e2e-include-drift', before);
await writeConfigFingerprint(engine, id, computeSourceConfigFingerprint(before));
await engine.executeRaw(
`UPDATE sources SET config = $1::text::jsonb WHERE id = $2`,
[JSON.stringify(after), id],
);
const stored = await readConfigFingerprint(engine, id);
const current = computeSourceConfigFingerprint(after);
expect(stored).not.toBe(current);
});
test('strategy drift detected', async () => {
const before = { strategy: 'markdown' };
const after = { strategy: 'code' };
const id = await makeSource('e2e-strategy-drift', before);
await writeConfigFingerprint(engine, id, computeSourceConfigFingerprint(before));
await engine.executeRaw(
`UPDATE sources SET config = $1::text::jsonb WHERE id = $2`,
[JSON.stringify(after), id],
);
expect(await readConfigFingerprint(engine, id))
.not.toBe(computeSourceConfigFingerprint(after));
});
test('mutating unrelated config field (federated) does NOT drift', async () => {
// The fingerprint hashes ONLY walk-affecting fields. Federation
// changes search visibility, not the walk set — must not invalidate
// the checkpoint.
const id = await makeSource('e2e-federated-toggle', {
federated: true,
exclude_globs: ['Templates/**'],
});
await writeConfigFingerprint(
engine,
id,
computeSourceConfigFingerprint({ exclude_globs: ['Templates/**'] }),
);
await engine.executeRaw(
`UPDATE sources SET config = $1::text::jsonb WHERE id = $2`,
[
JSON.stringify({ federated: false, exclude_globs: ['Templates/**'] }),
id,
],
);
const stored = await readConfigFingerprint(engine, id);
const current = computeSourceConfigFingerprint({
federated: false,
exclude_globs: ['Templates/**'],
});
expect(stored).toBe(current);
});
test('double-encoded JSONB config (the sources-add stringify bug) hashes equivalently to the parsed object', async () => {
// `gbrain sources add` writes `JSON.stringify(config)::jsonb`, which
// double-encodes the value into a JSON-string scalar (`"{\"x\":1}"`)
// rather than a proper JSONB object. The defensive reader in
// postgres-engine.ts:1274 + readSourceConfig parses the string back
// before the fingerprint sees it, so a double-encoded row and a
// properly-shaped row must fingerprint identically.
const cfg = { exclude_globs: ['Templates/**'], strategy: 'markdown' };
const direct = computeSourceConfigFingerprint(cfg);
// The pure compute fn handles a pre-parsed object; the persistence
// layer's job is to deliver a parsed object. We assert that the
// round-trip a real read would produce (parse the string scalar)
// hashes to the same value.
const parsed = JSON.parse(JSON.stringify(cfg));
expect(computeSourceConfigFingerprint(parsed)).toBe(direct);
});
});
+63
View File
@@ -355,6 +355,69 @@ describe('sync monorepo subdir-source support (#753/#774)', () => {
expect(await engine.getPage('wiki/draft-a')).toBeNull();
});
// ─────────────────────────────────────────────────────────────────────────
// --include: allow-list counterpart (#2156). Same scope-relative anchoring
// as --exclude; exclude applies after include.
// ─────────────────────────────────────────────────────────────────────────
test('--include: only matching files import on full sync', async () => {
const { performSync } = await import('../src/commands/sync.ts');
const result = await performSync(engine, {
repoPath,
include: ['wiki/**'],
noPull: true,
noEmbed: true,
full: true,
});
expect(result.status).toBe('first_sync');
expect(result.added).toBe(2); // wiki/page1 + wiki/page2; memory/* miss the allow-list
expect(await engine.getPage('wiki/page1')).not.toBeNull();
expect(await engine.getPage('memory/note1')).toBeNull();
});
test('--include applies to the incremental path too', async () => {
const { performSync } = await import('../src/commands/sync.ts');
const first = await performSync(engine, {
repoPath,
include: ['wiki/**'],
noPull: true,
noEmbed: true,
full: true,
});
expect(first.status).toBe('first_sync');
writeFileSync(join(repoPath, 'wiki', 'page3.md'), mdPage('Wiki Page 3'));
writeFileSync(join(repoPath, 'memory', 'note3.md'), mdPage('Memory Note 3'));
gitCommit(repoPath, 'more pages');
const second = await performSync(engine, {
repoPath,
include: ['wiki/**'],
noPull: true,
noEmbed: true,
});
expect(second.status).toBe('synced');
expect(second.added).toBe(1); // wiki/page3 only; memory/note3 misses the allow-list
expect(await engine.getPage('wiki/page3')).not.toBeNull();
expect(await engine.getPage('memory/note3')).toBeNull();
});
test('--exclude applies after --include (path in both is rejected)', async () => {
const { performSync } = await import('../src/commands/sync.ts');
const result = await performSync(engine, {
repoPath,
include: ['wiki/**'],
exclude: ['wiki/page2.md'],
noPull: true,
noEmbed: true,
full: true,
});
expect(result.status).toBe('first_sync');
expect(result.added).toBe(1); // page1 only: page2 included then excluded
expect(await engine.getPage('wiki/page1')).not.toBeNull();
expect(await engine.getPage('wiki/page2')).toBeNull();
});
// ─────────────────────────────────────────────────────────────────────────
// --exclude '**/*' emits warning (NAV-4)
// ─────────────────────────────────────────────────────────────────────────
+212
View File
@@ -0,0 +1,212 @@
/**
* TODO #3 `parseGlobList` defensive parse.
*
* `sources.config` is a JSONB column with no schema. The runtime can find
* anything in `config.include_globs` / `config.exclude_globs`:
* - A user `gbrain sources add` wrote `["people/**"]` (the happy path).
* - A stray hand-edit wrote `"people/**"` (string, not array).
* - A future migration's null default.
* - A test fixture that left the column at `{}`.
*
* The parse must produce `string[] | undefined` so the downstream
* `SyncOpts.include` / `SyncOpts.exclude` are either undefined (no filter)
* or a non-empty list of usable globs. Returning `[]` would make
* `commands/sync.ts:1454` engage the filter loop with an empty allow-list
* that classifies every path as `include-glob-miss`.
*/
import { describe, test, expect } from 'bun:test';
import { parseGlobList, mergeGlobs, computeSourceConfigFingerprint } from '../src/commands/sync.ts';
describe('parseGlobList — JSONB-safe coercion to string[] | undefined', () => {
test('happy path: array of strings round-trips identically', () => {
expect(parseGlobList(['people/**', 'companies/**'])).toEqual(['people/**', 'companies/**']);
});
test('single-element array returned as-is', () => {
expect(parseGlobList(['Templates/**'])).toEqual(['Templates/**']);
});
test('non-array values return undefined (string, object, number, null)', () => {
expect(parseGlobList('Templates/**')).toBeUndefined();
expect(parseGlobList({ globs: ['Templates/**'] })).toBeUndefined();
expect(parseGlobList(42)).toBeUndefined();
expect(parseGlobList(null)).toBeUndefined();
expect(parseGlobList(undefined)).toBeUndefined();
});
test('empty array returns undefined (no engagement of the filter loop)', () => {
// Critical: a literal `[]` must not slip through. Empty `include` in
// SyncableOptions silently passes everything (good), but empty
// `exclude` is fine too — the real motivation is to keep `SyncOpts`
// unset so callers can ignore the field entirely. Symmetric with the
// `omitted glob flags leave config untouched` regression guard in
// sources.test.ts.
expect(parseGlobList([])).toBeUndefined();
});
test('mixed array drops non-string entries and keeps the rest', () => {
expect(parseGlobList(['people/**', 42, null, 'companies/**'])).toEqual([
'people/**',
'companies/**',
]);
});
test('empty strings dropped (a `""` glob would match every path)', () => {
expect(parseGlobList(['', 'people/**', ''])).toEqual(['people/**']);
});
test('array of only empty strings collapses to undefined', () => {
expect(parseGlobList(['', '', ''])).toBeUndefined();
});
});
describe('mergeGlobs — CLI flags union with persisted source-config globs', () => {
test('both sides present: union, deduped, CLI first', () => {
expect(mergeGlobs(['a/**', 'b/**'], ['b/**', 'c/**'])).toEqual(['a/**', 'b/**', 'c/**']);
});
test('CLI only', () => {
expect(mergeGlobs(['a/**'], undefined)).toEqual(['a/**']);
});
test('persisted only', () => {
expect(mergeGlobs([], ['Templates/**'])).toEqual(['Templates/**']);
});
test('neither side: undefined so SyncOpts stays unset', () => {
expect(mergeGlobs([], undefined)).toBeUndefined();
});
});
/**
* #2157 follow-on (sources_config_fingerprint, migration v125).
*
* computeSourceConfigFingerprint hashes the walk-affecting fields of
* sources.config (strategy + include_globs + exclude_globs) so the
* "Already up to date" gate at performSync's git-HEAD equality check
* can detect drift and force a re-walk. These cases pin the contract
* the gate depends on:
*
* - Deterministic over equivalent inputs (order-insensitive,
* defensively-coerced via parseGlobList).
* - Sensitive to each walk-affecting field separately.
* - Insensitive to fields the walker doesn't read (federated,
* unrelated keys).
* - A toggle-and-revert is a no-op (returns to the original hash).
*
* Without the canonicalization the gate would fire spuriously on
* cosmetic changes (e.g. a user re-ordering their exclude list) and
* miss real drift (e.g. an add-then-remove that nets to a different
* effective set than the stored fingerprint).
*/
describe('computeSourceConfigFingerprint — walk-affecting config drift detector', () => {
test('empty config produces a stable hash', () => {
const a = computeSourceConfigFingerprint({});
const b = computeSourceConfigFingerprint({});
expect(a).toBe(b);
expect(a).toMatch(/^[a-f0-9]{64}$/);
});
test('null / undefined / missing config all hash the same', () => {
const empty = computeSourceConfigFingerprint({});
expect(computeSourceConfigFingerprint(null)).toBe(empty);
expect(computeSourceConfigFingerprint(undefined)).toBe(empty);
});
test('same config → same hash (deterministic)', () => {
const cfg = { strategy: 'markdown', exclude_globs: ['Templates/**', 'Photos/**'] };
expect(computeSourceConfigFingerprint(cfg)).toBe(computeSourceConfigFingerprint(cfg));
});
test('array order does not affect hash (canonical sort)', () => {
const a = computeSourceConfigFingerprint({ exclude_globs: ['a/**', 'b/**', 'c/**'] });
const b = computeSourceConfigFingerprint({ exclude_globs: ['c/**', 'a/**', 'b/**'] });
expect(a).toBe(b);
});
test('exclude_globs change → different hash', () => {
const a = computeSourceConfigFingerprint({ exclude_globs: ['Templates/**'] });
const b = computeSourceConfigFingerprint({ exclude_globs: ['Templates/**', 'Photos/**'] });
expect(a).not.toBe(b);
});
test('include_globs change → different hash', () => {
const a = computeSourceConfigFingerprint({ include_globs: ['people/**'] });
const b = computeSourceConfigFingerprint({ include_globs: ['people/**', 'companies/**'] });
expect(a).not.toBe(b);
});
test('strategy change → different hash', () => {
const a = computeSourceConfigFingerprint({ strategy: 'markdown' });
const b = computeSourceConfigFingerprint({ strategy: 'code' });
expect(a).not.toBe(b);
});
test('strategy unset vs set differ', () => {
const unset = computeSourceConfigFingerprint({});
const set = computeSourceConfigFingerprint({ strategy: 'markdown' });
expect(unset).not.toBe(set);
});
test('add-then-remove returns to original hash (toggle is a no-op)', () => {
const original = computeSourceConfigFingerprint({ exclude_globs: ['Templates/**'] });
const added = computeSourceConfigFingerprint({ exclude_globs: ['Templates/**', 'Photos/**'] });
const reverted = computeSourceConfigFingerprint({ exclude_globs: ['Templates/**'] });
expect(added).not.toBe(original);
expect(reverted).toBe(original);
});
test('non-walk-affecting fields are ignored (federated, unrelated keys)', () => {
const a = computeSourceConfigFingerprint({ exclude_globs: ['Templates/**'], federated: true });
const b = computeSourceConfigFingerprint({ exclude_globs: ['Templates/**'], federated: false });
const c = computeSourceConfigFingerprint({ exclude_globs: ['Templates/**'], some_unrelated_key: 'value' });
expect(a).toBe(b);
expect(a).toBe(c);
});
test('non-string strategy coerced to null (defensive)', () => {
// A hand-edited row could leave `strategy: 42` or `strategy: {}` — both
// collapse to the same shape as `strategy: undefined` so the fingerprint
// doesn't reflect a value the walker can't honor anyway.
const empty = computeSourceConfigFingerprint({});
expect(computeSourceConfigFingerprint({ strategy: 42 })).toBe(empty);
expect(computeSourceConfigFingerprint({ strategy: {} })).toBe(empty);
expect(computeSourceConfigFingerprint({ strategy: null })).toBe(empty);
});
test('defensive parsing: mixed-type glob arrays hash same as cleaned arrays', () => {
// parseGlobList drops non-string + empty entries; the fingerprint must
// reflect what the walker actually uses, not what the raw row says.
const dirty = computeSourceConfigFingerprint({
exclude_globs: ['Templates/**', 42, null, '', 'Photos/**'],
});
const clean = computeSourceConfigFingerprint({
exclude_globs: ['Templates/**', 'Photos/**'],
});
expect(dirty).toBe(clean);
});
test('empty array and missing field hash identically', () => {
const missing = computeSourceConfigFingerprint({});
const emptyArray = computeSourceConfigFingerprint({ exclude_globs: [] });
const emptyAfterClean = computeSourceConfigFingerprint({ exclude_globs: ['', '', ''] });
expect(emptyArray).toBe(missing);
expect(emptyAfterClean).toBe(missing);
});
test('non-array exclude_globs (string, object) hash same as missing', () => {
const missing = computeSourceConfigFingerprint({});
expect(computeSourceConfigFingerprint({ exclude_globs: 'Templates/**' })).toBe(missing);
expect(computeSourceConfigFingerprint({ exclude_globs: { foo: 'bar' } })).toBe(missing);
});
test('SHA-256 output shape: 64 hex characters', () => {
const fp = computeSourceConfigFingerprint({
strategy: 'markdown',
include_globs: ['people/**'],
exclude_globs: ['Templates/**', '.git/**'],
});
expect(fp).toMatch(/^[a-f0-9]{64}$/);
});
});