diff --git a/CHANGELOG.md b/CHANGELOG.md index 783295e34..cb8d65bd4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,53 @@ All notable changes to GBrain will be documented in this file. +## [0.22.1] - 2026-04-26 + +**Autopilot stops being a noisy neighbor.** + +Five hotfixes shipping together: incremental extract, cooperative cycle abort, supervisor watchdog reconnect, session-level connection timeouts, and server-side embed-stale filtering. The wave's theme is unified: gbrain's overnight maintenance loop was reading too much, ignoring abort signals, and quietly poisoning shared infrastructure when things went wrong. After this release the loop only reads pages that changed, bails cleanly when timeouts fire, and recovers from connection-pool poisoning without manual intervention. + +### For everyone + +These two fixes apply to both PGLite (default install) and Postgres / Supabase users: + +- **#417 incremental extract** — `gbrain dream` cycles no longer re-read every markdown file when only a handful changed. The cycle still walks the directory tree to build the link-resolution set (a fast `readdir` pass), but `readFileSync` runs only on pages sync flagged as added or modified. On a 54,461-page production brain this turned a 10-minute extract phase into a sub-second pass; on a 500-page brain you get the same proportional win. +- **#403 cycle abort** — when a cycle phase hits a per-job timeout, `runCycle` now bails at the next phase boundary instead of grinding through extract → embed → orphans while the worker thinks the job is done. A 30-second grace-then-evict safety net in `MinionWorker` frees the slot even if a future handler ignores the abort signal entirely. Cooperative — can't interrupt a phase mid-execution — but prevents the cascade that was wedging workers. + +### For Postgres / Supabase users + +Three fixes that no-op on PGLite (no network, no pooler, no per-connection state): + +- **#406 supervisor watchdog reconnect** — when the connection pool gets poisoned (PgBouncer rotation, Supabase pool bounce), the supervisor's watchdog now detects three consecutive health-check failures and calls `engine.reconnect()` to swap in a fresh pool. Workers crash cleanly on poisoned connections; supervisor catches it within ~3 health-check intervals (~3 minutes) instead of staying degraded until manual restart. Recovery is structural, not per-call magic. +- **#363 session timeouts** *(Contributed by @orendi84)* — every Postgres connection now sets `statement_timeout` and `idle_in_transaction_session_timeout` as connection-time startup parameters. An orphaned pgbouncer backend can no longer hold a `RowExclusiveLock` for hours and block schema migrations. Defaults: 5 minutes each. Override per-GUC via `GBRAIN_STATEMENT_TIMEOUT` / `GBRAIN_IDLE_TX_TIMEOUT` / `GBRAIN_CLIENT_CHECK_INTERVAL`. Closes #361. +- **#409 embed egress** *(Contributed by @atrevino47)* — `embed --stale` now filters server-side on `embedding IS NULL` instead of pulling every chunk's `vector(1536)` over the wire and discarding the unwanted ones client-side. On a fully-embedded 1.5K-page brain that's the difference between ~76 MB per call and a single `count()` round-trip. With autopilot firing every 5–10 minutes plus a 2-hour cron, one production user blew past Supabase's 5 GB free-tier ceiling at 102 GB used — that pattern is gone now. Two new `BrainEngine` methods (`countStaleChunks`, `listStaleChunks`) plus a consistency fix in `upsertChunks` so when `chunk_text` changes without a new embedding, both `embedding` and `embedded_at` reset to NULL together (no more "embedded_at says yes, embedding says NULL"). + +### Production proof point + +The wave was driven by a 54,461-page OpenClaw production deployment where extract took 600+ seconds and the queue stalled at 20–36 waiting jobs (all returning `skipped: cycle_already_running`). All five fixes ran as hotfixes there for 12+ hours stable before this release. The numbers are extreme; the underlying bugs are not. + +### Eng-review tightening + +The original #406 wrapped `executeRaw` in a per-call retry that auto-recovered from connection errors. Eng-review dropped that wrapper as unsound — a SQL-prefix regex isn't a safe idempotence boundary (writable CTEs, side-effecting SELECTs). What ships from #406 is the structural reconnect path, not the per-call retry. Recovery moves up one layer to the supervisor watchdog. See `TODOS.md` for the planned caller-opt-in retry follow-up. + +### Test coverage + +15 new test cases across `test/extract-incremental.test.ts` (new), `test/core/cycle.test.ts`, and `test/connection-resilience.test.ts`: +- 8 cases for `#417`: empty/undefined slugs, [a,b]-only reads, deleted-file handling, mode filter, dry-run, BATCH_SIZE flush, full-slug-set resolution. +- 4 cases for `#417` + Codex F2: cycle threads `pagesAffected` into extract, full-walk fallback, F2 noExtract gating (full cycle vs sync-only). +- 3 cases for D3: `executeRaw` has no per-call retry wrapper, `reconnect()` still exists, supervisor still has 3-strikes path. + +### To take advantage of v0.22.1 + +No manual step. PGLite users get the universal fixes automatically on next cycle. Postgres users additionally get session timeouts on the next pool reconnect, server-side stale filtering on the next `embed --stale`, and supervisor reconnect on the next pool poisoning event. + +```bash +gbrain upgrade +gbrain doctor # verify (optional) +``` + +If anything looks wrong post-upgrade, file an issue: https://github.com/garrytan/gbrain/issues with `gbrain doctor` output. + ## [0.22.0] - 2026-04-25 **Search stops getting swamped by chat logs. Curated pages win by default.** diff --git a/CLAUDE.md b/CLAUDE.md index 581083209..4f16da001 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -27,9 +27,9 @@ strict behavior when unset. - `src/core/engine-factory.ts` — Engine factory with dynamic imports (`'pglite'` | `'postgres'`) - `src/core/pglite-engine.ts` — PGLite (embedded Postgres 17.5 via WASM) implementation, all 40 BrainEngine methods. `addLinksBatch` / `addTimelineEntriesBatch` use multi-row `unnest()` with manual `$N` placeholders. As of v0.13.1, `connect()` wraps `PGlite.create()` in a try/catch that emits an actionable error naming the macOS 26.3 WASM bug (#223) and pointing at `gbrain doctor`; the lock is released on failure so the next process can retry cleanly. v0.22.0: `searchKeyword` and `searchKeywordChunks` multiply `ts_rank` by the source-factor CASE expression at the chunk-grain level; `searchVector` becomes a two-stage CTE — inner CTE keeps `ORDER BY cc.embedding <=> vec` so HNSW stays usable, outer SELECT re-ranks by `raw_score * source_factor`. Inner LIMIT scales with offset to preserve pagination contract. - `src/core/pglite-schema.ts` — PGLite-specific DDL (pgvector, pg_trgm, triggers) -- `src/core/postgres-engine.ts` — Postgres + pgvector implementation (Supabase / self-hosted). `addLinksBatch` / `addTimelineEntriesBatch` use `INSERT ... SELECT FROM unnest($1::text[], ...) JOIN pages ON CONFLICT DO NOTHING RETURNING 1` — 4-5 array params regardless of batch size, sidesteps the 65535-parameter cap. As of v0.12.3, `searchKeyword` / `searchVector` scope `statement_timeout` via `sql.begin` + `SET LOCAL` so the GUC dies with the transaction instead of leaking across the pooled postgres.js connection (contributed by @garagon). `getEmbeddingsByChunkIds` uses `tryParseEmbedding` so one corrupt row skips+warns instead of killing the query. v0.22.0: `searchKeyword`, `searchKeywordChunks`, and `searchVector` apply source-aware ranking by inlining the source-factor CASE and `NOT (col LIKE …)` hard-exclude clause from `src/core/search/sql-ranking.ts`. `searchVector` switches to a two-stage CTE (HNSW-safe inner ORDER BY, source-boost re-rank in the outer SELECT) and carries `p.source_id` through inner→outer for v0.18 multi-source callers. +- `src/core/postgres-engine.ts` — Postgres + pgvector implementation (Supabase / self-hosted). `addLinksBatch` / `addTimelineEntriesBatch` use `INSERT ... SELECT FROM unnest($1::text[], ...) JOIN pages ON CONFLICT DO NOTHING RETURNING 1` — 4-5 array params regardless of batch size, sidesteps the 65535-parameter cap. As of v0.12.3, `searchKeyword` / `searchVector` scope `statement_timeout` via `sql.begin` + `SET LOCAL` so the GUC dies with the transaction instead of leaking across the pooled postgres.js connection (contributed by @garagon). `getEmbeddingsByChunkIds` uses `tryParseEmbedding` so one corrupt row skips+warns instead of killing the query. v0.22.0: `searchKeyword`, `searchKeywordChunks`, and `searchVector` apply source-aware ranking by inlining the source-factor CASE and `NOT (col LIKE …)` hard-exclude clause from `src/core/search/sql-ranking.ts`. `searchVector` switches to a two-stage CTE (HNSW-safe inner ORDER BY, source-boost re-rank in the outer SELECT) and carries `p.source_id` through inner→outer for v0.18 multi-source callers. v0.22.1 (#406): `_savedConfig` retains the connect config; `reconnect()` tears down + recreates the pool from saved config (called by supervisor watchdog after 3 consecutive health-check failures). `executeRaw` is a single-statement passthrough — no per-call retry (D3 dropped that as unsound for non-idempotent statements; recovery is supervisor-driven). v0.22.1 (#363, contributed by @orendi84): `connect()` applies `resolveSessionTimeouts()` from `db.ts` as connection-time startup parameters (`statement_timeout`, `idle_in_transaction_session_timeout`) so orphan pgbouncer backends can't hold locks for hours. v0.22.1 (#409, contributed by @atrevino47): `countStaleChunks()` + `listStaleChunks()` server-side-filter on `embedding IS NULL` for `embed --stale`, eliminating ~76 MB/call client-side pull on a fully-embedded brain; `upsertChunks()` resets both `embedding` AND `embedded_at` to NULL when chunk_text changes without a new embedding (consistency). - `src/core/utils.ts` — Shared SQL utilities extracted from postgres-engine.ts. Exports `parseEmbedding(value)` (throws on unknown input, used by migration + ingest paths where data integrity matters) and as of v0.12.3 `tryParseEmbedding(value)` (returns `null` + warns once per process, used by search/rescore paths where availability matters more than strictness). -- `src/core/db.ts` — Connection management, schema initialization +- `src/core/db.ts` — Connection management, schema initialization. v0.22.1 (#363, contributed by @orendi84): `resolveSessionTimeouts()` returns `statement_timeout` + `idle_in_transaction_session_timeout` (defaults: 5min each, env-overridable via `GBRAIN_STATEMENT_TIMEOUT` / `GBRAIN_IDLE_TX_TIMEOUT` / `GBRAIN_CLIENT_CHECK_INTERVAL`). Both `connect()` (module singleton) and `PostgresEngine.connect()` (worker pool) consume the result via postgres.js's `connection` option, sending GUCs as startup parameters that survive PgBouncer transaction mode (unlike the prior `setSessionDefaults` post-pool SET, kept as a back-compat no-op shim). - `src/commands/migrate-engine.ts` — Bidirectional engine migration (`gbrain migrate --to supabase/pglite`) - `src/core/import-file.ts` — importFromFile + importFromContent (chunk + embed + tags) - `src/core/sync.ts` — Pure sync functions (manifest parsing, filtering, slug conversion) @@ -63,12 +63,14 @@ strict behavior when unset. - `src/core/transcription.ts` — Audio transcription: Groq Whisper (default), OpenAI fallback, ffmpeg segmentation for >25MB - `src/core/enrichment-service.ts` — Global enrichment service: entity slug generation, tier auto-escalation, batch throttling - `src/core/data-research.ts` — Recipe validation, field extraction (MRR/ARR regex), dedup, tracker parsing, HTML stripping -- `src/commands/extract.ts` — `gbrain extract links|timeline|all [--source fs|db]`: batch link/timeline extraction. fs walks markdown files, db walks pages from the engine (mutation-immune snapshot iteration; use this for live brains with no local checkout). As of v0.12.1 there is no in-memory dedup pre-load — candidates are buffered 100 at a time and flushed via `addLinksBatch` / `addTimelineEntriesBatch`; `ON CONFLICT DO NOTHING` enforces uniqueness at the DB layer, and the `created` counter returns real rows inserted (truthful on re-runs). +- `src/commands/embed.ts` — `gbrain embed [--stale|--all] [--slugs ...]`. v0.22.1 (#409, contributed by @atrevino47): `--stale` path now starts with `engine.countStaleChunks()` (single SELECT count(*) WHERE embedding IS NULL, ~50 bytes wire). On a fully-embedded brain that's a 1-line short-circuit — no further reads. When stale chunks exist, `engine.listStaleChunks()` returns just the chunks needing embeddings (slug + chunk_index + chunk_text + metadata, no `vector(1536)` payload). Caller groups by slug, embeds via OpenAI, re-upserts via `upsertChunks`. Replaces the prior page-walk that pulled every chunk's embedding column over the wire and discarded most. +- `src/commands/extract.ts` — `gbrain extract links|timeline|all [--source fs|db]`: batch link/timeline extraction. fs walks markdown files, db walks pages from the engine (mutation-immune snapshot iteration; use this for live brains with no local checkout). As of v0.12.1 there is no in-memory dedup pre-load — candidates are buffered 100 at a time and flushed via `addLinksBatch` / `addTimelineEntriesBatch`; `ON CONFLICT DO NOTHING` enforces uniqueness at the DB layer, and the `created` counter returns real rows inserted (truthful on re-runs). v0.22.1 (#417): `ExtractOpts.slugs?: string[]` enables incremental extract — when set, `extractForSlugs()` reads ONLY those slugs' files (single combined links+timeline pass) instead of the full directory walk. CLI `gbrain extract` keeps full-walk behavior; the cycle path threads sync's `pagesAffected` through. `walkMarkdownFiles(brainDir)` still runs at line 455 to build `allSlugs` for link resolution — see `TODOS.md` for replacing it with `engine.getAllSlugs()`. - `src/commands/graph-query.ts` — `gbrain graph-query [--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. 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/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. v0.22.1 (#403): per-job timeout fires `abort.abort(new Error('timeout'))` then a 30-second grace-then-evict safety net force-evicts the job from `inFlight` and marks it dead in DB if the handler ignores the abort signal — frees the slot even when a handler wedges (the 98-waiting-0-active prod incident driver). +- `src/core/minions/supervisor.ts` — MinionSupervisor process manager. Spawns `gbrain jobs work` as a child, restarts on crash with exponential backoff, periodic health check. v0.22.1 (#406): `consecutiveHealthFailures` counter; on 3 consecutive failures emits `health_warn` with `reason: 'db_connection_degraded'` and calls `engine.reconnect()` to swap in a fresh pool, then resets the counter. Worker exit classifier emits `likely_cause` field on `worker_exited` events: `oom_or_external_kill` (SIGKILL), `graceful_shutdown` (SIGTERM), `runtime_error` (code 1), `clean_exit` (code 0), `unknown`. - `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`). @@ -99,7 +101,7 @@ strict behavior when unset. - `src/core/migrate.ts` — schema-migration runner. Owns the `MIGRATIONS` array (source of truth for schema DDL). v0.14.2 extended the `Migration` interface with `sqlFor?: { postgres?, pglite? }` (engine-specific SQL overrides `sql`) and `transaction?: boolean` (set to false for `CREATE INDEX CONCURRENTLY`, which Postgres refuses inside a transaction; ignored on PGLite since it has no concurrent writers). Migration v14 (fix wave) uses a handler branching on `engine.kind` to run CONCURRENTLY on Postgres (with a pre-drop of any invalid remnant via `pg_index.indisvalid`) and plain `CREATE INDEX` on PGLite. v15 bumps `minion_jobs.max_stalled` default 1→5 and backfills existing non-terminal rows. - `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=` stripped. `getCliOptions()` / `setCliOptions()` expose a module-level singleton so commands reach the resolved flags without parameter threading. `cliOptsToProgressOptions()` maps to reporter options. `childGlobalFlags()` returns the flag suffix to append to `execSync('gbrain ...')` calls in migration orchestrators. `OperationContext.cliOpts` extends shared-op dispatch for MCP callers. -- `src/core/cycle.ts` — v0.17 brain maintenance cycle primitive. `runCycle(engine: BrainEngine | null, opts: CycleOpts): Promise` composes 6 phases in semantically-driven order (lint → backlinks → sync → extract → embed → orphans). Three callers: `gbrain dream` CLI, `gbrain autopilot` daemon's inline path, and the Minions `autopilot-cycle` handler (`src/commands/jobs.ts`). One source of truth for what the brain does overnight. Coordination via `gbrain_cycle_locks` DB table (TTL-based; works through PgBouncer transaction pooling, unlike session-scoped `pg_try_advisory_lock`) + `~/.gbrain/cycle.lock` file lock with PID-liveness for PGLite / engine=null mode. `CycleReport.schema_version: "1"` is the stable agent-consumable shape. `PhaseResult.error: { class, code, message, hint?, docs_url? }` is Stripe-API-tier structured failure info. `yieldBetweenPhases` hook awaited between every phase — Minions handler uses this to renew its job lock and prevent v0.14 stall-death regression. Engine nullable: filesystem phases (lint, backlinks) run without DB; DB phases skip with `status: "skipped", reason: "no_database"`. Lock-skip: read-only phase selections (`--phase orphans`) bypass the cycle lock. +- `src/core/cycle.ts` — v0.17 brain maintenance cycle primitive. `runCycle(engine: BrainEngine | null, opts: CycleOpts): Promise` composes 6 phases in semantically-driven order (lint → backlinks → sync → extract → embed → orphans). Three callers: `gbrain dream` CLI, `gbrain autopilot` daemon's inline path, and the Minions `autopilot-cycle` handler (`src/commands/jobs.ts`). One source of truth for what the brain does overnight. Coordination via `gbrain_cycle_locks` DB table (TTL-based; works through PgBouncer transaction pooling, unlike session-scoped `pg_try_advisory_lock`) + `~/.gbrain/cycle.lock` file lock with PID-liveness for PGLite / engine=null mode. `CycleReport.schema_version: "1"` is the stable agent-consumable shape. `PhaseResult.error: { class, code, message, hint?, docs_url? }` is Stripe-API-tier structured failure info. `yieldBetweenPhases` hook awaited between every phase — Minions handler uses this to renew its job lock and prevent v0.14 stall-death regression. Engine nullable: filesystem phases (lint, backlinks) run without DB; DB phases skip with `status: "skipped", reason: "no_database"`. Lock-skip: read-only phase selections (`--phase orphans`) bypass the cycle lock. v0.22.1 (#403): `CycleOpts.signal?: AbortSignal` propagates the worker's abort signal; `checkAborted()` fires between every phase and throws if the signal is aborted (cooperative — can't interrupt a phase mid-execution). v0.22.1 (#417): `runPhaseSync` returns `pagesAffected` via `SyncPhaseResult`; `runCycle` captures it and threads to `runPhaseExtract` as the 4th arg, enabling incremental extract on the cycle path. v0.22.1 (Codex F2): `runPhaseSync` takes `willRunExtractPhase: boolean` and sets `noExtract: phases.includes('extract')` so `gbrain dream --phase sync` doesn't silently lose extraction. - `src/commands/dream.ts` — v0.17 `gbrain dream` CLI. ~80-line thin alias over `runCycle`. brainDir resolution requires explicit `--dir` OR `sync.repo_path` config (no more walk-up-cwd-for-.git footgun). Flags: `--dry-run`, `--json`, `--phase `, `--pull`, `--dir `. Exit code 1 on status=failed (partial/warn not fatal — don't page on warnings). - `scripts/check-progress-to-stdout.sh` — CI guard against regressing to `\r`-on-stdout progress. Wired into `bun run test` via `scripts/check-progress-to-stdout.sh && bun test` in package.json. - `docs/progress-events.md` — Canonical JSON event schema reference. Stable from v0.15.2, additive only. @@ -401,6 +403,59 @@ in bulk paths, the CI guard will fail the build. `bun build --compile --outfile bin/gbrain src/cli.ts` +## Version locations (single source of truth: `VERSION` file) + +Every release advances the version in **five files at once**. Keep these in +sync. `/ship` enforces this via Step 12's idempotency check (VERSION vs +package.json drift), but the canonical list lives here so future runs and +the auto-update agent know where to look. + +**Required (every release must update all five):** + +| File | What lives there | Format | +|---|---|---| +| `VERSION` | The single source of truth. Read first by `/ship`, the binary, and CI version-gate. | Bare 4-digit string `MAJOR.MINOR.PATCH.MICRO` (e.g. `0.22.1`), no leading `v`, no trailing newline-sensitivity issues. | +| `package.json` | Bun/npm package version. `gbrain --version` reads it via the compiled binary's bundled package metadata. CI version-gate cross-checks this against `VERSION` and fails if they drift. | `"version": "0.22.1"` | +| `CHANGELOG.md` | Top entry header `## [0.22.1] - YYYY-MM-DD` plus the "To take advantage of v0.22.1" block. | Standard Keep-a-Changelog header. | +| `TODOS.md` | Any TODO entries that mention "follow-up from vX.Y.Z" use the version of the release that filed them. Update only when filing NEW follow-up TODOs. | Inline `vX.Y.Z` references in TODO bodies. | +| `CLAUDE.md` | The Key Files section's per-file annotations carry `vX.Y.Z (#NNN)` tags noting which release introduced a behavior. Update whenever a wave's annotations get folded in. | Inline `vX.Y.Z (#NNN, contributed by @user)` references. | + +**Auto-derived (no manual edit; refreshed by their own commands):** + +- `bun.lock` — root-package version is auto-pinned from `package.json`. After + bumping `package.json`, run `bun install` to refresh the lockfile. +- `llms-full.txt` / `llms.txt` — auto-generated documentation bundles. After + any release ship that touches the Key Files annotations in `CLAUDE.md`, + run `bun run build:llms` to regenerate. The bundles do not contain a + version pin per se; they reflect the current state of the docs they index. + +**Historical (DO NOT bump on release):** + +- `skills/migrations/v0.21.0.md` — migration files use the version they + shipped FROM as their filename. v0.21.0's migration always says v0.21.0. +- `src/commands/migrations/v0_21_0.ts` — same: migration code references + the schema version it migrates to. +- `test/migrations-v0_21_0.test.ts`, `test/migration-orchestrator-v0_21_0.test.ts`, + `test/migrate.test.ts` — migration tests reference historical migration + versions; these are correct as-is and should not move. +- `src/core/db.ts`, `src/core/migrate.ts`, `src/core/import-file.ts`, + `src/commands/reindex-code.ts` — code comments cite the release that + introduced a feature. Once written, these are historical record. +- `README.md` — references the latest published feature names by version + (e.g. "v0.21.0 Code Cathedral"); update only when the README's marketing + copy is intentionally being refreshed, NOT on every micro/patch bump. + +**The /ship workflow's version idempotency check:** Step 12 reads +`VERSION` and `package.json`, classifies as FRESH / ALREADY_BUMPED / +DRIFT_STALE_PKG / DRIFT_UNEXPECTED, and refuses to proceed on +DRIFT_UNEXPECTED. This is why the two must move together. + +**The CI version-gate** rejects pushes where `VERSION` and +`package.json` disagree, OR where `VERSION` is not strictly greater +than master's VERSION. If a queue collision claims your version on +master before yours lands, /ship's queue-aware allocator (Step 12) +will detect drift and re-bump on the next run. + ## Pre-ship requirements Before shipping (/ship) or reviewing (/review), always run the full test suite: diff --git a/TODOS.md b/TODOS.md index 9adf285e1..fb3dc62b4 100644 --- a/TODOS.md +++ b/TODOS.md @@ -473,3 +473,48 @@ iteration's residuals. ### Implement AWS Signature V4 for S3 storage backend **Completed:** v0.6.0 (2026-04-10) — replaced with @aws-sdk/client-s3 for proper SigV4 signing. + +### Caller-opt-in retry for `executeRaw` (D3 follow-up from v0.22.1) +**What:** Add `PostgresEngine.executeRawIdempotent(sql, params)` (or a `{retry: true}` parameter flag on `executeRaw`) so callers explicitly opt into auto-retry for statements they know are idempotent. Audit existing call sites and migrate the read-only ones (search, page fetches, etc.) to the new method. + +**Why:** Closes the gap left by D3's drop-the-wrapper decision in v0.22.1. The original #406 wrapped `executeRaw` in a regex-gated retry that was unsound for writable CTEs and side-effecting SELECTs. Recovery moved up to the supervisor watchdog, but per-call recovery for reads (the bulk of `executeRaw` traffic from MCP, search, page fetches) is gone. A caller-opt-in flag puts the idempotency decision where it belongs (at the call site, with full statement context). + +**Pros:** Restores per-call auto-recovery for reads without the phantom-write risk on mutations. Explicit > clever: each call site declares its own idempotency posture. Future caller-added mutations get safe-by-default behavior. + +**Cons:** Touches every existing `executeRaw` call site (~25). Requires careful audit — accidentally tagging a mutation as idempotent re-introduces the phantom-write bug. + +**Context:** Codex F3 demonstrated that `READ_ONLY_PREFIX = /^(\s|--.*\n)*(SELECT|WITH)\b/i` is unsound — `WITH x AS (UPDATE … RETURNING …) SELECT …` matches the prefix but updates a row; `SELECT pg_advisory_xact_lock(...)` is a SELECT with side effects. The plan-eng-review wrap-up in `~/.claude/plans/system-instruction-you-are-working-tender-horizon.md` has the full discussion. + +**Effort estimate:** M (human: ~1 day / CC: ~30 min including call-site audit). +**Priority:** P2 — current behavior (no retry, supervisor recovers within ~3 min) is acceptable but per-call recovery is a real ergonomic win. +**Depends on:** Nothing. + +### Replace `walkMarkdownFiles` with `engine.getAllSlugs()` in `extractForSlugs` (F1 follow-up from v0.22.1) +**What:** The cycle path's `extractForSlugs()` at `src/commands/extract.ts:455` still does a `walkMarkdownFiles(brainDir)` to build the `allSlugs` set for link resolution. On a 54K-page brain that's a single `readdir` traversal (~hundreds of ms — acceptable, dominated by the file-content-read elimination from #417). But `engine.getAllSlugs()` exists at `extract.ts:728` and produces the same set via a single SQL query (~tens of ms). + +**Why:** Eliminates the residual directory walk on every cycle. Codex F1 noted that the v0.22.1 plan's "cycle never re-walks the whole tree again" claim was overstated — it stops READING file contents but still walks the directory. This TODO closes that gap honestly. + +**Pros:** Cycle becomes O(slugs sync touched), not O(total brain size). No more readdir on a growing brain. ~5 LOC change. + +**Cons:** Crosses an FS-vs-DB consistency boundary in the FS-source extract path. Edge case: a file deleted from disk but still in DB. Currently `extractForSlugs` skips with `if (!existsSync(fullPath)) continue` — unchanged. But if a markdown file references a slug whose page exists in DB but file was deleted, the link would resolve via DB but the original extractor caught it. Needs a careful test for this case. + +**Context:** Codex plan-review during v0.22.1 wrap, verified at `extract.ts:455-456`. The plan-eng-review session captured the rationale. + +**Effort estimate:** S (human: ~2 hr / CC: ~10 min including the consistency-edge-case test). +**Priority:** P3 — pure perf, no correctness gap. +**Depends on:** Nothing. + +### `err.code`-based connection-error matching in `postgres-engine.ts` (B1 follow-up from v0.22.1) +**What:** The CONNECTION_ERROR_PATTERNS array (~12 strings: `ECONNREFUSED`, `connection terminated`, `password authentication failed`, etc.) matched against `err.message` and `err.code`. Replace with structured matching against `err.code` only, using postgres.js's typed error classes (`PostgresError` with structured codes). + +**Why:** String matching against error messages breaks on library upgrades (postgres.js could change its error message phrasing without bumping major). Code matching is durable. The Layer 1 cleanup follows: gbrain itself doesn't define connection-error codes; it should defer to postgres.js's classification. + +**Pros:** More durable across library updates. Less code (drop the 12-string array). Follows the typed-errors pattern v0.21.0 introduced (`src/core/errors.ts`). + +**Cons:** Requires verifying which `err.code` values postgres.js actually exposes for each connection-failure mode. May need fallback to message-substring matching for codes that postgres.js doesn't surface. + +**Context:** Section 2/B1 from the v0.22.1 plan-eng-review. After D3 dropped the per-call retry, `isConnectionError` is no longer in the hot path — only the supervisor watchdog cares about classifying connection errors, and it currently catches *anything*. This TODO is a cleanup pass when someone next touches that surface. + +**Effort estimate:** S (human: ~2 hr / CC: ~10 min). +**Priority:** P3. +**Depends on:** The above caller-opt-in retry (#1) is the natural co-lander since both touch the same error-classification surface. diff --git a/VERSION b/VERSION index 215740905..a723ece79 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.22.0 +0.22.1 diff --git a/llms-full.txt b/llms-full.txt index b74aa7d6c..c6ac1094b 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -106,9 +106,9 @@ strict behavior when unset. - `src/core/engine-factory.ts` — Engine factory with dynamic imports (`'pglite'` | `'postgres'`) - `src/core/pglite-engine.ts` — PGLite (embedded Postgres 17.5 via WASM) implementation, all 40 BrainEngine methods. `addLinksBatch` / `addTimelineEntriesBatch` use multi-row `unnest()` with manual `$N` placeholders. As of v0.13.1, `connect()` wraps `PGlite.create()` in a try/catch that emits an actionable error naming the macOS 26.3 WASM bug (#223) and pointing at `gbrain doctor`; the lock is released on failure so the next process can retry cleanly. v0.22.0: `searchKeyword` and `searchKeywordChunks` multiply `ts_rank` by the source-factor CASE expression at the chunk-grain level; `searchVector` becomes a two-stage CTE — inner CTE keeps `ORDER BY cc.embedding <=> vec` so HNSW stays usable, outer SELECT re-ranks by `raw_score * source_factor`. Inner LIMIT scales with offset to preserve pagination contract. - `src/core/pglite-schema.ts` — PGLite-specific DDL (pgvector, pg_trgm, triggers) -- `src/core/postgres-engine.ts` — Postgres + pgvector implementation (Supabase / self-hosted). `addLinksBatch` / `addTimelineEntriesBatch` use `INSERT ... SELECT FROM unnest($1::text[], ...) JOIN pages ON CONFLICT DO NOTHING RETURNING 1` — 4-5 array params regardless of batch size, sidesteps the 65535-parameter cap. As of v0.12.3, `searchKeyword` / `searchVector` scope `statement_timeout` via `sql.begin` + `SET LOCAL` so the GUC dies with the transaction instead of leaking across the pooled postgres.js connection (contributed by @garagon). `getEmbeddingsByChunkIds` uses `tryParseEmbedding` so one corrupt row skips+warns instead of killing the query. v0.22.0: `searchKeyword`, `searchKeywordChunks`, and `searchVector` apply source-aware ranking by inlining the source-factor CASE and `NOT (col LIKE …)` hard-exclude clause from `src/core/search/sql-ranking.ts`. `searchVector` switches to a two-stage CTE (HNSW-safe inner ORDER BY, source-boost re-rank in the outer SELECT) and carries `p.source_id` through inner→outer for v0.18 multi-source callers. +- `src/core/postgres-engine.ts` — Postgres + pgvector implementation (Supabase / self-hosted). `addLinksBatch` / `addTimelineEntriesBatch` use `INSERT ... SELECT FROM unnest($1::text[], ...) JOIN pages ON CONFLICT DO NOTHING RETURNING 1` — 4-5 array params regardless of batch size, sidesteps the 65535-parameter cap. As of v0.12.3, `searchKeyword` / `searchVector` scope `statement_timeout` via `sql.begin` + `SET LOCAL` so the GUC dies with the transaction instead of leaking across the pooled postgres.js connection (contributed by @garagon). `getEmbeddingsByChunkIds` uses `tryParseEmbedding` so one corrupt row skips+warns instead of killing the query. v0.22.0: `searchKeyword`, `searchKeywordChunks`, and `searchVector` apply source-aware ranking by inlining the source-factor CASE and `NOT (col LIKE …)` hard-exclude clause from `src/core/search/sql-ranking.ts`. `searchVector` switches to a two-stage CTE (HNSW-safe inner ORDER BY, source-boost re-rank in the outer SELECT) and carries `p.source_id` through inner→outer for v0.18 multi-source callers. v0.22.1 (#406): `_savedConfig` retains the connect config; `reconnect()` tears down + recreates the pool from saved config (called by supervisor watchdog after 3 consecutive health-check failures). `executeRaw` is a single-statement passthrough — no per-call retry (D3 dropped that as unsound for non-idempotent statements; recovery is supervisor-driven). v0.22.1 (#363, contributed by @orendi84): `connect()` applies `resolveSessionTimeouts()` from `db.ts` as connection-time startup parameters (`statement_timeout`, `idle_in_transaction_session_timeout`) so orphan pgbouncer backends can't hold locks for hours. v0.22.1 (#409, contributed by @atrevino47): `countStaleChunks()` + `listStaleChunks()` server-side-filter on `embedding IS NULL` for `embed --stale`, eliminating ~76 MB/call client-side pull on a fully-embedded brain; `upsertChunks()` resets both `embedding` AND `embedded_at` to NULL when chunk_text changes without a new embedding (consistency). - `src/core/utils.ts` — Shared SQL utilities extracted from postgres-engine.ts. Exports `parseEmbedding(value)` (throws on unknown input, used by migration + ingest paths where data integrity matters) and as of v0.12.3 `tryParseEmbedding(value)` (returns `null` + warns once per process, used by search/rescore paths where availability matters more than strictness). -- `src/core/db.ts` — Connection management, schema initialization +- `src/core/db.ts` — Connection management, schema initialization. v0.22.1 (#363, contributed by @orendi84): `resolveSessionTimeouts()` returns `statement_timeout` + `idle_in_transaction_session_timeout` (defaults: 5min each, env-overridable via `GBRAIN_STATEMENT_TIMEOUT` / `GBRAIN_IDLE_TX_TIMEOUT` / `GBRAIN_CLIENT_CHECK_INTERVAL`). Both `connect()` (module singleton) and `PostgresEngine.connect()` (worker pool) consume the result via postgres.js's `connection` option, sending GUCs as startup parameters that survive PgBouncer transaction mode (unlike the prior `setSessionDefaults` post-pool SET, kept as a back-compat no-op shim). - `src/commands/migrate-engine.ts` — Bidirectional engine migration (`gbrain migrate --to supabase/pglite`) - `src/core/import-file.ts` — importFromFile + importFromContent (chunk + embed + tags) - `src/core/sync.ts` — Pure sync functions (manifest parsing, filtering, slug conversion) @@ -142,12 +142,14 @@ strict behavior when unset. - `src/core/transcription.ts` — Audio transcription: Groq Whisper (default), OpenAI fallback, ffmpeg segmentation for >25MB - `src/core/enrichment-service.ts` — Global enrichment service: entity slug generation, tier auto-escalation, batch throttling - `src/core/data-research.ts` — Recipe validation, field extraction (MRR/ARR regex), dedup, tracker parsing, HTML stripping -- `src/commands/extract.ts` — `gbrain extract links|timeline|all [--source fs|db]`: batch link/timeline extraction. fs walks markdown files, db walks pages from the engine (mutation-immune snapshot iteration; use this for live brains with no local checkout). As of v0.12.1 there is no in-memory dedup pre-load — candidates are buffered 100 at a time and flushed via `addLinksBatch` / `addTimelineEntriesBatch`; `ON CONFLICT DO NOTHING` enforces uniqueness at the DB layer, and the `created` counter returns real rows inserted (truthful on re-runs). +- `src/commands/embed.ts` — `gbrain embed [--stale|--all] [--slugs ...]`. v0.22.1 (#409, contributed by @atrevino47): `--stale` path now starts with `engine.countStaleChunks()` (single SELECT count(*) WHERE embedding IS NULL, ~50 bytes wire). On a fully-embedded brain that's a 1-line short-circuit — no further reads. When stale chunks exist, `engine.listStaleChunks()` returns just the chunks needing embeddings (slug + chunk_index + chunk_text + metadata, no `vector(1536)` payload). Caller groups by slug, embeds via OpenAI, re-upserts via `upsertChunks`. Replaces the prior page-walk that pulled every chunk's embedding column over the wire and discarded most. +- `src/commands/extract.ts` — `gbrain extract links|timeline|all [--source fs|db]`: batch link/timeline extraction. fs walks markdown files, db walks pages from the engine (mutation-immune snapshot iteration; use this for live brains with no local checkout). As of v0.12.1 there is no in-memory dedup pre-load — candidates are buffered 100 at a time and flushed via `addLinksBatch` / `addTimelineEntriesBatch`; `ON CONFLICT DO NOTHING` enforces uniqueness at the DB layer, and the `created` counter returns real rows inserted (truthful on re-runs). v0.22.1 (#417): `ExtractOpts.slugs?: string[]` enables incremental extract — when set, `extractForSlugs()` reads ONLY those slugs' files (single combined links+timeline pass) instead of the full directory walk. CLI `gbrain extract` keeps full-walk behavior; the cycle path threads sync's `pagesAffected` through. `walkMarkdownFiles(brainDir)` still runs at line 455 to build `allSlugs` for link resolution — see `TODOS.md` for replacing it with `engine.getAllSlugs()`. - `src/commands/graph-query.ts` — `gbrain graph-query [--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. 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/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. v0.22.1 (#403): per-job timeout fires `abort.abort(new Error('timeout'))` then a 30-second grace-then-evict safety net force-evicts the job from `inFlight` and marks it dead in DB if the handler ignores the abort signal — frees the slot even when a handler wedges (the 98-waiting-0-active prod incident driver). +- `src/core/minions/supervisor.ts` — MinionSupervisor process manager. Spawns `gbrain jobs work` as a child, restarts on crash with exponential backoff, periodic health check. v0.22.1 (#406): `consecutiveHealthFailures` counter; on 3 consecutive failures emits `health_warn` with `reason: 'db_connection_degraded'` and calls `engine.reconnect()` to swap in a fresh pool, then resets the counter. Worker exit classifier emits `likely_cause` field on `worker_exited` events: `oom_or_external_kill` (SIGKILL), `graceful_shutdown` (SIGTERM), `runtime_error` (code 1), `clean_exit` (code 0), `unknown`. - `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`). @@ -178,7 +180,7 @@ strict behavior when unset. - `src/core/migrate.ts` — schema-migration runner. Owns the `MIGRATIONS` array (source of truth for schema DDL). v0.14.2 extended the `Migration` interface with `sqlFor?: { postgres?, pglite? }` (engine-specific SQL overrides `sql`) and `transaction?: boolean` (set to false for `CREATE INDEX CONCURRENTLY`, which Postgres refuses inside a transaction; ignored on PGLite since it has no concurrent writers). Migration v14 (fix wave) uses a handler branching on `engine.kind` to run CONCURRENTLY on Postgres (with a pre-drop of any invalid remnant via `pg_index.indisvalid`) and plain `CREATE INDEX` on PGLite. v15 bumps `minion_jobs.max_stalled` default 1→5 and backfills existing non-terminal rows. - `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=` stripped. `getCliOptions()` / `setCliOptions()` expose a module-level singleton so commands reach the resolved flags without parameter threading. `cliOptsToProgressOptions()` maps to reporter options. `childGlobalFlags()` returns the flag suffix to append to `execSync('gbrain ...')` calls in migration orchestrators. `OperationContext.cliOpts` extends shared-op dispatch for MCP callers. -- `src/core/cycle.ts` — v0.17 brain maintenance cycle primitive. `runCycle(engine: BrainEngine | null, opts: CycleOpts): Promise` composes 6 phases in semantically-driven order (lint → backlinks → sync → extract → embed → orphans). Three callers: `gbrain dream` CLI, `gbrain autopilot` daemon's inline path, and the Minions `autopilot-cycle` handler (`src/commands/jobs.ts`). One source of truth for what the brain does overnight. Coordination via `gbrain_cycle_locks` DB table (TTL-based; works through PgBouncer transaction pooling, unlike session-scoped `pg_try_advisory_lock`) + `~/.gbrain/cycle.lock` file lock with PID-liveness for PGLite / engine=null mode. `CycleReport.schema_version: "1"` is the stable agent-consumable shape. `PhaseResult.error: { class, code, message, hint?, docs_url? }` is Stripe-API-tier structured failure info. `yieldBetweenPhases` hook awaited between every phase — Minions handler uses this to renew its job lock and prevent v0.14 stall-death regression. Engine nullable: filesystem phases (lint, backlinks) run without DB; DB phases skip with `status: "skipped", reason: "no_database"`. Lock-skip: read-only phase selections (`--phase orphans`) bypass the cycle lock. +- `src/core/cycle.ts` — v0.17 brain maintenance cycle primitive. `runCycle(engine: BrainEngine | null, opts: CycleOpts): Promise` composes 6 phases in semantically-driven order (lint → backlinks → sync → extract → embed → orphans). Three callers: `gbrain dream` CLI, `gbrain autopilot` daemon's inline path, and the Minions `autopilot-cycle` handler (`src/commands/jobs.ts`). One source of truth for what the brain does overnight. Coordination via `gbrain_cycle_locks` DB table (TTL-based; works through PgBouncer transaction pooling, unlike session-scoped `pg_try_advisory_lock`) + `~/.gbrain/cycle.lock` file lock with PID-liveness for PGLite / engine=null mode. `CycleReport.schema_version: "1"` is the stable agent-consumable shape. `PhaseResult.error: { class, code, message, hint?, docs_url? }` is Stripe-API-tier structured failure info. `yieldBetweenPhases` hook awaited between every phase — Minions handler uses this to renew its job lock and prevent v0.14 stall-death regression. Engine nullable: filesystem phases (lint, backlinks) run without DB; DB phases skip with `status: "skipped", reason: "no_database"`. Lock-skip: read-only phase selections (`--phase orphans`) bypass the cycle lock. v0.22.1 (#403): `CycleOpts.signal?: AbortSignal` propagates the worker's abort signal; `checkAborted()` fires between every phase and throws if the signal is aborted (cooperative — can't interrupt a phase mid-execution). v0.22.1 (#417): `runPhaseSync` returns `pagesAffected` via `SyncPhaseResult`; `runCycle` captures it and threads to `runPhaseExtract` as the 4th arg, enabling incremental extract on the cycle path. v0.22.1 (Codex F2): `runPhaseSync` takes `willRunExtractPhase: boolean` and sets `noExtract: phases.includes('extract')` so `gbrain dream --phase sync` doesn't silently lose extraction. - `src/commands/dream.ts` — v0.17 `gbrain dream` CLI. ~80-line thin alias over `runCycle`. brainDir resolution requires explicit `--dir` OR `sync.repo_path` config (no more walk-up-cwd-for-.git footgun). Flags: `--dry-run`, `--json`, `--phase `, `--pull`, `--dir `. Exit code 1 on status=failed (partial/warn not fatal — don't page on warnings). - `scripts/check-progress-to-stdout.sh` — CI guard against regressing to `\r`-on-stdout progress. Wired into `bun run test` via `scripts/check-progress-to-stdout.sh && bun test` in package.json. - `docs/progress-events.md` — Canonical JSON event schema reference. Stable from v0.15.2, additive only. @@ -480,6 +482,59 @@ in bulk paths, the CI guard will fail the build. `bun build --compile --outfile bin/gbrain src/cli.ts` +## Version locations (single source of truth: `VERSION` file) + +Every release advances the version in **five files at once**. Keep these in +sync. `/ship` enforces this via Step 12's idempotency check (VERSION vs +package.json drift), but the canonical list lives here so future runs and +the auto-update agent know where to look. + +**Required (every release must update all five):** + +| File | What lives there | Format | +|---|---|---| +| `VERSION` | The single source of truth. Read first by `/ship`, the binary, and CI version-gate. | Bare 4-digit string `MAJOR.MINOR.PATCH.MICRO` (e.g. `0.22.1`), no leading `v`, no trailing newline-sensitivity issues. | +| `package.json` | Bun/npm package version. `gbrain --version` reads it via the compiled binary's bundled package metadata. CI version-gate cross-checks this against `VERSION` and fails if they drift. | `"version": "0.22.1"` | +| `CHANGELOG.md` | Top entry header `## [0.22.1] - YYYY-MM-DD` plus the "To take advantage of v0.22.1" block. | Standard Keep-a-Changelog header. | +| `TODOS.md` | Any TODO entries that mention "follow-up from vX.Y.Z" use the version of the release that filed them. Update only when filing NEW follow-up TODOs. | Inline `vX.Y.Z` references in TODO bodies. | +| `CLAUDE.md` | The Key Files section's per-file annotations carry `vX.Y.Z (#NNN)` tags noting which release introduced a behavior. Update whenever a wave's annotations get folded in. | Inline `vX.Y.Z (#NNN, contributed by @user)` references. | + +**Auto-derived (no manual edit; refreshed by their own commands):** + +- `bun.lock` — root-package version is auto-pinned from `package.json`. After + bumping `package.json`, run `bun install` to refresh the lockfile. +- `llms-full.txt` / `llms.txt` — auto-generated documentation bundles. After + any release ship that touches the Key Files annotations in `CLAUDE.md`, + run `bun run build:llms` to regenerate. The bundles do not contain a + version pin per se; they reflect the current state of the docs they index. + +**Historical (DO NOT bump on release):** + +- `skills/migrations/v0.21.0.md` — migration files use the version they + shipped FROM as their filename. v0.21.0's migration always says v0.21.0. +- `src/commands/migrations/v0_21_0.ts` — same: migration code references + the schema version it migrates to. +- `test/migrations-v0_21_0.test.ts`, `test/migration-orchestrator-v0_21_0.test.ts`, + `test/migrate.test.ts` — migration tests reference historical migration + versions; these are correct as-is and should not move. +- `src/core/db.ts`, `src/core/migrate.ts`, `src/core/import-file.ts`, + `src/commands/reindex-code.ts` — code comments cite the release that + introduced a feature. Once written, these are historical record. +- `README.md` — references the latest published feature names by version + (e.g. "v0.21.0 Code Cathedral"); update only when the README's marketing + copy is intentionally being refreshed, NOT on every micro/patch bump. + +**The /ship workflow's version idempotency check:** Step 12 reads +`VERSION` and `package.json`, classifies as FRESH / ALREADY_BUMPED / +DRIFT_STALE_PKG / DRIFT_UNEXPECTED, and refuses to proceed on +DRIFT_UNEXPECTED. This is why the two must move together. + +**The CI version-gate** rejects pushes where `VERSION` and +`package.json` disagree, OR where `VERSION` is not strictly greater +than master's VERSION. If a queue collision claims your version on +master before yours lands, /ship's queue-aware allocator (Step 12) +will detect drift and re-bump on the next run. + ## Pre-ship requirements Before shipping (/ship) or reviewing (/review), always run the full test suite: diff --git a/package.json b/package.json index 5536a24b6..eea700a78 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "gbrain", - "version": "0.21.0", + "version": "0.22.1", "description": "Postgres-native personal knowledge brain with hybrid RAG search", "type": "module", "main": "src/core/index.ts", diff --git a/src/commands/doctor.ts b/src/commands/doctor.ts index dd16bd24b..dccca7030 100644 --- a/src/commands/doctor.ts +++ b/src/commands/doctor.ts @@ -5,6 +5,7 @@ import { checkResolvable } from '../core/check-resolvable.ts'; import { autoFixDryViolations, type AutoFixReport, type FixOutcome } from '../core/dry-fix.ts'; import { findRepoRoot } from '../core/repo-root.ts'; import { loadCompletedMigrations } from '../core/preferences.ts'; +import { compareVersions } from './migrations/index.ts'; import { createProgress, startHeartbeat, type ProgressReporter } from '../core/progress.ts'; import { getCliOptions, cliOptsToProgressOptions } from '../core/cli-options.ts'; import type { DbUrlSource } from '../core/config.ts'; @@ -110,6 +111,15 @@ export async function runDoctor(engine: BrainEngine | null, args: string[], dbSo // Typical cause: v0.11.0 stopgap wrote a partial record but nobody ran // `gbrain apply-migrations --yes` afterward. This check fires on every // `gbrain doctor` invocation so your OpenClaw's health skill catches it. + // + // Forward-progress override: a partial entry for vX.Y.Z is treated as + // stale (not stuck) if there is a `complete` entry for any vA.B.C >= vX.Y.Z + // anywhere in the file. The reasoning: if a newer migration successfully + // landed, the install moved past the older partial — the old record is + // historical noise from a stopgap that never finished cleanly, but the + // schema clearly advanced. Without this, every install that went through + // a v0.11.0 stopgap and then upgraded carries the "MINIONS HALF-INSTALLED" + // flag forever, even on installs that have been at v0.22+ for months. try { const completed = loadCompletedMigrations(); const byVersion = new Map(); @@ -119,8 +129,17 @@ export async function runDoctor(engine: BrainEngine | null, args: string[], dbSo if (entry.status === 'partial') seen.partial = true; byVersion.set(entry.version, seen); } + const completedVersions = Array.from(byVersion.entries()) + .filter(([, s]) => s.complete) + .map(([v]) => v); const stuck = Array.from(byVersion.entries()) - .filter(([, s]) => s.partial && !s.complete) + .filter(([v, s]) => { + if (!s.partial || s.complete) return false; + // Forward-progress override: if any version >= v has completed, the + // partial is stale. compareVersions returns 1 when first arg is newer. + const supersededBy = completedVersions.find(cv => compareVersions(cv, v) >= 0); + return supersededBy === undefined; + }) .map(([v]) => v); if (stuck.length > 0) { checks.push({ diff --git a/src/commands/embed.ts b/src/commands/embed.ts index 0152ccf85..8813632b9 100644 --- a/src/commands/embed.ts +++ b/src/commands/embed.ts @@ -220,6 +220,23 @@ async function embedAll( result: EmbedResult, onProgress?: (done: number, total: number, embedded: number) => void, ) { + // ───────────────────────────────────────────────────────────── + // Stale-only fast path: avoid the listPages + per-page getChunks + // bomb that pulled every page row + every chunk's embedding column + // (~76 MB on a 1.5K-page brain) only to client-side-filter for + // chunks where embedding IS NULL. The new path issues one SQL + // pre-check + at most one slug-grouped SELECT excluding the + // (always-null on stale rows) embedding column. On a 100%-embedded + // brain (the autopilot common case) we exit after ~50 bytes wire. + // + // For --all (staleOnly=false) we keep the original behavior — the + // user is explicitly asking to re-embed everything, including + // chunks that already have embeddings. + // ───────────────────────────────────────────────────────────── + if (staleOnly) { + return await embedAllStale(engine, dryRun, result, onProgress); + } + const pages = await engine.listPages({ limit: 100000 }); let processed = 0; @@ -235,9 +252,7 @@ async function embedAll( async function embedOnePage(page: typeof pages[number]) { const chunks = await engine.getChunks(page.slug); - const toEmbed = staleOnly - ? chunks.filter(c => !c.embedded_at) - : chunks; + const toEmbed = chunks; // staleOnly path handled above via embedAllStale result.total_chunks += chunks.length; result.skipped += chunks.length - toEmbed.length; @@ -306,3 +321,120 @@ async function embedAll( console.log(`Embedded ${result.embedded} chunks across ${pages.length} pages`); } } + +/** + * SQL-side stale path: replaces the listPages + per-page getChunks + * walk with a count + slug-grouped SELECT. Preserves the existing + * functional contract (every chunk where embedding IS NULL gets + * embedded; nothing else is touched) without paying egress on + * already-embedded chunks. + * + * Why a separate function: the staleOnly path doesn't need + * listPages at all and groups by slug differently. Forking the + * function makes the read-bytes path explicit and keeps the --all + * path verbatim from prior behavior. + * + * Staleness predicate: `embedding IS NULL`. We deliberately do NOT + * use `embedded_at IS NULL` here — the bulk-import path can leave + * embedded_at populated while embedding is NULL (see upsertChunks + * consistency notes), and `embedding IS NULL` is the truth source + * for "this chunk needs an embedding". + */ +async function embedAllStale( + engine: BrainEngine, + dryRun: boolean, + result: EmbedResult, + onProgress?: (done: number, total: number, embedded: number) => void, +) { + // Pre-flight: 0 stale chunks → nothing to do, no further DB reads. + // Cheapest possible exit on the autopilot common case. + const staleCount = await engine.countStaleChunks(); + if (staleCount === 0) { + if (dryRun) { + console.log('[dry-run] Would embed 0 chunks (0 stale found)'); + } else { + console.log('Embedded 0 chunks (0 stale found)'); + } + return; + } + + // Pull only the stale chunks (no embedding column). + const staleRows = await engine.listStaleChunks(); + // Group by slug so each slug → array of stale chunks for batched embedding. + const bySlug = new Map(); + for (const row of staleRows) { + const list = bySlug.get(row.slug); + if (list) list.push(row); + else bySlug.set(row.slug, [row]); + } + + const slugs = Array.from(bySlug.keys()); + const totalStaleChunks = staleRows.length; + result.total_chunks += totalStaleChunks; + // skipped is "chunks we considered and skipped due to having an embedding". + // We never considered the non-stale chunks here, so leave skipped at 0. + // Callers reading EmbedResult who care about coverage should call + // engine.getStats() / engine.getHealth() afterward. + + if (dryRun) { + result.would_embed += totalStaleChunks; + result.pages_processed += slugs.length; + if (onProgress) { + // Emit a single tick to satisfy the contract (CLI progress reporters + // expect at least one start/finish pair). + onProgress(slugs.length, slugs.length, 0); + } + console.log(`[dry-run] Would embed ${totalStaleChunks} chunks across ${slugs.length} pages`); + return; + } + + const CONCURRENCY = parseInt(process.env.GBRAIN_EMBED_CONCURRENCY || '20', 10); + let processed = 0; + + async function embedOneSlug(slug: string) { + const stale = bySlug.get(slug)!; + try { + const embeddings = await embedBatch(stale.map(c => c.chunk_text)); + // CRITICAL: passing ONLY the stale indices to upsertChunks would + // delete every non-stale chunk on the same page (the != ALL filter + // wipes any chunk_index NOT in the input). To preserve them, we + // re-fetch existing chunks for this page and merge. Bounded by the + // stale slug count, not by total slugs — autopilot common case + // is 0 stale (pre-flight short-circuit, never reaches this path). + const existing = await engine.getChunks(slug); + const staleIdxToEmbedding = new Map(); + for (let j = 0; j < stale.length; j++) { + staleIdxToEmbedding.set(stale[j].chunk_index, embeddings[j]); + } + const merged: ChunkInput[] = existing.map(c => ({ + chunk_index: c.chunk_index, + chunk_text: c.chunk_text, + chunk_source: c.chunk_source, + // For stale chunks: pass the new embedding. + // For non-stale chunks: pass undefined → COALESCE preserves existing embedding. + embedding: staleIdxToEmbedding.get(c.chunk_index) ?? undefined, + token_count: c.token_count || Math.ceil(c.chunk_text.length / 4), + })); + await engine.upsertChunks(slug, merged); + result.embedded += stale.length; + } catch (e: unknown) { + console.error(`\n Error embedding ${slug}: ${e instanceof Error ? e.message : e}`); + } + processed++; + result.pages_processed++; + onProgress?.(processed, slugs.length, result.embedded); + } + + let nextIdx = 0; + async function worker() { + while (nextIdx < slugs.length) { + const idx = nextIdx++; + await embedOneSlug(slugs[idx]); + } + } + + const numWorkers = Math.min(CONCURRENCY, slugs.length); + await Promise.all(Array.from({ length: numWorkers }, () => worker())); + + console.log(`Embedded ${result.embedded} chunks across ${slugs.length} pages`); +} diff --git a/src/commands/extract.ts b/src/commands/extract.ts index f4a6ac41e..9b9d08baa 100644 --- a/src/commands/extract.ts +++ b/src/commands/extract.ts @@ -295,6 +295,13 @@ export interface ExtractOpts { dryRun?: boolean; /** Emit JSON (progress to stderr, result to stdout) instead of human text. */ jsonMode?: boolean; + /** + * Incremental mode: only extract from these specific slugs. + * When provided, skips the full directory walk and reads only the + * files corresponding to these slugs. Massive perf win on large brains. + * Pass undefined or omit for a full walk (CLI / first-run path). + */ + slugs?: string[]; } /** @@ -315,6 +322,21 @@ export async function runExtractCore(engine: BrainEngine, opts: ExtractOpts): Pr const jsonMode = !!opts.jsonMode; const result: ExtractResult = { links_created: 0, timeline_entries_created: 0, pages_processed: 0 }; + // Incremental path: if specific slugs provided, only extract from those files. + // This is the cycle path — sync tells us what changed, we only re-extract those. + if (opts.slugs !== undefined) { + if (opts.slugs.length === 0) { + // Nothing changed — skip entirely. + return result; + } + const r = await extractForSlugs(engine, opts.dir, opts.slugs, opts.mode, dryRun, jsonMode); + result.links_created = r.links_created; + result.timeline_entries_created = r.timeline_created; + result.pages_processed = r.pages; + return result; + } + + // Full walk path: CLI `gbrain extract` or first-run. if (opts.mode === 'links' || opts.mode === 'all') { const r = await extractLinksFromDir(engine, opts.dir, dryRun, jsonMode); result.links_created = r.created; @@ -411,6 +433,118 @@ export async function runExtract(engine: BrainEngine, args: string[]) { } } +/** + * Incremental extract: process only the specified slugs. + * + * Instead of walking 54K+ files, reads only the files that sync says changed. + * Still needs the full slug set for link resolution (resolveSlug needs to know + * all valid targets), but that's a single readdir, not 54K readFileSync calls. + * + * Combines links + timeline extraction in a single pass over each file — + * the full-walk path reads every file TWICE (once for links, once for timeline). + */ +async function extractForSlugs( + engine: BrainEngine, + brainDir: string, + slugs: string[], + mode: 'links' | 'timeline' | 'all', + dryRun: boolean, + jsonMode: boolean, +): Promise<{ links_created: number; timeline_created: number; pages: number }> { + // Build the full slug set for link resolution (fast: just readdir, no file reads) + const allFiles = walkMarkdownFiles(brainDir); + const allSlugs = new Set(allFiles.map(f => f.relPath.replace('.md', ''))); + + const doLinks = mode === 'links' || mode === 'all'; + const doTimeline = mode === 'timeline' || mode === 'all'; + + const progress = createProgress(cliOptsToProgressOptions(getCliOptions())); + progress.start('extract.incremental', slugs.length); + + let linksCreated = 0; + let timelineCreated = 0; + let pagesProcessed = 0; + + const linkBatch: LinkBatchInput[] = []; + const timelineBatch: TimelineBatchInput[] = []; + + async function flushLinks() { + if (linkBatch.length === 0) return; + try { + linksCreated += await engine.addLinksBatch(linkBatch); + } catch (e) { + const msg = e instanceof Error ? e.message : String(e); + if (!jsonMode) console.error(` link batch error (${linkBatch.length} rows lost): ${msg}`); + } finally { + linkBatch.length = 0; + } + } + + async function flushTimeline() { + if (timelineBatch.length === 0) return; + try { + timelineCreated += await engine.addTimelineEntriesBatch(timelineBatch); + } catch (e) { + const msg = e instanceof Error ? e.message : String(e); + if (!jsonMode) console.error(` timeline batch error (${timelineBatch.length} rows lost): ${msg}`); + } finally { + timelineBatch.length = 0; + } + } + + for (const slug of slugs) { + const relPath = slug + '.md'; + const fullPath = join(brainDir, relPath); + + try { + if (!existsSync(fullPath)) continue; // deleted file — sync already handled removal + const content = readFileSync(fullPath, 'utf-8'); + + // Links + if (doLinks) { + const links = await extractLinksFromFile(content, relPath, allSlugs); + for (const link of links) { + if (dryRun) { + if (!jsonMode) console.log(` ${link.from_slug} → ${link.to_slug} (${link.link_type})`); + linksCreated++; + } else { + linkBatch.push(link); + if (linkBatch.length >= BATCH_SIZE) await flushLinks(); + } + } + } + + // Timeline + if (doTimeline) { + const entries = extractTimelineFromContent(content, slug); + for (const entry of entries) { + if (dryRun) { + if (!jsonMode) console.log(` ${entry.slug}: ${entry.date} — ${entry.summary}`); + timelineCreated++; + } else { + timelineBatch.push({ slug: entry.slug, date: entry.date, source: entry.source, summary: entry.summary, detail: entry.detail }); + if (timelineBatch.length >= BATCH_SIZE) await flushTimeline(); + } + } + } + + pagesProcessed++; + } catch { /* skip unreadable */ } + progress.tick(1); + } + + await flushLinks(); + await flushTimeline(); + progress.finish(); + + if (!jsonMode) { + const label = dryRun ? '(dry run) would create' : 'created'; + console.log(`Incremental extract: ${label} ${linksCreated} link(s), ${timelineCreated} timeline entries from ${pagesProcessed}/${slugs.length} page(s)`); + } + + return { links_created: linksCreated, timeline_created: timelineCreated, pages: pagesProcessed }; +} + async function extractLinksFromDir( engine: BrainEngine, brainDir: string, dryRun: boolean, jsonMode: boolean, ): Promise<{ created: number; pages: number }> { diff --git a/src/commands/jobs.ts b/src/commands/jobs.ts index f82c3b300..8ef04ddba 100644 --- a/src/commands/jobs.ts +++ b/src/commands/jobs.ts @@ -913,6 +913,7 @@ export async function registerBuiltinHandlers(worker: MinionWorker, engine: Brai const report = await runCycle(engine, { brainDir: repoPath, pull: true, // autopilot daemon opts into git pull + signal: job.signal, // propagate abort so cycle bails on timeout/cancel yieldBetweenPhases: async () => { // Yield to the event loop so worker lock-renewal can fire. await new Promise(r => setImmediate(r)); diff --git a/src/core/cycle.ts b/src/core/cycle.ts index fdd914cb2..843b57f6b 100644 --- a/src/core/cycle.ts +++ b/src/core/cycle.ts @@ -140,6 +140,14 @@ export interface CycleOpts { * + refreshes the cycle-lock-table TTL. */ yieldBetweenPhases?: () => Promise; + /** + * AbortSignal from the Minions worker. When aborted (timeout, cancel, + * lock-loss), runCycle bails between phases and returns a 'failed' report + * instead of running the next phase. Without this, a timed-out + * autopilot-cycle handler ignores the abort and runs until the worker + * wedges (the 98-waiting-0-active incident on 2026-04-24). + */ + signal?: AbortSignal; } // ─── Lock primitives ─────────────────────────────────────────────── @@ -344,6 +352,20 @@ async function safeYield(hook?: () => Promise) { } } +/** + * Check if the abort signal has fired. Called between phases so that a + * timed-out Minions job bails promptly instead of grinding through all + * remaining phases while the worker thinks it's still at capacity. + */ +function checkAborted(signal?: AbortSignal): void { + if (signal?.aborted) { + const reason = signal.reason instanceof Error + ? signal.reason.message + : String(signal.reason || 'aborted'); + throw new Error(`[cycle] aborted between phases: ${reason}`); + } +} + // ─── Phase runners ───────────────────────────────────────────────── async function runPhaseLint(brainDir: string, dryRun: boolean): Promise { @@ -416,19 +438,29 @@ async function runPhaseBacklinks(brainDir: string, dryRun: boolean): Promise { + willRunExtractPhase: boolean, +): Promise { try { const { performSync } = await import('../commands/sync.ts'); const result = await performSync(engine, { repoPath: brainDir, dryRun, noPull: !pull, - noEmbed: true, // embed is a separate phase + noEmbed: true, // embed is a separate phase + noExtract: willRunExtractPhase, // dedupe ONLY when cycle's extract phase will also run. + // If extract isn't scheduled (e.g. `gbrain dream --phase sync`), + // sync's inline extract still runs to preserve prior behavior. }); const syncedCount = result.added + result.modified; return { @@ -448,6 +480,7 @@ async function runPhaseSync( syncStatus: result.status, dryRun, }, + pagesAffected: result.pagesAffected, }; } catch (e) { return { @@ -465,6 +498,7 @@ async function runPhaseExtract( engine: BrainEngine, brainDir: string, dryRun: boolean, + changedSlugs?: string[], ): Promise { try { const { runExtractCore } = await import('../commands/extract.ts'); @@ -480,15 +514,29 @@ async function runPhaseExtract( details: { dryRun: true, reason: 'no_dry_run_support' }, }; } - const result = await runExtractCore(engine, { mode: 'all', dir: brainDir }); + // Incremental path: if sync told us which slugs changed, only extract those. + // On a 54K-page brain this turns a 10-minute full walk into a sub-second pass. + const result = await runExtractCore(engine, { + mode: 'all', + dir: brainDir, + slugs: changedSlugs, // undefined = full walk (first run / manual) + }); const linksCreated = result?.links_created ?? 0; const timelineCreated = result?.timeline_entries_created ?? 0; + const incremental = changedSlugs !== undefined; return { phase: 'extract', status: 'ok', duration_ms: 0, - summary: `${linksCreated} link(s), ${timelineCreated} timeline entries`, - details: { linksCreated, timelineCreated, pages_processed: result?.pages_processed ?? 0 }, + summary: incremental + ? `${linksCreated} link(s), ${timelineCreated} timeline entries (incremental: ${changedSlugs.length} slugs)` + : `${linksCreated} link(s), ${timelineCreated} timeline entries`, + details: { + linksCreated, timelineCreated, + pages_processed: result?.pages_processed ?? 0, + incremental, + ...(incremental ? { slugs_targeted: changedSlugs.length } : {}), + }, }; } catch (e) { return { @@ -644,6 +692,7 @@ export async function runCycle( try { // ── Phase 1: lint ──────────────────────────────────────────── if (phases.includes('lint')) { + checkAborted(opts.signal); progress.start('cycle.lint'); const { result, duration_ms } = await timePhase(() => runPhaseLint(opts.brainDir, dryRun)); result.duration_ms = duration_ms; @@ -654,6 +703,7 @@ export async function runCycle( // ── Phase 2: backlinks ────────────────────────────────────── if (phases.includes('backlinks')) { + checkAborted(opts.signal); progress.start('cycle.backlinks'); const { result, duration_ms } = await timePhase(() => runPhaseBacklinks(opts.brainDir, dryRun)); result.duration_ms = duration_ms; @@ -663,7 +713,10 @@ export async function runCycle( } // ── Phase 3: sync ─────────────────────────────────────────── + // Track which slugs sync touched so extract can run incrementally. + let syncPagesAffected: string[] | undefined; if (phases.includes('sync')) { + checkAborted(opts.signal); if (!engine) { phaseResults.push({ phase: 'sync', @@ -674,8 +727,10 @@ export async function runCycle( }); } else { progress.start('cycle.sync'); - const { result, duration_ms } = await timePhase(() => runPhaseSync(engine, opts.brainDir, dryRun, pull)); + const { result, duration_ms } = await timePhase(() => runPhaseSync(engine, opts.brainDir, dryRun, pull, phases.includes('extract'))); result.duration_ms = duration_ms; + // Capture changed slugs for incremental extract. + syncPagesAffected = (result as SyncPhaseResult).pagesAffected; phaseResults.push(result); progress.finish(); } @@ -684,6 +739,7 @@ export async function runCycle( // ── Phase 4: extract ──────────────────────────────────────── if (phases.includes('extract')) { + checkAborted(opts.signal); if (!engine) { phaseResults.push({ phase: 'extract', @@ -693,8 +749,11 @@ export async function runCycle( details: { reason: 'no_database' }, }); } else { + // Pass changed slugs from sync for incremental extract. + // If sync didn't run (phases exclude it) or failed, syncPagesAffected + // is undefined → extract falls back to full walk (safe default). progress.start('cycle.extract'); - const { result, duration_ms } = await timePhase(() => runPhaseExtract(engine, opts.brainDir, dryRun)); + const { result, duration_ms } = await timePhase(() => runPhaseExtract(engine, opts.brainDir, dryRun, syncPagesAffected)); result.duration_ms = duration_ms; phaseResults.push(result); progress.finish(); @@ -704,6 +763,7 @@ export async function runCycle( // ── Phase 5: embed ────────────────────────────────────────── if (phases.includes('embed')) { + checkAborted(opts.signal); if (!engine) { phaseResults.push({ phase: 'embed', @@ -724,6 +784,7 @@ export async function runCycle( // ── Phase 6: orphans ──────────────────────────────────────── if (phases.includes('orphans')) { + checkAborted(opts.signal); if (!engine) { phaseResults.push({ phase: 'orphans', diff --git a/src/core/db.ts b/src/core/db.ts index ede0ea21f..6093d3878 100644 --- a/src/core/db.ts +++ b/src/core/db.ts @@ -72,26 +72,78 @@ export function resolvePoolSize(explicit?: number): number { } /** - * Apply session-level defaults to a fresh connection. Called from both - * the module-level `connect()` singleton and the PostgresEngine - * instance-level pool so the idle-in-transaction session timeout is set - * uniformly. + * Session-level GUCs applied to every new backend connection. Prevents + * orphan pgbouncer sessions from holding locks or running queries + * indefinitely when the postgres.js client disconnects mid-transaction + * (typical cause: autopilot SIGKILL'd by launchd, worker crash-loop, + * or transient network drop). * - * `idle_in_transaction_session_timeout = 5 min` was the v0.18.0 field - * report's headline production issue: a 24-hour idle connection was - * holding a lock on `pages` and blocking all DDL. 5 minutes is generous - * for any legitimate transaction but catches crashed writers. The GUC - * is session-scoped (safe for shared pools — no cross-statement leak). + * Observed failure mode these prevent: a single autopilot UPDATE on + * `minion_jobs.lock_until` left a pooler backend in `state='active'` + * / `wait_event='ClientRead'` for 24h+, holding a RowExclusiveLock + * that blocked every subsequent `ALTER TABLE minion_jobs ...`. * - * Wrapped in try/catch because some managed Postgres tenants restrict - * SET on the GUC; non-fatal if it fails. + * Defaults are conservative (chosen not to interfere with bulk work + * like long-running embed passes or CREATE INDEX on large tables): + * - statement_timeout = '5min' + * - idle_in_transaction_session_timeout = '5min' (matches v0.18.0 + * posture; #363's original 2min default was tightened to 5min on + * merge with v0.21.0's setSessionDefaults to avoid regressing + * long-running embed passes) + * + * Override per-GUC with env vars: + * - GBRAIN_STATEMENT_TIMEOUT + * - GBRAIN_IDLE_TX_TIMEOUT + * - GBRAIN_CLIENT_CHECK_INTERVAL (Postgres 14+; empty default - opt-in + * only since older self-hosted Postgres rejects this startup param) + * + * Set any env var to '0' or 'off' to disable that GUC entirely. + * + * Delivered via postgres.js's `connection` option, which sends these as + * startup parameters in the initial connection packet. Works correctly + * with PgBouncer session mode AND transaction mode: startup parameters + * pass through to the backend on connection creation and persist for the + * backend's lifetime (unlike `SET` commands which transaction-mode + * PgBouncer strips between transactions). + * + * Supersedes the v0.21.0 `setSessionDefaults(sql)` helper, which used + * a post-pool `SET` command. That approach is unreliable in PgBouncer + * transaction mode (transaction-mode poolers strip session-state SETs + * between transactions); startup parameters are durable. */ -export async function setSessionDefaults(sql: ReturnType): Promise { - try { - await sql`SET idle_in_transaction_session_timeout = '300000'`; - } catch { - // Non-fatal: some managed Postgres may restrict this GUC - } +const DEFAULT_STATEMENT_TIMEOUT = '5min'; +const DEFAULT_IDLE_TX_TIMEOUT = '5min'; + +export function resolveSessionTimeouts(): Record { + const out: Record = {}; + const add = (envKey: string, gucKey: string, defaultVal: string) => { + const raw = process.env[envKey]; + if (raw === '0' || raw === 'off') return; // explicitly disabled + const val = raw ?? defaultVal; + if (val) out[gucKey] = val; + }; + add('GBRAIN_STATEMENT_TIMEOUT', 'statement_timeout', DEFAULT_STATEMENT_TIMEOUT); + add('GBRAIN_IDLE_TX_TIMEOUT', 'idle_in_transaction_session_timeout', DEFAULT_IDLE_TX_TIMEOUT); + // client_connection_check_interval is opt-in: Postgres 14+ only, and some + // managed pooler tiers reject unknown startup parameters. Users can enable + // it explicitly once they know their Postgres version supports it. + add('GBRAIN_CLIENT_CHECK_INTERVAL', 'client_connection_check_interval', ''); + return out; +} + +/** + * Backward-compat shim for v0.21.0's `setSessionDefaults` callers. + * The current implementation no-ops because session timeouts are now + * applied at connection-startup time via `resolveSessionTimeouts()` + + * postgres.js's `connection` option (more durable across PgBouncer + * transaction mode). + * + * Kept as a callable function so existing call sites in `connect()` and + * `PostgresEngine.connect()` don't need to be touched on the merge — + * the work has already happened by the time this function would run. + */ +export async function setSessionDefaults(_sql: ReturnType): Promise { + // No-op: timeouts are now applied as startup parameters in resolveSessionTimeouts(). } export function getConnection(): ReturnType { @@ -125,6 +177,7 @@ export async function connect(config: EngineConfig): Promise { try { const prepare = resolvePrepare(url); + const timeouts = resolveSessionTimeouts(); const opts: Record = { max: resolvePoolSize(), idle_timeout: 20, @@ -134,6 +187,9 @@ export async function connect(config: EngineConfig): Promise { bigint: postgres.BigInt, }, }; + if (Object.keys(timeouts).length > 0) { + opts.connection = timeouts; + } if (typeof prepare === 'boolean') { opts.prepare = prepare; if (!prepare) { diff --git a/src/core/engine.ts b/src/core/engine.ts index 8e76f1f37..00c5f300b 100644 --- a/src/core/engine.ts +++ b/src/core/engine.ts @@ -1,6 +1,6 @@ import type { Page, PageInput, PageFilters, - Chunk, ChunkInput, + Chunk, ChunkInput, StaleChunkRow, SearchResult, SearchOpts, Link, GraphNode, GraphPath, TimelineEntry, TimelineInput, TimelineOpts, @@ -133,6 +133,21 @@ export interface BrainEngine { // Chunks upsertChunks(slug: string, chunks: ChunkInput[]): Promise; getChunks(slug: string): Promise; + /** + * Count chunks across the entire brain where embedded_at IS NULL. + * Pre-flight short-circuit for `embed --stale` so a 100%-embedded brain + * does no further work after a single SELECT count(*) (~50 bytes wire). + */ + countStaleChunks(): Promise; + /** + * Return every chunk where embedded_at IS NULL, with the metadata needed + * to call embedBatch + upsertChunks. The `embedding` column is omitted + * by design — stale rows have NULL embeddings, so shipping them wastes + * wire bytes for no gain. Caller groups by slug, embeds, and re-upserts. + * + * Bounded by an internal LIMIT of 100000 to mirror listPages. + */ + listStaleChunks(): Promise; deleteChunks(slug: string): Promise; // Links diff --git a/src/core/minions/supervisor.ts b/src/core/minions/supervisor.ts index 4da80b660..8de710d25 100644 --- a/src/core/minions/supervisor.ts +++ b/src/core/minions/supervisor.ts @@ -142,6 +142,7 @@ export class MinionSupervisor { private sigtermListener: (() => void) | null = null; private sigintListener: (() => void) | null = null; private lockAcquired = false; + private consecutiveHealthFailures = 0; constructor(engine: BrainEngine, opts: Partial & { cliPath: string }) { this.engine = engine; @@ -476,10 +477,26 @@ export class MinionSupervisor { } const exitReason = signal ? `signal ${signal}` : `code ${code ?? 'null'}`; + + // Classify the likely cause for easier debugging + let likelyCause: string; + if (signal === 'SIGKILL') { + likelyCause = 'oom_or_external_kill'; + } else if (signal === 'SIGTERM') { + likelyCause = 'graceful_shutdown'; + } else if (code === 1) { + likelyCause = 'runtime_error'; + } else if (code === 0) { + likelyCause = 'clean_exit'; + } else { + likelyCause = 'unknown'; + } + this.emit('worker_exited', { code: code ?? null, signal: signal ?? null, reason: exitReason, + likely_cause: likelyCause, crash_count: this.crashCount, max_crashes: this.opts.maxCrashes, run_duration_ms: runDuration, @@ -523,6 +540,9 @@ export class MinionSupervisor { [this.opts.queue], ); + // Reset consecutive failure counter on successful health check + this.consecutiveHealthFailures = 0; + 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); @@ -561,11 +581,41 @@ export class MinionSupervisor { }); } } catch (e) { - // Health check failures are non-fatal. - this.emit('health_error', { - error: e instanceof Error ? e.message : String(e), - queue: this.opts.queue, - }); + this.consecutiveHealthFailures++; + const errMsg = e instanceof Error ? e.message : String(e); + + if (this.consecutiveHealthFailures >= 3) { + // DB connection is likely dead. Emit a degraded warning. + this.emit('health_warn', { + reason: 'db_connection_degraded', + consecutive_failures: this.consecutiveHealthFailures, + error: errMsg, + queue: this.opts.queue, + }); + // Attempt to reconnect the engine if it supports it + try { + if ('reconnect' in this.engine && typeof (this.engine as Record).reconnect === 'function') { + await (this.engine as unknown as { reconnect(): Promise }).reconnect(); + this.consecutiveHealthFailures = 0; + this.emit('health_warn', { + reason: 'db_reconnected', + queue: this.opts.queue, + }); + } + } catch (reconnErr) { + this.emit('health_error', { + error: `reconnect failed: ${reconnErr instanceof Error ? reconnErr.message : String(reconnErr)}`, + reconnect_failed: true, + queue: this.opts.queue, + }); + } + } else { + // Non-fatal single failure + this.emit('health_error', { + error: errMsg, + queue: this.opts.queue, + }); + } } finally { this.healthInFlight = false; } diff --git a/src/core/minions/worker.ts b/src/core/minions/worker.ts index e6f8a9342..6db549282 100644 --- a/src/core/minions/worker.ts +++ b/src/core/minions/worker.ts @@ -277,12 +277,30 @@ export class MinionWorker { // The .finally clearTimeout below ensures process exit isn't delayed by a // dangling timer on normal completion. let timeoutTimer: ReturnType | null = null; + let graceTimer: ReturnType | null = null; if (job.timeout_ms != null) { timeoutTimer = setTimeout(() => { if (!abort.signal.aborted) { console.warn(`Job ${job.id} (${job.name}) hit per-job timeout (${job.timeout_ms}ms), aborting`); abort.abort(new Error('timeout')); } + // Safety net: if the handler doesn't resolve within 30s after abort, + // force-evict from inFlight so the worker can pick up new jobs. + // Without this, a handler that ignores AbortSignal wedges the worker + // forever (the 98-waiting-0-active incident on 2026-04-24). + graceTimer = setTimeout(() => { + if (this.inFlight.has(job.id)) { + console.warn( + `Job ${job.id} (${job.name}) did not exit within 30s of abort. ` + + `Force-evicting from inFlight to unblock worker. ` + + `The handler is still running but the worker will claim new jobs.` + ); + clearInterval(lockTimer); + this.inFlight.delete(job.id); + // Best-effort: mark as dead in DB so it doesn't get reclaimed + this.queue.failJob(job.id, lockToken, 'handler ignored abort signal (force-evicted)', 'dead').catch(() => {}); + } + }, 30_000); }, job.timeout_ms); } @@ -290,6 +308,7 @@ export class MinionWorker { .finally(() => { clearInterval(lockTimer); if (timeoutTimer) clearTimeout(timeoutTimer); + if (graceTimer) clearTimeout(graceTimer); this.inFlight.delete(job.id); }); diff --git a/src/core/pglite-engine.ts b/src/core/pglite-engine.ts index c1e6cb589..ddf67d793 100644 --- a/src/core/pglite-engine.ts +++ b/src/core/pglite-engine.ts @@ -9,7 +9,7 @@ import { PGLITE_SCHEMA_SQL } from './pglite-schema.ts'; import { acquireLock, releaseLock, type LockHandle } from './pglite-lock.ts'; import type { Page, PageInput, PageFilters, PageType, - Chunk, ChunkInput, + Chunk, ChunkInput, StaleChunkRow, SearchResult, SearchOpts, Link, GraphNode, GraphPath, TimelineEntry, TimelineInput, TimelineOpts, @@ -491,6 +491,9 @@ export class PGLiteEngine implements BrainEngine { } } + // CONSISTENCY: when chunk_text changes and no new embedding is supplied, BOTH embedding AND + // embedded_at must reset to NULL so `embed --stale` correctly picks up the row for re-embedding. + // See postgres-engine.ts upsertChunks for the full rationale — pglite mirrors it for parity. await this.db.query( `INSERT INTO content_chunks ${cols} VALUES ${rowParts.join(', ')} ON CONFLICT (page_id, chunk_index) DO UPDATE SET @@ -499,7 +502,10 @@ export class PGLiteEngine implements BrainEngine { embedding = CASE WHEN EXCLUDED.chunk_text != content_chunks.chunk_text THEN EXCLUDED.embedding ELSE COALESCE(EXCLUDED.embedding, content_chunks.embedding) END, model = COALESCE(EXCLUDED.model, content_chunks.model), token_count = EXCLUDED.token_count, - embedded_at = COALESCE(EXCLUDED.embedded_at, content_chunks.embedded_at), + embedded_at = CASE + WHEN EXCLUDED.chunk_text != content_chunks.chunk_text AND EXCLUDED.embedding IS NULL THEN NULL + ELSE COALESCE(EXCLUDED.embedded_at, content_chunks.embedded_at) + END, language = EXCLUDED.language, symbol_name = EXCLUDED.symbol_name, symbol_type = EXCLUDED.symbol_type, @@ -523,6 +529,29 @@ export class PGLiteEngine implements BrainEngine { return (rows as Record[]).map(r => rowToChunk(r)); } + async countStaleChunks(): Promise { + const { rows } = await this.db.query( + `SELECT count(*)::int AS count + FROM content_chunks + WHERE embedding IS NULL`, + ); + const count = (rows[0] as { count: number } | undefined)?.count ?? 0; + return Number(count); + } + + async listStaleChunks(): Promise { + const { rows } = await this.db.query( + `SELECT p.slug, cc.chunk_index, cc.chunk_text, cc.chunk_source, + cc.model, cc.token_count + FROM content_chunks cc + JOIN pages p ON p.id = cc.page_id + WHERE cc.embedding IS NULL + ORDER BY p.id, cc.chunk_index + LIMIT 100000`, + ); + return rows as unknown as StaleChunkRow[]; + } + async deleteChunks(slug: string): Promise { await this.db.query( `DELETE FROM content_chunks diff --git a/src/core/postgres-engine.ts b/src/core/postgres-engine.ts index 1cc38e59a..865e9a63e 100644 --- a/src/core/postgres-engine.ts +++ b/src/core/postgres-engine.ts @@ -5,7 +5,7 @@ import { runMigrations } from './migrate.ts'; import { SCHEMA_SQL } from './schema-embedded.ts'; import type { Page, PageInput, PageFilters, PageType, - Chunk, ChunkInput, + Chunk, ChunkInput, StaleChunkRow, SearchResult, SearchOpts, Link, GraphNode, GraphPath, TimelineEntry, TimelineInput, TimelineOpts, @@ -21,9 +21,23 @@ import { validateSlug, contentHash, rowToPage, rowToChunk, rowToSearchResult, pa import { resolveBoostMap, resolveHardExcludes } from './search/source-boost.ts'; import { buildSourceFactorCase, buildHardExcludeClause } from './search/sql-ranking.ts'; +// CONNECTION_ERROR_PATTERNS / isConnectionError were used by the per-call +// executeRaw retry that #406 originally shipped. Eng-review D3 dropped that +// retry as unsound (regex idempotence-boundary doesn't hold for writable +// CTEs or side-effecting SELECTs). Recovery now happens at the supervisor +// level (3-strikes-then-reconnect). The unit tests in +// test/connection-resilience.test.ts retain a self-contained copy of the +// helper so the regression-against-future-reintroduction guard still works. +// See TODOS.md item: "err.code-based connection-error matching" for the +// follow-up that will reintroduce a typed retry mechanism. + export class PostgresEngine implements BrainEngine { readonly kind = 'postgres' as const; private _sql: ReturnType | null = null; + /** Saved config for reconnection. */ + private _savedConfig: (EngineConfig & { poolSize?: number }) | null = null; + /** Whether a reconnect is in progress (prevents concurrent reconnects). */ + private _reconnecting = false; // Instance connection (for workers) or fall back to module global (backward compat) get sql(): ReturnType { @@ -33,6 +47,7 @@ export class PostgresEngine implements BrainEngine { // Lifecycle async connect(config: EngineConfig & { poolSize?: number }): Promise { + this._savedConfig = config; if (config.poolSize) { // Instance-level connection for worker isolation. resolvePoolSize lets // GBRAIN_POOL_SIZE cap below the caller's requested size when set — the @@ -45,12 +60,20 @@ export class PostgresEngine implements BrainEngine { // "prepared statement does not exist" under load just like the module // singleton did before v0.15.4. const prepare = db.resolvePrepare(url); + // Session timeouts (statement_timeout + idle_in_transaction_session_timeout) + // keep orphan pgbouncer backends from holding locks for hours when the + // postgres.js client disconnects mid-transaction. See resolveSessionTimeouts + // in db.ts for context + env var overrides. + const timeouts = db.resolveSessionTimeouts(); const opts: Record = { max: size, idle_timeout: 20, connect_timeout: 10, types: { bigint: postgres.BigInt }, }; + if (Object.keys(timeouts).length > 0) { + opts.connection = timeouts; + } if (typeof prepare === 'boolean') { opts.prepare = prepare; } @@ -572,7 +595,13 @@ export class PostgresEngine implements BrainEngine { } } - // Single statement upsert: preserves existing embeddings via COALESCE when new value is NULL + // Single statement upsert: preserves existing embeddings via COALESCE when new value is NULL. + // CONSISTENCY: when chunk_text changes and no new embedding is supplied, BOTH embedding AND + // embedded_at must reset to NULL so `embed --stale` correctly picks up the row for re-embedding. + // Without this, embedded_at lies (says "embedded" while embedding=NULL), and any staleness + // predicate on embedded_at would silently skip the row. This is why the egress fix predicates + // on `embedding IS NULL` rather than `embedded_at IS NULL` — and it's why we now keep both + // columns honest at write time. await sql.unsafe( `INSERT INTO content_chunks ${cols} VALUES ${rows.join(', ')} ON CONFLICT (page_id, chunk_index) DO UPDATE SET @@ -581,7 +610,10 @@ export class PostgresEngine implements BrainEngine { embedding = CASE WHEN EXCLUDED.chunk_text != content_chunks.chunk_text THEN EXCLUDED.embedding ELSE COALESCE(EXCLUDED.embedding, content_chunks.embedding) END, model = COALESCE(EXCLUDED.model, content_chunks.model), token_count = EXCLUDED.token_count, - embedded_at = COALESCE(EXCLUDED.embedded_at, content_chunks.embedded_at), + embedded_at = CASE + WHEN EXCLUDED.chunk_text != content_chunks.chunk_text AND EXCLUDED.embedding IS NULL THEN NULL + ELSE COALESCE(EXCLUDED.embedded_at, content_chunks.embedded_at) + END, language = EXCLUDED.language, symbol_name = EXCLUDED.symbol_name, symbol_type = EXCLUDED.symbol_type, @@ -605,6 +637,30 @@ export class PostgresEngine implements BrainEngine { return rows.map((r) => rowToChunk(r as Record)); } + async countStaleChunks(): Promise { + const sql = this.sql; + const [row] = await sql` + SELECT count(*)::int AS count + FROM content_chunks + WHERE embedding IS NULL + `; + return Number((row as { count?: number } | undefined)?.count ?? 0); + } + + async listStaleChunks(): Promise { + const sql = this.sql; + const rows = await sql` + SELECT p.slug, cc.chunk_index, cc.chunk_text, cc.chunk_source, + cc.model, cc.token_count + FROM content_chunks cc + JOIN pages p ON p.id = cc.page_id + WHERE cc.embedding IS NULL + ORDER BY p.id, cc.chunk_index + LIMIT 100000 + `; + return rows as unknown as StaleChunkRow[]; + } + async deleteChunks(slug: string): Promise { const sql = this.sql; await sql` @@ -1328,9 +1384,34 @@ export class PostgresEngine implements BrainEngine { return rows.map((r) => rowToChunk(r as Record, true)); } + /** + * Reconnect the engine by tearing down the current pool and creating a fresh one. + * No-ops if no saved config (module-singleton mode) or if already reconnecting. + */ + async reconnect(): Promise { + if (!this._savedConfig || this._reconnecting) return; + this._reconnecting = true; + try { + // Tear down old pool (best-effort — it may already be dead) + try { await this.disconnect(); } catch { /* swallow */ } + // Create fresh pool + await this.connect(this._savedConfig); + } finally { + this._reconnecting = false; + } + } + async executeRaw>(sql: string, params?: unknown[]): Promise { const conn = this.sql; return conn.unsafe(sql, params as Parameters[1]) as unknown as T[]; + // Pre-#406 behavior: throw on any error including connection death. + // Per-call auto-retry is not safe here because executeRaw is also used + // for non-transactional mutations (DELETE/UPDATE/INSERT in sources.ts, + // ALTER TABLE in migrations) where retrying after a connection-mid-statement + // death can phantom-write a row that already committed on the server. + // Recovery instead happens at the supervisor level: the watchdog detects + // 3 consecutive health-check failures and calls engine.reconnect() to + // swap in a fresh pool. See db.ts setSessionDefaults / supervisor.ts. } // ============================================================ diff --git a/src/core/types.ts b/src/core/types.ts index 6422d28d7..fda549558 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -70,6 +70,21 @@ export interface Chunk { symbol_name_qualified?: string | null; } +/** + * Lightweight row shape returned by `BrainEngine.listStaleChunks()`. + * Excludes the `embedding` column on purpose — only chunks needing + * an embedding come back, and we don't ship the (always-null on stale + * rows) embedding bytes over the wire. See `embed --stale` egress fix. + */ +export interface StaleChunkRow { + slug: string; + chunk_index: number; + chunk_text: string; + chunk_source: 'compiled_truth' | 'timeline'; + model: string | null; + token_count: number | null; +} + export interface ChunkInput { chunk_index: number; chunk_text: string; diff --git a/test/connection-resilience.test.ts b/test/connection-resilience.test.ts new file mode 100644 index 000000000..36f8b93f0 --- /dev/null +++ b/test/connection-resilience.test.ts @@ -0,0 +1,314 @@ +import { describe, it, expect } from 'bun:test'; + +/** + * Tests for connection resilience features: + * 1. PostgresEngine.executeRaw retries on connection errors + * 2. PostgresEngine.reconnect creates fresh connection pool + * 3. Supervisor health check tracks consecutive failures + * 4. Supervisor classifies worker exit reasons + */ + +// --- Unit tests for isConnectionError (extracted pattern) --- + +const CONNECTION_ERROR_PATTERNS = [ + 'ECONNREFUSED', + 'ECONNRESET', + 'EPIPE', + 'connection terminated', + 'Client has encountered a connection error', + 'password authentication failed', + 'Connection terminated unexpectedly', + 'no pg_hba.conf entry', + 'server closed the connection unexpectedly', + 'SSL connection has been closed unexpectedly', + 'connection is insecure', + 'too many connections', + 'remaining connection slots are reserved', +]; + +function isConnectionError(err: unknown): boolean { + if (!err) return false; + const msg = err instanceof Error ? err.message : String(err); + const code = (err as NodeJS.ErrnoException)?.code; + if (code && CONNECTION_ERROR_PATTERNS.includes(code)) return true; + return CONNECTION_ERROR_PATTERNS.some(p => msg.includes(p)); +} + +describe('isConnectionError', () => { + it('detects password authentication failure', () => { + expect(isConnectionError(new Error('password authentication failed for user "postgres"'))).toBe(true); + }); + + it('detects ECONNREFUSED via error code', () => { + const err = new Error('connect ECONNREFUSED 127.0.0.1:5432') as NodeJS.ErrnoException; + err.code = 'ECONNREFUSED'; + expect(isConnectionError(err)).toBe(true); + }); + + it('detects ECONNRESET via error code', () => { + const err = new Error('read ECONNRESET') as NodeJS.ErrnoException; + err.code = 'ECONNRESET'; + expect(isConnectionError(err)).toBe(true); + }); + + it('detects connection terminated message', () => { + expect(isConnectionError(new Error('connection terminated'))).toBe(true); + }); + + it('detects Connection terminated unexpectedly', () => { + expect(isConnectionError(new Error('Connection terminated unexpectedly'))).toBe(true); + }); + + it('detects server closed the connection', () => { + expect(isConnectionError(new Error('server closed the connection unexpectedly'))).toBe(true); + }); + + it('detects SSL connection closed', () => { + expect(isConnectionError(new Error('SSL connection has been closed unexpectedly'))).toBe(true); + }); + + it('detects too many connections', () => { + expect(isConnectionError(new Error('FATAL: too many connections for role "postgres"'))).toBe(true); + }); + + it('does not match regular query errors', () => { + expect(isConnectionError(new Error('relation "foo" does not exist'))).toBe(false); + }); + + it('does not match null/undefined', () => { + expect(isConnectionError(null)).toBe(false); + expect(isConnectionError(undefined)).toBe(false); + }); + + it('does not match syntax errors', () => { + expect(isConnectionError(new Error('syntax error at or near "SELECT"'))).toBe(false); + }); + + it('does not match constraint violations', () => { + expect(isConnectionError(new Error('duplicate key value violates unique constraint'))).toBe(false); + }); +}); + +// --- Unit tests for worker exit classification --- + +function classifyWorkerExit(code: number | null, signal: string | null): string { + if (signal === 'SIGKILL') return 'oom_or_external_kill'; + if (signal === 'SIGTERM') return 'graceful_shutdown'; + if (code === 1) return 'runtime_error'; + if (code === 0) return 'clean_exit'; + return 'unknown'; +} + +describe('classifyWorkerExit', () => { + it('classifies SIGKILL as OOM/external kill', () => { + expect(classifyWorkerExit(null, 'SIGKILL')).toBe('oom_or_external_kill'); + }); + + it('classifies SIGTERM as graceful shutdown', () => { + expect(classifyWorkerExit(null, 'SIGTERM')).toBe('graceful_shutdown'); + }); + + it('classifies exit code 1 as runtime error', () => { + expect(classifyWorkerExit(1, null)).toBe('runtime_error'); + }); + + it('classifies exit code 0 as clean exit', () => { + expect(classifyWorkerExit(0, null)).toBe('clean_exit'); + }); + + it('classifies unknown codes as unknown', () => { + expect(classifyWorkerExit(137, null)).toBe('unknown'); + expect(classifyWorkerExit(null, null)).toBe('unknown'); + }); + + // Signal takes precedence over code + it('SIGKILL takes precedence over any exit code', () => { + expect(classifyWorkerExit(1, 'SIGKILL')).toBe('oom_or_external_kill'); + }); +}); + +// --- Mock-based tests for reconnect logic --- + +describe('PostgresEngine reconnect behavior', () => { + it('reconnect flag prevents concurrent reconnections', async () => { + // Simulate the _reconnecting guard + let reconnecting = false; + let reconnectCount = 0; + + async function reconnect() { + if (reconnecting) return; + reconnecting = true; + try { + reconnectCount++; + await new Promise(r => setTimeout(r, 10)); + } finally { + reconnecting = false; + } + } + + // Fire 3 concurrent reconnects — only 1 should run + await Promise.all([reconnect(), reconnect(), reconnect()]); + expect(reconnectCount).toBe(1); + }); + + it('executeRaw retry does not infinite-loop on persistent connection failure', async () => { + // Simulate: first call fails (connection error), reconnect succeeds, + // but retry also fails with a NON-connection error + let callCount = 0; + + async function executeRawWithRetry(): Promise { + callCount++; + if (callCount === 1) { + throw new Error('connection terminated'); // connection error → triggers retry + } + if (callCount === 2) { + throw new Error('relation "foo" does not exist'); // NOT a connection error → throw + } + return [{ ok: true }]; + } + + try { + await (async () => { + try { + return await executeRawWithRetry(); + } catch (err) { + if (isConnectionError(err)) { + // "reconnect" would happen here + return await executeRawWithRetry(); + } + throw err; + } + })(); + } catch (err) { + expect((err as Error).message).toBe('relation "foo" does not exist'); + } + + expect(callCount).toBe(2); // Only 2 attempts, no infinite loop + }); + + it('executeRaw succeeds on retry after connection error', async () => { + let callCount = 0; + + async function executeRawWithRetry(): Promise { + callCount++; + if (callCount === 1) { + throw new Error('password authentication failed for user "postgres"'); + } + return [{ ok: true }]; + } + + const result = await (async () => { + try { + return await executeRawWithRetry(); + } catch (err) { + if (isConnectionError(err)) { + // reconnect would happen here + return await executeRawWithRetry(); + } + throw err; + } + })(); + + expect(result).toEqual([{ ok: true }]); + expect(callCount).toBe(2); + }); +}); + +// --- Supervisor health check failure tracking --- + +describe('Supervisor health check failure tracking', () => { + it('emits db_connection_degraded after 3 consecutive failures', () => { + let consecutiveFailures = 0; + const emitted: Array<{ event: string; reason?: string }> = []; + + function emit(event: string, fields: Record = {}) { + emitted.push({ event, ...fields } as { event: string; reason?: string }); + } + + // Simulate 3 health check failures + for (let i = 0; i < 4; i++) { + consecutiveFailures++; + if (consecutiveFailures >= 3) { + emit('health_warn', { reason: 'db_connection_degraded', consecutive_failures: consecutiveFailures }); + } else { + emit('health_error', { error: 'connection terminated' }); + } + } + + const degradedWarnings = emitted.filter(e => e.reason === 'db_connection_degraded'); + expect(degradedWarnings.length).toBe(2); // fires at count 3 and 4 + + // First two were regular health_error + expect(emitted[0].event).toBe('health_error'); + expect(emitted[1].event).toBe('health_error'); + // Third triggers the degraded warning + expect(emitted[2].reason).toBe('db_connection_degraded'); + }); + + it('resets failure counter on successful health check', () => { + let consecutiveFailures = 0; + + // 2 failures + consecutiveFailures++; + consecutiveFailures++; + expect(consecutiveFailures).toBe(2); + + // Success resets + consecutiveFailures = 0; + expect(consecutiveFailures).toBe(0); + + // 1 more failure — should not trigger degraded (need 3 consecutive) + consecutiveFailures++; + expect(consecutiveFailures).toBeLessThan(3); + }); +}); + +// ───────────────────────────────────────────────────────────────── +// Eng-review D3 regression guards — executeRaw retry wrapper dropped +// ───────────────────────────────────────────────────────────────── +// +// The original #406 wrapped PostgresEngine.executeRaw in a per-call +// try/catch that retried on connection errors. Eng-review D3 dropped +// that wrapper as unsound (regex idempotence boundary doesn't hold +// for writable CTEs or side-effecting SELECTs). Recovery now happens +// at the supervisor level via the 3-strikes-then-reconnect path. +// +// These guards prevent reintroduction of the per-call retry without +// a typed-idempotency boundary. + +import { readFileSync } from 'fs'; +import { resolve } from 'path'; + +describe('Eng-review D3 — executeRaw has no per-call retry wrapper', () => { + it('PostgresEngine.executeRaw is a single-statement passthrough (no try/catch on connection errors)', () => { + const src = readFileSync(resolve('src/core/postgres-engine.ts'), 'utf-8'); + + // Find the executeRaw method in the class (not the helper inside withReservedConnection) + // Pattern: must be a method on the class taking (sql, params) + const fnMatch = src.match(/async executeRaw>\(sql: string, params\?: unknown\[\]\): Promise \{([\s\S]*?)\n \}/); + expect(fnMatch).not.toBeNull(); + const body = fnMatch![1]; + + // Must not have any try/catch + expect(body).not.toContain('try {'); + expect(body).not.toContain('catch'); + // Must not call reconnect() from this method + expect(body).not.toContain('this.reconnect()'); + // Must call conn.unsafe directly + expect(body).toContain('conn.unsafe('); + }); + + it('PostgresEngine.reconnect() still exists for supervisor-driven recovery', () => { + const src = readFileSync(resolve('src/core/postgres-engine.ts'), 'utf-8'); + expect(src).toContain('async reconnect()'); + expect(src).toContain('await this.disconnect()'); + }); + + it('Supervisor still has the 3-strikes-then-reconnect path', () => { + const src = readFileSync(resolve('src/core/minions/supervisor.ts'), 'utf-8'); + expect(src).toContain('consecutiveHealthFailures'); + // Supervisor invokes reconnect via a typed cast after 3 consecutive failures. + expect(src).toMatch(/reconnect\(\): Promise/); + expect(src).toContain('this.consecutiveHealthFailures >= 3'); + }); +}); diff --git a/test/core/cycle.test.ts b/test/core/cycle.test.ts index 308a0325c..095e26872 100644 --- a/test/core/cycle.test.ts +++ b/test/core/cycle.test.ts @@ -17,8 +17,8 @@ import { existsSync, unlinkSync } from 'fs'; let lintCalls: Array<{ target: string; fix: boolean; dryRun: boolean | undefined }> = []; let backlinksCalls: Array<{ action: string; dir: string; dryRun: boolean | undefined }> = []; -let syncCalls: Array<{ dryRun: boolean | undefined; noPull: boolean | undefined }> = []; -let extractCalls: Array<{ mode: string; dir: string }> = []; +let syncCalls: Array<{ dryRun: boolean | undefined; noPull: boolean | undefined; noExtract: boolean | undefined }> = []; +let extractCalls: Array<{ mode: string; dir: string; slugs: string[] | undefined }> = []; let embedCalls: Array<{ stale: boolean | undefined; dryRun: boolean | undefined }> = []; let orphansCalls: number = 0; @@ -49,7 +49,7 @@ mock.module('../../src/commands/backlinks.ts', () => ({ // Mock sync mock.module('../../src/commands/sync.ts', () => ({ performSync: async (_engine: any, opts: any) => { - syncCalls.push({ dryRun: opts.dryRun, noPull: opts.noPull }); + syncCalls.push({ dryRun: opts.dryRun, noPull: opts.noPull, noExtract: opts.noExtract }); return { status: opts.dryRun ? 'dry_run' : 'synced', fromCommit: 'abcd', @@ -72,8 +72,8 @@ mock.module('../../src/commands/sync.ts', () => ({ // Mock extract mock.module('../../src/commands/extract.ts', () => ({ runExtractCore: async (_engine: any, opts: any) => { - extractCalls.push({ mode: opts.mode, dir: opts.dir }); - return { links_created: 7, timeline_entries_created: 3, pages_processed: 5 }; + extractCalls.push({ mode: opts.mode, dir: opts.dir, slugs: opts.slugs }); + return { links_created: 7, timeline_entries_created: 3, pages_processed: opts.slugs?.length ?? 5 }; }, walkMarkdownFiles: () => [], extractMarkdownLinks: () => [], @@ -392,3 +392,63 @@ describe('runCycle — yieldBetweenPhases hook', () => { expect(report.phases.length).toBe(6); }); }); + +// ───────────────────────────────────────────────────────────────── +// Wave regression guards (#417 + Codex F2) +// ───────────────────────────────────────────────────────────────── + +describe('runCycle — incremental extract slug propagation (#417)', () => { + beforeEach(async () => { + await truncateCycleLocks(sharedEngine); + syncCalls = []; + extractCalls = []; + }); + + test('cycle threads sync.pagesAffected into extract phase as the slugs argument', async () => { + // performSync mock returns pagesAffected = ['a', 'b']. The extract phase + // must receive those exact slugs, not undefined (which would trigger a full walk). + await runCycle(sharedEngine, { brainDir: '/tmp/brain' }); + + // Sync ran once + expect(syncCalls.length).toBe(1); + // Extract ran once with the slugs from sync (not undefined) + expect(extractCalls.length).toBe(1); + expect(extractCalls[0].slugs).toEqual(['a', 'b']); + }); + + test('extract phase falls back to full walk when sync was skipped (slugs undefined)', async () => { + // Run only the extract phase — sync didn't run, so syncPagesAffected + // is undefined and extract should walk the full directory (slugs:undefined). + await runCycle(sharedEngine, { brainDir: '/tmp/brain', phases: ['extract'] }); + + expect(syncCalls.length).toBe(0); + expect(extractCalls.length).toBe(1); + expect(extractCalls[0].slugs).toBeUndefined(); + }); +}); + +describe('runCycle — Codex F2: noExtract is gated on whether extract phase runs', () => { + beforeEach(async () => { + await truncateCycleLocks(sharedEngine); + syncCalls = []; + extractCalls = []; + }); + + test('full cycle (sync + extract): noExtract=true so sync skips inline extraction (extract phase handles it)', async () => { + await runCycle(sharedEngine, { brainDir: '/tmp/brain', phases: ['sync', 'extract'] }); + + expect(syncCalls.length).toBe(1); + expect(syncCalls[0].noExtract).toBe(true); // dedupe enabled + expect(extractCalls.length).toBe(1); // extract phase ran + }); + + test('phases:[sync] only: noExtract=false so sync runs inline extraction (no silent extract drop)', async () => { + await runCycle(sharedEngine, { brainDir: '/tmp/brain', phases: ['sync'] }); + + expect(syncCalls.length).toBe(1); + // Critical: noExtract must be false here. If it were true, the user just lost + // their extraction without any indication. This is the F2 regression guard. + expect(syncCalls[0].noExtract).toBe(false); + expect(extractCalls.length).toBe(0); // extract phase did NOT run + }); +}); diff --git a/test/cycle-abort.test.ts b/test/cycle-abort.test.ts new file mode 100644 index 000000000..f58de5ceb --- /dev/null +++ b/test/cycle-abort.test.ts @@ -0,0 +1,148 @@ +/** + * test/cycle-abort.test.ts — Verify runCycle respects AbortSignal. + * + * Regression test for the 2026-04-24 incident where 98 jobs piled up + * because autopilot-cycle's handler didn't propagate AbortSignal to + * runCycle, and runCycle had no signal-checking between phases. + * + * Tests the three-layer fix: + * 1. CycleOpts.signal — runCycle checks signal between phases + * 2. Handler wiring — autopilot-cycle passes job.signal + * 3. Worker force-eviction — last resort if handler ignores abort + * + * Layer 3 is tested in minions.test.ts (worker-level). This file + * covers layers 1 and 2 via the cycle interface. + */ + +import { describe, test, expect } from 'bun:test'; + +// We can't easily import runCycle with a real engine for unit tests, +// but we CAN test the checkAborted pattern and CycleOpts contract. + +describe('CycleOpts.signal contract (v0.20.5)', () => { + test('signal field exists on CycleOpts interface', async () => { + // Type-level test: importing the type should work + const mod = await import('../src/core/cycle.ts'); + // runCycle exists and is callable + expect(typeof mod.runCycle).toBe('function'); + }); + + test('runCycle accepts signal in opts without error', async () => { + // Verify runCycle doesn't crash when signal is passed but no engine + const { runCycle } = await import('../src/core/cycle.ts'); + const abort = new AbortController(); + + // Call with null engine + minimal opts — should return a report + // (phases that need engine will be skipped) + const report = await runCycle(null, { + brainDir: '/nonexistent-for-test', + phases: [], // empty phases = no work + signal: abort.signal, + }); + + expect(report.schema_version).toBe('1'); + expect(report.status).toBeDefined(); + }); + + test('runCycle bails on pre-aborted signal', async () => { + const { runCycle } = await import('../src/core/cycle.ts'); + const abort = new AbortController(); + abort.abort(new Error('timeout')); + + // With a pre-aborted signal and phases that would run, it should + // throw or return failed (depending on which phase catches it first) + try { + const report = await runCycle(null, { + brainDir: '/nonexistent-for-test', + phases: ['lint'], // lint doesn't need engine, would normally run + signal: abort.signal, + }); + // If it returns instead of throwing, status should reflect the abort + expect(['failed', 'partial']).toContain(report.status); + } catch (err) { + // checkAborted threw — this is the expected behavior + expect(err instanceof Error).toBe(true); + expect((err as Error).message).toContain('aborted'); + } + }); + + test('runCycle bails mid-flight when signal fires between phases', async () => { + const { runCycle } = await import('../src/core/cycle.ts'); + const abort = new AbortController(); + + // Abort after 50ms — should catch between phases + setTimeout(() => abort.abort(new Error('timeout')), 50); + + try { + const report = await runCycle(null, { + brainDir: '/nonexistent-for-test', + phases: ['lint', 'backlinks', 'orphans'], + signal: abort.signal, + yieldBetweenPhases: async () => { + // Slow yield to give the abort time to fire + await new Promise(r => setTimeout(r, 100)); + }, + }); + // If it returned cleanly, not all phases should have run + // (abort should have prevented later phases) + const completedPhases = report.phases.length; + expect(completedPhases).toBeLessThan(3); + } catch (err) { + // checkAborted threw between phases — expected + expect(err instanceof Error).toBe(true); + expect((err as Error).message).toContain('aborted'); + } + }); +}); + +describe('autopilot-cycle handler contract (v0.20.5)', () => { + test('handler registration passes signal to runCycle', async () => { + // Verify the handler code in jobs.ts includes job.signal + const fs = await import('fs'); + const jobsSource = fs.readFileSync( + new URL('../src/commands/jobs.ts', import.meta.url), + 'utf8', + ); + + // The autopilot-cycle handler MUST pass signal to runCycle + // This is a source-level regression guard + const handlerBlock = jobsSource.slice( + jobsSource.indexOf("worker.register('autopilot-cycle'"), + jobsSource.indexOf("worker.register('autopilot-cycle'") + 500, + ); + + expect(handlerBlock).toContain('signal: job.signal'); + }); + + test('worker.ts has force-eviction safety net after timeout', async () => { + // Verify the worker code includes the grace timer + const fs = await import('fs'); + const workerSource = fs.readFileSync( + new URL('../src/core/minions/worker.ts', import.meta.url), + 'utf8', + ); + + // Must have the force-eviction pattern + expect(workerSource).toContain('Force-evicting from inFlight'); + expect(workerSource).toContain('graceTimer'); + expect(workerSource).toContain('handler ignored abort signal'); + }); + + test('cycle.ts has checkAborted calls between phases', async () => { + // Verify the cycle code checks abort between every phase + const fs = await import('fs'); + const cycleSource = fs.readFileSync( + new URL('../src/core/cycle.ts', import.meta.url), + 'utf8', + ); + + // Count checkAborted calls in the runCycle function body + const runCycleBody = cycleSource.slice( + cycleSource.indexOf('export async function runCycle'), + ); + const checkCalls = (runCycleBody.match(/checkAborted\(opts\.signal\)/g) || []).length; + + // Should have at least 6 (one per phase) + expect(checkCalls).toBeGreaterThanOrEqual(6); + }); +}); diff --git a/test/doctor-minions-check.test.ts b/test/doctor-minions-check.test.ts index 654f9af69..112892127 100644 --- a/test/doctor-minions-check.test.ts +++ b/test/doctor-minions-check.test.ts @@ -146,7 +146,9 @@ describe('gbrain doctor — half-migrated Minions detection', () => { test('filesystem: multiple versions each need their own complete entry', () => { // v0.10 is fully migrated but v0.11 is only partial. Doctor should - // flag v0.11 by name. + // flag v0.11 by name. The forward-progress override only kicks in + // when a NEWER version completed; v0.10 is older than v0.11 so the + // partial still stands. const migrationsDir = join(tmp, '.gbrain', 'migrations'); mkdirSync(migrationsDir, { recursive: true }); writeFileSync( @@ -166,6 +168,65 @@ describe('gbrain doctor — half-migrated Minions detection', () => { expect(minions!.message).not.toContain('0.10.0'); }); + test('filesystem: stale partial superseded by newer complete → NO warning (forward-progress override)', () => { + // v0.16.0 completed AFTER v0.11.0 went partial. The schema clearly + // advanced past v0.11.0, so the partial record is stale historical + // noise — not a real "MINIONS HALF-INSTALLED" condition. + // + // Without this override, every install that ever went through a + // v0.11.0 stopgap and then upgraded carries the FAIL flag forever, + // even on installs that have been at v0.22+ for months. Real cause: + // long-running gbrain installs accumulate partial entries from + // historical stopgap runs; a doctor flag with no time decay or + // forward-progress detection becomes meaningless once you've + // moved past those versions. + const migrationsDir = join(tmp, '.gbrain', 'migrations'); + mkdirSync(migrationsDir, { recursive: true }); + writeFileSync( + join(migrationsDir, 'completed.jsonl'), + [ + JSON.stringify({ version: '0.16.0', status: 'complete', ts: '2026-04-26T06:13:50.825Z' }), + JSON.stringify({ version: '0.11.0', status: 'partial', ts: '2026-04-26T06:16:56.298Z' }), + JSON.stringify({ version: '0.11.0', status: 'partial', ts: '2026-04-26T06:19:03.617Z' }), + ].join('\n') + '\n', + ); + + const result = run(['doctor', '--fast', '--json']); + // No FAIL on minions_migration — the v0.11.0 partials are stale + // because v0.16.0 (a newer release) completed. + const checks = JSON.parse(result.stdout).checks as Array<{ name: string; status: string }>; + const minions = checks.find(c => c.name === 'minions_migration'); + if (minions) { + expect(minions.status).not.toBe('fail'); + } + // Critically: the test fixture would have caused exit 1 under the old + // (no-override) logic because of the stale partial flag. Under the new + // logic, doctor exits 0 (or only warns about non-related checks). + expect(result.exitCode).toBe(0); + }); + + test('filesystem: stale partial NOT superseded → still flagged', () => { + // The override only fires when a >= partial version has completed. + // Older completes (e.g. v0.10 complete + v0.16 partial) do NOT + // supersede the partial; the partial still indicates a real problem. + const migrationsDir = join(tmp, '.gbrain', 'migrations'); + mkdirSync(migrationsDir, { recursive: true }); + writeFileSync( + join(migrationsDir, 'completed.jsonl'), + [ + JSON.stringify({ version: '0.10.0', status: 'complete' }), + JSON.stringify({ version: '0.16.0', status: 'partial' }), + ].join('\n') + '\n', + ); + + const result = run(['doctor', '--fast', '--json']); + expect(result.exitCode).toBe(1); + const checks = JSON.parse(result.stdout).checks as Array<{ name: string; status: string; message: string }>; + const minions = checks.find(c => c.name === 'minions_migration'); + expect(minions!.status).toBe('fail'); + expect(minions!.message).toContain('0.16.0'); + }); + test('human output: prints MINIONS HALF-INSTALLED loud banner', () => { // Same fixture as the first test, but check the human-readable output // includes the exact banner phrase an OpenClaw host's cron script diff --git a/test/e2e/worker-abort-recovery.test.ts b/test/e2e/worker-abort-recovery.test.ts new file mode 100644 index 000000000..775126340 --- /dev/null +++ b/test/e2e/worker-abort-recovery.test.ts @@ -0,0 +1,178 @@ +/** + * test/e2e/worker-abort-recovery.test.ts — E2E smoke test for worker + * recovery after handler timeout. + * + * Exercises the full path: submit job → handler runs → timeout fires → + * abort propagates → worker recovers → claims next job. + * + * This is the end-to-end regression test for the 2026-04-24 incident + * where a stuck autopilot-cycle handler wedged the worker with 98 jobs + * waiting and 0 active. + * + * Uses PGLite (in-memory), no external services needed. + */ + +import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { PGLiteEngine } from '../../src/core/pglite-engine.ts'; +import { MinionQueue } from '../../src/core/minions/queue.ts'; +import { MinionWorker } from '../../src/core/minions/worker.ts'; + +let engine: PGLiteEngine; +let queue: MinionQueue; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({ database_url: '' }); + await engine.initSchema(); + queue = new MinionQueue(engine); +}); + +afterAll(async () => { + await engine.disconnect(); +}); + +beforeEach(async () => { + await engine.executeRaw('DELETE FROM minion_jobs'); +}); + +describe('E2E: worker abort recovery (2026-04-24 regression)', () => { + test('worker recovers from timed-out handler and processes next job', async () => { + // Step 1: Submit a slow job with a short timeout + const slowJob = await queue.add('slow-handler', { type: 'slow' }, { + timeout_ms: 200, + max_attempts: 1, + }); + + // Step 2: Submit a fast job that should run AFTER the slow one times out + const fastJob = await queue.add('fast-handler', { type: 'fast' }, { + max_attempts: 1, + }); + + let slowHandlerAborted = false; + let fastHandlerExecuted = false; + + const worker = new MinionWorker(engine, { + pollInterval: 50, + concurrency: 1, // Single slot — forces sequential execution + }); + + // Slow handler: respects AbortSignal (the fix path) + worker.register('slow-handler', async (ctx) => { + // Simulate expensive work (like extract scanning 54K pages) + while (!ctx.signal.aborted) { + await new Promise(r => setTimeout(r, 20)); + } + slowHandlerAborted = true; + throw ctx.signal.reason || new Error('aborted'); + }); + + // Fast handler: just completes + worker.register('fast-handler', async () => { + fastHandlerExecuted = true; + return { done: true }; + }); + + // Step 3: Start worker + const workerPromise = worker.start(); + + // Step 4: Wait for slow job timeout (200ms) + handler abort + fast job execution + await new Promise(r => setTimeout(r, 1000)); + + // Step 5: Stop worker + worker.stop(); + await workerPromise; + + // Step 6: Verify + expect(slowHandlerAborted).toBe(true); + expect(fastHandlerExecuted).toBe(true); + + const slowResult = await queue.getJob(slowJob.id); + expect(slowResult!.status).toBe('dead'); + + const fastResult = await queue.getJob(fastJob.id); + expect(fastResult!.status).toBe('completed'); + expect(fastResult!.result).toEqual({ done: true }); + }); + + test('concurrency=2 worker still processes jobs while one slot is timing out', async () => { + const slowJob = await queue.add('slow-c2', {}, { + timeout_ms: 200, + max_attempts: 1, + }); + const fastJob = await queue.add('fast-c2', {}, { max_attempts: 1 }); + + let slowAborted = false; + let fastDone = false; + + const worker = new MinionWorker(engine, { + pollInterval: 50, + concurrency: 2, // Two slots — fast job can run in parallel + }); + + worker.register('slow-c2', async (ctx) => { + while (!ctx.signal.aborted) { + await new Promise(r => setTimeout(r, 10)); + } + slowAborted = true; + throw new Error('aborted'); + }); + + worker.register('fast-c2', async () => { + fastDone = true; + return { fast: true }; + }); + + const workerPromise = worker.start(); + await new Promise(r => setTimeout(r, 600)); + worker.stop(); + await workerPromise; + + expect(slowAborted).toBe(true); + expect(fastDone).toBe(true); + + const slowResult = await queue.getJob(slowJob.id); + expect(slowResult!.status).toBe('dead'); + + const fastResult = await queue.getJob(fastJob.id); + expect(fastResult!.status).toBe('completed'); + }); + + test('multiple timeouts in sequence dont permanently wedge worker', async () => { + // Submit 3 slow jobs that all timeout + 1 fast job + // The fast job MUST execute + const slow1 = await queue.add('multi-slow', {}, { timeout_ms: 100, max_attempts: 1 }); + const slow2 = await queue.add('multi-slow', {}, { timeout_ms: 100, max_attempts: 1 }); + const slow3 = await queue.add('multi-slow', {}, { timeout_ms: 100, max_attempts: 1 }); + const fast = await queue.add('multi-fast', {}, { max_attempts: 1 }); + + let timeoutsHit = 0; + let fastDone = false; + + const worker = new MinionWorker(engine, { pollInterval: 50, concurrency: 1 }); + + worker.register('multi-slow', async (ctx) => { + while (!ctx.signal.aborted) { + await new Promise(r => setTimeout(r, 10)); + } + timeoutsHit++; + throw new Error('aborted'); + }); + + worker.register('multi-fast', async () => { + fastDone = true; + return { ok: true }; + }); + + const workerPromise = worker.start(); + // 3 slow jobs × (100ms timeout + overhead) + fast job + margin + await new Promise(r => setTimeout(r, 2000)); + worker.stop(); + await workerPromise; + + expect(timeoutsHit).toBe(3); + expect(fastDone).toBe(true); + + const fastResult = await queue.getJob(fast.id); + expect(fastResult!.status).toBe('completed'); + }); +}); diff --git a/test/embed.test.ts b/test/embed.test.ts index 0c9992307..05f800a8f 100644 --- a/test/embed.test.ts +++ b/test/embed.test.ts @@ -106,14 +106,18 @@ describe('runEmbed --all (parallel)', () => { }); test('skips pages whose chunks are all already embedded when --stale', async () => { - const pages = [{ slug: 'fresh' }, { slug: 'stale' }]; const chunksBySlug = new Map([ ['fresh', [{ chunk_index: 0, chunk_text: 'hi', chunk_source: 'compiled_truth', embedded_at: '2026-01-01', token_count: 1 }]], ['stale', [{ chunk_index: 0, chunk_text: 'hi', chunk_source: 'compiled_truth', embedded_at: null, token_count: 1 }]], ]); + // Stale path uses countStaleChunks + listStaleChunks (SQL-side filter), not listPages. + const stale = [ + { slug: 'stale', chunk_index: 0, chunk_text: 'hi', chunk_source: 'compiled_truth', model: null, token_count: 1 }, + ]; const engine = mockEngine({ - listPages: async () => pages, + countStaleChunks: async () => 1, + listStaleChunks: async () => stale, getChunks: async (slug: string) => chunksBySlug.get(slug) || [], upsertChunks: async () => {}, }); @@ -145,9 +149,16 @@ describe('runEmbedCore --dry-run never calls the embedding model', () => { ], ]), ); + // SQL-side stale path: 6 stale rows across 3 pages. + const stale = pages.flatMap(p => [ + { slug: p.slug, chunk_index: 0, chunk_text: 'a', chunk_source: 'compiled_truth', model: null, token_count: 1 }, + { slug: p.slug, chunk_index: 1, chunk_text: 'b', chunk_source: 'compiled_truth', model: null, token_count: 1 }, + ]); const upserts: string[] = []; const engine = mockEngine({ + countStaleChunks: async () => 6, + listStaleChunks: async () => stale, listPages: async () => pages, getChunks: async (slug: string) => chunksBySlug.get(slug) || [], upsertChunks: async (slug: string) => { upserts.push(slug); }, @@ -163,32 +174,25 @@ describe('runEmbedCore --dry-run never calls the embedding model', () => { expect(result.dryRun).toBe(true); expect(result.embedded).toBe(0); expect(result.would_embed).toBe(6); // 3 pages * 2 chunks each + // skipped is 0 in the new SQL-side path: we never considered non-stale chunks. expect(result.skipped).toBe(0); - expect(result.total_chunks).toBe(6); + expect(result.total_chunks).toBe(6); // only stale chunks counted in SQL-side path expect(result.pages_processed).toBe(3); }); - test('dry-run --stale correctly separates stale from already-embedded', async () => { + test('dry-run --stale correctly identifies stale chunks (SQL-side path)', async () => { const { runEmbedCore } = await import('../src/commands/embed.ts'); - const pages = [{ slug: 'fresh' }, { slug: 'partial' }, { slug: 'all-stale' }]; - const chunksBySlug = new Map([ - ['fresh', [ - { chunk_index: 0, chunk_text: 'a', chunk_source: 'compiled_truth', embedded_at: '2026-01-01', token_count: 1 }, - { chunk_index: 1, chunk_text: 'b', chunk_source: 'compiled_truth', embedded_at: '2026-01-01', token_count: 1 }, - ]], - ['partial', [ - { chunk_index: 0, chunk_text: 'a', chunk_source: 'compiled_truth', embedded_at: '2026-01-01', token_count: 1 }, - { chunk_index: 1, chunk_text: 'b', chunk_source: 'compiled_truth', embedded_at: null, token_count: 1 }, - ]], - ['all-stale', [ - { chunk_index: 0, chunk_text: 'a', chunk_source: 'compiled_truth', embedded_at: null, token_count: 1 }, - { chunk_index: 1, chunk_text: 'b', chunk_source: 'compiled_truth', embedded_at: null, token_count: 1 }, - ]], - ]); + // SQL-side stale: only the 3 chunks where embedding IS NULL come back, + // grouped by slug. 'fresh' page has no stale rows so it's not in the result. + const stale = [ + { slug: 'partial', chunk_index: 1, chunk_text: 'b', chunk_source: 'compiled_truth', model: null, token_count: 1 }, + { slug: 'all-stale', chunk_index: 0, chunk_text: 'a', chunk_source: 'compiled_truth', model: null, token_count: 1 }, + { slug: 'all-stale', chunk_index: 1, chunk_text: 'b', chunk_source: 'compiled_truth', model: null, token_count: 1 }, + ]; const engine = mockEngine({ - listPages: async () => pages, - getChunks: async (slug: string) => chunksBySlug.get(slug) || [], + countStaleChunks: async () => 3, + listStaleChunks: async () => stale, upsertChunks: async () => {}, }); @@ -197,9 +201,11 @@ describe('runEmbedCore --dry-run never calls the embedding model', () => { expect(totalEmbedCalls).toBe(0); expect(result.dryRun).toBe(true); expect(result.would_embed).toBe(3); // 1 from 'partial' + 2 from 'all-stale' - expect(result.skipped).toBe(3); // 2 from 'fresh' + 1 from 'partial' - expect(result.total_chunks).toBe(6); - expect(result.pages_processed).toBe(3); + // SQL-side path does not see non-stale chunks, so skipped=0 and total_chunks=stale-count. + // Callers wanting full coverage should call engine.getStats()/getHealth() afterward. + expect(result.skipped).toBe(0); + expect(result.total_chunks).toBe(3); + expect(result.pages_processed).toBe(2); // 'partial' + 'all-stale' }); test('dry-run --slugs on a single page counts stale chunks, no API calls', async () => { @@ -228,7 +234,6 @@ describe('runEmbedCore --dry-run never calls the embedding model', () => { test('non-dry-run path reports accurate embedded count (regression guard)', async () => { const { runEmbedCore } = await import('../src/commands/embed.ts'); - const pages = [{ slug: 'a' }, { slug: 'b' }]; const chunksBySlug = new Map([ ['a', [{ chunk_index: 0, chunk_text: 'a', chunk_source: 'compiled_truth', embedded_at: null, token_count: 1 }]], ['b', [ @@ -236,9 +241,15 @@ describe('runEmbedCore --dry-run never calls the embedding model', () => { { chunk_index: 1, chunk_text: 'y', chunk_source: 'compiled_truth', embedded_at: null, token_count: 1 }, ]], ]); + const stale = [ + { slug: 'a', chunk_index: 0, chunk_text: 'a', chunk_source: 'compiled_truth', model: null, token_count: 1 }, + { slug: 'b', chunk_index: 0, chunk_text: 'x', chunk_source: 'compiled_truth', model: null, token_count: 1 }, + { slug: 'b', chunk_index: 1, chunk_text: 'y', chunk_source: 'compiled_truth', model: null, token_count: 1 }, + ]; const engine = mockEngine({ - listPages: async () => pages, + countStaleChunks: async () => 3, + listStaleChunks: async () => stale, getChunks: async (slug: string) => chunksBySlug.get(slug) || [], upsertChunks: async () => {}, }); @@ -253,3 +264,139 @@ describe('runEmbedCore --dry-run never calls the embedding model', () => { expect(result.pages_processed).toBe(2); }); }); + +// ──────────────────────────────────────────────────────────────── +// runEmbedCore --stale egress fix: SQL-side staleness filter +// Replaces the listPages + per-page getChunks bomb with a count + +// slug-grouped SELECT. On a 100%-embedded brain, 0 listPages calls. +// ──────────────────────────────────────────────────────────────── + +describe('runEmbedCore --stale egress fix (SQL-side filter)', () => { + test('zero stale chunks: countStaleChunks short-circuits, listPages never called', async () => { + const { runEmbedCore } = await import('../src/commands/embed.ts'); + let listPagesCalled = false; + let getChunksCalled = false; + let listStaleCalled = false; + const engine = mockEngine({ + countStaleChunks: async () => 0, + listPages: async () => { listPagesCalled = true; return []; }, + getChunks: async () => { getChunksCalled = true; return []; }, + listStaleChunks: async () => { listStaleCalled = true; return []; }, + upsertChunks: async () => {}, + }); + + const result = await runEmbedCore(engine, { stale: true }); + + expect(result.embedded).toBe(0); + expect(result.pages_processed).toBe(0); + // The egress fix: NONE of these should have been called when count=0. + expect(listPagesCalled).toBe(false); + expect(getChunksCalled).toBe(false); + expect(listStaleCalled).toBe(false); + expect(totalEmbedCalls).toBe(0); + }); + + test('N stale chunks across M pages: only stale slugs re-fetched, exact stale set embedded, non-stale chunks preserved', async () => { + const { runEmbedCore } = await import('../src/commands/embed.ts'); + let listPagesCalled = false; + + const stale = [ + { slug: 'page-a', chunk_index: 0, chunk_text: 'x', chunk_source: 'compiled_truth' as const, model: null, token_count: null }, + { slug: 'page-b', chunk_index: 1, chunk_text: 'y', chunk_source: 'compiled_truth' as const, model: null, token_count: null }, + { slug: 'page-b', chunk_index: 2, chunk_text: 'z', chunk_source: 'compiled_truth' as const, model: null, token_count: null }, + ]; + // page-b has a FRESH chunk at index 0 that must be preserved through the upsert. + const fullChunks: Record = { + 'page-a': [ + { chunk_index: 0, chunk_text: 'x', chunk_source: 'compiled_truth', embedded_at: null, token_count: 1 }, + ], + 'page-b': [ + { chunk_index: 0, chunk_text: 'fresh', chunk_source: 'compiled_truth', embedded_at: '2026-01-01', token_count: 5 }, + { chunk_index: 1, chunk_text: 'y', chunk_source: 'compiled_truth', embedded_at: null, token_count: 1 }, + { chunk_index: 2, chunk_text: 'z', chunk_source: 'compiled_truth', embedded_at: null, token_count: 1 }, + ], + }; + const upsertCalls: Array<{ slug: string; chunks: any[] }> = []; + const engine = mockEngine({ + countStaleChunks: async () => 3, + listStaleChunks: async () => stale, + listPages: async () => { listPagesCalled = true; return []; }, + getChunks: async (slug: string) => fullChunks[slug] || [], + upsertChunks: async (slug: string, chunks: any[]) => { upsertCalls.push({ slug, chunks }); }, + }); + + const result = await runEmbedCore(engine, { stale: true }); + + // listPages must NOT be called in the SQL-side path. + expect(listPagesCalled).toBe(false); + // One embedBatch call per stale slug (a, b). + expect(totalEmbedCalls).toBe(2); + expect(result.embedded).toBe(3); + expect(result.pages_processed).toBe(2); + + // page-b's upsert MUST include the fresh chunk (chunk_index=0) — otherwise + // it would be deleted by the upsertChunks != ALL filter. Critical regression check. + const pageBUpsert = upsertCalls.find(u => u.slug === 'page-b'); + expect(pageBUpsert).toBeDefined(); + const freshChunkInUpsert = pageBUpsert!.chunks.find((c: any) => c.chunk_index === 0); + expect(freshChunkInUpsert).toBeDefined(); + // Fresh chunk has no `embedding` field (preserved via COALESCE in upsertChunks SQL). + expect(freshChunkInUpsert.embedding).toBeUndefined(); + // Previously-stale chunks come through WITH a new embedding. + const staleChunkInUpsert = pageBUpsert!.chunks.find((c: any) => c.chunk_index === 1); + expect(staleChunkInUpsert.embedding).toBeDefined(); + expect(staleChunkInUpsert.embedding).toBeInstanceOf(Float32Array); + }); + + test('--stale dry-run: counts stale via countStaleChunks, reports via listStaleChunks, no embedBatch or upsertChunks', async () => { + const { runEmbedCore } = await import('../src/commands/embed.ts'); + const stale = [ + { slug: 'page-a', chunk_index: 0, chunk_text: 'x', chunk_source: 'compiled_truth' as const, model: null, token_count: null }, + { slug: 'page-b', chunk_index: 0, chunk_text: 'y', chunk_source: 'compiled_truth' as const, model: null, token_count: null }, + ]; + const upserts: string[] = []; + const engine = mockEngine({ + countStaleChunks: async () => 2, + listStaleChunks: async () => stale, + upsertChunks: async (slug: string) => { upserts.push(slug); }, + }); + + const result = await runEmbedCore(engine, { stale: true, dryRun: true }); + + expect(totalEmbedCalls).toBe(0); + expect(upserts).toEqual([]); + expect(result.would_embed).toBe(2); + expect(result.pages_processed).toBe(2); + expect(result.dryRun).toBe(true); + }); + + test('--all (non-stale) path is byte-identical: walks listPages and embeds every chunk', async () => { + // Regression guard for the legacy --all path. Behavior must be byte-identical + // to pre-fix: listPages + per-page getChunks + embed every chunk. + const { runEmbedCore } = await import('../src/commands/embed.ts'); + let countStaleCalled = false; + let listStaleCalled = false; + const pages = [{ slug: 'a' }, { slug: 'b' }]; + const chunksBySlug = new Map([ + ['a', [{ chunk_index: 0, chunk_text: 'a', chunk_source: 'compiled_truth', embedded_at: '2026-01-01', token_count: 1 }]], + ['b', [{ chunk_index: 0, chunk_text: 'b', chunk_source: 'compiled_truth', embedded_at: null, token_count: 1 }]], + ]); + + const engine = mockEngine({ + countStaleChunks: async () => { countStaleCalled = true; return 1; }, + listStaleChunks: async () => { listStaleCalled = true; return []; }, + listPages: async () => pages, + getChunks: async (slug: string) => chunksBySlug.get(slug) || [], + upsertChunks: async () => {}, + }); + + const result = await runEmbedCore(engine, { all: true }); + + // --all path must NOT take the new short-circuit. + expect(countStaleCalled).toBe(false); + expect(listStaleCalled).toBe(false); + // Both pages get embedded, regardless of embedded_at — that's the --all contract. + expect(totalEmbedCalls).toBe(2); + expect(result.embedded).toBe(2); + }); +}); diff --git a/test/extract-incremental.test.ts b/test/extract-incremental.test.ts new file mode 100644 index 000000000..7116774f9 --- /dev/null +++ b/test/extract-incremental.test.ts @@ -0,0 +1,182 @@ +/** + * Regression guards for the incremental extract path (PR #417). + * + * Eng-review Step 5: 8 unit cases asserting `runExtractCore({ slugs })` + * processes only the requested slugs in the cycle path while + * `slugs: undefined` falls through to the existing full-walk behavior. + * + * All tests use PGLite/in-memory — no DB connection required. + */ +import { describe, test, expect, beforeEach, afterEach } from 'bun:test'; +import { mkdtempSync, writeFileSync, mkdirSync, rmSync } from 'fs'; +import { tmpdir } from 'os'; +import { join } from 'path'; +import { runExtractCore } from '../src/commands/extract.ts'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import type { BrainEngine } from '../src/core/engine.ts'; + +let engine: PGLiteEngine; +let tempDir: string; + +beforeEach(async () => { + engine = new PGLiteEngine(); + await engine.connect({ engine: 'pglite' }); + await engine.initSchema(); + tempDir = mkdtempSync(join(tmpdir(), 'gbrain-extract-test-')); + mkdirSync(join(tempDir, 'people'), { recursive: true }); + mkdirSync(join(tempDir, 'companies'), { recursive: true }); +}); + +afterEach(async () => { + await engine.disconnect(); + rmSync(tempDir, { recursive: true, force: true }); +}); + +async function seedPage(slug: string, body: string): Promise { + const [type, name] = slug.split('/'); + await engine.putPage(slug, { + type: type as 'person' | 'company', + title: name, + compiled_truth: body, + timeline: '', + frontmatter: {}, + content_hash: 'h', + }); + // Also write to disk so walkMarkdownFiles can find it + const filePath = join(tempDir, slug + '.md'); + mkdirSync(join(tempDir, type), { recursive: true }); + writeFileSync(filePath, body); +} + +describe('runExtractCore — incremental cycle path (#417)', () => { + test('1. slugs: [] returns immediately with zero counts (early-return path)', async () => { + await seedPage('people/alice-example', '# alice'); + const result = await runExtractCore(engine as unknown as BrainEngine, { + mode: 'all', + dir: tempDir, + slugs: [], + }); + expect(result.links_created).toBe(0); + expect(result.timeline_entries_created).toBe(0); + expect(result.pages_processed).toBe(0); + }); + + test('2. slugs: undefined falls through to full-walk path', async () => { + await seedPage('people/alice-example', '# alice\n\n[bob](people/bob-example)'); + await seedPage('people/bob-example', '# bob'); + const result = await runExtractCore(engine as unknown as BrainEngine, { + mode: 'all', + dir: tempDir, + }); + // Full walk processes everything found on disk + expect(result.pages_processed).toBeGreaterThan(0); + }); + + test('3. slugs: [a, b] reads only those two files (incremental processing)', async () => { + await seedPage('people/alice-example', '# alice'); + await seedPage('people/bob-example', '# bob'); + await seedPage('people/charlie-example', '# charlie'); + const result = await runExtractCore(engine as unknown as BrainEngine, { + mode: 'all', + dir: tempDir, + slugs: ['people/alice-example', 'people/bob-example'], + }); + // Only 2 files processed even though 3 exist on disk + expect(result.pages_processed).toBe(2); + }); + + test('4. Slug whose file no longer exists is silently skipped', async () => { + await seedPage('people/alice-example', '# alice'); + // people/ghost has no file on disk but is in the slugs list + const result = await runExtractCore(engine as unknown as BrainEngine, { + mode: 'all', + dir: tempDir, + slugs: ['people/alice-example', 'people/ghost'], + }); + // alice processed; ghost skipped (no file) + expect(result.pages_processed).toBe(1); + }); + + test('5. mode: links skips timeline extraction in incremental', async () => { + const body = '# alice\n\n## Timeline\n- 2026-01-01: started'; + await seedPage('people/alice-example', body); + const result = await runExtractCore(engine as unknown as BrainEngine, { + mode: 'links', + dir: tempDir, + slugs: ['people/alice-example'], + }); + // Timeline extraction skipped even though body contains a timeline + expect(result.timeline_entries_created).toBe(0); + }); + + test('6. dryRun: true does not invoke addLinksBatch / addTimelineEntriesBatch', async () => { + await seedPage('people/alice-example', '# alice\n\n[bob](people/bob-example)'); + await seedPage('people/bob-example', '# bob'); + + let linksBatchCalled = false; + let timelineBatchCalled = false; + const originalAddLinks = engine.addLinksBatch.bind(engine); + const originalAddTimeline = engine.addTimelineEntriesBatch.bind(engine); + (engine as unknown as { addLinksBatch: typeof originalAddLinks }).addLinksBatch = async (...args) => { + linksBatchCalled = true; + return originalAddLinks(...args); + }; + (engine as unknown as { addTimelineEntriesBatch: typeof originalAddTimeline }).addTimelineEntriesBatch = async (...args) => { + timelineBatchCalled = true; + return originalAddTimeline(...args); + }; + + await runExtractCore(engine as unknown as BrainEngine, { + mode: 'all', + dir: tempDir, + slugs: ['people/alice-example'], + dryRun: true, + }); + + expect(linksBatchCalled).toBe(false); + expect(timelineBatchCalled).toBe(false); + }); + + test('7. BATCH_SIZE flush — slugs producing >100 candidate links exercise the mid-iteration flush', async () => { + // BATCH_SIZE in extract.ts is 100. Create one slug with 150 outbound links. + const targets: string[] = []; + for (let i = 0; i < 150; i++) { + const target = `companies/co-${i}`; + targets.push(target); + await seedPage(target, `# co-${i}`); + } + const linkBlock = targets.map(t => `- [${t}](${t})`).join('\n'); + await seedPage('people/alice-example', `# alice\n\n${linkBlock}`); + + const result = await runExtractCore(engine as unknown as BrainEngine, { + mode: 'links', + dir: tempDir, + slugs: ['people/alice-example'], + }); + // The flush happens mid-iteration when batch hits 100; the remaining 50 flush at end. + // No exception means the flush path executed cleanly. + expect(result.pages_processed).toBe(1); + expect(result.links_created).toBeGreaterThanOrEqual(0); // Just confirms the flush path didn't blow up + }); + + test('8. Full-slug-set resolution — slug references file outside changed set', async () => { + // alice references bob, but only alice is in the incremental slugs list. + // The allSlugs set must still include bob (from walkMarkdownFiles) so + // resolveSlug succeeds; otherwise the link would silently drop. + // Markdown link pattern requires .md target. + await seedPage('people/alice-example', '# alice\n\n[bob](bob-example.md)'); + await seedPage('people/bob-example', '# bob'); + + const result = await runExtractCore(engine as unknown as BrainEngine, { + mode: 'links', + dir: tempDir, + slugs: ['people/alice-example'], + }); + + // Only alice's file was read, but the resulting link must reference bob + // (resolved via the full allSlugs set built from walkMarkdownFiles). + expect(result.pages_processed).toBe(1); + // Link from alice to bob was extracted successfully via the full allSlugs set + expect(result.links_created).toBeGreaterThan(0); + }); +}); diff --git a/test/migrate.test.ts b/test/migrate.test.ts index b500f149a..ed17bd78d 100644 --- a/test/migrate.test.ts +++ b/test/migrate.test.ts @@ -769,26 +769,35 @@ describe('PR #356 — apply-migrations pre-flight schema-version warning', () => }); }); -describe('PR #356 — setSessionDefaults is applied on both db.ts and postgres-engine.ts paths', () => { - test('structural: idle_in_transaction_session_timeout set via single helper', () => { - // After PR #356 extracted setSessionDefaults, both connect paths - // should call the helper, not inline the SET. Any regression - // that re-duplicates the block gets caught here. +describe('PR #356 + #363 — session timeouts applied via startup parameters', () => { + test('structural: setSessionDefaults exists for back-compat; resolveSessionTimeouts is the source of truth', () => { + // PR #356 introduced setSessionDefaults (post-pool SET). + // PR #363 superseded it with resolveSessionTimeouts (startup parameters, + // PgBouncer-transaction-mode-safe). The setSessionDefaults function is + // kept as a no-op shim for back-compat with existing call sites. const dbSrc = readFileSync(resolve('src/core/db.ts'), 'utf-8'); const pgSrc = readFileSync(resolve('src/core/postgres-engine.ts'), 'utf-8'); - // Helper is defined in db.ts + // Helper still exists for back-compat expect(dbSrc).toContain('export async function setSessionDefaults'); + // The new source-of-truth function exists + expect(dbSrc).toContain('export function resolveSessionTimeouts'); expect(dbSrc).toContain('idle_in_transaction_session_timeout'); - // connect() in db.ts calls the helper, doesn't inline the SET - // (the SET only appears inside the helper itself now). - const setMatches = dbSrc.match(/SET idle_in_transaction_session_timeout/g) || []; - expect(setMatches.length).toBe(1); // only in the helper + // Both connect paths call resolveSessionTimeouts() and feed it through + // postgres.js's connection option (startup parameters) + expect(dbSrc).toContain('resolveSessionTimeouts()'); + expect(pgSrc).toContain('resolveSessionTimeouts()'); - // postgres-engine.ts calls the helper too, doesn't duplicate + // setSessionDefaults still callable (no-op) so existing call sites + // don't break, but the SET command itself is gone — the work has + // already happened at connection startup time. expect(pgSrc).toContain('db.setSessionDefaults'); - expect(pgSrc).not.toContain("SET idle_in_transaction_session_timeout"); + + // Critically: no SET idle_in_transaction in source — startup parameters + // are the durable mechanism for PgBouncer transaction mode. + const setMatches = dbSrc.match(/SET idle_in_transaction_session_timeout/g) || []; + expect(setMatches.length).toBe(0); }); }); @@ -809,3 +818,83 @@ describe('PR #356 — non-transactional DDL runs via reserved connection', () => expect(fnBody).toContain("SET statement_timeout = '600000'"); }); }); + +// ───────────────────────────────────────────────────────────────── +// PR #363 regression guards — session timeouts via startup parameters +// resolveSessionTimeouts — GBRAIN_*_TIMEOUT env overrides +// ───────────────────────────────────────────────────────────────── +// +// Guards: orphan pgbouncer backends that hold table locks for hours when +// the postgres.js client disconnects mid-transaction. Session-level +// statement_timeout + idle_in_transaction_session_timeout delivered as +// startup parameters kill those backends on the server side. + +describe('resolveSessionTimeouts — env var overrides', () => { + const { resolveSessionTimeouts } = require('../src/core/db.ts'); + const origStatement = process.env.GBRAIN_STATEMENT_TIMEOUT; + const origIdleTx = process.env.GBRAIN_IDLE_TX_TIMEOUT; + const origCheck = process.env.GBRAIN_CLIENT_CHECK_INTERVAL; + + afterAll(() => { + const restore = (key: string, val: string | undefined) => { + if (val === undefined) delete process.env[key]; + else process.env[key] = val; + }; + restore('GBRAIN_STATEMENT_TIMEOUT', origStatement); + restore('GBRAIN_IDLE_TX_TIMEOUT', origIdleTx); + restore('GBRAIN_CLIENT_CHECK_INTERVAL', origCheck); + }); + + const resetEnv = () => { + delete process.env.GBRAIN_STATEMENT_TIMEOUT; + delete process.env.GBRAIN_IDLE_TX_TIMEOUT; + delete process.env.GBRAIN_CLIENT_CHECK_INTERVAL; + }; + + test('returns statement_timeout + idle_in_transaction defaults when unset', () => { + resetEnv(); + const t = resolveSessionTimeouts(); + expect(t.statement_timeout).toBe('5min'); + // Default bumped from #363's original 2min to 5min on merge with v0.21.0's + // setSessionDefaults posture, to avoid regressing long embed/CREATE INDEX + // passes that have legitimate idle gaps. + expect(t.idle_in_transaction_session_timeout).toBe('5min'); + // client_connection_check_interval is opt-in only (Postgres 14+) + expect(t.client_connection_check_interval).toBeUndefined(); + }); + + test('env vars override the defaults', () => { + resetEnv(); + process.env.GBRAIN_STATEMENT_TIMEOUT = '10min'; + process.env.GBRAIN_IDLE_TX_TIMEOUT = '30s'; + process.env.GBRAIN_CLIENT_CHECK_INTERVAL = '15s'; + const t = resolveSessionTimeouts(); + expect(t.statement_timeout).toBe('10min'); + expect(t.idle_in_transaction_session_timeout).toBe('30s'); + expect(t.client_connection_check_interval).toBe('15s'); + }); + + test("'0' disables a specific GUC", () => { + resetEnv(); + process.env.GBRAIN_STATEMENT_TIMEOUT = '0'; + const t = resolveSessionTimeouts(); + expect(t.statement_timeout).toBeUndefined(); + expect(t.idle_in_transaction_session_timeout).toBe('5min'); + }); + + test("'off' disables a specific GUC", () => { + resetEnv(); + process.env.GBRAIN_IDLE_TX_TIMEOUT = 'off'; + const t = resolveSessionTimeouts(); + expect(t.statement_timeout).toBe('5min'); + expect(t.idle_in_transaction_session_timeout).toBeUndefined(); + }); + + test('all three can be disabled independently', () => { + resetEnv(); + process.env.GBRAIN_STATEMENT_TIMEOUT = '0'; + process.env.GBRAIN_IDLE_TX_TIMEOUT = 'off'; + const t = resolveSessionTimeouts(); + expect(Object.keys(t)).toHaveLength(0); + }); +}); diff --git a/test/minions.test.ts b/test/minions.test.ts index ce377a1f8..68e87b675 100644 --- a/test/minions.test.ts +++ b/test/minions.test.ts @@ -1896,3 +1896,158 @@ describe('MinionQueue: v0.19.1 wall-clock + handleTimeouts non-interference (T1) expect(after?.status).toBe('active'); }); }); + +// --------------------------------------------------------------------------- +// Abort signal propagation + force-eviction (v0.20.5 cycle-abort fix) +// --------------------------------------------------------------------------- + +describe('MinionWorker: abort signal propagation (v0.20.5)', () => { + test('handler receiving abort signal can exit cleanly', async () => { + // Handler that respects AbortSignal + const job = await queue.add('abort-aware', {}, { timeout_ms: 150, max_attempts: 1 }); + let signalAborted = false; + + const worker = new MinionWorker(engine, { pollInterval: 50 }); + worker.register('abort-aware', async (ctx) => { + // Simulate long work that checks signal + while (!ctx.signal.aborted) { + await new Promise(r => setTimeout(r, 10)); + } + signalAborted = true; + throw ctx.signal.reason || new Error('aborted'); + }); + + const workerPromise = worker.start(); + // Wait for timeout (150ms) + handler to notice + margin + await new Promise(r => setTimeout(r, 500)); + worker.stop(); + await workerPromise; + + expect(signalAborted).toBe(true); + const result = await queue.getJob(job.id); + // Should be dead (max_attempts: 1, aborted) + expect(result!.status).toBe('dead'); + expect(result!.error_text).toContain('abort'); + }); + + test('handler ignoring abort signal still gets abort fired', async () => { + // Handler that IGNORES AbortSignal — the exact bug pattern. + // We verify the abort fires (the signal flips) even though the handler + // doesn't check it. The 30s force-eviction grace is too long for unit + // tests; the E2E test in test/e2e/worker-abort-recovery.test.ts covers + // the full force-eviction path. Here we just verify the abort signal + // is delivered to the handler context. + const job = await queue.add('abort-ignorer', {}, { timeout_ms: 100, max_attempts: 1 }); + let handlerStarted = false; + let signalWasAborted = false; + + const worker = new MinionWorker(engine, { pollInterval: 50 }); + worker.register('abort-ignorer', async (ctx) => { + handlerStarted = true; + // Wait a bit, then check if signal was aborted + await new Promise(r => setTimeout(r, 200)); + signalWasAborted = ctx.signal.aborted; + // Now exit (a well-behaved handler would do this) + if (ctx.signal.aborted) { + throw ctx.signal.reason || new Error('aborted'); + } + return { ok: true }; + }); + + const workerPromise = worker.start(); + await new Promise(r => setTimeout(r, 500)); + + expect(handlerStarted).toBe(true); + expect(signalWasAborted).toBe(true); + + worker.stop(); + await workerPromise; + + const result = await queue.getJob(job.id); + expect(result!.status).toBe('dead'); + }); + + test('worker claims new jobs after timeout eviction (no wedge)', async () => { + // The critical regression test: submit a slow job that times out, + // then submit a fast job. The fast job MUST execute. + const slowJob = await queue.add('slow-timeout', {}, { timeout_ms: 100, max_attempts: 1 }); + let slowAborted = false; + let fastExecuted = false; + + const worker = new MinionWorker(engine, { pollInterval: 50, concurrency: 1 }); + worker.register('slow-timeout', async (ctx) => { + // Respects abort but takes a moment + await new Promise(r => setTimeout(r, 50)); + while (!ctx.signal.aborted) { + await new Promise(r => setTimeout(r, 10)); + } + slowAborted = true; + throw new Error('aborted: timeout'); + }); + worker.register('fast-after', async () => { + fastExecuted = true; + return { fast: true }; + }); + + const workerPromise = worker.start(); + + // Wait for slow job to start and timeout + await new Promise(r => setTimeout(r, 300)); + + // Now submit the fast job — it should get claimed + const fastJob = await queue.add('fast-after', {}); + await new Promise(r => setTimeout(r, 300)); + + worker.stop(); + await workerPromise; + + expect(slowAborted).toBe(true); + expect(fastExecuted).toBe(true); + + const slowResult = await queue.getJob(slowJob.id); + expect(slowResult!.status).toBe('dead'); + + const fastResult = await queue.getJob(fastJob.id); + expect(fastResult!.status).toBe('completed'); + }); +}); + +// --------------------------------------------------------------------------- +// checkAborted (v0.20.5 cycle.ts) +// --------------------------------------------------------------------------- + +describe('checkAborted (v0.20.5 cycle signal)', () => { + // Import the function indirectly by testing the behavior pattern + test('undefined signal does not throw', () => { + // checkAborted is not exported, so we test through CycleOpts behavior. + // This test validates the pattern directly. The `as` cast keeps the + // union type intact — a bare `const signal = undefined` (or even + // `const signal: AbortSignal | undefined = undefined`) would narrow + // back to literal `undefined` via TS control-flow analysis and then + // reject the optional-chain access on it. + const signal = undefined as AbortSignal | undefined; + expect(() => { + if (signal?.aborted) throw new Error('aborted'); + }).not.toThrow(); + }); + + test('non-aborted signal does not throw', () => { + const abort = new AbortController(); + expect(() => { + if (abort.signal.aborted) throw new Error('aborted'); + }).not.toThrow(); + }); + + test('aborted signal throws with reason', () => { + const abort = new AbortController(); + abort.abort(new Error('timeout')); + expect(() => { + if (abort.signal.aborted) { + const reason = abort.signal.reason instanceof Error + ? abort.signal.reason.message + : String(abort.signal.reason || 'aborted'); + throw new Error(`[cycle] aborted between phases: ${reason}`); + } + }).toThrow('aborted between phases: timeout'); + }); +});