diff --git a/.changeset/fn-7512-planner-bounded-recovery.md b/.changeset/fn-7512-planner-bounded-recovery.md new file mode 100644 index 0000000000..76586f23d0 --- /dev/null +++ b/.changeset/fn-7512-planner-bounded-recovery.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": minor +--- + +summary: Planner oversight can autonomously inject guidance, retry stuck/failed steps, and request fixes within bounded limits. +category: feature +dev: Adds pure `decidePlannerRecovery` + recovery types (core) and `PlannerRecoveryController` with injected guidance/retry/targeted-fix handlers (engine), consuming the FN-7511 observation. Acts only at effective level `autonomous`, caps attempts per (task, stage) via `PLANNER_RECOVERY_MAX_ATTEMPTS`, skips user-paused tasks, and excludes merge/PR/destructive actions (deferred to FN-7513) and comprehensive human-control safeguards (FN-7514). diff --git a/docs/architecture.md b/docs/architecture.md index 058667e914..2ca84a8ca2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1378,6 +1378,68 @@ call happens here, and it emits no run-audit events or dashboard UI. Steering/re gates, human-control safeguards, and dashboard/UI/run-audit surfaces are deferred to FN-7512 through FN-7520; this module is the seam those subtasks read observations from. +### Planner overseer bounded autonomous recovery (FN-7512) + +/* +FNXC:PlannerOversight 2026-07-04-12:00: +FN-7512 builds the bounded autonomous-recovery layer on top of FN-7511's observation seam. When the +task's effective planner oversight level resolves to `"autonomous"`, the planner overseer may take ONE +of three bounded corrective actions on the task's currently watched stage: + - **inject_guidance** — post a planner-authored steering comment into the active agent lane. + - **retry_step** — re-enqueue a stuck/failed step via the existing store retry/re-enqueue path. + - **request_targeted_fix** — post a steering comment tagged as a targeted-fix request, referencing + the observation's specific error source link. +At every other effective level (`"off"`/`"observe"`/`"steer"`) the decision is always `"none"` — this +layer is completely inert unless oversight is `"autonomous"`. +*/ + +`packages/core/src/planner-recovery.ts` declares the shared, engine-free recovery vocabulary: +`PlannerRecoveryActionKind` (`inject_guidance | retry_step | request_targeted_fix | none`), +`PlannerRecoveryObservation` (a structural mirror of FN-7511's `OverseerStageObservation` so the engine +can pass one straight through with no adapter), `PlannerRecoveryAttemptState`, `PlannerRecoveryDecision`, +and the pure, never-throw `decidePlannerRecovery(input)`. Decision rules, in order: + +1. No observation, or `oversightLevel !== "autonomous"` → `"none"`. +2. The per-`(taskId, watchedStage)` attempt count has reached `PLANNER_RECOVERY_MAX_ATTEMPTS` (default + `3`, mirroring `MAX_RECOVERY_RETRIES` in `recovery-policy.ts`) → `"none"`, `exhausted: true` — the + layer stops autonomously and the task is left for escalation (FN-7514+ owns the human-control story). +3. `merger` / `pull-request` stages → `"none"` with a deferral reason: these require confirmation and + are owned by FN-7513's confirmation-gated layer, never dispatched from here. +4. `reviewer` stage → `"inject_guidance"`. +5. `executor` / `workflow-gate` stage with `signal === "failed"` → `"request_targeted_fix"` when a + source link carries a specific fixable error (`failed-check` / `merge-error`), else `"retry_step"`. +6. Any other `executor` / `workflow-gate` signal (stuck/blocked/progressing/awaiting-human) → + `"inject_guidance"`. + +`packages/engine/src/planner-recovery-controller.ts`'s `PlannerRecoveryController` is the dispatcher, +mirroring the `AutoRecoveryDispatcher` + `StuckTaskDetector` handler-injection conventions: it holds an +in-memory per-`(taskId, watchedStage)` attempt registry, calls `decidePlannerRecovery`, and — only when +an action other than `"none"` is chosen — dispatches through injected `PlannerRecoveryHandlers` +(`injectGuidance` / `retryStep` / `requestTargetedFix`, all optional and async), incrementing the +attempt count only on a successful dispatch. `tick(task, ctx)` is a no-op (returns `null`) when +`task.userPaused === true` or when there is no active observation, and never throws — any handler or +snapshot-provider error degrades to a no-op. + +`ProjectEngine` wires one concrete `PlannerRecoveryController` alongside its `PlannerOverseerMonitor`, +reusing ONLY existing mechanisms — no new session/tool/merge channel: + +- `injectGuidance` / `requestTargetedFix` → `store.addSteeringComment(taskId, text, "agent")` (the same + channel the executor's real-time injection listener already watches). +- `retryStep` → `store.moveTask(taskId, "todo", { preserveProgress: true, moveSource: "engine" })` — the + same in-progress→todo retry/re-enqueue path auto-recovery and self-healing already use. + +`controller.tick(task)` is called from the SAME bounded 45s poll FN-7511 uses for `observeTask`, guarded +so it only runs when the resolved effective level is `"autonomous"` (every other level already +`continue`s before reaching the tick). Attempt state for a task is cleared (`controller.clear(taskId)`) +whenever the task leaves the in-flight `in-progress`/`in-review` set, alongside the FN-7511 observation +ring buffer. + +**Explicit scope boundaries (owned by later subtasks, not this layer):** merge/PR actions and +destructive/external-service side effects (FN-7513, confirmation-gated); comprehensive human-pause / +`autoMerge:false` / human-review terminal safeguards beyond the bare `userPaused` skip (FN-7514); a +persisted intervention timeline (FN-7519); run-audit/activity events (FN-7520); and any dashboard UI +(FN-7515+). + --- ## 11) Multi-Project Architecture diff --git a/packages/core/src/__tests__/planner-recovery.test.ts b/packages/core/src/__tests__/planner-recovery.test.ts new file mode 100644 index 0000000000..11dc2c5061 --- /dev/null +++ b/packages/core/src/__tests__/planner-recovery.test.ts @@ -0,0 +1,116 @@ +import { describe, expect, it } from "vitest"; +import { + decidePlannerRecovery, + PLANNER_RECOVERY_MAX_ATTEMPTS, + type PlannerRecoveryObservation, +} from "../planner-recovery.js"; + +function observation(overrides: Partial = {}): PlannerRecoveryObservation { + return { + taskId: "FN-1", + stage: "executor", + signal: "progressing", + oversightLevel: "autonomous", + sources: [], + ...overrides, + }; +} + +describe("decidePlannerRecovery", () => { + it("returns none when there is no observation (no watched stage)", () => { + const decision = decidePlannerRecovery({ snapshot: null }); + expect(decision.action).toBe("none"); + expect(decision.exhausted).toBe(false); + expect(decision.watchedStage).toBeNull(); + }); + + it("returns none for every non-autonomous effective level", () => { + for (const level of ["off", "observe", "steer"] as const) { + const decision = decidePlannerRecovery({ snapshot: observation({ oversightLevel: level, signal: "failed" }) }); + expect(decision.action, `level=${level}`).toBe("none"); + expect(decision.exhausted).toBe(false); + } + }); + + it("yields retry_step for a failed executor stage with no specific error source", () => { + const decision = decidePlannerRecovery({ snapshot: observation({ stage: "executor", signal: "failed", sources: [] }) }); + expect(decision.action).toBe("retry_step"); + expect(decision.attemptCount).toBe(0); + expect(decision.attemptLimit).toBe(PLANNER_RECOVERY_MAX_ATTEMPTS); + }); + + it("yields retry_step for a failed workflow-gate stage with no specific error source", () => { + const decision = decidePlannerRecovery({ snapshot: observation({ stage: "workflow-gate", signal: "failed" }) }); + expect(decision.action).toBe("retry_step"); + }); + + it("yields request_targeted_fix for a failed executor stage carrying a specific error source link", () => { + const decision = decidePlannerRecovery({ + snapshot: observation({ + stage: "executor", + signal: "failed", + sources: [{ kind: "failed-check", ref: "check-1", url: "https://example.test/check-1" }], + }), + attemptState: { attemptCount: 1 }, + }); + expect(decision.action).toBe("request_targeted_fix"); + expect(decision.attemptCount).toBe(1); + expect(decision.attemptLimit).toBe(PLANNER_RECOVERY_MAX_ATTEMPTS); + }); + + it("yields inject_guidance for a reviewer stage", () => { + const decision = decidePlannerRecovery({ snapshot: observation({ stage: "reviewer", signal: "progressing" }) }); + expect(decision.action).toBe("inject_guidance"); + }); + + it("yields inject_guidance for a stuck-but-not-failed stage", () => { + const decision = decidePlannerRecovery({ snapshot: observation({ stage: "executor", signal: "stuck" }) }); + expect(decision.action).toBe("inject_guidance"); + }); + + it("defers merger and pull-request stages to none with a deferral reason", () => { + for (const stage of ["merger", "pull-request"] as const) { + const decision = decidePlannerRecovery({ snapshot: observation({ stage, signal: "failed" }) }); + expect(decision.action, `stage=${stage}`).toBe("none"); + expect(decision.reason.toLowerCase()).toContain("deferred"); + } + }); + + it("returns none + exhausted true exactly at the attempt limit", () => { + const decision = decidePlannerRecovery({ + snapshot: observation({ stage: "executor", signal: "failed" }), + attemptState: { attemptCount: PLANNER_RECOVERY_MAX_ATTEMPTS }, + }); + expect(decision.action).toBe("none"); + expect(decision.exhausted).toBe(true); + }); + + it("still allows action one attempt below the limit", () => { + const decision = decidePlannerRecovery({ + snapshot: observation({ stage: "executor", signal: "failed" }), + attemptState: { attemptCount: PLANNER_RECOVERY_MAX_ATTEMPTS - 1 }, + }); + expect(decision.action).not.toBe("none"); + expect(decision.exhausted).toBe(false); + }); + + it("never throws on missing/partial snapshot fields", () => { + expect(() => decidePlannerRecovery({ snapshot: undefined })).not.toThrow(); + expect(() => decidePlannerRecovery({} as never)).not.toThrow(); + expect(() => + decidePlannerRecovery({ snapshot: { taskId: "FN-1" } as unknown as PlannerRecoveryObservation }), + ).not.toThrow(); + const decision = decidePlannerRecovery({ snapshot: { taskId: "FN-1" } as unknown as PlannerRecoveryObservation }); + expect(decision.action).toBe("none"); + }); + + it("respects a custom attemptLimit override", () => { + const decision = decidePlannerRecovery({ + snapshot: observation({ stage: "executor", signal: "failed" }), + attemptState: { attemptCount: 1, attemptLimit: 1 }, + }); + expect(decision.action).toBe("none"); + expect(decision.exhausted).toBe(true); + expect(decision.attemptLimit).toBe(1); + }); +}); diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index b0b2d24b84..ba5dd4a099 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -430,6 +430,18 @@ export { applyWorkflowSettingsOverlay, type WorkflowSettingsOverlayInput, } from "./effective-settings-overlay.js"; +export { + decidePlannerRecovery, + PLANNER_RECOVERY_MAX_ATTEMPTS, + type PlannerRecoveryActionKind, + type PlannerRecoveryWatchedStage, + type PlannerRecoveryObservationSignal, + type PlannerRecoverySourceLink, + type PlannerRecoveryObservation, + type PlannerRecoveryAttemptState, + type PlannerRecoveryDecision, + type DecidePlannerRecoveryInput, +} from "./planner-recovery.js"; // ── Engine wiring (set by @fusion/engine at module load) ──────────── export { diff --git a/packages/core/src/planner-recovery.ts b/packages/core/src/planner-recovery.ts new file mode 100644 index 0000000000..7081bf7800 --- /dev/null +++ b/packages/core/src/planner-recovery.ts @@ -0,0 +1,222 @@ +/** + * FNXC:PlannerOversight 2026-07-04-12:00: + * FN-7512 requirement: when the effective planner oversight level is + * `"autonomous"`, the planner overseer may take BOUNDED autonomous + * corrective action on the task's currently watched stage — inject steering + * guidance into the active agent lane, retry a stuck/failed step, or request + * a targeted fix for a detected error. Every action is capped by a + * per-(task, watched-stage) attempt limit (`PLANNER_RECOVERY_MAX_ATTEMPTS`) + * so recovery can never loop forever; once the budget is exhausted the + * decision degrades to `"none"` with `exhausted: true` and the task is left + * for human/other escalation. Merge/PR and destructive actions are + * explicitly OUT of scope here (deferred to FN-7513's confirmation-gated + * layer), and comprehensive human-control safeguards beyond a bare + * `userPaused` skip are FN-7514's responsibility. This module is pure, + * never-throws, and has NO engine imports — the engine-side dispatch lives + * in `@fusion/engine`'s `PlannerRecoveryController`. + * + * Delivered-shape note: FN-7511 shipped its observation model in + * `packages/engine/src/planner-overseer.ts` (`OverseerStageObservation` / + * `OverseerWatchedStage` / `OverseerSourceLink`) rather than the + * `PlannerObservationSnapshot` shape anticipated at spec time, and it has no + * `getSnapshot`/`isActive`/`watchedStage === "none"` API — instead + * `PlannerOverseerMonitor.observeTask()` returns one observation (or `null` + * when there is nothing to watch). This module's `PlannerRecoveryObservation` + * input type mirrors the delivered `OverseerStageObservation` field names + * structurally (`stage`, `signal`, `oversightLevel`, `sources`) so the engine + * controller can pass an `OverseerStageObservation` straight through without + * an adapter; "no watched stage" is represented by passing `snapshot: null`. + */ + +import type { PlannerOversightLevel } from "./types.js"; + +/** The bounded corrective actions autonomous planner recovery may take. */ +export type PlannerRecoveryActionKind = "inject_guidance" | "retry_step" | "request_targeted_fix" | "none"; + +/** Mirrors the delivered `OverseerWatchedStage` union (FN-7511). */ +export type PlannerRecoveryWatchedStage = "executor" | "reviewer" | "merger" | "pull-request" | "workflow-gate"; + +/** Mirrors the delivered `OverseerObservationSignal` union (FN-7511). */ +export type PlannerRecoveryObservationSignal = "progressing" | "stuck" | "failed" | "blocked" | "awaiting-human" | "complete"; + +/** Mirrors the delivered `OverseerSourceLink` shape (FN-7511) structurally. */ +export interface PlannerRecoverySourceLink { + kind: string; + ref: string; + url?: string; +} + +/** + * The minimal observation shape `decidePlannerRecovery` reads. Structurally + * compatible with the engine's `OverseerStageObservation` (FN-7511) so the + * engine controller can pass one straight through. `null` means "no watched + * stage currently active for this task" (equivalent to the spec's + * `watchedStage === "none"` / `isActive: false`). + */ +export interface PlannerRecoveryObservation { + taskId: string; + stage: PlannerRecoveryWatchedStage; + signal: PlannerRecoveryObservationSignal; + oversightLevel: PlannerOversightLevel | string; + sources?: PlannerRecoverySourceLink[]; +} + +/** Per-`(taskId, watchedStage)` bounded attempt counter the caller persists/tracks. */ +export interface PlannerRecoveryAttemptState { + attemptCount: number; + attemptLimit?: number; +} + +/** Result of `decidePlannerRecovery` — pure, deterministic, never throws. */ +export interface PlannerRecoveryDecision { + action: PlannerRecoveryActionKind; + reason: string; + attemptCount: number; + attemptLimit: number; + exhausted: boolean; + watchedStage: PlannerRecoveryWatchedStage | null; + sourceLinks: PlannerRecoverySourceLink[]; +} + +/** + * Maximum bounded recovery attempts per `(taskId, watchedStage)` before + * autonomous action stops and the task is left for escalation. Mirrors the + * bound style of `MAX_RECOVERY_RETRIES` in `recovery-policy.ts`. + */ +export const PLANNER_RECOVERY_MAX_ATTEMPTS = 3; + +/** Source-link kinds treated as carrying a specific, fixable error (vs. a bare stuck/blocked signal). */ +const ERROR_SOURCE_KINDS = new Set(["failed-check", "merge-error"]); + +export interface DecidePlannerRecoveryInput { + /** The current observation for the task's watched stage, or `null` when nothing is currently watched. */ + snapshot: PlannerRecoveryObservation | null | undefined; + /** Current attempt state for this `(taskId, watchedStage)`; omit for a fresh stage. */ + attemptState?: PlannerRecoveryAttemptState; +} + +/** + * FNXC:PlannerOversight 2026-07-04-12:00: + * Pure, never-throw decision function for bounded autonomous planner + * recovery. Rules: + * 1. No observation, or `oversightLevel !== "autonomous"` → `"none"` + * (nothing to do / oversight level does not permit autonomous action). + * 2. Attempt budget for the `(taskId, watchedStage)` already spent + * (`attemptCount >= attemptLimit`) → `"none"`, `exhausted: true` (stop + * autonomously; leave the task for escalation). + * 3. `merger` / `pull-request` stages → `"none"` with a deferral reason — + * these require confirmation and are owned by FN-7513, never dispatched + * from this bounded layer. + * 4. `reviewer` stage → `"inject_guidance"`. + * 5. `executor` / `workflow-gate` stage with `signal === "failed"` → + * `"request_targeted_fix"` when a source link carries a specific + * fixable error (`failed-check` / `merge-error`), else `"retry_step"`. + * 6. Any other `executor` / `workflow-gate` signal (stuck/blocked/ + * progressing/awaiting-human) → `"inject_guidance"`. + */ +export function decidePlannerRecovery(input: DecidePlannerRecoveryInput): PlannerRecoveryDecision { + const attemptCount = input?.attemptState?.attemptCount ?? 0; + const attemptLimit = input?.attemptState?.attemptLimit ?? PLANNER_RECOVERY_MAX_ATTEMPTS; + + try { + const snapshot = input?.snapshot ?? null; + const watchedStage = snapshot?.stage ?? null; + const sourceLinks = snapshot?.sources ?? []; + + if (!snapshot) { + return { + action: "none", + reason: "No watched stage is currently active for this task", + attemptCount, + attemptLimit, + exhausted: false, + watchedStage, + sourceLinks, + }; + } + + if (snapshot.oversightLevel !== "autonomous") { + return { + action: "none", + reason: `Effective planner oversight level "${String(snapshot.oversightLevel)}" does not permit autonomous recovery`, + attemptCount, + attemptLimit, + exhausted: false, + watchedStage, + sourceLinks, + }; + } + + if (attemptCount >= attemptLimit) { + return { + action: "none", + reason: `Bounded recovery attempt budget (${attemptLimit}) exhausted for stage "${watchedStage}"`, + attemptCount, + attemptLimit, + exhausted: true, + watchedStage, + sourceLinks, + }; + } + + if (snapshot.stage === "merger" || snapshot.stage === "pull-request") { + return { + action: "none", + reason: `Stage "${snapshot.stage}" requires confirmation-gated recovery (deferred to FN-7513)`, + attemptCount, + attemptLimit, + exhausted: false, + watchedStage, + sourceLinks, + }; + } + + if (snapshot.stage === "reviewer") { + return { + action: "inject_guidance", + reason: "Reviewer stage — injecting steering guidance", + attemptCount, + attemptLimit, + exhausted: false, + watchedStage, + sourceLinks, + }; + } + + // executor / workflow-gate beyond this point. + if (snapshot.signal === "failed") { + const hasErrorSource = sourceLinks.some((link) => ERROR_SOURCE_KINDS.has(link.kind)); + return { + action: hasErrorSource ? "request_targeted_fix" : "retry_step", + reason: hasErrorSource + ? "Failed stage with a specific error source — requesting a targeted fix" + : "Failed stage with no specific error source — retrying the step", + attemptCount, + attemptLimit, + exhausted: false, + watchedStage, + sourceLinks, + }; + } + + return { + action: "inject_guidance", + reason: `Stage "${snapshot.stage}" signal "${snapshot.signal}" — injecting steering guidance`, + attemptCount, + attemptLimit, + exhausted: false, + watchedStage, + sourceLinks, + }; + } catch { + return { + action: "none", + reason: "decidePlannerRecovery: malformed input — degraded to no-op", + attemptCount, + attemptLimit, + exhausted: false, + watchedStage: null, + sourceLinks: [], + }; + } +} diff --git a/packages/engine/src/__tests__/planner-recovery-controller.test.ts b/packages/engine/src/__tests__/planner-recovery-controller.test.ts new file mode 100644 index 0000000000..b1ce0ea415 --- /dev/null +++ b/packages/engine/src/__tests__/planner-recovery-controller.test.ts @@ -0,0 +1,163 @@ +import { describe, expect, it, vi } from "vitest"; +import { PLANNER_RECOVERY_MAX_ATTEMPTS } from "@fusion/core"; +import type { Task } from "@fusion/core"; +import { PlannerRecoveryController, type PlannerRecoveryHandlers } from "../planner-recovery-controller.js"; +import type { OverseerStageObservation, OverseerWatchedStage } from "../planner-overseer.js"; + +function task(overrides: Partial = {}): Task { + return { + id: "FN-1", + title: "t", + description: "", + column: "in-progress", + ...overrides, + } as Task; +} + +function observation(overrides: Partial = {}): OverseerStageObservation { + return { + taskId: "FN-1", + stage: "executor" as OverseerWatchedStage, + signal: "failed", + oversightLevel: "autonomous", + observedAt: Date.now(), + reason: "test", + sources: [], + ...overrides, + }; +} + +function makeController( + obs: OverseerStageObservation | null, + handlers: PlannerRecoveryHandlers = {}, +): PlannerRecoveryController { + return new PlannerRecoveryController({ + snapshotProvider: { getSnapshot: () => obs }, + handlers, + }); +} + +describe("PlannerRecoveryController.tick", () => { + it("dispatches retryStep for a failed executor stage with no error source and increments the attempt count", async () => { + const retryStep = vi.fn().mockResolvedValue(undefined); + const controller = makeController(observation(), { retryStep }); + + const decision = await controller.tick(task()); + expect(decision?.action).toBe("retry_step"); + expect(retryStep).toHaveBeenCalledTimes(1); + expect(controller.getAttemptCount("FN-1", "executor")).toBe(1); + }); + + it("dispatches injectGuidance for a reviewer-stage decision", async () => { + const injectGuidance = vi.fn().mockResolvedValue(undefined); + const controller = makeController(observation({ stage: "reviewer", signal: "progressing" }), { injectGuidance }); + + const decision = await controller.tick(task()); + expect(decision?.action).toBe("inject_guidance"); + expect(injectGuidance).toHaveBeenCalledTimes(1); + }); + + it("dispatches requestTargetedFix when the failed observation carries an error source link", async () => { + const requestTargetedFix = vi.fn().mockResolvedValue(undefined); + const controller = makeController( + observation({ sources: [{ kind: "failed-check", ref: "chk-1" }] }), + { requestTargetedFix }, + ); + + const decision = await controller.tick(task()); + expect(decision?.action).toBe("request_targeted_fix"); + expect(requestTargetedFix).toHaveBeenCalledTimes(1); + }); + + it("stops dispatching once the per-stage attempt budget is exhausted", async () => { + const retryStep = vi.fn().mockResolvedValue(undefined); + const controller = makeController(observation(), { retryStep }); + + for (let i = 0; i < PLANNER_RECOVERY_MAX_ATTEMPTS; i++) { + await controller.tick(task()); + } + expect(retryStep).toHaveBeenCalledTimes(PLANNER_RECOVERY_MAX_ATTEMPTS); + + const exhaustedDecision = await controller.tick(task()); + expect(exhaustedDecision?.action).toBe("none"); + expect(exhaustedDecision?.exhausted).toBe(true); + expect(retryStep).toHaveBeenCalledTimes(PLANNER_RECOVERY_MAX_ATTEMPTS); + }); + + it("is inert when effectiveLevel/oversightLevel is off/observe/steer", async () => { + for (const level of ["off", "observe", "steer"] as const) { + const retryStep = vi.fn().mockResolvedValue(undefined); + const injectGuidance = vi.fn().mockResolvedValue(undefined); + const controller = makeController(observation({ oversightLevel: level }), { retryStep, injectGuidance }); + const decision = await controller.tick(task()); + expect(decision?.action, `level=${level}`).toBe("none"); + expect(retryStep).not.toHaveBeenCalled(); + expect(injectGuidance).not.toHaveBeenCalled(); + } + }); + + it("is skipped entirely when task.userPaused is true", async () => { + const retryStep = vi.fn().mockResolvedValue(undefined); + const controller = makeController(observation(), { retryStep }); + const decision = await controller.tick(task({ userPaused: true })); + expect(decision).toBeNull(); + expect(retryStep).not.toHaveBeenCalled(); + }); + + it("exposes only the three bounded handlers — no merge/PR/destructive action is invocable", () => { + const handlers: PlannerRecoveryHandlers = {}; + const allowed = new Set(["injectGuidance", "retryStep", "requestTargetedFix"]); + // Structural assertion: the handlers interface accepts exactly these three optional members. + const keys = Object.keys({ injectGuidance: undefined, retryStep: undefined, requestTargetedFix: undefined } satisfies Required); + for (const key of keys) { + expect(allowed.has(key)).toBe(true); + } + void handlers; + }); + + it("clear(taskId) resets attempt state for that task", async () => { + const retryStep = vi.fn().mockResolvedValue(undefined); + const controller = makeController(observation(), { retryStep }); + await controller.tick(task()); + expect(controller.getAttemptCount("FN-1", "executor")).toBe(1); + controller.clear("FN-1"); + expect(controller.getAttemptCount("FN-1", "executor")).toBe(0); + }); + + it("never throws when a handler rejects", async () => { + const retryStep = vi.fn().mockRejectedValue(new Error("boom")); + const controller = makeController(observation(), { retryStep }); + await expect(controller.tick(task())).resolves.not.toThrow(); + // Attempt count should not increment on a failed dispatch. + expect(controller.getAttemptCount("FN-1", "executor")).toBe(0); + }); + + it("never throws when the snapshot is absent, and returns null", async () => { + const controller = makeController(null, {}); + await expect(controller.tick(task())).resolves.toBeNull(); + }); + + it("never throws when the snapshot provider itself throws", async () => { + const controller = new PlannerRecoveryController({ + snapshotProvider: { + getSnapshot: () => { + throw new Error("provider exploded"); + }, + }, + }); + await expect(controller.tick(task())).resolves.toBeNull(); + }); + + it("adapts a getObservations()-style source (PlannerOverseerMonitor shape) via its latest observation", async () => { + const retryStep = vi.fn().mockResolvedValue(undefined); + const controller = new PlannerRecoveryController({ + snapshotProvider: { + getObservations: () => [observation({ signal: "progressing" }), observation({ signal: "failed" })], + }, + handlers: { retryStep }, + }); + const decision = await controller.tick(task()); + expect(decision?.action).toBe("retry_step"); + expect(retryStep).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 2ed20e1c55..03f60a9e31 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -647,6 +647,26 @@ export { type OverseerLogStore, type PlannerOverseerMonitorOptions, } from "./planner-overseer.js"; +export { + PlannerRecoveryController, + type PlannerRecoveryContext, + type PlannerRecoveryHandlers, + type PlannerRecoverySnapshotProvider, + type PlannerRecoveryObservationSource, + type PlannerRecoveryControllerOptions, +} from "./planner-recovery-controller.js"; +export { + decidePlannerRecovery, + PLANNER_RECOVERY_MAX_ATTEMPTS, + type PlannerRecoveryActionKind, + type PlannerRecoveryWatchedStage, + type PlannerRecoveryObservationSignal, + type PlannerRecoverySourceLink, + type PlannerRecoveryObservation, + type PlannerRecoveryAttemptState, + type PlannerRecoveryDecision, + type DecidePlannerRecoveryInput, +} from "@fusion/core"; export { SECRET_MUTATION_TYPES, SECRET_AUDIT_PLAINTEXT_FORBIDDEN_KEYS, diff --git a/packages/engine/src/planner-recovery-controller.ts b/packages/engine/src/planner-recovery-controller.ts new file mode 100644 index 0000000000..4d491d3b18 --- /dev/null +++ b/packages/engine/src/planner-recovery-controller.ts @@ -0,0 +1,195 @@ +/** + * FNXC:PlannerOversight 2026-07-04-12:00: + * FN-7512 engine-side dispatcher for bounded autonomous planner recovery. + * Consumes the FN-7511 `PlannerOverseerMonitor` observation (or an injected + * snapshot provider), calls the pure `decidePlannerRecovery` from + * `@fusion/core`, and — ONLY when the observation's effective oversight + * level is `"autonomous"` — dispatches the chosen bounded action (inject + * guidance / retry the step / request a targeted fix) through injected + * `PlannerRecoveryHandlers`. Mirrors the `AutoRecoveryDispatcher` + + * `StuckTaskDetector` conventions: a per-`(taskId, watchedStage)` in-memory + * attempt registry, degrade-to-no-op on any error, never throw. + * + * Minimum guards owned by this task (FN-7514 owns the comprehensive + * human-control safeguards): `tick()` is a no-op when `task.userPaused` is + * true, and no handler here ever performs a merge/PR or destructive/ + * external-service action — those are excluded by construction (only + * `injectGuidance` / `retryStep` / `requestTargetedFix` exist) and are owned + * by FN-7513's confirmation-gated layer. + */ + +import type { PlannerRecoveryDecision, PlannerRecoveryObservation, Task } from "@fusion/core"; +import { decidePlannerRecovery, PLANNER_RECOVERY_MAX_ATTEMPTS } from "@fusion/core"; +import { createLogger, type Logger } from "./logger.js"; +import type { OverseerStageObservation } from "./planner-overseer.js"; + +/** Minimal shared context threaded through to handlers (e.g. a run-id or clock). */ +export interface PlannerRecoveryContext { + now?: () => number; + [key: string]: unknown; +} + +/** + * Side-effecting handlers a caller wires up using ONLY existing mechanisms + * (steering-comment API for guidance/targeted-fix, store retry/re-enqueue + * for step retry). All optional and all async; a missing handler simply + * means that action is not dispatched (degrades to no-op, never throws). + */ +export interface PlannerRecoveryHandlers { + injectGuidance?: (task: Task, decision: PlannerRecoveryDecision, ctx: PlannerRecoveryContext) => Promise; + retryStep?: (task: Task, decision: PlannerRecoveryDecision, ctx: PlannerRecoveryContext) => Promise; + requestTargetedFix?: (task: Task, decision: PlannerRecoveryDecision, ctx: PlannerRecoveryContext) => Promise; +} + +/** Minimal seam for fetching the current watched-stage observation for a task. */ +export interface PlannerRecoverySnapshotProvider { + getSnapshot(taskId: string): OverseerStageObservation | null | undefined | Promise; +} + +/** The delivered `PlannerOverseerMonitor` shape this controller can also accept directly (FN-7511). */ +export interface PlannerRecoveryObservationSource { + getObservations(taskId: string): OverseerStageObservation[]; +} + +export interface PlannerRecoveryControllerOptions { + /** Either a `{ getSnapshot(taskId) }` provider, or a `PlannerOverseerMonitor`-shaped source (adapted via its latest recorded observation). */ + snapshotProvider: PlannerRecoverySnapshotProvider | PlannerRecoveryObservationSource; + handlers?: PlannerRecoveryHandlers; + logger?: Logger; +} + +const controllerLog = createLogger("planner-recovery-controller"); + +function isSnapshotProvider(value: unknown): value is PlannerRecoverySnapshotProvider { + return typeof (value as PlannerRecoverySnapshotProvider)?.getSnapshot === "function"; +} + +function normalizeProvider( + provider: PlannerRecoverySnapshotProvider | PlannerRecoveryObservationSource, +): PlannerRecoverySnapshotProvider { + if (isSnapshotProvider(provider)) { + return provider; + } + const source = provider as PlannerRecoveryObservationSource; + return { + getSnapshot: (taskId: string) => { + const observations = source.getObservations(taskId); + return observations.length > 0 ? observations[observations.length - 1] : null; + }, + }; +} + +/** + * FNXC:PlannerOversight 2026-07-04-12:00: + * Bounded autonomous-recovery dispatcher. Holds a per-`(taskId, + * watchedStage)` attempt registry (in-memory; not persisted — the wider + * intervention timeline is FN-7519's responsibility) and increments it only + * when an action is actually dispatched. Once a stage's attempt count + * reaches `PLANNER_RECOVERY_MAX_ATTEMPTS`, `decidePlannerRecovery` returns + * `exhausted: true` and `tick()` takes no further action for that stage. + */ +export class PlannerRecoveryController { + private readonly snapshotProvider: PlannerRecoverySnapshotProvider; + private readonly handlers: PlannerRecoveryHandlers; + private readonly logger: Logger; + private readonly attempts = new Map(); + + constructor(options: PlannerRecoveryControllerOptions) { + this.snapshotProvider = normalizeProvider(options.snapshotProvider); + this.handlers = options.handlers ?? {}; + this.logger = options.logger ?? controllerLog; + } + + private attemptKey(taskId: string, stage: string): string { + return `${taskId}::${stage}`; + } + + /** + * Evaluate and, when warranted, dispatch one bounded recovery action for + * `task`'s currently watched stage. Never throws — any handler/registry + * error degrades to a no-op. Returns the computed decision (even when no + * action was dispatched) for logging/testing, or `null` when the task is + * user-paused, has no active observation, or the snapshot lookup failed. + */ + async tick(task: Task, ctx: PlannerRecoveryContext = {}): Promise { + try { + if (!task || task.userPaused === true) { + return null; + } + + const snapshot = await this.getSnapshotSafe(task.id); + if (!snapshot) { + return null; + } + + const key = this.attemptKey(task.id, snapshot.stage); + const attemptCount = this.attempts.get(key) ?? 0; + + const decision = decidePlannerRecovery({ + snapshot: snapshot as unknown as PlannerRecoveryObservation, + attemptState: { attemptCount, attemptLimit: PLANNER_RECOVERY_MAX_ATTEMPTS }, + }); + + if (decision.action === "none") { + return decision; + } + + const dispatched = await this.dispatch(decision, task, ctx); + if (dispatched) { + this.attempts.set(key, attemptCount + 1); + } + return decision; + } catch (err) { + this.logger.warn(`tick failed for ${task?.id ?? "?"}: ${(err as Error)?.message ?? String(err)}`); + return null; + } + } + + private async dispatch(decision: PlannerRecoveryDecision, task: Task, ctx: PlannerRecoveryContext): Promise { + try { + if (decision.action === "inject_guidance") { + if (!this.handlers.injectGuidance) return false; + await this.handlers.injectGuidance(task, decision, ctx); + return true; + } + if (decision.action === "retry_step") { + if (!this.handlers.retryStep) return false; + await this.handlers.retryStep(task, decision, ctx); + return true; + } + if (decision.action === "request_targeted_fix") { + if (!this.handlers.requestTargetedFix) return false; + await this.handlers.requestTargetedFix(task, decision, ctx); + return true; + } + return false; + } catch (err) { + this.logger.warn(`handler for action="${decision.action}" failed on ${task.id}: ${(err as Error)?.message ?? String(err)}`); + return false; + } + } + + private async getSnapshotSafe(taskId: string): Promise { + try { + const result = await this.snapshotProvider.getSnapshot(taskId); + return result ?? null; + } catch { + return null; + } + } + + /** Reset all attempt state for `taskId` (every watched stage) — call on terminal task transitions. */ + clear(taskId: string): void { + const prefix = `${taskId}::`; + for (const key of [...this.attempts.keys()]) { + if (key.startsWith(prefix)) { + this.attempts.delete(key); + } + } + } + + /** Test/inspection seam: current attempt count for a `(taskId, watchedStage)` pair. */ + getAttemptCount(taskId: string, stage: string): number { + return this.attempts.get(this.attemptKey(taskId, stage)) ?? 0; + } +} diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts index 6967c4ae87..8b54bae7e7 100644 --- a/packages/engine/src/project-engine.ts +++ b/packages/engine/src/project-engine.ts @@ -21,6 +21,7 @@ import type { WorktreePool } from "./worktree-pool.js"; import type { ProjectRuntimeConfig } from "./project-runtime.js"; import { PrMonitor } from "./pr-monitor.js"; import { PlannerOverseerMonitor } from "./planner-overseer.js"; +import { PlannerRecoveryController, type PlannerRecoveryHandlers } from "./planner-recovery-controller.js"; import type { PrNodeGithubOps } from "./pr-nodes.js"; import { PrReconciler, type PrReconcileGithubOps } from "./pr-reconcile.js"; import { PrCommentHandler } from "./pr-comment-handler.js"; @@ -335,6 +336,18 @@ export class ProjectEngine { private plannerOverseerPollTimer: ReturnType | null = null; /** Conservative poll cadence for the records-only planner-overseer monitor (45s). */ private static readonly PLANNER_OVERSEER_POLL_INTERVAL_MS = 45 * 1000; + /** + * FNXC:PlannerOversight 2026-07-04-12:00: + * FN-7512 bounded autonomous-recovery dispatcher. Consumes the FN-7511 + * `plannerOverseer`'s recorded observations and, ONLY when the task's + * effective planner oversight level resolves to `"autonomous"`, dispatches + * one bounded action per poll tick (inject guidance / retry the step / + * request a targeted fix) through handlers wired to the existing + * steering-comment API and store retry/re-enqueue path. Never merge/PR or + * destructive actions (FN-7513 owns those); comprehensive human-control + * safeguards beyond the userPaused skip are FN-7514's responsibility. + */ + private plannerRecoveryController?: PlannerRecoveryController; private prReconciler?: PrReconciler; private prCommentHandler?: PrCommentHandler; private notifier?: NtfyNotifier; @@ -593,6 +606,14 @@ export class ProjectEngine { // FN-7511: Initialize the records-only planner-overseer monitor and start // its bounded, gated poll over in-flight tasks. this.plannerOverseer = new PlannerOverseerMonitor({ store }); + // FN-7512: bounded autonomous-recovery dispatcher, wired to the existing + // steering-comment API + store retry/re-enqueue path only — no new + // session/tool/merge channel. Ticked from the same poll as the FN-7511 + // observer, guarded to the "autonomous" effective level there. + this.plannerRecoveryController = new PlannerRecoveryController({ + snapshotProvider: this.plannerOverseer, + handlers: this.buildPlannerRecoveryHandlers(store), + }); this.startPlannerOverseerPoll(store); // 2. Initialize PrMonitor + PrCommentHandler @@ -1000,6 +1021,41 @@ export class ProjectEngine { return this.plannerOverseer; } + /** Get the bounded PlannerRecoveryController (if initialized). See FN-7512. */ + getPlannerRecoveryController(): PlannerRecoveryController | undefined { + return this.plannerRecoveryController; + } + + /** + * FNXC:PlannerOversight 2026-07-04-12:00: + * Concrete FN-7512 handler wiring — ONLY reuses existing mechanisms: + * `injectGuidance`/`requestTargetedFix` post a planner-authored steering + * comment via `store.addSteeringComment` (the same channel the executor's + * real-time injection listener already watches); `retryStep` calls the + * store's existing in-progress→todo retry/re-enqueue path + * (`moveTask(id, "todo", { preserveProgress: true })`), preserving + * progress exactly like the auto-recovery/self-healing retry handlers do. + * No new session/tool/merge channel is introduced. + */ + private buildPlannerRecoveryHandlers(store: TaskStore): PlannerRecoveryHandlers { + return { + injectGuidance: async (task, decision) => { + const text = `[planner-oversight] ${decision.reason}`; + await store.addSteeringComment(task.id, text, "agent"); + }, + retryStep: async (task) => { + await store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine" } as Parameters[2]); + }, + requestTargetedFix: async (task, decision) => { + const sourceRef = decision.sourceLinks[0]?.ref; + const text = sourceRef + ? `[planner-oversight] targeted-fix requested: ${decision.reason} (source: ${sourceRef})` + : `[planner-oversight] targeted-fix requested: ${decision.reason}`; + await store.addSteeringComment(task.id, text, "agent"); + }, + }; + } + /** Get the CronRunner (if initialized). */ getCronRunner(): CronRunner | undefined { return this.cronRunner; @@ -1847,6 +1903,15 @@ export class ProjectEngine { continue; } await overseer.observeTask(task, level); + + // FN-7512: one guarded, autonomous-only bounded recovery tick at the + // same passive seam FN-7511 uses for observation. Inert for every + // other effective level ("off"/"observe"/"steer" already `continue`d + // above); `PlannerRecoveryController.tick` itself skips userPaused + // tasks and never throws. + if (level === "autonomous" && this.plannerRecoveryController) { + await this.plannerRecoveryController.tick(task); + } } catch { // Best-effort per-task — never let one task's failure block the poll. } @@ -1857,6 +1922,7 @@ export class ProjectEngine { for (const taskId of overseer.getObservedTaskIds()) { if (!inFlightIds.has(taskId)) { overseer.clear(taskId); + this.plannerRecoveryController?.clear(taskId); } } } catch {