FN-7551: wire overseer decision points to engine emitOverseer* façade
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) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-7551-overseer-timeline-wiring.md
Normal file
7
.changeset/fn-7551-overseer-timeline-wiring.md
Normal file
@@ -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]).
|
||||
@@ -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 `<rebaseBaseSha>..HEAD` has zero attributable own commits but the source `fusion/<id>` tip still carries attributable own commits. `target` is the task ID; metadata includes `recordedSha`, `rebaseMergeBaseSha`, `sourceBranchRef`, `sourceBranchOwnCommitCount`, and `sourceBranchOwnCommitShas`.
|
||||
|
||||
@@ -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<string, string>;
|
||||
plannerEscalationEmitDedup: Set<string>;
|
||||
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> = {}): 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<Task> {
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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<void>;
|
||||
/**
|
||||
* 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<void>;
|
||||
}
|
||||
|
||||
/** 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)}`);
|
||||
|
||||
@@ -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<string, string>();
|
||||
/**
|
||||
* 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<string>();
|
||||
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<TaskStore["moveTask"]>[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 {
|
||||
|
||||
Reference in New Issue
Block a user