Compare commits

..
Author SHA1 Message Date
Garry Tan e22b8fb555 Merge remote-tracking branch 'origin/master' into feat/parallel-sync
# Conflicts:
#	CHANGELOG.md
#	CLAUDE.md
#	VERSION
#	llms-full.txt
#	package.json
#	src/commands/sync.ts
2026-04-29 22:49:21 -07:00
Garry Tan 2a9feb859f Merge remote-tracking branch 'origin/master' into feat/parallel-sync
# Conflicts:
#	CHANGELOG.md
#	CLAUDE.md
#	VERSION
#	llms-full.txt
#	package.json
2026-04-29 22:18:26 -07:00
Garry Tan 15b9316dbf Merge remote-tracking branch 'origin/master' into feat/parallel-sync
# Conflicts:
#	CHANGELOG.md
#	TODOS.md
#	VERSION
#	package.json
2026-04-29 11:37:09 -07:00
Garry TanandClaude Opus 4.7 f739de5521 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>
2026-04-29 08:23:44 -07:00
Garry TanandClaude Opus 4.7 02d585c0a4 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>
2026-04-28 20:05:17 -07:00
Garry Tan e573fa6988 Merge remote-tracking branch 'origin/master' into feat/parallel-sync
# Conflicts:
#	CHANGELOG.md
#	CLAUDE.md
#	VERSION
#	package.json
2026-04-28 19:56:25 -07:00
Garry Tan ff6320e552 Merge remote-tracking branch 'origin/master' into pr-490
# Conflicts:
#	CHANGELOG.md
#	CLAUDE.md
#	TODOS.md
#	VERSION
#	package.json
2026-04-28 08:53:54 -07:00
Garry TanandClaude Opus 4.7 36c750bbec 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>
2026-04-28 08:47:19 -07:00
Garry TanandClaude Opus 4.7 7f2c81f929 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>
2026-04-28 08:43:07 -07:00
Garry TanandClaude Opus 4.7 1353366b5f 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>
2026-04-28 08:42:54 -07:00
Garry TanandClaude Opus 4.7 93ae40dd3a 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>
2026-04-28 08:42:43 -07:00
Garry TanandClaude Opus 4.7 8fcd2737bf 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>
2026-04-28 08:42:30 -07:00
Garry TanandClaude Opus 4.7 b23f24f91b 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>
2026-04-28 08:42:21 -07:00
Garry TanandClaude Opus 4.7 10d96545a4 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>
2026-04-28 08:42:02 -07:00
root 6b2f3bc321 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.
2026-04-28 06:43:12 +00:00
17 changed files with 25 additions and 2229 deletions
-1
View File
@@ -18,4 +18,3 @@ eval/data/world-v1/world.html
# BrainBench amara-life-v1 Opus cache (regenerate via eval:generate-amara-life)
eval/data/amara-life-v1/_cache/
.claude/
export/
-186
View File
@@ -2,192 +2,6 @@
All notable changes to GBrain will be documented in this file.
## [0.22.15] - 2026-04-29
## **Throw bare markdown into your brain and it becomes properly typed knowledge. No YAML ceremony.**
A real 81K-page brain has 9,655 files with no frontmatter. They imported fine, but every one of them landed in the DB as `type: concept`, `title: <slugified-filename>`, no date, no source, no tags. Search ranking suffered. Type-filtered queries missed them. Entity resolution fell over.
This release adds path-aware frontmatter inference. `gbrain sync` now synthesizes type, date, source, and tags from the filesystem path and first heading the moment a bare-frontmatter file imports. No LLM call, fully deterministic, file on disk untouched. An Apple Note at `Apple Notes/2010-04-13 founders mtg.md` lands as `type: apple-note, title: founders mtg, date: 2010-04-13, source: apple-notes` instead of `type: concept, title: 2010 04 13 Founders Mtg`.
If you want the inference written back to git, the new `gbrain frontmatter generate <path> --fix` walks a brain dir, infers frontmatter for every file that lacks it, and writes back with `.bak` safety backups. Dry-run by default.
### The 9,655 numbers that matter
Measured against my actual brain (gbrain v0.22.8 + the new inference path).
| Behavior | Before v0.22.15 | After v0.22.15 |
|---|---|---|
| Files importing as `type: concept` (no frontmatter) | 9,655 | 0 |
| Apple Notes typed correctly (`apple-note`) | 0 | 5,861 |
| Calendar indexes typed correctly (`calendar-index`) | 0 | 3,201 |
| Therapy sessions typed + dated | 0 | 60 |
| Essay drafts typed + dated | 0 | 33 |
| LLM cost for the full reclassification | n/a | $0 |
The agent doing type-filtered queries on your brain (`type: person`, `type: meeting`, `type: essay`) now actually finds those pages instead of treating everything as `concept`.
### What this means for you
If you've been resisting frontmatter ceremony — same. Throw bare markdown into your brain and inference handles it. The rules table in `src/core/frontmatter-inference.ts` covers the obvious directories (`people/`, `companies/`, `daily/calendar/`, `writing/`, `meetings/`, `personal/`, etc.) plus a generic catch-all. Adding a new convention is one line in `DIRECTORY_RULES`.
## To take advantage of v0.22.15
`gbrain upgrade` should do this automatically. Then:
1. **Run a dry-run preview:**
```bash
gbrain frontmatter generate ~/brain
```
You'll see how many files would get inferred frontmatter and the breakdown by type.
2. **Optionally write back to git:**
```bash
gbrain frontmatter generate ~/brain --fix
```
Each modified file gets a `.bak` backup before rewrite.
3. **Re-sync to pick up the new metadata:**
```bash
gbrain sync ~/brain
```
Inferred frontmatter is folded into `content_hash`, so previously-bare files re-import once with proper types and re-embed. Subsequent syncs are idempotent.
4. **If anything looks off,** please file an issue: https://github.com/garrytan/gbrain/issues with the path of the misclassified file and the rule that matched.
### Itemized changes
#### Features
- `src/core/frontmatter-inference.ts` (new module) — Path-aware frontmatter synthesis. `DIRECTORY_RULES` table maps path prefixes to type/date/title/source/tags. First-match-wins. Date extraction from filenames (`YYYY-MM-DD` prefix or anywhere). Title extraction with date-prefix stripping and first-`#`-heading fallback (20-line window). YAML-safe serialization with quoting for special characters.
- `src/core/import-file.ts``importFromFile()` runs inference inline before `parseMarkdown()` when `opts.inferFrontmatter !== false` (default on). The synthesized frontmatter folds into the in-memory content for parsing, chunking, embedding, and content-hash computation. The file on disk is not modified.
- `src/commands/frontmatter.ts` — New `gbrain frontmatter generate <path> [--fix] [--dry-run] [--json]` subcommand. Walks a directory (skips `.git`, `node_modules`, `.obsidian`, symlinks), runs inference on every `.md` file without frontmatter, optionally writes back with `.bak` backups. Auto-detects brain root by walking up for `.git`. Shows per-type breakdown and first-10 examples.
#### Fixes
- `src/commands/frontmatter.ts:344``runGenerate` dynamic path import now includes `basename`. Single-file invocation (`gbrain frontmatter generate <file>`) previously crashed with `ReferenceError: basename is not defined` on the relative-path-empty fallback at line 437.
#### Tests
- `test/frontmatter-inference.test.ts` (new, 35 cases) — date extraction (5), title extraction from filenames (5) and headings (4 incl. 20-line boundary), inference for every directory rule (13 incl. Apple Notes subfolder tagging), serialization with YAML-safe quoting (4), `applyInference` integration (2), rule ordering and catch-all coverage (2).
## [0.22.14] - 2026-04-29
**Bare `gbrain jobs work` now self-monitors and fail-stops cleanly when its database dies or the queue stalls.**
**The wedged-worker class of bug — process alive, jobs piling up, your `pgrep` check happily green — is gone.**
A production brain (54K pages, Supabase Postgres, 3-concurrency worker under a cron-based PM)
hit it last week: worker process state=Sl at 13:15 UTC, stopped claiming jobs, 21 jobs stacked
in `waiting` over two hours, 5 autopilot-cycles dead-lettered at the 600s timeout, then 150
zombie processes accumulated over the container's 31-day life. The PM's `pgrep` saw a live
PID and reported green the entire time.
Pre-v0.22.14, bare `gbrain jobs work` had **zero** health monitoring. The supervisor (`gbrain
jobs supervisor`) had the right protections — DB liveness probes, stall detection, RSS
watchdog, reconnect on transient PgBouncer blips — but the supervisor wraps `jobs work` as a
child, and many production deployments run bare `jobs work` directly under systemd, Docker,
launchd, cron watchdog, or supervisord. That mode got nothing.
This release moves health monitoring into the bare worker itself, gated by `GBRAIN_SUPERVISED=1`
so it doesn't double up under the supervisor. When the worker detects it's wedged, it emits an
`'unhealthy'` event with a structured reason, and the CLI calls `process.exit(1)` so the external
PM restarts it cleanly. **This is fail-stop:** the worker exits and stays dead until your PM
brings it back. If you run bare `jobs work` without a restart loop, you need one now.
### The numbers that matter
Detection signatures the new health check catches, measured against the production incident
above (and the 30-day deployment running under the band-aid bash watchdog Garry deployed before
this fix):
| Failure mode | Before v0.22.14 | After v0.22.14 |
|---|---|---|
| DB connection death (Supabase/PgBouncer drop) | undetected; worker idles forever | 3 consecutive `SELECT 1` failures (≤3min) → `'unhealthy'`+exit |
| Hung DB probe (network partition) | timer wedged forever, monitoring silently disabled | 10s probe timeout per tick → counted as failure → exit at strike 3 |
| Worker stall (event loop alive, claim returns null) | undetected; jobs pile up in `waiting` | 5min warn, 10min `'unhealthy'`+exit (measured from last completion) |
| Memory leak (RSS climbing past 2GB) | undetected on bare workers | watchdog default 2048 MB triggers `gracefulShutdown('watchdog')` |
| Worker stalled but waiting jobs are unhandled type | ❌ false-positive exit (restart loop) | filter by registered handler names, no exit |
Operationally: from the band-aid bash watchdog Garry deployed before this fix, fresh worker
restart cleared 21 waiting → 0 in 2 minutes, then ran stable for 30+ min with 130 MB RSS,
autopilot-cycles completing in 0.20.6s instead of timing out at 600s.
### What this means for operators
Add a restart policy to your bare-worker invocation BEFORE upgrading. The new behavior is
fail-stop, not self-healing — without a restart loop, your worker will exit on the first DB
blip and stay dead. systemd `Restart=always`, Docker `restart: always`, launchd `KeepAlive`,
cron watchdog, supervisord `autorestart=true`. The migration walks every PM. If you're using
`gbrain jobs supervisor`, you're already protected — the supervisor handles spawn-on-crash
itself.
The default `--max-rss` for bare workers also bumped from 0 (off) to 2048 MB. If you ran bare
workers with intentionally large embed/import jobs, raise the limit (`--max-rss 4096`) or opt
out (`--max-rss 0`). The migration includes per-PM unit-file edits.
## To take advantage of v0.22.14
`gbrain upgrade` should do this automatically. If it didn't, or if `gbrain doctor` warns about
a bare worker exiting with watchdog signatures:
1. **Confirm your bare-worker invocations have a restart policy:**
```bash
# systemd
grep -E '^Restart=' ~/.config/systemd/user/gbrain-worker.service /etc/systemd/system/gbrain-worker.service 2>/dev/null
# crontab
crontab -l | grep "gbrain jobs work"
# launchctl
plutil -p ~/Library/LaunchAgents/com.user.gbrain-worker.plist | grep -A1 KeepAlive
```
2. **Decide on RSS posture:**
- Default 2048 MB matches supervisor behavior. Most bare workers fit.
- Embed/import jobs > 2GB? Pass `--max-rss 4096` (or higher).
- Intentionally unbounded? Pass `--max-rss 0`.
3. **Walk the migration:** `skills/migrations/v0.22.14.md` has the full per-PM table and a
verification block.
4. **Verify:**
```bash
gbrain jobs stats
gbrain doctor --json | jq '.checks[] | select(.name == "queue_health")'
```
Worker startup line should now read:
`Minion worker started (queue: default, concurrency: 3, watchdog: 2048MB, health-check: 60s)`
Under supervisor: the `health-check: Ns` segment is absent (supervisor handles it).
5. **If anything fails or numbers look wrong**, file an issue at
https://github.com/garrytan/gbrain/issues with `gbrain doctor` output and the contents of
`~/.gbrain/upgrade-errors.jsonl` if it exists.
### Itemized changes
#### Added
- `MinionWorkerOpts.{healthCheckInterval, stallWarnAfterMs, stallExitAfterMs, dbFailExitAfter, dbProbeTimeoutMs}` — five new tuning knobs. Defaults: 60s probe interval, 5min warn / 10min exit, 3 DB strikes, 10s per-probe timeout.
- `MinionWorker` now extends `EventEmitter`. Emits `'unhealthy'` with `{ reason: 'db_dead', consecutiveFailures, message } | { reason: 'stalled', waitingCount, idleMinutes }`. CLI subscribes; direct API consumers without a listener inherit a fail-stop fallback that calls `process.exit(1)` to preserve pre-refactor semantics.
- `gbrain jobs work --health-interval MS` — tune the self-health-check cadence (0 disables; rejects NaN/negative/sub-1000ms typos).
- `gbrain jobs supervisor --health-interval MS` — same flag, same validation, same `0 = disable` contract on the supervisor's own probe.
- `GBRAIN_SUPERVISED=1` env var on the supervisor's spawned worker child (skips the child's self-health timer to avoid double-monitoring).
- `gbrain doctor` `queue_health` subcheck reports RSS-watchdog kills in the last 24h via exact match on `error_text = 'aborted: watchdog'` scoped to `status IN ('dead','failed')`.
- `skills/migrations/v0.22.14.md` — full migration walkthrough with per-PM restart-policy preflight, RSS-posture decision tree, and per-system unit-file edits.
#### Changed
- **Default `--max-rss` for `gbrain jobs work`: 0 → 2048 MB.** Matches supervisor default. Catches memory-leak stalls that previously went undetected on bare workers. Opt out with `--max-rss 0`.
- **Bare-worker behavior is now fail-stop** when the DB is unreachable or the queue stalls. Pre-v0.22.14 the worker idled silently. Now it exits and relies on the external PM (systemd, Docker, launchd, cron, supervisord) to restart cleanly.
- Stall query at `worker.ts` filters by registered handler names (`AND name = ANY($2::text[])`) so workers don't false-positive when waiting jobs of unhandled names accumulate.
- Stall exit threshold measured from `lastCompletionTime` (not from when the warning fired), so 5min warn / 10min exit means total idle of 10 min — not 15 min.
- DB liveness probe wrapped in `Promise.race` against a 10s timeout so a hung `executeRaw` cannot wedge the recursive `setTimeout` chain forever.
- `setInterval` → recursive `setTimeout` with a `running` flag throughout. Eliminates timer-callback overlap on slow probes.
- `parseMaxRssFlag` returns `number | undefined` (was `number`) so callers distinguish absent from explicit-disable.
- `process.env.GBRAIN_SUPERVISED` check tightened from `!!env.X` to `=== '1'` (precise contract; no fuzzy matching on `'0'` or `'false'`).
- `MinionWorker` constructor throws when `stallExitAfterMs <= stallWarnAfterMs` so misconfigurations fail loudly at startup.
#### Fixed
- **Wedged-worker false-positive on heterogeneous queues** — workers registering only some handlers no longer interpret waiting jobs of other names as a stall. Repeated `process.exit(1)` → restart loop is gone.
- **Hung DB probe wedge** — pre-fix, a hung `executeRaw('SELECT 1')` kept the recursive `setTimeout` from rescheduling, silently disabling the entire health monitor. Post-fix, the probe times out and counts as a failure.
- **`--health-interval 0` no longer DB-hammers the supervisor.** Pre-fix, the documented "0 disables" contract was a lie — `setInterval(cb, 0)` schedules a tight loop. Now gated behind `> 0`.
- **Inline `jobs submit --follow` and `jobs smoke` no longer kill the user's CLI session** on a DB blip. Both now pass `healthCheckInterval: 0` so the no-listener fallback can't trip on one-shot runs.
- Doctor's RSS-watchdog hint matches the actual error_text signature (`'aborted: watchdog'`) instead of the wrong `'memory limit'` literal that never matched.
#### For contributors
- `MinionWorker extends EventEmitter` — if you import the class directly, the `on('unhealthy', ...)` event is now part of the public surface. The `UnhealthyReason` discriminated union is exported from `src/core/minions/worker.ts`.
- New regression-test infrastructure in `test/minions.test.ts`: `makeProbeEngine(overrides)` is a Proxy-based engine wrapper that intercepts `SELECT 1` and the stall `count(*)` query while passing every other call through to the real PGLite engine. Useful for any future test that needs to inject DB liveness or stall semantics without mocking the entire engine surface.
### Adjacent (separate PR, v0.22.15)
PR #503 catches the *symptom* of one specific failure mode. The cause-side fix — `runPhaseEmbed → embed.ts → embedBatch` not honoring `signal.aborted` between OpenAI batch calls — ships in v0.22.15 (highest-priority TODO; daily wedge driver). Plumbing is documented in `TODOS.md`.
## [0.22.13] - 2026-04-28
**Sync got faster, and the bookmark stopped lying.**
-112
View File
@@ -1,117 +1,5 @@
# TODOS
## minions / worker (v0.22.14 follow-ups)
### v0.22.15 — Embed cooperative-abort (HIGHEST PRIORITY — daily pain)
**Priority:** P0
**What:** Plumb `signal: AbortSignal` through `runPhaseEmbed`
`src/commands/embed.ts``embedBatch` in `src/core/embedding.ts`. Check
`signal?.aborted` between OpenAI batch calls (every ~100 texts, ~2s
real-time) and between slugs in the per-slug loop.
**Why:** Embed phase ignores `signal.aborted` between batches today. Job
wall-clock timeout fires → handler keeps running → cycle's finally block
unreachable → `gbrain_cycle_locks` row stays held indefinitely. Every
subsequent autopilot cron cycle sees `cycle_already_running` → skips. Lock
TTL is 30 min; new cycles give up before that. Doctor reports UNHEALTHY.
**The chain in production:** ~5min cron submits cycle → 22K stale pages →
embed phase takes 1015 min → 600s timeout fires → job dead-lettered → embed
keeps running → lock held → all subsequent cycles skip. Garry hits this
DAILY on his production brain.
**Pros:** Closes the daily wedge. Makes timeouts actually effective. Lets
operators bump worker timeouts confidently knowing abort actually stops
work.
**Cons:** Touching the embed hot path; small risk of botching the abort
checks. Mitigation: between-batch granularity (~2s), not per-text (too fine)
or per-slug (too coarse for 500+ chunk slugs).
**Context:** PR #503 (v0.22.14) catches the SYMPTOM (worker stalled, queue
piling up) via self-health-monitoring. This PR catches the CAUSE for one
specific failure class. Both fixes are needed; they're complementary, not
duplicative.
**Files to touch:**
- `src/core/cycle.ts:579``runPhaseEmbed(engine, dryRun)` → add
`signal?: AbortSignal` arg
- `src/core/cycle.ts:803` — pass `opts.signal` through
- `src/commands/embed.ts:~363` — accept signal, check between slugs
- `src/core/embedding.ts:51-56``embedBatch(texts, onProgress?, signal?)`,
check between for-loop iterations of `BATCH_SIZE` slices
**Tests required:**
1. embedBatch checks signal between OpenAI calls; aborts within one batch (~2s)
2. Per-slug loop in `embed.ts` checks signal between slugs
3. End-to-end: cycle handler with embed phase + signal aborted mid-flight →
finally runs → `gbrain_cycle_locks` row deleted
4. Regression: 1K+ chunks scenario — embed does NOT block lock release when
timeout fires
**Effort:** M (human: ~3 hr / CC: ~30 min).
**Depends on / blocked by:** Nothing. v0.22.14 ships first.
### v0.23+ — Bare-worker engine reconnect parity with supervisor
**Priority:** P2
**What:** Extract the supervisor's reconnect-then-fail pattern into
`MinionWorker` so bare workers can retry transient DB blips before exiting.
Today the supervisor calls `engine.reconnect()` after 3 consecutive DB health
failures (#406); the bare worker just emits `'unhealthy'` and the CLI calls
`process.exit(1)`.
**Why:** Bare-worker behavior is more disruptive than supervised behavior on
transient PgBouncer blips. A bare worker restarts the entire process; a
supervised worker just reconnects the pool. Operationally the supervisor
approach is gentler (no in-flight job loss, no PM restart latency).
**Pros:** Unifies bare and supervised behavior. Reduces process churn on
transient network blips.
**Cons:** More code in MinionWorker; risk of reconnect masking a real
problem. Mitigation: cap retry attempts, fall through to `'unhealthy'`
emission after the cap.
**Context:** Filed during v0.22.14 plan-eng-review. The asymmetry is
documented in v0.22.14 CHANGELOG as deliberate; this TODO captures the
"unify someday" intent.
**Effort:** S (human: ~2 hr / CC: ~20 min).
**Depends on / blocked by:** Nothing.
### v0.23+ — `minion_workers` heartbeat table for queue_health doctor (B7)
**Priority:** P3
**What:** Add a `minion_workers` table (`worker_id` PK, `hostname`,
`last_heartbeat`, `queue`, `concurrency`, `started_at`) so the existing
`queue_health` doctor check (Postgres path) can detect dead workers via
heartbeat staleness instead of relying on the indirect `lock_until` proxy.
**Why:** v0.19.1 added `queue_health` checks for stalled-active jobs and
waiting-depth threshold. The worker-heartbeat subcheck was deferred (B7)
because the `lock_until`-on-active-jobs proxy can't distinguish "worker
exited cleanly" from "worker idle" — a check that cries wolf erodes trust
in every doctor check. With a real heartbeat row, doctor can say "no worker
seen in N intervals" with confidence.
**Pros:** Doctor's `queue_health` becomes ground-truth. Detects "worker
container died but cron didn't restart it" scenario.
**Cons:** New table, schema migration, every health-tick UPSERTs. Costs
a write per worker per minute (default).
**Context:** Filed during v0.22.14 plan-eng-review. PR #503's self-health
monitoring is the worker-side liveness; this would be the queue-side
ground-truth.
**Effort:** M (human: ~1 day / CC: ~1 hr).
**Depends on / blocked by:** Schema migration system; nothing else.
## sync (v0.22.13 follow-up — PR #490 review)
### D-PR490-1 — Plumb resolved `database_url` through `SyncOpts`
+1 -1
View File
@@ -1 +1 @@
0.22.15
0.22.13
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "gbrain",
"version": "0.22.15",
"version": "0.22.13",
"description": "Postgres-native personal knowledge brain with hybrid RAG search",
"type": "module",
"main": "src/core/index.ts",
-210
View File
@@ -1,210 +0,0 @@
---
feature_pitch:
headline: Bare workers now self-monitor and fail-stop into your PM's restart loop
body: |
Bare `gbrain jobs work` now ships with the same health protection the
supervisor already had: DB liveness probes (with per-probe timeout so a
hung connection can't wedge the monitor), stall detection filtered by
registered handler names, and an RSS watchdog default of 2048 MB.
When the worker detects it's wedged (stuck pgbouncer connection, hung
event loop, stalled job claim), it emits `'unhealthy'` and the CLI calls
`process.exit(1)`. This is **fail-stop**: it requires an external process
manager (systemd, Docker `restart: always`, launchd `KeepAlive`, cron
watchdog) to bring the worker back. Without one, the process exits and
stays dead — that's a regression from pre-v0.22.14 self-healing.
Pre-v0.22.14 behavior: bare workers had ZERO health monitoring. A wedged
worker stayed alive doing nothing while jobs piled up in `waiting` and
your PM's `pgrep` check happily reported green.
If you're using `gbrain jobs supervisor`, you're already protected — the
supervisor handles spawn-on-crash itself. The fail-stop concern only
applies to direct `gbrain jobs work` invocations.
---
# v0.22.14 — Bare-worker self-health-monitoring
## ⚠️ Pre-flight: confirm you have a process supervisor
If you run `gbrain jobs work` directly (NOT under `gbrain jobs supervisor`),
verify your process manager is configured to restart the worker on exit
BEFORE upgrading:
| Manager | What to check |
|---|---|
| systemd | `Restart=always` (or `Restart=on-failure`) in the `.service` unit |
| Docker | `restart: always` / `restart: unless-stopped` in compose, OR `--restart` flag |
| launchd (macOS) | `<key>KeepAlive</key><true/>` in the plist |
| cron watchdog | Cron entry that re-spawns when `pgrep -f "gbrain jobs work"` is empty |
| supervisord | `autorestart=true` |
**If your bare worker has no restart loop, the v0.22.14 fail-stop behavior
will leave you with a dead worker after the first DB blip.** Either add a
restart policy OR switch to `gbrain jobs supervisor` (which spawns its own
child + restarts on crash internally).
## What ships
- DB liveness probes inside `gbrain jobs work` (60s interval, 3 strikes → exit)
- Stall detection (5min warn / 10min exit when waiting jobs accumulate but
in-flight is empty)
- `--max-rss` defaults to 2048 MB for bare workers (matches supervisor default;
was 0 = disabled)
- New `MinionWorkerOpts.{healthCheckInterval, stallWarnAfterMs,
stallExitAfterMs, dbFailExitAfter, dbProbeTimeoutMs}` for tuning (5 fields)
- `MinionWorker` now extends `EventEmitter`; emits `'unhealthy'` event with
a structured reason payload. **No-listener fallback**: if the caller does
not subscribe to `'unhealthy'`, the worker calls `process.exit(1)` itself
to preserve the pre-refactor fail-stop behavior. The CLI subscribes; direct
API consumers without a listener inherit the fail-stop default. Inline
paths (`jobs submit --follow`, `jobs smoke`) explicitly pass
`healthCheckInterval: 0` to disable the timer entirely so the fallback
cannot trip on a one-shot run.
- `GBRAIN_SUPERVISED=1` env var (set by supervisor on its child) causes the
child worker to skip its own health timer (no double-monitoring)
- Constructor validation: throws if `stallExitAfterMs <= stallWarnAfterMs`
(the contract requires strict greater-than)
## ⚠️ Breaking change you may need to act on
**`gbrain jobs work` now defaults `--max-rss 2048` (MB).** Pre-v0.22.14, bare
workers had no RSS limit. If you ran bare workers with intentionally large
embed/import working sets, your worker may now get killed by the watchdog after
upgrade.
**You need to act if any of the following is true:**
- You run `gbrain jobs work` directly under a process manager (systemd, Docker,
cron, supervisord, your own bash wrapper)
- Your worker handles large embed batches, multi-GB import jobs, or
memory-intensive subagent tasks that legitimately exceed 2 GB RSS
- You rely on bare-worker resident memory exceeding 2 GB
**You do NOT need to act if:**
- You run `gbrain jobs supervisor` (the supervisor already defaulted to
`--max-rss 2048` and has reconnect-then-fail logic on top)
- Your typical worker stays well under 2 GB RSS
## What the agent should do post-upgrade
### 1. Run the orchestrator (mechanical migration)
```bash
gbrain apply-migrations --yes
```
This is a no-op for v0.22.14 — there's no schema migration. The mechanical work
is in the binary; the user-side decision is the breaking-change handling below.
### 2. Detect bare-worker deployments
If the user runs OpenClaw, Hermes, or any agent fork that uses `gbrain jobs
work` directly (NOT under `gbrain jobs supervisor`), they need to know. Check:
```bash
# Look for bare worker invocations in cron, systemd, Docker, launchctl, etc.
grep -r "gbrain jobs work" ~/.config/systemd/ /etc/systemd/ 2>/dev/null | head
crontab -l 2>/dev/null | grep "gbrain jobs work" | head
launchctl list 2>/dev/null | grep gbrain | head
ps -ef | grep "gbrain jobs work" | grep -v grep | grep -v supervisor
```
If you find bare-worker invocations, surface them to the user.
### 3. Ask about RSS profile
Use AskUserQuestion to confirm whether the user's bare workers need >2 GB RSS:
> "v0.22.14 changes `gbrain jobs work` to default `--max-rss 2048` (MB).
> Pre-v0.22.14 bare workers had no limit. If your worker handles large embed
> batches or multi-GB imports, the watchdog may now kill it. Do you want
> to keep the new 2 GB default, raise the limit, or opt out entirely?"
Options:
- **A) Keep 2 GB default (recommended for most)** — protects against memory
leaks; restarts on overflow; matches supervisor behavior.
- **B) Raise to N GB (specify N)** — pass `--max-rss <N*1024>` to the worker
invocation.
- **C) Opt out** — pass `--max-rss 0`.
### 4. Apply the user's choice
For each bare-worker invocation, edit the unit/cron/launchctl/script to add
the chosen `--max-rss` flag.
**systemd (~/.config/systemd/user/gbrain-worker.service):**
```ini
ExecStart=/usr/local/bin/gbrain jobs work --queue default --concurrency 3 --max-rss 4096
# Or to opt out: --max-rss 0
```
Then `systemctl --user daemon-reload && systemctl --user restart gbrain-worker`.
**cron (`crontab -e`):**
```cron
@reboot /usr/local/bin/gbrain jobs work --queue default --concurrency 3 --max-rss 4096
```
**Docker compose:**
```yaml
command: ["gbrain", "jobs", "work", "--queue", "default", "--concurrency", "3", "--max-rss", "4096"]
```
**launchctl (~/Library/LaunchAgents/com.user.gbrain-worker.plist):**
```xml
<key>ProgramArguments</key>
<array>
<string>/usr/local/bin/gbrain</string>
<string>jobs</string>
<string>work</string>
<string>--max-rss</string>
<string>4096</string>
</array>
```
Then `launchctl unload ... && launchctl load ...`.
### 5. (Optional) Tune health-check thresholds
The new opts default to sensible values (60s probe interval, 5min warn / 10min
exit, 3 DB failures). If you have specific SLAs, you can pass `--health-interval
<ms>` to adjust the probe cadence. Stall thresholds are not yet CLI-exposed
(only the API; CLI flags coming in a follow-up).
To disable self-monitoring entirely (e.g. you have your own external health
checker):
```bash
gbrain jobs work --health-interval 0 --max-rss 0
```
### 6. Verify
```bash
gbrain jobs stats # queue should be flowing normally
gbrain doctor --json | jq '.' # no critical warnings
ps -o rss= -p $(pgrep -f "gbrain jobs work") | awk '{print $1/1024 " MB"}'
```
Worker startup log line should now show health-check status:
```
Minion worker started (queue: default, concurrency: 3, watchdog: 2048MB, health-check: 60s)
```
If running under supervisor, you'll see the watchdog but NOT the `health-check:
60s` segment (because `GBRAIN_SUPERVISED=1` skips the child's self-monitor).
### 7. If anything fails
Open an issue at https://github.com/garrytan/gbrain/issues with:
- Output of `gbrain doctor`
- Contents of `~/.gbrain/upgrade-errors.jsonl` if it exists
- Your bare-worker invocation (systemd unit / cron line / Dockerfile snippet)
- Which step broke
-32
View File
@@ -774,30 +774,6 @@ export async function runDoctor(engine: BrainEngine | null, args: string[], dbSo
ORDER BY depth DESC
LIMIT 5
`;
// Subcheck 3 (v0.22.14): RSS-watchdog kills in the last 24h. Bare workers
// newly default to --max-rss 2048 (was 0); operators who run large embed
// or import jobs may see kills that didn't happen pre-v0.22.14. We surface
// a hint when this signature appears so the upgrade path is obvious.
// Signature: when the watchdog trips, gracefulShutdown('watchdog') aborts
// in-flight jobs with `new Error('watchdog')`. The worker's failJob path
// (worker.ts:660-664) writes `error_text = 'aborted: watchdog'` for any
// job in-flight at the moment of the kill.
//
// We deliberately DO NOT do a loose `ILIKE '%watchdog%'`:
// 1. Parent jobs that inherit `on_child_fail='fail_parent'` get
// `"child job N failed: aborted: watchdog"` — counting that
// double-counts (child + parent) for one watchdog event.
// 2. Any user error_text containing the word "watchdog" matches.
// Match the exact prefix `'aborted: watchdog'` to scope this purely to
// the worker's own kill signature.
const rssKillRows: Array<{ cnt: number }> = await sql`
SELECT count(*)::int AS cnt
FROM minion_jobs
WHERE status IN ('dead', 'failed')
AND finished_at > now() - interval '24 hours'
AND error_text = 'aborted: watchdog'
`;
const rssKillCount = rssKillRows[0]?.cnt ?? 0;
const problems: string[] = [];
if (stalledRows.length > 0) {
@@ -818,14 +794,6 @@ export async function runDoctor(engine: BrainEngine | null, args: string[], dbSo
`Fix: set maxWaiting on the submitter (or raise GBRAIN_QUEUE_WAITING_THRESHOLD).`
);
}
if (rssKillCount > 0) {
problems.push(
`${rssKillCount} job(s) dead-lettered for RSS-watchdog memory-limit kills in last 24h. ` +
`v0.22.14 changed the bare-worker --max-rss default from 0 (off) to 2048 MB. ` +
`Fix: raise the limit (e.g. \`gbrain jobs work --max-rss 4096\`) or opt out (\`--max-rss 0\`). ` +
`See skills/migrations/v0.22.14.md.`
);
}
if (problems.length === 0) {
checks.push({
+1 -193
View File
@@ -49,10 +49,6 @@ export async function runFrontmatter(args: string[]): Promise<void> {
}
return;
}
if (sub === 'generate') {
await runGenerate(rest);
return;
}
if (sub === 'install-hook') {
const { runFrontmatterInstallHook } = await import('./frontmatter-install-hook.ts');
await runFrontmatterInstallHook(rest);
@@ -75,11 +71,10 @@ async function connectEngineForAudit(): Promise<BrainEngine> {
}
function printHelp() {
console.log(`gbrain frontmatter — frontmatter validation, audit, auto-repair, and generation
console.log(`gbrain frontmatter — frontmatter validation, audit, and auto-repair
Usage:
gbrain frontmatter validate <path> [--json] [--fix] [--dry-run]
gbrain frontmatter generate <path> [--fix] [--dry-run] [--json]
gbrain frontmatter audit [--source <id>] [--json]
gbrain frontmatter install-hook [--source <id>] [--force] [--uninstall]
@@ -96,26 +91,6 @@ validate
--dry-run Preview --fix without writing.
--json Emit a JSON envelope on stdout.
generate
Synthesize frontmatter for files that have none (MISSING_OPEN). Uses
directory-aware rules to infer type, title, date, source, and tags from
the filesystem path and file content. Zero LLM calls, fully deterministic.
Without --fix: dry-run preview showing what would be generated.
With --fix: writes frontmatter to files (with .bak safety backups).
Rules are defined in src/core/frontmatter-inference.ts DIRECTORY_RULES.
Add new directory conventions by adding rules to the table.
Examples:
gbrain frontmatter generate /path/to/brain # preview all
gbrain frontmatter generate /path/to/brain --fix # write all
gbrain frontmatter generate /path/to/brain/people/ --fix # just people/
--fix Write generated frontmatter to files (.bak safety backups).
--dry-run Preview without writing (default when --fix is omitted).
--json Emit JSON output.
audit
Read-only scan across all registered sources (or one with --source <id>).
Reports per-source counts grouped by error code. Use this in CI or doctor
@@ -322,170 +297,3 @@ function printAuditHumanReport(report: AuditReport): void {
console.log(`\nFix with: gbrain frontmatter validate <source-path> --fix`);
}
}
// ---------------------------------------------------------------------------
// generate — synthesize frontmatter for files that have none
// ---------------------------------------------------------------------------
async function runGenerate(args: string[]): Promise<void> {
const targetPath = args.find(a => !a.startsWith('-'));
const doFix = args.includes('--fix');
const dryRun = args.includes('--dry-run');
const jsonOut = args.includes('--json');
if (!targetPath) {
console.error('error: gbrain frontmatter generate requires a <path> argument');
console.error('usage: gbrain frontmatter generate <path> [--fix] [--dry-run] [--json]');
process.exitCode = 1;
return;
}
const { inferFrontmatter, serializeFrontmatter } = await import('../core/frontmatter-inference.ts');
const { resolve, relative, join, basename } = await import('path');
const { readFileSync, writeFileSync, copyFileSync, statSync, readdirSync, lstatSync } = await import('fs');
const rootPath = resolve(targetPath);
const isDir = statSync(rootPath).isDirectory();
// Find the brain root — walk up from targetPath looking for .git or known brain markers.
// Inference rules match against brain-root-relative paths (e.g., "people/alice.md").
let brainRoot = rootPath;
if (isDir) {
let candidate = rootPath;
for (let i = 0; i < 10; i++) {
try {
statSync(join(candidate, '.git'));
brainRoot = candidate;
break;
} catch {
const parent = resolve(candidate, '..');
if (parent === candidate) break;
candidate = parent;
}
}
}
interface GenerateResult {
path: string;
type: string;
title: string;
date?: string;
rule: string;
}
const results: GenerateResult[] = [];
let scanned = 0;
let skipped = 0;
let generated = 0;
let written = 0;
function processFile(absPath: string, relPath: string) {
scanned++;
if (!absPath.endsWith('.md')) return;
// Skip symlinks
try { if (lstatSync(absPath).isSymbolicLink()) return; } catch { return; }
let content: string;
try { content = readFileSync(absPath, 'utf-8'); } catch { return; }
const inferred = inferFrontmatter(relPath, content);
if (inferred.skipped) {
skipped++;
return;
}
generated++;
results.push({
path: relPath,
type: inferred.type,
title: inferred.title,
date: inferred.date,
rule: inferred.matchedRule || '(default)',
});
if (doFix && !dryRun) {
const fm = serializeFrontmatter(inferred);
const newContent = fm + '\n' + content;
// Safety: write .bak first
copyFileSync(absPath, absPath + '.bak');
writeFileSync(absPath, newContent, 'utf-8');
written++;
}
}
function walkDir(dir: string, rootForRel: string) {
let entries: string[];
try { entries = readdirSync(dir); } catch { return; }
for (const entry of entries) {
if (entry === '.git' || entry === 'node_modules' || entry === '.obsidian') continue;
const abs = join(dir, entry);
try {
const stat = statSync(abs);
if (stat.isDirectory()) {
walkDir(abs, rootForRel);
} else if (stat.isFile() && entry.endsWith('.md')) {
processFile(abs, relative(rootForRel, abs));
}
} catch { /* skip unreadable */ }
}
}
if (isDir) {
walkDir(rootPath, brainRoot);
} else {
const relPath = relative(brainRoot, rootPath) || basename(rootPath);
processFile(rootPath, relPath);
}
// Output
if (jsonOut) {
console.log(JSON.stringify({
scanned,
skipped,
generated,
written,
dryRun: !doFix || dryRun,
results: results.slice(0, 100), // Cap JSON output
totalResults: results.length,
}, null, 2));
return;
}
// Human-readable output
const mode = doFix && !dryRun ? 'WRITE' : 'DRY-RUN';
console.log(`\nFrontmatter generation (${mode})`);
console.log(` Scanned: ${scanned} files`);
console.log(` Already have frontmatter: ${skipped}`);
console.log(` Would generate: ${generated}`);
if (doFix && !dryRun) {
console.log(` Written: ${written} (with .bak backups)`);
}
// Show sample by type
const byType: Record<string, number> = {};
for (const r of results) {
byType[r.type] = (byType[r.type] || 0) + 1;
}
if (Object.keys(byType).length > 0) {
console.log(`\n By type:`);
for (const [type, count] of Object.entries(byType).sort(([, a], [, b]) => b - a)) {
console.log(` ${type}: ${count}`);
}
}
// Show first 10 examples
if (results.length > 0 && (!doFix || dryRun)) {
console.log(`\n Examples:`);
for (const r of results.slice(0, 10)) {
console.log(` ${r.path}`);
console.log(` → type: ${r.type}, title: "${r.title}"${r.date ? `, date: ${r.date}` : ''} [rule: ${r.rule}]`);
}
if (results.length > 10) {
console.log(` ... and ${results.length - 10} more`);
}
if (!doFix) {
console.log(`\n To write: gbrain frontmatter generate ${targetPath} --fix`);
}
}
}
+17 -96
View File
@@ -33,14 +33,14 @@ export function parseMaxWaitingFlag(args: string[]): number | undefined {
}
/** Parse `--max-rss N` (MB). Returns:
* - undefined if the flag is absent (caller decides the default)
* - 0 if the flag is absent (no watchdog by default for bare `jobs work`)
* - 0 if `--max-rss 0` (explicit disable)
* - the value if >= 256
* Errors and exits the process if the flag is non-numeric, negative, or
* positive but < 256 (likely a GB-vs-MB unit-confusion typo). */
export function parseMaxRssFlag(args: string[]): number | undefined {
export function parseMaxRssFlag(args: string[]): number {
const raw = parseFlag(args, '--max-rss');
if (raw === undefined) return undefined;
if (raw === undefined) return 0;
const parsed = parseInt(raw, 10);
if (!Number.isFinite(parsed) || parsed < 0) {
console.error(`Error: --max-rss must be a non-negative integer (MB), got "${raw}"`);
@@ -133,7 +133,6 @@ USAGE
gbrain jobs stats
gbrain jobs smoke
gbrain jobs work [--queue Q] [--concurrency N] [--max-rss MB]
[--health-interval MS]
gbrain jobs supervisor [start] [--detach] [--json]
[--concurrency N] [--queue Q] [--pid-file PATH]
[--max-crashes N] [--health-interval N]
@@ -315,15 +314,8 @@ HANDLER TYPES (built in)
if (follow) {
console.log(`Job #${job.id} submitted (${name}). Executing inline...`);
// Inline execution: run the job in this process. Disable the
// self-health-check timer — inline flows are one-shot and don't have
// a process manager to restart them. With the timer enabled and no
// 'unhealthy' listener, a DB blip would trip emitUnhealthy's
// no-listener fallback and call process.exit(1) from inside the
// library, killing the user's CLI session.
const worker = new MinionWorker(engine, {
queue: queueName, pollInterval: 100, healthCheckInterval: 0,
});
// Inline execution: run the job in this process
const worker = new MinionWorker(engine, { queue: queueName, pollInterval: 100 });
// Register built-in handlers
await registerBuiltinHandlers(worker, engine);
@@ -497,11 +489,7 @@ HANDLER TYPES (built in)
const sigkillRescue = hasFlag(args, '--sigkill-rescue');
const wedgeRescue = hasFlag(args, '--wedge-rescue');
// Smoke harness is short-lived and has no listener — disable the health
// timer so the no-listener fallback can't trip process.exit(1) mid-test.
const worker = new MinionWorker(engine, {
queue: 'smoke', pollInterval: 100, healthCheckInterval: 0,
});
const worker = new MinionWorker(engine, { queue: 'smoke', pollInterval: 100 });
worker.register('noop', async () => ({ ok: true, at: new Date().toISOString() }));
const job = await queue.add('noop', {}, { queue: 'smoke', max_attempts: 1 });
@@ -650,69 +638,19 @@ HANDLER TYPES (built in)
const queueName = parseFlag(args, '--queue') ?? 'default';
const concurrency = resolveWorkerConcurrency(args);
// --max-rss defaults to 2048 for bare workers (matching supervisor default).
// This catches memory-leak stalls that previously went undetected without
// a supervisor. Operators can opt out with `--max-rss 0`.
const maxRssExplicit = parseMaxRssFlag(args);
const maxRssMb = maxRssExplicit ?? 2048;
// --health-interval: self-health-check period in ms. 0 disables. Default: 60_000 (60s).
// Provides DB liveness probes + stall detection for bare workers.
// Automatically skipped when running under a supervisor (GBRAIN_SUPERVISED=1).
// Validated aggressively (parity with --max-rss): reject NaN/negative/non-integer
// values, and reject suspicious sub-1000ms values that are likely a unit-confusion
// typo (e.g. "--health-interval 60" thinking the unit is seconds).
const healthRaw = parseFlag(args, '--health-interval');
let healthCheckInterval = 60_000;
if (healthRaw !== undefined) {
const parsed = parseInt(healthRaw, 10);
if (!Number.isFinite(parsed) || parsed < 0) {
console.error(`Error: --health-interval must be a non-negative integer (ms), got "${healthRaw}"`);
process.exit(1);
}
if (parsed > 0 && parsed < 1000) {
console.error(
`Error: --health-interval ${parsed} is suspiciously low (likely a unit-confusion typo). ` +
`The flag takes milliseconds; for 60-second probes pass 60000. Use 0 to disable.`,
);
process.exit(1);
}
healthCheckInterval = parsed;
}
// --max-rss is opt-in for bare `gbrain jobs work` — preserves pre-v0.21 behavior
// for operators with legitimately large embed/import working sets. The supervisor
// path injects a default 2048; this code path does not.
const maxRssMb = parseMaxRssFlag(args);
try { await queue.ensureSchema(); }
catch (e) { console.error(e instanceof Error ? e.message : String(e)); process.exit(1); }
const worker = new MinionWorker(engine, {
queue: queueName, concurrency, maxRssMb, healthCheckInterval,
});
const worker = new MinionWorker(engine, { queue: queueName, concurrency, maxRssMb });
await registerBuiltinHandlers(worker, engine);
// Subscribe to self-health failures emitted by the worker. Library code
// (worker.ts) never calls process.exit directly so it stays embeddable;
// this CLI layer is the right place to terminate the process and let
// the external PM (systemd, Docker, cron watchdog) restart cleanly.
worker.on('unhealthy', (info) => {
if (info.reason === 'db_dead') {
console.error(
`[health] FATAL: DB unreachable after ${info.consecutiveFailures} probes (${info.message}). ` +
`Exiting for process-manager restart.`,
);
} else {
console.error(
`[health] FATAL: Worker stalled — ${info.waitingCount} waiting job(s) for ` +
`registered handlers, ${info.idleMinutes}m idle. Exiting for process-manager restart.`,
);
}
process.exit(1);
});
const isSupervisedChild = process.env.GBRAIN_SUPERVISED === '1';
const watchdogNote = maxRssMb > 0 ? `, watchdog: ${maxRssMb}MB` : '';
const healthNote = !isSupervisedChild && healthCheckInterval > 0
? `, health-check: ${Math.round(healthCheckInterval / 1000)}s`
: '';
console.log(`Minion worker started (queue: ${queueName}, concurrency: ${concurrency}${watchdogNote}${healthNote})`);
console.log(`Minion worker started (queue: ${queueName}, concurrency: ${concurrency}${watchdogNote})`);
console.log(`Registered handlers: ${worker.registeredNames.join(', ')}`);
await worker.start();
break;
@@ -849,32 +787,15 @@ HANDLER TYPES (built in)
const concurrency = parseInt(parseFlag(args, '--concurrency') ?? '2', 10);
const queueName = parseFlag(args, '--queue') ?? 'default';
const maxCrashes = parseInt(parseFlag(args, '--max-crashes') ?? '10', 10);
// --health-interval (supervisor): validate same as `jobs work` so NaN /
// negative / sub-1000ms typos fail-fast instead of silently disabling
// the supervisor's own health probe.
const supHealthRaw = parseFlag(args, '--health-interval');
let healthInterval = 60_000;
if (supHealthRaw !== undefined) {
const parsed = parseInt(supHealthRaw, 10);
if (!Number.isFinite(parsed) || parsed < 0) {
console.error(`Error: --health-interval must be a non-negative integer (ms), got "${supHealthRaw}"`);
process.exit(1);
}
if (parsed > 0 && parsed < 1000) {
console.error(
`Error: --health-interval ${parsed} is suspiciously low (likely a unit-confusion typo). ` +
`The flag takes milliseconds; for 60-second probes pass 60000. Use 0 to disable.`,
);
process.exit(1);
}
healthInterval = parsed;
}
const healthInterval = parseInt(parseFlag(args, '--health-interval') ?? '60000', 10);
const allowShellJobs = hasFlag(args, '--allow-shell-jobs') ||
!!process.env.GBRAIN_ALLOW_SHELL_JOBS;
const detach = hasFlag(args, '--detach');
// Supervisor defaults --max-rss 2048 (MB) — main production path uses
// the supervisor, so the watchdog is on by default here.
const maxRssMb = parseMaxRssFlag(args) ?? 2048;
// the supervisor, so the watchdog is on by default here. parseMaxRssFlag
// returns 0 when the flag is absent; substitute the supervisor default.
const maxRssRaw = parseMaxRssFlag(args);
const maxRssMb = parseFlag(args, '--max-rss') === undefined ? 2048 : maxRssRaw;
const cliPath = parseFlag(args, '--cli-path') ?? resolveGbrainCliPath();
-410
View File
@@ -1,410 +0,0 @@
/**
* Frontmatter inference synthesize YAML frontmatter from filesystem metadata.
*
* ## Why this exists
*
* GBrain's sync and import pipelines work fine without frontmatter gray-matter
* returns the full content as body, and `inferType`/`inferTitle` in markdown.ts
* provide fallbacks. But the inferred metadata is minimal:
*
* - `type` defaults to 'concept' for most paths
* - `title` is the slugified filename ("2010 04 13 Apr 13 Founders Mtg")
* - No `date` field, no `source` metadata, no folder-aware tagging
*
* This module provides **rich inference** directory-aware type mapping, date
* extraction from filenames, title cleanup (strip date prefixes, HTML entities),
* heading extraction from content, and source/folder tagging. It produces a
* complete frontmatter block that can be:
*
* 1. Written back to the file on disk (via `gbrain frontmatter generate --fix`)
* 2. Used at import time without modifying the file (DB-only inference)
* 3. Shown as a dry-run preview (via `gbrain frontmatter generate --dry-run`)
*
* ## Design principles
*
* - **Never overwrite existing frontmatter.** If a file already has `---`, skip it.
* - **Infer from filesystem first, content second.** Directory path type, filename date + title,
* first `#` heading title fallback, content entity hints.
* - **Deterministic.** Same file always produces the same frontmatter. No LLM calls, no network.
* - **Extensible via rules.** The `DIRECTORY_RULES` table maps path patterns to type + source + tags.
* Adding a new directory convention = adding one rule.
* - **Safe.** `.bak` files on write, `--dry-run` by default in CLI, idempotent.
*
* ## How it fits in the pipeline
*
* ```
* Sync/Import
* file has frontmatter? normal import (existing path)
* file has NO frontmatter?
* inferFrontmatter(filePath, content) synthesize frontmatter
* prepend to content import as usual
* optionally write back to disk (--write-back flag)
* ```
*
* The inference runs BEFORE `parseMarkdown`, so the downstream pipeline sees
* well-formed frontmatter and all the existing validation/chunking/embedding
* logic works unchanged.
*
* ## Directory rules table
*
* Each rule matches a path pattern (case-insensitive prefix) and provides:
* - `type`: page type for the brain schema
* - `source`: optional source tag (e.g., "apple-notes", "therapy")
* - `tags`: optional additional tags
* - `datePattern`: where to look for dates 'filename' (YYYY-MM-DD prefix),
* 'dirname' (parent dir name), or 'none'
* - `titleStrategy`: how to extract title 'filename' (strip date prefix),
* 'heading' (first # in content), 'filename-full' (no date strip)
*/
import { basename, dirname, relative } from 'path';
// ─── Types ───────────────────────────────────────────────────────────
export interface InferredFrontmatter {
title: string;
type: string;
date?: string;
source?: string;
tags?: string[];
/** True if the file already has frontmatter (inference skipped). */
skipped?: boolean;
/** The rule that matched, for debugging. */
matchedRule?: string;
}
export interface DirectoryRule {
/** Case-insensitive path prefix to match (e.g., 'apple notes/'). */
pathPrefix: string;
/** Page type to assign. */
type: string;
/** Optional source tag. */
source?: string;
/** Optional tags to add. */
tags?: string[];
/** Where to look for dates. Default: 'filename'. */
datePattern?: 'filename' | 'dirname' | 'none';
/** How to extract title. Default: 'filename'. */
titleStrategy?: 'filename' | 'heading' | 'filename-full';
}
// ─── Directory Rules ─────────────────────────────────────────────────
// Ordered from most specific to least specific. First match wins.
// Add new directory conventions here.
export const DIRECTORY_RULES: DirectoryRule[] = [
// Apple Notes — bulk import from Apple Notes app. Filenames are
// "YYYY-MM-DD Title.md" with HTML-styled content.
{
pathPrefix: 'apple notes/youtube shows/',
type: 'apple-note',
source: 'apple-notes',
tags: ['youtube', 'shows'],
datePattern: 'filename',
titleStrategy: 'filename',
},
{
pathPrefix: 'apple notes/yc/',
type: 'apple-note',
source: 'apple-notes',
tags: ['yc'],
datePattern: 'filename',
titleStrategy: 'filename',
},
{
pathPrefix: 'apple notes/archived/',
type: 'apple-note',
source: 'apple-notes',
tags: ['archived'],
datePattern: 'filename',
titleStrategy: 'filename',
},
{
pathPrefix: 'apple notes/politics/',
type: 'apple-note',
source: 'apple-notes',
tags: ['politics'],
datePattern: 'filename',
titleStrategy: 'filename',
},
{
pathPrefix: 'apple notes/pitch notes/',
type: 'apple-note',
source: 'apple-notes',
tags: ['pitch-notes'],
datePattern: 'filename',
titleStrategy: 'filename',
},
{
pathPrefix: 'apple notes/gstack/',
type: 'apple-note',
source: 'apple-notes',
tags: ['gstack'],
datePattern: 'filename',
titleStrategy: 'filename',
},
{
pathPrefix: 'apple notes/photo-cameras/',
type: 'apple-note',
source: 'apple-notes',
tags: ['photography'],
datePattern: 'filename',
titleStrategy: 'filename',
},
{
pathPrefix: 'apple notes/jan bowman notes/',
type: 'apple-note',
source: 'apple-notes',
tags: ['therapy', 'jan-bowman'],
datePattern: 'filename',
titleStrategy: 'filename',
},
// Catch-all for Apple Notes not in a subfolder
{
pathPrefix: 'apple notes/',
type: 'apple-note',
source: 'apple-notes',
datePattern: 'filename',
titleStrategy: 'filename',
},
// Calendar diarization files
{
pathPrefix: 'daily/calendar/',
type: 'calendar-index',
source: 'calendar',
datePattern: 'filename',
titleStrategy: 'filename',
},
// Personal sections
{
pathPrefix: 'personal/therapy/',
type: 'therapy-session',
source: 'therapy',
datePattern: 'filename',
titleStrategy: 'filename',
},
{
pathPrefix: 'personal/reflections/',
type: 'reflection',
source: 'personal',
datePattern: 'filename',
titleStrategy: 'heading',
},
{
pathPrefix: 'personal/',
type: 'personal',
source: 'personal',
datePattern: 'none',
titleStrategy: 'heading',
},
// Writing
{
pathPrefix: 'writing/essays/',
type: 'essay',
source: 'writing',
datePattern: 'filename',
titleStrategy: 'heading',
},
{
pathPrefix: 'writing/ideas/',
type: 'idea',
source: 'writing',
datePattern: 'filename',
titleStrategy: 'heading',
},
{
pathPrefix: 'writing/',
type: 'writing',
source: 'writing',
datePattern: 'filename',
titleStrategy: 'heading',
},
// Entity directories — these should already have frontmatter in most cases,
// but the 55 people pages etc. that don't get handled here.
{ pathPrefix: 'people/', type: 'person', titleStrategy: 'heading' },
{ pathPrefix: 'companies/', type: 'company', titleStrategy: 'heading' },
{ pathPrefix: 'projects/', type: 'project', titleStrategy: 'heading' },
{ pathPrefix: 'civic/', type: 'civic', titleStrategy: 'heading' },
{ pathPrefix: 'events/', type: 'event', titleStrategy: 'heading', datePattern: 'filename' },
{ pathPrefix: 'meetings/', type: 'meeting', titleStrategy: 'heading', datePattern: 'filename' },
{ pathPrefix: 'media/', type: 'media', titleStrategy: 'heading' },
// Catch-all for any remaining files
{ pathPrefix: '', type: 'note', titleStrategy: 'heading' },
];
// ─── Date extraction ─────────────────────────────────────────────────
/** Extract YYYY-MM-DD date from a filename like "2010-04-13 Apr 13 founders mtg.md" */
export function extractDateFromFilename(filename: string): string | null {
// Pattern 1: YYYY-MM-DD prefix (with - or space separator after)
const m1 = filename.match(/^(\d{4}-\d{2}-\d{2})[\s_-]/);
if (m1) return m1[1];
// Pattern 2: YYYY-MM-DD anywhere in filename
const m2 = filename.match(/(\d{4}-\d{2}-\d{2})/);
if (m2) return m2[1];
// Pattern 3: "YYYY MM DD" with spaces
const m3 = filename.match(/^(\d{4})\s+(\d{2})\s+(\d{2})\s/);
if (m3) return `${m3[1]}-${m3[2]}-${m3[3]}`;
return null;
}
// ─── Title extraction ────────────────────────────────────────────────
/** Extract title from filename, stripping date prefix and extension. */
export function extractTitleFromFilename(filename: string): string {
// Remove .md extension
let title = filename.replace(/\.md$/i, '');
// Strip YYYY-MM-DD prefix (with separator)
title = title.replace(/^\d{4}-\d{2}-\d{2}[\s_-]+/, '');
// Strip YYYY MM DD prefix (space-separated)
title = title.replace(/^\d{4}\s+\d{2}\s+\d{2}\s+/, '');
// Clean up: title case, replace dashes/underscores with spaces
title = title
.replace(/[-_]/g, ' ')
.replace(/\s+/g, ' ')
.trim();
// Don't title-case if it already has mixed case (e.g., "YC presidency")
if (title === title.toLowerCase() || title === title.toUpperCase()) {
title = title.replace(/\b\w/g, c => c.toUpperCase());
}
return title || 'Untitled';
}
/** Extract title from first heading (# ...) in content. */
export function extractTitleFromHeading(content: string): string | null {
const lines = content.split('\n');
for (const line of lines.slice(0, 20)) {
const m = line.match(/^#\s+(.+)/);
if (m) return m[1].trim();
}
return null;
}
// ─── Core inference ──────────────────────────────────────────────────
/**
* Infer frontmatter for a file that has none.
*
* @param relativePath - Path relative to brain root (e.g., "Apple Notes/2010-04-13 Apr 13 founders mtg.md")
* @param content - File content (may be empty)
* @returns Inferred frontmatter fields
*/
export function inferFrontmatter(relativePath: string, content: string): InferredFrontmatter {
// Check if file already has frontmatter
const firstNonEmpty = content.split('\n').find(l => l.trim().length > 0);
if (firstNonEmpty?.trim() === '---') {
return { title: '', type: '', skipped: true };
}
const lowerPath = relativePath.toLowerCase();
const filename = basename(relativePath);
// Find matching rule
let matchedRule: DirectoryRule | undefined;
for (const rule of DIRECTORY_RULES) {
if (lowerPath.startsWith(rule.pathPrefix.toLowerCase())) {
matchedRule = rule;
break;
}
}
// Default rule if none matched
if (!matchedRule) {
matchedRule = { pathPrefix: '', type: 'note', titleStrategy: 'heading' };
}
// Extract date
let date: string | undefined;
const datePattern = matchedRule.datePattern ?? 'filename';
if (datePattern === 'filename') {
date = extractDateFromFilename(filename) ?? undefined;
}
// Extract title
let title: string;
const titleStrategy = matchedRule.titleStrategy ?? 'filename';
if (titleStrategy === 'heading') {
title = extractTitleFromHeading(content) ?? extractTitleFromFilename(filename);
} else if (titleStrategy === 'filename-full') {
title = filename.replace(/\.md$/i, '').replace(/[-_]/g, ' ').trim();
} else {
title = extractTitleFromFilename(filename);
}
// Build tags from rule + subfolder
const tags = [...(matchedRule.tags ?? [])];
// Add subfolder as tag for Apple Notes (e.g., "YC", "Politics")
if (matchedRule.source === 'apple-notes' && matchedRule.pathPrefix === 'apple notes/') {
const parts = relativePath.split('/');
if (parts.length > 2) {
const subfolder = parts[1].toLowerCase().replace(/\s+/g, '-');
if (!tags.includes(subfolder)) tags.push(subfolder);
}
}
return {
title,
type: matchedRule.type,
date,
source: matchedRule.source,
tags: tags.length > 0 ? tags : undefined,
matchedRule: matchedRule.pathPrefix || '(default)',
};
}
/**
* Generate a YAML frontmatter block from inferred fields.
* Returns the `---\n...\n---\n` string to prepend to content.
*/
export function serializeFrontmatter(fm: InferredFrontmatter): string {
if (fm.skipped) return '';
const lines: string[] = ['---'];
// Title — quote if it contains special YAML chars
const needsQuote = /[:"'#\[\]{}|>&*!?,]/.test(fm.title);
lines.push(`title: ${needsQuote ? JSON.stringify(fm.title) : fm.title}`);
lines.push(`type: ${fm.type}`);
if (fm.date) {
lines.push(`date: "${fm.date}"`);
}
if (fm.source) {
lines.push(`source: ${fm.source}`);
}
if (fm.tags && fm.tags.length > 0) {
lines.push(`tags: [${fm.tags.map(t => JSON.stringify(t)).join(', ')}]`);
}
lines.push('---');
return lines.join('\n') + '\n';
}
/**
* Apply frontmatter inference to file content.
* Returns the content with frontmatter prepended, or the original content if it already has frontmatter.
*/
export function applyInference(relativePath: string, content: string): { content: string; inferred: InferredFrontmatter } {
const inferred = inferFrontmatter(relativePath, content);
if (inferred.skipped) {
return { content, inferred };
}
const fm = serializeFrontmatter(inferred);
return { content: fm + '\n' + content, inferred };
}
+2 -16
View File
@@ -339,7 +339,7 @@ export async function importFromFile(
engine: BrainEngine,
filePath: string,
relativePath: string,
opts: { noEmbed?: boolean; inferFrontmatter?: boolean } = {},
opts: { noEmbed?: boolean } = {},
): Promise<ImportResult> {
// Defense-in-depth: reject symlinks before reading content.
const lstat = lstatSync(filePath);
@@ -352,27 +352,13 @@ export async function importFromFile(
return { slug: relativePath, status: 'skipped', chunks: 0, error: `File too large (${stat.size} bytes)` };
}
let content = readFileSync(filePath, 'utf-8');
const content = readFileSync(filePath, 'utf-8');
// Route code files through the code import path
if (isCodeFilePath(relativePath)) {
return importCodeFile(engine, relativePath, content, opts);
}
// v0.22.8 — Frontmatter inference: if the file has no frontmatter and
// inference is enabled, synthesize it from the filesystem path + content.
// This turns bare markdown files into fully-typed, dated, tagged pages
// without requiring the user to manually add YAML headers.
// The inference is applied to the in-memory content only; the file on disk
// is not modified. Use `gbrain frontmatter generate --fix` to write back.
if (opts.inferFrontmatter !== false) {
const { applyInference } = await import('./frontmatter-inference.ts');
const { content: inferred, inferred: meta } = applyInference(relativePath, content);
if (!meta.skipped) {
content = inferred;
}
}
const parsed = parseMarkdown(content, relativePath);
// Enforce path-authoritative slug. parseMarkdown prefers frontmatter.slug over
+2 -11
View File
@@ -225,12 +225,8 @@ export class MinionSupervisor {
process.on('SIGTERM', this.sigtermListener);
process.on('SIGINT', this.sigintListener);
// 4. Health monitoring. Skip when healthInterval=0 — that's the explicit
// "disable" contract documented on `--health-interval 0`. setInterval(0)
// would be a tight DB-hammering loop, not the no-op users expect.
if (this.opts.healthInterval > 0) {
this.healthTimer = setInterval(() => { void this.healthCheck(); }, this.opts.healthInterval);
}
// 4. Health monitoring.
this.healthTimer = setInterval(() => { void this.healthCheck(); }, this.opts.healthInterval);
// 5. Announce start.
this.emit('started', {
@@ -431,11 +427,6 @@ export class MinionSupervisor {
} else {
delete env.GBRAIN_ALLOW_SHELL_JOBS;
}
// Signal to the child worker that it's running under a supervisor.
// The worker's self-health-check (DB probes, stall detection) is
// redundant when the supervisor already provides these — setting
// this env var causes the worker to skip its own health timer.
env.GBRAIN_SUPERVISED = '1';
this.lastStartTime = Date.now();
-19
View File
@@ -170,25 +170,6 @@ export interface MinionWorkerOpts {
* case where all concurrency slots are wedged with zero job completions
* so the per-job check never fires. */
rssCheckInterval?: number;
/** Self-health-check interval in ms. 0 = disabled. Default: 60000 (1 minute).
* Automatically disabled when running under a supervisor (GBRAIN_SUPERVISED=1).
* Provides DB liveness probes and stall detection for bare `gbrain jobs work`
* deployments managed by external process managers (systemd, Docker, cron). */
healthCheckInterval?: number;
/** Stall detection: ms of continuous idle (waiting>0, inFlight=0, no completions)
* before emitting the first warning. Default: 300000 (5 minutes). */
stallWarnAfterMs?: number;
/** Stall detection: ms of continuous idle before emitting `'unhealthy'` with
* reason='stalled'. Default: 600000 (10 minutes). Must be > stallWarnAfterMs. */
stallExitAfterMs?: number;
/** DB liveness probe: number of consecutive failed `SELECT 1` probes before
* emitting `'unhealthy'` with reason='db_dead'. Default: 3. */
dbFailExitAfter?: number;
/** Per-probe wall-clock timeout in ms. A `SELECT 1` that hangs longer than
* this counts as a failure (fed into dbFailExitAfter). Without this, a
* hung probe would wedge the recursive setTimeout chain forever and
* silently disable the health monitor. Default: 10000 (10 seconds). */
dbProbeTimeoutMs?: number;
}
// --- Job Context (passed to handlers) ---
+1 -209
View File
@@ -20,15 +20,8 @@ import { UnrecoverableError } from './types.ts';
import { MinionQueue } from './queue.ts';
import { calculateBackoff } from './backoff.ts';
import { randomUUID } from 'crypto';
import { EventEmitter } from 'events';
import { evaluateQuietHours, type QuietHoursConfig } from './quiet-hours.ts';
/** Reason payload emitted with `'unhealthy'` when self-health-check trips.
* CLI layer (jobs.ts:work) subscribes and decides whether to call process.exit. */
export type UnhealthyReason =
| { reason: 'db_dead'; consecutiveFailures: number; message: string }
| { reason: 'stalled'; waitingCount: number; idleMinutes: number };
/**
* Read the quiet_hours JSONB column off a MinionJob, if present. The
* column was added in schema migration v12; older rows + versions of
@@ -49,13 +42,7 @@ interface InFlightJob {
promise: Promise<void>;
}
/** Type-safe `on('unhealthy', ...)` for callers. */
export interface MinionWorker {
on(event: 'unhealthy', listener: (info: UnhealthyReason) => void): this;
emit(event: 'unhealthy', info: UnhealthyReason): boolean;
}
export class MinionWorker extends EventEmitter {
export class MinionWorker {
private queue: MinionQueue;
private handlers = new Map<string, MinionHandler>();
private running = false;
@@ -80,7 +67,6 @@ export class MinionWorker extends EventEmitter {
private engine: BrainEngine,
opts?: MinionWorkerOpts & MinionQueueOpts,
) {
super();
this.queue = new MinionQueue(engine, {
maxSpawnDepth: opts?.maxSpawnDepth,
maxAttachmentBytes: opts?.maxAttachmentBytes,
@@ -95,25 +81,7 @@ export class MinionWorker extends EventEmitter {
maxRssMb: opts?.maxRssMb ?? 0,
getRss: opts?.getRss ?? (() => process.memoryUsage().rss),
rssCheckInterval: opts?.rssCheckInterval ?? 60000,
healthCheckInterval: opts?.healthCheckInterval ?? 60000,
stallWarnAfterMs: opts?.stallWarnAfterMs ?? 5 * 60_000,
stallExitAfterMs: opts?.stallExitAfterMs ?? 10 * 60_000,
dbFailExitAfter: opts?.dbFailExitAfter ?? 3,
dbProbeTimeoutMs: opts?.dbProbeTimeoutMs ?? 10_000,
};
// Stall thresholds contract: exit MUST be strictly greater than warn.
// If exit <= warn, the warn-then-exit semantics break: a single tick at
// idle > warn would set stallWarningSince and the subsequent tick at
// idle > exit could fire immediately without giving operators visibility.
// Reject misconfigurations at construction time so the failure mode is
// a loud throw on startup rather than a quiet contract violation.
if (this.opts.stallExitAfterMs <= this.opts.stallWarnAfterMs) {
throw new Error(
`MinionWorkerOpts: stallExitAfterMs (${this.opts.stallExitAfterMs}) must be > ` +
`stallWarnAfterMs (${this.opts.stallWarnAfterMs}). ` +
`The contract is "warn first, exit later" — they cannot fire on the same tick.`,
);
}
}
/** Register a handler for a job type. */
@@ -126,28 +94,6 @@ export class MinionWorker extends EventEmitter {
return Array.from(this.handlers.keys());
}
/** Emit 'unhealthy' with a no-listener fallback. The default contract is
* fail-stop: pre-EventEmitter-refactor behavior was process.exit(1) inside
* the timer; the refactor moved that responsibility to the CLI subscriber.
* But direct API consumers without a listener would see emit() become a
* no-op AND `healthExited=true` permanently disabling monitoring a
* silent regression. Solution: if no one subscribed, log and exit
* ourselves so the worker dies and the PM restarts it. Subscribers
* override this default by adding a listener before start(). */
private emitUnhealthy(info: UnhealthyReason): void {
if (this.listenerCount('unhealthy') === 0) {
const detail = info.reason === 'db_dead'
? `DB unreachable (${info.consecutiveFailures} probes): ${info.message}`
: `worker stalled (${info.waitingCount} waiting, ${info.idleMinutes}m idle)`;
console.error(
`[health] FATAL: ${detail}. No 'unhealthy' listener registered; ` +
`defaulting to process.exit(1) for process-manager restart.`,
);
process.exit(1);
}
this.emit('unhealthy', info);
}
/** Start the worker loop. Blocks until stopped. */
async start(): Promise<void> {
if (this.handlers.size === 0) {
@@ -209,159 +155,6 @@ export class MinionWorker extends EventEmitter {
}, this.opts.rssCheckInterval);
}
// Self-health-check — provides supervisor-grade monitoring for bare workers.
// Disabled when running under a supervisor (GBRAIN_SUPERVISED=1) or when
// healthCheckInterval is 0. Catches two failure modes that leave the process
// alive but non-functional:
// 1. DB connection death (Supabase/PgBouncer drops, network blip)
// 2. Worker stall (event loop alive but not claiming/completing jobs)
//
// On failure, emits an `'unhealthy'` event with a structured reason. The
// CLI layer (`src/commands/jobs.ts:work`) subscribes and decides whether to
// call process.exit. Library code never calls process.exit directly so
// MinionWorker stays embeddable in non-CLI contexts (tests, other hosts).
//
// Timer pattern: recursive setTimeout with a `running` flag, not setInterval.
// setInterval queues callbacks even when the prior is still awaiting; on a
// hung DB probe that piles up overlapping async checks racing on
// `consecutiveDbFailures`. The recursive pattern guarantees one tick at a time.
const isSupervisedChild = process.env.GBRAIN_SUPERVISED === '1';
let healthTimer: ReturnType<typeof setTimeout> | null = null;
if (!isSupervisedChild && this.opts.healthCheckInterval > 0) {
let consecutiveDbFailures = 0;
let lastKnownCompleted = this.jobsCompleted;
let lastCompletionTime = Date.now();
let stallWarningSince: number | null = null;
let healthRunning = false;
let healthExited = false;
// Race executeRaw against a wall-clock deadline. A hung connection
// (network-partitioned PgBouncer, deadlocked backend) would otherwise
// hold the await forever — the recursive setTimeout's next tick is only
// scheduled in `finally`, so a hung probe would silently disable the
// entire health monitor. The timeout treats hangs as failures and feeds
// them into `dbFailExitAfter`.
const probeWithTimeout = async (): Promise<void> => {
const ac = new AbortController();
const timeoutMs = this.opts.dbProbeTimeoutMs;
const timer = setTimeout(() => ac.abort(), timeoutMs);
try {
await Promise.race([
this.engine.executeRaw('SELECT 1'),
new Promise<never>((_, reject) => {
ac.signal.addEventListener('abort', () => {
reject(new Error(`probe timeout after ${timeoutMs}ms`));
});
}),
]);
} finally {
clearTimeout(timer);
}
};
const runHealthCheck = async (): Promise<void> => {
if (healthRunning || !this.running || healthExited) return;
healthRunning = true;
try {
// --- 1. DB liveness probe ---
try {
await probeWithTimeout();
consecutiveDbFailures = 0;
} catch (e) {
consecutiveDbFailures++;
const msg = e instanceof Error ? e.message : String(e);
console.error(
`[health] DB probe failed (${consecutiveDbFailures}/${this.opts.dbFailExitAfter}): ${msg}`,
);
if (consecutiveDbFailures >= this.opts.dbFailExitAfter) {
console.error(
`[health] DB unreachable after ${this.opts.dbFailExitAfter} consecutive probes. ` +
`Emitting 'unhealthy' for process-manager restart.`,
);
healthExited = true;
this.emitUnhealthy({
reason: 'db_dead',
consecutiveFailures: consecutiveDbFailures,
message: msg,
});
}
return; // Skip stall check when DB is flaky
}
// --- 2. Stall detection ---
if (this.jobsCompleted > lastKnownCompleted) {
lastKnownCompleted = this.jobsCompleted;
lastCompletionTime = Date.now();
stallWarningSince = null;
}
const idleMs = Date.now() - lastCompletionTime;
// Only check for stalls when no jobs are in-flight and it's been a while
if (idleMs > this.opts.stallWarnAfterMs && this.inFlight.size === 0) {
try {
// Filter by registered handler names so a worker that doesn't
// claim a particular job-name doesn't false-positive when those
// jobs accumulate in `waiting`. Only counts work THIS worker would
// actually have claimed.
const handlerNames = this.registeredNames;
const rows = handlerNames.length === 0
? [] as { cnt: string }[]
: await this.engine.executeRaw<{ cnt: string }>(
`SELECT count(*)::text AS cnt FROM minion_jobs
WHERE status = 'waiting'
AND queue = $1
AND name = ANY($2::text[])`,
[this.opts.queue, handlerNames],
);
const waiting = parseInt(rows[0]?.cnt ?? '0', 10);
const idleMinutes = Math.round(idleMs / 60_000);
if (waiting > 0) {
// Two thresholds, both measured from `lastCompletionTime` (NOT
// from when the warning fired). With defaults (warn=5min,
// exit=10min), the first warning fires at idle=5min and the
// unhealthy emit fires at idle=10min — matching the contract
// documented in MinionWorkerOpts.
if (!stallWarningSince) {
stallWarningSince = Date.now();
console.warn(
`[health] Possible stall: ${waiting} waiting job(s) for ` +
`registered handlers, 0 in-flight, ${idleMinutes}m since last completion`,
);
} else if (idleMs > this.opts.stallExitAfterMs) {
console.error(
`[health] Worker stalled for ${Math.round(this.opts.stallExitAfterMs / 60_000)}+ ` +
`minutes with ${waiting} waiting job(s). Emitting 'unhealthy' for process-manager restart.`,
);
healthExited = true;
this.emitUnhealthy({
reason: 'stalled',
waitingCount: waiting,
idleMinutes,
});
}
} else {
stallWarningSince = null; // Queue empty (for our handlers) — not stalled, just idle
}
} catch {
// DB query failed — the liveness probe above will catch persistent failures
}
} else {
stallWarningSince = null;
}
} finally {
healthRunning = false;
if (this.running && !healthExited) {
healthTimer = setTimeout(runHealthCheck, this.opts.healthCheckInterval);
}
}
};
// First tick scheduled after one interval so newly-started workers have
// a chance to do real work before the stall clock starts ticking.
healthTimer = setTimeout(runHealthCheck, this.opts.healthCheckInterval);
}
try {
while (this.running) {
// Promote delayed jobs
@@ -408,7 +201,6 @@ export class MinionWorker extends EventEmitter {
} finally {
clearInterval(stalledTimer);
if (rssTimer) clearInterval(rssTimer);
if (healthTimer) clearTimeout(healthTimer); // recursive setTimeout pattern
process.removeListener('SIGTERM', shutdown);
process.removeListener('SIGINT', shutdown);
-283
View File
@@ -1,283 +0,0 @@
/**
* Tests for frontmatter-inference.ts the zero-friction ingest pipeline.
*
* Validates that files without frontmatter get correct type, title, date,
* source, and tags inferred from their filesystem path and content.
*/
import { describe, test, expect } from 'bun:test';
import {
inferFrontmatter,
extractDateFromFilename,
extractTitleFromFilename,
extractTitleFromHeading,
serializeFrontmatter,
applyInference,
DIRECTORY_RULES,
} from '../src/core/frontmatter-inference.ts';
// ── Date extraction ──────────────────────────────────────────────────
describe('extractDateFromFilename', () => {
test('extracts YYYY-MM-DD from date-prefixed filename', () => {
expect(extractDateFromFilename('2010-04-13 Apr 13 founders mtg.md')).toBe('2010-04-13');
});
test('extracts date with dash separator', () => {
expect(extractDateFromFilename('2024-01-30-therapy-session.md')).toBe('2024-01-30');
});
test('extracts date with underscore separator', () => {
expect(extractDateFromFilename('2023-06-15_meeting-notes.md')).toBe('2023-06-15');
});
test('returns null for no-date filename', () => {
expect(extractDateFromFilename('README.md')).toBe(null);
});
test('returns null for filename with numbers but no date', () => {
expect(extractDateFromFilename('chapter-1-intro.md')).toBe(null);
});
});
// ── Title extraction ─────────────────────────────────────────────────
describe('extractTitleFromFilename', () => {
test('strips date prefix and cleans up', () => {
expect(extractTitleFromFilename('2010-04-13 Apr 13 founders mtg.md')).toBe('Apr 13 founders mtg');
});
test('strips YYYY-MM-DD- prefix', () => {
expect(extractTitleFromFilename('2024-01-30-therapy-session.md')).toBe('Therapy Session');
});
test('handles filename without date', () => {
expect(extractTitleFromFilename('cognitive-distortions.md')).toBe('Cognitive Distortions');
});
test('preserves mixed case', () => {
expect(extractTitleFromFilename('YC presidency.md')).toBe('YC presidency');
});
test('returns Untitled for empty result', () => {
expect(extractTitleFromFilename('.md')).toBe('Untitled');
});
});
describe('extractTitleFromHeading', () => {
test('extracts first # heading', () => {
expect(extractTitleFromHeading('# Dhravya Shah\n\n> Founder of Supermemory')).toBe('Dhravya Shah');
});
test('ignores ## headings', () => {
expect(extractTitleFromHeading('Some text\n## Not this\n# This one')).toBe('This one');
});
test('returns null when no heading found', () => {
expect(extractTitleFromHeading('Just some text\nwithout headings')).toBe(null);
});
test('looks within first 20 lines only', () => {
const lines = Array(25).fill('text').join('\n') + '\n# Too Late';
expect(extractTitleFromHeading(lines)).toBe(null);
});
});
// ── Core inference ───────────────────────────────────────────────────
describe('inferFrontmatter', () => {
test('skips files that already have frontmatter', () => {
const result = inferFrontmatter('people/alice.md', '---\ntitle: Alice\n---\n# Alice');
expect(result.skipped).toBe(true);
});
test('Apple Notes: infers type, date, title, source', () => {
const result = inferFrontmatter(
'Apple Notes/2010-04-13 Apr 13 founders mtg.md',
'<span style="color:#000ff;">Top priority</span>',
);
expect(result.type).toBe('apple-note');
expect(result.date).toBe('2010-04-13');
expect(result.title).toBe('Apr 13 founders mtg');
expect(result.source).toBe('apple-notes');
});
test('Apple Notes/YC: adds yc tag', () => {
const result = inferFrontmatter(
'Apple Notes/YC/2022-08-04 Project 1783Y.md',
'Some content',
);
expect(result.type).toBe('apple-note');
expect(result.tags).toContain('yc');
expect(result.date).toBe('2022-08-04');
});
test('Apple Notes/Politics: adds politics tag', () => {
const result = inferFrontmatter(
'Apple Notes/Politics/2023-11-15 DA race notes.md',
'Some content',
);
expect(result.tags).toContain('politics');
});
test('people/ directory: type person, title from heading', () => {
const result = inferFrontmatter(
'people/dhravya-shah.md',
'# Dhravya Shah\n\n> Founder of Supermemory',
);
expect(result.type).toBe('person');
expect(result.title).toBe('Dhravya Shah');
});
test('people/ directory: falls back to filename when no heading', () => {
const result = inferFrontmatter(
'people/john-doe.md',
'Some text without a heading',
);
expect(result.type).toBe('person');
expect(result.title).toBe('John Doe');
});
test('personal/therapy: infers therapy-session type with date', () => {
const result = inferFrontmatter(
'personal/therapy/jan/2024-01-30.md',
'Session notes...',
);
expect(result.type).toBe('therapy-session');
expect(result.date).toBe('2024-01-30');
expect(result.source).toBe('therapy');
});
test('personal/reflections: infers reflection type, title from heading', () => {
const result = inferFrontmatter(
'personal/reflections/cognitive-distortions.md',
'# Cognitive Distortions\n\nA list of common...',
);
expect(result.type).toBe('reflection');
expect(result.title).toBe('Cognitive Distortions');
});
test('writing/essays: infers essay type', () => {
const result = inferFrontmatter(
'writing/essays/2024-03-15-on-being-remembered.md',
'# On Being Remembered Forever\n\nSome thoughts...',
);
expect(result.type).toBe('essay');
expect(result.title).toBe('On Being Remembered Forever');
expect(result.date).toBe('2024-03-15');
});
test('daily/calendar: infers calendar-index type', () => {
const result = inferFrontmatter(
'daily/calendar/2026-01-15-yc-office-hours.md',
'# Calendar Index\nSome calendar data',
);
expect(result.type).toBe('calendar-index');
expect(result.source).toBe('calendar');
});
test('companies/ directory: type company', () => {
const result = inferFrontmatter(
'companies/stripe.md',
'# Stripe\n\n> Online payments infrastructure',
);
expect(result.type).toBe('company');
expect(result.title).toBe('Stripe');
});
test('unknown directory: defaults to note type with heading title', () => {
const result = inferFrontmatter(
'random/some-file.md',
'# My Random Notes\n\nStuff here',
);
expect(result.type).toBe('note');
expect(result.title).toBe('My Random Notes');
});
test('handles empty content', () => {
const result = inferFrontmatter('notes/empty.md', '');
expect(result.type).toBe('note');
expect(result.title).toBe('Empty');
});
});
// ── Serialization ────────────────────────────────────────────────────
describe('serializeFrontmatter', () => {
test('generates valid YAML frontmatter', () => {
const fm = serializeFrontmatter({
title: 'Apr 13 founders mtg',
type: 'apple-note',
date: '2010-04-13',
source: 'apple-notes',
tags: ['yc'],
});
expect(fm).toContain('---');
expect(fm).toContain('title: Apr 13 founders mtg');
expect(fm).toContain('type: apple-note');
expect(fm).toContain('date: "2010-04-13"');
expect(fm).toContain('source: apple-notes');
expect(fm).toContain('tags: ["yc"]');
});
test('quotes title with special chars', () => {
const fm = serializeFrontmatter({
title: 'What\'s the deal: a "primer"',
type: 'note',
});
expect(fm).toContain('title: "What\'s the deal: a \\"primer\\""');
});
test('returns empty string for skipped files', () => {
expect(serializeFrontmatter({ title: '', type: '', skipped: true })).toBe('');
});
test('omits optional fields when absent', () => {
const fm = serializeFrontmatter({ title: 'Test', type: 'note' });
expect(fm).not.toContain('date');
expect(fm).not.toContain('source');
expect(fm).not.toContain('tags');
});
});
// ── Integration ──────────────────────────────────────────────────────
describe('applyInference', () => {
test('prepends frontmatter to content without it', () => {
const { content, inferred } = applyInference(
'people/alice-smith.md',
'# Alice Smith\n\n> Founder of FooBar',
);
expect(content).toMatch(/^---\n/);
expect(content).toContain('type: person');
expect(content).toContain('title: Alice Smith');
expect(content).toContain('# Alice Smith');
expect(inferred.skipped).toBeUndefined();
});
test('returns original content for files with frontmatter', () => {
const original = '---\ntitle: Bob\n---\n# Bob';
const { content, inferred } = applyInference('people/bob.md', original);
expect(content).toBe(original);
expect(inferred.skipped).toBe(true);
});
});
// ── Rules coverage ───────────────────────────────────────────────────
describe('DIRECTORY_RULES', () => {
test('has a catch-all rule with empty prefix', () => {
const catchAll = DIRECTORY_RULES.find(r => r.pathPrefix === '');
expect(catchAll).toBeDefined();
expect(catchAll!.type).toBe('note');
});
test('Apple Notes rules are more specific than the catch-all', () => {
const appleRules = DIRECTORY_RULES.filter(r => r.pathPrefix.startsWith('apple notes/'));
expect(appleRules.length).toBeGreaterThan(1); // subfolder rules + catch-all
// Subfolder rules should come before the generic apple notes/ rule
const ycIdx = DIRECTORY_RULES.findIndex(r => r.pathPrefix === 'apple notes/yc/');
const genericIdx = DIRECTORY_RULES.findIndex(r => r.pathPrefix === 'apple notes/');
expect(ycIdx).toBeLessThan(genericIdx);
});
});
-376
View File
@@ -2306,379 +2306,3 @@ describe('checkAborted (v0.20.5 cycle signal)', () => {
}).toThrow('aborted between phases: timeout');
});
});
// --- v0.22.14: Self-health-check for bare workers ---
describe('MinionWorker: self-health-check', () => {
test('health check is active when GBRAIN_SUPERVISED is not set', async () => {
// Save and clear the env var
const saved = process.env.GBRAIN_SUPERVISED;
delete process.env.GBRAIN_SUPERVISED;
try {
const worker = new MinionWorker(engine, {
queue: 'default',
concurrency: 1,
healthCheckInterval: 100, // fast for testing
pollInterval: 50,
stalledInterval: 10_000,
maxRssMb: 0,
});
worker.register('noop', async () => {});
await queue.add('noop', {});
const startPromise = worker.start();
// Let the health check fire at least once (100ms interval)
await new Promise(r => setTimeout(r, 300));
worker.stop();
await startPromise;
// Worker should have processed the job despite health check running
const completed = await queue.getJobs({ status: 'completed' });
expect(completed.length).toBeGreaterThanOrEqual(1);
} finally {
if (saved !== undefined) process.env.GBRAIN_SUPERVISED = saved;
else delete process.env.GBRAIN_SUPERVISED;
}
}, 10_000);
test('health check is skipped when GBRAIN_SUPERVISED=1', async () => {
const saved = process.env.GBRAIN_SUPERVISED;
process.env.GBRAIN_SUPERVISED = '1';
try {
const worker = new MinionWorker(engine, {
queue: 'default',
concurrency: 1,
healthCheckInterval: 100,
pollInterval: 50,
stalledInterval: 10_000,
maxRssMb: 0,
});
worker.register('noop', async () => {});
await queue.add('noop', {});
const startPromise = worker.start();
await new Promise(r => setTimeout(r, 300));
worker.stop();
await startPromise;
// Worker should still process jobs fine
const completed = await queue.getJobs({ status: 'completed' });
expect(completed.length).toBeGreaterThanOrEqual(1);
} finally {
if (saved !== undefined) process.env.GBRAIN_SUPERVISED = saved;
else delete process.env.GBRAIN_SUPERVISED;
}
}, 10_000);
test('healthCheckInterval=0 disables health check', async () => {
delete process.env.GBRAIN_SUPERVISED;
const worker = new MinionWorker(engine, {
queue: 'default',
concurrency: 1,
healthCheckInterval: 0,
pollInterval: 50,
stalledInterval: 10_000,
maxRssMb: 0,
});
worker.register('noop', async () => {});
await queue.add('noop', {});
const startPromise = worker.start();
await new Promise(r => setTimeout(r, 300));
worker.stop();
await startPromise;
const completed = await queue.getJobs({ status: 'completed' });
expect(completed.length).toBeGreaterThanOrEqual(1);
}, 10_000);
});
// --- v0.22.14: Self-health-check behavior tests (D7) ---
// These tests use a Proxy around the real engine so executeRaw can be
// intercepted by SQL pattern. SELECT 1 = liveness probe; the count(*) query
// = stall detection. Anything else passes through to the underlying engine.
interface ProbeOverrides {
/** When set, executeRaw('SELECT 1') uses this function instead of pass-through.
* Returning a thrown error simulates DB death; returning [{}] simulates success. */
selectOne?: () => Promise<unknown>;
/** When set, executeRaw of the stall-detection count(*) query returns this. */
countWaiting?: (handlers: string[]) => number;
/** Captures the last SQL string that matched the stall-count regex. Tests
* use this to assert the production SQL still contains `name = ANY(...)`
* so a future refactor that drops the predicate is caught. */
capturedStallSql?: { sql: string | null };
}
function makeProbeEngine(overrides: ProbeOverrides) {
return new Proxy(engine, {
get(target, prop, receiver) {
if (prop === 'executeRaw') {
return async (sql: string, params?: unknown[]): Promise<unknown[]> => {
if (overrides.selectOne && /^\s*SELECT\s+1\s*$/i.test(sql)) {
const r = await overrides.selectOne();
return Array.isArray(r) ? r : [r];
}
if (overrides.countWaiting && /count\(\*\).*minion_jobs.*WHERE\s+status\s*=\s*'waiting'/is.test(sql)) {
if (overrides.capturedStallSql) overrides.capturedStallSql.sql = sql;
const handlers = (params?.[1] as string[]) ?? [];
return [{ cnt: String(overrides.countWaiting(handlers)) }];
}
// Pass through to real engine for anything else (claim queries etc.)
return (target as unknown as { executeRaw: (s: string, p?: unknown[]) => Promise<unknown[]> })
.executeRaw(sql, params);
};
}
return Reflect.get(target, prop, receiver);
},
}) as unknown as PGLiteEngine;
}
describe('MinionWorker: self-health-check behavior (v0.22.14)', () => {
test('emits unhealthy{db_dead} after dbFailExitAfter consecutive DB probe failures', async () => {
delete process.env.GBRAIN_SUPERVISED;
let probeCount = 0;
const probeEngine = makeProbeEngine({
selectOne: async () => {
probeCount++;
throw new Error('connection terminated unexpectedly');
},
});
const worker = new MinionWorker(probeEngine, {
queue: 'default',
concurrency: 1,
healthCheckInterval: 30,
dbFailExitAfter: 3,
pollInterval: 50,
stalledInterval: 10_000,
maxRssMb: 0,
});
worker.register('noop', async () => {});
const events: Array<{ reason: string }> = [];
worker.on('unhealthy', (info) => { events.push(info); });
const startPromise = worker.start();
// 3 ticks at 30ms = 90ms; give extra slack.
await new Promise(r => setTimeout(r, 250));
worker.stop();
await startPromise;
expect(probeCount).toBeGreaterThanOrEqual(3);
expect(events.length).toBeGreaterThanOrEqual(1);
expect(events[0].reason).toBe('db_dead');
}, 10_000);
test('DB recovery resets the failure counter (no exit after intermittent failures)', async () => {
delete process.env.GBRAIN_SUPERVISED;
let probeCount = 0;
// Pattern: fail, fail, succeed (resets), fail, fail, then permanently succeed.
// No 3 consecutive failures, so dbFailExitAfter=3 must NOT trip.
const probeEngine = makeProbeEngine({
selectOne: async () => {
const idx = probeCount++;
if (idx === 0 || idx === 1 || idx === 3 || idx === 4) {
throw new Error('transient blip');
}
return [{ ok: 1 }];
},
});
const worker = new MinionWorker(probeEngine, {
queue: 'default',
concurrency: 1,
healthCheckInterval: 30,
dbFailExitAfter: 3,
pollInterval: 50,
stalledInterval: 10_000,
maxRssMb: 0,
});
worker.register('noop', async () => {});
const events: Array<{ reason: string }> = [];
worker.on('unhealthy', (info) => { events.push(info); });
const startPromise = worker.start();
await new Promise(r => setTimeout(r, 250));
worker.stop();
await startPromise;
// Counter should never have hit 3 consecutive — success at index 2 resets it.
const dbDeadEvents = events.filter(e => e.reason === 'db_dead');
expect(dbDeadEvents.length).toBe(0);
}, 10_000);
test('emits unhealthy{stalled} after stallExitAfterMs of continuous idle with waiting jobs', async () => {
delete process.env.GBRAIN_SUPERVISED;
const probeEngine = makeProbeEngine({
selectOne: async () => [{ ok: 1 }],
countWaiting: () => 5, // pretend 5 jobs are waiting for our handler names
});
const worker = new MinionWorker(probeEngine, {
queue: 'default',
concurrency: 1,
healthCheckInterval: 30,
stallWarnAfterMs: 50,
stallExitAfterMs: 100,
pollInterval: 50,
stalledInterval: 10_000,
maxRssMb: 0,
});
worker.register('noop', async () => {});
// Don't queue any real jobs — claim returns null, inFlight stays 0,
// jobsCompleted stays 0, idle clock advances.
const events: Array<{ reason: string; waitingCount?: number }> = [];
worker.on('unhealthy', (info) => { events.push(info); });
const startPromise = worker.start();
// Both thresholds measured from lastCompletionTime (corrected per codex r2):
// - tick @ +30ms: idle=30ms, < stallWarnAfterMs(50), no warn
// - tick @ +60ms: idle=60ms, > 50, warn fires (stallWarningSince set)
// - tick @ +90ms: idle=90ms, < stallExitAfterMs(100), no exit yet
// - tick @ +120ms: idle=120ms, > 100 → exit fires (unhealthy event)
// Wait 350ms which leaves comfortable slack for setTimeout drift.
await new Promise(r => setTimeout(r, 350));
worker.stop();
await startPromise;
const stalledEvents = events.filter(e => e.reason === 'stalled');
expect(stalledEvents.length).toBeGreaterThanOrEqual(1);
expect(stalledEvents[0].waitingCount).toBe(5);
// The idleMinutes payload should reflect total idle, not warn-since.
// With idle ~120ms at exit time, idleMinutes rounds to 0 — that's
// expected; the value is informative, not load-bearing.
}, 10_000);
test('inFlight > 0 blocks stall detection (long-running legitimate job)', async () => {
delete process.env.GBRAIN_SUPERVISED;
const probeEngine = makeProbeEngine({
selectOne: async () => [{ ok: 1 }],
countWaiting: () => 5,
});
const worker = new MinionWorker(probeEngine, {
queue: 'default',
concurrency: 1,
healthCheckInterval: 30,
stallWarnAfterMs: 50,
stallExitAfterMs: 100,
pollInterval: 50,
stalledInterval: 10_000,
maxRssMb: 0,
});
worker.register('noop', async () => {});
const events: Array<{ reason: string }> = [];
worker.on('unhealthy', (info) => { events.push(info); });
// Inject a fake in-flight entry directly. This bypasses the claim path
// (which goes through the proxy and complicates the cleanup race) and
// tests exactly what we want: the stall check's `inFlight.size === 0`
// gate when there's legitimate ongoing work.
const fakeInFlight = (worker as unknown as {
inFlight: Map<number, { lockTimer: NodeJS.Timeout; abort: AbortController; promise: Promise<void> }>
}).inFlight;
const fakeAbort = new AbortController();
const fakePromise = new Promise<void>(() => { /* never resolves */ });
const fakeTimer = setInterval(() => {}, 60_000); // dummy lock timer
fakeInFlight.set(99999, { lockTimer: fakeTimer, abort: fakeAbort, promise: fakePromise });
const startPromise = worker.start();
await new Promise(r => setTimeout(r, 350));
// Remove our fake entry before stop so the worker doesn't wait 30s for it.
clearInterval(fakeTimer);
fakeInFlight.delete(99999);
worker.stop();
await startPromise;
// No stall event should fire — inFlight.size > 0 gates the stall check.
const stalledEvents = events.filter(e => e.reason === 'stalled');
expect(stalledEvents.length).toBe(0);
}, 10_000);
test('regression (D1): waiting jobs of unregistered handler names do NOT trigger stall exit', async () => {
delete process.env.GBRAIN_SUPERVISED;
// The count(*) query is filtered by registered handler names. If handlers=['noop']
// and the queue has 5 'widget-fn' jobs, the SQL `name = ANY($2)` filter returns 0.
// The probe engine simulates this by checking handlers before returning a count;
// we ALSO capture the SQL to assert the predicate text is actually present (so a
// future refactor that silently drops `AND name = ANY(...)` is caught).
const capturedStallSql = { sql: null as string | null };
const probeEngine = makeProbeEngine({
selectOne: async () => [{ ok: 1 }],
countWaiting: (handlers) => handlers.includes('widget-fn') ? 5 : 0,
capturedStallSql,
});
const worker = new MinionWorker(probeEngine, {
queue: 'default',
concurrency: 1,
healthCheckInterval: 50,
stallWarnAfterMs: 100,
stallExitAfterMs: 200,
pollInterval: 50,
stalledInterval: 10_000,
maxRssMb: 0,
});
// Register 'noop' but pretend the queue is full of 'widget-fn' (unhandled).
worker.register('noop', async () => {});
const events: Array<{ reason: string }> = [];
worker.on('unhealthy', (info) => { events.push(info); });
const startPromise = worker.start();
// Window > stallExitAfterMs; if D1 fix wasn't applied, stall would fire.
await new Promise(r => setTimeout(r, 500));
worker.stop();
await startPromise;
// No stall event — the count for 'noop' handlers is 0, so worker is correctly idle.
const stalledEvents = events.filter(e => e.reason === 'stalled');
expect(stalledEvents.length).toBe(0);
// SQL shape assertion: the production query MUST filter by handler names.
// Without this assertion, a future change that drops the predicate would
// pass the no-event check above (the handler array would be irrelevant
// to the underlying DB but our probe just needs to return 0).
expect(capturedStallSql.sql).not.toBeNull();
expect(capturedStallSql.sql).toMatch(/name\s*=\s*ANY/i);
}, 10_000);
test('regression (R3): constructor throws when stallExitAfterMs <= stallWarnAfterMs', () => {
// The contract on MinionWorkerOpts.stallExitAfterMs says "Must be >
// stallWarnAfterMs". Without validation, an exit threshold equal to or
// less than the warn threshold made the configured exit time a lie
// (warn fires first, exit can't preempt). The constructor now throws
// loudly so misconfigurations fail at startup, not at idle-time.
expect(() => new MinionWorker(engine, {
stallWarnAfterMs: 200,
stallExitAfterMs: 100, // less than warn — invalid
})).toThrow(/stallExitAfterMs.*must be > stallWarnAfterMs/i);
expect(() => new MinionWorker(engine, {
stallWarnAfterMs: 100,
stallExitAfterMs: 100, // equal to warn — also invalid (must be strictly >)
})).toThrow(/stallExitAfterMs.*must be > stallWarnAfterMs/i);
// Sanity: defaults (5min warn / 10min exit) construct without throwing.
expect(() => new MinionWorker(engine, {})).not.toThrow();
});
});
-73
View File
@@ -328,79 +328,6 @@ describe('MinionSupervisor', () => {
}, 15_000);
});
describe('integration: GBRAIN_SUPERVISED env var (v0.22.14)', () => {
it('sets GBRAIN_SUPERVISED=1 on spawned worker child', async () => {
const outFile = join(tmpdir(), `gbrain-sup-supervised-${process.pid}-${Date.now()}.txt`);
try { unlinkSync(outFile); } catch { /* may not exist */ }
const h = makeHarness('supervised-env', `printf '%s\n' "\${GBRAIN_SUPERVISED-UNSET}" > "$OUT_FILE" ; exit 0`);
try {
const sup = spawnSupervisor(h, {
OUT_FILE: outFile,
SUP_MAX_CRASHES: '1',
});
await sup.exited;
expect(existsSync(outFile)).toBe(true);
const childSawEnv = readFileSync(outFile, 'utf8').trim();
expect(childSawEnv).toBe('1');
} finally {
try { unlinkSync(outFile); } catch { /* noop */ }
h.cleanup();
}
}, 15_000);
});
describe('regression (R3): healthInterval=0 disables timer (v0.22.14)', () => {
// Pre-fix: supervisor unconditionally called setInterval(callback, 0),
// which schedules a tight loop on the next event-loop tick. The
// operator-facing CLI claim "Use 0 to disable" was a lie — passing 0
// produced a DB-probe loop that hammered Postgres.
//
// Post-fix: setInterval is gated on healthInterval > 0. With 0, the
// supervisor runs its supervise loop normally with the health timer
// entirely absent.
//
// Assertion strategy: spawn the supervisor with SUP_HEALTH_INTERVAL_MS=0,
// a fast worker that exits cleanly, and SUP_MAX_CRASHES=1. A working fix
// should produce a single worker spawn → exit → supervisor shutdown
// sequence. If the tight-loop bug returned, the supervisor would still
// exit (max-crashes path) but the audit trail would show the tell-tale
// signature of an extremely high health-check call rate during the brief
// window before max-crashes fires. We assert the basic completion path
// and let CI's wall-clock detect any pathological CPU spike.
it('completes a normal supervise lifecycle with healthInterval=0', async () => {
const h = makeHarness('health-interval-zero', 'exit 0');
try {
const sup = spawnSupervisor(h, {
SUP_HEALTH_INTERVAL_MS: '0',
SUP_MAX_CRASHES: '1',
});
const start = Date.now();
const { code } = await sup.exited;
const elapsedMs = Date.now() - start;
// Clean exit (max-crashes path returns 1; this is fine — we just
// want to confirm the supervisor reached its terminal state without
// hanging or runaway looping).
expect(code).toBe(1);
// Sanity: a tight loop on setInterval(0) plus the spawn-respawn
// loop would still terminate at max-crashes, but it would be
// measurably slower than a clean run because the event loop is
// saturated with health-check callbacks. Cap the upper bound at
// 10s — clean runs typically finish in 12s.
expect(elapsedMs).toBeLessThan(10_000);
} finally {
h.cleanup();
}
}, 15_000);
});
describe('integration: --max-rss spawn args (v0.21)', () => {
it('passes --max-rss 2048 to spawned worker by default', async () => {
const outFile = join(tmpdir(), `gbrain-sup-maxrss-${process.pid}-${Date.now()}.txt`);