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
This commit is contained in:
Ziyang Guo
2026-08-08 19:03:36 +07:00
committed by GitHub
parent 3257758492
commit 86555e39bb
8 changed files with 68 additions and 11 deletions
+3 -3
View File
@@ -208,7 +208,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this
- `src/commands/graph-query.ts``gbrain graph-query <slug> [--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 <id> [--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 <prompt> [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 <job> [--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 </dev/null` non-empty-stdout contract).
- `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 </dev/null` non-empty-stdout contract).
- `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). 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-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).
- `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).
- `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`.
- `src/mcp/dispatch.ts` — shared tool-call dispatch consumed by both stdio (`server.ts`) and HTTP transports. Exports `dispatchToolCall(engine, name, params, opts)`, `buildOperationContext(engine, params, opts)`, `validateParams(op, params)`. Single source of truth for `(ctx, params)` handler arg order and the 5-field `OperationContext` shape (engine + config + logger + dryRun + remote). Defaults `remote: true` (untrusted); local CLI callers pass `remote: false`. Also exports `summarizeMcpParams(opName, params)` — privacy-preserving redactor for `mcp_request_log` and the admin SSE feed, returns `{redacted, kind, declared_keys, unknown_key_count, approx_bytes}`. Intersects submitted top-level keys against the operation's declared `params` allow-list (declared keys preserved sorted; unknown keys counted but never named, closing the attacker-controlled-key-name leak). Byte counts bucketed up to nearest 1KB so an attacker can't binary-search secret-content sizes by probing. Raw payload visibility is opt-in via `gbrain serve --http --log-full-params` (loud stderr warning). New logging paths route through this helper, not `JSON.stringify(params)`.
- `src/mcp/rate-limit.ts` — Bounded-LRU token-bucket limiter. `buildDefaultLimiters()` returns the two-bucket pipeline: pre-auth IP (30/60s, fires BEFORE the DB lookup so brute-force load against `access_tokens` is capped) + post-auth token-id (60/60s). Tracks `lastTouchedMs` separately from `lastRefillMs` so an exhausted key can't be reset by hammering past the TTL. LRU cap bounds memory under attacker-controlled key growth.
+5 -4
View File
@@ -33,6 +33,7 @@
import type { BrainEngine, SourceRow } from '../core/engine.ts';
import type { MinionQueue } from '../core/minions/queue.ts';
import { NON_GLOBAL_PHASES, GLOBAL_PHASES, LAST_GLOBAL_AT_KEY } from '../core/cycle.ts';
import { sourceConfigHasRemoteUrl } from '../core/sources-load.ts';
const FULL_CYCLE_FLOOR_MIN = 60;
@@ -430,13 +431,13 @@ export async function dispatchPerSource(
const dispatched: string[] = [];
for (const src of dispatch) {
try {
const remoteUrl = typeof src.config?.remote_url === 'string' ? src.config.remote_url : null;
const shouldPull = sourceConfigHasRemoteUrl(src.config);
const job = await queue.add(
'autopilot-cycle',
{
repoPath: opts.repoPath,
source_id: src.id,
pull: !!remoteUrl,
pull: shouldPull,
// #2194 fix #3 (cycle split): per-source cycles run ONLY source-scoped
// (+ mixed) phases. The brain-wide global phases (embed, orphans,
// purge, …) run once in autopilot-global-maintenance, not N times
@@ -465,11 +466,11 @@ export async function dispatchPerSource(
job_id: job.id,
mode: 'per_source',
source_id: src.id,
pull: !!remoteUrl,
pull: shouldPull,
slot: opts.slot,
}));
} else {
log(`[dispatch] job #${job.id} autopilot-cycle source=${src.id}${remoteUrl ? ' pull=yes' : ''}`);
log(`[dispatch] job #${job.id} autopilot-cycle source=${src.id}${shouldPull ? ' pull=yes' : ''}`);
}
} catch (e) {
// Per-source submit failure does NOT abort the tick (codex E1 F1
+2 -1
View File
@@ -739,7 +739,7 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
try {
const { isFederatedV2Enabled } = await import('../core/feature-flags.ts');
if (await isFederatedV2Enabled(engine)) {
const { loadAllSources } = await import('../core/sources-load.ts');
const { loadAllSources, sourceConfigHasRemoteUrl } = await import('../core/sources-load.ts');
const sources = await loadAllSources(engine);
const intervalMs = baseInterval * 1000;
const now = Date.now();
@@ -754,6 +754,7 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
{
sourceId: src.id,
repoPath: src.local_path,
pull: sourceConfigHasRemoteUrl(src.config),
auto_embed_backfill: true,
embed_reason: 'autopilot_freshness',
},
+12 -3
View File
@@ -22,6 +22,16 @@ function hasFlag(args: string[], flag: string): boolean {
return args.includes(flag);
}
/**
* Resolve the canonical positive-polarity pull flag while preserving queued
* jobs that still carry the legacy inverse `noPull` key.
*/
export function resolveJobPull(data: Record<string, unknown>): 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
+6
View File
@@ -148,6 +148,12 @@ export function parseSourceConfig(config: unknown): Record<string, unknown> {
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);
+7
View File
@@ -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
+8
View File
@@ -234,6 +234,14 @@ describe('dispatchPerSource — integration with stubbed engine + queue', () =>
expect((byId.get('local')!.data as Record<string, unknown>).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<string, unknown>).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;
+25
View File
@@ -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);
});
});