From 3d58260e1a0c30d4d51e9d2d56068f44d6e707d1 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sat, 4 Jul 2026 19:52:39 -0700 Subject: [PATCH] =?UTF-8?q?FN-7551:=20wire=20overseer=20decision=20points?= =?UTF-8?q?=20to=20engine=20emitOverseer*=20fa=C3=A7ade?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wires PlannerOverseerMonitor/PlannerRecoveryController decision points (human-control withholds, confirmation requests/resolutions, and related overseer stages) to the FN-7520 emitOverseer* façade using the real TaskStore, so the planner-oversight intervention timeline now populates from real engine activity instead of staying empty. - Add onConfirmationResolved handler to PlannerRecoveryController, invoked (best-effort, audit-only) from resolveConfirmation for both approved and denied outcomes. - Wire project-engine.ts to call emitOverseerObservation/emitOverseerEscalation/emitOverseerConfirmation at the real engine decision points, deduped per (task, stage[, signal]). - Add planner-overseer-intervention-wiring.test.ts covering the new wiring end-to-end. - Update docs/architecture.md to reflect the wiring. - Add changeset fn-7551-overseer-timeline-wiring.md (patch). Files changed: .changeset/fn-7551-overseer-timeline-wiring.md | 7 + docs/architecture.md | 2 +- .../planner-overseer-intervention-wiring.test.ts | 319 +++++++++++++++++++++ packages/engine/src/planner-recovery-controller.ts | 36 +++ packages/engine/src/project-engine.ts | 248 +++++++++++++++- 5 files changed, 607 insertions(+), 5 deletions(-) Fusion-Task-Id: FN-7551 Fusion-Task-Lineage: 8bcd103e-8797-4ef5-9b68-bd2daec8d26b Co-authored-by: Fusion (runfusion.ai) --- .../fn-7551-overseer-timeline-wiring.md | 7 + docs/architecture.md | 2 +- ...anner-overseer-intervention-wiring.test.ts | 319 ++++++++++++++++++ .../engine/src/planner-recovery-controller.ts | 36 ++ packages/engine/src/project-engine.ts | 248 +++++++++++++- 5 files changed, 607 insertions(+), 5 deletions(-) create mode 100644 .changeset/fn-7551-overseer-timeline-wiring.md create mode 100644 packages/engine/src/__tests__/planner-overseer-intervention-wiring.test.ts diff --git a/.changeset/fn-7551-overseer-timeline-wiring.md b/.changeset/fn-7551-overseer-timeline-wiring.md new file mode 100644 index 0000000000..82d4775708 --- /dev/null +++ b/.changeset/fn-7551-overseer-timeline-wiring.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Planner-oversight intervention timeline now populates from real engine activity. +category: fix +dev: Wires PlannerOverseerMonitor/PlannerRecoveryController decision points to the FN-7520 emitOverseer* façade with the real TaskStore; observation/escalation emission deduped per (task, stage[, signal]). diff --git a/docs/architecture.md b/docs/architecture.md index f3581af30e..329a035bb7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1097,7 +1097,7 @@ Mesh configuration and post-provision managed-node operations are registered sep ### Run Audit API The run-audit system records every mutation performed by the engine across four domains: - **Database** — task:create, task:update, task:move, etc. Node handoff/recovery emits structured events: `node:handoff:parked` (handoff denied/parked), `node:handoff:reassign-local` (local takeover approved), `node:handoff:reassign-any` (any-healthy takeover approved), and `node:lease:recovered` (abandoned lease cleared and task requeued). -- **Database / `overseer:intervention`** (FN-7519, emission façade FN-7520) — the planner-overseer intervention timeline's single canonical mutation type. `target` is the task ID; metadata carries the six intervention field groups (`stage`, `reason`, `action`, `outcome`, optional `attemptCount`/`attemptLimit`, optional `sourceLinks`). Written only via `recordPlannerIntervention` and read via `getPlannerInterventionTimeline`/`parseInterventionEntry` (`packages/core/src/planner-intervention.ts`) so no parallel audit store or timeline-mapping exists; surfaced read-only in the task-detail Intervention Timeline (`GET /tasks/:id/overseer/interventions`). FN-7520 adds the canonical `emitOverseerObservation` / `emitOverseerSteering` / `emitOverseerRecoveryAttempt` / `emitOverseerRetry` / `emitOverseerConfirmation` / `emitOverseerEscalation` emission façade (`packages/core/src/planner-overseer-events.ts`) that fixes each decision-point category's `action`/default `outcome` and funnels through `recordPlannerIntervention` — the single seam FN-7511/FN-7512/FN-7513 call rather than emitting `overseer:intervention` events inline. +- **Database / `overseer:intervention`** (FN-7519, emission façade FN-7520, runtime wiring FN-7551) — the planner-overseer intervention timeline's single canonical mutation type. `target` is the task ID; metadata carries the six intervention field groups (`stage`, `reason`, `action`, `outcome`, optional `attemptCount`/`attemptLimit`, optional `sourceLinks`). Written only via `recordPlannerIntervention` and read via `getPlannerInterventionTimeline`/`parseInterventionEntry` (`packages/core/src/planner-intervention.ts`) so no parallel audit store or timeline-mapping exists; surfaced read-only in the task-detail Intervention Timeline (`GET /tasks/:id/overseer/interventions`). FN-7520 adds the canonical `emitOverseerObservation` / `emitOverseerSteering` / `emitOverseerRecoveryAttempt` / `emitOverseerRetry` / `emitOverseerConfirmation` / `emitOverseerEscalation` emission façade (`packages/core/src/planner-overseer-events.ts`) that fixes each decision-point category's `action`/default `outcome` and funnels through `recordPlannerIntervention` — the single seam FN-7511/FN-7512/FN-7513 call rather than emitting `overseer:intervention` events inline. FN-7551 wires this façade to the LIVE runtime: `ProjectEngine`'s `PlannerOverseerMonitor#onObservation` callback (deduped per `(taskId, stage:signal)`), `buildPlannerRecoveryHandlers` (steering/retry/targeted-fix/confirmation-request, plus `PlannerRecoveryController`'s new optional `onConfirmationResolved` hook for the approve/deny resolution), and `pollPlannerOverseer`'s bounded-recovery-exhaustion escalation (deduped per `(taskId, stage)`) — so the intervention timeline now reflects real engine activity, not only synthetic unit-test entries. - **Git** — worktree:create, worktree:remove, `worktree:remove-fallback` (metadata `{ fallback: "filesystem-non-empty", error }` when native git removal falls back to filesystem removal + admin prune), commit:create, merge:resolve, merge:audit-failure, `worktree:reanchored`, and worktrunk lifecycle events (`worktree:worktrunk-install|create|sync|prune|remove`, plus `worktree:worktrunk-fallback`, `worktree:worktrunk-failure`, and `worktree:worktrunk-fallback-native`). Worktrunk events share metadata `{ op, binaryPath?, worktreePath?, durationMs?, exitCode?, stderrPreview?, installSource?, prunedCount? }` with `installSource` (`"release-binary" | "cargo"`) limited to successful `worktree:worktrunk-install` events and `prunedCount` limited to successful prune events when known. `worktree:worktrunk-install` is emitted only for true install actions; cache hits, configured `worktrunk.binaryPath` overrides, and `$PATH` resolutions intentionally remain silent. Dirty post-merge audit outcomes emit `merge:audit-failure` with metadata `{ mode, strategy, action, reason, issueCount, duplicateSubjectCount, touchedFileOverlapCount, verificationPassed, auditTargetLabel }`. FN-5279 adds `merge:reuse-handoff-acquired`, `merge:reuse-handoff-refused`, `merge:reuse-handoff-released`, and `merge:reuse-handoff-deferred-to-worktrunk` for task-worktree auto-merge handoff visibility. FN-5351 adds `merge:integration-worktree-state` (pre-handoff checkout/dirty snapshot for resolved integration branch), `merge:cwd-integration-fallback-refused` (terminal refusal park event), and `merge:integration-ref-advance` (integration ref advance outcome telemetry). - **Git / `merge:file-scope-violation`** — emitted by the merger when `FileScopeViolationError` aborts a squash. `target` is the task ID; metadata includes `stagedFiles`, `declaredScope`, `resetLabel`, `stagedFileCount`, and `declaredScopeCount`. Consumed by `fileScopeInvariantFailuresPerDay` in `GET /api/health/reliability` (FN-4360). - **Git / `merge:no-op-attribution-mismatch`** — emitted by the rebase landed-files attribution guard (FN-5304) when `..HEAD` has zero attributable own commits but the source `fusion/` tip still carries attributable own commits. `target` is the task ID; metadata includes `recordedSha`, `rebaseMergeBaseSha`, `sourceBranchRef`, `sourceBranchOwnCommitCount`, and `sourceBranchOwnCommitShas`. diff --git a/packages/engine/src/__tests__/planner-overseer-intervention-wiring.test.ts b/packages/engine/src/__tests__/planner-overseer-intervention-wiring.test.ts new file mode 100644 index 0000000000..364057cc77 --- /dev/null +++ b/packages/engine/src/__tests__/planner-overseer-intervention-wiring.test.ts @@ -0,0 +1,319 @@ +/** + * FNXC:PlannerOversight 2026-07-04-19:45: + * FN-7551 engine-level end-to-end test: proves real overseer decision points + * — observation, retry, targeted-fix, steering (reviewer), confirmation + * request, confirmation resolution, and bounded-recovery escalation — + * populate the `overseer:intervention` run-audit timeline via the ACTUAL + * production wiring in `project-engine.ts` (`PlannerOverseerMonitor#onObservation` + * → the private `emitOverseerObservationDeduped`, the private + * `buildPlannerRecoveryHandlers`, the private `emitOverseerEscalationDeduped`), + * against a REAL in-memory `TaskStore` — never by calling + * `emitOverseer*`/`recordPlannerIntervention` directly. + * + * Constructing a full `ProjectEngine` (via `start()`) pulls in cron/ + * notification/research/tunnel/automation subsystems that are impractical to + * boot in a unit test. Instead this file extracts the engine's REAL + * prototype methods via `Object.create(ProjectEngine.prototype)`, seeding + * only the two dedup-map instance fields those methods read/write (normally + * initialized by class-field initializers the constructor never runs here) + * — this exercises the exact same code that runs inside + * `pollPlannerOverseer`/`start()` in production, not a reimplementation. + */ + +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { TaskStore, getPlannerInterventionTimeline, type Task } from "@fusion/core"; +import { ProjectEngine } from "../project-engine.js"; +import { PlannerOverseerMonitor, type OverseerStageObservation } from "../planner-overseer.js"; +import { PlannerRecoveryController, type PlannerRecoveryHandlers, type PlannerRecoverySnapshotProvider } from "../planner-recovery-controller.js"; + +interface EngineOverseerInternals { + plannerObservationEmitDedup: Map; + plannerEscalationEmitDedup: Set; + buildPlannerRecoveryHandlers(store: TaskStore): PlannerRecoveryHandlers; + emitOverseerObservationDeduped(store: TaskStore, observation: OverseerStageObservation): void; + emitOverseerEscalationDeduped( + store: TaskStore, + taskId: string, + decision: { watchedStage: string | null; reason: string; attemptCount: number; attemptLimit: number; sourceLinks: unknown[] }, + ): void; +} + +/** Extracts the real `ProjectEngine` prototype methods FN-7551 wired without running its heavy constructor/`start()`. */ +function makeEngineInternals(): EngineOverseerInternals { + const engineLike = Object.create(ProjectEngine.prototype) as unknown as EngineOverseerInternals; + engineLike.plannerObservationEmitDedup = new Map(); + engineLike.plannerEscalationEmitDedup = new Set(); + return engineLike; +} + +/** Builds the real production wiring (monitor + recovery-controller handlers) against a concrete `store`, mirroring `ProjectEngine.start()`. */ +function wireRealEngineOverseer(store: TaskStore) { + const internals = makeEngineInternals(); + const monitor = new PlannerOverseerMonitor({ + store, + onObservation: (observation) => internals.emitOverseerObservationDeduped(store, observation), + }); + const handlers = internals.buildPlannerRecoveryHandlers(store); + const controllerFromMonitor = new PlannerRecoveryController({ snapshotProvider: monitor, handlers }); + + return { + monitor, + handlers, + controllerFromMonitor, + /** A controller wired with the SAME real handlers but a synthetic snapshot, for exercising decision branches the monitor's own signal-derivation cannot produce (e.g. a failed executor signal). */ + controllerWithSnapshot: (observation: OverseerStageObservation) => { + const provider: PlannerRecoverySnapshotProvider = { getSnapshot: () => observation }; + return new PlannerRecoveryController({ snapshotProvider: provider, handlers }); + }, + emitEscalation: ( + taskId: string, + decision: { watchedStage: string | null; reason: string; attemptCount: number; attemptLimit: number; sourceLinks: unknown[] }, + ) => internals.emitOverseerEscalationDeduped(store, taskId, decision), + }; +} + +function observation(overrides: Partial = {}): OverseerStageObservation { + return { + taskId: "T", + stage: "executor", + signal: "progressing", + oversightLevel: "autonomous", + observedAt: Date.now(), + reason: "test", + sources: [], + ...overrides, + }; +} + +describe("FN-7551 — overseer decision points populate the intervention timeline via the live wiring", () => { + let rootDir: string; + let globalDir: string; + let store: TaskStore; + + beforeEach(async () => { + rootDir = mkdtempSync(join(tmpdir(), "fn-7551-engine-root-")); + globalDir = mkdtempSync(join(tmpdir(), "fn-7551-engine-global-")); + store = new TaskStore(rootDir, globalDir, { inMemoryDb: true }); + await store.init(); + }); + + afterEach(() => { + store.stopWatching(); + store.close(); + rmSync(rootDir, { recursive: true, force: true }); + rmSync(globalDir, { recursive: true, force: true }); + }); + + async function seedTask(column: "in-progress" | "in-review" = "in-progress"): Promise { + const task = await store.createTask({ title: "T", description: "d" }); + await store.moveTask(task.id, "todo"); + await store.moveTask(task.id, "in-progress"); + if (column === "in-review") { + await store.moveTask(task.id, "in-review", { preserveProgress: true } as never); + } + return (await store.getTask(task.id))!; + } + + it("observation: emits exactly one entry per (stage, signal), dedupes an unchanged repeat, and appends on a changed signal", async () => { + const task = await seedTask("in-progress"); + const { monitor } = wireRealEngineOverseer(store); + + await monitor.observeTask(task, "autonomous"); + await monitor.observeTask(task, "autonomous"); + await monitor.observeTask(task, "autonomous"); + + let timeline = getPlannerInterventionTimeline(store, task.id); + expect(timeline).toHaveLength(1); + expect(timeline[0].action).toBe("observe"); + expect(timeline[0].stage).toBe("executor"); + + // Change the signal (progressing -> "blocked" via paused) and observe again — must append. + const pausedTask = { ...task, paused: true, pausedReason: "manual test pause" } as Task; + await monitor.observeTask(pausedTask, "autonomous"); + + timeline = getPlannerInterventionTimeline(store, task.id); + expect(timeline).toHaveLength(2); + expect(timeline[0].action).toBe("observe"); // newest-first + }); + + it("level 'off' never records an observation entry", async () => { + const task = await seedTask("in-progress"); + const { monitor } = wireRealEngineOverseer(store); + const result = await monitor.observeTask(task, "off"); + expect(result).toBeNull(); + expect(getPlannerInterventionTimeline(store, task.id)).toHaveLength(0); + }); + + it("failed executor with no error source dispatches retry_step and emits a retry entry with attemptCount/attemptLimit", async () => { + const task = await seedTask("in-progress"); + const { controllerWithSnapshot } = wireRealEngineOverseer(store); + const controller = controllerWithSnapshot(observation({ taskId: task.id, stage: "executor", signal: "failed", sources: [] })); + + const decision = await controller.tick(task); + expect(decision?.action).toBe("retry_step"); + + const timeline = getPlannerInterventionTimeline(store, task.id); + const retryEntry = timeline.find((e) => e.action === "retry"); + expect(retryEntry).toBeTruthy(); + expect(retryEntry?.stage).toBe("executor"); + expect(retryEntry?.attemptCount).toBe(1); + expect(retryEntry?.attemptLimit).toBe(3); + }); + + it("failed executor WITH an error source (failed-check) dispatches request_targeted_fix and emits a request-fix entry", async () => { + const task = await seedTask("in-progress"); + const { controllerWithSnapshot } = wireRealEngineOverseer(store); + const controller = controllerWithSnapshot( + observation({ + taskId: task.id, + stage: "executor", + signal: "failed", + sources: [{ kind: "failed-check", ref: "lint" }], + }), + ); + + const decision = await controller.tick(task); + expect(decision?.action).toBe("request_targeted_fix"); + + const timeline = getPlannerInterventionTimeline(store, task.id); + const fixEntry = timeline.find((e) => e.action === "request-fix"); + expect(fixEntry).toBeTruthy(); + expect(fixEntry?.attemptCount).toBe(1); + expect(fixEntry?.attemptLimit).toBe(3); + expect(fixEntry?.sourceLinks?.[0]?.target).toBe("lint"); + }); + + it("reviewer stage dispatches inject_guidance and emits an inject-guidance steering entry", async () => { + const task = await seedTask("in-review"); + const { controllerWithSnapshot } = wireRealEngineOverseer(store); + const controller = controllerWithSnapshot(observation({ taskId: task.id, stage: "reviewer", signal: "progressing" })); + + const decision = await controller.tick(task); + expect(decision?.action).toBe("inject_guidance"); + + const timeline = getPlannerInterventionTimeline(store, task.id); + const steeringEntry = timeline.find((e) => e.action === "inject-guidance"); + expect(steeringEntry).toBeTruthy(); + expect(steeringEntry?.stage).toBe("reviewer"); + }); + + it("merger/pull-request confirmation-required decision emits a request-confirmation entry; approving it emits a resolution entry with 'succeeded'", async () => { + const task = await seedTask("in-review"); + const { monitor, controllerFromMonitor: controller } = wireRealEngineOverseer(store); + await monitor.observeTask(task, "autonomous"); // merger stage (plain in-review, no PR/reviewState) + + const decision = await controller.tick(task); + expect(decision?.requiresConfirmation).toBe(true); + + let timeline = getPlannerInterventionTimeline(store, task.id); + const requestEntry = timeline.find((e) => e.action === "request-confirmation"); + expect(requestEntry).toBeTruthy(); + expect(requestEntry?.outcome).toBe("awaiting-confirmation"); + + const pending = controller.getPendingConfirmations(task.id); + expect(pending).toHaveLength(1); + + await controller.resolveConfirmation(task.id, pending[0].requestId, "approved", "test-user"); + + timeline = getPlannerInterventionTimeline(store, task.id); + const confirmationEntries = timeline.filter((e) => e.action === "request-confirmation"); + expect(confirmationEntries.length).toBeGreaterThanOrEqual(2); + expect(confirmationEntries.some((e) => e.outcome === "succeeded")).toBe(true); + }); + + it("denying a confirmation resolution emits a 'skipped' outcome entry", async () => { + const task = await seedTask("in-review"); + const { monitor, controllerFromMonitor: controller } = wireRealEngineOverseer(store); + await monitor.observeTask(task, "autonomous"); + await controller.tick(task); + const pending = controller.getPendingConfirmations(task.id); + expect(pending).toHaveLength(1); + + await controller.resolveConfirmation(task.id, pending[0].requestId, "denied"); + + const timeline = getPlannerInterventionTimeline(store, task.id); + const confirmationEntries = timeline.filter((e) => e.action === "request-confirmation"); + expect(confirmationEntries.some((e) => e.outcome === "skipped")).toBe(true); + }); + + it("bounded-recovery exhaustion emits exactly one escalate entry across repeated polls of the same exhausted stage", async () => { + const task = await seedTask("in-progress"); + const { emitEscalation } = wireRealEngineOverseer(store); + const exhaustedDecision = { + watchedStage: "executor", + reason: 'Bounded recovery attempt budget (3) exhausted for stage "executor"', + attemptCount: 3, + attemptLimit: 3, + sourceLinks: [], + }; + + emitEscalation(task.id, exhaustedDecision); + emitEscalation(task.id, exhaustedDecision); + emitEscalation(task.id, exhaustedDecision); + + const timeline = getPlannerInterventionTimeline(store, task.id); + const escalations = timeline.filter((e) => e.action === "escalate"); + expect(escalations).toHaveLength(1); + expect(escalations[0].outcome).toBe("failed"); + }); + + it("exhaustion actually reached through real tick()s (three denials) emits escalate exactly once thereafter", async () => { + const task = await seedTask("in-review"); + const { monitor, controllerFromMonitor: controller, emitEscalation } = wireRealEngineOverseer(store); + await monitor.observeTask(task, "autonomous"); + + for (let i = 0; i < 3; i += 1) { + const decision = await controller.tick(task); + expect(decision?.requiresConfirmation).toBe(true); + const pending = controller.getPendingConfirmations(task.id); + await controller.resolveConfirmation(task.id, pending[0].requestId, "denied"); + } + + const finalDecision = await controller.tick(task); + expect(finalDecision?.action).toBe("none"); + expect(finalDecision?.exhausted).toBe(true); + + // The poll wires escalation emission itself (project-engine.ts), so drive + // it explicitly here with the real decision object, twice, to prove the dedup. + emitEscalation(task.id, finalDecision!); + emitEscalation(task.id, finalDecision!); + + const timeline = getPlannerInterventionTimeline(store, task.id); + expect(timeline.filter((e) => e.action === "escalate")).toHaveLength(1); + }); + + it("oversight level 'off' and a human-control-withheld (userPaused) task never emit any steering/retry/fix/confirmation/escalation entry", async () => { + const task = await seedTask("in-review"); + const { monitor, controllerFromMonitor: controller } = wireRealEngineOverseer(store); + + // Level "off": monitor records nothing (mirrors the poll's `continue` before ever calling tick()). + const result = await monitor.observeTask(task, "off"); + expect(result).toBeNull(); + expect(getPlannerInterventionTimeline(store, task.id)).toHaveLength(0); + + // Human-control withheld (userPaused) — tick() must short-circuit before + // any confirmation classification/dispatch, so no intervention entry is + // ever recorded (the withhold itself is a separate no-action event this + // task does not touch). + const pausedTask = { ...task, userPaused: true } as Task; + const decision = await controller.tick(pausedTask); + expect(decision).toBeNull(); + expect(getPlannerInterventionTimeline(store, task.id)).toHaveLength(0); + }); + + it("a store/façade failure during observation emission never throws out of observeTask (best-effort contract)", async () => { + const task = await seedTask("in-progress"); + const throwingStore = { + ...store, + recordRunAuditEvent: () => { + throw new Error("boom"); + }, + } as unknown as TaskStore; + + const { monitor } = wireRealEngineOverseer(throwingStore); + await expect(monitor.observeTask(task, "autonomous")).resolves.not.toThrow(); + }); +}); diff --git a/packages/engine/src/planner-recovery-controller.ts b/packages/engine/src/planner-recovery-controller.ts index 58077310db..ee4c2e1aaf 100644 --- a/packages/engine/src/planner-recovery-controller.ts +++ b/packages/engine/src/planner-recovery-controller.ts @@ -30,6 +30,14 @@ * once per (taskId, watchedStage, reason) — repeated `tick()`s for the same * still-withheld reason do not re-emit until the reason changes or the task * leaves the withheld state, so the audit trail isn't spammed every poll. + * + * FNXC:PlannerOversight 2026-07-04-19:45: + * FN-7551 adds the optional `onConfirmationResolved` handler, invoked from + * `resolveConfirmation` for both "approved" and "denied" outcomes so the + * engine wiring (`project-engine.ts`) can emit a matching + * `emitOverseerConfirmation({..., outcome: "succeeded"|"skipped"})` + * resolution entry through the FN-7520 façade. Purely additive/audit-only — + * it does not change the approve/deny execution semantics above. */ import type { PlannerConfirmationRequest, PlannerRecoveryDecision, PlannerRecoveryObservation, Settings, Task } from "@fusion/core"; @@ -105,6 +113,24 @@ export interface PlannerRecoveryHandlers { decision: OverseerHumanControlDecision & { reason: OverseerHumanControlWithholdReason }, ctx: PlannerRecoveryContext, ) => Promise; + /** + * FNXC:PlannerOversight 2026-07-04-19:45: + * FN-7551: audit-only notification invoked from `resolveConfirmation` for + * BOTH `"approved"` and `"denied"` resolutions, mirroring the optional/ + * never-throw contract of `recordHumanControlWithheld`. This handler must + * NOT perform (or influence) the approve/deny execution path itself — + * that remains `executeMergePrAction`/`executeDestructiveExternalAction`, + * invoked separately from `executeApproved`. Callers wire this to + * `emitOverseerConfirmation({..., outcome: resolution === "approved" ? + * "succeeded" : "skipped"})` so the intervention timeline shows both the + * request and its resolution. Optional; a missing handler is a pure no-op. + */ + onConfirmationResolved?: ( + taskId: string, + request: PlannerConfirmationRequest, + resolution: "approved" | "denied", + ctx: PlannerRecoveryContext, + ) => Promise; } /** Minimal seam for fetching the current watched-stage observation for a task. */ @@ -461,6 +487,16 @@ export class PlannerRecoveryController { } } + // FN-7551: audit-only resolution notification, additive to the + // approve/deny execution above — never influences it. Best-effort. + if (this.handlers.onConfirmationResolved) { + try { + await this.handlers.onConfirmationResolved(taskId, resolved, resolution, ctx); + } catch (err) { + this.logger.warn(`onConfirmationResolved handler failed for ${taskId}: ${(err as Error)?.message ?? String(err)}`); + } + } + return resolved; } catch (err) { this.logger.warn(`resolveConfirmation failed for ${taskId}: ${(err as Error)?.message ?? String(err)}`); diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts index 88310ae4e8..2ee050dbd5 100644 --- a/packages/engine/src/project-engine.ts +++ b/packages/engine/src/project-engine.ts @@ -13,8 +13,27 @@ import type { ResearchSynthesisRequest, ResearchSynthesisResult, PlannerOverseerRuntimeSnapshot, + PlannerInterventionSourceLink, + PlannerOversightStage, +} from "@fusion/core"; +import { + allowsAutoMergeProcessing, + compareTasksByPriorityThenAgeAndId, + emitOverseerConfirmation, + emitOverseerEscalation, + emitOverseerObservation, + emitOverseerRecoveryAttempt, + emitOverseerRetry, + emitOverseerSteering, + getTaskHardMergeBlocker, + isSharedBranchGroupMemberIntegration, + isWorkspaceTask, + normalizeMergerMode, + resolveEffectivePlannerOversightLevel, + resolveEffectiveSettings, + resolveMaxAutoMergeRetries, + sortTasksByPriorityThenAgeAndId, } from "@fusion/core"; -import { allowsAutoMergeProcessing, compareTasksByPriorityThenAgeAndId, getTaskHardMergeBlocker, isSharedBranchGroupMemberIntegration, isWorkspaceTask, normalizeMergerMode, resolveEffectivePlannerOversightLevel, resolveEffectiveSettings, resolveMaxAutoMergeRetries, sortTasksByPriorityThenAgeAndId } from "@fusion/core"; import { assemblePlannerOverseerRuntimeSnapshot } from "./planner-overseer-runtime-snapshot.js"; import { execFile } from "node:child_process"; import { promisify } from "node:util"; @@ -351,6 +370,27 @@ export class ProjectEngine { * safeguards beyond the userPaused skip are FN-7514's responsibility. */ private plannerRecoveryController?: PlannerRecoveryController; + /** + * FNXC:PlannerOversight 2026-07-04-19:45: + * FN-7551 requirement: real overseer decision points (observation, + * steering/retry/targeted-fix, confirmation request+resolution, and + * bounded-recovery escalation) must emit exactly one `overseer:intervention` + * run-audit entry through the FN-7520 `emitOverseer*` façade with the real + * `TaskStore`, so the dashboard intervention timeline reflects live engine + * activity instead of only synthetic unit-test entries. Emission must be + * deduped so the 45s poll does not flood the timeline: this map tracks the + * last emitted `"stage:signal"` per taskId for observations, mirroring + * FN-7514's `lastWithheldReason` dedup pattern. Cleared alongside the + * monitor/controller ring buffers whenever a task leaves the in-flight set. + */ + private readonly plannerObservationEmitDedup = new Map(); + /** + * FNXC:PlannerOversight 2026-07-04-19:45: + * FN-7551: tracks `(taskId, stage)` pairs that have already had a bounded- + * recovery escalation emitted so a stage that stays exhausted across many + * subsequent polls emits exactly one `escalate` entry, not one per poll. + */ + private readonly plannerEscalationEmitDedup = new Set(); private prReconciler?: PrReconciler; private prCommentHandler?: PrCommentHandler; private notifier?: NtfyNotifier; @@ -608,7 +648,16 @@ 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 }); + this.plannerOverseer = new PlannerOverseerMonitor({ + store, + // FN-7551: emit one deduped `overseer:intervention` observation entry + // through the FN-7520 façade for each real observation the monitor + // records, using the real TaskStore. Best-effort — never throws (the + // monitor already swallows callback errors around `onObservation`). + onObservation: (observation) => { + this.emitOverseerObservationDeduped(store, observation); + }, + }); // 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 @@ -1123,6 +1172,12 @@ export class ProjectEngine { const updatedTask = await store.updateTask(taskId, { plannerOversightLevel: "off" }); this.plannerOverseer?.clear(taskId); this.plannerRecoveryController?.clear(taskId); + // FN-7551: release the observation/escalation emission-dedup state too, + // so if oversight is later re-enabled for this task, the first new + // observation/escalation emits rather than staying suppressed by stale + // dedup keys from before the stop. + this.plannerObservationEmitDedup.delete(taskId); + this.clearPlannerEscalationDedup(taskId); return { applied: true, reason: "stopped", task: updatedTask }; } catch (err) { @@ -1145,6 +1200,120 @@ export class ProjectEngine { return this.getPlannerOverseerRuntimeSnapshot(taskId); } + /** + * FNXC:PlannerOversight 2026-07-04-19:45: + * FN-7551 mapping helper: converts the `{kind, ref, url?}` source-link shape + * shared by `OverseerSourceLink` (FN-7511 observations) and + * `PlannerRecoverySourceLink` (FN-7512/FN-7513 decisions) into the FN-7520 + * façade's `PlannerInterventionSourceLink` shape (`{kind, label, target, url}`), + * using `ref` as both `label` and `target` when no richer label exists. + * Never throws; an empty/undefined input yields `undefined` so callers can + * omit `sourceLinks` entirely rather than pass an empty array. + */ + private toInterventionSourceLinks( + links: ReadonlyArray<{ kind: string; ref: string; url?: string }> | undefined, + ): PlannerInterventionSourceLink[] | undefined { + if (!links || links.length === 0) return undefined; + return links.map((link) => ({ + kind: link.kind as PlannerInterventionSourceLink["kind"], + label: link.ref || link.kind, + target: link.ref, + url: link.url, + })); + } + + /** + * FNXC:PlannerOversight 2026-07-04-19:45: + * FN-7551: best-effort wrapper shared by every non-observation emission + * call-site (steering/retry/targeted-fix/confirmation/escalation) — + * swallows and logs any façade/store failure so an audit-emission error + * never breaks the dispatching handler or the poll (mirrors the + * try/catch-degrade-to-no-op contract every FN-7512/FN-7513/FN-7514 + * handler already follows). + */ + private emitOverseerInterventionSafe(fn: () => void): void { + try { + fn(); + } catch (err) { + runtimeLog.warn(`Failed to emit overseer intervention: ${err instanceof Error ? err.message : String(err)}`); + } + } + + /** + * FNXC:PlannerOversight 2026-07-04-19:45: + * FN-7551: emits one `overseer:intervention` `observe` entry through + * `emitOverseerObservation` for a real `OverseerStageObservation`, deduped + * per `(taskId, stage:signal)` so a 45s poll of an unchanged watched stage + * does not append a new observation entry every cycle — only a changed + * `(stage, signal)` pair emits. Best-effort: any store/façade failure is + * swallowed so it never breaks `PlannerOverseerMonitor#observeTask`/the poll. + */ + private emitOverseerObservationDeduped(store: TaskStore, observation: import("./planner-overseer.js").OverseerStageObservation): void { + try { + const dedupKey = `${observation.stage}:${observation.signal}`; + const last = this.plannerObservationEmitDedup.get(observation.taskId); + if (last === dedupKey) { + return; + } + this.plannerObservationEmitDedup.set(observation.taskId, dedupKey); + emitOverseerObservation({ + store, + taskId: observation.taskId, + stage: observation.stage, + reason: observation.reason, + sourceLinks: this.toInterventionSourceLinks(observation.sources), + }); + } catch (err) { + runtimeLog.warn( + `Failed to emit overseer observation intervention for ${observation.taskId}: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + + /** + * FNXC:PlannerOversight 2026-07-04-19:45: + * FN-7551: emits one `overseer:intervention` `escalate` entry through + * `emitOverseerEscalation` when a `(taskId, stage)` pair's bounded-recovery + * budget is exhausted, deduped so it is emitted exactly once while the + * stage stays exhausted across subsequent polls (a stage that later + * un-exhausts — e.g. cleared via `clear(taskId)` on terminal transition — + * clears the dedup entry and may escalate again in a future exhaustion). + * Best-effort; never throws out of the poll. + */ + private emitOverseerEscalationDeduped( + store: TaskStore, + taskId: string, + decision: { watchedStage: PlannerOversightStage | null; reason: string; attemptCount: number; attemptLimit: number; sourceLinks: ReadonlyArray<{ kind: string; ref: string; url?: string }> }, + ): void { + if (!decision.watchedStage) return; + const dedupKey = `${taskId}::${decision.watchedStage}`; + if (this.plannerEscalationEmitDedup.has(dedupKey)) { + return; + } + this.plannerEscalationEmitDedup.add(dedupKey); + this.emitOverseerInterventionSafe(() => + emitOverseerEscalation({ + store, + taskId, + stage: decision.watchedStage as PlannerOversightStage, + reason: decision.reason, + attemptCount: decision.attemptCount, + attemptLimit: decision.attemptLimit, + sourceLinks: this.toInterventionSourceLinks(decision.sourceLinks), + }), + ); + } + + /** FN-7551: clears any escalation-dedup entries for `taskId` across every watched stage. */ + private clearPlannerEscalationDedup(taskId: string): void { + const prefix = `${taskId}::`; + for (const key of [...this.plannerEscalationEmitDedup]) { + if (key.startsWith(prefix)) { + this.plannerEscalationEmitDedup.delete(key); + } + } + } + /** * FNXC:PlannerOversight 2026-07-04-12:00: * Concrete FN-7512 handler wiring — ONLY reuses existing mechanisms: @@ -1161,9 +1330,34 @@ export class ProjectEngine { injectGuidance: async (task, decision) => { const text = `[planner-oversight] ${decision.reason}`; await store.addSteeringComment(task.id, text, "agent"); + // FN-7551: emit the steering intervention entry AFTER the steering + // comment succeeds, through the real store, so the timeline reflects + // the same guidance the agent actually saw. + this.emitOverseerInterventionSafe(() => + emitOverseerSteering({ + store, + taskId: task.id, + stage: (decision.watchedStage ?? "executor") as PlannerOversightStage, + reason: decision.reason, + sourceLinks: this.toInterventionSourceLinks(decision.sourceLinks), + }), + ); }, - retryStep: async (task) => { + retryStep: async (task, decision) => { await store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine" } as Parameters[2]); + // FN-7551: the attempt just dispatched — record it as attemptCount + 1 + // (decision.attemptCount is the count BEFORE this dispatch). + this.emitOverseerInterventionSafe(() => + emitOverseerRetry({ + store, + taskId: task.id, + stage: (decision.watchedStage ?? "executor") as PlannerOversightStage, + reason: decision.reason, + attemptCount: decision.attemptCount + 1, + attemptLimit: decision.attemptLimit, + sourceLinks: this.toInterventionSourceLinks(decision.sourceLinks), + }), + ); }, requestTargetedFix: async (task, decision) => { const sourceRef = decision.sourceLinks[0]?.ref; @@ -1171,6 +1365,17 @@ export class ProjectEngine { ? `[planner-oversight] targeted-fix requested: ${decision.reason} (source: ${sourceRef})` : `[planner-oversight] targeted-fix requested: ${decision.reason}`; await store.addSteeringComment(task.id, text, "agent"); + this.emitOverseerInterventionSafe(() => + emitOverseerRecoveryAttempt({ + store, + taskId: task.id, + stage: (decision.watchedStage ?? "executor") as PlannerOversightStage, + reason: decision.reason, + attemptCount: decision.attemptCount + 1, + attemptLimit: decision.attemptLimit, + sourceLinks: this.toInterventionSourceLinks(decision.sourceLinks), + }), + ); }, // FNXC:PlannerOversight 2026-07-04-13:00: // FN-7513 requirement: merge/PR actions beyond guidance/retry, and any @@ -1184,6 +1389,33 @@ export class ProjectEngine { requestConfirmation: async (task, request) => { const text = `[planner-oversight] confirmation required (${request.sideEffectClass}): ${request.reason}`; await store.addSteeringComment(task.id, text, "agent"); + this.emitOverseerInterventionSafe(() => + emitOverseerConfirmation({ + store, + taskId: task.id, + stage: request.watchedStage as PlannerOversightStage, + reason: request.reason, + sourceLinks: this.toInterventionSourceLinks(request.sourceLinks), + }), + ); + }, + // FNXC:PlannerOversight 2026-07-04-19:45: + // FN-7551: audit-only confirmation-RESOLUTION emission. Invoked from + // `PlannerRecoveryController.resolveConfirmation` for both "approved" + // and "denied" outcomes, mirroring the request-path emission above so + // the timeline shows both the request and its resolution. Never touches + // the approve/deny execution path itself. + onConfirmationResolved: async (taskId, request, resolution) => { + this.emitOverseerInterventionSafe(() => + emitOverseerConfirmation({ + store, + taskId, + stage: request.watchedStage as PlannerOversightStage, + reason: request.reason, + outcome: resolution === "approved" ? "succeeded" : "skipped", + sourceLinks: this.toInterventionSourceLinks(request.sourceLinks), + }), + ); }, // FNXC:PlannerOversight 2026-07-04-14:30: // FN-7513 code-review fix: a `"merge_pr"`-classified confirmation covers @@ -2111,7 +2343,13 @@ export class ProjectEngine { // human-review) BEFORE any action/confirmation classification — // never throws. if (level === "autonomous" && this.plannerRecoveryController) { - await this.plannerRecoveryController.tick(task, { settings: engineSettings }); + const decision = await this.plannerRecoveryController.tick(task, { settings: engineSettings }); + // FN-7551: bounded-recovery exhaustion is an escalation-worthy + // event — emit exactly one `escalate` entry per (taskId, stage) + // while the stage remains exhausted across subsequent polls. + if (decision?.exhausted && decision.watchedStage) { + this.emitOverseerEscalationDeduped(store, task.id, decision); + } } } catch { // Best-effort per-task — never let one task's failure block the poll. @@ -2124,6 +2362,8 @@ export class ProjectEngine { if (!inFlightIds.has(taskId)) { overseer.clear(taskId); this.plannerRecoveryController?.clear(taskId); + this.plannerObservationEmitDedup.delete(taskId); + this.clearPlannerEscalationDedup(taskId); } } } catch {