fix(autopilot): reopen targeted remediation slots (#4046) (#4067)

Co-authored-by: arisgysel-design <arisgysel-design@users.noreply.github.com>

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.

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<string, SourceFailure> = 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,
   };
 }
 

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;
+}

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,

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\(\)/);
   });
 

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<string, unknown> = {}): 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<typeof dispatchPerSource>[1];
+
+    const result = await dispatchPerSource(engine, queue, fanoutOpts);
+
+    expect(result.dispatched).toEqual([]);
+    expect(result.all_sources_fresh).toBe(false);
+  });
 });