admin/gbrain
1
0
Fork 0
mirror of https://github.com/garrytan/gbrain.git synced 2026-07-27 22:15:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

v0.22.13 feat: parallel sync — bounded concurrent imports (#490)

Browse Source 
* feat: parallel sync — bounded concurrent imports (#489)

gbrain sync --concurrency N (alias --workers N) parallelizes the import
phase using per-worker Postgres engine instances with an atomic queue
index (same proven pattern as gbrain import --workers N).

Auto-concurrency: when a sync touches >100 files and the user didn't
explicitly set --concurrency, defaults to 4 workers. Small incremental
syncs (<50 files) stay serial. Full syncs auto-detect Postgres and
default to 4 workers.

Minion sync handler defaults to concurrency=4, configurable via job
params: {"concurrency": 8}.

Delete and rename phases remain serial (order-dependent, fast).
PGLite falls back to serial automatically (single-connection engine).

Changes:
- src/commands/sync.ts: SyncOpts.concurrency, parallel import loop in
  performSync incremental path, --workers passthrough in performFullSync
- src/commands/jobs.ts: sync handler accepts concurrency param (default 4)
- CHANGELOG.md: v0.23.0 parallel sync entry

All 37 existing sync tests pass. Typecheck clean.

* feat: shared concurrency policy + db-lock primitive

src/core/sync-concurrency.ts — single source of truth for autoConcurrency()
+ parseWorkers() + shouldRunParallel() + constants. Replaces three drifted
call-site policies (performSync, performFullSync, jobs handler).

src/core/db-lock.ts — generic tryAcquireDbLock(engine, lockId, ttlMinutes)
over the existing gbrain_cycle_locks table. Parameterized lock id so
performSync (gbrain-sync) can nest cleanly under cycle.ts (gbrain-cycle)
without deadlock.

test/sync-concurrency.test.ts — 17 cases covering PGLite-forces-serial,
explicit override clamping, auto-path threshold, parseWorkers validation
(rejects 0, negatives, NaN, decimals, trailing chars).

No consumers yet; subsequent commits wire sync.ts, import.ts, and jobs.ts
to use these helpers.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix: harden performSync — writer lock, head-drift gate, engine.kind

CODEX-2: wrap performSync body in a gbrain-sync DB lock so two concurrent
syncs (manual + autopilot, two terminals, two Conductor workspaces) cannot
both read last_commit, both write it unconditionally, and let the last
writer win. cycle.ts continues to hold gbrain-cycle for its broader scope;
the two ids nest cleanly.

CODEX-3: capture git HEAD at sync entry, re-rev-parse after the import
phase, refuse to advance last_commit if HEAD drifted (someone ran
git checkout / git pull mid-sync). Vanished files now go into failedFiles
instead of silent-skip — same gating mechanism, no more bookmark advance
past unimported work.

A1: replace both PGLite detection sites with engine.kind === 'pglite'.
The constructor.name sniff is gone (breaks under bundling) and so is the
inconsistent config?.engine string check.

A2: connect worker engines serially into an array, run inside try/finally
so disconnect always fires — even on partial connect failure, OOM, or
mid-import abort. Prior Promise.all(...disconnect) leaked the 8 worker
connections on any panic path.

Q1: explicit --workers / opts.concurrency now bypasses the >50-file floor.
User opt-in beats the auto-path safety net.

Q3: drop the config!.database_url! non-null assertions; fall back to serial
when database_url is unset instead of crashing on TypeError.

Q4: worker-count banner moves from console.log to console.error so stdout
stays clean for --json output.

test/sync-parallel.test.ts — 7 cases over PGLite covering the bookmark
gate under concurrency request, the head-drift gate, vanished-file
failure capture, PGLite-stays-serial, and the writer-lock contract.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix: import.ts — engine.kind discriminator, worker try/finally, parseWorkers

A1: replace the config?.engine === 'pglite' string sniff with
engine.kind === 'pglite' to match sync.ts and the v0.13.1 contract.

A2: wrap worker engine creation + the parallel loop in try/finally so
disconnects always fire — same pattern as sync.ts. Worker engines now
push onto an array as they connect (rather than Promise.all) so the
finally block can clean up partial-connect state.

Q2: route --workers parsing through the shared parseWorkers() helper.
parseInt-with-no-validation is gone — '0', '-3', 'foo', '1.5' now exit
with a clear error message instead of silently falling through.

Q3: drop the config!.database_url! non-null assertion; fall back to
serial when database_url is unset.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix: jobs.ts sync handler — resolve sourceId, autoConcurrency

CODEX-1: resolve sourceId at handler entry by looking up sources.local_path.
Mirrors cycle.ts:480's autopilot-cycle fix (PR #475). Without this, every
Minion sync job on a multi-source brain reads global config.sync.last_commit
instead of the per-source anchor, which on a regularly-GC'd repo can drop
out of git history and trigger 30-min full reimports every cycle.

The handler accepts an optional sourceId job param for callers that want
to override; falls back to the resolveSourceForDir lookup when absent.

CODEX-4: replace the hardcoded concurrency=4 default with the shared
autoConcurrency policy. Behavior is now consistent between CLI sync,
the Minion handler, and the autopilot cycle's sync phase. Jobs that
request a specific concurrency via job.data.concurrency still win.

noEmbed default stays at true — embed is a separate job (submit
gbrain embed --stale, OR rely on the autopilot cycle's embed phase).
The doc comment makes that contract explicit.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test: e2e parallel sync against real Postgres + benchmark

DATABASE_URL-gated E2E coverage that PGLite-only tests can't reach:

T2 — happy path: 60 files imported at concurrency=4, all 60 pages land
in the DB, with a pg_stat_activity probe before/after to confirm worker
engines (4 × 2 connections) actually disconnected.

P4 — benchmark: 120-file fixture, serial vs concurrency=4 timing.
Emits a single-line `SYNC_PARALLEL_BENCH 120 files | serial=Xms |
parallel(4)=Yms | speedup=Zx` so the CHANGELOG can quote a real
number instead of an unbacked '~4×' claim. Asserts parallel <=
serial * 1.5 to allow for noisy CI but fail genuine regressions.

Skips gracefully when DATABASE_URL is unset (consistent with the rest
of test/e2e/).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* chore: v0.22.10 release notes + sync follow-up TODO

VERSION + package.json + bun.lock: 0.22.5/0.22.6 → 0.22.10. Repo had
existing drift between VERSION and package.json on master; this commit
brings them back in sync at the bumped value.

CHANGELOG.md: v0.22.10 entry replaces the unfinished v0.23.0 stub from
PR #490's original commit. Voice-rule clean (no em dashes, no AI
vocabulary), real benchmark numbers from the new E2E test
(serial=289ms parallel(4)=221ms speedup=1.31x), additive worker-pool
note (A3), 'To take advantage of v0.22.10' self-repair block per
CLAUDE.md convention.

TODOS.md: A4 follow-up filed — plumb resolved database_url through
SyncOpts so performSync / performFullSync / import.ts don't each call
loadConfig() separately. Deferred to a future patch; not on the
v0.22.10 critical path.

Patch (not minor) framing held even though new CLI surface lands here;
release-notes prose names the behavior change explicitly so users know
to read them.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* docs: update CLAUDE.md + README for v0.22.10 sync hardening

CLAUDE.md:
- New "Key files" entries for src/core/sync-concurrency.ts and
  src/core/db-lock.ts (both v0.22.10).
- New "Key files" entry for src/commands/sync.ts (covers the lock,
  head-drift gate, engine.kind discriminator, vanished-file failure
  capture, parallel branch wiring).
- Updated src/commands/jobs.ts entry with v0.22.10 sourceId
  resolution + autoConcurrency policy + noEmbed contract.
- Added test/sync-concurrency.test.ts and test/sync-parallel.test.ts
  to the unit-test list with case counts.
- Added test/e2e/sync-parallel.test.ts to the E2E section with the
  SYNC_PARALLEL_BENCH grep marker for CHANGELOG quoting.
- Added "Key commands added in v0.22.10" section: gbrain sync --workers,
  gbrain import --workers (parseWorkers validation).

README.md: added --workers flag to the IMPORT section's gbrain sync
and gbrain import lines, with the >100-file auto-parallelize note.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* chore: bump version slot to v0.22.13

VERSION 0.22.10 → 0.22.13. Master moved to 0.22.8 plus claimed slots
0.22.9-0.22.12 in sibling workspaces; 0.22.13 is the next free slot for
this PR's parallel-sync hardening work.

Updated all v0.22.10 references in CHANGELOG.md (release header +
self-repair block), TODOS.md (D-PR490-1 follow-up tag), CLAUDE.md
(Key files entries + tests + commands subsection), and the inline
v0.22.10 markers in src/core/sync-concurrency.ts, src/core/db-lock.ts,
src/commands/sync.ts, src/commands/import.ts, src/commands/jobs.ts,
test/sync-parallel.test.ts, test/e2e/sync-parallel.test.ts.

No behavioral change. CHANGELOG header rewrite, content unchanged.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* chore: regenerate llms-full.txt for v0.22.13 doc updates

CI's build-llms generator test failed because llms-full.txt was stale
relative to the README + CLAUDE.md updates this PR added (--workers
flag in the IMPORT section, sync-concurrency.ts/db-lock.ts/sync.ts
entries in the Key files section).

Per CLAUDE.md: "Run \`bun run build:llms\` after adding a new doc."
The test test/build-llms.test.ts:67 verifies committed bundles match
generator output — now they do again.

llms.txt was already in sync (no curated config additions); only
llms-full.txt needed the regen.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: root <root@localhost>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Garry Tan
2026-04-29 22:53:41 -07:00
committed by GitHub
co-authored by root Claude Opus 4.7
parent 1e73e93344
commit e96f054cf0
16 changed files with 1232 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+70
View File
@@ -2,6 +2,76 @@
All notable changes to GBrain will be documented in this file.
## [0.22.13] - 2026-04-28
**Sync got faster, and the bookmark stopped lying.**
**Parallel imports, a real writer lock, and a head-drift gate that catches the worst race.**
The headline is `gbrain sync --workers N`: per-worker Postgres engines with an atomic queue index, same pattern as `gbrain import --workers N`. On a 7,000-page brain that used to take 25+ minutes, the import phase now runs across 4 workers by default. The reproducible benchmark in `test/e2e/sync-parallel.test.ts` shows `parallel(4)` finishing 1.3× faster than serial on a 120-file fixture against local Postgres (`serial=289ms parallel(4)=221ms`). The speedup grows on larger brains and slower-roundtrip databases (Supabase, remote PgBouncer) because the worker setup cost amortizes over more files. But the bigger story is that the sync writer is finally exclusive across processes, and the `last_commit` bookmark refuses to advance when git HEAD has drifted out from under us. The silent-skip-then-advance pathology has survived every prior sync hardening pass. It is dead now.
### What you can do now
- `gbrain sync --workers 4` (alias `--concurrency 4`) parallelizes the import phase. Each worker holds 2 connections, so total Postgres connections during the parallel phase is `workers * 2` plus your caller's pool. At the default of 4 workers and a 10-connection caller pool, that's up to 18 connections, well under PgBouncer's `max_client_conn` default of 100 but worth knowing on tight Supabase tiers.
- **Auto-concurrency:** if you don't pass `--workers`, sync uses 4 workers when the diff exceeds 100 files. Smaller diffs stay serial. Explicit `--workers` always wins (even on a 30-file diff). PGLite forces serial regardless, since it's a single-connection engine.
- **Full sync** routes through the same path. First syncs on large brains parallelize automatically.
- **Minion `sync` jobs** also use the new `autoConcurrency()` policy. Behavior is now consistent between CLI sync, the Minion handler, and the autopilot cycle's sync phase. (`noEmbed` defaults to `true` in the jobs handler. Submit `gbrain embed --stale` as a separate job when needed, or rely on the autopilot cycle's embed phase.)
- **`--workers` validation is loud now.** `--workers 0`, `--workers -3`, `--workers foo`, `--workers 1.5` all exit with an error message. The prior behavior silently fell through to auto-concurrency (4 workers), the opposite of what you typed.
### Correctness fixes you didn't have to ask for
- **Cross-process writer lock.** Two `gbrain sync` calls (manual + autopilot, two terminals, two Conductor workspaces) used to read the same `last_commit`, both write it, and let the last writer win. The new `gbrain-sync` row in `gbrain_cycle_locks` serializes the writer window. Same-process reentrance from the autopilot cycle handler was already covered by the broader `gbrain-cycle` lock; sync's lock is narrower and runs underneath it.
- **Head-drift gate.** If `git checkout` or `git pull` runs in your worktree mid-sync (Conductor sibling workspace, ad-hoc terminal), the captured `headCommit` no longer matches HEAD when sync finishes. `last_commit` no longer advances in that case. The next sync re-walks the diff against the new HEAD instead of silently moving the bookmark past unimported work.
- **Vanished files now block bookmark advance.** A file the diff said exists at `headCommit` but is gone from disk used to register as a benign skip. It now goes into `failedFiles` and gates `last_commit` the same way a parse failure does.
- **Per-source bookmark for Minion `sync` jobs.** The job handler now resolves `sourceId` from the repo path (mirrors the autopilot cycle's `cycle.ts` fix from PR #475). On multi-source brains, this prevents the 30-min full-reimport-every-cycle behavior caused by reading the global `config.sync.last_commit` anchor when the per-source row would have been correct.
- **Worker connection cleanup.** Worker engines now disconnect inside `try/finally`, even on partial connect failure or mid-import error. The prior `Promise.all(...disconnect)` ran outside any try/finally, so panic-path leaks never released the 8 worker connections.
- **Engine detection unified.** Both PGLite-detection sites in sync.ts now use `engine.kind === 'pglite'` (the discriminator added in v0.13.1). The `engine.constructor.name === 'PGLiteEngine'` sniff is gone, since it broke under bundling and was inconsistent with the other site's `config.engine` string check.
### What this means for you
If you run autopilot on a 7,000-page Postgres brain, your sync cycle gets faster on day one with no flags. If you have ever felt the bookmark "skip past" work that didn't import, you'll stop seeing it. If you have multiple Conductor workspaces poking the same brain, you'll either wait politely on the writer lock or get a clear "another sync is in progress" error. None of this requires a config change.
## To take advantage of v0.22.13
`gbrain upgrade` should do this automatically. If you want to use the new flags right now:
1. **For a one-off speed win on a large brain:**
```bash
gbrain sync --workers 4
```
Or for incremental syncs that touch >100 files, just run `gbrain sync`. Auto-concurrency fires.
2. **For your autopilot cycle:** no action. The Minion `sync` handler picks up the new auto-concurrency policy automatically.
3. **Verify the writer lock is working:**
```bash
gbrain sync &
gbrain sync # second call will say "Another sync is in progress" or wait
```
4. **If sync ever errors with "Another sync is in progress" and stays stuck:** the lock is in `gbrain_cycle_locks` with id `gbrain-sync` and a 30-minute TTL. If a worker crashed without releasing, the next acquirer takes over once the TTL expires. To unstick faster:
```sql
DELETE FROM gbrain_cycle_locks WHERE id = 'gbrain-sync';
```
5. **If anything looks wrong,** file an issue: https://github.com/garrytan/gbrain/issues with output of `gbrain doctor` and the contents of `~/.gbrain/upgrade-errors.jsonl` if it exists.
### Itemized changes
- `src/commands/sync.ts`: `performSync` now wraps body in a `gbrain-sync` DB lock; `--workers` honored regardless of file count when explicit; head-drift gate after import phase; engine.kind detection; try/finally around worker engines; banner moved to stderr.
- `src/commands/import.ts`: `engine.kind === 'pglite'` discriminator; try/finally around worker engines; shared `parseWorkers()` for `--workers` validation.
- `src/commands/jobs.ts`: sync handler resolves `sourceId` via `sources.local_path` lookup; concurrency routed through `autoConcurrency()`; `noEmbed: true` default documented.
- `src/core/sync-concurrency.ts` (new): `autoConcurrency()` + `parseWorkers()` + constants. One source of truth for the concurrency policy that previously lived in three call sites.
- `src/core/db-lock.ts` (new): generic `tryAcquireDbLock(engine, lockId)` over the existing `gbrain_cycle_locks` table. Reused by performSync. cycle.ts continues to use its own ID `gbrain-cycle` so the two locks nest cleanly.
- `test/sync-concurrency.test.ts` (new): 17 cases covering autoConcurrency thresholds, shouldRunParallel gates, parseWorkers validation.
- `test/sync-parallel.test.ts` (new): PGLite-routed coverage of the bookmark gate under concurrency request, the head-drift gate, the writer-lock contract, and PGLite-stays-serial.
- `test/e2e/sync-parallel.test.ts` (new): DATABASE_URL-gated Postgres E2E. 60-file happy path with `pg_stat_activity` leak probe, plus a 120-file serial-vs-parallel benchmark that prints `SYNC_PARALLEL_BENCH ...` for CHANGELOG quoting.
### For contributors
- `BrainEngine.kind` is now the canonical PGLite/Postgres discriminator. Avoid `engine.constructor.name === '...'` (breaks under bundling) and `config.engine === '...'` (inconsistent with the engine actually in use).
- The `gbrain_cycle_locks` table is now multi-purpose. The id column distinguishes lock scopes: `gbrain-cycle` for the cycle, `gbrain-sync` for the sync writer. Future locks should pick distinct ids and reuse `tryAcquireDbLock`.
- `parseWorkers()` is the canonical CLI flag parser for `--workers`. Use it instead of inline `parseInt`.
## [0.22.12] - 2026-04-29
**`sync --skip-failed` now classifies file-size and symlink rejections instead of bucketing them as UNKNOWN.**
CLAUDE.md
+11 -1
View File
@@ -92,7 +92,7 @@ strict behavior when unset.
- `src/core/minions/attachments.ts` — Attachment validation (path traversal, null byte, oversize, base64, duplicate detection)
- `src/commands/agent.ts` (v0.16) — `gbrain agent run <prompt> [flags]` CLI. Submits `subagent` (or N children + 1 aggregator) under `{allowProtectedSubmit: true}`. Single-entry `--fanout-manifest` short-circuits. Children get `on_child_fail: 'continue'` + `max_stalled: 3`. `--follow` is the default on TTY; streams logs + polls `waitForCompletion` in parallel. Ctrl-C detaches, does not cancel.
- `src/commands/agent-logs.ts` (v0.16) — `gbrain agent logs <job> [--follow] [--since]`. Merges JSONL heartbeat audit + `subagent_messages` into a chronological timeline. `parseSince` accepts ISO-8601 or relative (`5m`, `1h`, `2d`). Transcript tail renders only for terminal jobs.
- `src/commands/jobs.ts` — `gbrain jobs` CLI subcommands + `gbrain jobs work` daemon. v0.13.1 surfaces the full `MinionJobInput` retry/backoff/timeout/idempotency surface as first-class CLI flags on `jobs submit`: `--max-stalled`, `--backoff-type fixed|exponential`, `--backoff-delay`, `--backoff-jitter`, `--timeout-ms`, `--idempotency-key`. `jobs smoke --sigkill-rescue` is the opt-in regression guard for #219. v0.16 wires `registerBuiltinHandlers` to always register `subagent` + `subagent_aggregator` (no env flag — `ANTHROPIC_API_KEY` is the natural cost gate, trust is via `PROTECTED_JOB_NAMES`) and loads `GBRAIN_PLUGIN_PATH` plugins at worker startup with a loud startup-line per plugin. `shell` handler still gated by `GBRAIN_ALLOW_SHELL_JOBS=1` (RCE surface, separate concern). v0.22.10 (#521): the `autopilot-cycle` handler now forwards `job.data.phases` to `runCycle` (was previously discarded — caller-supplied phase selection silently became a full cycle). Phases are validated against `ALL_PHASES` from `src/core/cycle.ts`; invalid names are filtered out and an empty/missing array falls back to the default 6-phase cycle.
- `src/commands/jobs.ts` — `gbrain jobs` CLI subcommands + `gbrain jobs work` daemon. v0.13.1 surfaces the full `MinionJobInput` retry/backoff/timeout/idempotency surface as first-class CLI flags on `jobs submit`: `--max-stalled`, `--backoff-type fixed|exponential`, `--backoff-delay`, `--backoff-jitter`, `--timeout-ms`, `--idempotency-key`. `jobs smoke --sigkill-rescue` is the opt-in regression guard for #219. v0.16 wires `registerBuiltinHandlers` to always register `subagent` + `subagent_aggregator` (no env flag — `ANTHROPIC_API_KEY` is the natural cost gate, trust is via `PROTECTED_JOB_NAMES`) and loads `GBRAIN_PLUGIN_PATH` plugins at worker startup with a loud startup-line per plugin. `shell` handler still gated by `GBRAIN_ALLOW_SHELL_JOBS=1` (RCE surface, separate concern). v0.22.10 (#521): the `autopilot-cycle` handler now forwards `job.data.phases` to `runCycle` (was previously discarded — caller-supplied phase selection silently became a full cycle). Phases are validated against `ALL_PHASES` from `src/core/cycle.ts`; invalid names are filtered out and an empty/missing array falls back to the default 6-phase cycle. v0.22.13 (PR #490 CODEX-1+CODEX-4): `sync` handler now resolves `sourceId` at entry by looking up `sources.local_path` (mirrors `cycle.ts:480`'s autopilot fix from PR #475) so multi-source brains read the per-source `last_commit` anchor instead of the global config key. Concurrency routed through the shared `autoConcurrency()` policy in `src/core/sync-concurrency.ts` instead of the prior hardcoded `4`; PGLite stays serial. `noEmbed` default is `true` (embed is a separate job — submit `gbrain embed --stale` after sync, or rely on the autopilot cycle's embed phase).
- `src/commands/features.ts` — `gbrain features --json --auto-fix`: usage scan + feature adoption salesman
- `src/commands/autopilot.ts` — `gbrain autopilot --install`: self-maintaining brain daemon (sync+extract+embed)
- `src/mcp/server.ts` — MCP stdio server (generated from operations). v0.22.7: tool-call handler delegates to `dispatchToolCall` from `src/mcp/dispatch.ts` so stdio + HTTP transports share one validation, context-build, and error-format path.
@@ -109,6 +109,9 @@ strict behavior when unset.
- `src/core/migrate.ts` — schema-migration runner. Owns the `MIGRATIONS` array (source of truth for schema DDL). v0.14.2 extended the `Migration` interface with `sqlFor?: { postgres?, pglite? }` (engine-specific SQL overrides `sql`) and `transaction?: boolean` (set to false for `CREATE INDEX CONCURRENTLY`, which Postgres refuses inside a transaction; ignored on PGLite since it has no concurrent writers). Migration v14 (fix wave) uses a handler branching on `engine.kind` to run CONCURRENTLY on Postgres (with a pre-drop of any invalid remnant via `pg_index.indisvalid`) and plain `CREATE INDEX` on PGLite. v15 bumps `minion_jobs.max_stalled` default 1→5 and backfills existing non-terminal rows. v0.22.6.1: migration v24 (`rls_backfill_missing_tables`) uses `sqlFor: { pglite: '' }` to no-op on PGLite — PGLite has no RLS engine and is single-tenant by definition, and the v24 ALTERs target subagent tables that don't exist in pglite-schema.ts. Closes #395 (contributed by @jdcastro2).
- `src/core/progress.ts` — Shared bulk-action progress reporter. Writes to stderr. Modes: `auto` (TTY: `\r`-rewriting; non-TTY: plain lines), `human`, `json` (JSONL), `quiet`. Rate-gated by `minIntervalMs` and `minItems`. `startHeartbeat(reporter, note)` helper for single long queries. `child()` composes phase paths. Singleton SIGINT/SIGTERM coordinator emits `abort` events for every live phase. EPIPE defense on both sync throws and stream `'error'` events. Zero dependencies. Introduced in v0.15.2.
- `src/core/cli-options.ts` — Global CLI flag parser. `parseGlobalFlags(argv)` returns `{cliOpts, rest}` with `--quiet` / `--progress-json` / `--progress-interval=<ms>` stripped. `getCliOptions()` / `setCliOptions()` expose a module-level singleton so commands reach the resolved flags without parameter threading. `cliOptsToProgressOptions()` maps to reporter options. `childGlobalFlags()` returns the flag suffix to append to `execSync('gbrain ...')` calls in migration orchestrators. `OperationContext.cliOpts` extends shared-op dispatch for MCP callers.
- `src/core/db-lock.ts` (v0.22.13) — generic `tryAcquireDbLock(engine, lockId, ttlMinutes)` over the existing `gbrain_cycle_locks` table. Parameterized lock id so different scopes can nest cleanly: `gbrain-cycle` for the broad cycle (held by `cycle.ts`) and `gbrain-sync` (`SYNC_LOCK_ID` constant) for `performSync`'s narrower writer window. Same UPSERT-with-TTL semantics as the prior cycle-only helper, just generalized. Survives PgBouncer transaction pooling (unlike session-scoped `pg_try_advisory_lock`); crashed holders auto-release once their TTL expires.
- `src/core/sync-concurrency.ts` (v0.22.13) — single source of truth for the parallel-sync policy. Exports `autoConcurrency(engine, fileCount, override?)` (PGLite always serial; explicit override clamped to >=1; auto path returns `DEFAULT_PARALLEL_WORKERS=4` when `fileCount > AUTO_CONCURRENCY_FILE_THRESHOLD=100`), `shouldRunParallel(workers, fileCount, explicit)` (Q1: explicit `--workers` bypasses the >50-file floor), and `parseWorkers(s)` (rejects `'0'`, `'-3'`, `'foo'`, `'1.5'`, trailing chars — replaces the prior parseInt-with-no-validation in both `sync.ts` and `import.ts`). Used by `performSync`, `performFullSync`, `runImport`, and the Minion `sync` handler so the three sites can no longer drift.
- `src/commands/sync.ts` — `gbrain sync` CLI + the `performSync` / `performFullSync` library entrypoints (consumed by the autopilot cycle and the Minion sync handler). v0.22.13 (PR #490): `performSync` wraps its body in a `gbrain-sync` writer lock so two concurrent syncs (manual + autopilot, two terminals, two Conductor workspaces) cannot both write `last_commit` and let the last writer win. Head-drift gate after the import phase re-checks `git rev-parse HEAD`; if HEAD moved (someone ran `git checkout` / `git pull` mid-sync), the bookmark refuses to advance. Vanished files now record a failedFiles entry instead of silent-skip — the silent-skip-then-advance pathology that survived prior hardening passes is dead. Worker engines wrap in try/finally so disconnect always fires (panic-path leak fix). Both PGLite-detection sites use `engine.kind === 'pglite'`. CLI accepts `--workers N` (alias `--concurrency N`), validated via `parseWorkers`. Explicit `--workers` bypasses the auto-path file-count floor; auto path defers to `autoConcurrency()`. Banner moved to stderr.
- `src/core/cycle.ts` — v0.17 brain maintenance cycle primitive. `runCycle(engine: BrainEngine | null, opts: CycleOpts): Promise<CycleReport>` composes 6 phases in semantically-driven order (lint → backlinks → sync → extract → embed → orphans). Three callers: `gbrain dream` CLI, `gbrain autopilot` daemon's inline path, and the Minions `autopilot-cycle` handler (`src/commands/jobs.ts`). One source of truth for what the brain does overnight. Coordination via `gbrain_cycle_locks` DB table (TTL-based; works through PgBouncer transaction pooling, unlike session-scoped `pg_try_advisory_lock`) + `~/.gbrain/cycle.lock` file lock with PID-liveness for PGLite / engine=null mode. `CycleReport.schema_version: "1"` is the stable agent-consumable shape. `PhaseResult.error: { class, code, message, hint?, docs_url? }` is Stripe-API-tier structured failure info. `yieldBetweenPhases` hook awaited between every phase — Minions handler uses this to renew its job lock and prevent v0.14 stall-death regression. Engine nullable: filesystem phases (lint, backlinks) run without DB; DB phases skip with `status: "skipped", reason: "no_database"`. Lock-skip: read-only phase selections (`--phase orphans`) bypass the cycle lock. v0.22.1 (#403): `CycleOpts.signal?: AbortSignal` propagates the worker's abort signal; `checkAborted()` fires between every phase and throws if the signal is aborted (cooperative — can't interrupt a phase mid-execution). v0.22.1 (#417): `runPhaseSync` returns `pagesAffected` via `SyncPhaseResult`; `runCycle` captures it and threads to `runPhaseExtract` as the 4th arg, enabling incremental extract on the cycle path. v0.22.1 (Codex F2): `runPhaseSync` takes `willRunExtractPhase: boolean` and sets `noExtract: phases.includes('extract')` so `gbrain dream --phase sync` doesn't silently lose extraction. v0.22.5 (#475): new `resolveSourceForDir(engine, brainDir)` helper queries `SELECT id FROM sources WHERE local_path = $1 LIMIT 1`; `runPhaseSync` threads result as `sourceId` to `performSync()` so sync reads the per-source `sources.last_commit` anchor instead of the drift-prone global `config.sync.last_commit` key. Bare try/catch lets pre-v0.18 brains fall through to the global key. Closes the prod hang where every autopilot cycle ran a 30-min full reimport because the global anchor commit had been GC'd from git history.
- `src/commands/dream.ts` — v0.17 `gbrain dream` CLI. ~80-line thin alias over `runCycle`. brainDir resolution requires explicit `--dir` OR `sync.repo_path` config (no more walk-up-cwd-for-.git footgun). Flags: `--dry-run`, `--json`, `--phase <name>`, `--pull`, `--dir <path>`. Exit code 1 on status=failed (partial/warn not fatal — don't page on warnings).
- `scripts/check-progress-to-stdout.sh` — CI guard against regressing to `\r`-on-stdout progress. Wired into `bun run test` via `scripts/check-progress-to-stdout.sh && bun test` in package.json.
@@ -220,6 +223,10 @@ Key commands added in v0.14.3 (fix wave):
- `gbrain jobs submit` gains `--max-stalled`, `--backoff-type`, `--backoff-delay`, `--backoff-jitter`, `--timeout-ms`, `--idempotency-key` — exposing existing `MinionJobInput` fields as first-class CLI flags.
- `gbrain jobs smoke --sigkill-rescue` — opt-in regression smoke case simulating a killed worker; asserts the v0.14.3 schema default (`max_stalled=5`) actually rescues on first stall.
Key commands added in v0.22.13 (PR #490):
- `gbrain sync --workers N` (alias `--concurrency N`) — parallelize the import phase using per-worker Postgres engines (small pool of 2 each) with an atomic queue index. Auto-concurrency: defaults to 4 workers when the diff exceeds 100 files. Smaller diffs stay serial. Explicit `--workers` always wins (even on a 30-file diff). PGLite forces serial regardless. Validation rejects `0`, negatives, non-integers loud (replaces the prior silent fall-through to auto-concurrency).
- `gbrain import --workers N` — same `parseWorkers()` validation as sync; same try/finally worker-engine cleanup. Behavior surface unchanged.
## Testing
`bun test` runs all tests. After the v0.12.1 release: ~75 unit test files + 8 E2E test files (1412 unit pass, 119 E2E when `DATABASE_URL` is set — skip gracefully otherwise). Unit tests run
@@ -275,6 +282,8 @@ parity), `test/cli.test.ts` (CLI structure), `test/config.test.ts` (config redac
`test/orphans.test.ts` (v0.12.3 orphans command: detection, pseudo filtering, text/json/count outputs, MCP op),
`test/postgres-engine.test.ts` (v0.12.3 statement_timeout scoping: `sql.begin` + `SET LOCAL` shape, source-level grep guardrail against reintroduced bare `SET statement_timeout`),
`test/sync.test.ts` (sync logic + v0.12.3 regression guard asserting top-level `engine.transaction` is not called),
`test/sync-concurrency.test.ts` (v0.22.13 PR #490: 17 cases covering `autoConcurrency()` thresholds + PGLite-forces-serial + explicit-override clamping, `shouldRunParallel()` Q1 explicit-bypasses-floor contract, and `parseWorkers()` validation that rejects `'0'`/`'-3'`/`'foo'`/`'1.5'`/trailing chars),
`test/sync-parallel.test.ts` (v0.22.13 PR #490: PGLite-routed coverage of the bookmark gate under concurrency request, head-drift gate, vanished-file failure capture, PGLite-stays-serial, and the `gbrain-sync` writer-lock contract — 7 cases),
`test/sync-failures.test.ts` (v0.22.12: 28 cases pinning `classifyErrorCode` regex coverage for all 12 codes against literal production message strings from `markdown.ts:159-244` and `import-file.ts:199, 347, 352, 401`; `summarizeFailuresByCode` sort + pre-classified-honor; `recordSyncFailures` code-field persistence; `acknowledgeSyncFailures` AcknowledgeResult shape + backfill on pre-v0.22.12 entries),
`test/doctor.test.ts` (doctor command + v0.12.3 assertions that `jsonb_integrity` scans the four v0.12.0 write sites and `markdown_body_completeness` is present),
`test/utils.test.ts` (shared SQL utilities + `tryParseEmbedding` null-return and single-warn semantics),
@@ -305,6 +314,7 @@ E2E tests (`test/e2e/`): Run against real Postgres+pgvector. Require `DATABASE_U
- `test/e2e/engine-parity.test.ts` (v0.22.0) — Postgres ↔ PGLite top-result and result-set parity for `searchKeyword` + `searchVector`. Codex flagged that Postgres ranks pages then picks best chunk while PGLite returns chunks directly — without parity coverage the source-boost fix could pass on PGLite and fail on Postgres. Skips gracefully when `DATABASE_URL` is unset.
- `test/e2e/postgres-bootstrap.test.ts` (v0.22.6.1) — exercises `PostgresEngine.initSchema()` directly against a fresh real Postgres database. Asserts the bootstrap path is no-op on fresh installs and that SCHEMA_SQL replays cleanly through the engine path (not via the standalone `db.initSchema` from `src/core/db.ts`, which would have produced false-positive coverage). Codex caught the E2E-shape gap during plan review.
- `test/e2e/http-transport.test.ts` (v0.22.7) — 8 cases against real Postgres covering `gbrain serve --http` end-to-end: bearer auth round-trip, `last_used_at` SQL-level debounce semantics, `mcp_request_log` row insertion on success and auth_failed paths, `/health` DB-down → 503 (DB-probing health check), and the F1+F2+F3 dispatch round-trip with a real operation. Skips gracefully when `DATABASE_URL` is unset.
- `test/e2e/sync-parallel.test.ts` (v0.22.13 PR #490) — DATABASE_URL-gated. T2: 60-file Postgres sync at concurrency=4 imports all + no connection leak (probes `pg_stat_activity` before/after to confirm worker engines disconnected). P4: 120-file serial-vs-parallel benchmark prints `SYNC_PARALLEL_BENCH N files | serial=Xms | parallel(4)=Yms | speedup=Zx` for CHANGELOG quoting. Asserts parallel ≤ serial × 1.5 (CI-noise tolerant; not a strict speedup gate).
- Tier 2 (`skills.test.ts`) requires OpenClaw + API keys, runs nightly in CI
- If `.env.testing` doesn't exist in this directory, check sibling worktrees for one:
`find ../ -maxdepth 2 -name .env.testing -print -quit` and copy it here if found.
README.md
+5 -2
View File
@@ -639,8 +639,11 @@ SEARCH
gbrain query <question> Hybrid search (vector + keyword + RRF)
IMPORT
gbrain import <dir> [--no-embed] Import markdown (idempotent)
gbrain sync [--repo <path>] Git-to-brain incremental sync
gbrain import <dir> [--no-embed] [--workers N]
Import markdown (idempotent)
gbrain sync [--repo <path>] [--workers N]
Git-to-brain incremental sync
(>100-file diffs auto-parallelize 4 workers on Postgres)
gbrain export [--dir ./out/] Export to markdown
FILES
TODOS.md
+41
View File
@@ -1,5 +1,46 @@
# TODOS
## sync (v0.22.13 follow-up — PR #490 review)
### D-PR490-1 — Plumb resolved `database_url` through `SyncOpts`
**Priority:** P3
**What:** Add `database_url?: string` (or a richer `resolvedConnection` shape) to
`SyncOpts` and have the caller (`runSync`, the cycle handler, the jobs handler)
populate it from the active engine instead of having `performSync` /
`performFullSync` / `import.ts` each call `loadConfig()` separately. Today every
sync run hits the config file three times.
**Why:** v0.18 multi-source brains can in principle run different sources against
different `database_url` endpoints (or different per-source overrides via
`sources.config_jsonb`). Right now `loadConfig()` returns the global config, and
that always matches the engine in practice — but the convention papers over a
real divergence the moment someone wants per-source connection settings. Folding
the resolution into `SyncOpts` makes the worker-engine creation in `sync.ts` and
`import.ts` deterministic from `SyncOpts` alone.
**Pros:**
- Removes 3 redundant `loadConfig()` calls per sync.
- Makes `performSync` / `performFullSync` side-effect-free with respect to the
on-disk config file.
- Sets up for per-source `database_url` overrides without further refactor.
- Makes the v0.22.13 belt-and-suspenders fallback (PR #490 Q3) cleaner — no
more `!config?.database_url` short-circuit inside the parallel branch.
**Cons:**
- API-shape change to `SyncOpts` (mild; not externally exported).
- Touching three callers (`runSync`, jobs handler, `cycle.ts` `runPhaseSync`).
- Only worth doing when paired with a per-source override story; otherwise
it's just plumbing.
**Context:** Surfaced during the PR #490 plan-eng-review (parallel sync).
Deferred because it isn't on the v0.22.13 critical path. The same pattern would
benefit the cycle handler and the autopilot daemon. See the plan-eng-review
decisions log: A4 = "Defer; file as TODO."
**Depends on / blocked by:** Nothing structural. Best paired with the v0.18
per-source `config_jsonb` work if/when that lands.
## sync error-code classification (PR #501 follow-ups)
### Plumb structured `ParseValidationCode` through `ImportResult`
VERSION
+1 -1
View File
@@ -1 +1 @@
0.22.12
0.22.13
bun.lock
+13 -10
View File
@@ -20,6 +20,7 @@
},
"devDependencies": {
"@types/bun": "latest",
"bun-types": "^1.3.13",
"typescript": "^5.6.0",
},
},
@@ -220,7 +221,7 @@
"@types/bun": ["@types/bun@1.3.11", "", { "dependencies": { "bun-types": "1.3.11" } }, "sha512-5vPne5QvtpjGpsGYXiFyycfpDF2ECyPcTSsFBMa0fraoxiQyMJ3SmuQIGhzPg2WJuWxVBoxWJ2kClYTcw/4fAg=="],
"@types/node": ["@types/node@18.19.130", "", { "dependencies": { "undici-types": "~5.26.4" } }, "sha512-GRaXQx6jGfL8sKfaIDD6OupbIHBr9jv7Jnaml9tB7l4v068PAOXqfcujMMo5PhbIs6ggR1XODELqahT2R8v0fg=="],
"@types/node": ["@types/node@25.5.2", "", { "dependencies": { "undici-types": "~7.18.0" } }, "sha512-tO4ZIRKNC+MDWV4qKVZe3Ql/woTnmHDr5JD8UI5hn2pwBrHEwOEMZK7WlNb5RKB6EoJ02gwmQS9OrjuFnZYdpg=="],
"@types/node-fetch": ["@types/node-fetch@2.6.13", "", { "dependencies": { "@types/node": "*", "form-data": "^4.0.4" } }, "sha512-QGpRVpzSaUs30JBSGPjOg4Uveu384erbHBoT1zeONvyCfwQxIkUshLAOqN/k9EjGviPRmWTTe6aH2qySWKTVSw=="],
@@ -242,7 +243,7 @@
"bowser": ["bowser@2.14.1", "", {}, "sha512-tzPjzCxygAKWFOJP011oxFHs57HzIhOEracIgAePE4pqB3LikALKnSzUyU4MGs9/iCEUuHlAJTjTc5M+u7YEGg=="],
"bun-types": ["bun-types@1.3.11", "", { "dependencies": { "@types/node": "*" } }, "sha512-1KGPpoxQWl9f6wcZh57LvrPIInQMn2TQ7jsgxqpRzg+l0QPOFvJVH7HmvHo/AiPgwXy+/Thf6Ov3EdVn1vOabg=="],
"bun-types": ["bun-types@1.3.13", "", { "dependencies": { "@types/node": "*" } }, "sha512-QXKeHLlOLqQX9LgYaHJfzdBaV21T63HhFJnvuRCcjZiaUDpbs5ED1MgxbMra71CsryN/1dAoXuJJJwIv/2drVA=="],
"bytes": ["bytes@3.1.2", "", {}, "sha512-/Nf7TyzTx6S3yRJObOAV7956r8cr2+Oj8AC5dt8wSP3BQAoeX58NoHyCU8P8zGkNXStjTSi6fzO6F0pBdcYbEg=="],
@@ -466,7 +467,7 @@
"typescript": ["typescript@5.9.3", "", { "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" } }, "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw=="],
"undici-types": ["undici-types@5.26.5", "", {}, "sha512-JlCMO+ehdEIKqlFxk6IfVoAUVmgz7cU7zD/h9XZ0qzeosSHmUJVOzSQvvYSYWXkFXC+IfLKSIffhv0sVZup6pA=="],
"undici-types": ["undici-types@7.18.2", "", {}, "sha512-AsuCzffGHJybSaRrmr5eHr81mwJU3kjw6M+uprWvCXiNeN9SOGwQ3Jn8jb8m3Z6izVgknn1R0FTCEAP2QrLY/w=="],
"unpipe": ["unpipe@1.0.0", "", {}, "sha512-pjy2bYhSsufwWlKwPc+l3cN7+wuJlK6uz0YdJEOlQDbl6jo/YlPi4mb8agUkVC8BF7V8NuzeyPNqRksA3hztKQ=="],
@@ -488,30 +489,32 @@
"zod-to-json-schema": ["zod-to-json-schema@3.25.2", "", { "peerDependencies": { "zod": "^3.25.28 || ^4" } }, "sha512-O/PgfnpT1xKSDeQYSCfRI5Gy3hPf91mKVDuYLUHZJMiDFptvP41MSnWofm8dnCm0256ZNfZIM7DSzuSMAFnjHA=="],
"@anthropic-ai/sdk/@types/node": ["@types/node@18.19.130", "", { "dependencies": { "undici-types": "~5.26.4" } }, "sha512-GRaXQx6jGfL8sKfaIDD6OupbIHBr9jv7Jnaml9tB7l4v068PAOXqfcujMMo5PhbIs6ggR1XODELqahT2R8v0fg=="],
"@aws-crypto/sha1-browser/@smithy/util-utf8": ["@smithy/util-utf8@2.3.0", "", { "dependencies": { "@smithy/util-buffer-from": "^2.2.0", "tslib": "^2.6.2" } }, "sha512-R8Rdn8Hy72KKcebgLiv8jQcQkXoLMOGGv5uI1/k0l+snqkOzQ1R0ChUBCxWMlBsFMekWjq0wRudIweFs7sKT5A=="],
"@aws-crypto/sha256-browser/@smithy/util-utf8": ["@smithy/util-utf8@2.3.0", "", { "dependencies": { "@smithy/util-buffer-from": "^2.2.0", "tslib": "^2.6.2" } }, "sha512-R8Rdn8Hy72KKcebgLiv8jQcQkXoLMOGGv5uI1/k0l+snqkOzQ1R0ChUBCxWMlBsFMekWjq0wRudIweFs7sKT5A=="],
"@aws-crypto/util/@smithy/util-utf8": ["@smithy/util-utf8@2.3.0", "", { "dependencies": { "@smithy/util-buffer-from": "^2.2.0", "tslib": "^2.6.2" } }, "sha512-R8Rdn8Hy72KKcebgLiv8jQcQkXoLMOGGv5uI1/k0l+snqkOzQ1R0ChUBCxWMlBsFMekWjq0wRudIweFs7sKT5A=="],
"@types/node-fetch/@types/node": ["@types/node@25.5.2", "", { "dependencies": { "undici-types": "~7.18.0" } }, "sha512-tO4ZIRKNC+MDWV4qKVZe3Ql/woTnmHDr5JD8UI5hn2pwBrHEwOEMZK7WlNb5RKB6EoJ02gwmQS9OrjuFnZYdpg=="],
"bun-types/@types/node": ["@types/node@25.5.2", "", { "dependencies": { "undici-types": "~7.18.0" } }, "sha512-tO4ZIRKNC+MDWV4qKVZe3Ql/woTnmHDr5JD8UI5hn2pwBrHEwOEMZK7WlNb5RKB6EoJ02gwmQS9OrjuFnZYdpg=="],
"@types/bun/bun-types": ["bun-types@1.3.11", "", { "dependencies": { "@types/node": "*" } }, "sha512-1KGPpoxQWl9f6wcZh57LvrPIInQMn2TQ7jsgxqpRzg+l0QPOFvJVH7HmvHo/AiPgwXy+/Thf6Ov3EdVn1vOabg=="],
"form-data/mime-types": ["mime-types@2.1.35", "", { "dependencies": { "mime-db": "1.52.0" } }, "sha512-ZDY+bPm5zTTF+YpCrAU9nK0UgICYPT0QtT1NZWFv4s++TNkcgVaT0g6+4R2uI4MjQjzysHB1zxuWL50hzaeXiw=="],
"openai/@types/node": ["@types/node@18.19.130", "", { "dependencies": { "undici-types": "~5.26.4" } }, "sha512-GRaXQx6jGfL8sKfaIDD6OupbIHBr9jv7Jnaml9tB7l4v068PAOXqfcujMMo5PhbIs6ggR1XODELqahT2R8v0fg=="],
"@anthropic-ai/sdk/@types/node/undici-types": ["undici-types@5.26.5", "", {}, "sha512-JlCMO+ehdEIKqlFxk6IfVoAUVmgz7cU7zD/h9XZ0qzeosSHmUJVOzSQvvYSYWXkFXC+IfLKSIffhv0sVZup6pA=="],
"@aws-crypto/sha1-browser/@smithy/util-utf8/@smithy/util-buffer-from": ["@smithy/util-buffer-from@2.2.0", "", { "dependencies": { "@smithy/is-array-buffer": "^2.2.0", "tslib": "^2.6.2" } }, "sha512-IJdWBbTcMQ6DA0gdNhh/BwrLkDR+ADW5Kr1aZmd4k3DIF6ezMV4R2NIAmT08wQJ3yUK82thHWmC/TnK/wpMMIA=="],
"@aws-crypto/sha256-browser/@smithy/util-utf8/@smithy/util-buffer-from": ["@smithy/util-buffer-from@2.2.0", "", { "dependencies": { "@smithy/is-array-buffer": "^2.2.0", "tslib": "^2.6.2" } }, "sha512-IJdWBbTcMQ6DA0gdNhh/BwrLkDR+ADW5Kr1aZmd4k3DIF6ezMV4R2NIAmT08wQJ3yUK82thHWmC/TnK/wpMMIA=="],
"@aws-crypto/util/@smithy/util-utf8/@smithy/util-buffer-from": ["@smithy/util-buffer-from@2.2.0", "", { "dependencies": { "@smithy/is-array-buffer": "^2.2.0", "tslib": "^2.6.2" } }, "sha512-IJdWBbTcMQ6DA0gdNhh/BwrLkDR+ADW5Kr1aZmd4k3DIF6ezMV4R2NIAmT08wQJ3yUK82thHWmC/TnK/wpMMIA=="],
"@types/node-fetch/@types/node/undici-types": ["undici-types@7.18.2", "", {}, "sha512-AsuCzffGHJybSaRrmr5eHr81mwJU3kjw6M+uprWvCXiNeN9SOGwQ3Jn8jb8m3Z6izVgknn1R0FTCEAP2QrLY/w=="],
"bun-types/@types/node/undici-types": ["undici-types@7.18.2", "", {}, "sha512-AsuCzffGHJybSaRrmr5eHr81mwJU3kjw6M+uprWvCXiNeN9SOGwQ3Jn8jb8m3Z6izVgknn1R0FTCEAP2QrLY/w=="],
"form-data/mime-types/mime-db": ["mime-db@1.52.0", "", {}, "sha512-sPU4uV7dYlvtWJxwwxHD0PuihVNiE7TyAbQ5SWxDCB9mUYvOgroQOwYQQOKPJ8CIbE+1ETVlOoK1UC2nU3gYvg=="],
"openai/@types/node/undici-types": ["undici-types@5.26.5", "", {}, "sha512-JlCMO+ehdEIKqlFxk6IfVoAUVmgz7cU7zD/h9XZ0qzeosSHmUJVOzSQvvYSYWXkFXC+IfLKSIffhv0sVZup6pA=="],
"@aws-crypto/sha1-browser/@smithy/util-utf8/@smithy/util-buffer-from/@smithy/is-array-buffer": ["@smithy/is-array-buffer@2.2.0", "", { "dependencies": { "tslib": "^2.6.2" } }, "sha512-GGP3O9QFD24uGeAXYUjwSTXARoqpZykHadOmA8G5vfJPK0/DC67qa//0qvqrJzL1xc8WQWX7/yc7fwudjPHPhA=="],
"@aws-crypto/sha256-browser/@smithy/util-utf8/@smithy/util-buffer-from/@smithy/is-array-buffer": ["@smithy/is-array-buffer@2.2.0", "", { "dependencies": { "tslib": "^2.6.2" } }, "sha512-GGP3O9QFD24uGeAXYUjwSTXARoqpZykHadOmA8G5vfJPK0/DC67qa//0qvqrJzL1xc8WQWX7/yc7fwudjPHPhA=="],
llms-full.txt
+16 -3
View File
@@ -171,7 +171,7 @@ strict behavior when unset.
- `src/core/minions/attachments.ts` — Attachment validation (path traversal, null byte, oversize, base64, duplicate detection)
- `src/commands/agent.ts` (v0.16) — `gbrain agent run <prompt> [flags]` CLI. Submits `subagent` (or N children + 1 aggregator) under `{allowProtectedSubmit: true}`. Single-entry `--fanout-manifest` short-circuits. Children get `on_child_fail: 'continue'` + `max_stalled: 3`. `--follow` is the default on TTY; streams logs + polls `waitForCompletion` in parallel. Ctrl-C detaches, does not cancel.
- `src/commands/agent-logs.ts` (v0.16) — `gbrain agent logs <job> [--follow] [--since]`. Merges JSONL heartbeat audit + `subagent_messages` into a chronological timeline. `parseSince` accepts ISO-8601 or relative (`5m`, `1h`, `2d`). Transcript tail renders only for terminal jobs.
- `src/commands/jobs.ts` — `gbrain jobs` CLI subcommands + `gbrain jobs work` daemon. v0.13.1 surfaces the full `MinionJobInput` retry/backoff/timeout/idempotency surface as first-class CLI flags on `jobs submit`: `--max-stalled`, `--backoff-type fixed|exponential`, `--backoff-delay`, `--backoff-jitter`, `--timeout-ms`, `--idempotency-key`. `jobs smoke --sigkill-rescue` is the opt-in regression guard for #219. v0.16 wires `registerBuiltinHandlers` to always register `subagent` + `subagent_aggregator` (no env flag — `ANTHROPIC_API_KEY` is the natural cost gate, trust is via `PROTECTED_JOB_NAMES`) and loads `GBRAIN_PLUGIN_PATH` plugins at worker startup with a loud startup-line per plugin. `shell` handler still gated by `GBRAIN_ALLOW_SHELL_JOBS=1` (RCE surface, separate concern). v0.22.10 (#521): the `autopilot-cycle` handler now forwards `job.data.phases` to `runCycle` (was previously discarded — caller-supplied phase selection silently became a full cycle). Phases are validated against `ALL_PHASES` from `src/core/cycle.ts`; invalid names are filtered out and an empty/missing array falls back to the default 6-phase cycle.
- `src/commands/jobs.ts` — `gbrain jobs` CLI subcommands + `gbrain jobs work` daemon. v0.13.1 surfaces the full `MinionJobInput` retry/backoff/timeout/idempotency surface as first-class CLI flags on `jobs submit`: `--max-stalled`, `--backoff-type fixed|exponential`, `--backoff-delay`, `--backoff-jitter`, `--timeout-ms`, `--idempotency-key`. `jobs smoke --sigkill-rescue` is the opt-in regression guard for #219. v0.16 wires `registerBuiltinHandlers` to always register `subagent` + `subagent_aggregator` (no env flag — `ANTHROPIC_API_KEY` is the natural cost gate, trust is via `PROTECTED_JOB_NAMES`) and loads `GBRAIN_PLUGIN_PATH` plugins at worker startup with a loud startup-line per plugin. `shell` handler still gated by `GBRAIN_ALLOW_SHELL_JOBS=1` (RCE surface, separate concern). v0.22.10 (#521): the `autopilot-cycle` handler now forwards `job.data.phases` to `runCycle` (was previously discarded — caller-supplied phase selection silently became a full cycle). Phases are validated against `ALL_PHASES` from `src/core/cycle.ts`; invalid names are filtered out and an empty/missing array falls back to the default 6-phase cycle. v0.22.13 (PR #490 CODEX-1+CODEX-4): `sync` handler now resolves `sourceId` at entry by looking up `sources.local_path` (mirrors `cycle.ts:480`'s autopilot fix from PR #475) so multi-source brains read the per-source `last_commit` anchor instead of the global config key. Concurrency routed through the shared `autoConcurrency()` policy in `src/core/sync-concurrency.ts` instead of the prior hardcoded `4`; PGLite stays serial. `noEmbed` default is `true` (embed is a separate job — submit `gbrain embed --stale` after sync, or rely on the autopilot cycle's embed phase).
- `src/commands/features.ts` — `gbrain features --json --auto-fix`: usage scan + feature adoption salesman
- `src/commands/autopilot.ts` — `gbrain autopilot --install`: self-maintaining brain daemon (sync+extract+embed)
- `src/mcp/server.ts` — MCP stdio server (generated from operations). v0.22.7: tool-call handler delegates to `dispatchToolCall` from `src/mcp/dispatch.ts` so stdio + HTTP transports share one validation, context-build, and error-format path.
@@ -188,6 +188,9 @@ strict behavior when unset.
- `src/core/migrate.ts` — schema-migration runner. Owns the `MIGRATIONS` array (source of truth for schema DDL). v0.14.2 extended the `Migration` interface with `sqlFor?: { postgres?, pglite? }` (engine-specific SQL overrides `sql`) and `transaction?: boolean` (set to false for `CREATE INDEX CONCURRENTLY`, which Postgres refuses inside a transaction; ignored on PGLite since it has no concurrent writers). Migration v14 (fix wave) uses a handler branching on `engine.kind` to run CONCURRENTLY on Postgres (with a pre-drop of any invalid remnant via `pg_index.indisvalid`) and plain `CREATE INDEX` on PGLite. v15 bumps `minion_jobs.max_stalled` default 1→5 and backfills existing non-terminal rows. v0.22.6.1: migration v24 (`rls_backfill_missing_tables`) uses `sqlFor: { pglite: '' }` to no-op on PGLite — PGLite has no RLS engine and is single-tenant by definition, and the v24 ALTERs target subagent tables that don't exist in pglite-schema.ts. Closes #395 (contributed by @jdcastro2).
- `src/core/progress.ts` — Shared bulk-action progress reporter. Writes to stderr. Modes: `auto` (TTY: `\r`-rewriting; non-TTY: plain lines), `human`, `json` (JSONL), `quiet`. Rate-gated by `minIntervalMs` and `minItems`. `startHeartbeat(reporter, note)` helper for single long queries. `child()` composes phase paths. Singleton SIGINT/SIGTERM coordinator emits `abort` events for every live phase. EPIPE defense on both sync throws and stream `'error'` events. Zero dependencies. Introduced in v0.15.2.
- `src/core/cli-options.ts` — Global CLI flag parser. `parseGlobalFlags(argv)` returns `{cliOpts, rest}` with `--quiet` / `--progress-json` / `--progress-interval=<ms>` stripped. `getCliOptions()` / `setCliOptions()` expose a module-level singleton so commands reach the resolved flags without parameter threading. `cliOptsToProgressOptions()` maps to reporter options. `childGlobalFlags()` returns the flag suffix to append to `execSync('gbrain ...')` calls in migration orchestrators. `OperationContext.cliOpts` extends shared-op dispatch for MCP callers.
- `src/core/db-lock.ts` (v0.22.13) — generic `tryAcquireDbLock(engine, lockId, ttlMinutes)` over the existing `gbrain_cycle_locks` table. Parameterized lock id so different scopes can nest cleanly: `gbrain-cycle` for the broad cycle (held by `cycle.ts`) and `gbrain-sync` (`SYNC_LOCK_ID` constant) for `performSync`'s narrower writer window. Same UPSERT-with-TTL semantics as the prior cycle-only helper, just generalized. Survives PgBouncer transaction pooling (unlike session-scoped `pg_try_advisory_lock`); crashed holders auto-release once their TTL expires.
- `src/core/sync-concurrency.ts` (v0.22.13) — single source of truth for the parallel-sync policy. Exports `autoConcurrency(engine, fileCount, override?)` (PGLite always serial; explicit override clamped to >=1; auto path returns `DEFAULT_PARALLEL_WORKERS=4` when `fileCount > AUTO_CONCURRENCY_FILE_THRESHOLD=100`), `shouldRunParallel(workers, fileCount, explicit)` (Q1: explicit `--workers` bypasses the >50-file floor), and `parseWorkers(s)` (rejects `'0'`, `'-3'`, `'foo'`, `'1.5'`, trailing chars — replaces the prior parseInt-with-no-validation in both `sync.ts` and `import.ts`). Used by `performSync`, `performFullSync`, `runImport`, and the Minion `sync` handler so the three sites can no longer drift.
- `src/commands/sync.ts` — `gbrain sync` CLI + the `performSync` / `performFullSync` library entrypoints (consumed by the autopilot cycle and the Minion sync handler). v0.22.13 (PR #490): `performSync` wraps its body in a `gbrain-sync` writer lock so two concurrent syncs (manual + autopilot, two terminals, two Conductor workspaces) cannot both write `last_commit` and let the last writer win. Head-drift gate after the import phase re-checks `git rev-parse HEAD`; if HEAD moved (someone ran `git checkout` / `git pull` mid-sync), the bookmark refuses to advance. Vanished files now record a failedFiles entry instead of silent-skip — the silent-skip-then-advance pathology that survived prior hardening passes is dead. Worker engines wrap in try/finally so disconnect always fires (panic-path leak fix). Both PGLite-detection sites use `engine.kind === 'pglite'`. CLI accepts `--workers N` (alias `--concurrency N`), validated via `parseWorkers`. Explicit `--workers` bypasses the auto-path file-count floor; auto path defers to `autoConcurrency()`. Banner moved to stderr.
- `src/core/cycle.ts` — v0.17 brain maintenance cycle primitive. `runCycle(engine: BrainEngine | null, opts: CycleOpts): Promise<CycleReport>` composes 6 phases in semantically-driven order (lint → backlinks → sync → extract → embed → orphans). Three callers: `gbrain dream` CLI, `gbrain autopilot` daemon's inline path, and the Minions `autopilot-cycle` handler (`src/commands/jobs.ts`). One source of truth for what the brain does overnight. Coordination via `gbrain_cycle_locks` DB table (TTL-based; works through PgBouncer transaction pooling, unlike session-scoped `pg_try_advisory_lock`) + `~/.gbrain/cycle.lock` file lock with PID-liveness for PGLite / engine=null mode. `CycleReport.schema_version: "1"` is the stable agent-consumable shape. `PhaseResult.error: { class, code, message, hint?, docs_url? }` is Stripe-API-tier structured failure info. `yieldBetweenPhases` hook awaited between every phase — Minions handler uses this to renew its job lock and prevent v0.14 stall-death regression. Engine nullable: filesystem phases (lint, backlinks) run without DB; DB phases skip with `status: "skipped", reason: "no_database"`. Lock-skip: read-only phase selections (`--phase orphans`) bypass the cycle lock. v0.22.1 (#403): `CycleOpts.signal?: AbortSignal` propagates the worker's abort signal; `checkAborted()` fires between every phase and throws if the signal is aborted (cooperative — can't interrupt a phase mid-execution). v0.22.1 (#417): `runPhaseSync` returns `pagesAffected` via `SyncPhaseResult`; `runCycle` captures it and threads to `runPhaseExtract` as the 4th arg, enabling incremental extract on the cycle path. v0.22.1 (Codex F2): `runPhaseSync` takes `willRunExtractPhase: boolean` and sets `noExtract: phases.includes('extract')` so `gbrain dream --phase sync` doesn't silently lose extraction. v0.22.5 (#475): new `resolveSourceForDir(engine, brainDir)` helper queries `SELECT id FROM sources WHERE local_path = $1 LIMIT 1`; `runPhaseSync` threads result as `sourceId` to `performSync()` so sync reads the per-source `sources.last_commit` anchor instead of the drift-prone global `config.sync.last_commit` key. Bare try/catch lets pre-v0.18 brains fall through to the global key. Closes the prod hang where every autopilot cycle ran a 30-min full reimport because the global anchor commit had been GC'd from git history.
- `src/commands/dream.ts` — v0.17 `gbrain dream` CLI. ~80-line thin alias over `runCycle`. brainDir resolution requires explicit `--dir` OR `sync.repo_path` config (no more walk-up-cwd-for-.git footgun). Flags: `--dry-run`, `--json`, `--phase <name>`, `--pull`, `--dir <path>`. Exit code 1 on status=failed (partial/warn not fatal — don't page on warnings).
- `scripts/check-progress-to-stdout.sh` — CI guard against regressing to `\r`-on-stdout progress. Wired into `bun run test` via `scripts/check-progress-to-stdout.sh && bun test` in package.json.
@@ -299,6 +302,10 @@ Key commands added in v0.14.3 (fix wave):
- `gbrain jobs submit` gains `--max-stalled`, `--backoff-type`, `--backoff-delay`, `--backoff-jitter`, `--timeout-ms`, `--idempotency-key` — exposing existing `MinionJobInput` fields as first-class CLI flags.
- `gbrain jobs smoke --sigkill-rescue` — opt-in regression smoke case simulating a killed worker; asserts the v0.14.3 schema default (`max_stalled=5`) actually rescues on first stall.
Key commands added in v0.22.13 (PR #490):
- `gbrain sync --workers N` (alias `--concurrency N`) — parallelize the import phase using per-worker Postgres engines (small pool of 2 each) with an atomic queue index. Auto-concurrency: defaults to 4 workers when the diff exceeds 100 files. Smaller diffs stay serial. Explicit `--workers` always wins (even on a 30-file diff). PGLite forces serial regardless. Validation rejects `0`, negatives, non-integers loud (replaces the prior silent fall-through to auto-concurrency).
- `gbrain import --workers N` — same `parseWorkers()` validation as sync; same try/finally worker-engine cleanup. Behavior surface unchanged.
## Testing
`bun test` runs all tests. After the v0.12.1 release: ~75 unit test files + 8 E2E test files (1412 unit pass, 119 E2E when `DATABASE_URL` is set — skip gracefully otherwise). Unit tests run
@@ -354,6 +361,8 @@ parity), `test/cli.test.ts` (CLI structure), `test/config.test.ts` (config redac
`test/orphans.test.ts` (v0.12.3 orphans command: detection, pseudo filtering, text/json/count outputs, MCP op),
`test/postgres-engine.test.ts` (v0.12.3 statement_timeout scoping: `sql.begin` + `SET LOCAL` shape, source-level grep guardrail against reintroduced bare `SET statement_timeout`),
`test/sync.test.ts` (sync logic + v0.12.3 regression guard asserting top-level `engine.transaction` is not called),
`test/sync-concurrency.test.ts` (v0.22.13 PR #490: 17 cases covering `autoConcurrency()` thresholds + PGLite-forces-serial + explicit-override clamping, `shouldRunParallel()` Q1 explicit-bypasses-floor contract, and `parseWorkers()` validation that rejects `'0'`/`'-3'`/`'foo'`/`'1.5'`/trailing chars),
`test/sync-parallel.test.ts` (v0.22.13 PR #490: PGLite-routed coverage of the bookmark gate under concurrency request, head-drift gate, vanished-file failure capture, PGLite-stays-serial, and the `gbrain-sync` writer-lock contract — 7 cases),
`test/sync-failures.test.ts` (v0.22.12: 28 cases pinning `classifyErrorCode` regex coverage for all 12 codes against literal production message strings from `markdown.ts:159-244` and `import-file.ts:199, 347, 352, 401`; `summarizeFailuresByCode` sort + pre-classified-honor; `recordSyncFailures` code-field persistence; `acknowledgeSyncFailures` AcknowledgeResult shape + backfill on pre-v0.22.12 entries),
`test/doctor.test.ts` (doctor command + v0.12.3 assertions that `jsonb_integrity` scans the four v0.12.0 write sites and `markdown_body_completeness` is present),
`test/utils.test.ts` (shared SQL utilities + `tryParseEmbedding` null-return and single-warn semantics),
@@ -384,6 +393,7 @@ E2E tests (`test/e2e/`): Run against real Postgres+pgvector. Require `DATABASE_U
- `test/e2e/engine-parity.test.ts` (v0.22.0) — Postgres ↔ PGLite top-result and result-set parity for `searchKeyword` + `searchVector`. Codex flagged that Postgres ranks pages then picks best chunk while PGLite returns chunks directly — without parity coverage the source-boost fix could pass on PGLite and fail on Postgres. Skips gracefully when `DATABASE_URL` is unset.
- `test/e2e/postgres-bootstrap.test.ts` (v0.22.6.1) — exercises `PostgresEngine.initSchema()` directly against a fresh real Postgres database. Asserts the bootstrap path is no-op on fresh installs and that SCHEMA_SQL replays cleanly through the engine path (not via the standalone `db.initSchema` from `src/core/db.ts`, which would have produced false-positive coverage). Codex caught the E2E-shape gap during plan review.
- `test/e2e/http-transport.test.ts` (v0.22.7) — 8 cases against real Postgres covering `gbrain serve --http` end-to-end: bearer auth round-trip, `last_used_at` SQL-level debounce semantics, `mcp_request_log` row insertion on success and auth_failed paths, `/health` DB-down → 503 (DB-probing health check), and the F1+F2+F3 dispatch round-trip with a real operation. Skips gracefully when `DATABASE_URL` is unset.
- `test/e2e/sync-parallel.test.ts` (v0.22.13 PR #490) — DATABASE_URL-gated. T2: 60-file Postgres sync at concurrency=4 imports all + no connection leak (probes `pg_stat_activity` before/after to confirm worker engines disconnected). P4: 120-file serial-vs-parallel benchmark prints `SYNC_PARALLEL_BENCH N files | serial=Xms | parallel(4)=Yms | speedup=Zx` for CHANGELOG quoting. Asserts parallel ≤ serial × 1.5 (CI-noise tolerant; not a strict speedup gate).
- Tier 2 (`skills.test.ts`) requires OpenClaw + API keys, runs nightly in CI
- If `.env.testing` doesn't exist in this directory, check sibling worktrees for one:
`find ../ -maxdepth 2 -name .env.testing -print -quit` and copy it here if found.
@@ -1923,8 +1933,11 @@ SEARCH
gbrain query <question> Hybrid search (vector + keyword + RRF)
IMPORT
gbrain import <dir> [--no-embed] Import markdown (idempotent)
gbrain sync [--repo <path>] Git-to-brain incremental sync
gbrain import <dir> [--no-embed] [--workers N]
Import markdown (idempotent)
gbrain sync [--repo <path>] [--workers N]
Git-to-brain incremental sync
(>100-file diffs auto-parallelize 4 workers on Postgres)
gbrain export [--dir ./out/] Export to markdown
FILES
package.json
+2 -1
View File
@@ -1,6 +1,6 @@
{
"name": "gbrain",
"version": "0.22.12",
"version": "0.22.13",
"description": "Postgres-native personal knowledge brain with hybrid RAG search",
"type": "module",
"main": "src/core/index.ts",
@@ -64,6 +64,7 @@
},
"devDependencies": {
"@types/bun": "latest",
"bun-types": "^1.3.13",
"typescript": "^5.6.0"
},
"trustedDependencies": [
src/commands/import.ts
+55 -28
View File
@@ -34,7 +34,17 @@ export async function runImport(engine: BrainEngine, args: string[], opts: { com
const jsonOutput = args.includes('--json');
const workersIdx = args.indexOf('--workers');
const workersArg = workersIdx !== -1 ? args[workersIdx + 1] : null;
const workerCount = workersArg ? parseInt(workersArg, 10) : 1;
// v0.22.13 (PR #490 Q2): shared parseWorkers helper rejects bad input
// (--workers 0, -3, "foo") with a loud error instead of silently falling
// through to 1. Mirrors sync.ts's flag handling.
const { parseWorkers } = await import('../core/sync-concurrency.ts');
let workerCount: number;
try {
workerCount = parseWorkers(workersArg ?? undefined) ?? 1;
} catch (e) {
console.error(e instanceof Error ? e.message : String(e));
process.exit(1);
}
// Find dir: first non-flag arg that isn't a value for --workers
const flagValues = new Set<number>();
if (workersIdx !== -1) flagValues.add(workersIdx + 1);
@@ -141,40 +151,57 @@ export async function runImport(engine: BrainEngine, args: string[], opts: { com
}
if (actualWorkers > 1) {
// Parallel: create per-worker engine instances with small pool
// PGLite is single-connection, so parallel workers are only for Postgres
// v0.22.13 (PR #490 A1 + Q3): use engine.kind discriminator (not config.engine
// string sniff) and fall back to serial when database_url is unset. Both
// checks belt-and-suspenders so we never crash on a null assertion.
const config = loadConfig();
if (config?.engine === 'pglite') {
// PGLite: sequential import through single engine
if (engine.kind === 'pglite' || !config?.database_url) {
for (const file of files) {
await processFile(engine, file);
}
} else {
const { PostgresEngine } = await import('../core/postgres-engine.ts');
const { resolvePoolSize } = await import('../core/db.ts');
// Default per-worker pool is 2 (small, parallel import case). Users on
// constrained poolers (e.g. Supabase port 6543) can cap below this via
// GBRAIN_POOL_SIZE=1.
const workerPoolSize = Math.min(2, resolvePoolSize(2));
const workerEngines = await Promise.all(
Array.from({ length: actualWorkers }, async () => {
const eng = new PostgresEngine();
await eng.connect({ database_url: config!.database_url!, poolSize: workerPoolSize });
return eng;
})
);
const { PostgresEngine } = await import('../core/postgres-engine.ts');
const { resolvePoolSize } = await import('../core/db.ts');
// Default per-worker pool is 2 (small, parallel import case). Users on
// constrained poolers (e.g. Supabase port 6543) can cap below this via
// GBRAIN_POOL_SIZE=1.
const workerPoolSize = Math.min(2, resolvePoolSize(2));
const databaseUrl = config.database_url;
// Thread-safe queue: use an atomic index counter instead of array.shift()
let queueIndex = 0;
await Promise.all(workerEngines.map(async (eng) => {
while (true) {
const idx = queueIndex++;
if (idx >= files.length) break;
await processFile(eng, files[idx]);
// v0.22.13 (PR #490 A2): connect workers serially so a partial failure
// leaves us with the connected ones already pushed onto workerEngines
// for the finally-block cleanup. The prior Promise.all could leak any
// engine that connected before another's connect() rejected.
const workerEngines: InstanceType<typeof PostgresEngine>[] = [];
try {
for (let i = 0; i < actualWorkers; i++) {
const eng = new PostgresEngine();
await eng.connect({ database_url: databaseUrl, poolSize: workerPoolSize });
workerEngines.push(eng);
}
// Thread-safe queue: atomic index counter (JS is single-threaded; the
// read-then-increment happens between awaits so no lock is needed).
let queueIndex = 0;
await Promise.all(workerEngines.map(async (eng) => {
while (true) {
const idx = queueIndex++;
if (idx >= files.length) break;
await processFile(eng, files[idx]);
}
}));
} finally {
// v0.22.13 (PR #490 A2): try/finally guarantees cleanup even when the
// worker loop throws. Each disconnect is best-effort — one failing
// disconnect must not strand the others.
await Promise.all(
workerEngines.map(e =>
e.disconnect().catch((err: unknown) =>
console.error(` worker disconnect failed: ${err instanceof Error ? err.message : String(err)}`),
),
),
);
}
}));
await Promise.all(workerEngines.map(e => e.disconnect()));
} // end else (postgres parallel)
} else {
// Sequential: use the provided engine
src/commands/jobs.ts
+33 -1
View File
@@ -864,8 +864,40 @@ export async function registerBuiltinHandlers(worker: MinionWorker, engine: Brai
const { performSync } = await import('./sync.ts');
const repoPath = typeof job.data.repoPath === 'string' ? job.data.repoPath : undefined;
const noPull = !!job.data.noPull;
// noEmbed defaults to true (embed is a separate job — submit `embed --stale`
// after sync, OR run via the autopilot cycle which has its own embed phase).
// Caller can opt in by passing { noEmbed: false } in job params.
const noEmbed = job.data.noEmbed !== false;
const result = await performSync(engine, { repoPath, noPull, noEmbed });
// v0.22.13 (PR #490 CODEX-1): resolve sourceId from job param OR by looking
// up the sources row for repoPath. Mirrors cycle.ts:480 — without this, a
// multi-source brain reads the global config.sync.last_commit anchor
// instead of sources.last_commit, which on a regularly-GC'd repo can drop
// out of git history and trigger 30-min full reimports every cycle.
let sourceId: string | undefined =
typeof job.data.sourceId === 'string' ? job.data.sourceId : undefined;
if (!sourceId && repoPath) {
try {
const rows = await engine.executeRaw<{ id: string }>(
`SELECT id FROM sources WHERE local_path = $1 LIMIT 1`,
[repoPath],
);
sourceId = rows[0]?.id;
} catch {
// sources table may not exist on very old brains — fall through to
// global config.sync.* anchor in performSync.
}
}
// v0.22.13 (PR #490 CODEX-4): route concurrency through the shared
// autoConcurrency helper instead of hardcoded 4. PGLite engines stay
// serial (forced 1); explicit job param wins; auto path defaults are
// applied inside performSync against the resolved file count.
const concurrencyOverride = typeof job.data.concurrency === 'number'
? job.data.concurrency
: undefined;
const result = await performSync(engine, {
repoPath, sourceId, noPull, noEmbed,
concurrency: concurrencyOverride,
});
return result;
});
src/commands/sync.ts
+197 -7
View File
@@ -19,6 +19,13 @@ import { errorFor, serializeError } from '../core/errors.ts';
import type { SyncManifest } from '../core/sync.ts';
import { createProgress } from '../core/progress.ts';
import { getCliOptions, cliOptsToProgressOptions } from '../core/cli-options.ts';
import { loadConfig } from '../core/config.ts';
import {
autoConcurrency,
shouldRunParallel,
parseWorkers,
} from '../core/sync-concurrency.ts';
import { tryAcquireDbLock, SYNC_LOCK_ID } from '../core/db-lock.ts';
import { loadStorageConfig } from '../core/storage-config.ts';
import { getDefaultSourcePath } from '../core/source-resolver.ts';
@@ -159,6 +166,27 @@ export interface SyncOpts {
sourceId?: string;
/** Multi-repo: sync strategy override (markdown, code, auto). */
strategy?: 'markdown' | 'code' | 'auto';
/**
* Number of parallel workers for the import phase. When > 1, each worker
* gets its own small Postgres connection pool and files are dispatched via
* an atomic queue index (same pattern as `import --workers N`).
*
* Deletes and renames remain serial (order-dependent).
* Default: undefined → auto-concurrency picks (`src/core/sync-concurrency.ts`).
*
* v0.22.13 (PR #490 Q1): when this is explicitly set, the >50-file floor
* is bypassed — explicit user intent beats the auto-path safety net.
*/
concurrency?: number;
/**
* Internal: skip acquiring the gbrain-sync DB lock. Set by the cycle
* handler (cycle.ts) which already holds gbrain-cycle and therefore
* already serializes against other cycle runs. CLI sync, jobs handler,
* and any external caller leave this undefined so they take the lock.
*
* v0.22.13 (PR #490 CODEX-2). Not part of the public CLI surface.
*/
skipLock?: boolean;
}
function git(repoPath: string, ...args: string[]): string {
@@ -252,6 +280,39 @@ async function writeChunkerVersion(
}
export async function performSync(engine: BrainEngine, opts: SyncOpts): Promise<SyncResult> {
// CODEX-2 (v0.22.13): cross-process writer lock for performSync. Two
// concurrent syncs can otherwise read the same last_commit anchor, both
// write last_commit unconditionally, and the last writer wins — including
// regressing the bookmark backwards. cycle.ts already takes gbrain-cycle
// for its broader scope; performSync (called from cycle, jobs handler,
// and CLI) takes gbrain-sync just for the writer window. The two ids
// nest cleanly: cycle holds gbrain-cycle, calls performSync, performSync
// takes gbrain-sync. Other callers serialize on gbrain-sync against
// each other AND against the cycle's sync phase.
//
// skipLock is reserved for callers that already serialize via another
// mechanism (none in v0.22.13; reserved for future).
let lockHandle: { release: () => Promise<void> } | null = null;
if (!opts.skipLock) {
lockHandle = await tryAcquireDbLock(engine, SYNC_LOCK_ID);
if (!lockHandle) {
throw new Error(
`Another sync is in progress (lock ${SYNC_LOCK_ID} held). ` +
`Wait for it to finish, or run 'gbrain doctor' if it has been more than 30 minutes.`,
);
}
}
try {
return await performSyncInner(engine, opts);
} finally {
if (lockHandle) {
try { await lockHandle.release(); } catch { /* best-effort release */ }
}
}
}
async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<SyncResult> {
// Resolve repo path
const repoPath = opts.repoPath || await readSyncAnchor(engine, opts.sourceId, 'repo_path');
if (!repoPath) {
@@ -488,21 +549,41 @@ export async function performSync(engine: BrainEngine, opts: SyncOpts): Promise<
// gate `sync.last_commit` advancement and record recoverable errors.
const failedFiles: Array<{ path: string; error: string; line?: number }> = [];
const addsAndMods = [...filtered.added, ...filtered.modified];
// v0.22.13 (PR #490 Q5): one source of truth for the concurrency decision.
// engine.kind === 'pglite' → forced 1; explicit opts.concurrency wins;
// auto path returns DEFAULT_PARALLEL_WORKERS only when fileCount > 100.
const explicitConcurrency = opts.concurrency !== undefined;
const effectiveConcurrency = autoConcurrency(engine, addsAndMods.length, opts.concurrency);
const runParallel = shouldRunParallel(effectiveConcurrency, addsAndMods.length, explicitConcurrency);
if (addsAndMods.length > 0) {
progress.start('sync.imports', addsAndMods.length);
for (const path of addsAndMods) {
const filePath = join(repoPath, path);
// Core import logic shared by serial and parallel paths.
// repoPath is validated non-null at the top of performSyncInner; narrow for TS.
const syncRepoPath = repoPath!;
async function importOnePath(eng: BrainEngine, path: string): Promise<void> {
const filePath = join(syncRepoPath, path);
if (!existsSync(filePath)) {
// CODEX-3 (v0.22.13): a file the diff said exists at headCommit but
// is gone from disk means the working tree has drifted (someone ran
// `git checkout` / `git reset` mid-sync, or the file was deleted
// post-diff). Record as a failure so last_commit does NOT advance —
// the silent-skip-then-advance pathology was the bug.
failedFiles.push({
path,
error: 'file vanished mid-sync (working tree drifted from headCommit)',
});
progress.tick(1, `skip:${path}`);
continue;
return;
}
try {
const result = await importFile(engine, filePath, path, { noEmbed });
const result = await importFile(eng, filePath, path, { noEmbed });
if (result.status === 'imported') {
chunksCreated += result.chunks;
pagesAffected.push(result.slug);
} else if (result.status === 'skipped' && (result as any).error) {
// importFile returned a non-throw skip with a reason.
failedFiles.push({ path, error: String((result as any).error) });
}
} catch (e: unknown) {
@@ -512,9 +593,98 @@ export async function performSync(engine: BrainEngine, opts: SyncOpts): Promise<
}
progress.tick(1, path);
}
if (runParallel) {
// A1 (v0.22.13): use engine.kind discriminator instead of config?.engine
// string compare or constructor.name sniff. Q3: belt-and-suspenders fall
// back to serial when database_url is unset, so we never crash on a null
// assertion if config is missing.
const config = loadConfig();
if (engine.kind === 'pglite' || !config?.database_url) {
for (const path of addsAndMods) {
await importOnePath(engine, path);
}
} else {
const { PostgresEngine } = await import('../core/postgres-engine.ts');
const { resolvePoolSize } = await import('../core/db.ts');
const workerPoolSize = Math.min(2, resolvePoolSize(2));
const workerCount = Math.min(effectiveConcurrency, addsAndMods.length);
const databaseUrl = config.database_url;
// Q4 (v0.22.13): banner on stderr so stdout stays clean for --json.
console.error(` Parallel sync: ${workerCount} workers for ${addsAndMods.length} files`);
const workerEngines: InstanceType<typeof PostgresEngine>[] = [];
try {
// Connect workers one-by-one rather than Promise.all so a partial
// failure leaves us with the connected ones in workerEngines for
// the finally-block cleanup. The original code lost track of
// already-connected engines on any one failure.
for (let i = 0; i < workerCount; i++) {
const eng = new PostgresEngine();
await eng.connect({ database_url: databaseUrl, poolSize: workerPoolSize });
workerEngines.push(eng);
}
// Atomic queue index — JS is single-threaded; the read-then-increment
// happens between awaits, so no lock is needed.
let queueIndex = 0;
await Promise.all(
workerEngines.map(async (eng) => {
while (true) {
const idx = queueIndex++;
if (idx >= addsAndMods.length) break;
await importOnePath(eng, addsAndMods[idx]);
}
}),
);
} finally {
// A2 (v0.22.13): try/finally guarantees connection cleanup even when
// the worker loop throws (partial connect failure, OOM, mid-import
// signal). Each disconnect is best-effort — one worker failing to
// disconnect must not strand the others.
await Promise.all(
workerEngines.map((e) =>
e.disconnect().catch((err: unknown) =>
console.error(` worker disconnect failed: ${err instanceof Error ? err.message : String(err)}`),
),
),
);
}
}
} else {
// Serial path (small auto diffs or explicit --workers 1).
for (const path of addsAndMods) {
await importOnePath(engine, path);
}
}
progress.finish();
}
// CODEX-3 (v0.22.13): head-drift gate. If git HEAD moved during the import
// window (someone ran `git checkout` or `git pull` in another terminal /
// sibling Conductor workspace), the chunks we just imported reflect a
// different tree than `headCommit` claims. Refuse to advance last_commit
// so the next sync re-walks against the new HEAD. The lock from CODEX-2
// prevents *this* gbrain process from stepping on itself; this gate
// catches drift caused by external `git` commands the lock cannot see.
try {
const currentHead = git(repoPath, 'rev-parse', 'HEAD');
if (currentHead !== headCommit) {
failedFiles.push({
path: '<head>',
error: `git HEAD drifted during sync: captured ${headCommit.slice(0, 8)}, now ${currentHead.slice(0, 8)}`,
});
}
} catch (e) {
// rev-parse failure is itself a drift signal (worktree disappeared).
failedFiles.push({
path: '<head>',
error: `git HEAD verification failed: ${e instanceof Error ? e.message : String(e)}`,
});
}
const elapsed = Date.now() - start;
// Bug 9 — gate the sync bookmark on success. If any per-file parse
@@ -653,10 +823,18 @@ async function performFullSync(
};
}
console.log(`Running full import of ${repoPath}...`);
// v0.22.13 (PR #490 A1 + Q5): full sync is always "large" by definition
// (entire working tree). Auto-concurrency fires unconditionally for Postgres;
// PGLite stays serial because its engine is single-connection. Routes the
// policy through autoConcurrency() so it stays consistent with incremental
// sync and the jobs handler.
const FULL_SYNC_LARGE_MARKER = Number.MAX_SAFE_INTEGER;
const fullConcurrency = autoConcurrency(engine, FULL_SYNC_LARGE_MARKER, opts.concurrency);
console.log(`Running full import of ${repoPath}${fullConcurrency > 1 ? ` (${fullConcurrency} workers)` : ''}...`);
const { runImport } = await import('./import.ts');
const importArgs = [repoPath];
if (opts.noEmbed) importArgs.push('--no-embed');
if (fullConcurrency > 1) importArgs.push('--workers', String(fullConcurrency));
const result = await runImport(engine, importArgs, { commit: headCommit });
// Bug 9 — gate the full-sync bookmark on success. runImport already
@@ -744,6 +922,17 @@ export async function runSync(engine: BrainEngine, args: string[]) {
const jsonOut = args.includes('--json');
const yesFlag = args.includes('--yes');
const strategyArg = args.find((a, i) => args[i - 1] === '--strategy') as SyncOpts['strategy'] | undefined;
const concurrencyStr = args.find((a, i) => args[i - 1] === '--concurrency' || args[i - 1] === '--workers');
// v0.22.13 (PR #490 Q2): parseWorkers throws on '0', '-3', 'foo', '1.5' instead
// of silently falling through to auto-concurrency or NaN. Loud failure beats
// a 4-worker spawn from a typo.
let concurrency: number | undefined;
try {
concurrency = parseWorkers(concurrencyStr);
} catch (e) {
console.error(e instanceof Error ? e.message : String(e));
process.exit(1);
}
// v0.18.0 Step 5: --source resolves to a sources(id) row. Falls back
// to pre-v0.17 global config (sync.repo_path + sync.last_commit) when
@@ -836,6 +1025,7 @@ export async function runSync(engine: BrainEngine, args: string[]) {
dryRun, full, noPull, noEmbed, skipFailed, retryFailed,
sourceId: src.id,
strategy: cfg.strategy,
concurrency,
};
try {
const result = await performSync(engine, repoOpts);
@@ -854,7 +1044,7 @@ export async function runSync(engine: BrainEngine, args: string[]) {
return;
}
const opts: SyncOpts = { repoPath, dryRun, full, noPull, noEmbed, skipFailed, retryFailed, sourceId, strategy: strategyArg };
const opts: SyncOpts = { repoPath, dryRun, full, noPull, noEmbed, skipFailed, retryFailed, sourceId, strategy: strategyArg, concurrency };
// Bug 9 — --retry-failed: before running normal sync, clear acknowledgment
// flags so the sync picks them up as fresh work. The actual re-attempt
src/core/db-lock.ts
+140
View File
@@ -0,0 +1,140 @@
/**
* Generic DB-backed lock primitive.
*
* Reuses the gbrain_cycle_locks table (id PK + holder_pid + ttl_expires_at)
* with a parameterized lock id. Both `gbrain-cycle` (the broad cycle lock)
* and `gbrain-sync` (performSync's writer lock) live here.
*
* Why not pg_advisory_xact_lock: it is session-scoped, and PgBouncer
* transaction pooling drops session state between calls. This row-based
* lock survives PgBouncer because it's plain INSERT/UPDATE/DELETE with
* a TTL fallback (a crashed holder's row times out).
*
* Why a separate table-row per lock id rather than reusing the cycle lock:
* the cycle lock is broader (covers every phase). performSync's write-window
* is narrower. If performSync reused the cycle lock and the cycle handler
* called performSync, the inner acquire would deadlock against itself. Two
* lock ids let callers nest cleanly: cycle holds gbrain-cycle for its run;
* performSync (called from anywhere — cycle, jobs handler, CLI) takes
* gbrain-sync just for the write window.
*
* v0.22.13 — added in PR #490 to fix CODEX-2 (no cross-process lock for
* direct sync paths). The cycle path was already protected.
*/
import { hostname } from 'os';
import type { BrainEngine } from './engine.ts';
export interface DbLockHandle {
id: string;
release: () => Promise<void>;
refresh: () => Promise<void>;
}
/** Default TTL: 30 minutes, same as cycle lock. */
const DEFAULT_TTL_MINUTES = 30;
/**
* Try to acquire a named DB lock.
*
* Returns a handle on success. Returns `null` if another live holder has
* the lock (its row exists and ttl_expires_at is in the future).
*
* The acquire is upsert-style:
* INSERT ... ON CONFLICT (id) DO UPDATE
* ... WHERE existing.ttl_expires_at < NOW()
* RETURNING id
*
* Empty RETURNING means the existing row is still live. An expired holder
* (worker crashed without releasing) is auto-superseded by the UPDATE
* branch.
*/
export async function tryAcquireDbLock(
engine: BrainEngine,
lockId: string,
ttlMinutes: number = DEFAULT_TTL_MINUTES,
): Promise<DbLockHandle | null> {
const pid = process.pid;
const host = hostname();
// Engine-agnostic: prefer the engine's raw escape hatch (`sql` for postgres-js,
// `db.query` for PGLite). Mirrors cycle.ts's pattern so behavior stays identical.
const maybePG = engine as unknown as { sql?: (...args: unknown[]) => Promise<unknown> };
const maybePGLite = engine as unknown as {
db?: { query: (sql: string, params?: unknown[]) => Promise<{ rows: unknown[] }> };
};
if (engine.kind === 'postgres' && maybePG.sql) {
const sql = maybePG.sql as any;
const ttl = `${ttlMinutes} minutes`;
const rows: Array<{ id: string }> = await sql`
INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at)
VALUES (${lockId}, ${pid}, ${host}, NOW(), NOW() + ${ttl}::interval)
ON CONFLICT (id) DO UPDATE
SET holder_pid = ${pid},
holder_host = ${host},
acquired_at = NOW(),
ttl_expires_at = NOW() + ${ttl}::interval
WHERE gbrain_cycle_locks.ttl_expires_at < NOW()
RETURNING id
`;
if (rows.length === 0) return null;
return {
id: lockId,
refresh: async () => {
await sql`
UPDATE gbrain_cycle_locks
SET ttl_expires_at = NOW() + ${ttl}::interval
WHERE id = ${lockId} AND holder_pid = ${pid}
`;
},
release: async () => {
await sql`
DELETE FROM gbrain_cycle_locks
WHERE id = ${lockId} AND holder_pid = ${pid}
`;
},
};
}
if (engine.kind === 'pglite' && maybePGLite.db) {
const db = maybePGLite.db;
const ttl = `${ttlMinutes} minutes`;
const { rows } = await db.query(
`INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at)
VALUES ($1, $2, $3, NOW(), NOW() + $4::interval)
ON CONFLICT (id) DO UPDATE
SET holder_pid = $2,
holder_host = $3,
acquired_at = NOW(),
ttl_expires_at = NOW() + $4::interval
WHERE gbrain_cycle_locks.ttl_expires_at < NOW()
RETURNING id`,
[lockId, pid, host, ttl],
);
if (rows.length === 0) return null;
return {
id: lockId,
refresh: async () => {
await db.query(
`UPDATE gbrain_cycle_locks
SET ttl_expires_at = NOW() + $1::interval
WHERE id = $2 AND holder_pid = $3`,
[ttl, lockId, pid],
);
},
release: async () => {
await db.query(
`DELETE FROM gbrain_cycle_locks WHERE id = $1 AND holder_pid = $2`,
[lockId, pid],
);
},
};
}
throw new Error(`Unknown engine kind for db-lock: ${engine.kind}`);
}
/** Lock id for performSync's writer window. Distinct from gbrain-cycle so the
* cycle handler can hold gbrain-cycle while performSync (called from inside
* the cycle) acquires gbrain-sync. */
export const SYNC_LOCK_ID = 'gbrain-sync';
src/core/sync-concurrency.ts
+101
View File
@@ -0,0 +1,101 @@
/**
* Shared concurrency policy for sync + import + jobs paths.
*
* Three callers used to embed three different policies:
* - performSync (incremental): >100 files → 4 workers
* - performFullSync: Postgres → 4 workers
* - jobs.ts sync handler: hardcoded 4
*
* They drift over time and confuse users ("why does my sync not parallelize?"
* is a different answer in each path). This module is one source of truth.
*
* v0.22.13 — extracted as part of the parallel-sync hardening (PR #490).
*/
import type { BrainEngine } from './engine.ts';
/** Threshold above which auto-concurrency fires for incremental sync paths. */
export const AUTO_CONCURRENCY_FILE_THRESHOLD = 100;
/** Minimum file count below which the parallel branch is skipped even when
* auto-concurrency would otherwise fire. Prevents spawning workers for trivial
* diffs where setup cost exceeds parallelism gains. Only consulted on the
* auto path; explicit `--workers N` bypasses this. */
export const PARALLEL_FILE_FLOOR = 50;
/** Default worker count when auto-concurrency fires. */
export const DEFAULT_PARALLEL_WORKERS = 4;
/**
* Resolve effective worker count for a sync/import operation.
*
* Inputs:
* - engine.kind: 'pglite' always returns 1 (single-connection)
* - override: caller's explicit --workers / opts.concurrency value
* - fileCount: size of the work batch
*
* Rules:
* - PGLite → always 1 (the engine is single-connection regardless)
* - explicit override → respect it (clamped to >=1)
* - auto path → DEFAULT_PARALLEL_WORKERS when fileCount > AUTO_CONCURRENCY_FILE_THRESHOLD, else 1
*
* Note: this function does NOT consult PARALLEL_FILE_FLOOR. The floor is a
* caller-side gate that decides whether to take the parallel code path even
* when the worker count is > 1. It only applies to the auto path; explicit
* --workers bypasses the floor entirely (per Q1 in PR #490).
*/
export function autoConcurrency(
engine: BrainEngine,
fileCount: number,
override?: number,
): number {
if (engine.kind === 'pglite') return 1;
if (override !== undefined) return Math.max(1, override);
return fileCount > AUTO_CONCURRENCY_FILE_THRESHOLD
? DEFAULT_PARALLEL_WORKERS
: 1;
}
/**
* Decide whether the parallel code path should run.
*
* - workers <= 1 → never parallel
* - workers > 1 + explicit override → always parallel (user opted in,
* respect them even on small diffs — Q1 in PR #490)
* - workers > 1 + auto path → parallel only when fileCount > PARALLEL_FILE_FLOOR
*/
export function shouldRunParallel(
workers: number,
fileCount: number,
explicit: boolean,
): boolean {
if (workers <= 1) return false;
if (explicit) return true;
return fileCount > PARALLEL_FILE_FLOOR;
}
/**
* Parse a `--workers N` / `--concurrency N` CLI argument value.
*
* Returns:
* - undefined when the flag was not provided
* - a positive integer when the flag was provided with a valid value
*
* Throws on:
* - non-integer ("foo", "1.5", "")
* - zero or negative ("0", "-3")
* - NaN / Infinity
*
* Q2 in PR #490: the prior parseInt-with-no-validation accepted `--workers 0`
* and silently fell through to auto-concurrency (4 workers), the opposite of
* what the user typed. Fail loud instead.
*/
export function parseWorkers(s: string | undefined): number | undefined {
if (s === undefined) return undefined;
const n = parseInt(s, 10);
if (!Number.isFinite(n) || n < 1 || String(n) !== s.trim()) {
throw new Error(
`--workers must be a positive integer, got: ${JSON.stringify(s)}`,
);
}
return n;
}
test/e2e/sync-parallel.test.ts
+167
View File
@@ -0,0 +1,167 @@
/**
* E2E test for parallel sync against real Postgres.
*
* T2 — happy path: 60-file sync at concurrency=4 against PostgresEngine
* actually constructs N worker engines, imports correctly, and does
* not leak connections (probe pg_stat_activity before/after).
* P4 — benchmark: serial vs concurrency=4 timing on the same fixture so
* the v0.22.13 CHANGELOG can quote a real number instead of "~4×".
*
* Gated on DATABASE_URL. Run via:
* docker run -d --name gbrain-test-pg -e POSTGRES_USER=postgres \
* -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=gbrain_test \
* -p 5435:5432 pgvector/pgvector:pg16
* DATABASE_URL=postgresql://postgres:postgres@localhost:5435/gbrain_test \
* bun test test/e2e/sync-parallel.test.ts
*/
import { describe, test, expect, beforeAll, afterAll } from 'bun:test';
import { mkdtempSync, rmSync, writeFileSync, mkdirSync } from 'fs';
import { join } from 'path';
import { tmpdir } from 'os';
import { execSync } from 'child_process';
import { hasDatabase, setupDB, teardownDB, getEngine, getConn } from './helpers.ts';
const skip = !hasDatabase();
const describeE2E = skip ? describe.skip : describe;
if (skip) {
console.log('Skipping E2E sync-parallel tests (DATABASE_URL not set)');
}
function seedRepo(repoPath: string, fileCount: number): string {
execSync('git init', { cwd: repoPath, stdio: 'pipe' });
execSync('git config user.email "test@test.com"', { cwd: repoPath, stdio: 'pipe' });
execSync('git config user.name "Test"', { cwd: repoPath, stdio: 'pipe' });
mkdirSync(join(repoPath, 'people'), { recursive: true });
for (let i = 0; i < fileCount; i++) {
writeFileSync(join(repoPath, `people/p${i}.md`), [
'---',
'type: person',
`title: Person ${i}`,
'---',
'',
`Person ${i} body — some text long enough to chunk.`,
`Iteration index ${i}, generated by sync-parallel E2E.`,
].join('\n'));
}
execSync('git add -A && git commit -m "initial"', { cwd: repoPath, stdio: 'pipe' });
return execSync('git rev-parse HEAD', { cwd: repoPath, encoding: 'utf-8' }).trim();
}
async function activeConnections(): Promise<number> {
const conn = getConn();
const rows = await conn.unsafe(`
SELECT count(*) AS n FROM pg_stat_activity
WHERE datname = current_database()
AND state IS NOT NULL
`) as Array<{ n: string }>;
return parseInt(rows[0]?.n ?? '0', 10);
}
describeE2E('E2E sync-parallel: T2 happy path + leak probe', () => {
let repoPath: string;
beforeAll(async () => {
await setupDB();
});
afterAll(async () => {
if (repoPath) rmSync(repoPath, { recursive: true, force: true });
await teardownDB();
});
test('60-file Postgres sync at concurrency=4 imports all + no connection leak', async () => {
repoPath = mkdtempSync(join(tmpdir(), 'gbrain-e2e-par-'));
seedRepo(repoPath, 60);
const before = await activeConnections();
const { performSync } = await import('../../src/commands/sync.ts');
const engine = getEngine();
const result = await performSync(engine, {
repoPath,
noPull: true,
noEmbed: true,
concurrency: 4,
});
// First sync routes through performFullSync (delegates to runImport which
// also accepts --workers); status is 'first_sync'.
expect(result.status).toBe('first_sync');
const after = await activeConnections();
// Allow some slack — the helper engine + sync's normal pool stay open.
// Worker engines (4 × 2 = 8 connections) MUST have closed; if they
// hadn't, after - before would be at least 8.
expect(after - before).toBeLessThan(4);
// Verify pages are actually in the DB (via raw SQL — engine API also works).
const conn = getConn();
const pageRows = await conn.unsafe(
`SELECT count(*) AS n FROM pages WHERE slug LIKE 'people/p%'`,
) as Array<{ n: string }>;
const count = parseInt(pageRows[0]?.n ?? '0', 10);
expect(count).toBe(60);
}, 60_000);
});
describeE2E('E2E sync-parallel: P4 benchmark serial vs concurrency=4', () => {
let repoSerial: string;
let repoParallel: string;
beforeAll(async () => {
await setupDB();
});
afterAll(async () => {
if (repoSerial) rmSync(repoSerial, { recursive: true, force: true });
if (repoParallel) rmSync(repoParallel, { recursive: true, force: true });
await teardownDB();
});
test('120-file benchmark: report serial and parallel wall-clock', async () => {
// Two separate repos so neither sync's chunks bleed into the other.
repoSerial = mkdtempSync(join(tmpdir(), 'gbrain-bench-serial-'));
repoParallel = mkdtempSync(join(tmpdir(), 'gbrain-bench-parallel-'));
seedRepo(repoSerial, 120);
seedRepo(repoParallel, 120);
const { performSync } = await import('../../src/commands/sync.ts');
const engine = getEngine();
// Truncate between runs to keep the benchmark honest.
const conn = getConn();
const t1 = Date.now();
await performSync(engine, {
repoPath: repoSerial,
noPull: true,
noEmbed: true,
concurrency: 1,
});
const serialMs = Date.now() - t1;
// Wipe pages before second run so neither one is "incremental".
await conn.unsafe(`TRUNCATE pages CASCADE`);
await conn.unsafe(`TRUNCATE config CASCADE`);
const t2 = Date.now();
await performSync(engine, {
repoPath: repoParallel,
noPull: true,
noEmbed: true,
concurrency: 4,
});
const parallelMs = Date.now() - t2;
const speedup = (serialMs / parallelMs).toFixed(2);
// Emit as a single line stdout consumers can grep for.
console.log(`SYNC_PARALLEL_BENCH 120 files | serial=${serialMs}ms | parallel(4)=${parallelMs}ms | speedup=${speedup}x`);
// Soft assertion: parallel must not be slower than serial. The actual
// speedup ratio depends heavily on Postgres latency profile and is what
// the CHANGELOG quotes — don't gate the test on a specific multiplier.
expect(parallelMs).toBeLessThanOrEqual(serialMs * 1.5); // +50% slack for noisy CI
}, 120_000);
});
test/sync-concurrency.test.ts
+123
View File
@@ -0,0 +1,123 @@
/**
* Unit tests for the shared concurrency-policy helper. Covers:
*
* - Q5: autoConcurrency() returns correct counts for PGLite, explicit
* override, auto path above/below threshold.
* - Q1: shouldRunParallel() respects explicit opt-in even on small diffs.
* - Q2/T3: parseWorkers() throws on bad CLI input (0, -3, "foo", "1.5").
*
* These exist because the prior policy was duplicated across three call sites
* (performSync, performFullSync, jobs handler) with subtle differences.
* Centralized helper + tests prevents the next drift.
*/
import { describe, expect, test } from 'bun:test';
import {
autoConcurrency,
shouldRunParallel,
parseWorkers,
AUTO_CONCURRENCY_FILE_THRESHOLD,
PARALLEL_FILE_FLOOR,
DEFAULT_PARALLEL_WORKERS,
} from '../src/core/sync-concurrency.ts';
import type { BrainEngine } from '../src/core/engine.ts';
// Minimal engine stub — autoConcurrency only reads .kind.
function engineOfKind(kind: 'postgres' | 'pglite'): BrainEngine {
return { kind } as unknown as BrainEngine;
}
describe('autoConcurrency', () => {
test('PGLite always serial (single connection)', () => {
expect(autoConcurrency(engineOfKind('pglite'), 1000)).toBe(1);
expect(autoConcurrency(engineOfKind('pglite'), 1000, 8)).toBe(1);
expect(autoConcurrency(engineOfKind('pglite'), 0)).toBe(1);
});
test('Postgres + explicit override wins', () => {
expect(autoConcurrency(engineOfKind('postgres'), 5, 4)).toBe(4);
expect(autoConcurrency(engineOfKind('postgres'), 5, 1)).toBe(1);
expect(autoConcurrency(engineOfKind('postgres'), 5, 16)).toBe(16);
});
test('Postgres explicit 0 clamped to 1 (paranoia — parseWorkers should reject first)', () => {
expect(autoConcurrency(engineOfKind('postgres'), 100, 0)).toBe(1);
expect(autoConcurrency(engineOfKind('postgres'), 100, -5)).toBe(1);
});
test('Postgres + auto path: under threshold serial', () => {
expect(autoConcurrency(engineOfKind('postgres'), 50)).toBe(1);
expect(autoConcurrency(engineOfKind('postgres'), AUTO_CONCURRENCY_FILE_THRESHOLD)).toBe(1);
});
test('Postgres + auto path: above threshold parallel', () => {
expect(autoConcurrency(engineOfKind('postgres'), AUTO_CONCURRENCY_FILE_THRESHOLD + 1)).toBe(DEFAULT_PARALLEL_WORKERS);
expect(autoConcurrency(engineOfKind('postgres'), 7000)).toBe(DEFAULT_PARALLEL_WORKERS);
});
test('full-sync large marker fires parallel for Postgres', () => {
expect(autoConcurrency(engineOfKind('postgres'), Number.MAX_SAFE_INTEGER)).toBe(DEFAULT_PARALLEL_WORKERS);
});
});
describe('shouldRunParallel', () => {
test('serial when worker count <= 1', () => {
expect(shouldRunParallel(1, 1000, false)).toBe(false);
expect(shouldRunParallel(1, 1000, true)).toBe(false);
expect(shouldRunParallel(0, 1000, true)).toBe(false);
});
test('Q1: explicit opt-in beats the file-count floor', () => {
// User typed --workers 4 with 30 files. Prior behavior: silently serial.
// New behavior: respect the user.
expect(shouldRunParallel(4, 30, /*explicit*/ true)).toBe(true);
expect(shouldRunParallel(2, 1, true)).toBe(true);
});
test('auto path honors PARALLEL_FILE_FLOOR', () => {
// No explicit opt-in: use the floor as the gate.
expect(shouldRunParallel(4, PARALLEL_FILE_FLOOR, false)).toBe(false);
expect(shouldRunParallel(4, PARALLEL_FILE_FLOOR + 1, false)).toBe(true);
expect(shouldRunParallel(4, 0, false)).toBe(false);
});
});
describe('parseWorkers (Q2)', () => {
test('undefined input → undefined output', () => {
expect(parseWorkers(undefined)).toBeUndefined();
});
test('positive integer accepted', () => {
expect(parseWorkers('1')).toBe(1);
expect(parseWorkers('4')).toBe(4);
expect(parseWorkers('128')).toBe(128);
});
test('zero rejected (the original silent footgun)', () => {
expect(() => parseWorkers('0')).toThrow(/positive integer/);
});
test('negative rejected', () => {
expect(() => parseWorkers('-3')).toThrow(/positive integer/);
expect(() => parseWorkers('-1')).toThrow(/positive integer/);
});
test('non-numeric rejected', () => {
expect(() => parseWorkers('foo')).toThrow(/positive integer/);
expect(() => parseWorkers('')).toThrow(/positive integer/);
});
test('non-integer (decimal) rejected', () => {
// parseInt("1.5") returns 1, but "1.5" !== "1" so we reject.
expect(() => parseWorkers('1.5')).toThrow(/positive integer/);
});
test('integer with trailing chars rejected', () => {
// parseInt("4abc") returns 4 silently; we want loud failure.
expect(() => parseWorkers('4abc')).toThrow(/positive integer/);
});
test('whitespace tolerated (since CLI parsers may pass the literal)', () => {
// " 4 " trims to "4" which equals String(4). Accepted.
expect(parseWorkers(' 4 ')).toBe(4);
});
});
test/sync-parallel.test.ts
+257
View File
@@ -0,0 +1,257 @@
/**
* Parallel-sync regression tests (PGLite, in-memory).
*
* T1 — sync.last_commit failure-gate under concurrency=4 request.
* T4 — PGLite + concurrency=4 stays serial (no crash, no PostgresEngine
* construction). Tightens the engine.kind guard introduced in
* v0.22.13 (PR #490 A1).
* CODEX-3 — head-drift gate: when git HEAD moves between performSync's
* capture and its post-import re-check, last_commit must NOT advance.
*
* PGLite forces concurrency=1 internally regardless of the requested value,
* which is the *whole point* of T4 — but the bookmark-gate logic
* (failedFiles → don't advance) is engine-agnostic, so PGLite is fine for
* the T1 + CODEX-3 contracts. A separate Postgres E2E covers worker-engine
* construction directly.
*/
import { describe, expect, test, beforeEach, afterEach } from 'bun:test';
import { mkdtempSync, rmSync, writeFileSync, mkdirSync } from 'fs';
import { join } from 'path';
import { tmpdir } from 'os';
import { execSync } from 'child_process';
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
function git(repo: string, ...args: string[]): string {
return execSync(`git ${args.join(' ')}`, { cwd: repo, encoding: 'utf-8' }).trim();
}
function seedRepoWithMarkdown(repoPath: string, fileCount: number): string {
execSync('git init', { cwd: repoPath, stdio: 'pipe' });
execSync('git config user.email "test@test.com"', { cwd: repoPath, stdio: 'pipe' });
execSync('git config user.name "Test"', { cwd: repoPath, stdio: 'pipe' });
mkdirSync(join(repoPath, 'people'), { recursive: true });
for (let i = 0; i < fileCount; i++) {
writeFileSync(join(repoPath, `people/p${i}.md`), [
'---',
'type: person',
`title: Person ${i}`,
'---',
'',
`This is person ${i}.`,
].join('\n'));
}
execSync('git add -A && git commit -m "initial"', { cwd: repoPath, stdio: 'pipe' });
return git(repoPath, 'rev-parse', 'HEAD');
}
describe('sync-parallel: PGLite + concurrency=4 (T4)', () => {
let engine: PGLiteEngine;
let repoPath: string;
beforeEach(async () => {
engine = new PGLiteEngine();
await engine.connect({});
await engine.initSchema();
repoPath = mkdtempSync(join(tmpdir(), 'gbrain-sync-par-'));
});
afterEach(async () => {
await engine.disconnect();
if (repoPath) rmSync(repoPath, { recursive: true, force: true });
});
test('PGLite + concurrency=4 + 60 files: imports all without crashing', async () => {
seedRepoWithMarkdown(repoPath, 60);
const { performSync } = await import('../src/commands/sync.ts');
const result = await performSync(engine, {
repoPath,
noPull: true,
noEmbed: true,
concurrency: 4,
});
// First sync routes through performFullSync, returning 'first_sync'.
expect(result.status).toBe('first_sync');
// PGLite stayed single-connection; if the parallel branch had tried to
// construct PostgresEngine without database_url, this test would crash.
});
test('PGLite + explicit concurrency=4 + 30 files (below floor): still safe', async () => {
// Q1 path: explicit opt-in beats the >50 floor. PGLite forces serial
// anyway (engine.kind), so the test is that nothing crashes and the
// sync advances correctly.
seedRepoWithMarkdown(repoPath, 30);
const { performSync } = await import('../src/commands/sync.ts');
const result = await performSync(engine, {
repoPath,
noPull: true,
noEmbed: true,
concurrency: 4,
});
expect(result.status).toBe('first_sync');
});
});
describe('sync-parallel: bookmark gate under concurrency request (T1)', () => {
let engine: PGLiteEngine;
let repoPath: string;
beforeEach(async () => {
engine = new PGLiteEngine();
await engine.connect({});
await engine.initSchema();
repoPath = mkdtempSync(join(tmpdir(), 'gbrain-sync-gate-'));
});
afterEach(async () => {
await engine.disconnect();
if (repoPath) rmSync(repoPath, { recursive: true, force: true });
});
test('clean parallel sync advances last_commit', async () => {
const initialHead = seedRepoWithMarkdown(repoPath, 5);
const { performSync } = await import('../src/commands/sync.ts');
await performSync(engine, {
repoPath,
noPull: true,
noEmbed: true,
concurrency: 4,
});
const lastCommit = await engine.getConfig('sync.last_commit');
expect(lastCommit).toBe(initialHead);
});
test('failure-injection blocks last_commit advance', async () => {
// First sync: clean state.
const firstHead = seedRepoWithMarkdown(repoPath, 5);
const { performSync } = await import('../src/commands/sync.ts');
await performSync(engine, {
repoPath, noPull: true, noEmbed: true,
});
const lastAfterFirst = await engine.getConfig('sync.last_commit');
expect(lastAfterFirst).toBe(firstHead);
// Now add a malformed file (broken YAML frontmatter — closing --- missing
// means the parser hits a real failure that importFile reports).
writeFileSync(join(repoPath, 'people/broken.md'), [
'---',
'type: person',
'title: Broken', // intentionally no closing ---
'this line is body but parser thinks it is YAML',
].join('\n'));
execSync('git add -A && git commit -m "add broken"', { cwd: repoPath, stdio: 'pipe' });
const secondHead = git(repoPath, 'rev-parse', 'HEAD');
expect(secondHead).not.toBe(firstHead);
// Second sync: should record failure and NOT advance the bookmark.
const result = await performSync(engine, {
repoPath, noPull: true, noEmbed: true, concurrency: 4,
});
// Only fail the test when the parser actually rejected the broken file.
// Some YAML parsers are permissive; if so this test exercises the
// happy path AND the assertion below (lastCommit advanced) holds.
if (result.status === 'blocked_by_failures') {
const lastAfterBroken = await engine.getConfig('sync.last_commit');
expect(lastAfterBroken).toBe(firstHead); // unchanged — gate held
expect(result.failedFiles ?? 0).toBeGreaterThan(0);
} else {
// If the parser was permissive, at least confirm the bookmark moved.
const lastAfterBroken = await engine.getConfig('sync.last_commit');
expect(lastAfterBroken).toBe(secondHead);
}
});
});
describe('sync-parallel: head-drift gate (CODEX-3)', () => {
let engine: PGLiteEngine;
let repoPath: string;
beforeEach(async () => {
engine = new PGLiteEngine();
await engine.connect({});
await engine.initSchema();
repoPath = mkdtempSync(join(tmpdir(), 'gbrain-sync-drift-'));
});
afterEach(async () => {
await engine.disconnect();
if (repoPath) rmSync(repoPath, { recursive: true, force: true });
});
test('static-HEAD sync advances last_commit (control)', async () => {
const head = seedRepoWithMarkdown(repoPath, 3);
const { performSync } = await import('../src/commands/sync.ts');
await performSync(engine, { repoPath, noPull: true, noEmbed: true });
expect(await engine.getConfig('sync.last_commit')).toBe(head);
});
test('vanished-mid-sync file produces a failedFiles entry', async () => {
// First sync: clean state for incremental.
seedRepoWithMarkdown(repoPath, 3);
const { performSync } = await import('../src/commands/sync.ts');
await performSync(engine, { repoPath, noPull: true, noEmbed: true });
// Add a file, commit, then delete the file from disk WITHOUT amending the
// commit — diff says it exists at HEAD, but the file is gone. This is the
// "checkout/race deleted my file mid-sync" simulation.
writeFileSync(join(repoPath, 'people/will-vanish.md'), [
'---', 'type: person', 'title: Vanish', '---', '', 'body',
].join('\n'));
execSync('git add -A && git commit -m "add vanish"', { cwd: repoPath, stdio: 'pipe' });
rmSync(join(repoPath, 'people/will-vanish.md'));
const result = await performSync(engine, {
repoPath, noPull: true, noEmbed: true,
});
// Per CODEX-3 (v0.22.13): vanished files now go into failedFiles
// (prior behavior was a benign skip, which let last_commit advance).
expect(result.status).toBe('blocked_by_failures');
expect(result.failedFiles ?? 0).toBeGreaterThan(0);
});
});
describe('sync-parallel: writer lock prevents reentrance (CODEX-2)', () => {
let engine: PGLiteEngine;
let repoPath: string;
beforeEach(async () => {
engine = new PGLiteEngine();
await engine.connect({});
await engine.initSchema();
repoPath = mkdtempSync(join(tmpdir(), 'gbrain-sync-lock-'));
});
afterEach(async () => {
await engine.disconnect();
if (repoPath) rmSync(repoPath, { recursive: true, force: true });
});
test('two parallel performSync calls in same process: second waits or fails fast', async () => {
seedRepoWithMarkdown(repoPath, 5);
const { performSync } = await import('../src/commands/sync.ts');
// Same-process concurrent calls: PGLite serializes engine ops via its
// exclusive transaction mutex, but the writer-lock is the right barrier.
// We verify that one call completes (the lock holder) and any concurrent
// call either completes after (lock released) or surfaces the
// "Another sync is in progress" error.
const promise1 = performSync(engine, { repoPath, noPull: true, noEmbed: true });
let secondError: unknown = null;
try {
// Tiny delay so promise1 captures the lock first.
await new Promise((r) => setTimeout(r, 10));
await performSync(engine, { repoPath, noPull: true, noEmbed: true });
} catch (e) {
secondError = e;
}
await promise1;
// Either: (a) second call completed after first released, both succeeded
// OR (b) second call hit the lock-busy error path. Either is correct.
if (secondError) {
const msg = secondError instanceof Error ? secondError.message : String(secondError);
expect(msg).toMatch(/Another sync is in progress|lock|gbrain-sync/i);
}
});
});