FN-7511: add planner overseer stage monitoring
Add records-only planner overseer monitoring across in-flight task lifecycle stages. - Add a PlannerOverseerMonitor with normalized observations and deterministic watched-stage resolution. - Wire ProjectEngine to poll in-progress and in-review tasks, gated by effective planner oversight level. - Document the monitoring seam and add focused coverage plus a release changeset. Files changed: .changeset/fn-7511-planner-overseer-monitoring.md | 7 + docs/architecture.md | 26 ++ docs/workflow-steps.md | 2 + .../engine/src/__tests__/planner-overseer.test.ts | 294 ++++++++++++++++++ packages/engine/src/index.ts | 12 + packages/engine/src/planner-overseer.ts | 338 +++++++++++++++++++++ packages/engine/src/project-engine.ts | 93 +++++- 7 files changed, 771 insertions(+), 1 deletion(-) Fusion-Task-Id: FN-7511 Fusion-Task-Lineage: 81b616cf-47e9-4769-b02d-fc7ebd3fcb2f Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-7511-planner-overseer-monitoring.md
Normal file
7
.changeset/fn-7511-planner-overseer-monitoring.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": minor
|
||||
---
|
||||
|
||||
summary: Planner oversight now monitors tasks across executor, reviewer, merger, pull-request, and workflow-gate stages.
|
||||
category: feature
|
||||
dev: Adds records-only PlannerOverseerMonitor + resolveWatchedStage + OverseerStageObservation in @fusion/engine, gated by resolveEffectivePlannerOversightLevel (off = no observation) and wired into ProjectEngine via a bounded poll. Steering/recovery and UI land in FN-7512/FN-7515+.
|
||||
@@ -1352,6 +1352,32 @@ Limits are controlled by project settings (`maxSpawnedAgentsPerParent`, `maxSpaw
|
||||
### Custom instructions
|
||||
`packages/engine/src/agent-instructions.ts` resolves per-agent instruction text/path with path-traversal and extension validation.
|
||||
|
||||
### Planner overseer monitoring (records-only)
|
||||
|
||||
/*
|
||||
FNXC:PlannerOversight 2026-07-04-00:00:
|
||||
FN-7511 delivers the monitoring foundation for a planner-oversight layer that watches an in-flight task's
|
||||
lifecycle without steering it. `packages/engine/src/planner-overseer.ts` declares the five watched stages
|
||||
(`OVERSEER_WATCHED_STAGES`: executor, reviewer, merger, pull-request, workflow-gate), a normalized
|
||||
`OverseerStageObservation` model, and a `resolveWatchedStage(task)` resolver with deterministic precedence
|
||||
(workflow-gate > pull-request > merger > reviewer > executor) so a task in a compound state resolves to
|
||||
exactly one stage. `PlannerOverseerMonitor#observeTask(task, level)` is the gating seam: when the task's
|
||||
effective planner oversight level (`resolveEffectivePlannerOversightLevel`, FN-7508/FN-7509/FN-7510) is
|
||||
`"off"`, nothing is recorded; otherwise exactly one observation is recorded into a bounded per-task ring
|
||||
buffer (default cap 20) and the optional `onObservation` callback is invoked best-effort.
|
||||
*/
|
||||
|
||||
`ProjectEngine` constructs a `PlannerOverseerMonitor` alongside `PrMonitor` and exposes it via
|
||||
`getPlannerOverseer()`. A bounded `setInterval` poll (45s cadence, cleared on `stop()`) walks the
|
||||
current `in-progress`/`in-review` tasks, resolves each task's effective planner oversight level, and
|
||||
calls `observeTask` — skipping tasks that resolve to `"off"` or to no watched stage. Observations for
|
||||
tasks that leave the in-flight set are dropped from the ring buffer on the next poll.
|
||||
|
||||
This layer is **records-only**: no lifecycle mutation, retry, merge, notification, or external-service
|
||||
call happens here, and it emits no run-audit events or dashboard UI. Steering/recovery, confirmation
|
||||
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.
|
||||
|
||||
---
|
||||
|
||||
## 11) Multi-Project Architecture
|
||||
|
||||
@@ -461,6 +461,8 @@ A gate node also has a `gateMode`:
|
||||
Defaults:
|
||||
- gates are `advisory` by default (advisory-by-default per FN-4368); opt in to `gate` by setting the node's `gateMode` in the [Workflow Editor](./workflow-editor.md).
|
||||
|
||||
A task paused awaiting a prompt/script gate's cli-approval or ask-input response (`pausedReason` prefixed `workflow-cli-approval:` or `workflow-input:`) is one of the five stages the records-only planner-overseer monitor watches (`workflow-gate`, `packages/engine/src/planner-overseer.ts`, FN-7511) — see **`docs/architecture.md` § "Planner overseer monitoring (records-only)"**.
|
||||
|
||||
## Built-In Quality Gates
|
||||
|
||||
<!--
|
||||
|
||||
294
packages/engine/src/__tests__/planner-overseer.test.ts
Normal file
294
packages/engine/src/__tests__/planner-overseer.test.ts
Normal file
@@ -0,0 +1,294 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
OVERSEER_WATCHED_STAGES,
|
||||
PlannerOverseerMonitor,
|
||||
resolveWatchedStage,
|
||||
type OverseerTaskRef,
|
||||
} from "../planner-overseer.js";
|
||||
|
||||
function taskFixture(overrides: Partial<OverseerTaskRef> = {}): OverseerTaskRef {
|
||||
return {
|
||||
id: "FN-1000",
|
||||
column: "in-progress",
|
||||
prInfo: undefined,
|
||||
reviewState: undefined,
|
||||
paused: false,
|
||||
pausedReason: undefined,
|
||||
workflowTransitionNotification: undefined,
|
||||
...overrides,
|
||||
} as OverseerTaskRef;
|
||||
}
|
||||
|
||||
describe("resolveWatchedStage", () => {
|
||||
it("resolves an active in-progress task to executor", () => {
|
||||
expect(resolveWatchedStage(taskFixture({ column: "in-progress" }))).toBe("executor");
|
||||
});
|
||||
|
||||
it("resolves in-review with a pending reviewState to reviewer", () => {
|
||||
expect(
|
||||
resolveWatchedStage(
|
||||
taskFixture({
|
||||
column: "in-review",
|
||||
reviewState: {
|
||||
source: "reviewer-agent",
|
||||
items: [],
|
||||
addressing: [],
|
||||
} as unknown as OverseerTaskRef["reviewState"],
|
||||
}),
|
||||
),
|
||||
).toBe("reviewer");
|
||||
});
|
||||
|
||||
it("resolves in-review with no reviewState/PR/gate marker to merger (awaiting integration)", () => {
|
||||
expect(resolveWatchedStage(taskFixture({ column: "in-review" }))).toBe("merger");
|
||||
});
|
||||
|
||||
it("resolves in-review with an explicit manual-merge-hold marker to merger", () => {
|
||||
expect(
|
||||
resolveWatchedStage(
|
||||
taskFixture({
|
||||
column: "in-review",
|
||||
reviewState: {
|
||||
source: "reviewer-agent",
|
||||
items: [],
|
||||
addressing: [],
|
||||
} as unknown as OverseerTaskRef["reviewState"],
|
||||
workflowTransitionNotification: {
|
||||
kind: "manual-merge-hold",
|
||||
column: "in-review",
|
||||
transitionId: "t-1",
|
||||
} as unknown as OverseerTaskRef["workflowTransitionNotification"],
|
||||
}),
|
||||
),
|
||||
).toBe("merger");
|
||||
});
|
||||
|
||||
it("resolves in-review with an active (open) PR to pull-request", () => {
|
||||
expect(
|
||||
resolveWatchedStage(
|
||||
taskFixture({
|
||||
column: "in-review",
|
||||
prInfo: {
|
||||
url: "https://github.com/o/r/pull/1",
|
||||
number: 1,
|
||||
status: "open",
|
||||
title: "t",
|
||||
headBranch: "h",
|
||||
baseBranch: "b",
|
||||
commentCount: 0,
|
||||
} as unknown as OverseerTaskRef["prInfo"],
|
||||
}),
|
||||
),
|
||||
).toBe("pull-request");
|
||||
});
|
||||
|
||||
it("resolves a paused workflow-cli-approval gate to workflow-gate regardless of column", () => {
|
||||
expect(
|
||||
resolveWatchedStage(
|
||||
taskFixture({
|
||||
column: "in-progress",
|
||||
paused: true,
|
||||
pausedReason: "workflow-cli-approval:build: npm run build",
|
||||
}),
|
||||
),
|
||||
).toBe("workflow-gate");
|
||||
});
|
||||
|
||||
it("resolves a paused workflow-input gate to workflow-gate", () => {
|
||||
expect(
|
||||
resolveWatchedStage(
|
||||
taskFixture({
|
||||
column: "in-review",
|
||||
paused: true,
|
||||
pausedReason: "workflow-input:ask: What environment?",
|
||||
}),
|
||||
),
|
||||
).toBe("workflow-gate");
|
||||
});
|
||||
|
||||
it("returns null for non-monitorable columns (todo/done/archived/triage)", () => {
|
||||
for (const column of ["todo", "done", "archived", "triage"] as const) {
|
||||
expect(resolveWatchedStage(taskFixture({ column }))).toBeNull();
|
||||
}
|
||||
});
|
||||
|
||||
it("deterministic precedence: workflow-gate wins over an active PR and reviewState in a compound state", () => {
|
||||
const compound = taskFixture({
|
||||
column: "in-review",
|
||||
paused: true,
|
||||
pausedReason: "workflow-cli-approval:deploy: run deploy",
|
||||
prInfo: {
|
||||
url: "https://github.com/o/r/pull/2",
|
||||
number: 2,
|
||||
status: "open",
|
||||
title: "t",
|
||||
headBranch: "h",
|
||||
baseBranch: "b",
|
||||
commentCount: 0,
|
||||
} as unknown as OverseerTaskRef["prInfo"],
|
||||
reviewState: {
|
||||
source: "reviewer-agent",
|
||||
items: [],
|
||||
addressing: [],
|
||||
} as unknown as OverseerTaskRef["reviewState"],
|
||||
});
|
||||
expect(resolveWatchedStage(compound)).toBe("workflow-gate");
|
||||
});
|
||||
|
||||
it("deterministic precedence: an active PR wins over reviewState when no gate is paused", () => {
|
||||
const compound = taskFixture({
|
||||
column: "in-review",
|
||||
prInfo: {
|
||||
url: "https://github.com/o/r/pull/3",
|
||||
number: 3,
|
||||
status: "open",
|
||||
title: "t",
|
||||
headBranch: "h",
|
||||
baseBranch: "b",
|
||||
commentCount: 0,
|
||||
} as unknown as OverseerTaskRef["prInfo"],
|
||||
reviewState: {
|
||||
source: "reviewer-agent",
|
||||
items: [],
|
||||
addressing: [],
|
||||
} as unknown as OverseerTaskRef["reviewState"],
|
||||
});
|
||||
expect(resolveWatchedStage(compound)).toBe("pull-request");
|
||||
});
|
||||
|
||||
it("never throws on a malformed/partial task (missing column/optional fields)", () => {
|
||||
expect(resolveWatchedStage({} as OverseerTaskRef)).toBeNull();
|
||||
expect(resolveWatchedStage(null)).toBeNull();
|
||||
expect(resolveWatchedStage(undefined)).toBeNull();
|
||||
expect(resolveWatchedStage({ id: "FN-1" } as OverseerTaskRef)).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe("PlannerOverseerMonitor.observeTask", () => {
|
||||
const stageFixtures: Array<{ stage: (typeof OVERSEER_WATCHED_STAGES)[number]; task: OverseerTaskRef }> = [
|
||||
{ stage: "executor", task: taskFixture({ column: "in-progress" }) },
|
||||
{
|
||||
stage: "reviewer",
|
||||
task: taskFixture({
|
||||
column: "in-review",
|
||||
reviewState: { source: "reviewer-agent", items: [], addressing: [] } as unknown as OverseerTaskRef["reviewState"],
|
||||
}),
|
||||
},
|
||||
{ stage: "merger", task: taskFixture({ column: "in-review" }) },
|
||||
{
|
||||
stage: "pull-request",
|
||||
task: taskFixture({
|
||||
column: "in-review",
|
||||
prInfo: {
|
||||
url: "https://github.com/o/r/pull/9",
|
||||
number: 9,
|
||||
status: "open",
|
||||
title: "t",
|
||||
headBranch: "h",
|
||||
baseBranch: "b",
|
||||
commentCount: 0,
|
||||
} as unknown as OverseerTaskRef["prInfo"],
|
||||
}),
|
||||
},
|
||||
{
|
||||
stage: "workflow-gate",
|
||||
task: taskFixture({ column: "in-progress", paused: true, pausedReason: "workflow-input:ask: env?" }),
|
||||
},
|
||||
];
|
||||
|
||||
it.each(stageFixtures)(
|
||||
"records exactly one observation for the $stage stage when level is not off",
|
||||
async ({ stage, task }) => {
|
||||
for (const level of ["observe", "steer", "autonomous"] as const) {
|
||||
const monitor = new PlannerOverseerMonitor();
|
||||
const observation = await monitor.observeTask(task, level);
|
||||
expect(observation).not.toBeNull();
|
||||
expect(observation?.stage).toBe(stage);
|
||||
expect(observation?.oversightLevel).toBe(level);
|
||||
expect(observation?.taskId).toBe(task.id);
|
||||
expect(observation?.sources.length).toBeGreaterThan(0);
|
||||
expect(monitor.getObservations(task.id)).toHaveLength(1);
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it.each(stageFixtures)("records nothing and returns null for the $stage stage when level is off", async ({ task }) => {
|
||||
const monitor = new PlannerOverseerMonitor();
|
||||
const observation = await monitor.observeTask(task, "off");
|
||||
expect(observation).toBeNull();
|
||||
expect(monitor.getObservations(task.id)).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("returns null and records nothing when no stage is monitorable", async () => {
|
||||
const monitor = new PlannerOverseerMonitor();
|
||||
const observation = await monitor.observeTask(taskFixture({ column: "todo" }), "autonomous");
|
||||
expect(observation).toBeNull();
|
||||
expect(monitor.getObservations("FN-1000")).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("invokes the onObservation callback when provided", async () => {
|
||||
const onObservation = vi.fn().mockResolvedValue(undefined);
|
||||
const monitor = new PlannerOverseerMonitor({ onObservation });
|
||||
const task = taskFixture({ column: "in-progress" });
|
||||
const observation = await monitor.observeTask(task, "observe");
|
||||
expect(onObservation).toHaveBeenCalledTimes(1);
|
||||
expect(onObservation).toHaveBeenCalledWith(observation);
|
||||
});
|
||||
|
||||
it("still resolves when the onObservation callback throws (best-effort)", async () => {
|
||||
const onObservation = vi.fn().mockRejectedValue(new Error("callback exploded"));
|
||||
const monitor = new PlannerOverseerMonitor({ onObservation });
|
||||
const task = taskFixture({ column: "in-progress" });
|
||||
await expect(monitor.observeTask(task, "observe")).resolves.not.toBeNull();
|
||||
expect(monitor.getObservations(task.id)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("records into the store best-effort and swallows logEntry failures", async () => {
|
||||
const store = { logEntry: vi.fn().mockRejectedValue(new Error("log failed")) };
|
||||
const monitor = new PlannerOverseerMonitor({ store });
|
||||
const task = taskFixture({ column: "in-progress" });
|
||||
await expect(monitor.observeTask(task, "observe")).resolves.not.toBeNull();
|
||||
expect(store.logEntry).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("bounds the per-task ring buffer to the configured cap, keeping the most recent N", async () => {
|
||||
const monitor = new PlannerOverseerMonitor({ maxObservationsPerTask: 3 });
|
||||
const task = taskFixture({ column: "in-progress" });
|
||||
const observations = [];
|
||||
for (let i = 0; i < 5; i++) {
|
||||
const obs = await monitor.observeTask(task, "observe");
|
||||
observations.push(obs);
|
||||
}
|
||||
const retained = monitor.getObservations(task.id);
|
||||
expect(retained).toHaveLength(3);
|
||||
// The three retained entries should be the last three recorded (index 2,3,4).
|
||||
expect(retained.map((o) => o.observedAt)).toEqual(
|
||||
[observations[2], observations[3], observations[4]].map((o) => o!.observedAt),
|
||||
);
|
||||
});
|
||||
|
||||
it("defaults the ring buffer cap to 20 entries per task", async () => {
|
||||
const monitor = new PlannerOverseerMonitor();
|
||||
const task = taskFixture({ column: "in-progress" });
|
||||
for (let i = 0; i < 25; i++) {
|
||||
await monitor.observeTask(task, "observe");
|
||||
}
|
||||
expect(monitor.getObservations(task.id)).toHaveLength(20);
|
||||
});
|
||||
|
||||
it("clear() removes retained observations for a task", async () => {
|
||||
const monitor = new PlannerOverseerMonitor();
|
||||
const task = taskFixture({ column: "in-progress" });
|
||||
await monitor.observeTask(task, "observe");
|
||||
expect(monitor.getObservations(task.id)).toHaveLength(1);
|
||||
monitor.clear(task.id);
|
||||
expect(monitor.getObservations(task.id)).toHaveLength(0);
|
||||
expect(monitor.getObservedTaskIds()).not.toContain(task.id);
|
||||
});
|
||||
|
||||
it("never throws on a malformed/partial task passed to observeTask (degrades to no-op)", async () => {
|
||||
const monitor = new PlannerOverseerMonitor();
|
||||
await expect(monitor.observeTask({} as OverseerTaskRef, "autonomous")).resolves.toBeNull();
|
||||
await expect(monitor.observeTask(undefined as unknown as OverseerTaskRef, "autonomous")).resolves.toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -635,6 +635,18 @@ export {
|
||||
type LLMSynthesisProviderOptions,
|
||||
} from "./research/providers/index.js";
|
||||
export { PrMonitor, type PrComment, type TrackedPr, type OnNewCommentsCallback } from "./pr-monitor.js";
|
||||
export {
|
||||
PlannerOverseerMonitor,
|
||||
OVERSEER_WATCHED_STAGES,
|
||||
resolveWatchedStage,
|
||||
type OverseerWatchedStage,
|
||||
type OverseerObservationSignal,
|
||||
type OverseerStageObservation,
|
||||
type OverseerSourceLink,
|
||||
type OverseerTaskRef,
|
||||
type OverseerLogStore,
|
||||
type PlannerOverseerMonitorOptions,
|
||||
} from "./planner-overseer.js";
|
||||
export {
|
||||
SECRET_MUTATION_TYPES,
|
||||
SECRET_AUDIT_PLAINTEXT_FORBIDDEN_KEYS,
|
||||
|
||||
338
packages/engine/src/planner-overseer.ts
Normal file
338
packages/engine/src/planner-overseer.ts
Normal file
@@ -0,0 +1,338 @@
|
||||
/**
|
||||
* FNXC:PlannerOversight 2026-07-04-00:00:
|
||||
* FN-7511 delivers the monitoring foundation for the planner overseer: an
|
||||
* engine module that watches an in-flight task's progression across five
|
||||
* lifecycle stages — executor, reviewer, merger, pull-request, and
|
||||
* workflow-gate — and records normalized `OverseerStageObservation`s gated by
|
||||
* the task's effective planner oversight level (`resolveEffectivePlannerOversightLevel`,
|
||||
* FN-7508/FN-7509/FN-7510). When the effective level is `"off"`, nothing is
|
||||
* recorded. This layer is records-only: it does not steer, retry, fix, gate,
|
||||
* or notify — those land in FN-7512 (steering/recovery), FN-7513
|
||||
* (confirmation gates), FN-7514 (human-control safeguards), and
|
||||
* FN-7515–FN-7520 (dashboard UI / run-audit events / intervention timeline).
|
||||
* The observation model + `PlannerOverseerMonitor` registry declared here is
|
||||
* the seam every later planner-oversight subtask reads from.
|
||||
*/
|
||||
|
||||
import type { PlannerOversightLevel, PrInfo, Task } from "@fusion/core";
|
||||
|
||||
/** Alias for the `Task.reviewState` shape without requiring a separate core export. */
|
||||
type OverseerTaskReviewState = NonNullable<Task["reviewState"]>;
|
||||
|
||||
/**
|
||||
* The five lifecycle stages the planner overseer watches. Precedence when a
|
||||
* task is in a compound state (see {@link resolveWatchedStage}):
|
||||
* workflow-gate > pull-request > merger > reviewer > executor.
|
||||
*/
|
||||
export const OVERSEER_WATCHED_STAGES = ["executor", "reviewer", "merger", "pull-request", "workflow-gate"] as const;
|
||||
export type OverseerWatchedStage = (typeof OVERSEER_WATCHED_STAGES)[number];
|
||||
|
||||
/** Normalized signal describing how a watched stage is currently progressing. */
|
||||
export type OverseerObservationSignal = "progressing" | "stuck" | "failed" | "blocked" | "awaiting-human" | "complete";
|
||||
|
||||
/** A link back to the concrete evidence an observation was derived from. */
|
||||
export interface OverseerSourceLink {
|
||||
kind: "agent-log" | "review-comment" | "failed-check" | "merge-error" | "pr-state";
|
||||
ref: string;
|
||||
url?: string;
|
||||
}
|
||||
|
||||
/** One normalized, oversight-gated observation of a task's current watched stage. */
|
||||
export interface OverseerStageObservation {
|
||||
taskId: string;
|
||||
stage: OverseerWatchedStage;
|
||||
signal: OverseerObservationSignal;
|
||||
oversightLevel: PlannerOversightLevel;
|
||||
observedAt: number;
|
||||
reason: string;
|
||||
sources: OverseerSourceLink[];
|
||||
}
|
||||
|
||||
/** The minimal task shape the stage resolver reads. Kept a `Pick` so callers
|
||||
* (and tests) can pass partial/malformed fixtures without satisfying the
|
||||
* full `Task` interface. */
|
||||
export type OverseerTaskRef = Pick<
|
||||
Task,
|
||||
"id" | "column" | "prInfo" | "reviewState" | "paused" | "pausedReason" | "workflowTransitionNotification"
|
||||
>;
|
||||
|
||||
/**
|
||||
* FNXC:PlannerOversight 2026-07-04-00:00:
|
||||
* Maps a task's delivered lifecycle state to exactly one watched stage, or
|
||||
* `null` when the task is not currently monitorable (e.g. `todo`, `done`,
|
||||
* `archived`, `triage`).
|
||||
*
|
||||
* Precedence (deterministic, so a compound state resolves to a single stable
|
||||
* stage): **workflow-gate > pull-request > merger > reviewer > executor**.
|
||||
* A task paused on a workflow prompt/script gate node is reported as
|
||||
* workflow-gate even if it also carries an open PR or pending review, because
|
||||
* the gate is the current blocking reason. Below that, an active PR takes
|
||||
* precedence over a bare merge/review classification since PR lifecycle is
|
||||
* the more specific state. Below that, an explicit merge-hold/merge-error
|
||||
* marker takes precedence over a generic pending-review read of `in-review`.
|
||||
*
|
||||
* Never throws — missing/partial fields degrade to `null`.
|
||||
*/
|
||||
export function resolveWatchedStage(task: Partial<OverseerTaskRef> | null | undefined): OverseerWatchedStage | null {
|
||||
try {
|
||||
if (!task) return null;
|
||||
|
||||
// workflow-gate: paused awaiting an explicit workflow prompt/script gate
|
||||
// input (cli-approval or ask-input gate), regardless of column.
|
||||
if (task.paused === true && typeof task.pausedReason === "string") {
|
||||
if (task.pausedReason.startsWith("workflow-cli-approval:") || task.pausedReason.startsWith("workflow-input:")) {
|
||||
return "workflow-gate";
|
||||
}
|
||||
}
|
||||
|
||||
const column = task.column;
|
||||
if (column !== "in-progress" && column !== "in-review") {
|
||||
return null;
|
||||
}
|
||||
|
||||
if (column === "in-progress") {
|
||||
return "executor";
|
||||
}
|
||||
|
||||
// column === "in-review" beyond this point.
|
||||
|
||||
// pull-request: an active (non-terminal) PR lifecycle takes precedence
|
||||
// over a plain merge/review read of in-review.
|
||||
const prInfo = task.prInfo;
|
||||
if (prInfo && typeof prInfo === "object" && prInfo.status !== "merged" && prInfo.status !== "closed") {
|
||||
return "pull-request";
|
||||
}
|
||||
|
||||
// merger: an explicit merge-hold notification marker or a recorded merge
|
||||
// error means the task is in the merge/integration phase.
|
||||
const marker = task.workflowTransitionNotification;
|
||||
if (marker && marker.kind === "manual-merge-hold") {
|
||||
return "merger";
|
||||
}
|
||||
if (prInfo && typeof prInfo === "object" && typeof prInfo.lastMergeError === "string" && prInfo.lastMergeError.length > 0) {
|
||||
return "merger";
|
||||
}
|
||||
|
||||
// reviewer: review is in progress / pending items.
|
||||
const reviewState = task.reviewState;
|
||||
if (reviewState && typeof reviewState === "object") {
|
||||
return "reviewer";
|
||||
}
|
||||
|
||||
// Plain in-review with no review state and no merge marker yet — treat as
|
||||
// the merge/integration phase (awaiting auto-merge).
|
||||
return "merger";
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function deriveSignalAndSources(
|
||||
taskId: string,
|
||||
stage: OverseerWatchedStage,
|
||||
task: Partial<OverseerTaskRef>,
|
||||
): { signal: OverseerObservationSignal; reason: string; sources: OverseerSourceLink[] } {
|
||||
switch (stage) {
|
||||
case "executor": {
|
||||
if (task.paused === true) {
|
||||
return {
|
||||
signal: "blocked",
|
||||
reason: task.pausedReason ? `Executor stage paused: ${task.pausedReason}` : "Executor stage paused",
|
||||
sources: [{ kind: "agent-log", ref: taskId }],
|
||||
};
|
||||
}
|
||||
return {
|
||||
signal: "progressing",
|
||||
reason: "Task is actively executing in-progress work",
|
||||
sources: [{ kind: "agent-log", ref: taskId }],
|
||||
};
|
||||
}
|
||||
case "reviewer": {
|
||||
const reviewState = task.reviewState as OverseerTaskReviewState | undefined;
|
||||
const summary = reviewState?.summary;
|
||||
const decision = summary && "reviewDecision" in summary ? summary.reviewDecision : undefined;
|
||||
if (decision === "CHANGES_REQUESTED") {
|
||||
return {
|
||||
signal: "blocked",
|
||||
reason: "Review requested changes",
|
||||
sources: [{ kind: "review-comment", ref: reviewState?.items?.[0]?.id ?? taskId }],
|
||||
};
|
||||
}
|
||||
return {
|
||||
signal: "progressing",
|
||||
reason: "Review in progress",
|
||||
sources: [{ kind: "review-comment", ref: reviewState?.items?.[0]?.id ?? taskId }],
|
||||
};
|
||||
}
|
||||
case "merger": {
|
||||
const prInfo = task.prInfo;
|
||||
if (prInfo?.lastMergeError) {
|
||||
return {
|
||||
signal: "failed",
|
||||
reason: `Merge failed: ${prInfo.lastMergeError}`,
|
||||
sources: [{ kind: "merge-error", ref: prInfo.lastMergeError }],
|
||||
};
|
||||
}
|
||||
if (task.workflowTransitionNotification?.kind === "manual-merge-hold") {
|
||||
return {
|
||||
signal: "awaiting-human",
|
||||
reason: "Held awaiting manual merge decision",
|
||||
sources: [{ kind: "merge-error", ref: task.workflowTransitionNotification.transitionId ?? taskId }],
|
||||
};
|
||||
}
|
||||
return {
|
||||
signal: "progressing",
|
||||
reason: "Task is in the merge/integration phase",
|
||||
sources: [{ kind: "merge-error", ref: taskId }],
|
||||
};
|
||||
}
|
||||
case "pull-request": {
|
||||
const prInfo = task.prInfo as PrInfo | undefined;
|
||||
if (prInfo?.checkRollup === "failure") {
|
||||
return {
|
||||
signal: "failed",
|
||||
reason: "PR checks failing",
|
||||
sources: [{ kind: "failed-check", ref: prInfo.url, url: prInfo.url }],
|
||||
};
|
||||
}
|
||||
return {
|
||||
signal: "progressing",
|
||||
reason: "PR lifecycle in progress",
|
||||
sources: [{ kind: "pr-state", ref: prInfo?.url ?? taskId, url: prInfo?.url }],
|
||||
};
|
||||
}
|
||||
case "workflow-gate": {
|
||||
return {
|
||||
signal: "awaiting-human",
|
||||
reason: task.pausedReason ? `Paused on workflow gate: ${task.pausedReason}` : "Paused on workflow gate",
|
||||
sources: [{ kind: "agent-log", ref: task.pausedReason ?? taskId }],
|
||||
};
|
||||
}
|
||||
default: {
|
||||
return {
|
||||
signal: "progressing",
|
||||
reason: "",
|
||||
sources: [],
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Minimal store seam the monitor records best-effort observations through —
|
||||
* mirrors `fallback-model-observer.ts`'s `FallbackLogStore` seam. */
|
||||
export interface OverseerLogStore {
|
||||
logEntry?(taskId: string, action: string): Promise<unknown>;
|
||||
appendAgentLog?(
|
||||
taskId: string,
|
||||
text: string,
|
||||
type: "text" | "thinking" | "tool" | "tool_result" | "tool_error",
|
||||
detail?: string,
|
||||
agent?: string,
|
||||
): Promise<unknown>;
|
||||
}
|
||||
|
||||
export interface PlannerOverseerMonitorOptions {
|
||||
store?: OverseerLogStore;
|
||||
onObservation?: (observation: OverseerStageObservation) => void | Promise<void>;
|
||||
/** Max observations retained per task in the in-memory ring buffer. Default: 20. */
|
||||
maxObservationsPerTask?: number;
|
||||
}
|
||||
|
||||
const DEFAULT_MAX_OBSERVATIONS_PER_TASK = 20;
|
||||
|
||||
/**
|
||||
* FNXC:PlannerOversight 2026-07-04-00:00:
|
||||
* Records-only monitor: watches a task's current lifecycle stage and, when
|
||||
* the effective oversight level is not `"off"`, records one normalized
|
||||
* `OverseerStageObservation` per call into a bounded per-task ring buffer and
|
||||
* invokes the optional `onObservation` callback best-effort. Never mutates
|
||||
* task lifecycle, never retries/fixes/merges/notifies — steering and
|
||||
* recovery are FN-7512+.
|
||||
*/
|
||||
export class PlannerOverseerMonitor {
|
||||
private readonly store?: OverseerLogStore;
|
||||
private readonly onObservation?: (observation: OverseerStageObservation) => void | Promise<void>;
|
||||
private readonly maxObservationsPerTask: number;
|
||||
private readonly observations = new Map<string, OverseerStageObservation[]>();
|
||||
|
||||
constructor(options: PlannerOverseerMonitorOptions = {}) {
|
||||
this.store = options.store;
|
||||
this.onObservation = options.onObservation;
|
||||
this.maxObservationsPerTask = options.maxObservationsPerTask ?? DEFAULT_MAX_OBSERVATIONS_PER_TASK;
|
||||
}
|
||||
|
||||
/**
|
||||
* Observe a task's current watched stage and record a gated observation.
|
||||
* Returns `null` when the level is `"off"` or when no stage is currently
|
||||
* monitorable. Never throws.
|
||||
*/
|
||||
async observeTask(task: OverseerTaskRef, level: PlannerOversightLevel): Promise<OverseerStageObservation | null> {
|
||||
try {
|
||||
if (level === "off") {
|
||||
return null;
|
||||
}
|
||||
|
||||
const stage = resolveWatchedStage(task);
|
||||
if (!stage) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const { signal, reason, sources } = deriveSignalAndSources(task.id, stage, task);
|
||||
const observation: OverseerStageObservation = {
|
||||
taskId: task.id,
|
||||
stage,
|
||||
signal,
|
||||
oversightLevel: level,
|
||||
observedAt: Date.now(),
|
||||
reason,
|
||||
sources,
|
||||
};
|
||||
|
||||
this.record(observation);
|
||||
|
||||
if (this.onObservation) {
|
||||
try {
|
||||
await this.onObservation(observation);
|
||||
} catch {
|
||||
// Best-effort — never let a consumer callback fail the monitor.
|
||||
}
|
||||
}
|
||||
|
||||
if (this.store?.logEntry) {
|
||||
await this.store
|
||||
.logEntry(task.id, `[planner-overseer] stage=${stage} signal=${signal}: ${reason}`)
|
||||
.catch(() => undefined);
|
||||
}
|
||||
|
||||
return observation;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private record(observation: OverseerStageObservation): void {
|
||||
const existing = this.observations.get(observation.taskId) ?? [];
|
||||
existing.push(observation);
|
||||
if (existing.length > this.maxObservationsPerTask) {
|
||||
existing.splice(0, existing.length - this.maxObservationsPerTask);
|
||||
}
|
||||
this.observations.set(observation.taskId, existing);
|
||||
}
|
||||
|
||||
/** Return the recorded observations for a task, oldest first. */
|
||||
getObservations(taskId: string): OverseerStageObservation[] {
|
||||
return [...(this.observations.get(taskId) ?? [])];
|
||||
}
|
||||
|
||||
/** Clear recorded observations for a task (e.g. on task completion). */
|
||||
clear(taskId: string): void {
|
||||
this.observations.delete(taskId);
|
||||
}
|
||||
|
||||
/** Task IDs that currently retain at least one recorded observation. Used
|
||||
* by the engine poll to release ring buffers for tasks that have left the
|
||||
* in-flight set. */
|
||||
getObservedTaskIds(): string[] {
|
||||
return [...this.observations.keys()];
|
||||
}
|
||||
}
|
||||
@@ -13,13 +13,14 @@ import type {
|
||||
ResearchSynthesisRequest,
|
||||
ResearchSynthesisResult,
|
||||
} from "@fusion/core";
|
||||
import { allowsAutoMergeProcessing, compareTasksByPriorityThenAgeAndId, getTaskHardMergeBlocker, isSharedBranchGroupMemberIntegration, isWorkspaceTask, normalizeMergerMode, resolveMaxAutoMergeRetries, sortTasksByPriorityThenAgeAndId } from "@fusion/core";
|
||||
import { allowsAutoMergeProcessing, compareTasksByPriorityThenAgeAndId, getTaskHardMergeBlocker, isSharedBranchGroupMemberIntegration, isWorkspaceTask, normalizeMergerMode, resolveEffectivePlannerOversightLevel, resolveEffectiveSettings, resolveMaxAutoMergeRetries, sortTasksByPriorityThenAgeAndId } from "@fusion/core";
|
||||
import { execFile } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import { InProcessRuntime } from "./runtimes/in-process-runtime.js";
|
||||
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 type { PrNodeGithubOps } from "./pr-nodes.js";
|
||||
import { PrReconciler, type PrReconcileGithubOps } from "./pr-reconcile.js";
|
||||
import { PrCommentHandler } from "./pr-comment-handler.js";
|
||||
@@ -322,6 +323,18 @@ export class ProjectEngine {
|
||||
private runtime: InProcessRuntime;
|
||||
private started = false;
|
||||
private prMonitor?: PrMonitor;
|
||||
/**
|
||||
* FNXC:PlannerOversight 2026-07-04-00:00:
|
||||
* FN-7511 records-only planner-overseer monitor. Watches in-flight tasks
|
||||
* (in-progress/in-review) across the executor/reviewer/merger/pull-request/
|
||||
* workflow-gate stages, gated by the task's effective planner oversight
|
||||
* level (`resolveEffectivePlannerOversightLevel`). No lifecycle mutation —
|
||||
* steering/recovery land in FN-7512+.
|
||||
*/
|
||||
private plannerOverseer?: PlannerOverseerMonitor;
|
||||
private plannerOverseerPollTimer: ReturnType<typeof setInterval> | null = null;
|
||||
/** Conservative poll cadence for the records-only planner-overseer monitor (45s). */
|
||||
private static readonly PLANNER_OVERSEER_POLL_INTERVAL_MS = 45 * 1000;
|
||||
private prReconciler?: PrReconciler;
|
||||
private prCommentHandler?: PrCommentHandler;
|
||||
private notifier?: NtfyNotifier;
|
||||
@@ -577,6 +590,11 @@ export class ProjectEngine {
|
||||
runtimeLog.warn(`Remote tunnel restore evaluation failed (continuing startup): ${message}`);
|
||||
}
|
||||
|
||||
// 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.startPlannerOverseerPoll(store);
|
||||
|
||||
// 2. Initialize PrMonitor + PrCommentHandler
|
||||
this.prMonitor = new PrMonitor();
|
||||
this.prCommentHandler = new PrCommentHandler(store);
|
||||
@@ -809,6 +827,7 @@ export class ProjectEngine {
|
||||
clearInterval(this.mergeActiveReconcileTimer);
|
||||
this.mergeActiveReconcileTimer = null;
|
||||
}
|
||||
this.stopPlannerOverseerPoll();
|
||||
|
||||
// Abort active/pending merge work before tearing down sessions.
|
||||
this.mergeAbortController?.abort();
|
||||
@@ -976,6 +995,11 @@ export class ProjectEngine {
|
||||
return this.prMonitor;
|
||||
}
|
||||
|
||||
/** Get the records-only PlannerOverseerMonitor (if initialized). See FN-7511. */
|
||||
getPlannerOverseer(): PlannerOverseerMonitor | undefined {
|
||||
return this.plannerOverseer;
|
||||
}
|
||||
|
||||
/** Get the CronRunner (if initialized). */
|
||||
getCronRunner(): CronRunner | undefined {
|
||||
return this.cronRunner;
|
||||
@@ -1780,6 +1804,73 @@ export class ProjectEngine {
|
||||
return cleared;
|
||||
}
|
||||
|
||||
/**
|
||||
* FNXC:PlannerOversight 2026-07-04-00:00:
|
||||
* Bounded, gated poll over in-flight tasks (in-progress/in-review). For
|
||||
* each task, resolves the effective planner oversight level and, unless it
|
||||
* is "off" or the task resolves to no watched stage, records one
|
||||
* `OverseerStageObservation` via `PlannerOverseerMonitor#observeTask`.
|
||||
* Records-only: no lifecycle mutation, retry, or notification here.
|
||||
* Cleared on `stop()`. Never an unbounded loop — a single bounded
|
||||
* `setInterval` at a conservative cadence.
|
||||
*/
|
||||
private startPlannerOverseerPoll(store: TaskStore): void {
|
||||
if (this.plannerOverseerPollTimer) {
|
||||
return;
|
||||
}
|
||||
this.plannerOverseerPollTimer = setInterval(() => {
|
||||
void this.pollPlannerOverseer(store);
|
||||
}, ProjectEngine.PLANNER_OVERSEER_POLL_INTERVAL_MS);
|
||||
}
|
||||
|
||||
private async pollPlannerOverseer(store: TaskStore): Promise<void> {
|
||||
if (!this.plannerOverseer || this.shuttingDown) {
|
||||
return;
|
||||
}
|
||||
const overseer = this.plannerOverseer;
|
||||
try {
|
||||
const [inProgress, inReview] = await Promise.all([
|
||||
store.listTasks({ column: "in-progress" }).catch(() => [] as Task[]),
|
||||
store.listTasks({ column: "in-review" }).catch(() => [] as Task[]),
|
||||
]);
|
||||
const inFlight = [...inProgress, ...inReview];
|
||||
const inFlightIds = new Set(inFlight.map((t) => t.id));
|
||||
|
||||
for (const task of inFlight) {
|
||||
try {
|
||||
const workflowEffective = await resolveEffectiveSettings(store, { id: task.id }).catch(() => ({}) as Record<string, unknown>);
|
||||
const level = resolveEffectivePlannerOversightLevel(
|
||||
task.plannerOversightLevel,
|
||||
workflowEffective.plannerOversightLevel as string | undefined,
|
||||
);
|
||||
if (level === "off") {
|
||||
continue;
|
||||
}
|
||||
await overseer.observeTask(task, level);
|
||||
} catch {
|
||||
// Best-effort per-task — never let one task's failure block the poll.
|
||||
}
|
||||
}
|
||||
|
||||
// Drop retained observations for tasks that have left the in-flight set
|
||||
// (moved to done/archived/failed/etc.) so the ring buffers don't leak.
|
||||
for (const taskId of overseer.getObservedTaskIds()) {
|
||||
if (!inFlightIds.has(taskId)) {
|
||||
overseer.clear(taskId);
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// Best-effort poll — degrade silently, never throw out of the interval.
|
||||
}
|
||||
}
|
||||
|
||||
private stopPlannerOverseerPoll(): void {
|
||||
if (this.plannerOverseerPollTimer) {
|
||||
clearInterval(this.plannerOverseerPollTimer);
|
||||
this.plannerOverseerPollTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
private scheduleMergeActiveReconciliation(intervalMs: number): void {
|
||||
if (!Number.isFinite(intervalMs) || intervalMs <= 0) {
|
||||
return;
|
||||
|
||||
Reference in New Issue
Block a user