From 638dd0d24763e320767c463c61060e39ebda438d Mon Sep 17 00:00:00 2001 From: arisgysel-design Date: Thu, 13 Aug 2026 14:44:16 +0200 Subject: [PATCH] fix(autopilot): reopen targeted remediation slots (#4046) (#4067) Co-authored-by: arisgysel-design --- docs/TESTING.md | 1 + docs/architecture/KEY_FILES.md | 2 +- src/commands/autopilot-fanout.ts | 32 ++++++-- src/commands/autopilot-remediation-policy.ts | 46 ++++++++++++ src/commands/autopilot.ts | 39 ++++++---- test/autopilot-fanout-wiring.test.ts | 12 ++- test/autopilot-fanout.test.ts | 78 ++++++++++++++++++++ 7 files changed, 188 insertions(+), 22 deletions(-) create mode 100644 src/commands/autopilot-remediation-policy.ts diff --git a/docs/TESTING.md b/docs/TESTING.md index c8f8319b6..f7debdb51 100644 --- a/docs/TESTING.md +++ b/docs/TESTING.md @@ -176,6 +176,7 @@ Unit tests and what they cover: - `test/watch-command.test.ts` — `gbrain watch` push transport (#2095): streaming loop, rolling window, session dedupe, `--json` JSONL shape, `channel: 'watch'` event logging, clean EOF return. Hermetic PGLite + injected line/write deps (no subprocess, no real stdin). - `test/watch-sigint.serial.test.ts` — `gbrain watch` SIGINT lifecycle against a real spawned CLI subprocess with a tmpdir brain. SERIAL: parallel unit shards flake on concurrent subprocess spawns (same rationale as `apply-migrations-pglite-spawn.serial.test.ts`). - `test/autopilot-launchd-lifecycle.serial.test.ts` — autopilot lifecycle behavior, not generated-string assertions: the full install → self-disable → status → reinstall → uninstall arc with `launchctl` replaced by an argv recorder and the generated wrapper executed by a REAL bash against a genuinely deleted repo (every platform), plus a darwin-only fail-SKIP describe against the real launchd under a per-run unique label (`GBRAIN_AUTOPILOT_LABEL`) so it can never collide with — or tear down — a real install on the host. Serial: spawns subprocesses and pins HOME/GBRAIN_HOME for the whole file. +- `test/autopilot-fanout.test.ts` — Autopilot fan-out and #4046 policy regression: targeted idempotency keys reopen per dispatch interval while stable doctor/remediate keys remain unchanged; the 60-minute full-cycle floor wins with a remaining small plan, and an all-fresh restart check advances the process-local clock without masking failed stale-source submissions. - `test/agent-scheduler-contract.serial.test.ts` — the documented external agent-scheduler shell chain (`gbrain sync --repo X && gbrain embed --stale`, live-sync.md / INSTALL_FOR_AGENTS.md Step 7) driven end-to-end through a real `/bin/sh` against a keyless PGLite brain: the `&&` short-circuit IS the contract (argv arrays can't exercise it), the keyless bare stale embed exits 0, and the pull-failure case that must break the chain does. Anti-vacuity: the fixture commits a real page and every read-back asserts pages >= 1. Serial: real spawned CLI + tmpdir HOME. - `test/cli-format-volunteer.test.ts` — `formatResult`'s `volunteer_context` human rendering: pointer lines with confidence/arm/rationale, the empty-result message, the approximate stats summary. - `test/config.test.ts` — config redaction. diff --git a/docs/architecture/KEY_FILES.md b/docs/architecture/KEY_FILES.md index 21342f776..0646e86db 100644 --- a/docs/architecture/KEY_FILES.md +++ b/docs/architecture/KEY_FILES.md @@ -268,7 +268,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `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 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. `--status` is heartbeat-truth, not artifact-presence, and is dispatched ENGINE-FREE in cli.ts before connectEngine (`runAutopilotStatus`; a live PGLite daemon's exclusive DB lock or a DB outage must not take down the alarm — `--uninstall` rides the same pre-engine dispatch): pure `classifyAutopilotStatus` maps `{installed, disabledReason, pausedReason, heartbeatAgeSeconds, intervalSeconds}` → `not_installed`/`disabled`/`paused`/`never_run`/`stale`/`fresh` with exit codes 0/1/2 via `autopilotStatusExitCode` (disabled=2; paused/stale/never_run=1). Stale tolerance = 6×interval — the adaptive scheduler sleeps 2×interval on healthy brains, so 3× flapped false alarms; a non-finite or non-positive interval falls back to 1800s at BOTH the dispatch parse and the pure layer (NaN staleAfter would read a dead daemon as fresh), and the daemon's own interval parse carries the same guard (NaN reached setTimeout as 0 = busy-loop). `paused` outranks the heartbeat states because the tick loop refreshes its heartbeat BEFORE honoring the pause marker; the tick loop also SELF-CLEARS a migrate-signed marker whose recorded pid is dead (SIGKILL orphans heal within one poll) and exits-for-relaunch when the file-plane engine identity diverges from boot (`autopilotEngineIdentity`; null/torn reads are skipped, so a concurrent config write cannot restart it). `detectInstalledTarget()` probes plist/systemd-unit/start-script/crontab, where the crontab probe uses pure `crontabIndicatesAutopilotInstall` (a cron'd status-monitor line must not read as an install). The generated wrapper embeds `generateSelfDisableGuard(repo, target)`: repo-dir test (never `.git` — a file in worktrees), THREE consecutive misses required before disabling (strike counter file; one absent external/cloud volume at login must not permanently kill the install), marker write BEFORE the supervisor-specific stop (`launchctl bootout` / `systemctl disable --now`; cron/container just exit), all interpolations single-quoted shell literals. Wrapper + artifacts resolve through `gbrainHomePath()` and a GBRAIN_HOME install bakes the env var into the wrapper (supervisors don't pass the installer's env). `--install` and `--uninstall` clear both markers. launchd label + lock/marker/strike paths resolve through `src/core/autopilot-paths.ts`. Pinned by `test/autopilot-self-disable-and-status.test.ts` + `test/autopilot-launchd-lifecycle.serial.test.ts` (shimmed lifecycle on all platforms; REAL launchd on darwin). +- `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: `src/commands/autopilot-remediation-policy.ts` owns the routing policy and interval-scoped targeted-remediation keys. Each tick: cheap `engine.getHealth()` + `computeRecommendations()`, then route by shape — healthy/no-plan brains sleep until the 60-minute floor; an overdue full cycle runs regardless of score or a non-empty plan; fresh plans with `<=3` steps and `<5min` stay targeted; large/slow plans or score `<70` use the full cycle. Targeted jobs append `:autopilot:` to the stable recommendation key, preserving doctor/remediate checkpoint semantics while reopening the slot after a completed interval (#4046). The `gbrain-cycle` lock prevents targeted submissions and the full cycle from running concurrently; `maxWaiting: 1` bounds the cross-window backlog when a handler runs longer than one interval. After restart, an all-fresh persisted fan-out advances the process-local full-cycle clock so a remaining targeted plan resumes on the next tick; failed or cooldown-gated stale sources do not. Pinned by `test/autopilot-fanout.test.ts` and `test/autopilot-fanout-wiring.test.ts`. `--status` is heartbeat-truth, not artifact-presence, and is dispatched ENGINE-FREE in cli.ts before connectEngine (`runAutopilotStatus`; a live PGLite daemon's exclusive DB lock or a DB outage must not take down the alarm — `--uninstall` rides the same pre-engine dispatch): pure `classifyAutopilotStatus` maps `{installed, disabledReason, pausedReason, heartbeatAgeSeconds, intervalSeconds}` → `not_installed`/`disabled`/`paused`/`never_run`/`stale`/`fresh` with exit codes 0/1/2 via `autopilotStatusExitCode` (disabled=2; paused/stale/never_run=1). Stale tolerance = 6×interval — the adaptive scheduler sleeps 2×interval on healthy brains, so 3× flapped false alarms; a non-finite or non-positive interval falls back to 1800s at BOTH the dispatch parse and the pure layer (NaN staleAfter would read a dead daemon as fresh), and the daemon's own interval parse carries the same guard (NaN reached setTimeout as 0 = busy-loop). `paused` outranks the heartbeat states because the tick loop refreshes its heartbeat BEFORE honoring the pause marker; the tick loop also SELF-CLEARS a migrate-signed marker whose recorded pid is dead (SIGKILL orphans heal within one poll) and exits-for-relaunch when the file-plane engine identity diverges from boot (`autopilotEngineIdentity`; null/torn reads are skipped, so a concurrent config write cannot restart it). `detectInstalledTarget()` probes plist/systemd-unit/start-script/crontab, where the crontab probe uses pure `crontabIndicatesAutopilotInstall` (a cron'd status-monitor line must not read as an install). The generated wrapper embeds `generateSelfDisableGuard(repo, target)`: repo-dir test (never `.git` — a file in worktrees), THREE consecutive misses required before disabling (strike counter file; one absent external/cloud volume at login must not permanently kill the install), marker write BEFORE the supervisor-specific stop (`launchctl bootout` / `systemctl disable --now`; cron/container just exit), all interpolations single-quoted shell literals. Wrapper + artifacts resolve through `gbrainHomePath()` and a GBRAIN_HOME install bakes the env var into the wrapper (supervisors don't pass the installer's env). `--install` and `--uninstall` clear both markers. launchd label + lock/marker/strike paths resolve through `src/core/autopilot-paths.ts`. Pinned by `test/autopilot-self-disable-and-status.test.ts` + `test/autopilot-launchd-lifecycle.serial.test.ts` (shimmed lifecycle on all platforms; REAL launchd on darwin). - `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. diff --git a/src/commands/autopilot-fanout.ts b/src/commands/autopilot-fanout.ts index bf19cd022..e9adf9977 100644 --- a/src/commands/autopilot-fanout.ts +++ b/src/commands/autopilot-fanout.ts @@ -34,8 +34,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; +import { AUTOPILOT_FULL_CYCLE_FLOOR_MINUTES } from './autopilot-remediation-policy.ts'; // #2194 fix #2: failure cooldown. A source whose autopilot-cycle keeps // failing/timing-out re-dispatches every tick today (only SUCCESS gates @@ -81,6 +80,8 @@ export interface FanoutResult { /** True when this tick fell back to the legacy single-job path * (no sources rows / engine empty). */ legacy_fallback: boolean; + /** True when every enumerated source is inside the freshness window. */ + all_sources_fresh: boolean; } /** @@ -180,7 +181,11 @@ export function readLastFullCycleAt(src: SourceRow): Date | null { * a brain may have fresh sync but stale extract/embed. The 60-min floor on * full-cycle is the canonical freshness signal for autopilot dispatch. */ -export function isSourceStale(src: SourceRow, now = Date.now(), floorMin = FULL_CYCLE_FLOOR_MIN): boolean { +export function isSourceStale( + src: SourceRow, + now = Date.now(), + floorMin = AUTOPILOT_FULL_CYCLE_FLOOR_MINUTES, +): boolean { const last = readLastFullCycleAt(src); if (last === null) return true; const ageMin = (now - last.getTime()) / 60_000; @@ -328,7 +333,7 @@ export function selectSourcesForDispatch( sources: SourceRow[], fanoutMax: number, now = Date.now(), - floorMin = FULL_CYCLE_FLOOR_MIN, + floorMin = AUTOPILOT_FULL_CYCLE_FLOOR_MINUTES, recentFailures: Map = new Map(), cooldownOpts: CooldownOpts = { baseMin: FAILURE_COOLDOWN_BASE_MIN, capMin: FAILURE_COOLDOWN_CAP_MIN }, ): { dispatch: SourceRow[]; skippedFresh: SourceRow[]; skippedCap: SourceRow[]; skippedCooldown: SourceRow[] } { @@ -406,7 +411,14 @@ export async function dispatchPerSource( } else { log(`[dispatch] job #${job.id} autopilot-cycle (legacy single-source)`); } - return { dispatched: [], skipped_fresh: [], skipped_cap: [], skipped_cooldown: [], legacy_fallback: true }; + return { + dispatched: [], + skipped_fresh: [], + skipped_cap: [], + skipped_cooldown: [], + legacy_fallback: true, + all_sources_fresh: false, + }; } // #2194 fix #2: load recent per-source failures + cooldown knobs so a @@ -426,7 +438,14 @@ export async function dispatchPerSource( } const { dispatch, skippedFresh, skippedCap, skippedCooldown } = - selectSourcesForDispatch(sources, opts.fanoutMax, Date.now(), FULL_CYCLE_FLOOR_MIN, recentFailures, cooldownOpts); + selectSourcesForDispatch( + sources, + opts.fanoutMax, + Date.now(), + AUTOPILOT_FULL_CYCLE_FLOOR_MINUTES, + recentFailures, + cooldownOpts, + ); const dispatched: string[] = []; for (const src of dispatch) { @@ -509,6 +528,7 @@ export async function dispatchPerSource( skipped_cap: skippedCap.map(s => s.id), skipped_cooldown: skippedCooldown.map(s => s.id), legacy_fallback: false, + all_sources_fresh: skippedFresh.length === sources.length, }; } diff --git a/src/commands/autopilot-remediation-policy.ts b/src/commands/autopilot-remediation-policy.ts new file mode 100644 index 000000000..340126ffb --- /dev/null +++ b/src/commands/autopilot-remediation-policy.ts @@ -0,0 +1,46 @@ +export const AUTOPILOT_FULL_CYCLE_FLOOR_MINUTES = 60; + +export interface AutopilotRemediationPlanShape { + score: number; + planLength: number; + estimatedSeconds: number; + minutesSinceLastFull: number; +} + +/** + * Keep recommendation keys stable for doctor/remediate checkpoints while + * giving Autopilot a fresh single-flight slot on every dispatch interval. + */ +export function autopilotRemediationIdempotencyKey( + recommendationKey: string, + dispatchSlot: string, +): string { + return `${recommendationKey}:autopilot:${dispatchSlot}`; +} + +/** + * A full cycle is a freshness invariant, independent of the current score or + * targeted plan. Large/slow/severely degraded plans retain the existing + * hammer behavior before the freshness floor is reached. + */ +export function shouldRunAutopilotFullCycle({ + score, + planLength, + estimatedSeconds, + minutesSinceLastFull, +}: AutopilotRemediationPlanShape): boolean { + return minutesSinceLastFull >= AUTOPILOT_FULL_CYCLE_FLOOR_MINUTES + || planLength > 3 + || estimatedSeconds >= 300 + || score < 70; +} + +export function shouldSleepHealthyAutopilot( + score: number, + planLength: number, + minutesSinceLastFull: number, +): boolean { + return score >= 95 + && planLength === 0 + && minutesSinceLastFull < AUTOPILOT_FULL_CYCLE_FLOOR_MINUTES; +} diff --git a/src/commands/autopilot.ts b/src/commands/autopilot.ts index 6314b9b2e..390305cbc 100644 --- a/src/commands/autopilot.ts +++ b/src/commands/autopilot.ts @@ -41,6 +41,11 @@ import { evaluateQuietHours } from '../core/minions/quiet-hours.ts'; import { inspectLock } from '../core/db-lock.ts'; import { registerCleanup } from '../core/process-cleanup.ts'; import { resolveAutopilotDispatchTimeoutMs } from './autopilot-timeout.ts'; +import { + autopilotRemediationIdempotencyKey, + shouldRunAutopilotFullCycle, + shouldSleepHealthyAutopilot, +} from './autopilot-remediation-policy.ts'; // Path helpers live in a LEAF core module so other commands (gbrain migrate) // can read the daemon's state files without importing this one — a dynamic // import of a command module drags its whole flag surface into the importer's @@ -875,8 +880,8 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) { // // New logic: compute the remediation plan (cheap; no full doctor // walk), then route to the right level of intervention: - // - Score >= 95 + empty plan: full cycle every 60min (phase- - // coupling exercise), otherwise sleep. + // - Full cycle every 60min regardless of score/plan (phase- + // coupling + freshness invariant); healthy brains sleep before it. // - Small plan (<=3 steps, <5min): submit individual handlers. // - Large plan or low score: full autopilot-cycle (the hammer). // @@ -1121,16 +1126,16 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) { const estTotal = plan.reduce((s, r) => s + r.est_seconds, 0); // Track time since last full cycle for the 60-min floor. - const FULL_CYCLE_FLOOR_MIN = 60; const minutesSinceLastFull = (Date.now() - lastFullCycleAt) / 60000; - const shouldFullCycle = - (score >= 95 && plan.length === 0 && minutesSinceLastFull >= FULL_CYCLE_FLOOR_MIN) || - plan.length > 3 || - estTotal >= 300 || - score < 70; + const shouldFullCycle = shouldRunAutopilotFullCycle({ + score, + planLength: plan.length, + estimatedSeconds: estTotal, + minutesSinceLastFull, + }); - const shouldSleep = score >= 95 && plan.length === 0 && minutesSinceLastFull < FULL_CYCLE_FLOOR_MIN; + const shouldSleep = shouldSleepHealthyAutopilot(score, plan.length, minutesSinceLastFull); if (shouldSleep) { if (jsonMode) { @@ -1181,7 +1186,11 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) { if (jsonMode) process.stderr.write(JSON.stringify({ event: 'global_maintenance_dispatch_failed', error: e instanceof Error ? e.message : String(e) }) + '\n'); } } - if (result.dispatched.length > 0 || result.legacy_fallback) { + // On restart the process-local clock starts overdue. If persisted + // source timestamps say every source is fresh, advance the local + // clock too; otherwise a non-empty targeted plan would be skipped + // on every tick until the persisted 60-minute window elapsed. + if (result.dispatched.length > 0 || result.legacy_fallback || result.all_sources_fresh) { lastFullCycleAt = Date.now(); } if (jsonMode) { @@ -1205,15 +1214,17 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) { } } else { // Small targeted plan — submit individual handlers per step. - // D9 content-hash idempotency keys (from computeRecommendations). - // maxWaiting:1 per submit per codex #17 (closes the backpressure - // gap the prior implementation had for targeted submits). + // Recommendation keys stay stable for doctor/remediate checkpoints; + // Autopilot adds the dispatch interval so completed rows cannot hold + // the remediation slot forever (#4046). + // maxWaiting:1 per submit per codex #17 bounds the cross-window + // backlog if a targeted handler runs longer than one interval. for (const step of plan) { try { const isProtected = !!step.protected; const submitOpts = { queue: 'default', - idempotency_key: step.idempotency_key, + idempotency_key: autopilotRemediationIdempotencyKey(step.idempotency_key, slot), max_attempts: 2, timeout_ms: timeoutMs, maxWaiting: 1, diff --git a/test/autopilot-fanout-wiring.test.ts b/test/autopilot-fanout-wiring.test.ts index 21b2a14bb..9d1017655 100644 --- a/test/autopilot-fanout-wiring.test.ts +++ b/test/autopilot-fanout-wiring.test.ts @@ -80,6 +80,13 @@ describe('autopilot.ts ↔ dispatchPerSource wiring', () => { expect(freshnessBlock).toContain('pull: sourceConfigHasRemoteUrl(src.config)'); }); + test('#4046: targeted dispatch scopes stable recommendation keys to the interval', () => { + expect(AUTOPILOT_SRC).toContain( + 'idempotency_key: autopilotRemediationIdempotencyKey(step.idempotency_key, slot)', + ); + expect(AUTOPILOT_SRC).not.toContain('idempotency_key: step.idempotency_key,'); + }); + 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 @@ -101,9 +108,12 @@ describe('autopilot.ts ↔ dispatchPerSource wiring', () => { expect(dispatchGlobalCall).not.toMatch(/\{\s*repoPath,\s*slot,\s*timeoutMs,/); }); - test('updates lastFullCycleAt on dispatch (so the 60-min floor is honored)', () => { + test('updates lastFullCycleAt after dispatch or an all-fresh restart check', () => { // After the dispatchPerSource call, the lastFullCycleAt module var // must update so the next tick doesn't immediately re-fan-out. + expect(AUTOPILOT_SRC).toMatch( + /result\.dispatched\.length > 0 \|\| result\.legacy_fallback \|\| result\.all_sources_fresh/, + ); expect(AUTOPILOT_SRC).toMatch(/lastFullCycleAt\s*=\s*Date\.now\(\)/); }); diff --git a/test/autopilot-fanout.test.ts b/test/autopilot-fanout.test.ts index d38aee88a..07b8b2d32 100644 --- a/test/autopilot-fanout.test.ts +++ b/test/autopilot-fanout.test.ts @@ -18,6 +18,11 @@ import { resolveFanoutMax, dispatchPerSource, } from '../src/commands/autopilot-fanout.ts'; +import { + autopilotRemediationIdempotencyKey, + shouldRunAutopilotFullCycle, + shouldSleepHealthyAutopilot, +} from '../src/commands/autopilot-remediation-policy.ts'; import type { SourceRow, BrainEngine } from '../src/core/engine.ts'; function src(id: string, last_full_cycle_at?: string | null, extra: Record = {}): SourceRow { @@ -74,6 +79,66 @@ describe('isSourceStale', () => { }); }); +describe('Autopilot remediation policy (#4046)', () => { + test('targeted remediation keys reopen in each dispatch interval', () => { + const recommendationKey = 'default:sync:deadbeef'; + const firstSlot = '2026-08-13T06:00:00.000Z'; + const nextSlot = '2026-08-13T06:05:00.000Z'; + + expect(autopilotRemediationIdempotencyKey(recommendationKey, firstSlot)).toBe( + 'default:sync:deadbeef:autopilot:2026-08-13T06:00:00.000Z', + ); + expect(autopilotRemediationIdempotencyKey(recommendationKey, nextSlot)).not.toBe( + autopilotRemediationIdempotencyKey(recommendationKey, firstSlot), + ); + }); + + test('an overdue full cycle wins even while a small remediation plan exists', () => { + expect(shouldRunAutopilotFullCycle({ + score: 94, + planLength: 2, + estimatedSeconds: 30, + minutesSinceLastFull: 61, + })).toBe(true); + }); + + test('a fresh small remediation plan stays targeted', () => { + expect(shouldRunAutopilotFullCycle({ + score: 94, + planLength: 2, + estimatedSeconds: 30, + minutesSinceLastFull: 10, + })).toBe(false); + }); + + test('only a fresh healthy brain with no plan sleeps', () => { + expect(shouldSleepHealthyAutopilot(95, 0, 59)).toBe(true); + expect(shouldSleepHealthyAutopilot(95, 0, 60)).toBe(false); + expect(shouldSleepHealthyAutopilot(95, 1, 10)).toBe(false); + }); + + test('large, slow, or severely degraded plans still use the full cycle', () => { + expect(shouldRunAutopilotFullCycle({ + score: 90, + planLength: 4, + estimatedSeconds: 30, + minutesSinceLastFull: 10, + })).toBe(true); + expect(shouldRunAutopilotFullCycle({ + score: 90, + planLength: 2, + estimatedSeconds: 300, + minutesSinceLastFull: 10, + })).toBe(true); + expect(shouldRunAutopilotFullCycle({ + score: 69, + planLength: 1, + estimatedSeconds: 30, + minutesSinceLastFull: 10, + })).toBe(true); + }); +}); + describe('selectSourcesForDispatch', () => { const NOW = Date.parse('2026-05-22T12:00:00.000Z'); const fresh = (id: string, agoMin: number) => @@ -313,6 +378,19 @@ describe('dispatchPerSource — integration with stubbed engine + queue', () => const result = await dispatchPerSource(engine, queue, fanoutOpts); expect(result.dispatched.length).toBe(0); expect(result.skipped_fresh.length).toBe(2); + expect(result.all_sources_fresh).toBe(true); expect(added.length).toBe(0); }); + + test('a failed stale-source submission is not misclassified as all fresh', async () => { + const { engine, fanoutOpts } = makeStubs([src('stale')]); + const queue = { + add: async () => { throw new Error('queue unavailable'); }, + } as unknown as Parameters[1]; + + const result = await dispatchPerSource(engine, queue, fanoutOpts); + + expect(result.dispatched).toEqual([]); + expect(result.all_sources_fresh).toBe(false); + }); });