From 86555e39bbd365c3e9138c3712b40af35047ad53 Mon Sep 17 00:00:00 2001 From: Ziyang Guo <121015044+RerankerGuo@users.noreply.github.com> Date: Sat, 8 Aug 2026 20:03:36 +0800 Subject: [PATCH] fix(autopilot): honor source pull policy (#3861) Fixes #3835 Use a canonical positive pull flag for sync and autopilot-cycle jobs while preserving legacy noPull payloads. Parse source config consistently so local-only sources skip pulls and PGLite remote sources still pull. Test: bun test test/job-pull-policy.test.ts test/autopilot-fanout.test.ts test/autopilot-fanout-wiring.test.ts test/sources-load.test.ts --- docs/architecture/KEY_FILES.md | 6 +++--- src/commands/autopilot-fanout.ts | 9 +++++---- src/commands/autopilot.ts | 3 ++- src/commands/jobs.ts | 15 ++++++++++++--- src/core/sources-load.ts | 6 ++++++ test/autopilot-fanout-wiring.test.ts | 7 +++++++ test/autopilot-fanout.test.ts | 8 ++++++++ test/job-pull-policy.test.ts | 25 +++++++++++++++++++++++++ 8 files changed, 68 insertions(+), 11 deletions(-) create mode 100644 test/job-pull-policy.test.ts diff --git a/docs/architecture/KEY_FILES.md b/docs/architecture/KEY_FILES.md index 5f086270c..95a8ebb4c 100644 --- a/docs/architecture/KEY_FILES.md +++ b/docs/architecture/KEY_FILES.md @@ -208,7 +208,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/commands/graph-query.ts` — `gbrain graph-query [--type T] [--depth N] [--direction in|out|both] [--include-foreign]`: typed-edge relationship traversal (renders indented tree). Foreign-edge footer always present (`X foreign edges (use --include-foreign to traverse)`) so cross-source edges never disappear silently; `--include-foreign` widens the SQL filter to walk them. Pinned by `test/graph-query.test.ts`. - `src/commands/sources.ts` — `gbrain sources {list,add,remove,archive,restore,archived,purge,current,status,audit}`. `current [--json]` calls `resolveSourceWithTier()` and prints `source_id`, `tier` (`flag | env | dotfile | local_path | brain_default | seed_default`), and optional `detail` (decision table in `skills/conventions/brain-routing.md`). `status [--json]` — read-only per-source dashboard (last sync, staleness, page count, embedding coverage, unacked failures); thin wrapper around `buildSyncStatusReport` + `printSyncStatusReport` from `src/commands/sync.ts`; `--json` emits stable `{schema_version: 1, sources, ...}` on stdout; filters input to `local_path IS NOT NULL AND archived IS NOT TRUE`. `audit [--json]` — read-only dry-run disk scan for size distribution + would-blocks + junk-pattern hits WITHOUT touching the DB; walks `sources.local_path`, reads each markdown file, runs `assessContent()` from `src/core/content-sanity.ts`, aggregates by verdict (`ok | warn_oversize | hard_block_junk_pattern`). The live `runStatus` health table gains a `BACKFILL` column between `EMBED` and `FAILS` (`active(N)` beats `queued(N)` beats `idle`, from `SourceMetrics.backfill_active` / `backfill_queued` in `src/core/source-health.ts`) so operators see deferred `embed-backfill` minion work after `sync --all` exits 0; `jobCountsBySource` in `source-health.ts` widens its `minion_jobs` SQL with two `COUNT(*) FILTER (WHERE name = 'embed-backfill' AND ...)` aggregates (best-effort, all-0 on pre-minions brains). Pinned by `test/content-sanity.test.ts`, `test/import-file-content-sanity.test.ts`, `test/source-health.test.ts`. - `src/commands/reindex-frontmatter.ts` — `gbrain reindex-frontmatter`. `reindexFrontmatterCli(engine, args)` takes the ALREADY-CONNECTED engine from cli.ts's dispatch (#1963); it must never build/connect its own engine — a second connect on the same PGLite data dir self-deadlocks on the data-dir lock (this process already holds it) and timed out 100% of the time on PGLite. Same rule applies to `runBackfillCommand(engine, args)` in `src/commands/backfill.ts` and any future command dispatched from cli.ts's engine-connected switch. Pinned by `test/reindex-frontmatter-connect.test.ts` (library path) and `test/reindex-frontmatter-pglite-spawn.serial.test.ts` (CLI dispatch seam, both commands). -- `src/core/source-config-sql.ts` + `src/core/sources-load.ts` — canonical recovery for historical non-object `sources.config` values. The application reader unwraps nested JSON strings and merges recoverable array fragments left-to-right; the shared SQL expression mirrors that policy atomically for both engines, source config updates, archive/restore, and the paste-ready `source_config_shape` doctor repair. `localFederatedSourceIds` reads config through the same parser so stdio/CLI federation cannot silently disagree with `sources list`. Invalid fragments degrade to `{}` rather than throwing. Pinned by `test/sources-load.test.ts`, `test/list-all-sources.test.ts`, `test/local-federated-search-scope.test.ts`, `test/destructive-guard.test.ts`, and `test/doctor-source-config-shape.test.ts`. +- `src/core/source-config-sql.ts` + `src/core/sources-load.ts` — canonical recovery for historical non-object `sources.config` values. The application reader unwraps nested JSON strings and merges recoverable array fragments left-to-right; the shared SQL expression mirrors that policy atomically for both engines, source config updates, archive/restore, and the paste-ready `source_config_shape` doctor repair. `localFederatedSourceIds` reads config through the same parser so stdio/CLI federation cannot silently disagree with `sources list`. `sourceConfigHasRemoteUrl` uses that parser for autopilot pull policy, including PGLite's JSON-string config shape. Invalid fragments degrade to `{}` rather than throwing. Pinned by `test/sources-load.test.ts`, `test/job-pull-policy.test.ts`, `test/list-all-sources.test.ts`, `test/local-federated-search-scope.test.ts`, `test/destructive-guard.test.ts`, and `test/doctor-source-config-shape.test.ts`. - `src/core/source-resolver.ts` — 6-tier source resolution. `resolveSourceWithTier(engine, explicit, cwd)` returns `{ source_id, tier: SourceTier, detail? }` alongside `resolveSourceId()` (unchanged). `SOURCE_TIER_NAMES = ['flag', 'env', 'dotfile', 'local_path', 'sole_non_default', 'brain_default', 'seed_default']` (7 entries; order matches priority). Tier `sole_non_default` slots between `local_path` and `brain_default`: when NO `sources.default` config is set AND exactly one registered source has `local_path` AND isn't `'default'`, auto-route to it; archived sources excluded (try/catch for pre-v34 brains); private `pickSoleNonDefaultSource(engine)` shared by both resolver entry points so they cannot drift. Exported `formatSoleNonDefaultNudge(sourceId): string | null` builds the user-facing stderr nudge (null when `GBRAIN_NO_SOLE_NON_DEFAULT_NUDGE=1`). `src/commands/sync.ts:1497-1519` calls `resolveSourceWithTier` unconditionally so the tier fires; `src/commands/import.ts:96-128` mirrors with the tier-gated nudge. Consumed by `gbrain sources current`, `import --source-id`, `extract --source-id`, and the `source_routing_health` doctor check. Pinned by `test/source-resolver-with-tier.test.ts` (`withEnv()` per test-isolation lint), `test/source-resolver-sole-non-default.test.ts` (14 cases), `test/sync-sole-non-default-routing.test.ts` (3 PGLite cases driving real `runSync`). - `src/core/sync.ts` extension — `isSyncable` factored through private `classifySync(path, opts): SyncableReason | null`; exported companion `unsyncableReason(path, opts)` returns the same tagged reason or null when syncable. `SYNC_SKIP_FILES` is a named export (the four canonical metafile basenames `schema.md`, `index.md`, `log.md`, `README.md`). `SyncableReason` union: `'metafile' | 'strategy' | 'pruned-dir' | 'include-glob-miss' | 'exclude-glob-hit'`. The `commands/sync.ts` cleanup loop guards on `unsyncableReason(path)` being `'metafile'` OR `'pruned-dir'` (#2404) so previously-indexed metafile pages AND deliberately put-created pages under pruned dirs survive every re-sync. Does NOT cover `manifest.deleted` (the upstream filter already strips metafiles). Pinned by `test/sync-isSyncable-shape.test.ts` (15 cases, duality contract) + `test/sync-metafile-skip.serial.test.ts` (3 PGLite cases incl. the renamed `.md → .txt` negative). - `src/core/import-file.ts` extension — identity-based dedup pre-check at `:427-490`. Calls `engine.findDuplicatePage?.(sourceId, {hash, frontmatterId})` (optional `?` so test doubles compile). Posture: SKIP when `frontmatter.id` matches (true external duplicate from overlapping ingest roots), WARN-ALWAYS on content_hash collision with different/missing `frontmatter.id` (templates and daily logs may legitimately share text), FAIL CLOSED on lookup error, bypass via `--force-rechunk`. Soft-deleted pages excluded at the engine layer so tombstones don't block legitimate re-imports under new slugs. Pinned by `test/import-dedup-frontmatter-id.test.ts` (11 cases). @@ -273,9 +273,9 @@ 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. 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 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 ): boolean { + if (typeof data.pull === 'boolean') return data.pull; + if (typeof data.noPull === 'boolean') return !data.noPull; + return true; +} + /** * Long-lived workers outlive operator config changes. Re-stamp the AI gateway * from DB-backed model config immediately before queued jobs enter gateway-backed @@ -1414,7 +1424,7 @@ export async function registerBuiltinHandlers( worker.register('sync', async (job) => { const { performSync } = await import('./sync.ts'); const repoPath = typeof job.data.repoPath === 'string' ? job.data.repoPath : undefined; - const noPull = !!job.data.noPull; + const noPull = !resolveJobPull(job.data); // noEmbed defaults to true (embed is a separate job — submit `embed --stale` // after sync, OR run via the autopilot cycle which has its own embed phase). // Caller can opt in by passing { noEmbed: false } in job params. @@ -1855,8 +1865,7 @@ export async function registerBuiltinHandlers( ? (job.data.phases as string[]).filter(p => validPhases.has(p as any)) : undefined; - // Pull default: legacy `true` for back-compat; explicit boolean wins. - const pull = typeof job.data.pull === 'boolean' ? job.data.pull : true; + const pull = resolveJobPull(job.data); // #2194 fix #2 / codex #5 (D4): claim-time cooldown guard. A job already // queued or retrying (max_attempts:2) can reach the worker after the diff --git a/src/core/sources-load.ts b/src/core/sources-load.ts index 3c3b3aa9a..a4bd92428 100644 --- a/src/core/sources-load.ts +++ b/src/core/sources-load.ts @@ -148,6 +148,12 @@ export function parseSourceConfig(config: unknown): Record { return value ?? {}; } +/** True iff config declares a non-empty remote URL. */ +export function sourceConfigHasRemoteUrl(config: unknown): boolean { + const remoteUrl = parseSourceConfig(config).remote_url; + return typeof remoteUrl === 'string' && remoteUrl.trim().length > 0; +} + /** True iff the source's config.federated field is the literal boolean true. */ export function isSourceFederated(config: unknown): boolean { const parsed = parseSourceConfig(config); diff --git a/test/autopilot-fanout-wiring.test.ts b/test/autopilot-fanout-wiring.test.ts index 8bee073a7..21b2a14bb 100644 --- a/test/autopilot-fanout-wiring.test.ts +++ b/test/autopilot-fanout-wiring.test.ts @@ -73,6 +73,13 @@ describe('autopilot.ts ↔ dispatchPerSource wiring', () => { ); }); + test('freshness sync dispatch uses the parsed source config for pull policy', () => { + const freshnessIdx = AUTOPILOT_SRC.indexOf('idempotency_key: `autopilot-sync:'); + expect(freshnessIdx).toBeGreaterThan(-1); + const freshnessBlock = AUTOPILOT_SRC.slice(Math.max(0, freshnessIdx - 700), freshnessIdx + 200); + expect(freshnessBlock).toContain('pull: sourceConfigHasRemoteUrl(src.config)'); + }); + test('#2781: dispatchGlobalMaintenance gets the full-cycle floor, not the outer (non-full-cycle) timeoutMs', () => { // Live #2781 regression, found in review: dispatchGlobalMaintenance's // call used the object-shorthand `timeoutMs`, which resolved to the diff --git a/test/autopilot-fanout.test.ts b/test/autopilot-fanout.test.ts index 2e4fee561..d38aee88a 100644 --- a/test/autopilot-fanout.test.ts +++ b/test/autopilot-fanout.test.ts @@ -234,6 +234,14 @@ describe('dispatchPerSource — integration with stubbed engine + queue', () => expect((byId.get('local')!.data as Record).pull).toBe(false); }); + test('pull: true when PGLite returns source.config as a JSON string', async () => { + const remote = src('remote'); + remote.config = '{"remote_url":"https://github.com/x/y"}' as unknown as SourceRow['config']; + const { engine, queue, added, fanoutOpts } = makeStubs([remote]); + await dispatchPerSource(engine, queue, fanoutOpts); + expect((added[0].data as Record).pull).toBe(true); + }); + test('fanoutMax cap: 3 sources, fanoutMax=1, 1 dispatched + 2 in skippedCap', async () => { const { engine, queue, added, fanoutOpts } = makeStubs([src('a'), src('b'), src('c')]); fanoutOpts.fanoutMax = 1; diff --git a/test/job-pull-policy.test.ts b/test/job-pull-policy.test.ts new file mode 100644 index 000000000..1739946c4 --- /dev/null +++ b/test/job-pull-policy.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, test } from 'bun:test'; +import { resolveJobPull } from '../src/commands/jobs.ts'; +import { sourceConfigHasRemoteUrl } from '../src/core/sources-load.ts'; + +describe('job pull policy', () => { + test('canonical pull takes precedence over the legacy noPull key', () => { + expect(resolveJobPull({ pull: false })).toBe(false); + expect(resolveJobPull({ pull: true })).toBe(true); + expect(resolveJobPull({ pull: false, noPull: false })).toBe(false); + expect(resolveJobPull({ pull: true, noPull: true })).toBe(true); + }); + + test('legacy noPull remains backward compatible', () => { + expect(resolveJobPull({ noPull: true })).toBe(false); + expect(resolveJobPull({ noPull: false })).toBe(true); + expect(resolveJobPull({})).toBe(true); + }); + + test('remote_url detection handles object and PGLite string configs', () => { + expect(sourceConfigHasRemoteUrl({ remote_url: 'https://example.invalid/brain.git' })).toBe(true); + expect(sourceConfigHasRemoteUrl('{"remote_url":"https://example.invalid/brain.git"}')).toBe(true); + expect(sourceConfigHasRemoteUrl({ remote_url: ' ' })).toBe(false); + expect(sourceConfigHasRemoteUrl('{}')).toBe(false); + }); +});