From bd4c976a85a3cab0adf05dc9233aa75a7be9bbbe Mon Sep 17 00:00:00 2001 From: test Date: Thu, 13 Aug 2026 10:41:33 -0700 Subject: [PATCH] fix(sync,webhook): consume deferred link extraction above the size gate (#2849) (#3561) Wave-assembled from PR #3561 by @time-attack. Co-Authored-By: Garry Tan --- docs/architecture/KEY_FILES.md | 4 +- src/commands/extract.ts | 6 +- src/commands/jobs.ts | 39 ++- src/commands/sync.ts | 64 ++++- ...sync-deferred-extract-queue.serial.test.ts | 253 ++++++++++++++++++ 5 files changed, 358 insertions(+), 8 deletions(-) create mode 100644 test/sync-deferred-extract-queue.serial.test.ts diff --git a/docs/architecture/KEY_FILES.md b/docs/architecture/KEY_FILES.md index 69e0e6243..37ae2e593 100644 --- a/docs/architecture/KEY_FILES.md +++ b/docs/architecture/KEY_FILES.md @@ -46,7 +46,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/commands/reindex-code.ts` — `gbrain reindex --code [--source ID] [--dry-run] [--yes] [--json] [--force] [--no-embed]`. Walks `pages WHERE type = 'code'` in 100-row batches, replays through `importCodeFile` for chunk + embed + content_hash folding. Idempotent unless `--force` bypasses the content_hash early-return. Cost-preview model field reads `getEmbeddingModelName()` from the gateway so preview reflects what the gateway will actually embed with. An informational stderr nudge inside `runReindexCode` (so dry-run + execute both surface it): when the configured embedding model isn't code-tuned (allowlist `{'voyage-code-3'}`, case-insensitive bare match), prints a recommendation to switch to `voyage:voyage-code-3`; suppress with `GBRAIN_NO_CODE_MODEL_NUDGE=1`, `--no-embed`, or `--json`. Pure `shouldNudgeCodeModel(bareName)` returns a tagged `NudgeDecision` union (takes the bare model name, emits qualified `voyage:voyage-code-3` for the paste-ready `gbrain config set` line). When `--yes` is absent and the caller is non-TTY or passed `--json`, the cost gate refuses (exit 2, no spend) via the pure exported `buildCostRefusal({json, previewMsg, preview, costUsd, model}): {stdout?, stderr?}` — JSON envelope only when `--json` is explicit, otherwise a human refusal on stderr (the spend guardrail is independent of the output format). `spend.posture=tokenmax` OR an explicit `--max-cost off`/`unlimited` makes the gate informational and proceeds (#2139); `--max-cost off` also disables the runtime BudgetTracker cap. Pinned by `test/ai/voyage-code-3-recipe.test.ts`, `test/reindex-code-nudge.serial.test.ts`, `test/reindex-code-model-source.serial.test.ts` (IRON-RULE regression for the cost-preview fix), `test/reindex-cost-refusal.test.ts`. - `src/core/fts-language.ts` — Single source for the Postgres text-search configuration name used by FTS. `getFtsLanguage()` resolves `GBRAIN_FTS_LANGUAGE` (default `english`), validates against `/^[a-z][a-z0-9_]*$/` (tsvector config names can't be bound as parameters, so the value is interpolated into raw SQL — the allowlist regex is the injection guard; invalid values warn once and fall back to `english`), and caches on first read (`resetFtsLanguageCache()` is test-only). Consumed by both engines' `searchKeyword`/`searchKeywordChunks` (`websearch_to_tsquery` query side), the `configurable_fts_language` migration, and `reindex-search-vector` (write-side trigger functions). Pinned by `test/fts-language.serial.test.ts` + `test/fts-language-migration.serial.test.ts` (includes the `'; DROP TABLE pages; --` injection cases). - `src/commands/reindex-search-vector.ts` — `gbrain reindex-search-vector [--dry-run] [--yes] [--json]`. Escape hatch for changing `GBRAIN_FTS_LANGUAGE` after the `configurable_fts_language` migration has run (the migration shows applied and is skipped): recreates `update_page_search_vector` + `update_chunk_search_vector` with the configured language — bodies mirror the migration's and KEEP the `SET search_path = pg_catalog, public` hardening (CREATE OR REPLACE resets proconfig) — then backfills `pages` (UPDATE-to-self re-fires the trigger) and `content_chunks` (direct vector recompute) in id-keyset batches of `BACKFILL_BATCH_SIZE` (5000) via `UPDATE … WHERE id IN (SELECT … LIMIT n) RETURNING id`, streaming phases `reindex_search_vector.pages`/`.chunks` through the shared progress reporter (stderr). Confirmation gate: `--yes`, or an interactive TTY [y/N]; `--json` does NOT bypass the gate (non-TTY without `--yes` refuses with a ConfirmationRequired envelope, exit 2). Idempotent. Pinned by `test/reindex-search-vector.serial.test.ts`. -- `src/commands/sync.ts` — `gbrain sync` CLI + the `performSync` / `performFullSync` library entrypoints (consumed by the autopilot cycle and the Minion sync handler). `performSync` runs under a writer lock: per-source `gbrain-sync:` whenever `opts.sourceId` is set, wrapped in `withRefreshingLock` from `src/core/db-lock.ts` so long-running sources (250K+ chunks) don't lose the lock at the 30-min TTL mid-run; the bare no-source path uses the SAME refreshing lock (#1794 — it was previously a non-refreshing `tryAcquireDbLock`, stealable mid-run during an incident); `SyncOpts.lockId?: string` is the explicit override. The lock refresh AND its health probe route through the DIRECT session pool so Supavisor transaction-pool exhaustion (`EMAXCONNSESSION`) can't kill renewal; takeover is heartbeat-aware (it will NOT steal a holder whose `last_refreshed_at` is within `GBRAIN_LOCK_STEAL_GRACE_SECONDS`, defending an alive-but-starved holder); the import loop yields the event loop every `GBRAIN_SYNC_YIELD_EVERY` files (`setTimeout(0)`, not `setImmediate` — Bun starves the timers phase) so the refresh `setInterval` heartbeat fires mid-import. This lock-identity invariant prevents a `sync --all` per-source worker racing `sync --source foo` on the global lock from corrupting the same source. `performSync` throws a typed `SyncLockBusyError` when the writer lock is held; the Minion `sync` handler (`src/commands/jobs.ts`) catches it and marks the job SKIPPED (not failed) so a cron/autopilot tick defers to the holder without polluting crash metrics. `performSyncInner` is RESUMABLE (incremental path): it drains a PINNED target commit (`lastCommit..pin`), banking drained file paths via `appendCompleted` (append-only delta into the `op_checkpoint_paths` child table, migration v115 — one row per path, O(delta) not the old O(N²) full-array rewrite), keyed by `syncFingerprint({sourceId, lastCommit})` from `src/core/op-checkpoint.ts` (paths under `op:'sync'`; the pinned target under `op:'sync-target'`), and advances `last_commit`/`last_sync_at` ONLY at full import completion. Checkpoint writes route through the DIRECT session pool + bounded retry so they survive `EMAXCONNSESSION`; the flush cadence is first-file then every `GBRAIN_SYNC_CHECKPOINT_EVERY` (default 1000) files OR `GBRAIN_SYNC_CHECKPOINT_SECONDS` (default 10s), with a race-safe `pendingCheckpointPaths` delta (single-flight swap, re-merge on failure) under parallel workers; a SIGTERM banks the in-flight delta via a no-retry one-shot (`appendCompletedOnce`, ordered before lock release through `registerCleanup`); and sustained flush failure aborts the run with `reason:'checkpoint_unavailable'` after `GBRAIN_SYNC_MAX_CHECKPOINT_FAILURES` consecutive fails rather than importing work it can never bank (every partial/blocked exit logs the banked-file count). A sync killed mid-import banks its progress, leaves the anchor unmoved (the source stays correctly stale to the autopilot scheduler — `last_sync_at` is never bumped on a partial), and the next run `resumeFilter`s the same fixed diff to skip done files. The pin is the checkpoint's stored target when still reachable from HEAD, else live HEAD (a history rewrite / reset re-pins); completion advances to the pin, NOT live HEAD, so commits landing past the pin are a clean next-sync diff — this closes the cross-run staleness window. After import a pin-reachability gate (`git merge-base --is-ancestor pin HEAD`) replaces the old strict head-drift gate: forward commits on top of the pin (e.g. a background enrich process committing to the same repo every ~2 min) are SAFE and no longer block the run; only a real rewrite (pin not an ancestor of HEAD) blocks and discards the checkpoint. A file added in `lastCommit..pin` but gone from disk (deleted by a commit after the pin) is SKIPPED and marked completed, not recorded as a failure. Downstream extract/facts/embed are size-gated: inline only for `totalChanges <= 100`; large syncs defer to the resumable `extract --stale` watermark + `embed --stale`/backfill + the facts cycle phases, so a 44K-page facts/embed pass never re-blocks import convergence (sync convergence == import convergence). Worker engines wrap in try/finally so disconnect always fires; both PGLite-detection sites use `engine.kind === 'pglite'`. CLI accepts `--workers N` (alias `--concurrency N`) validated via `parseWorkers` (explicit bypasses the file-count floor; auto path defers to `autoConcurrency()`). The newest-first descending-lex order uses `sortNewestFirst(addsAndMods)` from `src/core/sort-newest-first.ts` (shared with `gbrain import`). `gbrain sync --all` runs a continuous worker pool: `parseWorkers`-validated `--parallel N` (default `min(sourceCount, --workers, DEFAULT_PARALLEL_SOURCES=4)`), long-lived async workers pulling from a shared FIFO queue (no head-of-line blocking), per-source `withSourcePrefix(src.id, ...)` so every `slog`/`serr` line carries `[]`; `--skip-failed`/`--retry-failed` are scoped per source (`acknowledgeFailures(sourceId)`; `--all` acks every source, single-source acks only its own) and run UNDER parallel — the #1939 failure ledger is per-`(source_id, path)` and serialized through `withLedgerLock`, so the old "not supported under parallel, re-run with --serial" refusal is lifted (#2139, which also removed the forcing-function that pushed recovery syncs to `--serial` and thus armed the inline cost gate); a connection-budget stderr warning fires when `parallel × workers × 2 > 16` (the `× 2 per-file pool` factor: each per-file worker opens its own `PostgresEngine` with `poolSize=2`). Exports `resolveParallelism`, `syncOneSource`, `buildSyncStatusReport`, `printSyncStatusReport`, `SyncStatusReport` back the `gbrain sources status` dashboard. `--json` envelope `{schema_version: 1, sources, parallel, ok_count, error_count, skipped_count}` on stdout; human banners route to stderr via `humanSink` so `jq` parses cleanly. Exit matrix: 0 all ok (sources skipped by `--missing-path skip` count as ok), 1 any error. `--missing-path ` (default fail) handles sources whose `local_path` does not exist on this machine — machine-specific state in a brain-wide table, so a brain registered from several machines fails every foreign source on every run; `skip` classifies them `skipped_missing_path` (⊘ line, envelope entry with `local_path`, excluded from `error_count` and the rc gate) via the exported pure helpers `parseMissingPathMode` + `partitionMissingPathSources`, pinned by `test/sync-all-missing-path.test.ts`; default `fail` stays loud because on a single-machine brain a missing path usually means an unmounted volume. (The non-TTY cost gate no longer exits 2 — it auto-defers; #2139.) The dashboard SQL is `content_chunks ch JOIN pages pg ON pg.id = ch.page_id WHERE pg.deleted_at IS NULL` with `archived = false` at the caller; embedding column resolved via `resolveEmbeddingColumn(undefined, cfg)` from `src/core/search/embedding-column.ts` so Voyage/multimodal/non-default-column brains count against the column they use; errors propagate (no swallow-catch). The sync delete loop is interleaved per-batch resolve+delete using `engine.resolveSlugsByPaths` + `engine.deletePages` from `src/core/engine.ts` (73K-delete commit: ~146K SQL round-trips → ~292, closing the cascade-staleness class where one big-delete commit jammed every other source's sync); per-batch try-catch decomposes batch DELETE failures to per-slug `deletePage` fallback, unrecoverable per-slug failures land in `failedFiles`; `pagesAffected` filters to confirmed-deleted slugs. The rename loop gets the same batched slug-resolve; a rename whose destination is un-syncable folds the source path into the delete set so the old page can't orphan (#1970). An entry-time bookmark-reachability guard distinguishes a gc'd anchor (`cat-file` fails → `performFullSync`) from a history-rewrite anchor that is merely no-longer-an-ancestor but still on disk: the latter is diffed tree-to-tree directly (`git diff lastCommit..pin` is an endpoint-tree compare, ancestry not required) so a force-push / `master`→`main` consolidation imports only the real delta instead of re-walking the whole tree forever (#1970); an oversized or failed diff degrades to `performFullSync`. `performFullSync` is itself authoritative for deletes — after an advancing full import it purges file-backed pages (`source_path != null` AND strategy-aware `isSyncable`) whose source file no longer exists, sparing `put_page`/manual pages (null `source_path`) and metafiles. The stale-file decision routes through the pure, exported `planReconcileDeletes(rows, currentFiles, isSyncablePath)`: it normalizes path separators on both sides of the membership test (a Windows `path.relative` backslash path vs a git-derived forward-slash `source_path` would otherwise mark every page stale and wipe the source) and computes a mass-delete signal — when the reconcile would delete more than `MASS_RECONCILE_RATIO` (50%) of the file-backed pages the strategy manages, on a source holding more than `MASS_RECONCILE_MIN_PAGES` (20) of them, the delete is REFUSED with a loud stderr warning (that shape is almost always a path-comparison bug or the wrong repo path, not a genuine bulk deletion); `GBRAIN_ALLOW_MASS_RECONCILE=1` restores the unguarded delete for genuinely intended bulk removals. Pinned by `test/sync-reconcile-mass-delete.test.ts`. Below the valve, stale pages are partitioned by git history via exported `listEverCommittedPaths(repoPath)` (one `git log --all --no-renames --diff-filter=A --name-only` pass; null on non-git dirs → unchanged behavior): a stale path that EVER existed in history was genuinely deleted → reconciled; a path with NO history is DB-only write-through (never committed/pushed, e.g. lost to a fresh clone) → the page is KEPT and its markdown re-exported to the working tree via `writePageThrough`, with a stderr hint to commit it (#2426 — "absent from git" is the symptom of the missing write-through commit, not evidence the content is disposable). Pinned by `test/sync-reconcile-db-only.serial.test.ts`. `resolveSlugByPathOrSourcePath` at `sync.ts:267` delegates to `engine.resolveSlugsByPaths` when `sourceId` is set, keeping legacy `executeRaw` fallback for the no-sourceId path. `failedFiles` is hoisted to the top of `performSyncInner` so both delete-decompose and import loops feed the same bookmark gate. The cost gate is the shared `runInlineCostGate` (one implementation on BOTH the `--all` and single-source paths; runs at the command layer, never inside `performSync`), mode-aware via `willEmbedSynchronously` + posture-aware `shouldBlockSync` from `src/core/embedding.ts` (#2139). The DEFERRED path (v2 on, parallel) is INFORMATIONAL (embedding goes to per-source `embed-backfill` jobs with their own `$X/source/24h` cap, default $25 via `SPEND_CAP_CONFIG_KEY` from `embed-backfill-submit.ts`; prints the cap + backlog + queued-job count, NEVER exits 2). The INLINE path (v2 off, or `--serial`) gates on the DELTA estimate vs `sync.cost_gate_min_usd` (default $0.50): below floor proceeds; above floor in a TTY prompts `[y/N]`; above floor in a non-TTY/`--json` session AUTO-DEFERS embeds to capped backfill jobs and exits 0 (NEVER exit 2 — the wedged-cron fix); `spend.posture=tokenmax` makes it informational and proceeds inline. The estimate MIRRORS EXECUTION instead of pricing the whole tree: `estimateInlineNewTokens` routes through the shared `computeSyncDelta` (`src/core/sync-delta.ts`) — fetch-first against `origin/`, prices only the committed delta (a dirty-but-caught-up tree → $0), with a full-tree CEILING only on the fail-open rungs (chunker drift, first sync, git-unavailable) honestly labeled; `--full` adds the stale backlog (full sync sweeps it inline). Return shape carries `estimateKind: 'delta'|'ceiling'|'mixed'|'unchanged'` + `ceilingReasons`. Helpers `resolveCostGateFloorUsd(engine)` + `resolveBackfillCapUsd(engine)` resolve via `parseUsdLimit` (`off`/`unlimited` → `Infinity`; floor accepts `0` = block-on-any-spend). JSON envelopes carry `mode` + `gate` discriminators (`dry_run | deferred_notice | below_floor | auto_deferred_embeds | posture_tokenmax`) + a paste-ready `hint`; `Infinity` floors/caps render as the string `'unlimited'` (never raw, which JSON-serializes to `null`); `SyncStatusReportSource` gains `backfill_queued`/`backfill_active`/`backfill_last_completed_at`; cost previews read `getEmbeddingModelName()` (no hardcoded OpenAI). Format splits on the explicit `--json` flag only (human text otherwise). `SyncOpts.noSchemaPack` (CLI `--no-schema-pack`, threaded through `performSync` AND `syncOneSource`) skips `loadActivePack` so pages fall back to legacy prefix typing — an escape hatch when a suspect pack regex wedges a sync. A per-file BEGIN heartbeat `if (process.env.GBRAIN_SYNC_TRACE) serr('[sync] begin import: ')` fires BEFORE `importFile` (the `progress.tick` fires only AFTER) so a stuck file is a begin-line with no matching completion. Triage doc: `docs/architecture/serve-sync-concurrency.md` (PGLite single-writer serve↔sync contention + the `GBRAIN_SYNC_TRACE` + `--no-schema-pack` recipes). Pinned by `test/e2e/sync-status-pglite.test.ts` (IRON-RULE: PGLite seeds 2 sources × pages × chunks, soft-deletes 1 page, archives 1 source, validates the SQL excludes both AND uses the active embedding column), `test/sync-cost-gate.serial.test.ts`, `test/sync-cost-preview.test.ts`. Runaway-sync protection: `resolveSyncHardDeadline(args, {isTty, env, defaultNonTtySec?})` resolves a wall-clock hard deadline (precedence `--no-hard-deadline` > `--hard-deadline ` > `--timeout `(non-`--all`, which auto-arms the backstop) > `GBRAIN_SYNC_MAX_RUNTIME_SECONDS` env > non-TTY default 3600s > none; `HARD_DEADLINE_GRACE_SEC=30`). `src/cli.ts` installs the out-of-band watchdog (see `src/core/process-watchdog.ts`) for the sync command BEFORE `connectEngine` and disposes it in the dispatch `finally`, so even an event-loop-starved sync — or a connect-phase hang — is SIGTERM-then-SIGKILLed by the deadline instead of orphaning under cron. `runSync` registers a SIGINT handler that aborts an interrupt `AbortController` composed via `composeAbortSignals(...)` (an `AbortSignal.any` wrapper over the defined signals) with the per-source `--timeout` signal, so Ctrl-C returns a clean `partial` and releases the lock through the normal `finally` (process-cleanup.ts owns SIGTERM lock-release; the watchdog owns the hard kill). `withRefreshingLock` `unref()`s its refresh `setInterval`. The spin's own root cause is not yet pinned (leading lead is catastrophic-backtracking in a pack link-inference regex, bounded by the redos-guard); the watchdog heartbeat plus the existing `[gbrain phase]` breadcrumbs are the diagnosis surface. Pinned by `test/sync-hard-deadline.test.ts` (resolution precedence + `composeAbortSignals`). Monorepo subdir sources (#753/#774): `--src-subpath ` (or a repo path that IS a subdir — auto-discovery via `discoverGitRoot`, i.e. `git rev-parse --show-toplevel`) splits the repo path into `gitContextRoot` (all git ops: pull/diff/rev-parse/cat-file) and `syncScopeRoot` (walk/import/delete/rename scope); scoped syncs use git-root-relative slugs + `source_path` (full sync threads `slugRoot` into `runImport`) so full and incremental agree; NAV-1/NAV-2 realpath containment rejects `../`-traversal and symlinked scopes resolving outside the repo BEFORE any git op, and a per-file realpath guard (`isPathSafe`) refuses symlink-escape files in the incremental drain and rename reimport (fail-closed into `failedFiles`, so the bookmark can't advance past an escape); the full-sync reconcile is scope-restricted so a scoped sync never sweeps out-of-scope pages. `--exclude ` (repeatable) filters scope-relative paths in both full and incremental paths; exclusion never deletes previously-imported pages (conservative, matching the #1433 metafile posture); an all-excluded run warns loudly (NAV-4). A warn-and-continue internal `git pull` failure (non-timeout class — e.g. a local-path origin rejected by `protocol.file.allow=never`) still falls through to sync the local working tree, but a ZERO-import run after a failed pull returns `partial` with `reason: 'pull_failed'` instead of `up_to_date`: `last_commit` AND the `last_sync_at` heartbeat stay frozen (so doctor `sync_freshness` / `sources status` staleness fires), the single-source CLI exits non-zero, `sync --all` exits non-zero if any source hit it (JSON envelope carries the per-source `reason`), and the autopilot cycle's sync phase maps it to `warn`. Timeout-class partials keep their pre-existing exit-0 / phase-`ok` semantics (they converge on retry; a failing pull does not). Pinned by `test/sync-pull-failed-anchor.serial.test.ts`. `resolveSlugByPathOrSourcePath`: Resolves a slug by `pages.source_path` first (returns the stored slug for frontmatter-fallback pages whose path doesn't derive a slug), then falls back to `resolveSlugForPath(path)`. Threaded into all 4 delete/rename call sites (`performSync`'s un-syncable cleanup at ~:531, deletes at ~:603, rename oldSlug at ~:622). Without this, emoji-only / Thai / Arabic filenames whose slug came from frontmatter would orphan on delete/rename (the delete path would compute the wrong path-derived slug). Best-effort query — pre-migration brains fall through to the legacy path. +- `src/commands/sync.ts` — `gbrain sync` CLI + the `performSync` / `performFullSync` library entrypoints (consumed by the autopilot cycle and the Minion sync handler). `performSync` runs under a writer lock: per-source `gbrain-sync:` whenever `opts.sourceId` is set, wrapped in `withRefreshingLock` from `src/core/db-lock.ts` so long-running sources (250K+ chunks) don't lose the lock at the 30-min TTL mid-run; the bare no-source path uses the SAME refreshing lock (#1794 — it was previously a non-refreshing `tryAcquireDbLock`, stealable mid-run during an incident); `SyncOpts.lockId?: string` is the explicit override. The lock refresh AND its health probe route through the DIRECT session pool so Supavisor transaction-pool exhaustion (`EMAXCONNSESSION`) can't kill renewal; takeover is heartbeat-aware (it will NOT steal a holder whose `last_refreshed_at` is within `GBRAIN_LOCK_STEAL_GRACE_SECONDS`, defending an alive-but-starved holder); the import loop yields the event loop every `GBRAIN_SYNC_YIELD_EVERY` files (`setTimeout(0)`, not `setImmediate` — Bun starves the timers phase) so the refresh `setInterval` heartbeat fires mid-import. This lock-identity invariant prevents a `sync --all` per-source worker racing `sync --source foo` on the global lock from corrupting the same source. `performSync` throws a typed `SyncLockBusyError` when the writer lock is held; the Minion `sync` handler (`src/commands/jobs.ts`) catches it and marks the job SKIPPED (not failed) so a cron/autopilot tick defers to the holder without polluting crash metrics. `performSyncInner` is RESUMABLE (incremental path): it drains a PINNED target commit (`lastCommit..pin`), banking drained file paths via `appendCompleted` (append-only delta into the `op_checkpoint_paths` child table, migration v115 — one row per path, O(delta) not the old O(N²) full-array rewrite), keyed by `syncFingerprint({sourceId, lastCommit})` from `src/core/op-checkpoint.ts` (paths under `op:'sync'`; the pinned target under `op:'sync-target'`), and advances `last_commit`/`last_sync_at` ONLY at full import completion. Checkpoint writes route through the DIRECT session pool + bounded retry so they survive `EMAXCONNSESSION`; the flush cadence is first-file then every `GBRAIN_SYNC_CHECKPOINT_EVERY` (default 1000) files OR `GBRAIN_SYNC_CHECKPOINT_SECONDS` (default 10s), with a race-safe `pendingCheckpointPaths` delta (single-flight swap, re-merge on failure) under parallel workers; a SIGTERM banks the in-flight delta via a no-retry one-shot (`appendCompletedOnce`, ordered before lock release through `registerCleanup`); and sustained flush failure aborts the run with `reason:'checkpoint_unavailable'` after `GBRAIN_SYNC_MAX_CHECKPOINT_FAILURES` consecutive fails rather than importing work it can never bank (every partial/blocked exit logs the banked-file count). A sync killed mid-import banks its progress, leaves the anchor unmoved (the source stays correctly stale to the autopilot scheduler — `last_sync_at` is never bumped on a partial), and the next run `resumeFilter`s the same fixed diff to skip done files. The pin is the checkpoint's stored target when still reachable from HEAD, else live HEAD (a history rewrite / reset re-pins); completion advances to the pin, NOT live HEAD, so commits landing past the pin are a clean next-sync diff — this closes the cross-run staleness window. After import a pin-reachability gate (`git merge-base --is-ancestor pin HEAD`) replaces the old strict head-drift gate: forward commits on top of the pin (e.g. a background enrich process committing to the same repo every ~2 min) are SAFE and no longer block the run; only a real rewrite (pin not an ancestor of HEAD) blocks and discards the checkpoint. A file added in `lastCommit..pin` but gone from disk (deleted by a commit after the pin) is SKIPPED and marked completed, not recorded as a failure. Downstream extract/facts/embed are size-gated: inline only for `totalChanges <= 100`; large syncs defer to the resumable `extract --stale` watermark + `embed --stale`/backfill + the facts cycle phases, so a 44K-page facts/embed pass never re-blocks import convergence (sync convergence == import convergence). Worker engines wrap in try/finally so disconnect always fires; both PGLite-detection sites use `engine.kind === 'pglite'`. CLI accepts `--workers N` (alias `--concurrency N`) validated via `parseWorkers` (explicit bypasses the file-count floor; auto path defers to `autoConcurrency()`). The newest-first descending-lex order uses `sortNewestFirst(addsAndMods)` from `src/core/sort-newest-first.ts` (shared with `gbrain import`). `gbrain sync --all` runs a continuous worker pool: `parseWorkers`-validated `--parallel N` (default `min(sourceCount, --workers, DEFAULT_PARALLEL_SOURCES=4)`), long-lived async workers pulling from a shared FIFO queue (no head-of-line blocking), per-source `withSourcePrefix(src.id, ...)` so every `slog`/`serr` line carries `[]`; `--skip-failed`/`--retry-failed` are scoped per source (`acknowledgeFailures(sourceId)`; `--all` acks every source, single-source acks only its own) and run UNDER parallel — the #1939 failure ledger is per-`(source_id, path)` and serialized through `withLedgerLock`, so the old "not supported under parallel, re-run with --serial" refusal is lifted (#2139, which also removed the forcing-function that pushed recovery syncs to `--serial` and thus armed the inline cost gate); a connection-budget stderr warning fires when `parallel × workers × 2 > 16` (the `× 2 per-file pool` factor: each per-file worker opens its own `PostgresEngine` with `poolSize=2`). Exports `resolveParallelism`, `syncOneSource`, `buildSyncStatusReport`, `printSyncStatusReport`, `SyncStatusReport` back the `gbrain sources status` dashboard. `--json` envelope `{schema_version: 1, sources, parallel, ok_count, error_count, skipped_count}` on stdout; human banners route to stderr via `humanSink` so `jq` parses cleanly. Exit matrix: 0 all ok (sources skipped by `--missing-path skip` count as ok), 1 any error. `--missing-path ` (default fail) handles sources whose `local_path` does not exist on this machine — machine-specific state in a brain-wide table, so a brain registered from several machines fails every foreign source on every run; `skip` classifies them `skipped_missing_path` (⊘ line, envelope entry with `local_path`, excluded from `error_count` and the rc gate) via the exported pure helpers `parseMissingPathMode` + `partitionMissingPathSources`, pinned by `test/sync-all-missing-path.test.ts`; default `fail` stays loud because on a single-machine brain a missing path usually means an unmounted volume. (The non-TTY cost gate no longer exits 2 — it auto-defers; #2139.) The dashboard SQL is `content_chunks ch JOIN pages pg ON pg.id = ch.page_id WHERE pg.deleted_at IS NULL` with `archived = false` at the caller; embedding column resolved via `resolveEmbeddingColumn(undefined, cfg)` from `src/core/search/embedding-column.ts` so Voyage/multimodal/non-default-column brains count against the column they use; errors propagate (no swallow-catch). The sync delete loop is interleaved per-batch resolve+delete using `engine.resolveSlugsByPaths` + `engine.deletePages` from `src/core/engine.ts` (73K-delete commit: ~146K SQL round-trips → ~292, closing the cascade-staleness class where one big-delete commit jammed every other source's sync); per-batch try-catch decomposes batch DELETE failures to per-slug `deletePage` fallback, unrecoverable per-slug failures land in `failedFiles`; `pagesAffected` filters to confirmed-deleted slugs. The rename loop gets the same batched slug-resolve; a rename whose destination is un-syncable folds the source path into the delete set so the old page can't orphan (#1970). An entry-time bookmark-reachability guard distinguishes a gc'd anchor (`cat-file` fails → `performFullSync`) from a history-rewrite anchor that is merely no-longer-an-ancestor but still on disk: the latter is diffed tree-to-tree directly (`git diff lastCommit..pin` is an endpoint-tree compare, ancestry not required) so a force-push / `master`→`main` consolidation imports only the real delta instead of re-walking the whole tree forever (#1970); an oversized or failed diff degrades to `performFullSync`. `performFullSync` is itself authoritative for deletes — after an advancing full import it purges file-backed pages (`source_path != null` AND strategy-aware `isSyncable`) whose source file no longer exists, sparing `put_page`/manual pages (null `source_path`) and metafiles. The stale-file decision routes through the pure, exported `planReconcileDeletes(rows, currentFiles, isSyncablePath)`: it normalizes path separators on both sides of the membership test (a Windows `path.relative` backslash path vs a git-derived forward-slash `source_path` would otherwise mark every page stale and wipe the source) and computes a mass-delete signal — when the reconcile would delete more than `MASS_RECONCILE_RATIO` (50%) of the file-backed pages the strategy manages, on a source holding more than `MASS_RECONCILE_MIN_PAGES` (20) of them, the delete is REFUSED with a loud stderr warning (that shape is almost always a path-comparison bug or the wrong repo path, not a genuine bulk deletion); `GBRAIN_ALLOW_MASS_RECONCILE=1` restores the unguarded delete for genuinely intended bulk removals. Pinned by `test/sync-reconcile-mass-delete.test.ts`. Below the valve, stale pages are partitioned by git history via exported `listEverCommittedPaths(repoPath)` (one `git log --all --no-renames --diff-filter=A --name-only` pass; null on non-git dirs → unchanged behavior): a stale path that EVER existed in history was genuinely deleted → reconciled; a path with NO history is DB-only write-through (never committed/pushed, e.g. lost to a fresh clone) → the page is KEPT and its markdown re-exported to the working tree via `writePageThrough`, with a stderr hint to commit it (#2426 — "absent from git" is the symptom of the missing write-through commit, not evidence the content is disposable). Pinned by `test/sync-reconcile-db-only.serial.test.ts`. `resolveSlugByPathOrSourcePath` at `sync.ts:267` delegates to `engine.resolveSlugsByPaths` when `sourceId` is set, keeping legacy `executeRaw` fallback for the no-sourceId path. `failedFiles` is hoisted to the top of `performSyncInner` so both delete-decompose and import loops feed the same bookmark gate. The cost gate is the shared `runInlineCostGate` (one implementation on BOTH the `--all` and single-source paths; runs at the command layer, never inside `performSync`), mode-aware via `willEmbedSynchronously` + posture-aware `shouldBlockSync` from `src/core/embedding.ts` (#2139). The DEFERRED path (v2 on, parallel) is INFORMATIONAL (embedding goes to per-source `embed-backfill` jobs with their own `$X/source/24h` cap, default $25 via `SPEND_CAP_CONFIG_KEY` from `embed-backfill-submit.ts`; prints the cap + backlog + queued-job count, NEVER exits 2). The INLINE path (v2 off, or `--serial`) gates on the DELTA estimate vs `sync.cost_gate_min_usd` (default $0.50): below floor proceeds; above floor in a TTY prompts `[y/N]`; above floor in a non-TTY/`--json` session AUTO-DEFERS embeds to capped backfill jobs and exits 0 (NEVER exit 2 — the wedged-cron fix); `spend.posture=tokenmax` makes it informational and proceeds inline. The estimate MIRRORS EXECUTION instead of pricing the whole tree: `estimateInlineNewTokens` routes through the shared `computeSyncDelta` (`src/core/sync-delta.ts`) — fetch-first against `origin/`, prices only the committed delta (a dirty-but-caught-up tree → $0), with a full-tree CEILING only on the fail-open rungs (chunker drift, first sync, git-unavailable) honestly labeled; `--full` adds the stale backlog (full sync sweeps it inline). Return shape carries `estimateKind: 'delta'|'ceiling'|'mixed'|'unchanged'` + `ceilingReasons`. Helpers `resolveCostGateFloorUsd(engine)` + `resolveBackfillCapUsd(engine)` resolve via `parseUsdLimit` (`off`/`unlimited` → `Infinity`; floor accepts `0` = block-on-any-spend). JSON envelopes carry `mode` + `gate` discriminators (`dry_run | deferred_notice | below_floor | auto_deferred_embeds | posture_tokenmax`) + a paste-ready `hint`; `Infinity` floors/caps render as the string `'unlimited'` (never raw, which JSON-serializes to `null`); `SyncStatusReportSource` gains `backfill_queued`/`backfill_active`/`backfill_last_completed_at`; cost previews read `getEmbeddingModelName()` (no hardcoded OpenAI). Format splits on the explicit `--json` flag only (human text otherwise). `SyncOpts.noSchemaPack` (CLI `--no-schema-pack`, threaded through `performSync` AND `syncOneSource`) skips `loadActivePack` so pages fall back to legacy prefix typing — an escape hatch when a suspect pack regex wedges a sync. A per-file BEGIN heartbeat `if (process.env.GBRAIN_SYNC_TRACE) serr('[sync] begin import: ')` fires BEFORE `importFile` (the `progress.tick` fires only AFTER) so a stuck file is a begin-line with no matching completion. Triage doc: `docs/architecture/serve-sync-concurrency.md` (PGLite single-writer serve↔sync contention + the `GBRAIN_SYNC_TRACE` + `--no-schema-pack` recipes). Pinned by `test/e2e/sync-status-pglite.test.ts` (IRON-RULE: PGLite seeds 2 sources × pages × chunks, soft-deletes 1 page, archives 1 source, validates the SQL excludes both AND uses the active embedding column), `test/sync-cost-gate.serial.test.ts`, `test/sync-cost-preview.test.ts`. Runaway-sync protection: `resolveSyncHardDeadline(args, {isTty, env, defaultNonTtySec?})` resolves a wall-clock hard deadline (precedence `--no-hard-deadline` > `--hard-deadline ` > `--timeout `(non-`--all`, which auto-arms the backstop) > `GBRAIN_SYNC_MAX_RUNTIME_SECONDS` env > non-TTY default 3600s > none; `HARD_DEADLINE_GRACE_SEC=30`). `src/cli.ts` installs the out-of-band watchdog (see `src/core/process-watchdog.ts`) for the sync command BEFORE `connectEngine` and disposes it in the dispatch `finally`, so even an event-loop-starved sync — or a connect-phase hang — is SIGTERM-then-SIGKILLed by the deadline instead of orphaning under cron. `runSync` registers a SIGINT handler that aborts an interrupt `AbortController` composed via `composeAbortSignals(...)` (an `AbortSignal.any` wrapper over the defined signals) with the per-source `--timeout` signal, so Ctrl-C returns a clean `partial` and releases the lock through the normal `finally` (process-cleanup.ts owns SIGTERM lock-release; the watchdog owns the hard kill). `withRefreshingLock` `unref()`s its refresh `setInterval`. The spin's own root cause is not yet pinned (leading lead is catastrophic-backtracking in a pack link-inference regex, bounded by the redos-guard); the watchdog heartbeat plus the existing `[gbrain phase]` breadcrumbs are the diagnosis surface. Pinned by `test/sync-hard-deadline.test.ts` (resolution precedence + `composeAbortSignals`). Monorepo subdir sources (#753/#774): `--src-subpath ` (or a repo path that IS a subdir — auto-discovery via `discoverGitRoot`, i.e. `git rev-parse --show-toplevel`) splits the repo path into `gitContextRoot` (all git ops: pull/diff/rev-parse/cat-file) and `syncScopeRoot` (walk/import/delete/rename scope); scoped syncs use git-root-relative slugs + `source_path` (full sync threads `slugRoot` into `runImport`) so full and incremental agree; NAV-1/NAV-2 realpath containment rejects `../`-traversal and symlinked scopes resolving outside the repo BEFORE any git op, and a per-file realpath guard (`isPathSafe`) refuses symlink-escape files in the incremental drain and rename reimport (fail-closed into `failedFiles`, so the bookmark can't advance past an escape); the full-sync reconcile is scope-restricted so a scoped sync never sweeps out-of-scope pages. `--exclude ` (repeatable) filters scope-relative paths in both full and incremental paths; exclusion never deletes previously-imported pages (conservative, matching the #1433 metafile posture); an all-excluded run warns loudly (NAV-4). A warn-and-continue internal `git pull` failure (non-timeout class — e.g. a local-path origin rejected by `protocol.file.allow=never`) still falls through to sync the local working tree, but a ZERO-import run after a failed pull returns `partial` with `reason: 'pull_failed'` instead of `up_to_date`: `last_commit` AND the `last_sync_at` heartbeat stay frozen (so doctor `sync_freshness` / `sources status` staleness fires), the single-source CLI exits non-zero, `sync --all` exits non-zero if any source hit it (JSON envelope carries the per-source `reason`), and the autopilot cycle's sync phase maps it to `warn`. Timeout-class partials keep their pre-existing exit-0 / phase-`ok` semantics (they converge on retry; a failing pull does not). Pinned by `test/sync-pull-failed-anchor.serial.test.ts`. `resolveSlugByPathOrSourcePath`: Resolves a slug by `pages.source_path` first (returns the stored slug for frontmatter-fallback pages whose path doesn't derive a slug), then falls back to `resolveSlugForPath(path)`. Threaded into all 4 delete/rename call sites (`performSync`'s un-syncable cleanup at ~:531, deletes at ~:603, rename oldSlug at ~:622). Without this, emoji-only / Thai / Arabic filenames whose slug came from frontmatter would orphan on delete/rename (the delete path would compute the wrong path-derived slug). Best-effort query — pre-migration brains fall through to the legacy path. #2849: above the size gate (totalChanges > 100) the deferred link/timeline extraction is DURABLY QUEUED, not just hinted — the defer branch submits an `extract` Minion job `{stale: true, sourceId?, deferred_commit: pin}` keyed `extract-stale::` (repeat submissions toward the same drained pin coalesce; deliberately NO maxWaiting — an unscoped payload's coalesce filter matches ANY waiting extract job and would silently drop the sweep), `timeout_ms` derived from extract.ts's exported `STALE_TIME_BUDGET_MS` + headroom. The returned row is verified to be a live `{stale:true}` job (waiting/delayed/active) before the log claims "queued"; a finished row occupying the key slot (a prior sweep toward the same pin that completed before this run's pages landed — the checkpoint-resume / blocked-advance re-sync case) triggers a fresh submission under a run-unique key so those pages never strand stale. Submission is best-effort (failure falls back to the hint; pages stay stale + doctor-visible, never mis-stamped). Pinned by `test/sync-deferred-extract-queue.serial.test.ts`. - `src/core/sources-ops.ts` — Multi-source registration + clone-lifecycle ops (`addSource`, `recloneIfMissing`, `defaultCloneDir`, `isOwnedClone`, `unownedHint`). **Reclone-ownership invariant (must-never-violate): gbrain may only delete/re-clone a clone it created, NEVER a user working tree.** `recloneIfMissing` deletes `local_path`, so it gates on `isOwnedClone(src)` and throws a `SourceOpError('unmanaged_path', ...)` BEFORE any filesystem op when ownership is unprovable — fail-closed. Ownership is proven by `config.managed_clone === true` (written by `addSource`'s `--url` path, covering default-location and `--clone-dir` clones) OR `local_path === defaultCloneDir(id)` (back-compat for pre-marker clones, via exact normalized-path equality, symlink-free). A row with `remote_url` + an unowned `local_path` (a user-registered working tree, e.g. `sources add --path`) is refused untouched; re-add with `--url` to regain auto-reclone. The reclone is EXDEV-safe: clone into a SIBLING temp of `local_path` (not the shared `clones/.tmp`, which may sit on a different mount than a `--clone-dir` target), then swap (move old aside → move new in → drop old) so `local_path` is never left missing-and-unrecoverable; on swap failure the original is restored, and if restore fails the error names the `aside` path so it's never reflexively deleted. A TOCTOU re-check re-confirms ownership immediately before the destructive move and rejects a symlink leaf swapped in after the entry check (`symlink_escape`). `unownedHint(src, state)` is the shared recovery message used by both the core error and the `gbrain sync --source` CLI error; `gbrain sources restore` special-cases `unmanaged_path` to print "DB row restored; gbrain syncs this path read-only" instead of the misleading "try sync to recover" guidance. `SourceOpErrorCode` includes `unmanaged_path`. Pinned by `test/sources-ops.test.ts`, `test/sources-resync-recovery.test.ts`. - `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 `tryParseEmbedding(value)` (returns `null` + warns once per process, used by search/rescore paths where availability matters more than strictness). `isUndefinedColumnError(err)` predicate — pattern-matches Postgres SQLSTATE 42703 / "column ... does not exist" with engine-driver shape variation tolerated; replaces bare `catch {}` blocks in `oauth-provider.ts` so genuine errors (lock timeout, network blip, permission denied) propagate while column-missing falls through to the legacy fallback. `validateSourceId(id)` throws on anything outside `^[a-z0-9_-]+$`, used by the per-source disk-layout fix in patterns.ts/synthesize.ts before any `join(brainDir, '.sources', source_id, slug+'.md')` so source_id can't traverse out of brainDir. `rowToSearchResult` projects email `message_id` / `thread_id` metadata and exposes `source_subject` only when a non-empty Message-ID proves the page is an email, so generated page titles never become authoritative email subjects. `rowToPage` populates the required `Page.source_id` from the SELECT projection (`scripts/check-source-id-projection.sh` enforces every projection feeding `rowToPage` includes the column). - `src/core/db.ts` — Connection management, schema initialization. `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 (`setSessionDefaults` kept as a back-compat no-op shim). `connect()` returns `Promise` — `true` iff THIS call created the module singleton, `false` if it joined an existing one; the decision is atomic (no `await` between the `if (sql)` null-check and the synchronous `sql = postgres(...)` assignment), so two concurrent module connects can't both claim creation. `PostgresEngine` stores the return as its `_ownsModuleSingleton` token and only the creating engine may `db.disconnect()` the singleton — a borrower probe engine (lint/doctor config-lift) no-ops its disconnect, so its teardown can't null the connection the long-lived cycle owner is still using (the dream-cycle "connect() has not been called" failure). The module `sql` is only ever nulled by `db.disconnect()` (postgres.js auto-reconnects its own internal pool and never touches our reference). `disconnect()` snapshots + nulls `sql` before awaiting the pool end so a concurrent connect can't join a pool that's already closing. The end routes through `endPoolBounded(pool)` (#1972) — a gbrain-owned `Promise.race` of `pool.end({ timeout: POOL_END_TIMEOUT_SECONDS })` against a hard timer — so a PgBouncer transaction-mode drain that never settles can't hang teardown — the #2084 contract (finishCliTeardown's computed-deadline backstop + flushThenExit's fence-and-grace exit in cli-force-exit.ts) bounds it and delivers pending stdout before exit. `connection-manager.ts` ends its direct + read pools concurrently through the same helper so the per-pool bounds don't stack. @@ -264,7 +264,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/core/minions/attachments.ts` — Attachment validation (path traversal, null byte, oversize, base64, duplicate detection). - `src/commands/agent.ts` — `gbrain agent run [flags]` CLI. Submits `subagent` (or N children + 1 aggregator) under `{allowProtectedSubmit: true}`. Single-entry `--fanout-manifest` short-circuits. Children get `on_child_fail: 'continue'` + `max_stalled: 3`. `--follow` is the default on TTY; streams logs + polls `waitForCompletion` in parallel. Ctrl-C detaches, does not cancel. - `src/commands/agent-logs.ts` — `gbrain agent logs [--follow] [--since]`. Merges JSONL heartbeat audit + `subagent_messages` into a chronological timeline. `parseSince` accepts ISO-8601 or relative (`5m`, `1h`, `2d`). Transcript tail renders only for terminal jobs. -- `src/commands/jobs.ts` — `gbrain jobs` CLI subcommands + `gbrain jobs work` daemon. `case 'work'` wraps `worker.start()` in try/finally and owns engine lifecycle — calls `engine.disconnect()` on shutdown with loud error logging (the worker must not disconnect an engine it doesn't own; pool slots free immediately on shutdown rather than waiting for TCP keepalive). `jobs submit` surfaces the full `MinionJobInput` retry/backoff/timeout/idempotency surface as flags: `--max-stalled`, `--backoff-type fixed|exponential`, `--backoff-delay`, `--backoff-jitter`, `--timeout-ms`, `--idempotency-key`. `jobs smoke --sigkill-rescue` is the SIGKILL-rescue regression guard. `registerBuiltinHandlers` always registers `subagent` + `subagent_aggregator` (no env flag — `ANTHROPIC_API_KEY` is the cost gate, trust is via `PROTECTED_JOB_NAMES`) and loads `GBRAIN_PLUGIN_PATH` plugins at startup with a loud per-plugin line; `shell` handler still gated by `GBRAIN_ALLOW_SHELL_JOBS=1` (RCE surface). The `autopilot-cycle` handler forwards `job.data.phases` to `runCycle`, validated against `ALL_PHASES` from `src/core/cycle.ts` (invalid names filtered; empty/missing falls back to the default cycle); when `source_id` is set it binds `brainDir` to that source's `local_path` (null for a pure-DB source, never the global repo — the #2194/#2227 mixed-scope fix) and checks `isSourceInCooldown` before `runCycle`, returning a no-op `skipped` (not a failure) for a source still in its failure cooldown. The sibling `autopilot-global-maintenance` handler runs the brain-wide `GLOBAL_PHASES` once (no `sourceId`, `pull:false`) and stamps `autopilot.last_global_at` on success. `resolveJobPull` gives both cycle and standalone sync jobs one positive-polarity `pull` contract while preserving queued payloads that still carry the inverse legacy `noPull` key; explicit `pull` wins. The `sync` handler resolves `sourceId` at entry from `sources.local_path` (mirrors `cycle.ts:480`) so multi-source brains read the per-source `last_commit` anchor; concurrency routes through `autoConcurrency()` in `src/core/sync-concurrency.ts` (PGLite stays serial); `noEmbed` default is `true`. `gbrain jobs supervisor status` at `jobs.ts:803-826` consumes `summarizeCrashes()` from `src/core/minions/handlers/supervisor-audit.ts` for parity with `gbrain doctor`: JSON adds `crashes_by_cause: {runtime_error, oom_or_external_kill, unknown, legacy}` + `clean_exits_24h`; human output gains per-cause + clean-exits lines. Pinned by `test/job-pull-policy.test.ts` and 4 source-grep wiring assertions in `test/doctor.test.ts` requiring `crashes_by_cause` + `clean_exits_24h=` in both `doctor.ts` and `jobs.ts`. `gbrain jobs watch` decouples its two output axes: `--json` picks FORMAT (human default, never gated on isTTY), `--follow` picks LOOP (default `isTTY && !json`). Non-TTY with no flags prints ONE human snapshot then exits (clean for subagent/pipe/cron); `--follow` opts into a continuous stream (human plain per tick, or JSONL with `--json`); a TTY with no flags keeps the live ANSI dashboard. Resolution is the pure `resolveWatchMode(opts, isTTY): {json, follow, useAnsiDashboard}` in `src/commands/jobs-watch.ts`; the dispatch wires `--follow`. Pinned by `test/jobs-watch-mode.test.ts` (format×loop matrix incl. the TTY+`--json`-one-shot case) + `test/e2e/non-tty-output.serial.test.ts` (the `cmd 0` with forward progress. - `src/commands/features.ts` — `gbrain features --json --auto-fix`: usage scan + feature adoption salesman. - `src/commands/autopilot.ts` — `gbrain autopilot --install`: self-maintaining brain daemon (sync+extract+embed). Freshness sync jobs always send an explicit positive-polarity `pull` value derived from the source's parsed `remote_url`, so local-only sources skip pull and PGLite JSON-string configs behave like Postgres objects. Consumes `detectTini()` from `src/core/minions/spawn-helpers.ts`, resolved once at startup. Composes a `ChildWorkerSupervisor` instance for spawn-and-respawn (no inline `crashCount`/`startWorker`/`child.on('exit')`); `--max-rss 2048` and `maxCrashes: 5` preserved. `onMaxCrashesExceeded` routes through autopilot's own `shutdown('max_crashes')` so the autopilot lockfile gets cleaned up. `shutdown()` drains via `childSupervisor.killChild('SIGTERM')` + `awaitChildExit(35_000)`. Pinned by `test/autopilot-fanout-wiring.test.ts` and `test/autopilot-supervisor-wiring.test.ts` (6 static-shape guards: composes ChildWorkerSupervisor not legacy names, `--max-rss 2048` in argv, `maxCrashes: 5` literal, shutdown-via-callback, no workerProc reference). tick body invokes `runNightlyQualityProbe` when `cfg.autopilot.nightly_quality_probe.enabled === true` (default OFF — opt-in to protect API spend). NO scheduler-side rate-limit check — `runNightlyQualityProbe`'s internal `shouldRunNightly` (reading the audit JSONL) is the single source of truth. Probe call wrapped in try/catch that logs via `logError` and does NOT bump `consecutiveErrors` (probe failure is informational, never crashes the loop). Default `max_usd` cap = 5. Pinned by `test/autopilot-nightly-probe-wiring.test.ts`. per-source `extract_atoms` auto-drain. Postgres-only block after the freshness fan-out: gated on `autopilot.auto_drain.enabled` (default true) AND `!packDeclaresPhase(engine,'extract_atoms')` (the silent-backlog condition) AND per-source `countExtractAtomsBacklog > threshold` (default 25) AND a daily cap `floor(max_usd_per_day / ~$0.30)`. Enumerates `loadAllSources`. Submits the PROTECTED `extract-atoms-drain` job (`{allowProtectedSubmit:true}`) with a UTC-day time-sloted idempotency key `autopilot-extract-atoms-drain::` (a static key would block the source after the first job completed). `src/core/minions/protected-names.ts` adds `extract-atoms-drain`; `src/commands/jobs.ts` registers the handler (thin wrapper over `runExtractAtomsDrainForSource`, `LockUnavailableError` → `{deferred:true}`); `src/core/config.ts` adds the `autopilot.auto_drain.*` config keys + the `autopilot.` key prefix. Pinned by `test/extract-atoms-drain-handler.test.ts`, `test/autopilot-auto-drain-wiring.test.ts`. federated-brain co-existence + launchd hygiene. (1) `LOCK_PATH` resolves via `gbrainPath('autopilot.lock')` so it honors `GBRAIN_HOME` (two brains can run autopilot simultaneously without lock-stealing); lock file stores PID, startup checks `kill -0 ` before refusing to start (stale lock from a crashed process no longer blocks). (2) exported `classifyReconnectError(err)` returns `'recoverable' | 'unrecoverable'`; unrecoverable causes `process.exit(0)` so launchd backs off instead of looping `config.database_url undefined`. (3) exported pure `generateLaunchdPlist(wrapperPath, home)` sets `ThrottleInterval=300` so launchd respects the exit-0 backoff. Pinned by `test/autopilot-lock-path.test.ts` + `test/autopilot-reconnect-classifier.test.ts`. targeted-submit loop instead of blanket `autopilot-cycle` dispatch. Each tick: cheap `engine.getHealth()` (single SQL count) + `computeRecommendations()`, then route by shape — `score >= 95 AND no plan AND <60min since last full` → sleep; `score >= 95 AND >=60min` → submit `autopilot-cycle` (60-min floor exercises phase-coupling invariants on healthy brains); `plan <= 3 steps AND est <5min` → submit individual handlers; `plan large OR score < 70` → submit full `autopilot-cycle`. The `gbrain-cycle` lock ensures targeted submissions and the full cycle can't run concurrently. `maxWaiting: 1` per submit closes the queue-fan-out vector. - `src/mcp/server.ts` — MCP stdio server (generated from operations). Tool-call handler delegates to `dispatchToolCall` from `src/mcp/dispatch.ts` so stdio + HTTP transports share one validation, context-build, and error-format path. Stdin `'end'` / `'close'` shutdown hooks are skipped when `process.env.MCP_STDIO === '1'` — gateway-piped stdio MCP wrappers (OpenClaw's `bundle-mcp`) pipe the handshake then close their stdin half, which would otherwise kill the server before the first tool call; signal handlers (SIGTERM/SIGINT/SIGHUP) + the parent-process watchdog still cover legitimate disconnects. `src/commands/serve.ts` exposes `ServeOptions.mcpStdio?: boolean` as a test seam so the guard is exercisable without process.env mutation. Pinned by `test/serve-stdio-lifecycle.test.ts`. diff --git a/src/commands/extract.ts b/src/commands/extract.ts index cfb70b384..5caa3d81a 100644 --- a/src/commands/extract.ts +++ b/src/commands/extract.ts @@ -81,8 +81,10 @@ const BATCH_SIZE = 100; const STALE_BATCH_SIZE = Math.max(1, Number(process.env.GBRAIN_EXTRACT_STALE_BATCH) || 25); // v0.42.7: wall-clock budget for one `extract --stale` invocation (default // 30 min). `--catch-up` removes the cap (loops until 0 stale). Mirrors -// embedAllStale's time-budget shape. -const STALE_TIME_BUDGET_MS = Math.max(1000, Number(process.env.GBRAIN_EXTRACT_TIME_BUDGET_MS) || 30 * 60 * 1000); +// embedAllStale's time-budget shape. Exported so the #2849 deferred-sweep +// submitters (sync's size-gate defer branch + the jobs continuation chain) +// derive their job timeout_ms from the SAME budget instead of hardcoding. +export const STALE_TIME_BUDGET_MS = Math.max(1000, Number(process.env.GBRAIN_EXTRACT_TIME_BUDGET_MS) || 30 * 60 * 1000); /** * v0.42.7 (#1696): best-effort extraction stamp for the source-correct write diff --git a/src/commands/jobs.ts b/src/commands/jobs.ts index 9e89f92b4..8490a3dfc 100644 --- a/src/commands/jobs.ts +++ b/src/commands/jobs.ts @@ -1708,7 +1708,44 @@ export async function registerBuiltinHandlers( }); worker.register('extract', async (job) => { - const { runExtractCore } = await import('./extract.ts'); + const { runExtractCore, extractStaleFromDB, STALE_TIME_BUDGET_MS } = await import('./extract.ts'); + // #2849: stale mode — the durable follow-up for extraction deferred by + // performSync's size gate (totalChanges > 100). Runs the same DB-source + // watermark sweep as `gbrain extract --stale`, scoped to the source the + // sync that deferred it was scoped to (job.data.sourceId; absent = + // unscoped, matching what the CLI hint tells a default-brain operator + // to run). The sweep is checkout-less + idempotent, so retries and + // overlapping submissions converge. + if (job.data.stale === true) { + const sourceIdFilter = typeof job.data.sourceId === 'string' ? job.data.sourceId : undefined; + const r = await extractStaleFromDB(engine, { + dryRun: !!job.data.dryRun, + jsonMode: false, + includeFrontmatter: false, + sourceIdFilter, + catchUp: false, + }); + // Internal 30-min budget hit with work remaining → chain a + // continuation job so a very large deferred backlog converges without + // waiting for the next sync. Forward-progress guard (pagesProcessed > + // 0) prevents an infinite chain if the sweep can't advance. + if (!job.data.dryRun && r.staleRemaining > 0 && r.pagesProcessed > 0) { + try { + const queue = new MinionQueue(engine); + // NO maxWaiting: with an unscoped (NULL-sourceId) payload the + // coalesce filter matches ANY waiting 'extract' job and would + // swallow the continuation. Each completed sweep chains at most + // one continuation and the sweep is an idempotent watermark scan, + // so there is no pile-up to guard against. + await queue.add( + 'extract', + { ...job.data, continuation_of: job.id }, + { timeout_ms: STALE_TIME_BUDGET_MS + 5 * 60 * 1000 }, + ); + } catch { /* best-effort: next sync/manual sweep picks up the rest */ } + } + return { stale: true, source_id: sourceIdFilter ?? null, ...r }; + } const mode = (typeof job.data.mode === 'string' && ['links', 'timeline', 'all'].includes(job.data.mode)) ? (job.data.mode as 'links' | 'timeline' | 'all') : 'all'; diff --git a/src/commands/sync.ts b/src/commands/sync.ts index b120ef625..b61543cca 100644 --- a/src/commands/sync.ts +++ b/src/commands/sync.ts @@ -3647,10 +3647,68 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise 100 && pagesAffected.length > 0) { + // #2849: above the size gate the deferred extraction must be DURABLY + // QUEUED, not just hinted. The autopilot cycle's extract phase is + // slug-scoped (an up_to_date follow-up sync hands it an empty + // pagesAffected), so a webhook-driven large sync left + // `links_extracted_at` unstamped FOREVER unless an operator ran + // `gbrain extract --stale` by hand. Submit a source-scoped stale-sweep + // job bound to the consumed commit (idempotency key) so repeated + // webhook deliveries / sync retries of the same commit coalesce onto + // one job. The sweep itself is the watermark scan — it picks up the + // pages this run imported AND any banked across resumed runs. + // Best-effort: queue submission failure falls back to the hint-only + // behavior (the pages stay stale + visible to doctor, never mis-stamped). + let queuedJobId: number | string | null = null; + try { + const { MinionQueue } = await import('../core/minions/queue.ts'); + const { STALE_TIME_BUDGET_MS } = await import('./extract.ts'); + const queue = new MinionQueue(engine); + const payload = { + stale: true, + ...(opts.sourceId ? { sourceId: opts.sourceId } : {}), + reason: 'sync_size_gate', + // Bound to the PIN this run drained to (== headCommit unless resuming + // a stored target), not live HEAD — the sweep covers what we imported. + deferred_commit: pin, + }; + // The stale sweep has its own internal wall-clock budget + // (GBRAIN_EXTRACT_TIME_BUDGET_MS-derived); without an explicit + // timeout_ms the job would inherit the tight null-default and get + // wall-clock-killed mid-sweep (#1737 class). 5-min headroom. + const timeoutMs = STALE_TIME_BUDGET_MS + 5 * 60 * 1000; + // NO maxWaiting here: with an unscoped (NULL-sourceId) payload the + // queue's coalesce filter matches ANY waiting 'extract' job (e.g. a + // remediation-submitted {mode:'links'} row) and returns THAT job — + // silently dropping the sweep while we log "queued". The idempotency + // key alone is the dedup for repeat submissions toward the same pin. + const key = `extract-stale:${opts.sourceId ?? 'default'}:${pin}`; + const isLiveSweep = (j: { status: string; data: Record }): boolean => + j.data?.stale === true && ['waiting', 'delayed', 'active'].includes(j.status); + let job = await queue.add('extract', payload, { idempotency_key: key, timeout_ms: timeoutMs }); + if (!isLiveSweep(job)) { + // The key slot holds a FINISHED row: a prior sweep toward this pin + // that completed BEFORE this run's pages landed (checkpoint-resume / + // blocked-advance re-sync of the same target). Those pages went + // stale after that sweep's watermark pass, so coalescing onto the + // finished row would strand them — queue a fresh sweep under a + // run-unique key. (An 'active' sweep is safe to coalesce onto: its + // end-of-run staleRemaining re-count chains a continuation.) + job = await queue.add('extract', payload, { + idempotency_key: `${key}:${Date.now()}`, + timeout_ms: timeoutMs, + }); + } + // Only claim "queued" once we verified the returned row IS a live + // stale sweep — never trust queue.add's row blind. + if (isLiveSweep(job)) queuedJobId = job.id; + } catch { /* best-effort — hint below still tells the operator */ } slog( - ` Large sync: deferring link/timeline extraction. ` + - `Run 'gbrain extract --stale${opts.sourceId ? ` --source-id ${opts.sourceId}` : ''}' ` + - `(or let the autopilot cycle's extract phase sweep it).`, + ` Large sync: deferring link/timeline extraction` + + (queuedJobId != null + ? ` — queued stale-sweep job #${queuedJobId} (source: ${opts.sourceId ?? 'default'}); a running jobs worker will consume it.` + : `.`) + + ` Run 'gbrain extract --stale${opts.sourceId ? ` --source-id ${opts.sourceId}` : ''}' to extract now.`, ); } if (!opts.noExtract && totalChanges <= 100 && pagesAffected.length > 0) { diff --git a/test/sync-deferred-extract-queue.serial.test.ts b/test/sync-deferred-extract-queue.serial.test.ts new file mode 100644 index 000000000..173834010 --- /dev/null +++ b/test/sync-deferred-extract-queue.serial.test.ts @@ -0,0 +1,253 @@ +/** + * #2849 — a sync above the size gate (totalChanges > 100) must leave link + * extraction DURABLY QUEUED, not just hinted. + * + * PR #2850 fixed the sub-gate case (webhook + trigger submit noExtract:false), + * but above the gate performSync only printed a "deferring link/timeline + * extraction" hint and left `links_extracted_at` unstamped. The autopilot + * cycle's extract phase is slug-scoped (runExtractCore({slugs: + * syncPagesAffected})), so the NEXT cycle's up_to_date sync hands it an empty + * slug list and it processes 0 pages — the staleness is permanent until an + * operator runs `gbrain extract --stale` by hand. + * + * Post-fix, the deferral branch submits a source-scoped `extract` minion job + * with { stale: true }, bound to the consumed commit via idempotency key, and + * the extract handler routes stale jobs through extractStaleFromDB. These + * tests FAIL on master: no job row exists after a >100-file sync, and the + * handler ignores { stale: true } (it would walk a directory instead). + * + * Marked .serial.test.ts — spawns git subprocesses + shares one PGLite engine. + */ + +import { describe, test, expect, beforeAll, afterAll, beforeEach, afterEach } from 'bun:test'; +import { mkdtempSync, writeFileSync, rmSync, mkdirSync } from 'fs'; +import { execSync } from 'child_process'; +import { tmpdir } from 'os'; +import { join } from 'path'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { resetPgliteState } from './helpers/reset-pglite.ts'; + +let engine: PGLiteEngine; +let repoPath: string; +// resetPgliteState truncates the config table, wiping the `version` key +// MinionQueue.ensureSchema gates on — captured once and reseeded per test. +let schemaVersion: string | null = null; + +function git(cmd: string): void { execSync(cmd, { cwd: repoPath, stdio: 'pipe' }); } + +function headCommit(): string { + return execSync('git rev-parse HEAD', { cwd: repoPath }).toString().trim(); +} + +async function stampOf(slug: string): Promise { + const rows = await engine.executeRaw<{ links_extracted_at: string | null }>( + `SELECT links_extracted_at FROM pages WHERE slug = $1 AND source_id = 'default'`, [slug], + ); + return rows[0]?.links_extracted_at ?? null; +} + +async function staleExtractJobs(): Promise; idempotency_key: string | null }>> { + const rows = await engine.executeRaw<{ id: number; name: string; status: string; data: unknown; idempotency_key: string | null }>( + `SELECT id, name, status, data, idempotency_key FROM minion_jobs WHERE name = 'extract'`, + ); + return rows + .map(r => ({ + ...r, + data: (typeof r.data === 'string' ? JSON.parse(r.data) : r.data) as Record, + })) + .filter(r => r.data.stale === true); +} + +/** Seed repo with an initial commit, then add `n` pages that link to people/alice. */ +function writeLinkedPages(n: number): void { + mkdirSync(join(repoPath, 'notes'), { recursive: true }); + for (let i = 0; i < n; i++) { + writeFileSync(join(repoPath, `notes/page-${i}.md`), [ + '---', 'type: concept', `title: Page ${i}`, '---', '', + `[Alice](../people/alice.md) appears in note ${i}.`, + ].join('\n')); + } +} + +describe('#2849 — size-gated sync durably queues the deferred extraction', () => { + beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); + schemaVersion = await engine.getConfig('version'); + }, 60_000); + + afterAll(async () => { + if (engine) await engine.disconnect(); + }, 60_000); + + beforeEach(async () => { + await resetPgliteState(engine); + if (schemaVersion) await engine.setConfig('version', schemaVersion); + await engine.executeRaw(`DELETE FROM minion_jobs`).catch(() => {}); + repoPath = mkdtempSync(join(tmpdir(), 'gbrain-defer-')); + execSync('git init', { cwd: repoPath, stdio: 'pipe' }); + execSync('git config user.email "t@t.com"', { cwd: repoPath, stdio: 'pipe' }); + execSync('git config user.name "T"', { cwd: repoPath, stdio: 'pipe' }); + mkdirSync(join(repoPath, 'people'), { recursive: true }); + writeFileSync(join(repoPath, 'people/alice.md'), [ + '---', 'type: person', 'title: Alice', '---', '', 'Alice is a founder.', + ].join('\n')); + git('git add -A && git commit -m "initial"'); + const { performSync } = await import('../src/commands/sync.ts'); + await performSync(engine, { repoPath, full: true, noPull: true, noEmbed: true }); + }); + + afterEach(() => { + if (repoPath) rmSync(repoPath, { recursive: true, force: true }); + }); + + test('>100-change sync submits a commit-bound, source-scoped stale-extract job', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + writeLinkedPages(101); + git('git add -A && git commit -m "big drop"'); + const result = await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(['synced', 'first_sync']).toContain(result.status); + + // Above the gate: no inline extract, page unstamped (pre-existing, by design). + expect(await stampOf('notes/page-0')).toBeNull(); + + // THE FIX: the deferral is banked as a durable minion job… + const jobs = await staleExtractJobs(); + expect(jobs.length).toBe(1); + // …bound to the consumed commit so webhook redeliveries coalesce… + expect(jobs[0].idempotency_key).toBe(`extract-stale:default:${headCommit()}`); + expect(jobs[0].data.deferred_commit).toBe(headCommit()); + // …and it carries an explicit wall-clock budget covering the sweep. + const rows = await engine.executeRaw<{ timeout_ms: number | null }>( + `SELECT timeout_ms FROM minion_jobs WHERE id = $1`, [jobs[0].id], + ); + expect(rows[0].timeout_ms).toBeGreaterThanOrEqual(30 * 60 * 1000); + }, 120_000); + + test('re-syncing the same commit range coalesces onto the waiting job (exactly one)', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + const baseCommit = headCommit(); + writeLinkedPages(101); + git('git add -A && git commit -m "big drop"'); + await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(await staleExtractJobs()).toHaveLength(1); + // Rewind the anchor and re-drain the SAME range incrementally: the defer + // branch fires again with the same pin, the idempotency fast path hands + // back the still-waiting job, and no second row appears. Exactly 1 — + // `<= 1` would also pass in the coalesce-drop failure state (0 jobs). + // Garble the stored hashes so the re-drain actually re-imports (the + // content_hash short-circuit would otherwise leave pagesAffected empty + // and never reach the defer branch). + await engine.setConfig('sync.last_commit', baseCommit); + await engine.executeRaw(`UPDATE pages SET content_hash = 'stale-test' WHERE slug LIKE 'notes/%'`); + await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + const jobs = await staleExtractJobs(); + expect(jobs.length).toBe(1); + expect(jobs[0].status).toBe('waiting'); + }, 120_000); + + test('an unrelated waiting extract job does NOT swallow the stale sweep', async () => { + // Blocker-1 regression (#3561 review): the original submission used + // maxWaiting: 1, whose (name, queue, NULL-sourceId) coalesce filter + // matches ANY waiting 'extract' row — a remediation-submitted + // {mode:'links'} job made queue.add return THAT row and the stale sweep + // was silently dropped while the log claimed "queued". + const { MinionQueue } = await import('../src/core/minions/queue.ts'); + const queue = new MinionQueue(engine); + await queue.add('extract', { mode: 'links' }); + const { performSync } = await import('../src/commands/sync.ts'); + writeLinkedPages(101); + git('git add -A && git commit -m "big drop"'); + await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + const jobs = await staleExtractJobs(); + expect(jobs.length).toBe(1); + expect(jobs[0].status).toBe('waiting'); + expect(jobs[0].idempotency_key).toBe(`extract-stale:default:${headCommit()}`); + }, 120_000); + + test('a completed sweep for the same pin does not strand a re-synced range — a fresh job is queued', async () => { + // Blocker-3 regression (#3561 review): the idempotency fast path returns + // a COMPLETED row as-is. A re-sync of the same range (checkpoint-resume / + // blocked-advance) re-imports pages AFTER that sweep's watermark pass, so + // coalescing onto the finished row leaves them stale forever. The defer + // branch must detect the non-live row and queue a fresh sweep under a + // run-unique key. + const { performSync } = await import('../src/commands/sync.ts'); + const baseCommit = headCommit(); + writeLinkedPages(101); + git('git add -A && git commit -m "big drop"'); + await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + const [first] = await staleExtractJobs(); + await engine.executeRaw(`UPDATE minion_jobs SET status = 'completed' WHERE id = $1`, [first.id]); + // Rewind and re-drain the same range (garbled hashes force the + // re-import, mirroring a failed-file retry / checkpoint-resume drain): + // pages' updated_at moves past the "completed" sweep, so a live sweep + // must exist afterwards. + await engine.setConfig('sync.last_commit', baseCommit); + await engine.executeRaw(`UPDATE pages SET content_hash = 'stale-test' WHERE slug LIKE 'notes/%'`); + await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + const jobs = await staleExtractJobs(); + const waiting = jobs.filter(j => j.status === 'waiting'); + expect(waiting.length).toBe(1); + // Run-unique key: base key + suffix, never a bare collision with the old row. + expect(waiting[0].idempotency_key).toStartWith(`extract-stale:default:${headCommit()}:`); + }, 120_000); + + test('sub-gate sync does NOT queue a stale-extract job (inline extract still owns it)', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + writeLinkedPages(3); + git('git add -A && git commit -m "small drop"'); + await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(await staleExtractJobs()).toHaveLength(0); + // Inline path stamped the pages (the #1696 contract, unchanged). + expect(await stampOf('notes/page-0')).not.toBeNull(); + }, 120_000); + + test('--no-extract suppresses the queued job too', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + writeLinkedPages(101); + git('git add -A && git commit -m "big drop"'); + await performSync(engine, { repoPath, noPull: true, noEmbed: true, noExtract: true }); + expect(await staleExtractJobs()).toHaveLength(0); + }, 120_000); + + test('the extract handler consumes { stale: true } jobs via extractStaleFromDB and stamps the pages', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + writeLinkedPages(101); + git('git add -A && git commit -m "big drop"'); + await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(await stampOf('notes/page-7')).toBeNull(); + expect(await engine.getLinks('notes/page-7')).toHaveLength(0); + + // Run the registered handler exactly as a jobs worker would. + const { MinionWorker } = await import('../src/core/minions/worker.ts'); + const { registerBuiltinHandlers } = await import('../src/commands/jobs.ts'); + const worker = new MinionWorker(engine, { queue: 'test' }); + await registerBuiltinHandlers(worker, engine); + const handler = (worker as unknown as { handlers: Map Promise> }) + .handlers.get('extract'); + expect(handler).toBeDefined(); + + const [job] = await staleExtractJobs(); + const result = await handler!({ + id: job.id, + name: 'extract', + data: job.data, + updateProgress: async () => {}, + signal: { aborted: false }, + }) as { stale: boolean; pagesProcessed: number; staleRemaining: number }; + + // Behavioral discrimination vs master: master's handler ignores + // { stale: true } and dir-walks instead — it never stamps the watermark + // and returns a runExtractCore shape without `stale`. + expect(result.stale).toBe(true); + expect(result.pagesProcessed).toBeGreaterThanOrEqual(101); + expect(result.staleRemaining).toBe(0); + + // The deferred work actually converged: links exist + watermark stamped. + const links = await engine.getLinks('notes/page-7'); + expect(links.some(l => l.to_slug === 'people/alice')).toBe(true); + expect(await stampOf('notes/page-7')).not.toBeNull(); + }, 180_000); +});