mirror of
https://github.com/garrytan/gbrain.git
synced 2026-08-15 01:12:20 +00:00
Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Garry Tan
|
6776df2447 | ||
|
root
|
19dbf152c2 | ||
|
root
|
0c48010d79 | ||
|
|
d838d4792b | ||
|
|
e3f704229b |
+206
@@ -2,6 +2,212 @@
|
||||
|
||||
All notable changes to GBrain will be documented in this file.
|
||||
|
||||
## [0.20.3] - 2026-04-24
|
||||
|
||||
## **Your queue now rescues itself when a wedged worker holds a row lock. Wall-clock sweep kills the job that stall detection can't see.**
|
||||
## **`maxWaiting` is race-proof, observable, and reachable from the CLI — three bugs in one patch.**
|
||||
|
||||
A production autopilot-cycle job wedged for over an hour on a single OpenClaw deployment because the worker's handler got stuck mid-transaction holding a row lock. Both eviction paths were blocked: the stall detector's `FOR UPDATE SKIP LOCKED` pass skipped the row-locked candidate, and the timeout sweep's `lock_until > now()` predicate disqualified the job once lock-renewal had been blocked. Neither could see the job. The shell-job pipeline starved completely behind the wedge.
|
||||
|
||||
v0.19.0 shipped the wall-clock sweep as the third-layer kill shot: drop both constraints, evict on `started_at` alone, worst case at `2 × timeout_ms + stalledInterval`. This release locks down three correctness holes the v0.19.0 PR introduced — then closes the observability gap that let the incident run to minute 90 in the first place.
|
||||
|
||||
### The queue-resilience numbers that matter
|
||||
|
||||
Measured against the real incident on 2026-04-23 (OpenClaw autopilot + shell-job pipeline, Postgres engine, concurrency=1 worker).
|
||||
|
||||
| Behavior | Before v0.20.3 | After v0.20.3 |
|
||||
|---|---|---|
|
||||
| Wedged worker escape window | 90+ minutes (manual kill) | `~2 × timeout_ms + 30s` sweep interval |
|
||||
| Per-name waiting pile during wedge | 18 deferred per-slot jobs | capped at `maxWaiting` |
|
||||
| `maxWaiting` under concurrent submit (2 submitters, cap=2) | up to 3 rows (TOCTOU race) | exactly 2 rows (advisory-lock serialization) |
|
||||
| Same name across queues | cross-queue bleed — `shell` suppressed by `default` | isolated per `(name, queue)` |
|
||||
| `GBRAIN_WORKER_CONCURRENCY=foo` | silent wedge (`inFlight < NaN` false) | clamped to 1, loud stderr warning |
|
||||
| `gbrain jobs submit --max-waiting 2` | flag didn't exist | wired through to MinionJobInput |
|
||||
| Silent coalesce events | invisible | JSONL audit at `~/.gbrain/audit/backpressure-YYYY-Www.jsonl` |
|
||||
| `gbrain doctor` visibility into wedge | no check | new `queue_health` with 2 subchecks |
|
||||
|
||||
The two big shifts: (1) every silent-failure vector the v0.19.0 patches introduced now has a loud signal — JSONL audit files, doctor check, stderr warnings, peer-liveness probe. (2) `maxWaiting` is now actually a cap under concurrency, not a soft suggestion. A future multi-submitter pattern (parallel workspaces, dispatched children, OpenClaw + ycli cron) doesn't walk through it.
|
||||
|
||||
### What this means for OpenClaw users
|
||||
|
||||
If you're running `gbrain autopilot` on a daily-driver deployment, the wall-clock sweep is the difference between a 90-minute outage and a 30-second one. The `queue_health` doctor check means the next time your queue wedges, you notice in minute 2 instead of minute 90. If you've been writing programmatic Minion submitters and setting `maxWaiting`, it's worth re-reading the JSONL audit file the next time your agent does anything "interesting" — you'll see exactly which submission coalesces into which returned job.
|
||||
|
||||
## To take advantage of v0.20.3
|
||||
|
||||
`gbrain upgrade` handles the binary. You MUST restart long-running worker daemons so the new sweep runs in-process — the wall-clock eviction is a method on `MinionQueue`, not a cron job, so it only fires inside a worker loop.
|
||||
|
||||
1. **Upgrade the binary:**
|
||||
```bash
|
||||
gbrain upgrade
|
||||
```
|
||||
2. **Restart autopilot + workers:**
|
||||
```bash
|
||||
# systemd / launchd / OpenClaw service-manager: restart the unit.
|
||||
# Manual: kill the old `gbrain autopilot` and `gbrain jobs work`, start new ones.
|
||||
```
|
||||
3. **Verify:**
|
||||
```bash
|
||||
gbrain jobs smoke --wedge-rescue # exercises the new wall-clock path
|
||||
gbrain doctor --json | jq '.checks[] | select(.name == "queue_health")'
|
||||
```
|
||||
4. **If `gbrain doctor` flags anything unexpected,** please file an issue:
|
||||
https://github.com/garrytan/gbrain/issues with:
|
||||
- output of `gbrain doctor`
|
||||
- contents of `~/.gbrain/audit/backpressure-*.jsonl` (redact freely)
|
||||
- what commands you ran leading up to the wedge
|
||||
|
||||
### Itemized changes
|
||||
|
||||
**Queue core** (`src/core/minions/queue.ts`)
|
||||
- `maxWaiting` coalesce path wraps `count → select → insert` in `pg_advisory_xact_lock` keyed on `(name, queue)`. Concurrent submitters for the SAME key serialize; different keys stay parallel. Lock auto-releases on transaction commit/rollback — no cleanup path to leak. Fixes TOCTOU race caught by adversarial review.
|
||||
- `maxWaiting` count and select now filter on `queue` in addition to `name`. Pre-v0.20.3 code filtered on name alone, so a waiting `autopilot-cycle` in `queue=default` would suppress submissions to `queue=shell` with the same name. Cross-queue bleed is gone.
|
||||
|
||||
**Backpressure observability** (new `src/core/minions/backpressure-audit.ts`)
|
||||
- Every coalesce event writes one JSONL line to `~/.gbrain/audit/backpressure-YYYY-Www.jsonl` (ISO-week rotation, override dir via `GBRAIN_AUDIT_DIR`, mirrors the v0.14 shell-audit pattern).
|
||||
- Fields: `ts, queue, name, waiting_count, max_waiting, decision='coalesced', returned_job_id`.
|
||||
- Best-effort: write failures log to stderr but never block submission.
|
||||
|
||||
**CLI** (`src/commands/jobs.ts`)
|
||||
- New `--max-waiting N` flag on `gbrain jobs submit`. Clamps to `[1, 100]`, mirrors the existing `--max-stalled` wiring. The `MinionJobInput.maxWaiting` field was programmatic-only before; now it's reachable from the command line too.
|
||||
- `resolveWorkerConcurrency` clamps against invalid input. `parseInt` returns `NaN` for `"foo"`, `0` for `"0"`, negatives for `"-5"` — all of which silently wedge a worker (`inFlight.size < NaN/0/negative` is always false). Now clamped to ≥1 with a loud stderr warning naming the bad value. One typo in a systemd unit no longer reproduces the 90-minute outage.
|
||||
- New `gbrain jobs smoke --wedge-rescue` opt-in case. Forges a wedged-worker row state, invokes `handleStalled` + `handleTimeouts` + `handleWallClockTimeouts` in sequence, asserts only the wall-clock sweep evicts. Mirrors the v0.14.3 `--sigkill-rescue` shape.
|
||||
|
||||
**Doctor** (`src/commands/doctor.ts`)
|
||||
- New `queue_health` check (Postgres-only; PGLite skips with `Skipped (PGLite — no multi-process worker surface)`).
|
||||
- Subcheck 1 — **stalled-forever**: flags active jobs whose `started_at` is older than 1 hour. Reports the top 5 by start time with `gbrain jobs get/cancel <id>` fix hints.
|
||||
- Subcheck 2 — **waiting-depth**: flags per-name queues whose waiting count exceeds threshold. Default 10, overridable via `GBRAIN_QUEUE_WAITING_THRESHOLD` env. Reports the top 5 by depth with "consider setting maxWaiting on the submitter" fix hint.
|
||||
- Worker-heartbeat staleness subcheck intentionally deferred to follow-up because `lock_until`-on-active-jobs is a lossy proxy. A check that cries wolf erodes trust in every other doctor subcheck. Needs a `minion_workers` table to produce ground-truth signal.
|
||||
|
||||
**Autopilot** (`src/commands/autopilot.ts`)
|
||||
- `--no-worker` mode gains a peer-worker-liveness probe. Every cycle runs a cheap `SELECT count(*)` checking for active jobs with `lock_until` refreshed in the last 2 minutes. After 3 consecutive idle ticks, logs a loud `WARNING` naming the silent-wedge vector (`--no-worker` set but no worker running). Re-arms once a live signal returns, so a healthy-but-idle worker doesn't trigger spam.
|
||||
- Probe is documented as a proxy, not ground truth — idle worker with no active jobs reads as "no worker." The ground-truth fix needs a `minion_workers` heartbeat table (tracked as follow-up).
|
||||
|
||||
**Docs**
|
||||
- New `docs/guides/queue-operations-runbook.md`: the "my queue looks wedged — what do I run?" reference. One viewport, in order of escalation. What each `queue_health` subcheck means. Self-check for the `--no-worker + no-worker-running` footgun.
|
||||
- `CLAUDE.md` Key-files section updated for the new `handleWallClockTimeouts` method (v0.19.0, described here for the first time), the new `backpressure-audit.ts` module, the updated `maxWaiting` semantics, and the new `queue_health` doctor check.
|
||||
|
||||
**Tests** (`test/minions.test.ts`)
|
||||
- 23 new unit cases. Wall-clock sweep (3 cases + non-interference with `handleTimeouts`). `maxWaiting` (coalesce, clamp 0 → 1, floor 1.7 → 1, concurrent-submitter race via `Promise.all`, cross-queue isolation, unset fallthrough). Concurrency clamp (7 cases including `NaN`/`0`/negative). `parseMaxWaitingFlag` (5 cases). Backpressure audit file write. All 143 minions tests pass.
|
||||
- E2E wall-clock case against real Postgres is next on the roadmap (needs a second-connection row-lock helper; the unit-level coverage above exercises the sweep mechanics directly).
|
||||
|
||||
### For contributors
|
||||
|
||||
- The v0.19.0 PR's narrative framed the 18-job pileup as "duplicate submissions from a cron loop with no idempotency key." That framing was wrong. Autopilot already sets `idempotency_key: autopilot-cycle:${slot}` where slot is a 5-minute tick boundary — within-slot duplicates are structurally impossible. The 18 jobs were 18 different slots stacking up behind the wedged one. `maxWaiting` still caps the pile; the incident just wasn't about idempotency. Adversarial review caught this before v0.20.3 shipped.
|
||||
- Follow-up issues tracked: B2 (autopilot heartbeat file), B3 (doctor `--fix` learns queue rescue), B4 (backpressure counts surfaced in `jobs stats`), B5 (cross-cutting "health-delivery-agent" pattern), B7 (`minion_workers` heartbeat table — unblocks both the dropped `queue_health` subcheck and a ground-truth `--no-worker` probe), P1 (composite indexes `(status, started_at)` and `(status, name)` on `minion_jobs` — currently the new sweeps fall back to `idx_minion_jobs_status`, selective enough on healthy queues, worth tightening in v0.20.4).
|
||||
|
||||
Full plan with CEO + Eng + Codex adversarial decisions lives at `~/.claude/plans/` for the operators who care about how this release was reviewed.
|
||||
|
||||
## [0.20.2] - 2026-04-24
|
||||
|
||||
## **`gbrain jobs supervisor` is now a self-healing daemon you can actually drive. The Minions worker stops dying silently.**
|
||||
## **Three commands an agent can run: `start --detach`, `status --json`, `stop`. Crash loops are bounded, audit events are JSONL, and the health check finally reports real data.**
|
||||
|
||||
`gbrain jobs work` has always been the worker that drains your Minions queue. Problem: it dies (OOM, connection blip, panic) and nobody notices until jobs pile up. The old answer was `nohup` plus a 68-line bash watchdog script from the deployment guide, and it shipped its own bugs (restart-loop traps, log-parsing stall detection, zero audit trail).
|
||||
|
||||
v0.20.2 ships the replacement: `gbrain jobs supervisor` is a first-class CLI with atomic PID locking, exponential backoff, structured audit events at `~/.gbrain/audit/supervisor-YYYY-Www.jsonl`, and three subcommands that make it drivable by an OpenClaw or Hermes agent in three turns. The old bash watchdog is gone.
|
||||
|
||||
### The numbers that matter
|
||||
|
||||
Before v0.20.2, an agent driving the supervisor needed ~10 turns of shell archaeology (PID file scraping, `pgrep -f`, `kill -0`, log grep) just to start and stop the worker reliably. After v0.20.2, it's three commands with machine-parseable output.
|
||||
|
||||
| Capability | Before v0.20.2 | After v0.20.2 |
|
||||
|---|---|---|
|
||||
| Keeping the worker alive | `nohup` + `minion-watchdog.sh` (68 lines of bash, restart-loop bug, log-scrape health) | `gbrain jobs supervisor` (first-class CLI with atomic PID lock, exponential backoff, JSONL audit) |
|
||||
| PID file locking | `existsSync + readFileSync + writeFileSync` TOCTOU race | Atomic `O_CREAT|O_EXCL` via `openSync('wx')` — kernel-atomic mutex |
|
||||
| Stalled-jobs health alert | Queried `status='stalled'` — returned 0 rows forever (dead code) | Queries `status='active' AND lock_until < now()`, scoped to the supervised queue |
|
||||
| Shell-exec env inheritance | Child inherited `GBRAIN_ALLOW_SHELL_JOBS=1` from parent shell regardless of CLI flag | Explicit `else delete env.GBRAIN_ALLOW_SHELL_JOBS` when not opted in + regression test |
|
||||
| Agent discovery TTHW | ~10 turns of shell-scraping (cat PID / pgrep / kill -0 / log grep) | 3 turns: `start --detach` → `status --json` → `stop` |
|
||||
| Lifecycle observability | `console.log` with human prefixes, zero audit trail | JSONL events on stderr + `~/.gbrain/audit/supervisor-YYYY-Www.jsonl` + `gbrain doctor` integration |
|
||||
| Exit codes | undocumented; agent couldn't distinguish "already running" from "gave up" | Four documented codes: `0` clean, `1` max-crashes, `2` lock-held, `3` PID-unwritable |
|
||||
| Test coverage of the supervisor itself | ~15% (backoff math + PID helpers only) | Integration tests covering crash-restart, max-crashes drain, SIGTERM-during-backoff, env-inheritance regression |
|
||||
|
||||
The supervisor's own reliability claims are now testable. Every lifecycle event (`started`, `worker_spawned`, `worker_exited`, `backoff`, `health_warn`, `max_crashes_exceeded`, `shutting_down`, `stopped`, `worker_spawn_failed`) lands in a weekly-rotated JSONL file that `gbrain doctor` reads to surface a `supervisor` health check.
|
||||
|
||||
### What this means for your deployment
|
||||
|
||||
If you were using the old `nohup`/`minion-watchdog.sh` pattern:
|
||||
|
||||
1. **Stop the old watchdog:** `sudo kill $(head -n1 /tmp/gbrain-worker.pid) 2>/dev/null && crontab -e` and delete the watchdog cron line.
|
||||
2. **Delete the script:** `sudo rm -f /usr/local/bin/minion-watchdog.sh /tmp/gbrain-worker.pid /tmp/gbrain-worker.log`.
|
||||
3. **Start the supervisor:** `gbrain jobs supervisor start --detach --json` — or on systemd, reinstall the unit (now calls `gbrain jobs supervisor`).
|
||||
4. **Verify:** `gbrain doctor` reports a `supervisor` check; `gbrain jobs supervisor status --json` returns `running:true`.
|
||||
|
||||
For containers (Fly / Railway / Render / Heroku): the shipped `Procfile` and `fly.toml.partial` now call `gbrain jobs supervisor`. The platform restarts the container on host events, the supervisor restarts the worker on in-process crashes. Two-layer supervision with clean separation.
|
||||
|
||||
For OpenClaw / Hermes / Cursor agents driving the supervisor: you no longer need a shell skill to drive the worker. Every piece of state — liveness, crash history, max-crashes exhaustion — is a machine-parseable JSON response. Start with `gbrain jobs supervisor status --json | jq`.
|
||||
|
||||
## To take advantage of v0.20.2
|
||||
|
||||
`gbrain upgrade` pulls the binary. Nothing else is required if you're currently running `gbrain jobs work` directly or using systemd — the new supervisor is opt-in. To migrate:
|
||||
|
||||
1. **Verify the binary:**
|
||||
```bash
|
||||
gbrain --version # should say 0.20.2
|
||||
gbrain jobs supervisor --help | head -20
|
||||
```
|
||||
2. **Start the supervisor (detached, agent-friendly):**
|
||||
```bash
|
||||
gbrain jobs supervisor start --detach --json
|
||||
# → {"event":"started","supervisor_pid":1234,"pid_file":"/Users/you/.gbrain/supervisor.pid","detached":true}
|
||||
```
|
||||
3. **Check health:**
|
||||
```bash
|
||||
gbrain jobs supervisor status --json
|
||||
gbrain doctor | grep supervisor
|
||||
```
|
||||
4. **Stop when done:**
|
||||
```bash
|
||||
gbrain jobs supervisor stop
|
||||
```
|
||||
5. **(Optional) Migrate off the old watchdog:** see `docs/guides/minions-deployment.md` "Upgrading from an older deployment" for the cron-to-supervisor migration.
|
||||
|
||||
If `gbrain jobs supervisor status` reports `running:false` unexpectedly, or `gbrain doctor` flags a `supervisor` failure, file an issue at https://github.com/garrytan/gbrain/issues with:
|
||||
- output of `gbrain doctor`
|
||||
- the last ~50 lines of `~/.gbrain/audit/supervisor-*.jsonl`
|
||||
- which step broke
|
||||
|
||||
### Itemized changes
|
||||
|
||||
**`gbrain jobs supervisor`:**
|
||||
- New subcommands: `start [--detach] [--json]`, `status [--json]`, `stop [--json]`. Foreground use is unchanged (back-compat).
|
||||
- New flags: `--allow-shell-jobs` (explicit opt-in, replaces env-var sniffing), `--cli-path PATH` (override auto-resolution), `--json` (JSONL lifecycle events on stderr), `GBRAIN_SUPERVISOR_PID_FILE` env var (overrides default PID path).
|
||||
- Exit codes documented in `--help`: `0` clean, `1` max-crashes, `2` lock-held, `3` PID-unwritable.
|
||||
- Default PID path moved from `/tmp/gbrain-supervisor.pid` to `~/.gbrain/supervisor.pid` with automatic parent-directory creation.
|
||||
|
||||
**Safety fixes (codex adversarial review + eng review):**
|
||||
- Atomic PID lock via `openSync(path, 'wx')` — two supervisors starting simultaneously can no longer both win the race.
|
||||
- `stalled` health check query rewritten from unreachable `status='stalled'` to `status='active' AND lock_until < now()` matching `queue.ts:848 handleStalled()`.
|
||||
- Health queries now scoped to `WHERE queue = $1` — multi-queue deployments see the right queue.
|
||||
- Unified exit path via `shutdown(reason, exitCode)` — max-crashes drains gracefully instead of bypassing cleanup via `process.exit(1)`.
|
||||
- Listener ref tracking: `SIGTERM`/`SIGINT` handlers removed on shutdown for clean test lifecycle.
|
||||
|
||||
**Security hardening:**
|
||||
- `allowShellJobs` class default flipped `true` → `false`.
|
||||
- Child env now has `GBRAIN_ALLOW_SHELL_JOBS` explicitly deleted when `allowShellJobs:false` (was: silently inherited from parent shell).
|
||||
- Integration regression test locks this against future refactors.
|
||||
|
||||
**Observability:**
|
||||
- New `src/core/minions/handlers/supervisor-audit.ts` with ISO-week rotation (mirrors `shell-audit.ts` / `subagent-audit.ts` pattern).
|
||||
- Every supervisor emission (started, worker_spawned, worker_exited, worker_spawn_failed, backoff, health_warn, health_error, max_crashes_exceeded, shutting_down, stopped) written to `~/.gbrain/audit/supervisor-YYYY-Www.jsonl`.
|
||||
- `gbrain doctor` gains a `supervisor` check that reads the audit file and reports `running` / `last_start` / `crashes_24h` / `max_crashes_exceeded` with thresholds (ok / warn at 3+ crashes / fail on max-crashes event).
|
||||
|
||||
**Documentation:**
|
||||
- `docs/guides/minions-deployment.md` rewritten: supervisor is the canonical answer; which-supervisor-when decision table (container / systemd / dev laptop); three-command agent pattern; migration block from the old watchdog.
|
||||
- `README.md` Operations section gains a paragraph on `gbrain jobs supervisor`.
|
||||
- `docs/guides/minions-deployment-snippets/{systemd.service,Procfile,fly.toml.partial}` now invoke `gbrain jobs supervisor` instead of raw `gbrain jobs work`.
|
||||
- `docs/guides/minions-deployment-snippets/minion-watchdog.sh` deleted — subsumed by the supervisor.
|
||||
|
||||
**Tests:**
|
||||
- `test/supervisor.test.ts`: 7 → 13 tests. Four new integration tests exercise real `spawn()` lifecycles via shell-script fakes (crash-restart happy path, max-crashes-via-shutdown with audit assertions, SIGTERM-during-backoff clean exit, `GBRAIN_ALLOW_SHELL_JOBS` inheritance regression — positive + negative).
|
||||
- `test/fixtures/supervisor-runner.ts`: new standalone runner that constructs a supervisor from env vars so integration tests can observe `process.exit` without killing the test runner.
|
||||
|
||||
**For contributors:**
|
||||
- The `MinionSupervisor` class has a test-only `_backoffFloorMs` override for fast crash-loop tests. Not exposed via CLI.
|
||||
- `onEvent: (emission) => void` is an injectable hook on `SupervisorOpts` — Lane C's audit writer uses it; future observability integrations can too.
|
||||
- `autopilot.ts` migration to `MinionSupervisor` is explicitly deferred (follow-up PR): the current `start()` API blocks, which deadlocks autopilot's interval loop. Codex's review flagged this; the fix is a non-blocking-start API redesign, not a drop-in substitution.
|
||||
|
||||
Credit: original supervisor feature built by OpenClaw (PR #364 initial commit). Review wave + code-level fixes + daemon-manager CLI + observability boomerang + integration tests shipped via /autoplan (CEO + DX + Eng + Codex adversarial) followed by a 20-item multi-lane implementation plan.
|
||||
|
||||
## [0.20.0] - 2026-04-23
|
||||
|
||||
## **BrainBench moves out. gbrain gets its install surface back.**
|
||||
|
||||
@@ -62,12 +62,13 @@ strict behavior when unset.
|
||||
- `src/commands/graph-query.ts` — `gbrain graph-query <slug> [--type T] [--depth N] [--direction in|out|both]`: typed-edge relationship traversal (renders indented tree)
|
||||
- `src/core/link-extraction.ts` — shared library for the v0.12.0 graph layer. extractEntityRefs (canonical, replaces backlinks.ts duplicate) matches both `[Name](people/slug)` markdown links and Obsidian `[[people/slug|Name]]` wikilinks as of v0.12.3. extractPageLinks, inferLinkType heuristics (attended/works_at/invested_in/founded/advises/source/mentions), parseTimelineEntries, isAutoLinkEnabled config helper. `DIR_PATTERN` covers `people`, `companies`, `deals`, `topics`, `concepts`, `projects`, `entities`, `tech`, `finance`, `personal`, `openclaw`. Used by extract.ts, operations.ts auto-link post-hook, and backlinks.ts.
|
||||
- `src/core/minions/` — Minions job queue: BullMQ-inspired, Postgres-native (queue, worker, backoff, types, protected-names, quiet-hours, stagger, handlers/shell).
|
||||
- `src/core/minions/queue.ts` — MinionQueue class (submit, claim, complete, fail, stall detection, parent-child, depth/child-cap, per-job timeouts, cascade-kill, attachments, idempotency keys, child_done inbox, removeOnComplete/Fail). `add()` takes a 4th `trusted` arg (separate from `opts` to prevent spread leakage); protected names in `PROTECTED_JOB_NAMES` require `{allowProtectedSubmit: true}` and the check runs trim-normalized (whitespace-bypass safe). v0.14.1 #219: `add()` plumbs `max_stalled` through with a `[1, 100]` clamp; omitted values let the schema DEFAULT (5) kick in.
|
||||
- `src/core/minions/queue.ts` — MinionQueue class (submit, claim, complete, fail, stall detection, parent-child, depth/child-cap, per-job timeouts, cascade-kill, attachments, idempotency keys, child_done inbox, removeOnComplete/Fail). `add()` takes a 4th `trusted` arg (separate from `opts` to prevent spread leakage); protected names in `PROTECTED_JOB_NAMES` require `{allowProtectedSubmit: true}` and the check runs trim-normalized (whitespace-bypass safe). v0.14.1 #219: `add()` plumbs `max_stalled` through with a `[1, 100]` clamp; omitted values let the schema DEFAULT (5) kick in. v0.19.0: `handleWallClockTimeouts(lockDurationMs)` is Layer 3 kill shot for jobs where `FOR UPDATE SKIP LOCKED` stall detection and the timeout sweep both fail to evict (wedged worker holding a row lock via a pending transaction). v0.19.1: `maxWaiting` coalesce path now uses `pg_advisory_xact_lock` keyed on `(name, queue)` to serialize concurrent submits for the same key, and filters on `queue` in addition to `name` so cross-queue same-name jobs don't suppress each other.
|
||||
- `src/core/minions/worker.ts` — MinionWorker class (handler registry, lock renewal, graceful shutdown, timeout safety net). v0.14.0 abort-path fix: aborted jobs now call `failJob` with reason (`timeout`/`cancel`/`lock-lost`/`shutdown`) instead of returning silently. `shutdownAbort` (instance field) fires on process SIGTERM/SIGINT and propagates to `ctx.shutdownSignal` — shell handler listens to it; non-shell handlers don't.
|
||||
- `src/core/minions/types.ts` — `MinionJobInput` + `MinionJobStatus` + handler context types. `MinionJobInput.max_stalled` (new in v0.14.1) is optional; omitted values let the schema DEFAULT (5) kick in, provided values are clamped to `[1, 100]`.
|
||||
- `src/core/minions/protected-names.ts` — side-effect-free constant module exporting `PROTECTED_JOB_NAMES` + `isProtectedJobName()`. Kept pure so queue core can import without loading handler modules.
|
||||
- `src/core/minions/handlers/shell.ts` — `shell` job handler. Spawns `/bin/sh -c cmd` (absolute path, PATH-override-safe) or `argv[0] argv[1..]` (no shell). Env allowlist: `PATH, HOME, USER, LANG, TZ, NODE_ENV` + caller `env:` overrides. UTF-8-safe stdout/stderr tail via `string_decoder.StringDecoder`. Abort (either `ctx.signal` or `ctx.shutdownSignal`) fires SIGTERM → 5s grace → SIGKILL on child. Requires `GBRAIN_ALLOW_SHELL_JOBS=1` on worker (gated by `registerBuiltinHandlers`).
|
||||
- `src/core/minions/handlers/shell-audit.ts` — per-submission JSONL audit trail at `~/.gbrain/audit/shell-jobs-YYYY-Www.jsonl` (ISO-week rotation; override via `GBRAIN_AUDIT_DIR`). Best-effort: `mkdirSync(recursive)` + `appendFileSync`; failures logged to stderr, submission not blocked. Logs cmd (first 80 chars) or argv (JSON array). Never logs env values.
|
||||
- `src/core/minions/backpressure-audit.ts` (v0.19.1) — sibling of shell-audit.ts for `maxWaiting` coalesce events. JSONL at `~/.gbrain/audit/backpressure-YYYY-Www.jsonl`. Fires one line per coalesce with `(queue, name, waiting_count, max_waiting, returned_job_id, ts)`. Closes the silent-drop vector the v0.19.0 maxWaiting guard introduced.
|
||||
- `src/core/minions/handlers/subagent.ts` (v0.15) — LLM-loop handler. Two-phase tool persistence (pending → complete/failed), replay reconciliation for mid-dispatch crashes, dual-signal abort (`ctx.signal` + `ctx.shutdownSignal`), Anthropic prompt caching on system + tool defs. `makeSubagentHandler({engine, client?, ...})` factory; `MessagesClient` is an injectable interface the real SDK implements structurally. Throws `RateLeaseUnavailableError` (renewable) when rate-lease capacity is full.
|
||||
- `src/core/minions/handlers/subagent-aggregator.ts` (v0.15) — `subagent_aggregator` handler. Claims AFTER all children resolve (queue changes guarantee every terminal child posts a `child_done` inbox message with outcome). Reads inbox via `ctx.readInbox()`, builds deterministic mixed-outcome markdown summary. No LLM call in v0.15.
|
||||
- `src/core/minions/handlers/subagent-audit.ts` (v0.15) — JSONL audit + heartbeat writer at `~/.gbrain/audit/subagent-jobs-YYYY-Www.jsonl`. Events: `submission` (one line per submit) + `heartbeat` (per turn boundary: `llm_call_started | llm_call_completed | tool_called | tool_result | tool_failed`). Never logs prompts or tool inputs. `readSubagentAuditForJob(jobId, {sinceIso})` is the readback path for `gbrain agent logs`.
|
||||
@@ -89,7 +90,7 @@ strict behavior when unset.
|
||||
- `src/commands/migrations/` — TS migration registry (compiled into the binary; no filesystem walk of `skills/migrations/*.md` needed at runtime). `index.ts` lists migrations in semver order. `v0_11_0.ts` = Minions adoption orchestrator (8 phases). `v0_12_0.ts` = Knowledge Graph auto-wire orchestrator (5 phases: schema → config check → backfill links → backfill timeline → verify). `phaseASchema` has a 600s timeout (bumped from 60s in v0.12.1 for duplicate-heavy brains). `v0_12_2.ts` = JSONB double-encode repair orchestrator (4 phases: schema → repair-jsonb → verify → record). `v0_14_0.ts` = shell-jobs + autopilot cooperative (2 phases: schema ALTER minion_jobs.max_stalled SET DEFAULT 3 — superseded by v0.14.3's schema-level DEFAULT 5 + UPDATE backfill; pending-host-work ping for skills/migrations/v0.14.0.md). All orchestrators are idempotent and resumable from `partial` status. As of v0.14.2 (Bug 3), the RUNNER owns all ledger writes — orchestrators return `OrchestratorResult` and `apply-migrations.ts` persists a canonical `{version, status, phases}` shape after return. Orchestrators no longer call `appendCompletedMigration` directly. `statusForVersion` prefers `complete` over `partial` (never regresses). 3 consecutive partials → wedged → `--force-retry <version>` writes a `'retry'` reset marker. v0.14.3 (fix wave) ships schema-only migrations v14 (`pages_updated_at_index`) + v15 (`minion_jobs_max_stalled_default_5` with UPDATE backfill) via the `MIGRATIONS` array in `src/core/migrate.ts` — no orchestrator phases needed.
|
||||
- `src/commands/repair-jsonb.ts` — `gbrain repair-jsonb [--dry-run] [--json]`: rewrites `jsonb_typeof='string'` rows in place across 5 affected columns (pages.frontmatter, raw_data.data, ingest_log.pages_updated, files.metadata, page_versions.frontmatter). Fixes v0.12.0 double-encode bug on Postgres; PGLite no-ops. Idempotent.
|
||||
- `src/commands/orphans.ts` — `gbrain orphans [--json] [--count] [--include-pseudo]`: surfaces pages with zero inbound wikilinks, grouped by domain. Auto-generated/raw/pseudo pages filtered by default. Also exposed as `find_orphans` MCP operation. Shipped in v0.12.3 (contributed by @knee5).
|
||||
- `src/commands/doctor.ts` — `gbrain doctor [--json] [--fast] [--fix] [--dry-run] [--index-audit]`: health checks. v0.12.3 added `jsonb_integrity` + `markdown_body_completeness` reliability checks. v0.14.1: `--fix` delegates inlined cross-cutting rules to `> **Convention:** see [path](path).` callouts (pipes DRY violations into `src/core/dry-fix.ts`); `--fix --dry-run` previews without writing. v0.14.2: `schema_version` check fails loudly when `version=0` (migrations never ran — the #218 `bun install -g` signature) and routes users to `gbrain apply-migrations --yes`; new opt-in `--index-audit` flag (Postgres-only) reports zero-scan indexes from `pg_stat_user_indexes` (informational only, no auto-drop). v0.15.2: every DB check is wrapped in a progress phase; `markdown_body_completeness` runs under a 1s heartbeat timer so 10+ min scans are observable on 50K-page brains. Fix hints point at `gbrain repair-jsonb`, `gbrain sync --force`, and `gbrain apply-migrations`.
|
||||
- `src/commands/doctor.ts` — `gbrain doctor [--json] [--fast] [--fix] [--dry-run] [--index-audit]`: health checks. v0.12.3 added `jsonb_integrity` + `markdown_body_completeness` reliability checks. v0.14.1: `--fix` delegates inlined cross-cutting rules to `> **Convention:** see [path](path).` callouts (pipes DRY violations into `src/core/dry-fix.ts`); `--fix --dry-run` previews without writing. v0.14.2: `schema_version` check fails loudly when `version=0` (migrations never ran — the #218 `bun install -g` signature) and routes users to `gbrain apply-migrations --yes`; new opt-in `--index-audit` flag (Postgres-only) reports zero-scan indexes from `pg_stat_user_indexes` (informational only, no auto-drop). v0.15.2: every DB check is wrapped in a progress phase; `markdown_body_completeness` runs under a 1s heartbeat timer so 10+ min scans are observable on 50K-page brains. v0.19.1 added `queue_health` (Postgres-only) with two subchecks: stalled-forever active jobs (started_at > 1h) and waiting-depth-per-name > threshold (default 10, override via `GBRAIN_QUEUE_WAITING_THRESHOLD`). Worker-heartbeat subcheck intentionally deferred to follow-up B7 because it needs a `minion_workers` table to produce ground-truth signal. Fix hints point at `gbrain repair-jsonb`, `gbrain sync --force`, `gbrain apply-migrations`, and `gbrain jobs get/cancel <id>`.
|
||||
- `src/core/migrate.ts` — schema-migration runner. Owns the `MIGRATIONS` array (source of truth for schema DDL). v0.14.2 extended the `Migration` interface with `sqlFor?: { postgres?, pglite? }` (engine-specific SQL overrides `sql`) and `transaction?: boolean` (set to false for `CREATE INDEX CONCURRENTLY`, which Postgres refuses inside a transaction; ignored on PGLite since it has no concurrent writers). Migration v14 (fix wave) uses a handler branching on `engine.kind` to run CONCURRENTLY on Postgres (with a pre-drop of any invalid remnant via `pg_index.indisvalid`) and plain `CREATE INDEX` on PGLite. v15 bumps `minion_jobs.max_stalled` default 1→5 and backfills existing non-terminal rows.
|
||||
- `src/core/progress.ts` — Shared bulk-action progress reporter. Writes to stderr. Modes: `auto` (TTY: `\r`-rewriting; non-TTY: plain lines), `human`, `json` (JSONL), `quiet`. Rate-gated by `minIntervalMs` and `minItems`. `startHeartbeat(reporter, note)` helper for single long queries. `child()` composes phase paths. Singleton SIGINT/SIGTERM coordinator emits `abort` events for every live phase. EPIPE defense on both sync throws and stream `'error'` events. Zero dependencies. Introduced in v0.15.2.
|
||||
- `src/core/cli-options.ts` — Global CLI flag parser. `parseGlobalFlags(argv)` returns `{cliOpts, rest}` with `--quiet` / `--progress-json` / `--progress-interval=<ms>` stripped. `getCliOptions()` / `setCliOptions()` expose a module-level singleton so commands reach the resolved flags without parameter threading. `cliOptsToProgressOptions()` maps to reporter options. `childGlobalFlags()` returns the flag suffix to append to `execSync('gbrain ...')` calls in migration orchestrators. `OperationContext.cliOpts` extends shared-op dispatch for MCP callers.
|
||||
|
||||
@@ -211,9 +211,12 @@ The six daily pains — spawn storms, agents that stop responding, forgotten dis
|
||||
gbrain jobs smoke # verify install
|
||||
gbrain jobs submit sync --params '{}' # fire a background job
|
||||
gbrain jobs stats # health dashboard
|
||||
gbrain jobs work --concurrency 4 # start a worker (Postgres only)
|
||||
gbrain jobs supervisor --concurrency 4 # canonical: auto-restarting worker (Postgres only)
|
||||
gbrain jobs work --concurrency 4 # raw worker (no crash recovery — prefer `supervisor`)
|
||||
```
|
||||
|
||||
`gbrain jobs supervisor` keeps the worker alive across crashes with exponential backoff, atomic PID locking, structured audit events at `~/.gbrain/audit/supervisor-*.jsonl`, and a `start --detach` / `status --json` / `stop` subcommand surface for agents. In containers it runs as PID 1; on systemd hosts it's the child of `gbrain-worker.service`. Full deployment guide: [`docs/guides/minions-deployment.md`](docs/guides/minions-deployment.md).
|
||||
|
||||
Read [`skills/minion-orchestrator/SKILL.md`](skills/minion-orchestrator/SKILL.md) for parent-child DAGs, fan-in collection, steering via inbox.
|
||||
|
||||
**Minions is not incrementally better than sub-agents for background work. It's categorically different.** 753ms vs gateway timeout. $0 vs tokens. 100% vs couldn't-spawn. If your agent does deterministic work on a schedule, it runs on Minions now.
|
||||
|
||||
@@ -7,4 +7,7 @@
|
||||
# DATABASE_URL=postgresql://...
|
||||
# GBRAIN_ALLOW_SHELL_JOBS=1 # only if submitting shell jobs
|
||||
|
||||
worker: gbrain jobs work --concurrency 2
|
||||
# Two-layer supervision: the platform restarts the container on host
|
||||
# events (OOM, deploy); `gbrain jobs supervisor` restarts the worker
|
||||
# on in-process crashes with exponential backoff.
|
||||
worker: gbrain jobs supervisor --concurrency 2
|
||||
|
||||
@@ -5,10 +5,12 @@
|
||||
# fly secrets set GBRAIN_ALLOW_SHELL_JOBS=1 # only if submitting shell jobs
|
||||
# fly secrets set ANTHROPIC_API_KEY=... # optional
|
||||
#
|
||||
# Fly.io auto-restarts the process on crash — no watchdog needed.
|
||||
# Two-layer supervision: Fly restarts the VM on host events; the
|
||||
# `gbrain jobs supervisor` process restarts the worker on in-process
|
||||
# crashes with exponential backoff and a structured audit trail.
|
||||
|
||||
[processes]
|
||||
worker = "gbrain jobs work --concurrency 2"
|
||||
worker = "gbrain jobs supervisor --concurrency 2"
|
||||
|
||||
# Scale the worker process to 1 machine (job queue serializes work; more
|
||||
# machines means higher concurrency but also more Postgres connections).
|
||||
|
||||
@@ -1,68 +0,0 @@
|
||||
#!/bin/bash
|
||||
# minion-watchdog.sh — restart gbrain jobs work if the process is dead or
|
||||
# has logged a shutdown marker since its last start.
|
||||
#
|
||||
# Fixes the v0.16.1 restart-loop bug: old shutdown lines from previous
|
||||
# restarts stayed in the unrotated log and every tick re-matched them
|
||||
# forever. This version writes a restart epoch to line 2 of the PID file
|
||||
# and only considers log lines newer than that epoch.
|
||||
#
|
||||
# Run every 5 minutes from crontab. See docs/guides/minions-deployment.md.
|
||||
set -u
|
||||
|
||||
PID_FILE="${GBRAIN_WORKER_PID_FILE:-/tmp/gbrain-worker.pid}"
|
||||
LOG_FILE="${GBRAIN_WORKER_LOG_FILE:-/tmp/gbrain-worker.log}"
|
||||
GBRAIN="${GBRAIN_BIN:-/usr/local/bin/gbrain}"
|
||||
CONCURRENCY="${GBRAIN_WORKER_CONCURRENCY:-2}"
|
||||
|
||||
start_worker() {
|
||||
# stderr merged so banner lines ("[minion worker] shell handler enabled",
|
||||
# "worker shutting down") all land in $LOG_FILE.
|
||||
nohup "$GBRAIN" jobs work --concurrency "$CONCURRENCY" \
|
||||
> "$LOG_FILE" 2>&1 &
|
||||
local pid=$!
|
||||
# Line 1: PID. Line 2: restart epoch (seconds since 1970).
|
||||
# Readers that want just PID use `head -n1 "$PID_FILE"`.
|
||||
printf '%s\n%s\n' "$pid" "$(date +%s)" > "$PID_FILE"
|
||||
}
|
||||
|
||||
shutdown_since_restart() {
|
||||
# Only match shutdown lines logged AFTER the most recent restart epoch.
|
||||
# Worker log lines start with ISO-8601 UTC timestamps ("2026-04-21T19:05:12Z ...").
|
||||
local restart_epoch
|
||||
restart_epoch=$(sed -n '2p' "$PID_FILE" 2>/dev/null || echo 0)
|
||||
[ -z "$restart_epoch" ] && restart_epoch=0
|
||||
|
||||
# POSIX-portable regex (no {n} intervals — mawk on Debian/Ubuntu rejects them).
|
||||
awk -v since="$restart_epoch" '
|
||||
match($0, /^[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]T[0-9:.+Z-]+/) {
|
||||
ts_str = substr($0, RSTART, RLENGTH)
|
||||
cmd = "date -d \"" ts_str "\" +%s 2>/dev/null"
|
||||
cmd | getline ts
|
||||
close(cmd)
|
||||
if (ts + 0 > since + 0) print
|
||||
}
|
||||
' "$LOG_FILE" 2>/dev/null | grep -q "worker stopped\|worker shutting down"
|
||||
}
|
||||
|
||||
if [ -f "$PID_FILE" ]; then
|
||||
PID=$(head -n1 "$PID_FILE")
|
||||
if [ -n "$PID" ] && kill -0 "$PID" 2>/dev/null; then
|
||||
# Process alive — check whether the worker logged an internal shutdown
|
||||
# AFTER the last start. If yes, worker is dead-inside; restart.
|
||||
if shutdown_since_restart; then
|
||||
kill "$PID" 2>/dev/null
|
||||
# 10s grace: covers shell handler's 5s child SIGTERM→SIGKILL window
|
||||
# and leaves room for in-flight jobs to flush. Bump to 30 if your
|
||||
# jobs run > 10s.
|
||||
sleep 10
|
||||
kill -9 "$PID" 2>/dev/null
|
||||
start_worker
|
||||
fi
|
||||
else
|
||||
# PID file exists but process is gone (crash / kill -9 / reboot).
|
||||
start_worker
|
||||
fi
|
||||
else
|
||||
start_worker
|
||||
fi
|
||||
@@ -15,9 +15,13 @@ WorkingDirectory=/srv/gbrain
|
||||
# Env file is mode 600, owned by User=. Do not put secrets in this unit.
|
||||
EnvironmentFile=/etc/gbrain.env
|
||||
|
||||
ExecStart=/usr/local/bin/gbrain jobs work --concurrency 2
|
||||
# Two-layer supervision: systemd restarts `gbrain jobs supervisor` on host
|
||||
# events (reboot, unit crash); the supervisor restarts `gbrain jobs work`
|
||||
# on in-process crashes with exponential backoff + structured audit.
|
||||
ExecStart=/usr/local/bin/gbrain jobs supervisor --concurrency 2
|
||||
|
||||
# Replaces the cron watchdog. systemd restarts on any non-zero exit.
|
||||
# systemd restarts the supervisor on any non-zero exit. The supervisor
|
||||
# itself handles worker-level crash recovery.
|
||||
Restart=always
|
||||
RestartSec=10s
|
||||
|
||||
@@ -38,7 +42,9 @@ NoNewPrivileges=true
|
||||
PrivateTmp=true
|
||||
ProtectSystem=strict
|
||||
ProtectHome=read-only
|
||||
ReadWritePaths=/srv/gbrain
|
||||
# ReadWritePaths must include the brain workspace AND ~/.gbrain (PID file +
|
||||
# audit log written by the supervisor).
|
||||
ReadWritePaths=/srv/gbrain /home/gbrain/.gbrain
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
|
||||
+208
-199
@@ -1,7 +1,7 @@
|
||||
# Minions Worker Deployment Guide
|
||||
|
||||
Deploy `gbrain jobs work` so it stays running across crashes, reboots, and
|
||||
Postgres connection blips. Written for agents to execute line-by-line.
|
||||
Keep `gbrain jobs work` running across crashes, reboots, and Postgres
|
||||
connection blips. Written for agents to execute line-by-line.
|
||||
|
||||
## The problem
|
||||
|
||||
@@ -12,10 +12,61 @@ The persistent worker can die silently from:
|
||||
- Bun process crashes with no automatic restart.
|
||||
- Internal event-loop death (PID alive, worker loop stopped).
|
||||
|
||||
When the worker dies, submitted jobs sit in `waiting` forever. Nothing in
|
||||
gbrain core auto-restarts the worker — that's what this guide wires up.
|
||||
When the worker dies, submitted jobs sit in `waiting` forever. The
|
||||
canonical answer is `gbrain jobs supervisor` — a first-class CLI that
|
||||
spawns `gbrain jobs work` as a child and auto-restarts it on crash.
|
||||
|
||||
## Variables used in this guide
|
||||
## Worker supervision
|
||||
|
||||
### The canonical pattern
|
||||
|
||||
`gbrain jobs supervisor` is an auto-restarting wrapper around
|
||||
`gbrain jobs work`. It writes a PID file, restarts the worker on crash
|
||||
with exponential backoff (1s → 60s cap), emits lifecycle events to an
|
||||
audit file, and drains gracefully on SIGTERM (35s worker-drain window
|
||||
before SIGKILL). Exit codes are documented so agents can branch on them.
|
||||
|
||||
**Typical commands:**
|
||||
|
||||
```bash
|
||||
# Start in the foreground (blocks; Ctrl-C to stop).
|
||||
gbrain jobs supervisor --concurrency 4
|
||||
|
||||
# Start detached — returns {"event":"started","supervisor_pid":…} on stdout.
|
||||
gbrain jobs supervisor start --detach --json
|
||||
|
||||
# Check liveness without reading log files.
|
||||
gbrain jobs supervisor status --json
|
||||
|
||||
# Graceful stop (SIGTERM + drain wait + SIGKILL fallback).
|
||||
gbrain jobs supervisor stop
|
||||
```
|
||||
|
||||
**Exit codes:**
|
||||
|
||||
| Code | Meaning |
|
||||
|---|---|
|
||||
| 0 | Clean shutdown (SIGTERM/SIGINT received, worker drained) |
|
||||
| 1 | Max crashes exceeded (worker kept dying) |
|
||||
| 2 | Another supervisor holds the PID lock |
|
||||
| 3 | PID file unwritable (permission / path error) |
|
||||
|
||||
An agent seeing exit=2 can safely treat it as "one is already running";
|
||||
exit=1 should page a human.
|
||||
|
||||
### Which supervisor when?
|
||||
|
||||
The supervisor solves in-process crash recovery. Platform-level
|
||||
supervision (systemd, Fly, Render) handles host-level failures. You
|
||||
usually want both.
|
||||
|
||||
| Environment | Recommendation |
|
||||
|---|---|
|
||||
| **Container (Fly / Railway / Render / Heroku)** | `gbrain jobs supervisor` runs as PID 1. The platform restarts the container on OOM / host loss; supervisor restarts the worker on crash. See [Fly.io](#flyio) / [Render / Railway / Heroku](#render--railway--heroku). |
|
||||
| **Linux VM with systemd** | Two-layer recommended: systemd supervises `gbrain jobs supervisor`, which in turn supervises `gbrain jobs work`. Buys you automatic restart on reboot (systemd) plus fast crash recovery (supervisor). See [systemd](#systemd). |
|
||||
| **Dev laptop / macOS** | `gbrain jobs supervisor` in a terminal. Ctrl-C stops it. No system-level setup needed. |
|
||||
|
||||
### Variables used in this guide
|
||||
|
||||
Substitute these once before copy-pasting any snippet.
|
||||
|
||||
@@ -23,142 +74,122 @@ Substitute these once before copy-pasting any snippet.
|
||||
|---|---|---|
|
||||
| `$GBRAIN_BIN` | Absolute path to the `gbrain` binary | `$(command -v gbrain)` — often `/usr/local/bin/gbrain` or `~/.bun/bin/gbrain` |
|
||||
| `$GBRAIN_WORKER_USER` | OS user that owns the worker process | the same user that ran `gbrain init`; never `root` |
|
||||
| `$GBRAIN_WORKER_PID_FILE` | Worker PID + restart-epoch file | `/tmp/gbrain-worker.pid` (or `/var/run/gbrain/worker.pid` for systemd) |
|
||||
| `$GBRAIN_WORKER_LOG_FILE` | Worker log sink (stdout + stderr merged) | `/tmp/gbrain-worker.log` (or `/var/log/gbrain/worker.log`) |
|
||||
| `$GBRAIN_WORKSPACE` | `cwd` for shell jobs submitted by this deployment | absolute path, e.g. `/srv/my-brain` |
|
||||
| `$GBRAIN_ENV_FILE` | Secrets file sourced by crontab / systemd | `/etc/gbrain.env` (mode 600) |
|
||||
| `$GBRAIN_ENV_FILE` | Secrets file sourced by systemd / shell | `/etc/gbrain.env` (mode 600) |
|
||||
|
||||
## Preconditions
|
||||
### Preconditions
|
||||
|
||||
Run these before Step 1 of any option. Fail fast if something is wrong.
|
||||
Run these before any deployment step.
|
||||
|
||||
```bash
|
||||
# 1. gbrain is on PATH and resolves to an absolute location.
|
||||
command -v gbrain || { echo "gbrain not on PATH. Install, then retry."; exit 1; }
|
||||
|
||||
# 2. DATABASE_URL points at reachable Postgres (or PGLite path exists).
|
||||
# 2. DATABASE_URL points at reachable Postgres.
|
||||
# (Supervisor is Postgres-only. PGLite's exclusive file lock blocks the
|
||||
# separate worker process. If `config.engine === 'pglite'` the CLI rejects
|
||||
# with a clear error.)
|
||||
gbrain doctor --fast --json | jq '.checks[] | select(.name=="db_connectivity")'
|
||||
|
||||
# 3. Schema is up to date. If version=0 or status=="fail", fix it first:
|
||||
# 3. Schema is up to date. If version=0 or status=="fail":
|
||||
# gbrain apply-migrations --yes
|
||||
gbrain doctor --fast --json | jq '.checks[] | select(.name=="schema_version")'
|
||||
|
||||
# 4. You have write access to at least one crontab mechanism.
|
||||
crontab -l >/dev/null 2>&1 && echo "user crontab OK"
|
||||
[ -w /etc/crontab ] && echo "/etc/crontab OK"
|
||||
|
||||
# 5. If you plan to submit `shell` jobs, the WORKER process needs
|
||||
# GBRAIN_ALLOW_SHELL_JOBS=1 (submitters do not). The handler is gated
|
||||
# in registerBuiltinHandlers(); without the flag the worker startup
|
||||
# line reads "shell handler disabled (...)".
|
||||
# 4. If you plan to submit `shell` jobs, pass --allow-shell-jobs to the
|
||||
# supervisor (or export GBRAIN_ALLOW_SHELL_JOBS=1 before starting).
|
||||
# Without the flag, the shell handler is disabled at worker startup.
|
||||
```
|
||||
|
||||
## Which option?
|
||||
## Agent usage (OpenClaw / Hermes / Cursor / Codex)
|
||||
|
||||
- Your workload runs LLM subagents (`gbrain agent run`) or jobs that take
|
||||
> 30 s → **Option 1** (watchdog cron + persistent worker).
|
||||
- Your workload is short deterministic scripts on a fixed schedule (every
|
||||
3 h, daily, weekly) → **Option 2** (inline `--follow`).
|
||||
- You don't have shell access to a long-running box (Fly/Render/Railway,
|
||||
or any systemd host) → **Option 3** (service manager — replaces cron).
|
||||
|
||||
## Option 1: watchdog cron + persistent worker
|
||||
|
||||
A 5-minute cron checks whether the worker process is alive **and** whether
|
||||
it has logged an internal shutdown since its last start. Restarts if either
|
||||
condition fails.
|
||||
|
||||
### 1a. Install the env file (secrets stay out of crontab)
|
||||
|
||||
Never paste `DATABASE_URL` or API keys into crontab. `/etc/crontab` is
|
||||
mode 644 (world-readable); user crontabs under `/var/spool/cron/` are
|
||||
readable by `root`. Use the shipped env-file template:
|
||||
Three-command pattern an agent can drive without shell archaeology:
|
||||
|
||||
```bash
|
||||
sudo install -m 600 -o $GBRAIN_WORKER_USER -g $GBRAIN_WORKER_USER \
|
||||
# Start (returns PIDs + pid_file on stdout as JSON, then detaches)
|
||||
gbrain jobs supervisor start --detach --json
|
||||
# → {"event":"started","supervisor_pid":1234,"worker_pid":1235,"pid_file":"/Users/you/.gbrain/supervisor.pid"}
|
||||
|
||||
# Check health (machine-parseable JSON, no log scraping)
|
||||
gbrain jobs supervisor status --json
|
||||
# → {"running":true,"supervisor_pid":1234,"last_start":"2026-04-23T15:30:22Z","crashes_24h":0, ...}
|
||||
|
||||
# Stop cleanly (SIGTERM + 35s drain + SIGKILL fallback)
|
||||
gbrain jobs supervisor stop
|
||||
```
|
||||
|
||||
Every lifecycle event (spawn, crash, backoff, health warning, max-crashes,
|
||||
shutdown) is also written to `${GBRAIN_AUDIT_DIR:-~/.gbrain/audit}/supervisor-YYYY-Www.jsonl`
|
||||
for historical inspection. `gbrain doctor` reads that file and surfaces
|
||||
a `supervisor` check in its health report.
|
||||
|
||||
## Deployment: systemd
|
||||
|
||||
For long-running Linux VMs with shell access.
|
||||
|
||||
```bash
|
||||
# Create the worker user if it doesn't exist.
|
||||
sudo useradd --system --home "$GBRAIN_WORKSPACE" --shell /usr/sbin/nologin gbrain \
|
||||
2>/dev/null || true
|
||||
sudo mkdir -p "$GBRAIN_WORKSPACE" && sudo chown gbrain:gbrain "$GBRAIN_WORKSPACE"
|
||||
|
||||
# Install the env file (secrets stay out of the unit file).
|
||||
sudo install -m 600 -o gbrain -g gbrain \
|
||||
docs/guides/minions-deployment-snippets/gbrain.env.example /etc/gbrain.env
|
||||
sudoedit /etc/gbrain.env
|
||||
# Fill in DATABASE_URL, optional GBRAIN_ALLOW_SHELL_JOBS=1.
|
||||
|
||||
# Install the unit file, substituting /srv/gbrain → your workspace path.
|
||||
sudo install -m 644 docs/guides/minions-deployment-snippets/systemd.service \
|
||||
/etc/systemd/system/gbrain-worker.service
|
||||
sudo sed -i "s|/srv/gbrain|$GBRAIN_WORKSPACE|g" \
|
||||
/etc/systemd/system/gbrain-worker.service
|
||||
|
||||
sudo systemctl daemon-reload
|
||||
sudo systemctl enable --now gbrain-worker
|
||||
sudo systemctl status gbrain-worker
|
||||
journalctl -u gbrain-worker -n 50
|
||||
```
|
||||
|
||||
Fill in the connection string and `GBRAIN_ALLOW_SHELL_JOBS=1` (if
|
||||
applicable). See
|
||||
[`gbrain.env.example`](./minions-deployment-snippets/gbrain.env.example)
|
||||
for the full list.
|
||||
The shipped unit file invokes `gbrain jobs supervisor` (not `gbrain jobs work`
|
||||
directly) so you get two-layer supervision: systemd restarts the supervisor
|
||||
on host reboot, supervisor restarts the worker on in-process crash.
|
||||
|
||||
### 1b. Install the watchdog script
|
||||
`Restart=always` + `RestartSec=10s` handle the supervisor-level recovery.
|
||||
The unit runs as unprivileged `gbrain` with `PrivateTmp`, `ProtectSystem=strict`,
|
||||
and `ReadWritePaths=$GBRAIN_WORKSPACE,$HOME/.gbrain` (for the PID file and
|
||||
audit log). `LimitNOFILE=65535` covers Bun + Postgres pool + concurrent
|
||||
LLM subagent calls without hitting the default 1024 cap.
|
||||
|
||||
The [`minion-watchdog.sh`](./minions-deployment-snippets/minion-watchdog.sh)
|
||||
ships in-repo and writes a two-line PID file (PID on line 1, restart epoch
|
||||
on line 2). The restart-epoch marker is how the watchdog distinguishes
|
||||
stale shutdown lines in the log from current ones — without it, every tick
|
||||
after the first restart would match an old `worker shutting down` line and
|
||||
loop forever.
|
||||
|
||||
Requires GNU coreutils (Linux default). On macOS/BSD install via
|
||||
`brew install coreutils` and alias `date` to `gdate` in the cron env if you
|
||||
want to test the watchdog locally; production Linux boxes work as-is.
|
||||
## Deployment: Fly.io
|
||||
|
||||
```bash
|
||||
sudo install -m 755 -o $GBRAIN_WORKER_USER -g $GBRAIN_WORKER_USER \
|
||||
docs/guides/minions-deployment-snippets/minion-watchdog.sh \
|
||||
/usr/local/bin/minion-watchdog.sh
|
||||
# Merge the [processes] block from fly.toml.partial into your fly.toml.
|
||||
cat docs/guides/minions-deployment-snippets/fly.toml.partial >> fly.toml
|
||||
# Review + edit as needed.
|
||||
|
||||
# Set secrets (Fly handles restart on crash).
|
||||
fly secrets set DATABASE_URL='postgres://…' GBRAIN_ALLOW_SHELL_JOBS=1
|
||||
```
|
||||
|
||||
### 1c. Wire into cron
|
||||
The `[processes]` block runs `gbrain jobs supervisor` as PID 1. Fly
|
||||
restarts the container on host failure; the supervisor restarts the
|
||||
worker on in-process crash.
|
||||
|
||||
Pick the form that matches the crontab you're editing.
|
||||
## Deployment: Render / Railway / Heroku
|
||||
|
||||
**If you ran `crontab -e`** (user crontab — 5-field, no user column):
|
||||
Drop [`Procfile`](./minions-deployment-snippets/Procfile) at the repo
|
||||
root. The shipped Procfile calls `gbrain jobs supervisor`. Set
|
||||
`DATABASE_URL` + optional `GBRAIN_ALLOW_SHELL_JOBS=1` via the platform's
|
||||
env UI or CLI.
|
||||
|
||||
```
|
||||
SHELL=/bin/bash
|
||||
PATH=/usr/local/bin:/usr/bin:/bin
|
||||
BASH_ENV=/etc/gbrain.env
|
||||
*/5 * * * * /usr/local/bin/minion-watchdog.sh
|
||||
```
|
||||
## Deployment: inline `--follow` (no persistent worker)
|
||||
|
||||
**If you edited `/etc/crontab` directly** (system crontab — 6-field, with
|
||||
user column):
|
||||
|
||||
```
|
||||
SHELL=/bin/bash
|
||||
PATH=/usr/local/bin:/usr/bin:/bin
|
||||
BASH_ENV=/etc/gbrain.env
|
||||
*/5 * * * * gbrain /usr/local/bin/minion-watchdog.sh
|
||||
```
|
||||
|
||||
In both forms, `BASH_ENV=/etc/gbrain.env` tells non-interactive bash to
|
||||
source the env file before running the watchdog — that's how the
|
||||
connection string and `GBRAIN_ALLOW_SHELL_JOBS` reach the worker without
|
||||
landing in the world-readable crontab itself.
|
||||
|
||||
### 1d. Log rotation
|
||||
|
||||
The watchdog appends to the worker log across restarts. If you expect the
|
||||
file to grow unbounded, rotate it externally with `logrotate`:
|
||||
|
||||
```
|
||||
# /etc/logrotate.d/gbrain-worker
|
||||
/tmp/gbrain-worker.log {
|
||||
daily
|
||||
rotate 7
|
||||
missingok
|
||||
notifempty
|
||||
copytruncate
|
||||
}
|
||||
```
|
||||
|
||||
`copytruncate` is important — the watchdog's restart-epoch check survives
|
||||
it (the epoch is compared against in-log timestamps, not file inode).
|
||||
|
||||
## Option 2: inline `--follow` (no persistent worker)
|
||||
|
||||
Each cron run brings its own temporary worker. `--follow` starts one on
|
||||
the queue and blocks until the just-submitted job reaches a terminal state
|
||||
(`completed` / `failed` / `dead` / `cancelled`). 2-3 s startup overhead
|
||||
per job; negligible vs job duration for scheduled work.
|
||||
|
||||
Example: nightly brain enrichment as a shell job.
|
||||
For short deterministic scripts on a fixed schedule where you don't need
|
||||
a persistent worker between runs. Each cron run brings its own temporary
|
||||
worker. `--follow` starts one on the queue and blocks until the
|
||||
just-submitted job reaches a terminal state (`completed` / `failed` /
|
||||
`dead` / `cancelled`). 2-3 s startup overhead per job; negligible vs job
|
||||
duration for scheduled work.
|
||||
|
||||
```bash
|
||||
GBRAIN_ALLOW_SHELL_JOBS=1 gbrain jobs submit shell \
|
||||
@@ -170,85 +201,56 @@ GBRAIN_ALLOW_SHELL_JOBS=1 gbrain jobs submit shell \
|
||||
|
||||
Replace `gbrain embed --stale` with whichever gbrain subcommand you're
|
||||
scheduling (`sync`, `extract`, `orphans`, `doctor`, `check-backlinks`,
|
||||
`lint`, `autopilot`). If you're shelling out to a non-gbrain binary,
|
||||
keep its absolute path in the `cmd`.
|
||||
|
||||
**Shared-queue gotcha.** If other jobs are already waiting on the same
|
||||
queue with higher priority or earlier `created_at`, the temporary worker
|
||||
processes those first before reaching yours. `--follow` still exits only
|
||||
when YOUR job finishes. For strict single-job semantics on shared queues,
|
||||
`lint`, `autopilot`). For strict single-job semantics on shared queues,
|
||||
use a dedicated queue name like `nightly-enrich` above.
|
||||
|
||||
## Option 3: service manager (systemd / Fly / Render / Railway)
|
||||
## Upgrading from an older deployment
|
||||
|
||||
Replaces the watchdog entirely. No cron, no PID file, no restart-loop.
|
||||
The service manager owns liveness.
|
||||
### From `minion-watchdog.sh` (pre-v0.20)
|
||||
|
||||
### systemd (Linux hosts with shell access)
|
||||
Earlier versions of this guide shipped a 68-line bash watchdog
|
||||
(`minion-watchdog.sh`). It's been replaced by `gbrain jobs supervisor`
|
||||
which handles everything the script did, plus atomic PID locking,
|
||||
structured audit events, queue-scoped health checks, and graceful
|
||||
drain on SIGTERM.
|
||||
|
||||
**Migration:**
|
||||
|
||||
```bash
|
||||
# Create the worker user if it doesn't exist.
|
||||
sudo useradd --system --home "$GBRAIN_WORKSPACE" --shell /usr/sbin/nologin gbrain \
|
||||
2>/dev/null || true
|
||||
sudo mkdir -p "$GBRAIN_WORKSPACE" && sudo chown gbrain:gbrain "$GBRAIN_WORKSPACE"
|
||||
# 1. Stop and remove the old watchdog.
|
||||
sudo kill $(head -n1 /tmp/gbrain-worker.pid) 2>/dev/null
|
||||
sudo rm -f /usr/local/bin/minion-watchdog.sh /tmp/gbrain-worker.pid \
|
||||
/tmp/gbrain-worker.log
|
||||
crontab -e # delete the "*/5 * * * * /usr/local/bin/minion-watchdog.sh" line
|
||||
|
||||
# Install the unit file, substituting /srv/gbrain → your workspace path.
|
||||
sudo install -m 644 docs/guides/minions-deployment-snippets/systemd.service \
|
||||
/etc/systemd/system/gbrain-worker.service
|
||||
sudo sed -i "s|/srv/gbrain|$GBRAIN_WORKSPACE|g" \
|
||||
/etc/systemd/system/gbrain-worker.service
|
||||
# 2. Start the supervisor (systemd users: reinstall the unit from
|
||||
# docs/guides/minions-deployment-snippets/systemd.service, which
|
||||
# now calls `gbrain jobs supervisor`).
|
||||
gbrain jobs supervisor start --detach --json
|
||||
# Or: sudo systemctl restart gbrain-worker
|
||||
|
||||
# See 1a above for /etc/gbrain.env install.
|
||||
sudo systemctl daemon-reload
|
||||
sudo systemctl enable --now gbrain-worker
|
||||
sudo systemctl status gbrain-worker
|
||||
journalctl -u gbrain-worker -n 50
|
||||
# 3. Verify.
|
||||
gbrain jobs supervisor status --json
|
||||
gbrain doctor # 'supervisor' check should report running=true
|
||||
```
|
||||
|
||||
`Restart=always` + `RestartSec=10s` give you crash-loop recovery. The unit
|
||||
runs as an unprivileged `gbrain` user with `PrivateTmp`, `ProtectSystem=strict`,
|
||||
and `ReadWritePaths=$GBRAIN_WORKSPACE`. `LimitNOFILE=65535` in the shipped
|
||||
unit covers Bun + Postgres pool + concurrent LLM subagent calls without
|
||||
hitting the default 1024 cap.
|
||||
### Schema / migration hygiene
|
||||
|
||||
### Fly.io
|
||||
Regardless of which deployment path you're upgrading from:
|
||||
|
||||
Merge the `[processes]` block from
|
||||
[`fly.toml.partial`](./minions-deployment-snippets/fly.toml.partial) into
|
||||
your existing `fly.toml`. Set secrets with `fly secrets set` —
|
||||
Fly auto-restarts the process on crash.
|
||||
|
||||
### Render / Railway / Heroku
|
||||
|
||||
Drop [`Procfile`](./minions-deployment-snippets/Procfile) at the repo root.
|
||||
Set the connection string and `GBRAIN_ALLOW_SHELL_JOBS=1` via the
|
||||
platform's env UI or CLI.
|
||||
|
||||
## Upgrading an existing deployment
|
||||
|
||||
If you deployed on v0.13.x or earlier, walk this checklist:
|
||||
|
||||
1. **Stop the worker before upgrading.**
|
||||
`kill $(head -n1 /tmp/gbrain-worker.pid)` and wait for the process to
|
||||
exit. Skipping this risks an in-flight job landing partial schema.
|
||||
1. **Stop the worker before upgrading.** `gbrain jobs supervisor stop`
|
||||
(or `sudo systemctl stop gbrain-worker`). Skipping this risks an
|
||||
in-flight job landing partial schema.
|
||||
2. **Run `gbrain upgrade`**. Then `gbrain apply-migrations --yes` if
|
||||
`gbrain doctor` reports any migration as `partial` or `pending`.
|
||||
3. **If you run shell jobs:** from v0.14 onward, the worker requires
|
||||
`GBRAIN_ALLOW_SHELL_JOBS=1` to register the `shell` handler. Add it to
|
||||
`/etc/gbrain.env`. Submitters don't need the flag; only the worker does.
|
||||
4. **If you tuned your watchdog for `max_stalled=1`:** v0.14.3 migration
|
||||
v15 raised the schema default to 5 and backfilled existing non-terminal
|
||||
rows. A watchdog tuned around 1-strike dead-lettering will now
|
||||
over-restart because it takes 5 misses to dead-letter. Switch to the
|
||||
shipped watchdog (which keys on log markers, not job state).
|
||||
5. **If your v0.16.1 watchdog is still running:** it has a restart-loop
|
||||
bug (old shutdown lines in the unrotated log re-match every 5 min
|
||||
forever). Install the current `minion-watchdog.sh` from this guide's
|
||||
snippets — it writes a restart epoch into the PID file and only
|
||||
considers log lines newer than that epoch.
|
||||
6. **Verify.** `gbrain doctor` should report zero `pending` or `partial`
|
||||
migrations. `gbrain jobs stats` should show no unexplained growth in
|
||||
`dead` between pre- and post-upgrade.
|
||||
3. **If you run shell jobs:** from v0.14 onward, pass
|
||||
`--allow-shell-jobs` to the supervisor (or keep
|
||||
`GBRAIN_ALLOW_SHELL_JOBS=1` in `/etc/gbrain.env`). Submitters don't
|
||||
need the flag; only the worker does.
|
||||
4. **Verify.** `gbrain doctor` should report zero `pending` or `partial`
|
||||
migrations plus a healthy `supervisor` check. `gbrain jobs stats`
|
||||
should show no unexplained growth in `dead` between pre- and
|
||||
post-upgrade.
|
||||
|
||||
## Known issues
|
||||
|
||||
@@ -261,9 +263,10 @@ silently. The stall detector then dead-letters the job after
|
||||
|
||||
**Current defaults that make this worse:**
|
||||
|
||||
- `lockDuration: 30000` (30 s) — too short for long jobs during connection blips.
|
||||
- `max_stalled: 5` (schema column default on master — see `src/schema.sql`
|
||||
and `src/core/pglite-schema.ts`). Five missed heartbeats before dead-letter.
|
||||
- `lockDuration: 30000` (30 s) — too short for long jobs during
|
||||
connection blips.
|
||||
- `max_stalled: 5` (schema column default — see `src/schema.sql` and
|
||||
`src/core/pglite-schema.ts`). Five missed heartbeats before dead-letter.
|
||||
- `stalledInterval: 30000` (30 s) — checks too aggressively.
|
||||
|
||||
**Tune per-job today.** `gbrain jobs submit` accepts `--max-stalled N`,
|
||||
@@ -271,9 +274,6 @@ silently. The stall detector then dead-letters the job after
|
||||
`--backoff-jitter 0..1`, and `--timeout-ms N` as first-class flags
|
||||
(since v0.13.1). These write onto the job row at submit time — which is
|
||||
what `handleStalled()` reads — so per-job tuning is the real knob today.
|
||||
Worker-level `--lock-duration` / `--stall-interval` are on the roadmap;
|
||||
until they land, rely on per-job `--max-stalled` plus the watchdog (or
|
||||
systemd) for worker health.
|
||||
|
||||
### DO NOT pass `maxStalledCount` to `MinionWorker`
|
||||
|
||||
@@ -284,16 +284,16 @@ Use `gbrain jobs submit --max-stalled N` per-job instead.
|
||||
### Zombie shell children
|
||||
|
||||
When the Bun worker crashes hard, child processes from shell jobs can
|
||||
become zombies. The watchdog's 10 s `SIGTERM → SIGKILL` window covers the
|
||||
shell handler's 5 s child-kill grace (`KILL_GRACE_MS`). For long-running
|
||||
shell jobs, bump the watchdog's `sleep 10` to `sleep 30` so the worker
|
||||
has time to flush in-flight jobs before the kill.
|
||||
become zombies. The supervisor's SIGTERM → 35s drain → SIGKILL window
|
||||
covers the shell handler's 5 s child-kill grace (`KILL_GRACE_MS`). For
|
||||
long-running shell jobs, prefer timeouts via `--timeout-ms` on submit
|
||||
over relying on hard kills.
|
||||
|
||||
## Smoke test
|
||||
|
||||
```bash
|
||||
# Worker alive?
|
||||
kill -0 $(head -n1 /tmp/gbrain-worker.pid) 2>/dev/null && echo ALIVE || echo DEAD
|
||||
# Supervisor alive?
|
||||
gbrain jobs supervisor status --json | jq .running
|
||||
|
||||
# Aggregate queue health.
|
||||
gbrain jobs stats
|
||||
@@ -304,20 +304,29 @@ gbrain jobs list --status active --limit 10
|
||||
# Dead-lettered jobs.
|
||||
gbrain jobs list --status dead --limit 10
|
||||
|
||||
# Shell handler registered? (stderr banner merged into log via 2>&1.)
|
||||
grep "shell handler enabled" /tmp/gbrain-worker.log
|
||||
# Shell handler registered? (check supervisor audit log or worker stderr.)
|
||||
gbrain jobs supervisor status --json | jq '.worker_config.allow_shell_jobs'
|
||||
```
|
||||
|
||||
## Uninstall
|
||||
|
||||
- **Option 1 (watchdog cron):** `crontab -e`, delete the watchdog line.
|
||||
`kill $(head -n1 /tmp/gbrain-worker.pid) && rm /tmp/gbrain-worker.pid`.
|
||||
Optionally `sudo rm /etc/gbrain.env /usr/local/bin/minion-watchdog.sh`.
|
||||
- **Option 2 (inline `--follow`):** remove the cron entry. Nothing else to
|
||||
clean up — temporary workers exit with their jobs.
|
||||
- **Option 3 (systemd):** `sudo systemctl disable --now gbrain-worker`,
|
||||
then `sudo rm /etc/systemd/system/gbrain-worker.service /etc/gbrain.env`,
|
||||
then `sudo systemctl daemon-reload`.
|
||||
- **Option 3 (Fly/Render/Railway):** delete the `worker` process from
|
||||
`fly.toml` / `Procfile` and redeploy. Secrets set via `fly secrets`
|
||||
persist until `fly secrets unset`.
|
||||
**`gbrain jobs supervisor`** (foreground or `--detach`):
|
||||
|
||||
```bash
|
||||
gbrain jobs supervisor stop
|
||||
```
|
||||
|
||||
**systemd:**
|
||||
|
||||
```bash
|
||||
sudo systemctl disable --now gbrain-worker
|
||||
sudo rm /etc/systemd/system/gbrain-worker.service /etc/gbrain.env
|
||||
sudo systemctl daemon-reload
|
||||
```
|
||||
|
||||
**Fly / Render / Railway:** delete the `worker` process from `fly.toml`
|
||||
/ `Procfile` and redeploy. Secrets set via `fly secrets` persist until
|
||||
`fly secrets unset`.
|
||||
|
||||
**Inline `--follow`:** remove the cron entry. Nothing else to clean up
|
||||
— temporary workers exit with their jobs.
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
# Queue operations runbook
|
||||
|
||||
"My queue looks wedged — what do I run?" The commands below are in the order
|
||||
you probably want them. Shipped with v0.19.1 after a production incident
|
||||
where the queue held for 90+ minutes before the operator noticed.
|
||||
|
||||
## First signal: jobs aren't running
|
||||
|
||||
```bash
|
||||
gbrain doctor --json | jq '.checks[] | select(.name == "queue_health")'
|
||||
```
|
||||
|
||||
`queue_health` flags two patterns:
|
||||
|
||||
- **stalled-forever**: active job whose `started_at` is older than 1h.
|
||||
- **waiting-depth**: any per-name queue deeper than 10 (override via
|
||||
`GBRAIN_QUEUE_WAITING_THRESHOLD`). Signals a missing `maxWaiting`.
|
||||
|
||||
## Triage commands
|
||||
|
||||
```bash
|
||||
# Who's active right now?
|
||||
gbrain jobs list --status active
|
||||
|
||||
# Who's waiting, biggest pile first?
|
||||
gbrain jobs list --status waiting --limit 50
|
||||
|
||||
# What's wrong with a specific job?
|
||||
gbrain jobs get <id>
|
||||
```
|
||||
|
||||
## Rescue actions (in order of escalation)
|
||||
|
||||
```bash
|
||||
# Force-kill a single stuck job:
|
||||
gbrain jobs cancel <id>
|
||||
|
||||
# Clear a specific job entirely (last resort):
|
||||
gbrain jobs delete <id>
|
||||
|
||||
# Health smoke on the mechanism itself:
|
||||
gbrain jobs smoke --wedge-rescue
|
||||
```
|
||||
|
||||
## What each subcheck means
|
||||
|
||||
- **stalled-forever** — A worker claimed a job, started executing, and has
|
||||
held the row for over an hour. The wall-clock sweep evicts jobs past
|
||||
2× `timeout_ms`; if one's still active, either no `timeout_ms` was set
|
||||
or the sweep is newly deployed and this job predates it. Cancel it.
|
||||
- **waiting-depth** — Submitters are piling up jobs faster than workers
|
||||
drain them. Set `--max-waiting N` on the submission or on the programmatic
|
||||
`queue.add()` call. If you want a taller pile, raise the threshold via
|
||||
`GBRAIN_QUEUE_WAITING_THRESHOLD=50 gbrain doctor`.
|
||||
|
||||
## Self-check: is a worker even running?
|
||||
|
||||
```bash
|
||||
# If you're running autopilot with --no-worker, check that your external
|
||||
# worker (systemd / Docker / OpenClaw service-manager) is alive:
|
||||
gbrain jobs list --status active | head -5
|
||||
```
|
||||
|
||||
If the list is empty AND your submissions keep piling up, no worker is
|
||||
claiming. Start one:
|
||||
|
||||
```bash
|
||||
GBRAIN_ALLOW_SHELL_JOBS=1 gbrain jobs work --concurrency 4
|
||||
```
|
||||
|
||||
## Follow-ups tracked for v0.20+
|
||||
|
||||
- B7 — `minion_workers` heartbeat table for ground-truth liveness (the
|
||||
`--no-worker` probe and the dropped `queue_health` worker-heartbeat
|
||||
subcheck both need this).
|
||||
- B3 — `gbrain doctor --fix` learns to rescue queue wedges.
|
||||
+215
-202
@@ -141,12 +141,13 @@ strict behavior when unset.
|
||||
- `src/commands/graph-query.ts` — `gbrain graph-query <slug> [--type T] [--depth N] [--direction in|out|both]`: typed-edge relationship traversal (renders indented tree)
|
||||
- `src/core/link-extraction.ts` — shared library for the v0.12.0 graph layer. extractEntityRefs (canonical, replaces backlinks.ts duplicate) matches both `[Name](people/slug)` markdown links and Obsidian `[[people/slug|Name]]` wikilinks as of v0.12.3. extractPageLinks, inferLinkType heuristics (attended/works_at/invested_in/founded/advises/source/mentions), parseTimelineEntries, isAutoLinkEnabled config helper. `DIR_PATTERN` covers `people`, `companies`, `deals`, `topics`, `concepts`, `projects`, `entities`, `tech`, `finance`, `personal`, `openclaw`. Used by extract.ts, operations.ts auto-link post-hook, and backlinks.ts.
|
||||
- `src/core/minions/` — Minions job queue: BullMQ-inspired, Postgres-native (queue, worker, backoff, types, protected-names, quiet-hours, stagger, handlers/shell).
|
||||
- `src/core/minions/queue.ts` — MinionQueue class (submit, claim, complete, fail, stall detection, parent-child, depth/child-cap, per-job timeouts, cascade-kill, attachments, idempotency keys, child_done inbox, removeOnComplete/Fail). `add()` takes a 4th `trusted` arg (separate from `opts` to prevent spread leakage); protected names in `PROTECTED_JOB_NAMES` require `{allowProtectedSubmit: true}` and the check runs trim-normalized (whitespace-bypass safe). v0.14.1 #219: `add()` plumbs `max_stalled` through with a `[1, 100]` clamp; omitted values let the schema DEFAULT (5) kick in.
|
||||
- `src/core/minions/queue.ts` — MinionQueue class (submit, claim, complete, fail, stall detection, parent-child, depth/child-cap, per-job timeouts, cascade-kill, attachments, idempotency keys, child_done inbox, removeOnComplete/Fail). `add()` takes a 4th `trusted` arg (separate from `opts` to prevent spread leakage); protected names in `PROTECTED_JOB_NAMES` require `{allowProtectedSubmit: true}` and the check runs trim-normalized (whitespace-bypass safe). v0.14.1 #219: `add()` plumbs `max_stalled` through with a `[1, 100]` clamp; omitted values let the schema DEFAULT (5) kick in. v0.19.0: `handleWallClockTimeouts(lockDurationMs)` is Layer 3 kill shot for jobs where `FOR UPDATE SKIP LOCKED` stall detection and the timeout sweep both fail to evict (wedged worker holding a row lock via a pending transaction). v0.19.1: `maxWaiting` coalesce path now uses `pg_advisory_xact_lock` keyed on `(name, queue)` to serialize concurrent submits for the same key, and filters on `queue` in addition to `name` so cross-queue same-name jobs don't suppress each other.
|
||||
- `src/core/minions/worker.ts` — MinionWorker class (handler registry, lock renewal, graceful shutdown, timeout safety net). v0.14.0 abort-path fix: aborted jobs now call `failJob` with reason (`timeout`/`cancel`/`lock-lost`/`shutdown`) instead of returning silently. `shutdownAbort` (instance field) fires on process SIGTERM/SIGINT and propagates to `ctx.shutdownSignal` — shell handler listens to it; non-shell handlers don't.
|
||||
- `src/core/minions/types.ts` — `MinionJobInput` + `MinionJobStatus` + handler context types. `MinionJobInput.max_stalled` (new in v0.14.1) is optional; omitted values let the schema DEFAULT (5) kick in, provided values are clamped to `[1, 100]`.
|
||||
- `src/core/minions/protected-names.ts` — side-effect-free constant module exporting `PROTECTED_JOB_NAMES` + `isProtectedJobName()`. Kept pure so queue core can import without loading handler modules.
|
||||
- `src/core/minions/handlers/shell.ts` — `shell` job handler. Spawns `/bin/sh -c cmd` (absolute path, PATH-override-safe) or `argv[0] argv[1..]` (no shell). Env allowlist: `PATH, HOME, USER, LANG, TZ, NODE_ENV` + caller `env:` overrides. UTF-8-safe stdout/stderr tail via `string_decoder.StringDecoder`. Abort (either `ctx.signal` or `ctx.shutdownSignal`) fires SIGTERM → 5s grace → SIGKILL on child. Requires `GBRAIN_ALLOW_SHELL_JOBS=1` on worker (gated by `registerBuiltinHandlers`).
|
||||
- `src/core/minions/handlers/shell-audit.ts` — per-submission JSONL audit trail at `~/.gbrain/audit/shell-jobs-YYYY-Www.jsonl` (ISO-week rotation; override via `GBRAIN_AUDIT_DIR`). Best-effort: `mkdirSync(recursive)` + `appendFileSync`; failures logged to stderr, submission not blocked. Logs cmd (first 80 chars) or argv (JSON array). Never logs env values.
|
||||
- `src/core/minions/backpressure-audit.ts` (v0.19.1) — sibling of shell-audit.ts for `maxWaiting` coalesce events. JSONL at `~/.gbrain/audit/backpressure-YYYY-Www.jsonl`. Fires one line per coalesce with `(queue, name, waiting_count, max_waiting, returned_job_id, ts)`. Closes the silent-drop vector the v0.19.0 maxWaiting guard introduced.
|
||||
- `src/core/minions/handlers/subagent.ts` (v0.15) — LLM-loop handler. Two-phase tool persistence (pending → complete/failed), replay reconciliation for mid-dispatch crashes, dual-signal abort (`ctx.signal` + `ctx.shutdownSignal`), Anthropic prompt caching on system + tool defs. `makeSubagentHandler({engine, client?, ...})` factory; `MessagesClient` is an injectable interface the real SDK implements structurally. Throws `RateLeaseUnavailableError` (renewable) when rate-lease capacity is full.
|
||||
- `src/core/minions/handlers/subagent-aggregator.ts` (v0.15) — `subagent_aggregator` handler. Claims AFTER all children resolve (queue changes guarantee every terminal child posts a `child_done` inbox message with outcome). Reads inbox via `ctx.readInbox()`, builds deterministic mixed-outcome markdown summary. No LLM call in v0.15.
|
||||
- `src/core/minions/handlers/subagent-audit.ts` (v0.15) — JSONL audit + heartbeat writer at `~/.gbrain/audit/subagent-jobs-YYYY-Www.jsonl`. Events: `submission` (one line per submit) + `heartbeat` (per turn boundary: `llm_call_started | llm_call_completed | tool_called | tool_result | tool_failed`). Never logs prompts or tool inputs. `readSubagentAuditForJob(jobId, {sinceIso})` is the readback path for `gbrain agent logs`.
|
||||
@@ -168,7 +169,7 @@ strict behavior when unset.
|
||||
- `src/commands/migrations/` — TS migration registry (compiled into the binary; no filesystem walk of `skills/migrations/*.md` needed at runtime). `index.ts` lists migrations in semver order. `v0_11_0.ts` = Minions adoption orchestrator (8 phases). `v0_12_0.ts` = Knowledge Graph auto-wire orchestrator (5 phases: schema → config check → backfill links → backfill timeline → verify). `phaseASchema` has a 600s timeout (bumped from 60s in v0.12.1 for duplicate-heavy brains). `v0_12_2.ts` = JSONB double-encode repair orchestrator (4 phases: schema → repair-jsonb → verify → record). `v0_14_0.ts` = shell-jobs + autopilot cooperative (2 phases: schema ALTER minion_jobs.max_stalled SET DEFAULT 3 — superseded by v0.14.3's schema-level DEFAULT 5 + UPDATE backfill; pending-host-work ping for skills/migrations/v0.14.0.md). All orchestrators are idempotent and resumable from `partial` status. As of v0.14.2 (Bug 3), the RUNNER owns all ledger writes — orchestrators return `OrchestratorResult` and `apply-migrations.ts` persists a canonical `{version, status, phases}` shape after return. Orchestrators no longer call `appendCompletedMigration` directly. `statusForVersion` prefers `complete` over `partial` (never regresses). 3 consecutive partials → wedged → `--force-retry <version>` writes a `'retry'` reset marker. v0.14.3 (fix wave) ships schema-only migrations v14 (`pages_updated_at_index`) + v15 (`minion_jobs_max_stalled_default_5` with UPDATE backfill) via the `MIGRATIONS` array in `src/core/migrate.ts` — no orchestrator phases needed.
|
||||
- `src/commands/repair-jsonb.ts` — `gbrain repair-jsonb [--dry-run] [--json]`: rewrites `jsonb_typeof='string'` rows in place across 5 affected columns (pages.frontmatter, raw_data.data, ingest_log.pages_updated, files.metadata, page_versions.frontmatter). Fixes v0.12.0 double-encode bug on Postgres; PGLite no-ops. Idempotent.
|
||||
- `src/commands/orphans.ts` — `gbrain orphans [--json] [--count] [--include-pseudo]`: surfaces pages with zero inbound wikilinks, grouped by domain. Auto-generated/raw/pseudo pages filtered by default. Also exposed as `find_orphans` MCP operation. Shipped in v0.12.3 (contributed by @knee5).
|
||||
- `src/commands/doctor.ts` — `gbrain doctor [--json] [--fast] [--fix] [--dry-run] [--index-audit]`: health checks. v0.12.3 added `jsonb_integrity` + `markdown_body_completeness` reliability checks. v0.14.1: `--fix` delegates inlined cross-cutting rules to `> **Convention:** see [path](path).` callouts (pipes DRY violations into `src/core/dry-fix.ts`); `--fix --dry-run` previews without writing. v0.14.2: `schema_version` check fails loudly when `version=0` (migrations never ran — the #218 `bun install -g` signature) and routes users to `gbrain apply-migrations --yes`; new opt-in `--index-audit` flag (Postgres-only) reports zero-scan indexes from `pg_stat_user_indexes` (informational only, no auto-drop). v0.15.2: every DB check is wrapped in a progress phase; `markdown_body_completeness` runs under a 1s heartbeat timer so 10+ min scans are observable on 50K-page brains. Fix hints point at `gbrain repair-jsonb`, `gbrain sync --force`, and `gbrain apply-migrations`.
|
||||
- `src/commands/doctor.ts` — `gbrain doctor [--json] [--fast] [--fix] [--dry-run] [--index-audit]`: health checks. v0.12.3 added `jsonb_integrity` + `markdown_body_completeness` reliability checks. v0.14.1: `--fix` delegates inlined cross-cutting rules to `> **Convention:** see [path](path).` callouts (pipes DRY violations into `src/core/dry-fix.ts`); `--fix --dry-run` previews without writing. v0.14.2: `schema_version` check fails loudly when `version=0` (migrations never ran — the #218 `bun install -g` signature) and routes users to `gbrain apply-migrations --yes`; new opt-in `--index-audit` flag (Postgres-only) reports zero-scan indexes from `pg_stat_user_indexes` (informational only, no auto-drop). v0.15.2: every DB check is wrapped in a progress phase; `markdown_body_completeness` runs under a 1s heartbeat timer so 10+ min scans are observable on 50K-page brains. v0.19.1 added `queue_health` (Postgres-only) with two subchecks: stalled-forever active jobs (started_at > 1h) and waiting-depth-per-name > threshold (default 10, override via `GBRAIN_QUEUE_WAITING_THRESHOLD`). Worker-heartbeat subcheck intentionally deferred to follow-up B7 because it needs a `minion_workers` table to produce ground-truth signal. Fix hints point at `gbrain repair-jsonb`, `gbrain sync --force`, `gbrain apply-migrations`, and `gbrain jobs get/cancel <id>`.
|
||||
- `src/core/migrate.ts` — schema-migration runner. Owns the `MIGRATIONS` array (source of truth for schema DDL). v0.14.2 extended the `Migration` interface with `sqlFor?: { postgres?, pglite? }` (engine-specific SQL overrides `sql`) and `transaction?: boolean` (set to false for `CREATE INDEX CONCURRENTLY`, which Postgres refuses inside a transaction; ignored on PGLite since it has no concurrent writers). Migration v14 (fix wave) uses a handler branching on `engine.kind` to run CONCURRENTLY on Postgres (with a pre-drop of any invalid remnant via `pg_index.indisvalid`) and plain `CREATE INDEX` on PGLite. v15 bumps `minion_jobs.max_stalled` default 1→5 and backfills existing non-terminal rows.
|
||||
- `src/core/progress.ts` — Shared bulk-action progress reporter. Writes to stderr. Modes: `auto` (TTY: `\r`-rewriting; non-TTY: plain lines), `human`, `json` (JSONL), `quiet`. Rate-gated by `minIntervalMs` and `minItems`. `startHeartbeat(reporter, note)` helper for single long queries. `child()` composes phase paths. Singleton SIGINT/SIGTERM coordinator emits `abort` events for every live phase. EPIPE defense on both sync throws and stream `'error'` events. Zero dependencies. Introduced in v0.15.2.
|
||||
- `src/core/cli-options.ts` — Global CLI flag parser. `parseGlobalFlags(argv)` returns `{cliOpts, rest}` with `--quiet` / `--progress-json` / `--progress-interval=<ms>` stripped. `getCliOptions()` / `setCliOptions()` expose a module-level singleton so commands reach the resolved flags without parameter threading. `cliOptsToProgressOptions()` maps to reporter options. `childGlobalFlags()` returns the flag suffix to append to `execSync('gbrain ...')` calls in migration orchestrators. `OperationContext.cliOpts` extends shared-op dispatch for MCP callers.
|
||||
@@ -1403,9 +1404,12 @@ The six daily pains — spawn storms, agents that stop responding, forgotten dis
|
||||
gbrain jobs smoke # verify install
|
||||
gbrain jobs submit sync --params '{}' # fire a background job
|
||||
gbrain jobs stats # health dashboard
|
||||
gbrain jobs work --concurrency 4 # start a worker (Postgres only)
|
||||
gbrain jobs supervisor --concurrency 4 # canonical: auto-restarting worker (Postgres only)
|
||||
gbrain jobs work --concurrency 4 # raw worker (no crash recovery — prefer `supervisor`)
|
||||
```
|
||||
|
||||
`gbrain jobs supervisor` keeps the worker alive across crashes with exponential backoff, atomic PID locking, structured audit events at `~/.gbrain/audit/supervisor-*.jsonl`, and a `start --detach` / `status --json` / `stop` subcommand surface for agents. In containers it runs as PID 1; on systemd hosts it's the child of `gbrain-worker.service`. Full deployment guide: [`docs/guides/minions-deployment.md`](docs/guides/minions-deployment.md).
|
||||
|
||||
Read [`skills/minion-orchestrator/SKILL.md`](skills/minion-orchestrator/SKILL.md) for parent-child DAGs, fan-in collection, steering via inbox.
|
||||
|
||||
**Minions is not incrementally better than sub-agents for background work. It's categorically different.** 753ms vs gateway timeout. $0 vs tokens. 100% vs couldn't-spawn. If your agent does deterministic work on a schedule, it runs on Minions now.
|
||||
@@ -3496,8 +3500,8 @@ Source: https://raw.githubusercontent.com/garrytan/gbrain/master/docs/guides/min
|
||||
|
||||
# Minions Worker Deployment Guide
|
||||
|
||||
Deploy `gbrain jobs work` so it stays running across crashes, reboots, and
|
||||
Postgres connection blips. Written for agents to execute line-by-line.
|
||||
Keep `gbrain jobs work` running across crashes, reboots, and Postgres
|
||||
connection blips. Written for agents to execute line-by-line.
|
||||
|
||||
## The problem
|
||||
|
||||
@@ -3508,10 +3512,61 @@ The persistent worker can die silently from:
|
||||
- Bun process crashes with no automatic restart.
|
||||
- Internal event-loop death (PID alive, worker loop stopped).
|
||||
|
||||
When the worker dies, submitted jobs sit in `waiting` forever. Nothing in
|
||||
gbrain core auto-restarts the worker — that's what this guide wires up.
|
||||
When the worker dies, submitted jobs sit in `waiting` forever. The
|
||||
canonical answer is `gbrain jobs supervisor` — a first-class CLI that
|
||||
spawns `gbrain jobs work` as a child and auto-restarts it on crash.
|
||||
|
||||
## Variables used in this guide
|
||||
## Worker supervision
|
||||
|
||||
### The canonical pattern
|
||||
|
||||
`gbrain jobs supervisor` is an auto-restarting wrapper around
|
||||
`gbrain jobs work`. It writes a PID file, restarts the worker on crash
|
||||
with exponential backoff (1s → 60s cap), emits lifecycle events to an
|
||||
audit file, and drains gracefully on SIGTERM (35s worker-drain window
|
||||
before SIGKILL). Exit codes are documented so agents can branch on them.
|
||||
|
||||
**Typical commands:**
|
||||
|
||||
```bash
|
||||
# Start in the foreground (blocks; Ctrl-C to stop).
|
||||
gbrain jobs supervisor --concurrency 4
|
||||
|
||||
# Start detached — returns {"event":"started","supervisor_pid":…} on stdout.
|
||||
gbrain jobs supervisor start --detach --json
|
||||
|
||||
# Check liveness without reading log files.
|
||||
gbrain jobs supervisor status --json
|
||||
|
||||
# Graceful stop (SIGTERM + drain wait + SIGKILL fallback).
|
||||
gbrain jobs supervisor stop
|
||||
```
|
||||
|
||||
**Exit codes:**
|
||||
|
||||
| Code | Meaning |
|
||||
|---|---|
|
||||
| 0 | Clean shutdown (SIGTERM/SIGINT received, worker drained) |
|
||||
| 1 | Max crashes exceeded (worker kept dying) |
|
||||
| 2 | Another supervisor holds the PID lock |
|
||||
| 3 | PID file unwritable (permission / path error) |
|
||||
|
||||
An agent seeing exit=2 can safely treat it as "one is already running";
|
||||
exit=1 should page a human.
|
||||
|
||||
### Which supervisor when?
|
||||
|
||||
The supervisor solves in-process crash recovery. Platform-level
|
||||
supervision (systemd, Fly, Render) handles host-level failures. You
|
||||
usually want both.
|
||||
|
||||
| Environment | Recommendation |
|
||||
|---|---|
|
||||
| **Container (Fly / Railway / Render / Heroku)** | `gbrain jobs supervisor` runs as PID 1. The platform restarts the container on OOM / host loss; supervisor restarts the worker on crash. See [Fly.io](#flyio) / [Render / Railway / Heroku](#render--railway--heroku). |
|
||||
| **Linux VM with systemd** | Two-layer recommended: systemd supervises `gbrain jobs supervisor`, which in turn supervises `gbrain jobs work`. Buys you automatic restart on reboot (systemd) plus fast crash recovery (supervisor). See [systemd](#systemd). |
|
||||
| **Dev laptop / macOS** | `gbrain jobs supervisor` in a terminal. Ctrl-C stops it. No system-level setup needed. |
|
||||
|
||||
### Variables used in this guide
|
||||
|
||||
Substitute these once before copy-pasting any snippet.
|
||||
|
||||
@@ -3519,142 +3574,122 @@ Substitute these once before copy-pasting any snippet.
|
||||
|---|---|---|
|
||||
| `$GBRAIN_BIN` | Absolute path to the `gbrain` binary | `$(command -v gbrain)` — often `/usr/local/bin/gbrain` or `~/.bun/bin/gbrain` |
|
||||
| `$GBRAIN_WORKER_USER` | OS user that owns the worker process | the same user that ran `gbrain init`; never `root` |
|
||||
| `$GBRAIN_WORKER_PID_FILE` | Worker PID + restart-epoch file | `/tmp/gbrain-worker.pid` (or `/var/run/gbrain/worker.pid` for systemd) |
|
||||
| `$GBRAIN_WORKER_LOG_FILE` | Worker log sink (stdout + stderr merged) | `/tmp/gbrain-worker.log` (or `/var/log/gbrain/worker.log`) |
|
||||
| `$GBRAIN_WORKSPACE` | `cwd` for shell jobs submitted by this deployment | absolute path, e.g. `/srv/my-brain` |
|
||||
| `$GBRAIN_ENV_FILE` | Secrets file sourced by crontab / systemd | `/etc/gbrain.env` (mode 600) |
|
||||
| `$GBRAIN_ENV_FILE` | Secrets file sourced by systemd / shell | `/etc/gbrain.env` (mode 600) |
|
||||
|
||||
## Preconditions
|
||||
### Preconditions
|
||||
|
||||
Run these before Step 1 of any option. Fail fast if something is wrong.
|
||||
Run these before any deployment step.
|
||||
|
||||
```bash
|
||||
# 1. gbrain is on PATH and resolves to an absolute location.
|
||||
command -v gbrain || { echo "gbrain not on PATH. Install, then retry."; exit 1; }
|
||||
|
||||
# 2. DATABASE_URL points at reachable Postgres (or PGLite path exists).
|
||||
# 2. DATABASE_URL points at reachable Postgres.
|
||||
# (Supervisor is Postgres-only. PGLite's exclusive file lock blocks the
|
||||
# separate worker process. If `config.engine === 'pglite'` the CLI rejects
|
||||
# with a clear error.)
|
||||
gbrain doctor --fast --json | jq '.checks[] | select(.name=="db_connectivity")'
|
||||
|
||||
# 3. Schema is up to date. If version=0 or status=="fail", fix it first:
|
||||
# 3. Schema is up to date. If version=0 or status=="fail":
|
||||
# gbrain apply-migrations --yes
|
||||
gbrain doctor --fast --json | jq '.checks[] | select(.name=="schema_version")'
|
||||
|
||||
# 4. You have write access to at least one crontab mechanism.
|
||||
crontab -l >/dev/null 2>&1 && echo "user crontab OK"
|
||||
[ -w /etc/crontab ] && echo "/etc/crontab OK"
|
||||
|
||||
# 5. If you plan to submit `shell` jobs, the WORKER process needs
|
||||
# GBRAIN_ALLOW_SHELL_JOBS=1 (submitters do not). The handler is gated
|
||||
# in registerBuiltinHandlers(); without the flag the worker startup
|
||||
# line reads "shell handler disabled (...)".
|
||||
# 4. If you plan to submit `shell` jobs, pass --allow-shell-jobs to the
|
||||
# supervisor (or export GBRAIN_ALLOW_SHELL_JOBS=1 before starting).
|
||||
# Without the flag, the shell handler is disabled at worker startup.
|
||||
```
|
||||
|
||||
## Which option?
|
||||
## Agent usage (OpenClaw / Hermes / Cursor / Codex)
|
||||
|
||||
- Your workload runs LLM subagents (`gbrain agent run`) or jobs that take
|
||||
> 30 s → **Option 1** (watchdog cron + persistent worker).
|
||||
- Your workload is short deterministic scripts on a fixed schedule (every
|
||||
3 h, daily, weekly) → **Option 2** (inline `--follow`).
|
||||
- You don't have shell access to a long-running box (Fly/Render/Railway,
|
||||
or any systemd host) → **Option 3** (service manager — replaces cron).
|
||||
|
||||
## Option 1: watchdog cron + persistent worker
|
||||
|
||||
A 5-minute cron checks whether the worker process is alive **and** whether
|
||||
it has logged an internal shutdown since its last start. Restarts if either
|
||||
condition fails.
|
||||
|
||||
### 1a. Install the env file (secrets stay out of crontab)
|
||||
|
||||
Never paste `DATABASE_URL` or API keys into crontab. `/etc/crontab` is
|
||||
mode 644 (world-readable); user crontabs under `/var/spool/cron/` are
|
||||
readable by `root`. Use the shipped env-file template:
|
||||
Three-command pattern an agent can drive without shell archaeology:
|
||||
|
||||
```bash
|
||||
sudo install -m 600 -o $GBRAIN_WORKER_USER -g $GBRAIN_WORKER_USER \
|
||||
# Start (returns PIDs + pid_file on stdout as JSON, then detaches)
|
||||
gbrain jobs supervisor start --detach --json
|
||||
# → {"event":"started","supervisor_pid":1234,"worker_pid":1235,"pid_file":"/Users/you/.gbrain/supervisor.pid"}
|
||||
|
||||
# Check health (machine-parseable JSON, no log scraping)
|
||||
gbrain jobs supervisor status --json
|
||||
# → {"running":true,"supervisor_pid":1234,"last_start":"2026-04-23T15:30:22Z","crashes_24h":0, ...}
|
||||
|
||||
# Stop cleanly (SIGTERM + 35s drain + SIGKILL fallback)
|
||||
gbrain jobs supervisor stop
|
||||
```
|
||||
|
||||
Every lifecycle event (spawn, crash, backoff, health warning, max-crashes,
|
||||
shutdown) is also written to `${GBRAIN_AUDIT_DIR:-~/.gbrain/audit}/supervisor-YYYY-Www.jsonl`
|
||||
for historical inspection. `gbrain doctor` reads that file and surfaces
|
||||
a `supervisor` check in its health report.
|
||||
|
||||
## Deployment: systemd
|
||||
|
||||
For long-running Linux VMs with shell access.
|
||||
|
||||
```bash
|
||||
# Create the worker user if it doesn't exist.
|
||||
sudo useradd --system --home "$GBRAIN_WORKSPACE" --shell /usr/sbin/nologin gbrain \
|
||||
2>/dev/null || true
|
||||
sudo mkdir -p "$GBRAIN_WORKSPACE" && sudo chown gbrain:gbrain "$GBRAIN_WORKSPACE"
|
||||
|
||||
# Install the env file (secrets stay out of the unit file).
|
||||
sudo install -m 600 -o gbrain -g gbrain \
|
||||
docs/guides/minions-deployment-snippets/gbrain.env.example /etc/gbrain.env
|
||||
sudoedit /etc/gbrain.env
|
||||
# Fill in DATABASE_URL, optional GBRAIN_ALLOW_SHELL_JOBS=1.
|
||||
|
||||
# Install the unit file, substituting /srv/gbrain → your workspace path.
|
||||
sudo install -m 644 docs/guides/minions-deployment-snippets/systemd.service \
|
||||
/etc/systemd/system/gbrain-worker.service
|
||||
sudo sed -i "s|/srv/gbrain|$GBRAIN_WORKSPACE|g" \
|
||||
/etc/systemd/system/gbrain-worker.service
|
||||
|
||||
sudo systemctl daemon-reload
|
||||
sudo systemctl enable --now gbrain-worker
|
||||
sudo systemctl status gbrain-worker
|
||||
journalctl -u gbrain-worker -n 50
|
||||
```
|
||||
|
||||
Fill in the connection string and `GBRAIN_ALLOW_SHELL_JOBS=1` (if
|
||||
applicable). See
|
||||
[`gbrain.env.example`](./minions-deployment-snippets/gbrain.env.example)
|
||||
for the full list.
|
||||
The shipped unit file invokes `gbrain jobs supervisor` (not `gbrain jobs work`
|
||||
directly) so you get two-layer supervision: systemd restarts the supervisor
|
||||
on host reboot, supervisor restarts the worker on in-process crash.
|
||||
|
||||
### 1b. Install the watchdog script
|
||||
`Restart=always` + `RestartSec=10s` handle the supervisor-level recovery.
|
||||
The unit runs as unprivileged `gbrain` with `PrivateTmp`, `ProtectSystem=strict`,
|
||||
and `ReadWritePaths=$GBRAIN_WORKSPACE,$HOME/.gbrain` (for the PID file and
|
||||
audit log). `LimitNOFILE=65535` covers Bun + Postgres pool + concurrent
|
||||
LLM subagent calls without hitting the default 1024 cap.
|
||||
|
||||
The [`minion-watchdog.sh`](./minions-deployment-snippets/minion-watchdog.sh)
|
||||
ships in-repo and writes a two-line PID file (PID on line 1, restart epoch
|
||||
on line 2). The restart-epoch marker is how the watchdog distinguishes
|
||||
stale shutdown lines in the log from current ones — without it, every tick
|
||||
after the first restart would match an old `worker shutting down` line and
|
||||
loop forever.
|
||||
|
||||
Requires GNU coreutils (Linux default). On macOS/BSD install via
|
||||
`brew install coreutils` and alias `date` to `gdate` in the cron env if you
|
||||
want to test the watchdog locally; production Linux boxes work as-is.
|
||||
## Deployment: Fly.io
|
||||
|
||||
```bash
|
||||
sudo install -m 755 -o $GBRAIN_WORKER_USER -g $GBRAIN_WORKER_USER \
|
||||
docs/guides/minions-deployment-snippets/minion-watchdog.sh \
|
||||
/usr/local/bin/minion-watchdog.sh
|
||||
# Merge the [processes] block from fly.toml.partial into your fly.toml.
|
||||
cat docs/guides/minions-deployment-snippets/fly.toml.partial >> fly.toml
|
||||
# Review + edit as needed.
|
||||
|
||||
# Set secrets (Fly handles restart on crash).
|
||||
fly secrets set DATABASE_URL='postgres://…' GBRAIN_ALLOW_SHELL_JOBS=1
|
||||
```
|
||||
|
||||
### 1c. Wire into cron
|
||||
The `[processes]` block runs `gbrain jobs supervisor` as PID 1. Fly
|
||||
restarts the container on host failure; the supervisor restarts the
|
||||
worker on in-process crash.
|
||||
|
||||
Pick the form that matches the crontab you're editing.
|
||||
## Deployment: Render / Railway / Heroku
|
||||
|
||||
**If you ran `crontab -e`** (user crontab — 5-field, no user column):
|
||||
Drop [`Procfile`](./minions-deployment-snippets/Procfile) at the repo
|
||||
root. The shipped Procfile calls `gbrain jobs supervisor`. Set
|
||||
`DATABASE_URL` + optional `GBRAIN_ALLOW_SHELL_JOBS=1` via the platform's
|
||||
env UI or CLI.
|
||||
|
||||
```
|
||||
SHELL=/bin/bash
|
||||
PATH=/usr/local/bin:/usr/bin:/bin
|
||||
BASH_ENV=/etc/gbrain.env
|
||||
*/5 * * * * /usr/local/bin/minion-watchdog.sh
|
||||
```
|
||||
## Deployment: inline `--follow` (no persistent worker)
|
||||
|
||||
**If you edited `/etc/crontab` directly** (system crontab — 6-field, with
|
||||
user column):
|
||||
|
||||
```
|
||||
SHELL=/bin/bash
|
||||
PATH=/usr/local/bin:/usr/bin:/bin
|
||||
BASH_ENV=/etc/gbrain.env
|
||||
*/5 * * * * gbrain /usr/local/bin/minion-watchdog.sh
|
||||
```
|
||||
|
||||
In both forms, `BASH_ENV=/etc/gbrain.env` tells non-interactive bash to
|
||||
source the env file before running the watchdog — that's how the
|
||||
connection string and `GBRAIN_ALLOW_SHELL_JOBS` reach the worker without
|
||||
landing in the world-readable crontab itself.
|
||||
|
||||
### 1d. Log rotation
|
||||
|
||||
The watchdog appends to the worker log across restarts. If you expect the
|
||||
file to grow unbounded, rotate it externally with `logrotate`:
|
||||
|
||||
```
|
||||
# /etc/logrotate.d/gbrain-worker
|
||||
/tmp/gbrain-worker.log {
|
||||
daily
|
||||
rotate 7
|
||||
missingok
|
||||
notifempty
|
||||
copytruncate
|
||||
}
|
||||
```
|
||||
|
||||
`copytruncate` is important — the watchdog's restart-epoch check survives
|
||||
it (the epoch is compared against in-log timestamps, not file inode).
|
||||
|
||||
## Option 2: inline `--follow` (no persistent worker)
|
||||
|
||||
Each cron run brings its own temporary worker. `--follow` starts one on
|
||||
the queue and blocks until the just-submitted job reaches a terminal state
|
||||
(`completed` / `failed` / `dead` / `cancelled`). 2-3 s startup overhead
|
||||
per job; negligible vs job duration for scheduled work.
|
||||
|
||||
Example: nightly brain enrichment as a shell job.
|
||||
For short deterministic scripts on a fixed schedule where you don't need
|
||||
a persistent worker between runs. Each cron run brings its own temporary
|
||||
worker. `--follow` starts one on the queue and blocks until the
|
||||
just-submitted job reaches a terminal state (`completed` / `failed` /
|
||||
`dead` / `cancelled`). 2-3 s startup overhead per job; negligible vs job
|
||||
duration for scheduled work.
|
||||
|
||||
```bash
|
||||
GBRAIN_ALLOW_SHELL_JOBS=1 gbrain jobs submit shell \
|
||||
@@ -3666,85 +3701,56 @@ GBRAIN_ALLOW_SHELL_JOBS=1 gbrain jobs submit shell \
|
||||
|
||||
Replace `gbrain embed --stale` with whichever gbrain subcommand you're
|
||||
scheduling (`sync`, `extract`, `orphans`, `doctor`, `check-backlinks`,
|
||||
`lint`, `autopilot`). If you're shelling out to a non-gbrain binary,
|
||||
keep its absolute path in the `cmd`.
|
||||
|
||||
**Shared-queue gotcha.** If other jobs are already waiting on the same
|
||||
queue with higher priority or earlier `created_at`, the temporary worker
|
||||
processes those first before reaching yours. `--follow` still exits only
|
||||
when YOUR job finishes. For strict single-job semantics on shared queues,
|
||||
`lint`, `autopilot`). For strict single-job semantics on shared queues,
|
||||
use a dedicated queue name like `nightly-enrich` above.
|
||||
|
||||
## Option 3: service manager (systemd / Fly / Render / Railway)
|
||||
## Upgrading from an older deployment
|
||||
|
||||
Replaces the watchdog entirely. No cron, no PID file, no restart-loop.
|
||||
The service manager owns liveness.
|
||||
### From `minion-watchdog.sh` (pre-v0.20)
|
||||
|
||||
### systemd (Linux hosts with shell access)
|
||||
Earlier versions of this guide shipped a 68-line bash watchdog
|
||||
(`minion-watchdog.sh`). It's been replaced by `gbrain jobs supervisor`
|
||||
which handles everything the script did, plus atomic PID locking,
|
||||
structured audit events, queue-scoped health checks, and graceful
|
||||
drain on SIGTERM.
|
||||
|
||||
**Migration:**
|
||||
|
||||
```bash
|
||||
# Create the worker user if it doesn't exist.
|
||||
sudo useradd --system --home "$GBRAIN_WORKSPACE" --shell /usr/sbin/nologin gbrain \
|
||||
2>/dev/null || true
|
||||
sudo mkdir -p "$GBRAIN_WORKSPACE" && sudo chown gbrain:gbrain "$GBRAIN_WORKSPACE"
|
||||
# 1. Stop and remove the old watchdog.
|
||||
sudo kill $(head -n1 /tmp/gbrain-worker.pid) 2>/dev/null
|
||||
sudo rm -f /usr/local/bin/minion-watchdog.sh /tmp/gbrain-worker.pid \
|
||||
/tmp/gbrain-worker.log
|
||||
crontab -e # delete the "*/5 * * * * /usr/local/bin/minion-watchdog.sh" line
|
||||
|
||||
# Install the unit file, substituting /srv/gbrain → your workspace path.
|
||||
sudo install -m 644 docs/guides/minions-deployment-snippets/systemd.service \
|
||||
/etc/systemd/system/gbrain-worker.service
|
||||
sudo sed -i "s|/srv/gbrain|$GBRAIN_WORKSPACE|g" \
|
||||
/etc/systemd/system/gbrain-worker.service
|
||||
# 2. Start the supervisor (systemd users: reinstall the unit from
|
||||
# docs/guides/minions-deployment-snippets/systemd.service, which
|
||||
# now calls `gbrain jobs supervisor`).
|
||||
gbrain jobs supervisor start --detach --json
|
||||
# Or: sudo systemctl restart gbrain-worker
|
||||
|
||||
# See 1a above for /etc/gbrain.env install.
|
||||
sudo systemctl daemon-reload
|
||||
sudo systemctl enable --now gbrain-worker
|
||||
sudo systemctl status gbrain-worker
|
||||
journalctl -u gbrain-worker -n 50
|
||||
# 3. Verify.
|
||||
gbrain jobs supervisor status --json
|
||||
gbrain doctor # 'supervisor' check should report running=true
|
||||
```
|
||||
|
||||
`Restart=always` + `RestartSec=10s` give you crash-loop recovery. The unit
|
||||
runs as an unprivileged `gbrain` user with `PrivateTmp`, `ProtectSystem=strict`,
|
||||
and `ReadWritePaths=$GBRAIN_WORKSPACE`. `LimitNOFILE=65535` in the shipped
|
||||
unit covers Bun + Postgres pool + concurrent LLM subagent calls without
|
||||
hitting the default 1024 cap.
|
||||
### Schema / migration hygiene
|
||||
|
||||
### Fly.io
|
||||
Regardless of which deployment path you're upgrading from:
|
||||
|
||||
Merge the `[processes]` block from
|
||||
[`fly.toml.partial`](./minions-deployment-snippets/fly.toml.partial) into
|
||||
your existing `fly.toml`. Set secrets with `fly secrets set` —
|
||||
Fly auto-restarts the process on crash.
|
||||
|
||||
### Render / Railway / Heroku
|
||||
|
||||
Drop [`Procfile`](./minions-deployment-snippets/Procfile) at the repo root.
|
||||
Set the connection string and `GBRAIN_ALLOW_SHELL_JOBS=1` via the
|
||||
platform's env UI or CLI.
|
||||
|
||||
## Upgrading an existing deployment
|
||||
|
||||
If you deployed on v0.13.x or earlier, walk this checklist:
|
||||
|
||||
1. **Stop the worker before upgrading.**
|
||||
`kill $(head -n1 /tmp/gbrain-worker.pid)` and wait for the process to
|
||||
exit. Skipping this risks an in-flight job landing partial schema.
|
||||
1. **Stop the worker before upgrading.** `gbrain jobs supervisor stop`
|
||||
(or `sudo systemctl stop gbrain-worker`). Skipping this risks an
|
||||
in-flight job landing partial schema.
|
||||
2. **Run `gbrain upgrade`**. Then `gbrain apply-migrations --yes` if
|
||||
`gbrain doctor` reports any migration as `partial` or `pending`.
|
||||
3. **If you run shell jobs:** from v0.14 onward, the worker requires
|
||||
`GBRAIN_ALLOW_SHELL_JOBS=1` to register the `shell` handler. Add it to
|
||||
`/etc/gbrain.env`. Submitters don't need the flag; only the worker does.
|
||||
4. **If you tuned your watchdog for `max_stalled=1`:** v0.14.3 migration
|
||||
v15 raised the schema default to 5 and backfilled existing non-terminal
|
||||
rows. A watchdog tuned around 1-strike dead-lettering will now
|
||||
over-restart because it takes 5 misses to dead-letter. Switch to the
|
||||
shipped watchdog (which keys on log markers, not job state).
|
||||
5. **If your v0.16.1 watchdog is still running:** it has a restart-loop
|
||||
bug (old shutdown lines in the unrotated log re-match every 5 min
|
||||
forever). Install the current `minion-watchdog.sh` from this guide's
|
||||
snippets — it writes a restart epoch into the PID file and only
|
||||
considers log lines newer than that epoch.
|
||||
6. **Verify.** `gbrain doctor` should report zero `pending` or `partial`
|
||||
migrations. `gbrain jobs stats` should show no unexplained growth in
|
||||
`dead` between pre- and post-upgrade.
|
||||
3. **If you run shell jobs:** from v0.14 onward, pass
|
||||
`--allow-shell-jobs` to the supervisor (or keep
|
||||
`GBRAIN_ALLOW_SHELL_JOBS=1` in `/etc/gbrain.env`). Submitters don't
|
||||
need the flag; only the worker does.
|
||||
4. **Verify.** `gbrain doctor` should report zero `pending` or `partial`
|
||||
migrations plus a healthy `supervisor` check. `gbrain jobs stats`
|
||||
should show no unexplained growth in `dead` between pre- and
|
||||
post-upgrade.
|
||||
|
||||
## Known issues
|
||||
|
||||
@@ -3757,9 +3763,10 @@ silently. The stall detector then dead-letters the job after
|
||||
|
||||
**Current defaults that make this worse:**
|
||||
|
||||
- `lockDuration: 30000` (30 s) — too short for long jobs during connection blips.
|
||||
- `max_stalled: 5` (schema column default on master — see `src/schema.sql`
|
||||
and `src/core/pglite-schema.ts`). Five missed heartbeats before dead-letter.
|
||||
- `lockDuration: 30000` (30 s) — too short for long jobs during
|
||||
connection blips.
|
||||
- `max_stalled: 5` (schema column default — see `src/schema.sql` and
|
||||
`src/core/pglite-schema.ts`). Five missed heartbeats before dead-letter.
|
||||
- `stalledInterval: 30000` (30 s) — checks too aggressively.
|
||||
|
||||
**Tune per-job today.** `gbrain jobs submit` accepts `--max-stalled N`,
|
||||
@@ -3767,9 +3774,6 @@ silently. The stall detector then dead-letters the job after
|
||||
`--backoff-jitter 0..1`, and `--timeout-ms N` as first-class flags
|
||||
(since v0.13.1). These write onto the job row at submit time — which is
|
||||
what `handleStalled()` reads — so per-job tuning is the real knob today.
|
||||
Worker-level `--lock-duration` / `--stall-interval` are on the roadmap;
|
||||
until they land, rely on per-job `--max-stalled` plus the watchdog (or
|
||||
systemd) for worker health.
|
||||
|
||||
### DO NOT pass `maxStalledCount` to `MinionWorker`
|
||||
|
||||
@@ -3780,16 +3784,16 @@ Use `gbrain jobs submit --max-stalled N` per-job instead.
|
||||
### Zombie shell children
|
||||
|
||||
When the Bun worker crashes hard, child processes from shell jobs can
|
||||
become zombies. The watchdog's 10 s `SIGTERM → SIGKILL` window covers the
|
||||
shell handler's 5 s child-kill grace (`KILL_GRACE_MS`). For long-running
|
||||
shell jobs, bump the watchdog's `sleep 10` to `sleep 30` so the worker
|
||||
has time to flush in-flight jobs before the kill.
|
||||
become zombies. The supervisor's SIGTERM → 35s drain → SIGKILL window
|
||||
covers the shell handler's 5 s child-kill grace (`KILL_GRACE_MS`). For
|
||||
long-running shell jobs, prefer timeouts via `--timeout-ms` on submit
|
||||
over relying on hard kills.
|
||||
|
||||
## Smoke test
|
||||
|
||||
```bash
|
||||
# Worker alive?
|
||||
kill -0 $(head -n1 /tmp/gbrain-worker.pid) 2>/dev/null && echo ALIVE || echo DEAD
|
||||
# Supervisor alive?
|
||||
gbrain jobs supervisor status --json | jq .running
|
||||
|
||||
# Aggregate queue health.
|
||||
gbrain jobs stats
|
||||
@@ -3800,23 +3804,32 @@ gbrain jobs list --status active --limit 10
|
||||
# Dead-lettered jobs.
|
||||
gbrain jobs list --status dead --limit 10
|
||||
|
||||
# Shell handler registered? (stderr banner merged into log via 2>&1.)
|
||||
grep "shell handler enabled" /tmp/gbrain-worker.log
|
||||
# Shell handler registered? (check supervisor audit log or worker stderr.)
|
||||
gbrain jobs supervisor status --json | jq '.worker_config.allow_shell_jobs'
|
||||
```
|
||||
|
||||
## Uninstall
|
||||
|
||||
- **Option 1 (watchdog cron):** `crontab -e`, delete the watchdog line.
|
||||
`kill $(head -n1 /tmp/gbrain-worker.pid) && rm /tmp/gbrain-worker.pid`.
|
||||
Optionally `sudo rm /etc/gbrain.env /usr/local/bin/minion-watchdog.sh`.
|
||||
- **Option 2 (inline `--follow`):** remove the cron entry. Nothing else to
|
||||
clean up — temporary workers exit with their jobs.
|
||||
- **Option 3 (systemd):** `sudo systemctl disable --now gbrain-worker`,
|
||||
then `sudo rm /etc/systemd/system/gbrain-worker.service /etc/gbrain.env`,
|
||||
then `sudo systemctl daemon-reload`.
|
||||
- **Option 3 (Fly/Render/Railway):** delete the `worker` process from
|
||||
`fly.toml` / `Procfile` and redeploy. Secrets set via `fly secrets`
|
||||
persist until `fly secrets unset`.
|
||||
**`gbrain jobs supervisor`** (foreground or `--detach`):
|
||||
|
||||
```bash
|
||||
gbrain jobs supervisor stop
|
||||
```
|
||||
|
||||
**systemd:**
|
||||
|
||||
```bash
|
||||
sudo systemctl disable --now gbrain-worker
|
||||
sudo rm /etc/systemd/system/gbrain-worker.service /etc/gbrain.env
|
||||
sudo systemctl daemon-reload
|
||||
```
|
||||
|
||||
**Fly / Render / Railway:** delete the `worker` process from `fly.toml`
|
||||
/ `Procfile` and redeploy. Secrets set via `fly secrets` persist until
|
||||
`fly secrets unset`.
|
||||
|
||||
**Inline `--follow`:** remove the cron entry. Nothing else to clean up
|
||||
— temporary workers exit with their jobs.
|
||||
|
||||
---
|
||||
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "gbrain",
|
||||
"version": "0.20.0",
|
||||
"version": "0.20.3",
|
||||
"description": "Postgres-native personal knowledge brain with hybrid RAG search",
|
||||
"type": "module",
|
||||
"main": "src/core/index.ts",
|
||||
|
||||
+8
-5
@@ -13,13 +13,14 @@ This is the dispatcher. Skills are the implementation. **Read the skill file bef
|
||||
|
||||
| Trigger | Skill |
|
||||
|---------|-------|
|
||||
| "What do we know about", "tell me about", "search for" | `skills/query/SKILL.md` |
|
||||
| "What do we know about", "tell me about", "search for", "search the brain", "brain search", "background on", "notes on this", "who is" | `skills/query/SKILL.md` |
|
||||
| "Who knows who", "relationship between", "connections", "graph query" | `skills/query/SKILL.md` (use graph-query) |
|
||||
| Creating/enriching a person or company page | `skills/enrich/SKILL.md` |
|
||||
| Where does a new file go? Filing rules | `skills/repo-architecture/SKILL.md` |
|
||||
| Fix broken citations in brain pages | `skills/citation-fixer/SKILL.md` |
|
||||
| "Fix broken citations", "citations are broken", "fix citations", "citation audit" | `skills/citation-fixer/SKILL.md` |
|
||||
| "Research", "track", "extract from email", "investor updates", "donations" | `skills/data-research/SKILL.md` |
|
||||
| Share a brain page as a link | `skills/publish/SKILL.md` |
|
||||
| "validate frontmatter", "check frontmatter", "brain lint", "fix frontmatter" | `skills/frontmatter-guard/SKILL.md` |
|
||||
|
||||
## Content & media ingestion
|
||||
|
||||
@@ -58,7 +59,7 @@ This is the dispatcher. Skills are the implementation. **Read the skill file bef
|
||||
| Cross-modal review, second opinion | `skills/cross-modal-review/SKILL.md` |
|
||||
| "Validate skills", skill health check | `skills/testing/SKILL.md` |
|
||||
| Webhook setup, external event processing | `skills/webhook-transforms/SKILL.md` |
|
||||
| "Spawn agent", "background task", "parallel tasks", "steer agent", "pause/resume agent" | `skills/minion-orchestrator/SKILL.md` |
|
||||
| "Spawn agent", "background task", "parallel tasks", "steer agent", "pause/resume agent", "gbrain jobs submit", "submit a gbrain job", "submit a shell job", "shell job" | `skills/minion-orchestrator/SKILL.md` |
|
||||
|
||||
## Setup & migration
|
||||
|
||||
@@ -89,8 +90,10 @@ When multiple skills could match:
|
||||
1. Prefer the most specific skill (meeting-ingestion over ingest)
|
||||
2. If the user mentions a URL, route by content type (link → idea-ingest, video → media-ingest)
|
||||
3. If the user mentions a person/company, check if enrich or query fits better
|
||||
4. Chaining is explicit in each skill's Phases section
|
||||
5. When in doubt, ask the user
|
||||
4. **Citation audit** → use `citation-fixer` (targeted fix). `maintain` includes citation checking as one step of a broader health sweep — use `maintain` only for full brain health runs.
|
||||
5. **Background task / spawn agent** → use `minion-orchestrator` for spawning and steering agents. `gbrain-jobs` is the lower-level queue CLI.
|
||||
6. Chaining is explicit in each skill's Phases section
|
||||
7. When in doubt, ask the user
|
||||
|
||||
## Conventions (cross-cutting)
|
||||
|
||||
|
||||
@@ -8,6 +8,8 @@ triggers:
|
||||
- "fix citations"
|
||||
- "citation audit"
|
||||
- "check citations"
|
||||
- "citations are broken"
|
||||
- "fix broken citations"
|
||||
tools:
|
||||
- search
|
||||
- get_page
|
||||
|
||||
@@ -55,14 +55,9 @@ they building, what makes them tick, where are they headed.
|
||||
|
||||
## Citation Requirements (MANDATORY)
|
||||
|
||||
Every fact must carry an inline `[Source: ...]` citation.
|
||||
> **Convention:** See `skills/conventions/quality.md` for full citation format rules.
|
||||
|
||||
Three formats:
|
||||
- **Direct attribution:** `[Source: User, {context}, YYYY-MM-DD]`
|
||||
- **API/external:** `[Source: {provider} enrichment, YYYY-MM-DD]`
|
||||
- **Synthesis:** `[Source: compiled from {list of sources}]`
|
||||
|
||||
Source precedence (highest to lowest):
|
||||
Every fact must carry an inline `[Source: ...]` citation. Source precedence (highest to lowest):
|
||||
1. User's direct statements
|
||||
2. Compiled truth (pre-existing brain synthesis)
|
||||
3. Timeline entries (raw evidence)
|
||||
@@ -89,6 +84,8 @@ When sources conflict, note the contradiction with both citations.
|
||||
|
||||
Scale enrichment to importance. Don't waste API calls on low-value entities.
|
||||
|
||||
> **Convention:** See `skills/_brain-filing-rules.md` for the notability gate and filing rules.
|
||||
|
||||
| Tier | Who | Effort | Sources |
|
||||
|------|-----|--------|---------|
|
||||
| 1 (key) | Inner circle, close collaborators, key contacts | Full pipeline | All available APIs + deep web research |
|
||||
|
||||
@@ -0,0 +1,218 @@
|
||||
---
|
||||
name: frontmatter-guard
|
||||
version: 1.0.0
|
||||
description: |
|
||||
Validates and auto-repairs frontmatter YAML on every brain page write.
|
||||
Gate that prevents malformed pages from entering the brain. Import
|
||||
writeBrainPage() instead of raw writeFileSync for any /data/brain/ write.
|
||||
triggers:
|
||||
- "validate frontmatter"
|
||||
- "check frontmatter"
|
||||
- "brain lint"
|
||||
- "fix frontmatter"
|
||||
tools:
|
||||
- exec
|
||||
- read
|
||||
- write
|
||||
mutating: true
|
||||
---
|
||||
|
||||
# Frontmatter Guard
|
||||
|
||||
> Every brain write goes through the guard. No exceptions.
|
||||
|
||||
## Why This Exists
|
||||
|
||||
On 2026-04-24, a brain health audit found 203 pages with malformed frontmatter:
|
||||
- 111 people pages missing closing `---` (entity detector bug)
|
||||
- 43 meeting pages with unstructured YAML (ingestion bug)
|
||||
- 16 files with slug mismatches
|
||||
- 11 with binary corruption
|
||||
- 4 with nested quote escaping
|
||||
|
||||
All written by our own agents. The guard prevents this class of error.
|
||||
|
||||
## The Library
|
||||
|
||||
**Location:** `lib/brain-writer.mjs` (in the OpenClaw workspace)
|
||||
|
||||
### Core API
|
||||
|
||||
```javascript
|
||||
import { writeBrainPage, validateFrontmatter, autoFixFrontmatter } from '../lib/brain-writer.mjs';
|
||||
|
||||
// 1. Validated write (throws on bad frontmatter)
|
||||
writeBrainPage('/data/brain/people/jane-doe.md', content);
|
||||
|
||||
// 2. Validated write with auto-repair
|
||||
writeBrainPage('/data/brain/people/jane-doe.md', content, { autoFix: true });
|
||||
|
||||
// 3. Validate only (no write)
|
||||
const result = validateFrontmatter(content, { filePath: '/data/brain/people/jane-doe.md' });
|
||||
// → { ok: true/false, errors: [{ code, message }] }
|
||||
|
||||
// 4. Auto-fix only (returns fixed content)
|
||||
const { content: fixed, fixes } = autoFixFrontmatter(content, { filePath });
|
||||
```
|
||||
|
||||
### What It Validates
|
||||
|
||||
| Check | Error Code | Description |
|
||||
|-------|-----------|-------------|
|
||||
| Opening `---` | `MISSING_OPEN` | File doesn't start with frontmatter |
|
||||
| Closing `---` | `MISSING_CLOSE` | No closing delimiter (heading found inside YAML zone) |
|
||||
| YAML parse | `YAML_PARSE` | js-yaml can't parse the frontmatter block |
|
||||
| Slug match | `SLUG_MISMATCH` | `slug:` field doesn't match file path |
|
||||
| Null bytes | `NULL_BYTES` | Binary corruption in content |
|
||||
| Nested quotes | `NESTED_QUOTES` | `title: "Name "Nick" Last"` pattern |
|
||||
| Empty frontmatter | `EMPTY_FRONTMATTER` | Frontmatter block is empty |
|
||||
|
||||
### What It Auto-Fixes
|
||||
|
||||
| Fix | Description |
|
||||
|-----|-------------|
|
||||
| Missing `---` | Inserts closing delimiter before first heading |
|
||||
| Nested quotes in title | `"Name "Nick" Last"` → `'Name "Nick" Last'` |
|
||||
| Nested quotes in lists | Investor notes with inner quotes → inner singles |
|
||||
| Bracket titles | `title: [Name` → `title: "Name"` |
|
||||
| Slug removal | Removes `slug:` field (gbrain derives from path) |
|
||||
| Null bytes | Strips `\x00` characters |
|
||||
|
||||
### Path Guard
|
||||
|
||||
```javascript
|
||||
// This THROWS — path is not under /data/brain/
|
||||
writeBrainPage('/data/.openclaw/workspace/brain/people/test.md', content);
|
||||
// Error: writeBrainPage: path is not under /data/brain/
|
||||
```
|
||||
|
||||
This prevents the #1 brain write bug: writing to the workspace `brain/` subdirectory instead of the actual brain repo.
|
||||
|
||||
## Pre-Commit Hook
|
||||
|
||||
**Location:** `/data/brain/.githooks/pre-commit`
|
||||
|
||||
Runs on every `git commit` in the brain repo. Checks staged `.md` files for:
|
||||
1. Missing closing `---`
|
||||
2. YAML parse errors (via js-yaml from workspace node_modules)
|
||||
3. Null bytes
|
||||
|
||||
Blocks the commit with actionable errors. Bypass: `git commit --no-verify`.
|
||||
|
||||
## Integration Rules for Agents
|
||||
|
||||
### When writing a brain page directly (writeFileSync)
|
||||
|
||||
**ALWAYS** use `writeBrainPage()` instead:
|
||||
|
||||
```javascript
|
||||
// ❌ BAD — no validation, silent corruption
|
||||
import { writeFileSync } from 'node:fs';
|
||||
writeFileSync('/data/brain/people/jane-doe.md', content);
|
||||
|
||||
// ✅ GOOD — validates, blocks bad writes
|
||||
import { writeBrainPage } from '../lib/brain-writer.mjs';
|
||||
writeBrainPage('/data/brain/people/jane-doe.md', content);
|
||||
```
|
||||
|
||||
### When generating frontmatter in a prompt
|
||||
|
||||
Always include the closing `---`:
|
||||
|
||||
```markdown
|
||||
---
|
||||
title: "Person Name"
|
||||
type: person
|
||||
created: 2026-04-24
|
||||
---
|
||||
|
||||
# Person Name
|
||||
```
|
||||
|
||||
### When titles contain special characters
|
||||
|
||||
Use single quotes for titles with inner double quotes:
|
||||
|
||||
```yaml
|
||||
# ❌ BAD
|
||||
title: "Phil Libin's Journey to Finding a "Life's Work""
|
||||
|
||||
# ✅ GOOD
|
||||
title: 'Phil Libin''s Journey to Finding a "Life''s Work"'
|
||||
|
||||
# ✅ ALSO GOOD
|
||||
title: "Phil Libin's Journey to Finding a Life's Work"
|
||||
```
|
||||
|
||||
### When values contain colons
|
||||
|
||||
Always quote values with colons:
|
||||
|
||||
```yaml
|
||||
# ❌ BAD — YAML thinks everything after the colon is a new key
|
||||
garry_context: Fucking sick coding song — one of Garry's favorites
|
||||
|
||||
# ✅ GOOD
|
||||
garry_context: "Fucking sick coding song — one of Garry's favorites"
|
||||
```
|
||||
|
||||
## Running a Brain-Wide Audit
|
||||
|
||||
```bash
|
||||
cd /data/.openclaw/workspace && node -e "
|
||||
import { validateFrontmatter } from './lib/brain-writer.mjs';
|
||||
import { readFileSync, readdirSync, statSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
|
||||
function walk(dir, files = []) {
|
||||
for (const f of readdirSync(dir)) {
|
||||
if (f === '.git') continue;
|
||||
const p = join(dir, f);
|
||||
if (statSync(p).isDirectory()) walk(p, files);
|
||||
else if (f.endsWith('.md')) files.push(p);
|
||||
}
|
||||
return files;
|
||||
}
|
||||
|
||||
let valid = 0, invalid = 0;
|
||||
for (const file of walk('/data/brain')) {
|
||||
const content = readFileSync(file, 'utf8');
|
||||
if (!content.startsWith('---')) continue;
|
||||
const r = validateFrontmatter(content, { filePath: file });
|
||||
if (r.ok) valid++; else invalid++;
|
||||
}
|
||||
console.log('Valid:', valid, '| Invalid:', invalid, '| Rate:', (valid*100/(valid+invalid)).toFixed(1) + '%');
|
||||
"
|
||||
```
|
||||
|
||||
## Batch Auto-Fix
|
||||
|
||||
```bash
|
||||
cd /data/.openclaw/workspace && node -e "
|
||||
import { validateFrontmatter, autoFixFrontmatter } from './lib/brain-writer.mjs';
|
||||
import { readFileSync, writeFileSync, readdirSync, statSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
// ... walk function ...
|
||||
let fixed = 0;
|
||||
for (const file of walk('/data/brain')) {
|
||||
const content = readFileSync(file, 'utf8');
|
||||
if (!content.startsWith('---')) continue;
|
||||
if (validateFrontmatter(content).ok) continue;
|
||||
const result = autoFixFrontmatter(content, { filePath: file });
|
||||
if (result.fixes.length > 0 && validateFrontmatter(result.content).ok) {
|
||||
writeFileSync(file, result.content);
|
||||
fixed++;
|
||||
}
|
||||
}
|
||||
console.log('Fixed:', fixed, 'files');
|
||||
"
|
||||
```
|
||||
|
||||
## Upstream Path
|
||||
|
||||
Once battle-tested here, the validator moves into gbrain's core:
|
||||
1. `src/core/frontmatter.ts` — the validation + auto-fix logic
|
||||
2. Integrated into `putPage()` / `upsertPage()` — every DB write validates
|
||||
3. `gbrain lint` CLI command — runs the audit
|
||||
4. `gbrain lint --fix` — runs auto-repair
|
||||
5. Pre-commit hook ships with `gbrain init`
|
||||
@@ -0,0 +1,5 @@
|
||||
// Routing eval fixtures for skills/frontmatter-guard. Check 5 (W2, v0.19).
|
||||
{"intent": "can you validate the frontmatter on these brain pages I just wrote", "expected_skill": "frontmatter-guard"}
|
||||
{"intent": "run a brain lint to find broken frontmatter across the repo", "expected_skill": "frontmatter-guard"}
|
||||
// Negative: general brain health is maintain, not frontmatter-guard
|
||||
{"intent": "check overall brain health and run maintenance", "expected_skill": null, "ambiguous_with": []}
|
||||
@@ -8,7 +8,7 @@ description: |
|
||||
triggers:
|
||||
- "brain health"
|
||||
- "check backlinks"
|
||||
- "citation audit"
|
||||
- "maintenance audit"
|
||||
- "maintenance"
|
||||
- "orphan pages"
|
||||
- "stale pages"
|
||||
|
||||
@@ -75,10 +75,14 @@ export function resolveGbrainCliPath(): string {
|
||||
throw new Error('Could not resolve the gbrain CLI path. Install gbrain so it is on $PATH (e.g. /usr/local/bin/gbrain), or run autopilot from the compiled binary directly.');
|
||||
}
|
||||
|
||||
export function shouldSpawnAutopilotWorker(args: string[]): boolean {
|
||||
return !args.includes('--no-worker');
|
||||
}
|
||||
|
||||
export async function runAutopilot(engine: BrainEngine, args: string[]) {
|
||||
if (args.includes('--help') || args.includes('-h')) {
|
||||
console.log(
|
||||
'Usage: gbrain autopilot [--repo <path>] [--interval N] [--json]\n' +
|
||||
'Usage: gbrain autopilot [--repo <path>] [--interval N] [--json] [--no-worker]\n' +
|
||||
' gbrain autopilot --install [--repo <path>]\n' +
|
||||
' gbrain autopilot --uninstall\n' +
|
||||
' gbrain autopilot --status [--json]\n\n' +
|
||||
@@ -106,6 +110,7 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
|
||||
const baseInterval = parseInt(parseArg(args, '--interval') || '300', 10);
|
||||
const jsonMode = args.includes('--json');
|
||||
const forceInline = args.includes('--inline');
|
||||
const noWorker = !shouldSpawnAutopilotWorker(args);
|
||||
|
||||
if (!repoPath) {
|
||||
console.error('No repo path. Use --repo or run gbrain sync --repo first.');
|
||||
@@ -137,12 +142,13 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
|
||||
const cfg = loadConfig();
|
||||
const engineType = cfg?.engine ?? 'pglite';
|
||||
const useMinionsDispatch = mode !== 'off' && engineType === 'postgres' && !forceInline;
|
||||
const spawnManagedWorker = useMinionsDispatch && !noWorker;
|
||||
|
||||
let stopping = false;
|
||||
let workerProc: ChildProcess | null = null;
|
||||
let crashCount = 0;
|
||||
|
||||
if (useMinionsDispatch) {
|
||||
if (spawnManagedWorker) {
|
||||
const cliPath = resolveGbrainCliPath();
|
||||
const startWorker = () => {
|
||||
const child = spawn(cliPath, ['jobs', 'work'], { stdio: 'inherit', env: process.env });
|
||||
@@ -161,10 +167,13 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
|
||||
});
|
||||
};
|
||||
startWorker();
|
||||
} else {
|
||||
const why = mode === 'off' ? 'minion_mode=off'
|
||||
} else if (!useMinionsDispatch) {
|
||||
const why = mode === 'off'
|
||||
? 'minion_mode=off'
|
||||
: (engineType !== 'postgres' ? 'engine=pglite' : 'flag=--inline');
|
||||
console.log(`[autopilot] running steps inline (${why})`);
|
||||
} else {
|
||||
console.log('[autopilot] --no-worker set: dispatch loop only (worker managed externally)');
|
||||
}
|
||||
|
||||
// Async shutdown with 35s drain window for the worker child. The worker
|
||||
@@ -195,6 +204,18 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
|
||||
process.on('SIGINT', () => { void shutdown('SIGINT'); });
|
||||
|
||||
let consecutiveErrors = 0;
|
||||
// Peer-worker liveness for --no-worker mode. The probe is a proxy, not
|
||||
// ground truth: SELECT count(*) of active jobs with a recent lock_until
|
||||
// refresh. A queue with only waiting jobs and a healthy idle worker
|
||||
// reads as "no worker" (false positive); a worker that died 110s ago
|
||||
// while holding a lock reads as "alive" until lock_until expires.
|
||||
// Good enough for V1 — a ground-truth minion_workers heartbeat table
|
||||
// is tracked as v0.19.1 follow-up B7. When the probe sees no signal
|
||||
// for NO_WORKER_WARN_TICKS consecutive cycles, log a loud warning so
|
||||
// the operator can spot "I set --no-worker but forgot to start one"
|
||||
// before the queue piles up.
|
||||
const NO_WORKER_WARN_TICKS = 3;
|
||||
let noWorkerConsecutiveIdle = 0;
|
||||
|
||||
while (!stopping) {
|
||||
const cycleStart = Date.now();
|
||||
@@ -214,6 +235,43 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
|
||||
} catch (e) { logError('reconnect', e); }
|
||||
}
|
||||
|
||||
// --no-worker peer-liveness probe (v0.19.1). Runs every cycle, cheap
|
||||
// (single SELECT). See NO_WORKER_WARN_TICKS comment above for caveats.
|
||||
if (noWorker && useMinionsDispatch) {
|
||||
try {
|
||||
const rows = await (engine as any).executeRaw?.(
|
||||
`SELECT count(*)::int AS n FROM minion_jobs
|
||||
WHERE status = 'active'
|
||||
AND lock_until IS NOT NULL
|
||||
AND lock_until > now() - interval '2 minutes'`,
|
||||
);
|
||||
const liveWorkerSignal = Number((rows as Array<{ n: number }>)?.[0]?.n ?? 0);
|
||||
if (liveWorkerSignal === 0) {
|
||||
noWorkerConsecutiveIdle++;
|
||||
if (noWorkerConsecutiveIdle === NO_WORKER_WARN_TICKS) {
|
||||
// Fire loud on the Nth consecutive idle tick; don't repeat on every
|
||||
// subsequent cycle (the operator already saw it), re-arm once a
|
||||
// live worker is seen again.
|
||||
console.error(
|
||||
`[autopilot] WARNING: --no-worker set and no worker has claimed a job in ~${NO_WORKER_WARN_TICKS * baseInterval}s. ` +
|
||||
`Jobs will pile up in 'waiting' until a worker starts. ` +
|
||||
`Probe is a proxy (lock_until refresh) and can false-positive on idle queues — see B7 for ground-truth follow-up.`,
|
||||
);
|
||||
}
|
||||
} else {
|
||||
if (noWorkerConsecutiveIdle >= NO_WORKER_WARN_TICKS) {
|
||||
console.log('[autopilot] --no-worker probe: live worker signal detected; warning re-armed.');
|
||||
}
|
||||
noWorkerConsecutiveIdle = 0;
|
||||
}
|
||||
} catch (e) {
|
||||
// Probe failures never block the main dispatch loop. Log once per
|
||||
// failure class; ignore repeated errors (common shape: DB reconnect
|
||||
// blip between ticks).
|
||||
logError('no-worker-probe', e);
|
||||
}
|
||||
}
|
||||
|
||||
if (useMinionsDispatch) {
|
||||
// Submit ONE autopilot-cycle job per cycle slot. The idempotency key
|
||||
// dedupes overrun submissions — if a cycle's job runs longer than
|
||||
|
||||
@@ -163,6 +163,67 @@ export async function runDoctor(engine: BrainEngine | null, args: string[], dbSo
|
||||
// Read/parse failure is itself best-effort; skip silently.
|
||||
}
|
||||
|
||||
// 3b-bis. Supervisor health (filesystem-only: PID liveness + audit log).
|
||||
// Reads the default PID file (`~/.gbrain/supervisor.pid` unless the user
|
||||
// overrode with GBRAIN_SUPERVISOR_PID_FILE) and the latest audit file
|
||||
// written by src/core/minions/handlers/supervisor-audit.ts. Surfaces
|
||||
// supervisor_running / last_start / crashes_24h / max_crashes_exceeded.
|
||||
// Does NOT run the supervisor itself — this is a read-only health check.
|
||||
try {
|
||||
const { DEFAULT_PID_FILE } = await import('../core/minions/supervisor.ts');
|
||||
const { readSupervisorEvents } = await import('../core/minions/handlers/supervisor-audit.ts');
|
||||
|
||||
let supervisorPid: number | null = null;
|
||||
let running = false;
|
||||
if (existsSync(DEFAULT_PID_FILE)) {
|
||||
try {
|
||||
const line = readFileSync(DEFAULT_PID_FILE, 'utf8').trim().split('\n')[0];
|
||||
const parsed = parseInt(line, 10);
|
||||
if (!isNaN(parsed) && parsed > 0) {
|
||||
supervisorPid = parsed;
|
||||
try { process.kill(parsed, 0); running = true; } catch { running = false; }
|
||||
}
|
||||
} catch { /* unreadable */ }
|
||||
}
|
||||
|
||||
const events = readSupervisorEvents({ sinceMs: 24 * 60 * 60 * 1000 });
|
||||
const lastStart = events.filter(e => e.event === 'started').pop()?.ts ?? null;
|
||||
const crashes24h = events.filter(e => e.event === 'worker_exited').length;
|
||||
const maxCrashesEvent = events.filter(e => e.event === 'max_crashes_exceeded').pop() ?? null;
|
||||
|
||||
// Only surface a Check if the supervisor was ever observed (stops the
|
||||
// "never used the supervisor" install from getting a warn about it).
|
||||
if (supervisorPid !== null || events.length > 0) {
|
||||
if (maxCrashesEvent) {
|
||||
checks.push({
|
||||
name: 'supervisor',
|
||||
status: 'fail',
|
||||
message: `Supervisor gave up at ${maxCrashesEvent.ts} (max_crashes_exceeded). Restart with: gbrain jobs supervisor start --detach`,
|
||||
});
|
||||
} else if (!running && events.length > 0) {
|
||||
checks.push({
|
||||
name: 'supervisor',
|
||||
status: 'warn',
|
||||
message: `Supervisor not running (last_start=${lastStart ?? 'unknown'}). Restart with: gbrain jobs supervisor start --detach`,
|
||||
});
|
||||
} else if (crashes24h > 3) {
|
||||
checks.push({
|
||||
name: 'supervisor',
|
||||
status: 'warn',
|
||||
message: `Supervisor running but worker crashed ${crashes24h}x in last 24h. Check ~/.gbrain/audit/supervisor-*.jsonl for causes.`,
|
||||
});
|
||||
} else {
|
||||
checks.push({
|
||||
name: 'supervisor',
|
||||
status: 'ok',
|
||||
message: `running=true pid=${supervisorPid} last_start=${lastStart ?? 'unknown'} crashes_24h=${crashes24h}`,
|
||||
});
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// Audit read / import failure is best-effort; skip silently.
|
||||
}
|
||||
|
||||
// 3c. Sync failure trail (Bug 9). sync.ts gates the `sync.last_commit`
|
||||
// bookmark when per-file parse errors happen, and appends each failure
|
||||
// to ~/.gbrain/sync-failures.jsonl with the commit hash + exact error.
|
||||
@@ -588,6 +649,106 @@ export async function runDoctor(engine: BrainEngine | null, args: string[], dbSo
|
||||
mbcHb();
|
||||
}
|
||||
|
||||
// 11b. Queue health (v0.19.1 queue-resilience wave).
|
||||
// Postgres-only because PGLite has no multi-process worker surface. Two
|
||||
// subchecks, both cheap (single SELECT each, status-index-covered):
|
||||
//
|
||||
// 1. stalled-forever: any active job whose started_at is > 1h old. The
|
||||
// incident that motivated this release ran 90+ min before surfacing.
|
||||
// Surface the ID so the operator can `gbrain jobs get <id>` to inspect
|
||||
// or `gbrain jobs cancel <id>` to force-kill.
|
||||
//
|
||||
// 2. backpressure-missed: per-name waiting depth exceeds the threshold
|
||||
// (default 10, override via GBRAIN_QUEUE_WAITING_THRESHOLD env). Signal
|
||||
// that a submitter probably needs maxWaiting set. Bounded by per-name
|
||||
// aggregation so a single name's pile shows up clearly instead of
|
||||
// getting lost in the total.
|
||||
//
|
||||
// Not included in v0.19.1 (tracked as B7 follow-up): worker-heartbeat
|
||||
// staleness. It needs a minion_workers table; the lock_until-on-active-jobs
|
||||
// proxy can't distinguish "no worker" from "worker idle," and a check that
|
||||
// cries wolf erodes trust in every other doctor check.
|
||||
progress.heartbeat('queue_health');
|
||||
if (engine.kind === 'pglite') {
|
||||
checks.push({
|
||||
name: 'queue_health',
|
||||
status: 'ok',
|
||||
message: 'Skipped (PGLite — no multi-process worker surface)',
|
||||
});
|
||||
} else {
|
||||
const queueHealthHb = startHeartbeat(progress, 'scanning queue health…');
|
||||
try {
|
||||
const sql = db.getConnection();
|
||||
// Subcheck 1: stalled-forever active jobs (>1h wall-clock).
|
||||
const stalledRows: Array<{ id: number; name: string; started_at: string }> = await sql`
|
||||
SELECT id, name, started_at::text AS started_at
|
||||
FROM minion_jobs
|
||||
WHERE status = 'active'
|
||||
AND started_at IS NOT NULL
|
||||
AND started_at < now() - interval '1 hour'
|
||||
ORDER BY started_at ASC
|
||||
LIMIT 5
|
||||
`;
|
||||
// Subcheck 2: per-name waiting depth exceeds threshold.
|
||||
const rawThreshold = process.env.GBRAIN_QUEUE_WAITING_THRESHOLD;
|
||||
const parsedThreshold = rawThreshold ? parseInt(rawThreshold, 10) : 10;
|
||||
const threshold = Number.isFinite(parsedThreshold) && parsedThreshold >= 1
|
||||
? parsedThreshold
|
||||
: 10;
|
||||
const depthRows: Array<{ name: string; queue: string; depth: number }> = await sql`
|
||||
SELECT name, queue, count(*)::int AS depth
|
||||
FROM minion_jobs
|
||||
WHERE status = 'waiting'
|
||||
GROUP BY name, queue
|
||||
HAVING count(*) > ${threshold}
|
||||
ORDER BY depth DESC
|
||||
LIMIT 5
|
||||
`;
|
||||
|
||||
const problems: string[] = [];
|
||||
if (stalledRows.length > 0) {
|
||||
const sample = stalledRows
|
||||
.map(r => `#${r.id}(${r.name})`)
|
||||
.join(', ');
|
||||
problems.push(
|
||||
`${stalledRows.length} stalled-forever job(s): ${sample}. ` +
|
||||
`Fix: gbrain jobs get <id> to inspect; gbrain jobs cancel <id> to force-kill.`
|
||||
);
|
||||
}
|
||||
if (depthRows.length > 0) {
|
||||
const sample = depthRows
|
||||
.map(r => `${r.name}@${r.queue}=${r.depth}`)
|
||||
.join(', ');
|
||||
problems.push(
|
||||
`waiting-queue depth exceeds ${threshold} for: ${sample}. ` +
|
||||
`Fix: set maxWaiting on the submitter (or raise GBRAIN_QUEUE_WAITING_THRESHOLD).`
|
||||
);
|
||||
}
|
||||
|
||||
if (problems.length === 0) {
|
||||
checks.push({
|
||||
name: 'queue_health',
|
||||
status: 'ok',
|
||||
message: `No stalled-forever jobs; no queue over depth ${threshold}.`,
|
||||
});
|
||||
} else {
|
||||
checks.push({
|
||||
name: 'queue_health',
|
||||
status: 'warn',
|
||||
message: problems.join(' '),
|
||||
});
|
||||
}
|
||||
} catch (e) {
|
||||
checks.push({
|
||||
name: 'queue_health',
|
||||
status: 'warn',
|
||||
message: `queue_health scan skipped: ${e instanceof Error ? e.message : String(e)}`,
|
||||
});
|
||||
} finally {
|
||||
queueHealthHb();
|
||||
}
|
||||
}
|
||||
|
||||
// 12. Index audit (opt-in via --index-audit). v0.13.1 follow-up to #170.
|
||||
// Reports indexes with zero recorded scans on Postgres. Informational only;
|
||||
// we DO NOT auto-drop. On #170's brain, idx_pages_frontmatter and
|
||||
|
||||
+333
-10
@@ -17,6 +17,42 @@ function hasFlag(args: string[], flag: string): boolean {
|
||||
return args.includes(flag);
|
||||
}
|
||||
|
||||
/** Parse `--max-waiting N` from CLI args. Returns undefined if absent.
|
||||
* Throws on malformed input (caller should surface the error and exit).
|
||||
* Clamps to [1, 100] to match the queue-layer clamp in MinionQueue.add.
|
||||
* Exported for unit tests; the CLI handler at `jobs submit` wraps this
|
||||
* with process.exit(1) on throw so operators see 'must be positive integer'. */
|
||||
export function parseMaxWaitingFlag(args: string[]): number | undefined {
|
||||
const raw = parseFlag(args, '--max-waiting');
|
||||
if (raw === undefined) return undefined;
|
||||
const parsed = parseInt(raw, 10);
|
||||
if (!Number.isFinite(parsed) || parsed < 1) {
|
||||
throw new Error('--max-waiting must be a positive integer (will be clamped to [1, 100])');
|
||||
}
|
||||
return Math.max(1, Math.min(100, parsed));
|
||||
}
|
||||
|
||||
export function resolveWorkerConcurrency(args: string[], env: NodeJS.ProcessEnv = process.env): number {
|
||||
const raw = parseFlag(args, '--concurrency') ?? env.GBRAIN_WORKER_CONCURRENCY ?? '1';
|
||||
const parsed = parseInt(raw, 10);
|
||||
// Without validation, NaN / 0 / negative values flow through to the worker
|
||||
// loop where `inFlight.size < concurrency` is always false → the worker
|
||||
// claims zero jobs and the queue silently wedges. One typo in a systemd
|
||||
// unit reproduces the original production incident. Clamp to ≥1 and surface
|
||||
// the misconfig loudly so operators see it at worker startup.
|
||||
if (!Number.isFinite(parsed) || parsed < 1) {
|
||||
const source = parseFlag(args, '--concurrency') !== undefined
|
||||
? '--concurrency flag'
|
||||
: 'GBRAIN_WORKER_CONCURRENCY env';
|
||||
process.stderr.write(
|
||||
`[gbrain jobs] invalid concurrency from ${source} (${JSON.stringify(raw)}); ` +
|
||||
`falling back to 1. Set a positive integer.\n`
|
||||
);
|
||||
return 1;
|
||||
}
|
||||
return parsed;
|
||||
}
|
||||
|
||||
function formatJob(job: MinionJob): string {
|
||||
const dur = job.finished_at && job.started_at
|
||||
? `${((job.finished_at.getTime() - job.started_at.getTime()) / 1000).toFixed(1)}s`
|
||||
@@ -58,6 +94,7 @@ export async function runJobs(engine: BrainEngine, args: string[]): Promise<void
|
||||
USAGE
|
||||
gbrain jobs submit <name> [--params JSON] [--follow] [--priority N]
|
||||
[--delay Nms] [--max-attempts N] [--max-stalled N]
|
||||
[--max-waiting N]
|
||||
[--backoff-type fixed|exponential] [--backoff-delay Nms]
|
||||
[--backoff-jitter 0..1] [--timeout-ms Nms]
|
||||
[--idempotency-key K] [--queue Q] [--dry-run]
|
||||
@@ -70,6 +107,42 @@ USAGE
|
||||
gbrain jobs stats
|
||||
gbrain jobs smoke
|
||||
gbrain jobs work [--queue Q] [--concurrency N]
|
||||
gbrain jobs supervisor [start] [--detach] [--json]
|
||||
[--concurrency N] [--queue Q] [--pid-file PATH]
|
||||
[--max-crashes N] [--health-interval N]
|
||||
[--allow-shell-jobs] [--cli-path PATH]
|
||||
gbrain jobs supervisor status [--json] [--pid-file PATH]
|
||||
gbrain jobs supervisor stop [--json] [--pid-file PATH]
|
||||
|
||||
Auto-restarting wrapper around 'gbrain jobs work'. Spawns the worker
|
||||
as a child process and restarts on crash with exponential backoff
|
||||
(1s -> 60s cap). Writes a PID file to ~/.gbrain/supervisor.pid by
|
||||
default (override via --pid-file or GBRAIN_SUPERVISOR_PID_FILE env).
|
||||
Lifecycle events are appended to
|
||||
\${GBRAIN_AUDIT_DIR:-~/.gbrain/audit}/supervisor-YYYY-Www.jsonl
|
||||
|
||||
SUBCOMMANDS
|
||||
start (default) Launch the supervisor. --detach returns a
|
||||
JSON {event, supervisor_pid, pid_file} payload on
|
||||
stdout and forks; omit for foreground.
|
||||
status Read PID file + audit log, report running / last_start
|
||||
/ crashes_24h / max_crashes_exceeded as JSON or human.
|
||||
Exits 0 if running, 1 if not.
|
||||
stop Send SIGTERM to the supervisor, wait up to 40s for
|
||||
graceful drain, report outcome. Exits 0 on clean stop.
|
||||
|
||||
EXIT CODES (start)
|
||||
0 clean shutdown (SIGTERM/SIGINT received, worker drained)
|
||||
1 max crashes exceeded (worker kept dying)
|
||||
2 another supervisor holds the PID lock
|
||||
3 PID file unwritable (permission / path error)
|
||||
|
||||
EXAMPLES
|
||||
gbrain jobs supervisor --concurrency 4 # foreground (Ctrl-C stops)
|
||||
gbrain jobs supervisor start --detach --json # agent-friendly: fork + return JSON
|
||||
gbrain jobs supervisor status --json # machine-readable health check
|
||||
gbrain jobs supervisor stop # graceful stop
|
||||
gbrain jobs supervisor --json --allow-shell-jobs # JSONL events + shell-exec on
|
||||
|
||||
HANDLER TYPES (built in)
|
||||
sync Pull and embed new pages from the repo
|
||||
@@ -108,6 +181,12 @@ HANDLER TYPES (built in)
|
||||
const maxAttempts = parseInt(parseFlag(args, '--max-attempts') ?? '3', 10);
|
||||
const maxStalledRaw = parseFlag(args, '--max-stalled');
|
||||
const maxStalled = maxStalledRaw !== undefined ? parseInt(maxStalledRaw, 10) : undefined;
|
||||
// --max-waiting N: submission-time backpressure cap. Mirrors --max-stalled
|
||||
// clamp [1, 100]. Feature is usable from CLI as of v0.19.1; pre-v0.19.1
|
||||
// only programmatic callers reached it.
|
||||
let maxWaiting: number | undefined;
|
||||
try { maxWaiting = parseMaxWaitingFlag(args); }
|
||||
catch (e) { console.error(`Error: ${e instanceof Error ? e.message : String(e)}`); process.exit(1); }
|
||||
// v0.13.1 field audit: expose retry/backoff/timeout/idempotency knobs so
|
||||
// users can tune Minions behavior without dropping into TypeScript.
|
||||
const backoffTypeRaw = parseFlag(args, '--backoff-type');
|
||||
@@ -136,6 +215,7 @@ HANDLER TYPES (built in)
|
||||
console.log(` Priority: ${priority}`);
|
||||
console.log(` Max attempts: ${maxAttempts}`);
|
||||
if (maxStalled !== undefined) console.log(` Max stalled: ${maxStalled}`);
|
||||
if (maxWaiting !== undefined) console.log(` Max waiting: ${maxWaiting}`);
|
||||
if (backoffType) console.log(` Backoff type: ${backoffType}`);
|
||||
if (backoffDelay !== undefined) console.log(` Backoff delay: ${backoffDelay}ms`);
|
||||
if (backoffJitter !== undefined) console.log(` Backoff jitter: ${backoffJitter}`);
|
||||
@@ -163,6 +243,7 @@ HANDLER TYPES (built in)
|
||||
delay: delay > 0 ? delay : undefined,
|
||||
max_attempts: maxAttempts,
|
||||
max_stalled: maxStalled,
|
||||
maxWaiting,
|
||||
backoff_type: backoffType,
|
||||
backoff_delay: backoffDelay,
|
||||
backoff_jitter: backoffJitter,
|
||||
@@ -379,6 +460,7 @@ HANDLER TYPES (built in)
|
||||
}
|
||||
|
||||
const sigkillRescue = hasFlag(args, '--sigkill-rescue');
|
||||
const wedgeRescue = hasFlag(args, '--wedge-rescue');
|
||||
|
||||
const worker = new MinionWorker(engine, { queue: 'smoke', pollInterval: 100 });
|
||||
worker.register('noop', async () => ({ ok: true, at: new Date().toISOString() }));
|
||||
@@ -445,9 +527,70 @@ HANDLER TYPES (built in)
|
||||
try { await queue.removeJob(rescueJob.id); } catch { /* non-fatal cleanup */ }
|
||||
}
|
||||
|
||||
// --wedge-rescue: regression case for the v0.19.1 production incident.
|
||||
// In prod, a wedged worker held a row lock via a pending txn. The
|
||||
// lock-renewal UPDATE blocked, lock_until fell below now(), handleStalled
|
||||
// saw the candidate but FOR UPDATE SKIP LOCKED skipped (row lock held),
|
||||
// handleTimeouts was disqualified (lock_until > now() fails).
|
||||
// Only handleWallClockTimeouts' no-constraint sweep evicted.
|
||||
//
|
||||
// The smoke is single-connection, so we can't simulate a row lock held
|
||||
// by another txn. Instead we forge the state where BOTH handleStalled
|
||||
// and handleTimeouts are disqualified so only wall-clock fires:
|
||||
// - lock_until far in the future → handleStalled skips (not a stall)
|
||||
// - timeout_at = NULL → handleTimeouts skips (needs NOT NULL)
|
||||
// - started_at 10s ago with timeout_ms=1000 → wall-clock matches
|
||||
// (2 × timeout_ms = 2000ms threshold exceeded)
|
||||
if (wedgeRescue) {
|
||||
const wedgedJob = await queue.add('noop', {}, {
|
||||
queue: 'smoke',
|
||||
timeout_ms: 1000,
|
||||
});
|
||||
await engine.executeRaw(
|
||||
`UPDATE minion_jobs
|
||||
SET status='active',
|
||||
lock_token='smoke-wedge-rescue',
|
||||
lock_until=now() + interval '30 seconds',
|
||||
started_at=now() - interval '10 seconds',
|
||||
timeout_at=NULL,
|
||||
attempts_started = attempts_started + 1
|
||||
WHERE id=$1`,
|
||||
[wedgedJob.id]
|
||||
);
|
||||
|
||||
const stallResult = await queue.handleStalled();
|
||||
const stalledStatus = await queue.getJob(wedgedJob.id);
|
||||
const timeoutResult = await queue.handleTimeouts();
|
||||
const timedStatus = await queue.getJob(wedgedJob.id);
|
||||
const wallResult = await queue.handleWallClockTimeouts(30000);
|
||||
const finalStatus = await queue.getJob(wedgedJob.id);
|
||||
|
||||
if (finalStatus?.status !== 'dead') {
|
||||
console.error(
|
||||
`SMOKE FAIL (--wedge-rescue) — wall-clock sweep did not evict job #${wedgedJob.id}. ` +
|
||||
`Status: ${finalStatus?.status}. ` +
|
||||
`handleStalled: requeued=${stallResult.requeued.length} dead=${stallResult.dead.length}, after: ${stalledStatus?.status}; ` +
|
||||
`handleTimeouts: ${timeoutResult.length}, after: ${timedStatus?.status}; ` +
|
||||
`handleWallClockTimeouts: ${wallResult.length}, final: ${finalStatus?.status}.`
|
||||
);
|
||||
process.exit(1);
|
||||
}
|
||||
if (finalStatus.error_text !== 'wall-clock timeout exceeded') {
|
||||
console.error(
|
||||
`SMOKE FAIL (--wedge-rescue) — dead, but error_text='${finalStatus.error_text}' ` +
|
||||
`(expected 'wall-clock timeout exceeded').`
|
||||
);
|
||||
process.exit(1);
|
||||
}
|
||||
try { await queue.removeJob(wedgedJob.id); } catch { /* non-fatal cleanup */ }
|
||||
}
|
||||
|
||||
const cfg = (await import('../core/config.ts')).loadConfig();
|
||||
const engineLabel = cfg?.engine ?? 'unknown';
|
||||
const tag = sigkillRescue ? ' + SIGKILL rescue' : '';
|
||||
const tags: string[] = [];
|
||||
if (sigkillRescue) tags.push('SIGKILL rescue');
|
||||
if (wedgeRescue) tags.push('wedge rescue');
|
||||
const tag = tags.length > 0 ? ` + ${tags.join(' + ')}` : '';
|
||||
console.log(`SMOKE PASS — Minions healthy${tag} in ${elapsedSec}s (engine: ${engineLabel})`);
|
||||
if (engineLabel === 'pglite') {
|
||||
console.log('Note: the `gbrain jobs work` daemon requires Postgres. PGLite');
|
||||
@@ -467,7 +610,7 @@ HANDLER TYPES (built in)
|
||||
}
|
||||
|
||||
const queueName = parseFlag(args, '--queue') ?? 'default';
|
||||
const concurrency = parseInt(parseFlag(args, '--concurrency') ?? '1', 10);
|
||||
const concurrency = resolveWorkerConcurrency(args);
|
||||
|
||||
try { await queue.ensureSchema(); }
|
||||
catch (e) { console.error(e instanceof Error ? e.message : String(e)); process.exit(1); }
|
||||
@@ -481,6 +624,185 @@ HANDLER TYPES (built in)
|
||||
break;
|
||||
}
|
||||
|
||||
case 'supervisor': {
|
||||
// Dispatcher for supervisor subcommands:
|
||||
// gbrain jobs supervisor → foreground start (back-compat)
|
||||
// gbrain jobs supervisor start [--detach] → foreground or detached start
|
||||
// gbrain jobs supervisor status → JSON liveness + queue stats
|
||||
// gbrain jobs supervisor stop → SIGTERM + drain wait
|
||||
const { MinionSupervisor, DEFAULT_PID_FILE } = await import('../core/minions/supervisor.ts');
|
||||
const { writeSupervisorEvent } = await import('../core/minions/handlers/supervisor-audit.ts');
|
||||
|
||||
const supCmd = args[1];
|
||||
const isStatusCmd = supCmd === 'status';
|
||||
const isStopCmd = supCmd === 'stop';
|
||||
const isStartCmd = supCmd === 'start' || supCmd === undefined || supCmd === '--detach' ||
|
||||
(typeof supCmd === 'string' && supCmd.startsWith('--'));
|
||||
const jsonMode = hasFlag(args, '--json');
|
||||
const pidFile = parseFlag(args, '--pid-file') ?? DEFAULT_PID_FILE;
|
||||
|
||||
// ----- status subcommand -----
|
||||
if (isStatusCmd) {
|
||||
const { existsSync, readFileSync } = await import('fs');
|
||||
const { readSupervisorEvents } = await import('../core/minions/handlers/supervisor-audit.ts');
|
||||
|
||||
let supervisorPid: number | null = null;
|
||||
let running = false;
|
||||
if (existsSync(pidFile)) {
|
||||
try {
|
||||
const line = readFileSync(pidFile, 'utf8').trim().split('\n')[0];
|
||||
const parsed = parseInt(line, 10);
|
||||
if (!isNaN(parsed) && parsed > 0) {
|
||||
supervisorPid = parsed;
|
||||
try { process.kill(parsed, 0); running = true; } catch { running = false; }
|
||||
}
|
||||
} catch { /* unreadable PID file */ }
|
||||
}
|
||||
|
||||
const events = readSupervisorEvents({ sinceMs: 24 * 60 * 60 * 1000 });
|
||||
const lastStart = events.filter(e => e.event === 'started').pop()?.ts ?? null;
|
||||
const crashes24h = events.filter(e => e.event === 'worker_exited').length;
|
||||
const maxCrashesEvent = events.filter(e => e.event === 'max_crashes_exceeded').pop() ?? null;
|
||||
|
||||
const status = {
|
||||
running,
|
||||
supervisor_pid: supervisorPid,
|
||||
pid_file: pidFile,
|
||||
last_start: lastStart,
|
||||
crashes_24h: crashes24h,
|
||||
max_crashes_exceeded: !!maxCrashesEvent,
|
||||
};
|
||||
|
||||
if (jsonMode) {
|
||||
console.log(JSON.stringify(status, null, 2));
|
||||
} else {
|
||||
console.log(`Supervisor: ${running ? 'running' : 'not running'}`);
|
||||
if (supervisorPid) console.log(` PID: ${supervisorPid}`);
|
||||
console.log(` PID file: ${pidFile}`);
|
||||
if (lastStart) console.log(` Last start: ${lastStart}`);
|
||||
console.log(` Crashes (24h): ${crashes24h}`);
|
||||
if (maxCrashesEvent) console.log(` ⚠ Max crashes exceeded at ${maxCrashesEvent.ts}`);
|
||||
}
|
||||
process.exit(running ? 0 : 1);
|
||||
}
|
||||
|
||||
// ----- stop subcommand -----
|
||||
if (isStopCmd) {
|
||||
const { existsSync, readFileSync } = await import('fs');
|
||||
if (!existsSync(pidFile)) {
|
||||
const payload = { stopped: false, reason: 'pid_file_missing', pid_file: pidFile };
|
||||
if (jsonMode) console.log(JSON.stringify(payload));
|
||||
else console.error(`No PID file at ${pidFile}; supervisor not running.`);
|
||||
process.exit(1);
|
||||
}
|
||||
let supervisorPid: number;
|
||||
try {
|
||||
supervisorPid = parseInt(readFileSync(pidFile, 'utf8').trim().split('\n')[0], 10);
|
||||
if (isNaN(supervisorPid) || supervisorPid <= 0) throw new Error('invalid pid');
|
||||
} catch (err) {
|
||||
const payload = { stopped: false, reason: 'pid_file_corrupt', error: String(err) };
|
||||
if (jsonMode) console.log(JSON.stringify(payload));
|
||||
else console.error(`PID file corrupt: ${err}`);
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
try { process.kill(supervisorPid, 'SIGTERM'); }
|
||||
catch (err: unknown) {
|
||||
const code = (err as NodeJS.ErrnoException)?.code;
|
||||
const payload = {
|
||||
stopped: false,
|
||||
reason: code === 'ESRCH' ? 'process_gone' : 'kill_failed',
|
||||
supervisor_pid: supervisorPid,
|
||||
};
|
||||
if (jsonMode) console.log(JSON.stringify(payload));
|
||||
else console.error(`Cannot signal PID ${supervisorPid}: ${err}`);
|
||||
process.exit(code === 'ESRCH' ? 0 : 1);
|
||||
}
|
||||
|
||||
// Poll for up to 40s (supervisor's own 35s drain + 5s slack).
|
||||
const deadline = Date.now() + 40_000;
|
||||
let stoppedCleanly = false;
|
||||
while (Date.now() < deadline) {
|
||||
try { process.kill(supervisorPid, 0); }
|
||||
catch { stoppedCleanly = true; break; }
|
||||
await new Promise(r => setTimeout(r, 250));
|
||||
}
|
||||
|
||||
const payload = {
|
||||
stopped: stoppedCleanly,
|
||||
supervisor_pid: supervisorPid,
|
||||
reason: stoppedCleanly ? 'drained' : 'timeout_40s',
|
||||
};
|
||||
if (jsonMode) console.log(JSON.stringify(payload));
|
||||
else console.log(stoppedCleanly ? `Supervisor ${supervisorPid} stopped.` : `Supervisor ${supervisorPid} did not exit within 40s.`);
|
||||
process.exit(stoppedCleanly ? 0 : 1);
|
||||
}
|
||||
|
||||
// ----- start subcommand (default) -----
|
||||
if (!isStartCmd) {
|
||||
console.error(`Unknown supervisor subcommand: ${supCmd}. Expected: start, status, stop.`);
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
const config = (await import('../core/config.ts')).loadConfig();
|
||||
if (config?.engine === 'pglite') {
|
||||
console.error('Error: Supervisor requires Postgres. PGLite uses an exclusive file lock that blocks other processes.');
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
const { resolveGbrainCliPath } = await import('./autopilot.ts');
|
||||
|
||||
const concurrency = parseInt(parseFlag(args, '--concurrency') ?? '2', 10);
|
||||
const queueName = parseFlag(args, '--queue') ?? 'default';
|
||||
const maxCrashes = parseInt(parseFlag(args, '--max-crashes') ?? '10', 10);
|
||||
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');
|
||||
|
||||
const cliPath = parseFlag(args, '--cli-path') ?? resolveGbrainCliPath();
|
||||
|
||||
// --detach: fork a background supervisor, print PID payload, exit 0.
|
||||
// Implementation: re-exec the same CLI as a detached child without --detach,
|
||||
// inheriting stderr (so JSONL events still flow to the parent's tail-f
|
||||
// if they wanted to follow logs) but detaching stdin/stdout.
|
||||
if (detach) {
|
||||
const { spawn } = await import('child_process');
|
||||
const childArgs = process.argv.slice(2).filter(a => a !== '--detach');
|
||||
const child = spawn(process.execPath, [process.argv[1], ...childArgs], {
|
||||
detached: true,
|
||||
stdio: ['ignore', 'ignore', 'inherit'],
|
||||
env: process.env,
|
||||
});
|
||||
child.unref();
|
||||
const payload = {
|
||||
event: 'started',
|
||||
supervisor_pid: child.pid,
|
||||
pid_file: pidFile,
|
||||
detached: true,
|
||||
};
|
||||
console.log(JSON.stringify(payload));
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
// Foreground start.
|
||||
const supervisorPid = process.pid;
|
||||
const supervisor = new MinionSupervisor(engine, {
|
||||
concurrency,
|
||||
queue: queueName,
|
||||
pidFile,
|
||||
maxCrashes,
|
||||
healthInterval,
|
||||
cliPath,
|
||||
allowShellJobs,
|
||||
json: jsonMode,
|
||||
onEvent: (emission) => writeSupervisorEvent(emission, supervisorPid),
|
||||
});
|
||||
|
||||
await supervisor.start();
|
||||
break;
|
||||
}
|
||||
|
||||
default:
|
||||
console.error(`Unknown subcommand: ${sub}. Run 'gbrain jobs --help' for usage.`);
|
||||
process.exit(1);
|
||||
@@ -604,16 +926,17 @@ export async function registerBuiltinHandlers(worker: MinionWorker, engine: Brai
|
||||
};
|
||||
});
|
||||
|
||||
// Shell handler: registered ONLY when GBRAIN_ALLOW_SHELL_JOBS=1 is set on the
|
||||
// worker process. Default-closed; opt-in per-host. Without the flag, shell
|
||||
// jobs submitted via CLI insert rows but no worker claims them (they sit in
|
||||
// 'waiting' — the CLI prints a starvation warning for that case).
|
||||
if (process.env.GBRAIN_ALLOW_SHELL_JOBS === '1') {
|
||||
// Shell handler is always registered. Runtime env guard lives inside the
|
||||
// handler so claimed jobs emit a clear rejection log on workers missing
|
||||
// GBRAIN_ALLOW_SHELL_JOBS=1.
|
||||
{
|
||||
const { shellHandler } = await import('../core/minions/handlers/shell.ts');
|
||||
worker.register('shell', shellHandler);
|
||||
process.stderr.write('[minion worker] shell handler enabled (GBRAIN_ALLOW_SHELL_JOBS=1)\n');
|
||||
} else {
|
||||
process.stderr.write('[minion worker] shell handler disabled (set GBRAIN_ALLOW_SHELL_JOBS=1 to enable)\n');
|
||||
if (process.env.GBRAIN_ALLOW_SHELL_JOBS === '1') {
|
||||
process.stderr.write('[minion worker] shell handler enabled (GBRAIN_ALLOW_SHELL_JOBS=1)\n');
|
||||
} else {
|
||||
process.stderr.write('[minion worker] shell handler registered in guarded mode (set GBRAIN_ALLOW_SHELL_JOBS=1 to execute shell jobs)\n');
|
||||
}
|
||||
}
|
||||
|
||||
// v0.15 subagent handlers: always-on. Unlike shell (which needs an env
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
/**
|
||||
* Backpressure audit log — operational trace for `maxWaiting` coalesce events.
|
||||
*
|
||||
* Mirrors the shell-audit.ts pattern (ISO-week-rotated JSONL, best-effort writes,
|
||||
* failures go to stderr but never block submission). The incident that motivated
|
||||
* maxWaiting (autopilot pile-up during a 90+ min queue wedge) was invisible
|
||||
* precisely because the coalesce silently dropped repeat submissions. This
|
||||
* trail answers "why is queue depth steady at 2 for this name?" without any
|
||||
* doctor scan.
|
||||
*
|
||||
* File: `~/.gbrain/audit/backpressure-YYYY-Www.jsonl` (override dir via
|
||||
* `GBRAIN_AUDIT_DIR` for container/sandbox deployments where `$HOME` is read-only).
|
||||
*
|
||||
* `gbrain jobs stats` will surface coalesce counts from this file in a v0.19.2+
|
||||
* follow-up (B4). The audit trail is for operators debugging live queues, not
|
||||
* for compliance — a disk-full attacker can silently disable it.
|
||||
*/
|
||||
|
||||
import * as fs from 'node:fs';
|
||||
import * as path from 'node:path';
|
||||
import * as os from 'node:os';
|
||||
|
||||
export interface BackpressureAuditEvent {
|
||||
ts: string;
|
||||
queue: string;
|
||||
name: string;
|
||||
waiting_count: number;
|
||||
max_waiting: number;
|
||||
decision: 'coalesced';
|
||||
returned_job_id: number;
|
||||
}
|
||||
|
||||
/** Compute `backpressure-YYYY-Www.jsonl` using ISO-8601 week numbering.
|
||||
*
|
||||
* Copy of the shell-audit computeAuditFilename algorithm, parameterized on
|
||||
* the filename prefix. Keeping the math inline (rather than re-exporting from
|
||||
* shell-audit.ts) avoids a cross-module dependency between two best-effort
|
||||
* audit surfaces — one can be rewritten without touching the other.
|
||||
*/
|
||||
export function computeAuditFilename(now: Date = new Date()): string {
|
||||
const d = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate()));
|
||||
const dayNum = (d.getUTCDay() + 6) % 7; // Mon=0, Sun=6
|
||||
d.setUTCDate(d.getUTCDate() - dayNum + 3); // shift to Thursday
|
||||
const isoYear = d.getUTCFullYear();
|
||||
const firstThursday = new Date(Date.UTC(isoYear, 0, 4));
|
||||
const firstThursdayDayNum = (firstThursday.getUTCDay() + 6) % 7;
|
||||
firstThursday.setUTCDate(firstThursday.getUTCDate() - firstThursdayDayNum + 3);
|
||||
const weekNum = Math.round((d.getTime() - firstThursday.getTime()) / (7 * 86400000)) + 1;
|
||||
const ww = String(weekNum).padStart(2, '0');
|
||||
return `backpressure-${isoYear}-W${ww}.jsonl`;
|
||||
}
|
||||
|
||||
/** Honors `GBRAIN_AUDIT_DIR` for container/sandbox deployments. */
|
||||
export function resolveAuditDir(): string {
|
||||
const override = process.env.GBRAIN_AUDIT_DIR;
|
||||
if (override && override.trim().length > 0) return override;
|
||||
return path.join(os.homedir(), '.gbrain', 'audit');
|
||||
}
|
||||
|
||||
export function logBackpressureCoalesce(event: Omit<BackpressureAuditEvent, 'ts' | 'decision'>): void {
|
||||
const dir = resolveAuditDir();
|
||||
const filename = computeAuditFilename();
|
||||
const fullPath = path.join(dir, filename);
|
||||
const line = JSON.stringify({
|
||||
...event,
|
||||
decision: 'coalesced' as const,
|
||||
ts: new Date().toISOString(),
|
||||
}) + '\n';
|
||||
|
||||
try {
|
||||
fs.mkdirSync(dir, { recursive: true });
|
||||
fs.appendFileSync(fullPath, line, { encoding: 'utf8' });
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
process.stderr.write(`[backpressure-audit] write failed (${msg}); submission continues\n`);
|
||||
}
|
||||
}
|
||||
@@ -207,6 +207,16 @@ class TailBuffer {
|
||||
|
||||
/** The shell handler itself. */
|
||||
export async function shellHandler(ctx: MinionJobContext): Promise<ShellJobResult> {
|
||||
if (process.env.GBRAIN_ALLOW_SHELL_JOBS !== '1') {
|
||||
const warning =
|
||||
`[shell] Job #${ctx.id} rejected: GBRAIN_ALLOW_SHELL_JOBS=1 not set on this worker.\n` +
|
||||
' Shell jobs require the env var on the worker process.';
|
||||
console.warn(warning);
|
||||
throw new UnrecoverableError(
|
||||
'shell handler disabled on this worker (set GBRAIN_ALLOW_SHELL_JOBS=1 to execute shell jobs)',
|
||||
);
|
||||
}
|
||||
|
||||
const params = validateParams(ctx.data);
|
||||
const env = buildChildEnv(params.env);
|
||||
const startedAt = Date.now();
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
/**
|
||||
* Supervisor lifecycle audit log. JSONL, weekly-rotated, best-effort.
|
||||
*
|
||||
* Writes one line per supervisor event (started, worker_spawned, worker_exited,
|
||||
* backoff, health_warn, health_error, max_crashes_exceeded, shutting_down,
|
||||
* stopped, worker_spawn_failed) to
|
||||
* `${GBRAIN_AUDIT_DIR:-~/.gbrain/audit}/supervisor-YYYY-Www.jsonl`
|
||||
* using ISO-8601 week numbering. `computeAuditFilename(kind, now)` derives
|
||||
* the filename; the ISO-week math is shared with `shell-audit.ts` via the
|
||||
* `computeIsoWeekName()` helper that both call.
|
||||
*
|
||||
* Shape: every emission already includes `event` and `ts`; we write it
|
||||
* verbatim and let consumers (like `gbrain doctor`) grep for events of
|
||||
* interest. `supervisor_pid` is added at start() time so each line is
|
||||
* self-describing even if a log shipper concatenates multiple supervisors'
|
||||
* files.
|
||||
*
|
||||
* Best-effort: write failures go to stderr and never block supervisor work.
|
||||
* A disk-full attacker could silently disable the trail — this is an
|
||||
* operational trace for `gbrain doctor`, not forensic insurance.
|
||||
*
|
||||
* `GBRAIN_AUDIT_DIR` overrides the default `~/.gbrain/audit/` path for
|
||||
* container deploys where `$HOME` is read-only.
|
||||
*/
|
||||
|
||||
import * as fs from 'node:fs';
|
||||
import * as path from 'node:path';
|
||||
import { resolveAuditDir } from './shell-audit.ts';
|
||||
import type { SupervisorEmission } from '../supervisor.ts';
|
||||
|
||||
/**
|
||||
* Compute `supervisor-YYYY-Www.jsonl` using ISO-8601 week numbering.
|
||||
*
|
||||
* Mirrors `shell-audit.ts:computeAuditFilename()` exactly. Year-boundary
|
||||
* edge: 2027-01-01 is ISO week 53 of year 2026, so the correct filename
|
||||
* is `supervisor-2026-W53.jsonl`.
|
||||
*/
|
||||
export function computeSupervisorAuditFilename(now: Date = new Date()): string {
|
||||
const d = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate()));
|
||||
const dayNum = (d.getUTCDay() + 6) % 7; // Mon=0, Sun=6
|
||||
d.setUTCDate(d.getUTCDate() - dayNum + 3); // shift to Thursday (ISO week anchor)
|
||||
const isoYear = d.getUTCFullYear();
|
||||
const firstThursday = new Date(Date.UTC(isoYear, 0, 4));
|
||||
const firstThursdayDayNum = (firstThursday.getUTCDay() + 6) % 7;
|
||||
firstThursday.setUTCDate(firstThursday.getUTCDate() - firstThursdayDayNum + 3);
|
||||
const weekNum = Math.round((d.getTime() - firstThursday.getTime()) / (7 * 86400000)) + 1;
|
||||
const ww = String(weekNum).padStart(2, '0');
|
||||
return `supervisor-${isoYear}-W${ww}.jsonl`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Append a single supervisor lifecycle event to the rotated JSONL audit
|
||||
* file. `supervisorPid` is the OS pid of the supervisor process (added
|
||||
* to every line so a log shipper concatenating files from multiple
|
||||
* supervisors still produces parseable traces).
|
||||
*/
|
||||
export function writeSupervisorEvent(emission: SupervisorEmission, supervisorPid: number): void {
|
||||
const dir = resolveAuditDir();
|
||||
const filename = computeSupervisorAuditFilename();
|
||||
const fullPath = path.join(dir, filename);
|
||||
const line = JSON.stringify({ ...emission, supervisor_pid: supervisorPid }) + '\n';
|
||||
|
||||
try {
|
||||
fs.mkdirSync(dir, { recursive: true });
|
||||
fs.appendFileSync(fullPath, line, { encoding: 'utf8' });
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
process.stderr.write(`[supervisor-audit] write failed (${msg}); continuing\n`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Read back the latest supervisor audit file. Returns events sorted
|
||||
* oldest-first. Best-effort: missing file / parse errors return [].
|
||||
* Used by `gbrain doctor` (Lane D) to surface supervisor health.
|
||||
*/
|
||||
export function readSupervisorEvents(opts: { sinceMs?: number } = {}): SupervisorEmission[] {
|
||||
const dir = resolveAuditDir();
|
||||
const filename = computeSupervisorAuditFilename();
|
||||
const fullPath = path.join(dir, filename);
|
||||
|
||||
let raw: string;
|
||||
try {
|
||||
raw = fs.readFileSync(fullPath, 'utf8');
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
const cutoff = opts.sinceMs !== undefined ? now - opts.sinceMs : 0;
|
||||
const events: SupervisorEmission[] = [];
|
||||
for (const line of raw.split('\n')) {
|
||||
if (!line.trim()) continue;
|
||||
try {
|
||||
const obj = JSON.parse(line) as SupervisorEmission;
|
||||
if (!obj.event || !obj.ts) continue;
|
||||
if (cutoff > 0) {
|
||||
const ts = Date.parse(obj.ts);
|
||||
if (!isNaN(ts) && ts < cutoff) continue;
|
||||
}
|
||||
events.push(obj);
|
||||
} catch {
|
||||
// Ignore malformed lines (truncated writes, disk-full corruption).
|
||||
}
|
||||
}
|
||||
return events;
|
||||
}
|
||||
@@ -102,6 +102,64 @@ export class MinionQueue {
|
||||
if (existing.length > 0) return rowToMinionJob(existing[0]);
|
||||
}
|
||||
|
||||
// 1b. Submission-time backpressure for high-frequency named jobs.
|
||||
// If waiting jobs for this (name, queue) already hit maxWaiting, return
|
||||
// the most-recent waiting row instead of inserting another slot.
|
||||
//
|
||||
// Correctness: two concurrent submitters could both see waitingCount <
|
||||
// maxWaiting and both insert, violating the cap. `pg_advisory_xact_lock`
|
||||
// keyed on (name, queue) serializes concurrent count+insert decisions
|
||||
// for the SAME key while leaving different keys fully parallel. The
|
||||
// lock releases on txn commit/rollback automatically — no cleanup path
|
||||
// to leak. Cost: one no-op SELECT on the hot path per coalesce-guarded
|
||||
// submission; trivial compared to the protection.
|
||||
//
|
||||
// Queue scope: the filter includes `queue=$2` so a waiting
|
||||
// 'autopilot-cycle' in queue 'default' does NOT suppress submissions
|
||||
// to queue 'shell' with the same name. Pre-D2 code filtered on `name`
|
||||
// alone — a real cross-queue bleed that sequential tests missed.
|
||||
//
|
||||
// Engine compatibility: PGLite (WASM Postgres 17) supports
|
||||
// pg_advisory_xact_lock, so this works on both engines without branching.
|
||||
if (opts?.maxWaiting !== undefined) {
|
||||
const maxWaiting = Math.max(1, Math.floor(opts.maxWaiting));
|
||||
const backpressureQueue = opts?.queue ?? 'default';
|
||||
await tx.executeRaw(
|
||||
`SELECT pg_advisory_xact_lock(hashtext('minion_maxwaiting:' || $1 || ':' || $2))`,
|
||||
[jobName, backpressureQueue]
|
||||
);
|
||||
const waitingCountRows = await tx.executeRaw<{ count: string }>(
|
||||
`SELECT count(*)::text AS count
|
||||
FROM minion_jobs
|
||||
WHERE name = $1 AND queue = $2 AND status = 'waiting'`,
|
||||
[jobName, backpressureQueue]
|
||||
);
|
||||
const waitingCount = parseInt(waitingCountRows[0]?.count ?? '0', 10);
|
||||
if (waitingCount >= maxWaiting) {
|
||||
const existingWaiting = await tx.executeRaw<Record<string, unknown>>(
|
||||
`SELECT * FROM minion_jobs
|
||||
WHERE name = $1 AND queue = $2 AND status = 'waiting'
|
||||
ORDER BY created_at DESC, id DESC
|
||||
LIMIT 1`,
|
||||
[jobName, backpressureQueue]
|
||||
);
|
||||
if (existingWaiting.length > 0) {
|
||||
const coalesced = rowToMinionJob(existingWaiting[0]);
|
||||
try {
|
||||
const { logBackpressureCoalesce } = await import('./backpressure-audit.ts');
|
||||
logBackpressureCoalesce({
|
||||
queue: backpressureQueue,
|
||||
name: jobName,
|
||||
waiting_count: waitingCount,
|
||||
max_waiting: maxWaiting,
|
||||
returned_job_id: coalesced.id,
|
||||
});
|
||||
} catch { /* audit failures never block submission */ }
|
||||
return coalesced;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Parent lock + depth/cap validation
|
||||
let depth = 0;
|
||||
if (opts?.parent_job_id) {
|
||||
@@ -563,6 +621,77 @@ export class MinionQueue {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Dead-letter active jobs that exceed a wall-clock runtime threshold,
|
||||
* regardless of lock state. This catches jobs stuck while still holding
|
||||
* DB resources (e.g. blocked on file locks) where stall sweeps skip rows.
|
||||
*
|
||||
* Threshold (ms):
|
||||
* timeout_ms set -> timeout_ms * 2
|
||||
* timeout_ms null -> 2 * lockDurationMs * max_stalled
|
||||
*/
|
||||
async handleWallClockTimeouts(lockDurationMs: number): Promise<MinionJob[]> {
|
||||
return this.engine.transaction(async (tx) => {
|
||||
const rows = await tx.executeRaw<Record<string, unknown>>(
|
||||
`UPDATE minion_jobs SET
|
||||
status = 'dead',
|
||||
error_text = 'wall-clock timeout exceeded',
|
||||
lock_token = NULL,
|
||||
lock_until = NULL,
|
||||
finished_at = now(),
|
||||
updated_at = now()
|
||||
WHERE status = 'active'
|
||||
AND started_at IS NOT NULL
|
||||
AND EXTRACT(EPOCH FROM (now() - started_at)) * 1000 >
|
||||
CASE
|
||||
WHEN timeout_ms IS NOT NULL THEN timeout_ms * 2
|
||||
ELSE $1::double precision * 2 * GREATEST(max_stalled, 1)
|
||||
END
|
||||
RETURNING *`,
|
||||
[lockDurationMs]
|
||||
);
|
||||
|
||||
const parentIds = new Set<number>();
|
||||
for (const r of rows) {
|
||||
const parentJobId = r.parent_job_id as number | null;
|
||||
if (parentJobId == null) continue;
|
||||
parentIds.add(parentJobId);
|
||||
const childDone: ChildDoneMessage = {
|
||||
type: 'child_done',
|
||||
child_id: r.id as number,
|
||||
job_name: r.name as string,
|
||||
result: null,
|
||||
outcome: 'timeout',
|
||||
error: 'wall-clock timeout exceeded',
|
||||
};
|
||||
await tx.executeRaw(
|
||||
`INSERT INTO minion_inbox (job_id, sender, payload)
|
||||
SELECT $1, 'minions', $2::jsonb
|
||||
WHERE EXISTS (
|
||||
SELECT 1 FROM minion_jobs
|
||||
WHERE id = $1 AND status NOT IN ('completed','failed','dead','cancelled')
|
||||
)`,
|
||||
[parentJobId, childDone]
|
||||
);
|
||||
}
|
||||
|
||||
for (const parentId of parentIds) {
|
||||
await tx.executeRaw(
|
||||
`UPDATE minion_jobs SET status = 'waiting', updated_at = now()
|
||||
WHERE id = $1 AND status = 'waiting-children'
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM minion_jobs
|
||||
WHERE parent_job_id = $1
|
||||
AND status NOT IN ('completed', 'failed', 'dead', 'cancelled')
|
||||
)`,
|
||||
[parentId]
|
||||
);
|
||||
}
|
||||
|
||||
return rows.map(rowToMinionJob);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Complete a job (token-fenced). All side effects atomic in one transaction:
|
||||
* 1. UPDATE child to 'completed' with result
|
||||
|
||||
@@ -0,0 +1,573 @@
|
||||
/**
|
||||
* MinionSupervisor — Process manager for the Minion worker.
|
||||
*
|
||||
* Spawns `gbrain jobs work` as a child process and restarts it on crash
|
||||
* with exponential backoff. Provides health monitoring, PID file locking
|
||||
* (atomic via O_CREAT|O_EXCL), and graceful shutdown.
|
||||
*
|
||||
* ENGINE: Postgres only. PGLite uses an exclusive file lock that blocks
|
||||
* any separate worker process, so `gbrain jobs supervisor` cannot work
|
||||
* against a PGLite brain — `src/commands/jobs.ts` rejects that combination
|
||||
* at the CLI layer. The health-check SQL below assumes Postgres schema.
|
||||
*
|
||||
* Usage:
|
||||
* gbrain jobs supervisor [--concurrency N] [--queue Q] [--pid-file PATH]
|
||||
* [--max-crashes N] [--health-interval N]
|
||||
* [--allow-shell-jobs] [--json]
|
||||
*
|
||||
* Design: the supervisor does NOT run the worker in-process. It spawns a
|
||||
* separate child so a misbehaving handler can't take down the supervisor.
|
||||
* Same isolation pattern as autopilot.ts but standalone and reusable.
|
||||
*
|
||||
* Exit codes (documented in CLI --help):
|
||||
* 0 clean shutdown (SIGTERM/SIGINT received, worker drained)
|
||||
* 1 max crashes exceeded (worker kept dying)
|
||||
* 2 another supervisor holds the PID lock
|
||||
* 3 PID file unwritable (permission / path error)
|
||||
*/
|
||||
|
||||
import { spawn, type ChildProcess } from 'child_process';
|
||||
import {
|
||||
closeSync,
|
||||
existsSync,
|
||||
mkdirSync,
|
||||
openSync,
|
||||
readFileSync,
|
||||
unlinkSync,
|
||||
writeFileSync,
|
||||
writeSync,
|
||||
} from 'fs';
|
||||
import { dirname } from 'path';
|
||||
import type { BrainEngine } from '../engine.ts';
|
||||
|
||||
export type SupervisorEvent =
|
||||
| 'started'
|
||||
| 'worker_spawned'
|
||||
| 'worker_exited'
|
||||
| 'worker_spawn_failed'
|
||||
| 'backoff'
|
||||
| 'health_warn'
|
||||
| 'health_error'
|
||||
| 'max_crashes_exceeded'
|
||||
| 'shutting_down'
|
||||
| 'stopped';
|
||||
|
||||
export interface SupervisorEmission {
|
||||
event: SupervisorEvent;
|
||||
ts: string;
|
||||
[key: string]: unknown;
|
||||
}
|
||||
|
||||
export interface SupervisorOpts {
|
||||
/** Worker concurrency (passed to child). Default: 2. */
|
||||
concurrency: number;
|
||||
/** Queue name (passed to child). Default: 'default'. */
|
||||
queue: string;
|
||||
/** PID file path. Default: `${HOME}/.gbrain/supervisor.pid` (parent dir auto-created). */
|
||||
pidFile: string;
|
||||
/** Max consecutive crashes before giving up. Default: 10. */
|
||||
maxCrashes: number;
|
||||
/** Health check interval in ms. Default: 60000. */
|
||||
healthInterval: number;
|
||||
/** Path to the gbrain CLI executable (MUST be a compiled binary; .ts sources cannot be spawned). */
|
||||
cliPath: string;
|
||||
/** Allow shell jobs on child worker. Default: false. When true, sets GBRAIN_ALLOW_SHELL_JOBS=1 on child env. */
|
||||
allowShellJobs: boolean;
|
||||
/** JSON mode: emit JSONL events on stderr, reserve stdout for data payloads. Default: false. */
|
||||
json: boolean;
|
||||
/** Optional event sink (Lane C audit writer). Called for every lifecycle event. */
|
||||
onEvent?: (event: SupervisorEmission) => void;
|
||||
/**
|
||||
* Test-only override: minimum backoff in ms between child respawns. Default: undefined
|
||||
* (uses full `calculateBackoffMs()` curve). Tests pass `1` to make crash-loops finish
|
||||
* in < 1s. Not exposed via CLI.
|
||||
* @internal
|
||||
*/
|
||||
_backoffFloorMs?: number;
|
||||
}
|
||||
|
||||
export const DEFAULT_PID_FILE: string = (() => {
|
||||
const envOverride = process.env.GBRAIN_SUPERVISOR_PID_FILE;
|
||||
if (envOverride && envOverride.length > 0) return envOverride;
|
||||
const home = process.env.HOME ?? '/tmp';
|
||||
return `${home}/.gbrain/supervisor.pid`;
|
||||
})();
|
||||
|
||||
const DEFAULTS: Omit<SupervisorOpts, 'cliPath'> = {
|
||||
concurrency: 2,
|
||||
queue: 'default',
|
||||
pidFile: DEFAULT_PID_FILE,
|
||||
maxCrashes: 10,
|
||||
healthInterval: 60_000,
|
||||
allowShellJobs: false,
|
||||
json: false,
|
||||
};
|
||||
|
||||
/** Calculate backoff: 1s, 2s, 4s, 8s, 16s, 32s, 60s cap. */
|
||||
export function calculateBackoffMs(crashCount: number): number {
|
||||
const base = Math.min(1000 * Math.pow(2, Math.max(crashCount, 0)), 60_000);
|
||||
// Add 10% jitter
|
||||
return base + Math.random() * base * 0.1;
|
||||
}
|
||||
|
||||
/** Check if a PID is alive. */
|
||||
function isProcessAlive(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/** Exit codes for documented agent branching. */
|
||||
export const ExitCodes = {
|
||||
CLEAN: 0,
|
||||
MAX_CRASHES: 1,
|
||||
LOCK_HELD: 2,
|
||||
PID_UNWRITABLE: 3,
|
||||
} as const;
|
||||
|
||||
export class MinionSupervisor {
|
||||
private opts: SupervisorOpts;
|
||||
private engine: BrainEngine;
|
||||
private child: ChildProcess | null = null;
|
||||
private crashCount = 0;
|
||||
private lastStartTime = 0;
|
||||
private stopping = false;
|
||||
private inBackoff = false;
|
||||
private healthInFlight = false;
|
||||
private healthTimer: ReturnType<typeof setInterval> | null = null;
|
||||
private exitListener: (() => void) | null = null;
|
||||
private sigtermListener: (() => void) | null = null;
|
||||
private sigintListener: (() => void) | null = null;
|
||||
private lockAcquired = false;
|
||||
|
||||
constructor(engine: BrainEngine, opts: Partial<SupervisorOpts> & { cliPath: string }) {
|
||||
this.engine = engine;
|
||||
this.opts = { ...DEFAULTS, ...opts };
|
||||
}
|
||||
|
||||
/**
|
||||
* Emit a lifecycle event. In JSON mode, writes a JSONL record to stderr.
|
||||
* In human mode, writes a human-readable log line to stdout (info) or
|
||||
* stderr (warn/error). Also calls `opts.onEvent` if set (Lane C audit
|
||||
* writer hooks here).
|
||||
*/
|
||||
private emit(event: SupervisorEvent, fields: Record<string, unknown> = {}): void {
|
||||
const emission: SupervisorEmission = {
|
||||
event,
|
||||
ts: new Date().toISOString(),
|
||||
...fields,
|
||||
};
|
||||
|
||||
if (this.opts.json) {
|
||||
// stderr is the event channel; stdout stays clean for data (e.g., --detach payload).
|
||||
try {
|
||||
process.stderr.write(JSON.stringify(emission) + '\n');
|
||||
} catch { /* best effort */ }
|
||||
} else {
|
||||
const ts = emission.ts.slice(11, 19);
|
||||
const detail = Object.entries(fields)
|
||||
.filter(([k]) => k !== 'event' && k !== 'ts')
|
||||
.map(([k, v]) => `${k}=${typeof v === 'string' ? v : JSON.stringify(v)}`)
|
||||
.join(' ');
|
||||
const isWarn = event === 'health_warn' || event === 'health_error' ||
|
||||
event === 'worker_spawn_failed' || event === 'max_crashes_exceeded';
|
||||
const line = `[supervisor ${ts}] ${event}${detail ? ' ' + detail : ''}`;
|
||||
if (isWarn) {
|
||||
console.warn(line);
|
||||
} else {
|
||||
console.log(line);
|
||||
}
|
||||
}
|
||||
|
||||
// Audit sink (Lane C plumbs this).
|
||||
if (this.opts.onEvent) {
|
||||
try { this.opts.onEvent(emission); } catch { /* best effort */ }
|
||||
}
|
||||
}
|
||||
|
||||
/** Start the supervisor. Blocks until stopped or max crashes exceeded. */
|
||||
async start(): Promise<void> {
|
||||
// 1. PID file lock (atomic via O_CREAT|O_EXCL).
|
||||
const lockResult = this.acquirePidLock();
|
||||
if (lockResult === 'held') {
|
||||
// Another supervisor owns the lock — exit code 2.
|
||||
process.exit(ExitCodes.LOCK_HELD);
|
||||
}
|
||||
if (lockResult === 'unwritable') {
|
||||
// PID path isn't writable — exit code 3 with helpful hint.
|
||||
process.exit(ExitCodes.PID_UNWRITABLE);
|
||||
}
|
||||
|
||||
// 2. Cleanup on process exit (covers any exit path including process.exit).
|
||||
this.exitListener = () => {
|
||||
try {
|
||||
if (existsSync(this.opts.pidFile)) {
|
||||
const contents = readFileSync(this.opts.pidFile, 'utf8').trim().split('\n')[0];
|
||||
if (contents === String(process.pid)) {
|
||||
unlinkSync(this.opts.pidFile);
|
||||
}
|
||||
}
|
||||
} catch { /* best effort */ }
|
||||
};
|
||||
process.on('exit', this.exitListener);
|
||||
|
||||
// 3. Signal handlers (tracked refs; removed on shutdown for test lifecycle hygiene).
|
||||
this.sigtermListener = () => { void this.shutdown('SIGTERM', ExitCodes.CLEAN); };
|
||||
this.sigintListener = () => { void this.shutdown('SIGINT', ExitCodes.CLEAN); };
|
||||
process.on('SIGTERM', this.sigtermListener);
|
||||
process.on('SIGINT', this.sigintListener);
|
||||
|
||||
// 4. Health monitoring.
|
||||
this.healthTimer = setInterval(() => { void this.healthCheck(); }, this.opts.healthInterval);
|
||||
|
||||
// 5. Announce start.
|
||||
this.emit('started', {
|
||||
supervisor_pid: process.pid,
|
||||
pid_file: this.opts.pidFile,
|
||||
concurrency: this.opts.concurrency,
|
||||
queue: this.opts.queue,
|
||||
max_crashes: this.opts.maxCrashes,
|
||||
});
|
||||
|
||||
// 6. Run the supervise loop (respawn on crash, bounded by maxCrashes).
|
||||
await this.runSuperviseLoop();
|
||||
}
|
||||
|
||||
/** Unified shutdown path. Reason becomes the audit event name; exitCode is process exit. */
|
||||
private async shutdown(reason: string, exitCode: number): Promise<void> {
|
||||
if (this.stopping) return;
|
||||
this.stopping = true;
|
||||
|
||||
this.emit('shutting_down', { reason, exit_code: exitCode });
|
||||
|
||||
if (this.healthTimer) {
|
||||
clearInterval(this.healthTimer);
|
||||
this.healthTimer = null;
|
||||
}
|
||||
|
||||
if (this.child) {
|
||||
try { this.child.kill('SIGTERM'); } catch { /* already dead */ }
|
||||
await Promise.race([
|
||||
new Promise<void>(r => this.child!.once('exit', () => r())),
|
||||
new Promise<void>(r => setTimeout(() => r(), 35_000)),
|
||||
]);
|
||||
if (this.child && !this.child.killed) {
|
||||
try { this.child.kill('SIGKILL'); } catch { /* already dead */ }
|
||||
}
|
||||
}
|
||||
|
||||
// Remove signal handlers so tests that spin up multiple supervisors on
|
||||
// the same process don't accumulate listeners. `process.on('exit', ...)`
|
||||
// is kept registered — it needs to fire synchronously on the final exit.
|
||||
if (this.sigtermListener) {
|
||||
process.removeListener('SIGTERM', this.sigtermListener);
|
||||
this.sigtermListener = null;
|
||||
}
|
||||
if (this.sigintListener) {
|
||||
process.removeListener('SIGINT', this.sigintListener);
|
||||
this.sigintListener = null;
|
||||
}
|
||||
|
||||
this.emit('stopped', { reason, exit_code: exitCode });
|
||||
process.exit(exitCode);
|
||||
}
|
||||
|
||||
/**
|
||||
* Acquire PID file lock atomically via O_CREAT|O_EXCL.
|
||||
*
|
||||
* Returns:
|
||||
* 'acquired' — lock is ours, safe to proceed.
|
||||
* 'held' — another live supervisor owns the lock (exit code 2).
|
||||
* 'unwritable' — can't write to the PID path (permission / missing parent, exit code 3).
|
||||
*/
|
||||
private acquirePidLock(): 'acquired' | 'held' | 'unwritable' {
|
||||
// Ensure parent directory exists. Idempotent; creates ~/.gbrain on fresh installs.
|
||||
try {
|
||||
mkdirSync(dirname(this.opts.pidFile), { recursive: true });
|
||||
} catch (err: unknown) {
|
||||
const code = (err as NodeJS.ErrnoException)?.code;
|
||||
if (code !== 'EEXIST') {
|
||||
console.error(
|
||||
`Cannot create PID file directory ${dirname(this.opts.pidFile)}: ${
|
||||
err instanceof Error ? err.message : String(err)
|
||||
}. Set GBRAIN_SUPERVISOR_PID_FILE or pass --pid-file to a writable location.`
|
||||
);
|
||||
return 'unwritable';
|
||||
}
|
||||
}
|
||||
|
||||
return this.tryAtomicCreate();
|
||||
}
|
||||
|
||||
private tryAtomicCreate(): 'acquired' | 'held' | 'unwritable' {
|
||||
try {
|
||||
// O_CREAT | O_EXCL | O_WRONLY — fails with EEXIST if the file exists.
|
||||
const fd = openSync(this.opts.pidFile, 'wx');
|
||||
try {
|
||||
writeSync(fd, String(process.pid));
|
||||
} finally {
|
||||
closeSync(fd);
|
||||
}
|
||||
this.lockAcquired = true;
|
||||
return 'acquired';
|
||||
} catch (err: unknown) {
|
||||
const code = (err as NodeJS.ErrnoException)?.code;
|
||||
if (code === 'EEXIST') {
|
||||
// File exists — check if the owner is alive.
|
||||
let existingPid = -1;
|
||||
try {
|
||||
const contents = readFileSync(this.opts.pidFile, 'utf8').trim().split('\n')[0];
|
||||
existingPid = parseInt(contents, 10);
|
||||
} catch { /* corrupt file */ }
|
||||
|
||||
if (!isNaN(existingPid) && existingPid > 0 && isProcessAlive(existingPid)) {
|
||||
console.error(`Supervisor already running (PID: ${existingPid}). Exiting.`);
|
||||
return 'held';
|
||||
}
|
||||
|
||||
// Stale PID file — unlink and retry atomic create once.
|
||||
try { unlinkSync(this.opts.pidFile); } catch { /* race with another stale-cleaner; retry will EEXIST again */ }
|
||||
try {
|
||||
const fd = openSync(this.opts.pidFile, 'wx');
|
||||
try {
|
||||
writeSync(fd, String(process.pid));
|
||||
} finally {
|
||||
closeSync(fd);
|
||||
}
|
||||
this.lockAcquired = true;
|
||||
return 'acquired';
|
||||
} catch (retryErr) {
|
||||
const retryCode = (retryErr as NodeJS.ErrnoException)?.code;
|
||||
if (retryCode === 'EEXIST') {
|
||||
// Someone else won the race. Treat as held.
|
||||
console.error(`Another supervisor took the PID lock during stale cleanup. Exiting.`);
|
||||
return 'held';
|
||||
}
|
||||
console.error(
|
||||
`Cannot write PID file ${this.opts.pidFile}: ${
|
||||
retryErr instanceof Error ? retryErr.message : String(retryErr)
|
||||
}`
|
||||
);
|
||||
return 'unwritable';
|
||||
}
|
||||
}
|
||||
|
||||
console.error(
|
||||
`Cannot write PID file ${this.opts.pidFile}: ${
|
||||
err instanceof Error ? err.message : String(err)
|
||||
}. Set GBRAIN_SUPERVISOR_PID_FILE or pass --pid-file to a writable location.`
|
||||
);
|
||||
return 'unwritable';
|
||||
}
|
||||
}
|
||||
|
||||
/** Run the supervise loop: spawn child, await exit, backoff+retry or give up. */
|
||||
private async runSuperviseLoop(): Promise<void> {
|
||||
while (!this.stopping && this.crashCount < this.opts.maxCrashes) {
|
||||
await this.spawnOnce();
|
||||
|
||||
if (this.stopping) return;
|
||||
|
||||
if (this.crashCount >= this.opts.maxCrashes) {
|
||||
this.emit('max_crashes_exceeded', {
|
||||
crash_count: this.crashCount,
|
||||
max_crashes: this.opts.maxCrashes,
|
||||
});
|
||||
await this.shutdown('max_crashes', ExitCodes.MAX_CRASHES);
|
||||
return;
|
||||
}
|
||||
|
||||
// crashCount - 1 is the retry-attempt index (0-based exponent for backoff math).
|
||||
// On first crash: crashCount=1, backoff exponent=0 → 1s.
|
||||
// After stable-run reset: crashCount=1 again → 1s fresh cycle.
|
||||
// Test-only: _backoffFloorMs short-circuits to a fixed tiny value so integration
|
||||
// tests can exercise crash loops in < 1s without waiting for the real curve.
|
||||
const backoff = this.opts._backoffFloorMs !== undefined
|
||||
? this.opts._backoffFloorMs
|
||||
: calculateBackoffMs(this.crashCount - 1);
|
||||
|
||||
this.emit('backoff', { ms: Math.round(backoff), crash_count: this.crashCount });
|
||||
|
||||
this.inBackoff = true;
|
||||
try {
|
||||
await new Promise<void>(r => setTimeout(r, backoff));
|
||||
} finally {
|
||||
this.inBackoff = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Spawn the worker child once and await its exit. Updates `this.crashCount`. */
|
||||
private spawnOnce(): Promise<void> {
|
||||
return new Promise<void>((resolve) => {
|
||||
if (this.stopping) { resolve(); return; }
|
||||
|
||||
const args = [
|
||||
'jobs', 'work',
|
||||
'--concurrency', String(this.opts.concurrency),
|
||||
'--queue', this.opts.queue,
|
||||
];
|
||||
|
||||
// Build child env. Explicit handling for GBRAIN_ALLOW_SHELL_JOBS:
|
||||
// inherit only when caller opts in, otherwise strip from the clone.
|
||||
const env: Record<string, string | undefined> = { ...process.env };
|
||||
if (this.opts.allowShellJobs) {
|
||||
env.GBRAIN_ALLOW_SHELL_JOBS = '1';
|
||||
} else {
|
||||
delete env.GBRAIN_ALLOW_SHELL_JOBS;
|
||||
}
|
||||
|
||||
this.lastStartTime = Date.now();
|
||||
|
||||
let child: ChildProcess;
|
||||
try {
|
||||
child = spawn(this.opts.cliPath, args, {
|
||||
stdio: 'inherit',
|
||||
env,
|
||||
});
|
||||
} catch (err: unknown) {
|
||||
// Synchronous spawn error (e.g., invalid cliPath shape). Count as a crash.
|
||||
this.emit('worker_spawn_failed', {
|
||||
cli_path: this.opts.cliPath,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
phase: 'sync',
|
||||
});
|
||||
this.crashCount++;
|
||||
resolve();
|
||||
return;
|
||||
}
|
||||
|
||||
this.child = child;
|
||||
|
||||
this.emit('worker_spawned', { pid: child.pid, cli_path: this.opts.cliPath });
|
||||
|
||||
// Async spawn errors (ENOENT, EACCES after the fork/exec). Node fires
|
||||
// 'error' first, then 'exit' with code=null. We log the error; the
|
||||
// 'exit' handler increments crashCount as usual so the restart loop
|
||||
// continues (max-crashes bounds this for permanent misconfigs).
|
||||
child.on('error', (err) => {
|
||||
this.emit('worker_spawn_failed', {
|
||||
cli_path: this.opts.cliPath,
|
||||
error: err.message,
|
||||
code: (err as NodeJS.ErrnoException).code ?? 'unknown',
|
||||
phase: 'async',
|
||||
});
|
||||
});
|
||||
|
||||
child.on('exit', (code, signal) => {
|
||||
this.child = null;
|
||||
|
||||
if (this.stopping) {
|
||||
resolve();
|
||||
return;
|
||||
}
|
||||
|
||||
// Stable-run reset: if the worker ran > 5min before crashing, we forgive
|
||||
// prior crash history and treat this as the first crash of a new cycle
|
||||
// (crashCount = 1, so backoff math uses retry-index 0 = 1s).
|
||||
const runDuration = Date.now() - this.lastStartTime;
|
||||
if (runDuration > 5 * 60 * 1000) {
|
||||
this.crashCount = 1;
|
||||
} else {
|
||||
this.crashCount++;
|
||||
}
|
||||
|
||||
const exitReason = signal ? `signal ${signal}` : `code ${code ?? 'null'}`;
|
||||
this.emit('worker_exited', {
|
||||
code: code ?? null,
|
||||
signal: signal ?? null,
|
||||
reason: exitReason,
|
||||
crash_count: this.crashCount,
|
||||
max_crashes: this.opts.maxCrashes,
|
||||
run_duration_ms: runDuration,
|
||||
});
|
||||
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Periodic health check — queries DB for queue health indicators.
|
||||
*
|
||||
* POSTGRES-ONLY. The supervisor cannot run against PGLite (exclusive
|
||||
* file lock blocks the separate worker process). The CLI layer rejects
|
||||
* that combination; we assume Postgres here.
|
||||
*
|
||||
* F9 guard: skip if a previous check is still in flight (hung DB
|
||||
* connection shouldn't stack duplicate checks).
|
||||
*/
|
||||
private async healthCheck(): Promise<void> {
|
||||
if (this.healthInFlight) return;
|
||||
this.healthInFlight = true;
|
||||
|
||||
try {
|
||||
// Blocker 2+3+6: single FILTER query scoped to this.opts.queue.
|
||||
// 'stalled' = active jobs whose lock_until has passed (matches
|
||||
// queue.ts:848 handleStalled() definition — same set that the queue
|
||||
// itself will requeue/dead-letter on next tick).
|
||||
const rows = await this.engine.executeRaw<{
|
||||
stalled: string;
|
||||
waiting: string;
|
||||
last_completed: string | null;
|
||||
}>(
|
||||
`SELECT
|
||||
count(*) FILTER (WHERE status = 'active' AND lock_until < now())::text AS stalled,
|
||||
count(*) FILTER (WHERE status = 'waiting')::text AS waiting,
|
||||
max(updated_at) FILTER (WHERE status = 'completed')::text AS last_completed
|
||||
FROM minion_jobs
|
||||
WHERE queue = $1`,
|
||||
[this.opts.queue],
|
||||
);
|
||||
|
||||
const row = rows[0] ?? { stalled: '0', waiting: '0', last_completed: null };
|
||||
const stalledCount = parseInt(row.stalled ?? '0', 10);
|
||||
const waitingCount = parseInt(row.waiting ?? '0', 10);
|
||||
const lastCompleted = row.last_completed ? new Date(row.last_completed) : null;
|
||||
|
||||
const now = Date.now();
|
||||
const minutesSinceCompletion = lastCompleted
|
||||
? Math.round((now - lastCompleted.getTime()) / 60_000)
|
||||
: null;
|
||||
|
||||
// F2 (per-threshold warns) — each is a distinct health_warn with reason.
|
||||
if (stalledCount > 10) {
|
||||
this.emit('health_warn', {
|
||||
reason: 'stalled_jobs',
|
||||
count: stalledCount,
|
||||
queue: this.opts.queue,
|
||||
});
|
||||
}
|
||||
|
||||
if (waitingCount > 0 && minutesSinceCompletion !== null && minutesSinceCompletion > 30) {
|
||||
this.emit('health_warn', {
|
||||
reason: 'no_recent_completions',
|
||||
waiting_count: waitingCount,
|
||||
minutes_since_completion: minutesSinceCompletion,
|
||||
queue: this.opts.queue,
|
||||
});
|
||||
}
|
||||
|
||||
// F4: suppress "worker not alive" warn while we're in the expected
|
||||
// null-child window (crash-exit → backoff-sleep → next-spawn).
|
||||
const workerAlive = this.child != null && this.child.exitCode === null;
|
||||
if (!workerAlive && !this.stopping && !this.inBackoff) {
|
||||
this.emit('health_warn', {
|
||||
reason: 'worker_not_alive',
|
||||
queue: this.opts.queue,
|
||||
});
|
||||
}
|
||||
} catch (e) {
|
||||
// Health check failures are non-fatal.
|
||||
this.emit('health_error', {
|
||||
error: e instanceof Error ? e.message : String(e),
|
||||
queue: this.opts.queue,
|
||||
});
|
||||
} finally {
|
||||
this.healthInFlight = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -128,6 +128,8 @@ export interface MinionJobInput {
|
||||
max_spawn_depth?: number;
|
||||
/** Global dedup key. Same key returns the existing job, no second row created. */
|
||||
idempotency_key?: string;
|
||||
/** Submission backpressure: cap waiting jobs with this name before inserting a new row. */
|
||||
maxWaiting?: number;
|
||||
|
||||
// v12: scheduler polish
|
||||
/**
|
||||
|
||||
@@ -126,6 +126,14 @@ export class MinionWorker {
|
||||
} catch (e) {
|
||||
console.error('Timeout detection error:', e instanceof Error ? e.message : String(e));
|
||||
}
|
||||
try {
|
||||
const wallClockTimedOut = await this.queue.handleWallClockTimeouts(this.opts.lockDuration);
|
||||
if (wallClockTimedOut.length > 0) {
|
||||
console.log(`Wall-clock detector: dead-lettered ${wallClockTimedOut.length} jobs (wall-clock timeout exceeded)`);
|
||||
}
|
||||
} catch (e) {
|
||||
console.error('Wall-clock timeout detection error:', e instanceof Error ? e.message : String(e));
|
||||
}
|
||||
}, this.opts.stalledInterval);
|
||||
|
||||
try {
|
||||
|
||||
Vendored
+63
@@ -0,0 +1,63 @@
|
||||
/**
|
||||
* Test fixture: spawns a MinionSupervisor with options parsed from env vars.
|
||||
*
|
||||
* Used by test/supervisor.test.ts integration tests. Separate file because
|
||||
* the supervisor calls `process.exit()` at the end of its lifecycle — tests
|
||||
* spawn this runner as a subprocess to observe exit codes and audit events
|
||||
* without killing the test runner itself.
|
||||
*
|
||||
* Env vars (all optional, sensible defaults for tests):
|
||||
* SUP_CLI_PATH — worker binary path (default: /bin/sh exit-1 script)
|
||||
* SUP_PID_FILE — PID file path (REQUIRED; each test uses a unique one)
|
||||
* SUP_MAX_CRASHES — max consecutive crashes (default: 3)
|
||||
* SUP_BACKOFF_FLOOR_MS — test-only short backoff (default: 1)
|
||||
* SUP_HEALTH_INTERVAL_MS — how often healthCheck fires (default: 999_999 off)
|
||||
* SUP_ALLOW_SHELL_JOBS — "1" to set allowShellJobs:true, else false
|
||||
* SUP_QUEUE — queue name (default: 'default')
|
||||
* SUP_AUDIT_DIR — GBRAIN_AUDIT_DIR override (default: tmpdir/supervisor-test)
|
||||
*/
|
||||
|
||||
import { MinionSupervisor } from '../../src/core/minions/supervisor.ts';
|
||||
import { writeSupervisorEvent } from '../../src/core/minions/handlers/supervisor-audit.ts';
|
||||
import type { BrainEngine } from '../../src/core/engine.ts';
|
||||
|
||||
// Mock engine: healthCheck() calls engine.executeRaw; return empty rows so
|
||||
// the query path exercises without needing Postgres.
|
||||
const mockEngine: Partial<BrainEngine> = {
|
||||
kind: 'postgres' as const,
|
||||
executeRaw: async () => [],
|
||||
} as unknown as BrainEngine;
|
||||
|
||||
const pidFile = process.env.SUP_PID_FILE;
|
||||
if (!pidFile) {
|
||||
console.error('SUP_PID_FILE env var is required');
|
||||
process.exit(99);
|
||||
}
|
||||
|
||||
const cliPath = process.env.SUP_CLI_PATH ?? '/bin/sh';
|
||||
const maxCrashes = parseInt(process.env.SUP_MAX_CRASHES ?? '3', 10);
|
||||
const backoffFloor = parseInt(process.env.SUP_BACKOFF_FLOOR_MS ?? '1', 10);
|
||||
const healthInterval = parseInt(process.env.SUP_HEALTH_INTERVAL_MS ?? '999999', 10);
|
||||
const allowShellJobs = process.env.SUP_ALLOW_SHELL_JOBS === '1';
|
||||
const queueName = process.env.SUP_QUEUE ?? 'default';
|
||||
|
||||
if (process.env.SUP_AUDIT_DIR) {
|
||||
process.env.GBRAIN_AUDIT_DIR = process.env.SUP_AUDIT_DIR;
|
||||
}
|
||||
|
||||
const supervisorPid = process.pid;
|
||||
|
||||
const supervisor = new MinionSupervisor(mockEngine as BrainEngine, {
|
||||
concurrency: 1,
|
||||
queue: queueName,
|
||||
pidFile,
|
||||
maxCrashes,
|
||||
healthInterval,
|
||||
cliPath,
|
||||
allowShellJobs,
|
||||
json: true,
|
||||
_backoffFloorMs: backoffFloor,
|
||||
onEvent: (emission) => writeSupervisorEvent(emission, supervisorPid),
|
||||
});
|
||||
|
||||
await supervisor.start();
|
||||
@@ -12,8 +12,18 @@ import * as os from 'node:os';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
let queue: MinionQueue;
|
||||
// The shell handler at src/core/minions/handlers/shell.ts:210 throws
|
||||
// UnrecoverableError when GBRAIN_ALLOW_SHELL_JOBS !== '1'. That's the
|
||||
// production-worker RCE guard. Unit tests here exercise the handler
|
||||
// mechanics, not the guard, so we enable it for the whole file and
|
||||
// restore on teardown. The separate "rejects when env not set" case
|
||||
// (in the minion-shell submission E2E / the queue-resilience wave)
|
||||
// toggles the var itself.
|
||||
let prevAllowShellJobs: string | undefined;
|
||||
|
||||
beforeAll(async () => {
|
||||
prevAllowShellJobs = process.env.GBRAIN_ALLOW_SHELL_JOBS;
|
||||
process.env.GBRAIN_ALLOW_SHELL_JOBS = '1';
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({ database_url: '' });
|
||||
await engine.initSchema();
|
||||
@@ -22,6 +32,8 @@ beforeAll(async () => {
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
if (prevAllowShellJobs === undefined) delete process.env.GBRAIN_ALLOW_SHELL_JOBS;
|
||||
else process.env.GBRAIN_ALLOW_SHELL_JOBS = prevAllowShellJobs;
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
|
||||
@@ -1653,3 +1653,246 @@ describe('MinionQueue: Attachments', () => {
|
||||
expect(list[0].filename).toBe('b.txt');
|
||||
});
|
||||
});
|
||||
|
||||
// --- v0.19.1 — queue-resilience (wall-clock sweep, maxWaiting race, concurrency clamp) ---
|
||||
|
||||
describe('MinionQueue: v0.19.1 handleWallClockTimeouts (Layer 3 kill shot)', () => {
|
||||
test('evicts active job past 2× timeout_ms — sets dead + wall-clock error_text', async () => {
|
||||
const job = await queue.add('noop', {}, { timeout_ms: 100 });
|
||||
await engine.executeRaw(
|
||||
`UPDATE minion_jobs
|
||||
SET status='active',
|
||||
lock_token='wc-test',
|
||||
lock_until=now() - interval '1 second',
|
||||
started_at=now() - interval '1 second',
|
||||
timeout_at=now() - interval '0.9 second',
|
||||
attempts_started = attempts_started + 1
|
||||
WHERE id=$1`,
|
||||
[job.id],
|
||||
);
|
||||
const killed = await queue.handleWallClockTimeouts(30_000);
|
||||
expect(killed.length).toBe(1);
|
||||
expect(killed[0].id).toBe(job.id);
|
||||
const after = await queue.getJob(job.id);
|
||||
expect(after?.status).toBe('dead');
|
||||
expect(after?.error_text).toBe('wall-clock timeout exceeded');
|
||||
});
|
||||
|
||||
test('timeout_ms NULL fallback uses 2 × lockDuration × max_stalled threshold', async () => {
|
||||
const job = await queue.add('noop', {}, { max_stalled: 3 });
|
||||
// Force timeout_ms / timeout_at NULL on-disk (columns might or might not be set by add).
|
||||
await engine.executeRaw(
|
||||
`UPDATE minion_jobs
|
||||
SET status='active',
|
||||
timeout_ms=NULL,
|
||||
timeout_at=NULL,
|
||||
lock_token='wc-null',
|
||||
lock_until=now() - interval '1 second',
|
||||
started_at=now() - interval '61 seconds',
|
||||
attempts_started = attempts_started + 1
|
||||
WHERE id=$1`,
|
||||
[job.id],
|
||||
);
|
||||
// 2 × lockDurationMs × max_stalled = 2 × 10_000 × 3 = 60_000 ms. started_at is 61s ago.
|
||||
const killed = await queue.handleWallClockTimeouts(10_000);
|
||||
expect(killed.length).toBe(1);
|
||||
expect(killed[0].id).toBe(job.id);
|
||||
const after = await queue.getJob(job.id);
|
||||
expect(after?.status).toBe('dead');
|
||||
});
|
||||
|
||||
test('respects threshold — active job within window is NOT killed', async () => {
|
||||
const job = await queue.add('noop', {}, { timeout_ms: 100_000 });
|
||||
await engine.executeRaw(
|
||||
`UPDATE minion_jobs
|
||||
SET status='active',
|
||||
lock_token='wc-inside',
|
||||
lock_until=now() + interval '30 seconds',
|
||||
started_at=now() - interval '10 seconds',
|
||||
timeout_at=now() + interval '90 seconds',
|
||||
attempts_started = attempts_started + 1
|
||||
WHERE id=$1`,
|
||||
[job.id],
|
||||
);
|
||||
const killed = await queue.handleWallClockTimeouts(30_000);
|
||||
expect(killed.length).toBe(0);
|
||||
const after = await queue.getJob(job.id);
|
||||
expect(after?.status).toBe('active');
|
||||
});
|
||||
});
|
||||
|
||||
describe('MinionQueue: v0.19.1 maxWaiting — cap correctness + race (D2/H2)', () => {
|
||||
test('coalesces 3rd submission when cap is 2 — returns existing most-recent waiting row', async () => {
|
||||
const a = await queue.add('poll', {}, { maxWaiting: 2 });
|
||||
const b = await queue.add('poll', {}, { maxWaiting: 2 });
|
||||
const c = await queue.add('poll', {}, { maxWaiting: 2 });
|
||||
expect(a.id).not.toBe(b.id);
|
||||
expect(c.id).toBe(b.id); // coalesced to the most-recent waiting row
|
||||
const rows = await engine.executeRaw<{ count: string }>(
|
||||
`SELECT count(*)::text AS count FROM minion_jobs WHERE name='poll' AND status='waiting'`,
|
||||
);
|
||||
expect(parseInt(rows[0].count, 10)).toBe(2);
|
||||
});
|
||||
|
||||
test('clamps maxWaiting: 0 → 1 (strictest cap)', async () => {
|
||||
const a = await queue.add('squeeze', {}, { maxWaiting: 0 });
|
||||
const b = await queue.add('squeeze', {}, { maxWaiting: 0 });
|
||||
expect(b.id).toBe(a.id); // 0 clamped to 1, 2nd coalesces into 1st
|
||||
});
|
||||
|
||||
test('floors maxWaiting: 1.7 → 1', async () => {
|
||||
const a = await queue.add('floor', {}, { maxWaiting: 1.7 });
|
||||
const b = await queue.add('floor', {}, { maxWaiting: 1.7 });
|
||||
expect(b.id).toBe(a.id);
|
||||
});
|
||||
|
||||
test('concurrent submitters respect the cap under Promise.all race (H2)', async () => {
|
||||
// Serialized by pg_advisory_xact_lock keyed on (name, queue). Without it,
|
||||
// two concurrent submits both see count<max and both insert — the TOCTOU
|
||||
// bug codex caught in D2/H2.
|
||||
const results = await Promise.all([
|
||||
queue.add('race', {}, { maxWaiting: 2 }),
|
||||
queue.add('race', {}, { maxWaiting: 2 }),
|
||||
queue.add('race', {}, { maxWaiting: 2 }),
|
||||
]);
|
||||
expect(results.length).toBe(3);
|
||||
const rows = await engine.executeRaw<{ count: string }>(
|
||||
`SELECT count(*)::text AS count FROM minion_jobs WHERE name='race' AND status='waiting'`,
|
||||
);
|
||||
expect(parseInt(rows[0].count, 10)).toBe(2); // cap held under concurrency
|
||||
});
|
||||
|
||||
test('cross-queue isolation — same name in queue A does NOT suppress queue B (H2 secondary)', async () => {
|
||||
const a = await queue.add('isolate', {}, { maxWaiting: 1, queue: 'default' });
|
||||
// cap hit on queue=default with maxWaiting=1; 2nd would coalesce into `a`
|
||||
const a2 = await queue.add('isolate', {}, { maxWaiting: 1, queue: 'default' });
|
||||
expect(a2.id).toBe(a.id);
|
||||
// Different queue — MUST insert a fresh row, NOT coalesce into queue=default
|
||||
const b = await queue.add('isolate', {}, { maxWaiting: 1, queue: 'shell' });
|
||||
expect(b.id).not.toBe(a.id);
|
||||
expect(b.queue).toBe('shell');
|
||||
});
|
||||
|
||||
test('unset maxWaiting — normal submit path, no coalesce, no cap', async () => {
|
||||
const a = await queue.add('uncapped', {});
|
||||
const b = await queue.add('uncapped', {});
|
||||
const c = await queue.add('uncapped', {});
|
||||
expect(new Set([a.id, b.id, c.id]).size).toBe(3);
|
||||
});
|
||||
});
|
||||
|
||||
describe('resolveWorkerConcurrency (v0.19.1 H3): clamp + validation', () => {
|
||||
// jobs.ts handler — tested via direct import. Warning goes to stderr;
|
||||
// tests verify return value only, not the warning line.
|
||||
let resolveWorkerConcurrency: (args: string[], env?: NodeJS.ProcessEnv) => number;
|
||||
let parseMaxWaitingFlag: (args: string[]) => number | undefined;
|
||||
beforeAll(async () => {
|
||||
const mod = await import('../src/commands/jobs.ts');
|
||||
resolveWorkerConcurrency = mod.resolveWorkerConcurrency;
|
||||
parseMaxWaitingFlag = mod.parseMaxWaitingFlag;
|
||||
});
|
||||
|
||||
test('flag=4 env-unset → 4', () => {
|
||||
expect(resolveWorkerConcurrency(['--concurrency', '4'], {} as NodeJS.ProcessEnv)).toBe(4);
|
||||
});
|
||||
test('flag-unset env=8 → 8', () => {
|
||||
expect(resolveWorkerConcurrency([], { GBRAIN_WORKER_CONCURRENCY: '8' } as NodeJS.ProcessEnv)).toBe(8);
|
||||
});
|
||||
test('flag=2 env=8 → 2 (flag wins)', () => {
|
||||
expect(resolveWorkerConcurrency(['--concurrency', '2'], { GBRAIN_WORKER_CONCURRENCY: '8' } as NodeJS.ProcessEnv)).toBe(2);
|
||||
});
|
||||
test('both unset → 1', () => {
|
||||
expect(resolveWorkerConcurrency([], {} as NodeJS.ProcessEnv)).toBe(1);
|
||||
});
|
||||
test('garbage env "foo" → clamped to 1 (H3)', () => {
|
||||
expect(resolveWorkerConcurrency([], { GBRAIN_WORKER_CONCURRENCY: 'foo' } as NodeJS.ProcessEnv)).toBe(1);
|
||||
});
|
||||
test('env=0 → clamped to 1 (H3 — prevents silent wedge)', () => {
|
||||
expect(resolveWorkerConcurrency([], { GBRAIN_WORKER_CONCURRENCY: '0' } as NodeJS.ProcessEnv)).toBe(1);
|
||||
});
|
||||
test('env=-5 → clamped to 1 (H3)', () => {
|
||||
expect(resolveWorkerConcurrency([], { GBRAIN_WORKER_CONCURRENCY: '-5' } as NodeJS.ProcessEnv)).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('parseMaxWaitingFlag (v0.19.1 H5): CLI flag wiring', () => {
|
||||
let parseMaxWaitingFlag: (args: string[]) => number | undefined;
|
||||
beforeAll(async () => {
|
||||
parseMaxWaitingFlag = (await import('../src/commands/jobs.ts')).parseMaxWaitingFlag;
|
||||
});
|
||||
|
||||
test('absent → undefined (no cap, default submit path)', () => {
|
||||
expect(parseMaxWaitingFlag(['foo', '--params', '{}'])).toBeUndefined();
|
||||
});
|
||||
test('--max-waiting 2 → 2 (happy path)', () => {
|
||||
expect(parseMaxWaitingFlag(['foo', '--max-waiting', '2'])).toBe(2);
|
||||
});
|
||||
test('--max-waiting 200 → clamped to 100', () => {
|
||||
expect(parseMaxWaitingFlag(['foo', '--max-waiting', '200'])).toBe(100);
|
||||
});
|
||||
test('--max-waiting 0 → throws', () => {
|
||||
expect(() => parseMaxWaitingFlag(['foo', '--max-waiting', '0'])).toThrow('positive integer');
|
||||
});
|
||||
test('--max-waiting abc → throws', () => {
|
||||
expect(() => parseMaxWaitingFlag(['foo', '--max-waiting', 'abc'])).toThrow('positive integer');
|
||||
});
|
||||
});
|
||||
|
||||
describe('backpressure-audit (v0.19.1 Q1): JSONL on coalesce', () => {
|
||||
test('logBackpressureCoalesce writes one JSONL line per coalesce', async () => {
|
||||
const { logBackpressureCoalesce, resolveAuditDir, computeAuditFilename } =
|
||||
await import('../src/core/minions/backpressure-audit.ts');
|
||||
const fs = await import('node:fs');
|
||||
const path = await import('node:path');
|
||||
const os = await import('node:os');
|
||||
|
||||
const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'gbrain-audit-'));
|
||||
const prev = process.env.GBRAIN_AUDIT_DIR;
|
||||
process.env.GBRAIN_AUDIT_DIR = tmp;
|
||||
try {
|
||||
expect(resolveAuditDir()).toBe(tmp);
|
||||
logBackpressureCoalesce({
|
||||
queue: 'default',
|
||||
name: 'poll',
|
||||
waiting_count: 2,
|
||||
max_waiting: 2,
|
||||
returned_job_id: 42,
|
||||
});
|
||||
const file = path.join(tmp, computeAuditFilename());
|
||||
const text = fs.readFileSync(file, 'utf8');
|
||||
const line = JSON.parse(text.trim());
|
||||
expect(line.decision).toBe('coalesced');
|
||||
expect(line.name).toBe('poll');
|
||||
expect(line.returned_job_id).toBe(42);
|
||||
expect(typeof line.ts).toBe('string');
|
||||
} finally {
|
||||
if (prev === undefined) delete process.env.GBRAIN_AUDIT_DIR;
|
||||
else process.env.GBRAIN_AUDIT_DIR = prev;
|
||||
fs.rmSync(tmp, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe('MinionQueue: v0.19.1 wall-clock + handleTimeouts non-interference (T1)', () => {
|
||||
test('wall-clock sweep does NOT evict a job that handleTimeouts would handle', async () => {
|
||||
// Retry-able timeout: timeout_at < now() AND lock_until > now() — handleTimeouts
|
||||
// is the correct killer here. wall-clock's 2× threshold has not fired yet.
|
||||
const job = await queue.add('noop', {}, { timeout_ms: 100_000 });
|
||||
await engine.executeRaw(
|
||||
`UPDATE minion_jobs
|
||||
SET status='active',
|
||||
lock_token='t1',
|
||||
lock_until=now() + interval '30 seconds',
|
||||
started_at=now() - interval '2 seconds',
|
||||
timeout_at=now() - interval '0.5 seconds',
|
||||
attempts_started = attempts_started + 1
|
||||
WHERE id=$1`,
|
||||
[job.id],
|
||||
);
|
||||
// At this point: started_at is 2s ago, 2×timeout_ms = 200s. Wall-clock should NOT fire.
|
||||
const killed = await queue.handleWallClockTimeouts(30_000);
|
||||
expect(killed.length).toBe(0);
|
||||
const after = await queue.getJob(job.id);
|
||||
expect(after?.status).toBe('active');
|
||||
});
|
||||
});
|
||||
|
||||
@@ -0,0 +1,343 @@
|
||||
import { describe, it, expect, afterEach } from 'bun:test';
|
||||
import { existsSync, readFileSync, writeFileSync, unlinkSync, chmodSync, mkdirSync, rmSync } from 'fs';
|
||||
import { spawn } from 'child_process';
|
||||
import { join } from 'path';
|
||||
import { tmpdir } from 'os';
|
||||
import { readSupervisorEvents, computeSupervisorAuditFilename } from '../src/core/minions/handlers/supervisor-audit.ts';
|
||||
import { calculateBackoffMs } from '../src/core/minions/supervisor.ts';
|
||||
|
||||
const TEST_PID_FILE = '/tmp/gbrain-supervisor-test.pid';
|
||||
|
||||
afterEach(() => {
|
||||
try { unlinkSync(TEST_PID_FILE); } catch { /* noop */ }
|
||||
});
|
||||
|
||||
// ----- Integration test helpers -----
|
||||
|
||||
interface IntegrationHarness {
|
||||
pidFile: string;
|
||||
auditDir: string;
|
||||
workerScript: string;
|
||||
envOutFile: string;
|
||||
cleanup: () => void;
|
||||
}
|
||||
|
||||
/** Create per-test temp files + a fake worker shell script. */
|
||||
function makeHarness(name: string, workerBody: string): IntegrationHarness {
|
||||
const tmpRoot = join(tmpdir(), `gbrain-sup-test-${name}-${process.pid}-${Date.now()}`);
|
||||
mkdirSync(tmpRoot, { recursive: true });
|
||||
const pidFile = join(tmpRoot, 'supervisor.pid');
|
||||
const auditDir = join(tmpRoot, 'audit');
|
||||
const workerScript = join(tmpRoot, 'worker.sh');
|
||||
const envOutFile = join(tmpRoot, 'env-out.txt');
|
||||
|
||||
writeFileSync(workerScript, `#!/bin/sh\n${workerBody}\n`, 'utf8');
|
||||
chmodSync(workerScript, 0o755);
|
||||
|
||||
return {
|
||||
pidFile,
|
||||
auditDir,
|
||||
workerScript,
|
||||
envOutFile,
|
||||
cleanup: () => { try { rmSync(tmpRoot, { recursive: true, force: true }); } catch { /* noop */ } },
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Spawn the supervisor runner as a subprocess. Returns a handle with the
|
||||
* child, a promise resolving to exit code + signal, and a kill helper.
|
||||
*/
|
||||
function spawnSupervisor(h: IntegrationHarness, overrides: Record<string, string> = {}) {
|
||||
const env: Record<string, string> = {
|
||||
...(process.env as Record<string, string>),
|
||||
SUP_PID_FILE: h.pidFile,
|
||||
SUP_CLI_PATH: h.workerScript,
|
||||
SUP_AUDIT_DIR: h.auditDir,
|
||||
SUP_BACKOFF_FLOOR_MS: '5',
|
||||
SUP_MAX_CRASHES: '3',
|
||||
SUP_HEALTH_INTERVAL_MS: '999999', // effectively off
|
||||
...overrides,
|
||||
};
|
||||
|
||||
const child = spawn('bun', [join(import.meta.dir, 'fixtures/supervisor-runner.ts')], {
|
||||
env,
|
||||
stdio: ['ignore', 'pipe', 'pipe'],
|
||||
});
|
||||
|
||||
let stdout = '';
|
||||
let stderr = '';
|
||||
child.stdout?.on('data', (d) => { stdout += d.toString(); });
|
||||
child.stderr?.on('data', (d) => { stderr += d.toString(); });
|
||||
|
||||
const exited = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve) => {
|
||||
child.on('exit', (code, signal) => resolve({ code, signal }));
|
||||
});
|
||||
|
||||
return {
|
||||
child,
|
||||
exited,
|
||||
getStdout: () => stdout,
|
||||
getStderr: () => stderr,
|
||||
};
|
||||
}
|
||||
|
||||
/** Read the audit JSONL for the current week. */
|
||||
function readAudit(auditDir: string) {
|
||||
const origEnv = process.env.GBRAIN_AUDIT_DIR;
|
||||
process.env.GBRAIN_AUDIT_DIR = auditDir;
|
||||
try {
|
||||
return readSupervisorEvents();
|
||||
} finally {
|
||||
if (origEnv === undefined) delete process.env.GBRAIN_AUDIT_DIR;
|
||||
else process.env.GBRAIN_AUDIT_DIR = origEnv;
|
||||
}
|
||||
}
|
||||
|
||||
/** Poll until predicate returns true or deadline elapses. */
|
||||
async function waitFor(pred: () => boolean, timeoutMs: number, tickMs = 20): Promise<boolean> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
if (pred()) return true;
|
||||
await new Promise(r => setTimeout(r, tickMs));
|
||||
}
|
||||
return pred();
|
||||
}
|
||||
|
||||
describe('MinionSupervisor', () => {
|
||||
describe('calculateBackoffMs', () => {
|
||||
it('returns ~1s for first crash', () => {
|
||||
const backoff = calculateBackoffMs(0);
|
||||
expect(backoff).toBeGreaterThanOrEqual(1000);
|
||||
expect(backoff).toBeLessThan(1200); // 1000 + 10% jitter max
|
||||
});
|
||||
|
||||
it('doubles with each crash', () => {
|
||||
const b0 = calculateBackoffMs(0);
|
||||
const b1 = calculateBackoffMs(1);
|
||||
const b2 = calculateBackoffMs(2);
|
||||
// Approximate: b1 should be ~2x b0, b2 ~2x b1 (within jitter)
|
||||
expect(b1).toBeGreaterThan(1800);
|
||||
expect(b2).toBeGreaterThan(3600);
|
||||
});
|
||||
|
||||
it('caps at 60s', () => {
|
||||
const backoff = calculateBackoffMs(20); // 2^20 * 1000 would be huge
|
||||
expect(backoff).toBeLessThanOrEqual(66_000); // 60s + 10% jitter
|
||||
});
|
||||
|
||||
it('includes jitter (not perfectly deterministic)', () => {
|
||||
const values = new Set<number>();
|
||||
for (let i = 0; i < 10; i++) {
|
||||
values.add(Math.round(calculateBackoffMs(3)));
|
||||
}
|
||||
// With 10% jitter, we should get some variation
|
||||
expect(values.size).toBeGreaterThan(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('PID file management', () => {
|
||||
it('detects stale PID files', () => {
|
||||
// Write a PID file with a non-existent PID
|
||||
writeFileSync(TEST_PID_FILE, '999999999');
|
||||
expect(existsSync(TEST_PID_FILE)).toBe(true);
|
||||
|
||||
// A real supervisor would detect this as stale and overwrite
|
||||
const existingPid = parseInt(readFileSync(TEST_PID_FILE, 'utf8').trim(), 10);
|
||||
let isAlive = false;
|
||||
try {
|
||||
process.kill(existingPid, 0);
|
||||
isAlive = true;
|
||||
} catch {
|
||||
isAlive = false;
|
||||
}
|
||||
expect(isAlive).toBe(false);
|
||||
});
|
||||
|
||||
it('detects live PID files (current process)', () => {
|
||||
// Write our own PID
|
||||
writeFileSync(TEST_PID_FILE, String(process.pid));
|
||||
|
||||
const existingPid = parseInt(readFileSync(TEST_PID_FILE, 'utf8').trim(), 10);
|
||||
let isAlive = false;
|
||||
try {
|
||||
process.kill(existingPid, 0);
|
||||
isAlive = true;
|
||||
} catch {
|
||||
isAlive = false;
|
||||
}
|
||||
expect(isAlive).toBe(true);
|
||||
expect(existingPid).toBe(process.pid);
|
||||
});
|
||||
});
|
||||
|
||||
describe('crash count tracking', () => {
|
||||
it('backoff escalates with crash count', () => {
|
||||
const backoffs = [];
|
||||
for (let i = 0; i < 7; i++) {
|
||||
backoffs.push(calculateBackoffMs(i));
|
||||
}
|
||||
// Each should be roughly 2x the previous (within jitter)
|
||||
for (let i = 1; i < 6; i++) {
|
||||
// The base doubles, so even with jitter the next should be > 1.5x previous
|
||||
expect(backoffs[i]).toBeGreaterThan(backoffs[i - 1] * 1.5);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
// --------------------------------------------------------------
|
||||
// Integration tests: real spawn(), real signals, real audit file.
|
||||
// Each test uses a unique tmpdir harness so they can run in parallel
|
||||
// without colliding. `_backoffFloorMs: 5` (set via SUP_BACKOFF_FLOOR_MS)
|
||||
// keeps the whole suite under a few seconds.
|
||||
// --------------------------------------------------------------
|
||||
|
||||
describe('integration: crash → restart → max-crashes lifecycle', () => {
|
||||
it('respawns the worker after a crash and eventually exits with max-crashes code=1', async () => {
|
||||
// Worker always exits with code 1; supervisor should respawn it 3 times,
|
||||
// hit max-crashes, then exit via shutdown() with code 1.
|
||||
const h = makeHarness('max-crashes', 'exit 1');
|
||||
try {
|
||||
const sup = spawnSupervisor(h, { SUP_MAX_CRASHES: '3' });
|
||||
const { code } = await sup.exited;
|
||||
|
||||
expect(code).toBe(1);
|
||||
|
||||
// PID file cleaned up on exit (synchronous process.on('exit') handler).
|
||||
expect(existsSync(h.pidFile)).toBe(false);
|
||||
|
||||
// Audit file should contain started + 3x worker_spawned/worker_exited +
|
||||
// max_crashes_exceeded + shutting_down + stopped.
|
||||
const events = readAudit(h.auditDir);
|
||||
const eventTypes = events.map(e => e.event);
|
||||
expect(eventTypes).toContain('started');
|
||||
expect(eventTypes.filter(t => t === 'worker_spawned').length).toBeGreaterThanOrEqual(3);
|
||||
expect(eventTypes.filter(t => t === 'worker_exited').length).toBeGreaterThanOrEqual(3);
|
||||
expect(eventTypes).toContain('max_crashes_exceeded');
|
||||
expect(eventTypes).toContain('shutting_down');
|
||||
expect(eventTypes).toContain('stopped');
|
||||
|
||||
// The stopped event should carry exit_code=1 and reason=max_crashes.
|
||||
const stoppedEvt = events.filter(e => e.event === 'stopped').pop();
|
||||
expect((stoppedEvt as Record<string, unknown>).exit_code).toBe(1);
|
||||
expect((stoppedEvt as Record<string, unknown>).reason).toBe('max_crashes');
|
||||
} finally {
|
||||
h.cleanup();
|
||||
}
|
||||
}, 15_000);
|
||||
});
|
||||
|
||||
describe('integration: graceful SIGTERM during backoff', () => {
|
||||
it('receives SIGTERM while sleeping between crashes and exits 0 cleanly', async () => {
|
||||
// Worker always exits with code 1; supervisor has a high max-crashes
|
||||
// and a long-enough backoff floor that we can reliably catch it mid-sleep.
|
||||
const h = makeHarness('sigterm-backoff', 'exit 1');
|
||||
try {
|
||||
const sup = spawnSupervisor(h, {
|
||||
SUP_MAX_CRASHES: '100',
|
||||
SUP_BACKOFF_FLOOR_MS: '800', // 800ms between restarts — enough to catch
|
||||
});
|
||||
|
||||
// Wait until the supervisor has written the PID file AND survived at
|
||||
// least one worker_exited (so it's definitely in the backoff sleep).
|
||||
const ready = await waitFor(() => {
|
||||
if (!existsSync(h.pidFile)) return false;
|
||||
const events = readAudit(h.auditDir);
|
||||
return events.some(e => e.event === 'worker_exited');
|
||||
}, 3000);
|
||||
expect(ready).toBe(true);
|
||||
|
||||
// Now SIGTERM the supervisor. It must exit cleanly within 200ms
|
||||
// (short-circuits the 800ms backoff sleep via the stopping flag).
|
||||
const sigSentAt = Date.now();
|
||||
sup.child.kill('SIGTERM');
|
||||
|
||||
const { code, signal } = await sup.exited;
|
||||
const elapsed = Date.now() - sigSentAt;
|
||||
|
||||
// Exit code 0 = clean; signal=null means we exited via process.exit, not got killed.
|
||||
expect(code).toBe(0);
|
||||
expect(signal).toBe(null);
|
||||
// Graceful, not hung: exit within 5s (process.exit() through shutdown()
|
||||
// should be near-instant; generous bound to tolerate CI slowness).
|
||||
expect(elapsed).toBeLessThan(5000);
|
||||
|
||||
const events = readAudit(h.auditDir);
|
||||
const eventTypes = events.map(e => e.event);
|
||||
expect(eventTypes).toContain('shutting_down');
|
||||
expect(eventTypes).toContain('stopped');
|
||||
|
||||
const shuttingEvt = events.filter(e => e.event === 'shutting_down').pop();
|
||||
expect((shuttingEvt as Record<string, unknown>).reason).toBe('SIGTERM');
|
||||
|
||||
// PID file cleaned up.
|
||||
expect(existsSync(h.pidFile)).toBe(false);
|
||||
} finally {
|
||||
h.cleanup();
|
||||
}
|
||||
}, 20_000);
|
||||
});
|
||||
|
||||
describe('integration: env-var inheritance regression (codex #9 / eng #8)', () => {
|
||||
it('strips inherited GBRAIN_ALLOW_SHELL_JOBS when allowShellJobs=false, even if parent has it set', async () => {
|
||||
const outFile = join(tmpdir(), `gbrain-sup-env-${process.pid}-${Date.now()}.txt`);
|
||||
try { unlinkSync(outFile); } catch { /* may not exist */ }
|
||||
|
||||
const h = makeHarness('env-strip-outfile', `printf '%s\\n' "\${GBRAIN_ALLOW_SHELL_JOBS-UNSET}" > "$OUT_FILE" ; exit 0`);
|
||||
|
||||
try {
|
||||
const sup = spawnSupervisor(h, {
|
||||
OUT_FILE: outFile,
|
||||
GBRAIN_ALLOW_SHELL_JOBS: '1', // parent has it
|
||||
SUP_ALLOW_SHELL_JOBS: '0', // supervisor says NO
|
||||
SUP_MAX_CRASHES: '1',
|
||||
});
|
||||
|
||||
await sup.exited;
|
||||
|
||||
// Worker should have written "UNSET" (parent env var stripped from child).
|
||||
expect(existsSync(outFile)).toBe(true);
|
||||
const childSawEnv = readFileSync(outFile, 'utf8').trim();
|
||||
expect(childSawEnv).toBe('UNSET');
|
||||
} finally {
|
||||
try { unlinkSync(outFile); } catch { /* noop */ }
|
||||
h.cleanup();
|
||||
}
|
||||
}, 15_000);
|
||||
|
||||
it('DOES pass GBRAIN_ALLOW_SHELL_JOBS to child when allowShellJobs is true', async () => {
|
||||
const outFile = join(tmpdir(), `gbrain-sup-env-ok-${process.pid}-${Date.now()}.txt`);
|
||||
try { unlinkSync(outFile); } catch { /* may not exist */ }
|
||||
|
||||
const h = makeHarness('env-pass-on-opt-in', `printf '%s\\n' "\${GBRAIN_ALLOW_SHELL_JOBS-UNSET}" > "$OUT_FILE" ; exit 0`);
|
||||
|
||||
try {
|
||||
const sup = spawnSupervisor(h, {
|
||||
OUT_FILE: outFile,
|
||||
SUP_ALLOW_SHELL_JOBS: '1',
|
||||
SUP_MAX_CRASHES: '1',
|
||||
});
|
||||
|
||||
await sup.exited;
|
||||
|
||||
expect(existsSync(outFile)).toBe(true);
|
||||
expect(readFileSync(outFile, 'utf8').trim()).toBe('1');
|
||||
} finally {
|
||||
try { unlinkSync(outFile); } catch { /* noop */ }
|
||||
h.cleanup();
|
||||
}
|
||||
}, 15_000);
|
||||
});
|
||||
|
||||
describe('integration: audit file rotation + helper', () => {
|
||||
it('computeSupervisorAuditFilename returns supervisor-YYYY-Www.jsonl format', () => {
|
||||
const jan15_2026 = new Date(Date.UTC(2026, 0, 15)); // Thu
|
||||
expect(computeSupervisorAuditFilename(jan15_2026)).toMatch(/^supervisor-2026-W\d\d\.jsonl$/);
|
||||
});
|
||||
|
||||
it('year-boundary ISO week: 2027-01-01 reports as 2026-W53', () => {
|
||||
const jan1_2027 = new Date(Date.UTC(2027, 0, 1));
|
||||
// ISO week: 2027-01-01 is Friday of W53 of 2026
|
||||
expect(computeSupervisorAuditFilename(jan1_2027)).toBe('supervisor-2026-W53.jsonl');
|
||||
});
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user