diff --git a/.changeset/fn-7520-planner-overseer-events.md b/.changeset/fn-7520-planner-overseer-events.md new file mode 100644 index 0000000000..b7850edd77 --- /dev/null +++ b/.changeset/fn-7520-planner-overseer-events.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": minor +--- + +summary: Emit planner-overseer run-audit events for observations, steering, retries, recovery, confirmations, and escalations. +category: feature +dev: New core emitters (emitOverseerObservation/Steering/RecoveryAttempt/Retry/Confirmation/Escalation) in planner-overseer-events.ts, each mapping its decision-point to the correct intervention action/outcome and delegating to FN-7519's recordPlannerIntervention under the overseer:intervention mutation. Producer call-sites land in FN-7511/FN-7512/FN-7513. diff --git a/docs/architecture.md b/docs/architecture.md index 208fdfd0df..f3581af30e 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) — 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`). This task ships the shape and record/read helpers only — FN-7520 wires the actual emission call-sites at overseer decision points. +- **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. - **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/core/src/__tests__/planner-overseer-events.test.ts b/packages/core/src/__tests__/planner-overseer-events.test.ts new file mode 100644 index 0000000000..5aad7646d6 --- /dev/null +++ b/packages/core/src/__tests__/planner-overseer-events.test.ts @@ -0,0 +1,236 @@ +import { describe, it, expect } from "vitest"; +import { + emitOverseerObservation, + emitOverseerSteering, + emitOverseerRecoveryAttempt, + emitOverseerRetry, + emitOverseerConfirmation, + emitOverseerEscalation, +} from "../planner-overseer-events.js"; +import { getPlannerInterventionTimeline, type PlannerInterventionStore } from "../planner-intervention.js"; +import { OVERSEER_INTERVENTION_MUTATION } from "../types.js"; +import type { RunAuditEvent, RunAuditEventFilter, RunAuditEventInput } from "../types.js"; + +/* +FNXC:PlannerOversight 2026-07-04-19:30: +FN-7520 unit tests for the canonical `emitOverseer*` emission façade. Uses the +same narrow in-memory fake store pattern as FN-7519's +`planner-intervention.test.ts` (only depends on the +`recordRunAuditEvent`/`getRunAuditEvents` seam) to keep this suite fast per +the project's "Do Not Add Slow Tests" standing rule. +*/ +class FakeRunAuditStore implements PlannerInterventionStore { + events: RunAuditEvent[] = []; + private counter = 0; + + recordRunAuditEvent(input: RunAuditEventInput): RunAuditEvent { + const event: RunAuditEvent = { + id: `evt-${++this.counter}`, + timestamp: input.timestamp ?? new Date(Date.now() + this.counter).toISOString(), + taskId: input.taskId, + agentId: input.agentId, + runId: input.runId, + domain: input.domain, + mutationType: input.mutationType, + target: input.target, + metadata: input.metadata, + }; + this.events.push(event); + return event; + } + + getRunAuditEvents(options: RunAuditEventFilter = {}): RunAuditEvent[] { + return this.events + .filter((event) => (options.taskId ? event.taskId === options.taskId : true)) + .filter((event) => (options.mutationType ? event.mutationType === options.mutationType : true)) + .slice() + .sort((a, b) => (a.timestamp < b.timestamp ? 1 : a.timestamp > b.timestamp ? -1 : 0)) + .slice(0, options.limit ?? undefined); + } +} + +describe("emitOverseer* façade", () => { + it("emitOverseerObservation records action=observe, default outcome=succeeded, no attempt fields", () => { + const store = new FakeRunAuditStore(); + emitOverseerObservation({ + store, + taskId: "FN-1", + runId: "run-1", + stage: "executor", + reason: "Executor is progressing normally", + }); + + const timeline = getPlannerInterventionTimeline(store, "FN-1"); + expect(timeline).toHaveLength(1); + expect(timeline[0].action).toBe("observe"); + expect(timeline[0].outcome).toBe("succeeded"); + expect(timeline[0].stage).toBe("executor"); + expect(timeline[0].reason).toBe("Executor is progressing normally"); + expect(timeline[0].attemptCount).toBeUndefined(); + expect(timeline[0].attemptLimit).toBeUndefined(); + }); + + it("emitOverseerSteering records action=inject-guidance, default outcome=pending", () => { + const store = new FakeRunAuditStore(); + emitOverseerSteering({ + store, + taskId: "FN-2", + stage: "executor", + reason: "Injecting guidance to unblock stalled step", + }); + + const timeline = getPlannerInterventionTimeline(store, "FN-2"); + expect(timeline[0].action).toBe("inject-guidance"); + expect(timeline[0].outcome).toBe("pending"); + }); + + it("emitOverseerRecoveryAttempt records action=request-fix, default outcome=pending, persists attempt fields", () => { + const store = new FakeRunAuditStore(); + emitOverseerRecoveryAttempt({ + store, + taskId: "FN-3", + stage: "reviewer", + reason: "Requesting fix for failing review checks", + attemptCount: 1, + attemptLimit: 3, + }); + + const timeline = getPlannerInterventionTimeline(store, "FN-3"); + expect(timeline[0].action).toBe("request-fix"); + expect(timeline[0].outcome).toBe("pending"); + expect(timeline[0].attemptCount).toBe(1); + expect(timeline[0].attemptLimit).toBe(3); + }); + + it("emitOverseerRetry records action=retry, default outcome=pending, persists attempt fields", () => { + const store = new FakeRunAuditStore(); + emitOverseerRetry({ + store, + taskId: "FN-4", + stage: "merger", + reason: "Retrying stuck merge step", + attemptCount: 2, + attemptLimit: 4, + }); + + const timeline = getPlannerInterventionTimeline(store, "FN-4"); + expect(timeline[0].action).toBe("retry"); + expect(timeline[0].outcome).toBe("pending"); + expect(timeline[0].attemptCount).toBe(2); + expect(timeline[0].attemptLimit).toBe(4); + }); + + it("emitOverseerConfirmation records action=request-confirmation, default outcome=awaiting-confirmation", () => { + const store = new FakeRunAuditStore(); + emitOverseerConfirmation({ + store, + taskId: "FN-5", + stage: "pull-request", + reason: "Requesting human confirmation before merge", + }); + + const timeline = getPlannerInterventionTimeline(store, "FN-5"); + expect(timeline[0].action).toBe("request-confirmation"); + expect(timeline[0].outcome).toBe("awaiting-confirmation"); + }); + + it("emitOverseerEscalation records action=escalate, default outcome=failed", () => { + const store = new FakeRunAuditStore(); + emitOverseerEscalation({ + store, + taskId: "FN-6", + stage: "workflow-gate", + reason: "Bounded recovery exhausted; escalating to human", + }); + + const timeline = getPlannerInterventionTimeline(store, "FN-6"); + expect(timeline[0].action).toBe("escalate"); + expect(timeline[0].outcome).toBe("failed"); + }); + + it("an explicit outcome overrides each emitter's default", () => { + const store = new FakeRunAuditStore(); + emitOverseerObservation({ + store, + taskId: "FN-7", + stage: "executor", + reason: "Observation with overridden outcome", + outcome: "failed", + }); + emitOverseerEscalation({ + store, + taskId: "FN-7", + stage: "workflow-gate", + reason: "Escalation bypassed by human-control guard", + outcome: "skipped", + }); + + const timeline = getPlannerInterventionTimeline(store, "FN-7"); + // Newest-first ordering. + expect(timeline[0].action).toBe("escalate"); + expect(timeline[0].outcome).toBe("skipped"); + expect(timeline[1].action).toBe("observe"); + expect(timeline[1].outcome).toBe("failed"); + }); + + it("sourceLinks round-trip through metadata for every kind used by producers", () => { + const store = new FakeRunAuditStore(); + emitOverseerRecoveryAttempt({ + store, + taskId: "FN-8", + stage: "reviewer", + reason: "Requesting fix with linked evidence", + attemptCount: 1, + attemptLimit: 2, + sourceLinks: [ + { kind: "agent-log", label: "Agent log excerpt" }, + { kind: "failed-check", label: "Failing lint check" }, + { kind: "merge-error", label: "Merge conflict detail" }, + { kind: "pr-state", label: "PR review state" }, + ], + }); + + const timeline = getPlannerInterventionTimeline(store, "FN-8"); + expect(timeline[0].sourceLinks).toEqual([ + { kind: "agent-log", label: "Agent log excerpt", target: undefined, url: undefined }, + { kind: "failed-check", label: "Failing lint check", target: undefined, url: undefined }, + { kind: "merge-error", label: "Merge conflict detail", target: undefined, url: undefined }, + { kind: "pr-state", label: "PR review state", target: undefined, url: undefined }, + ]); + }); + + it("is non-throwing when only the minimal required fields are supplied", () => { + const store = new FakeRunAuditStore(); + expect(() => + emitOverseerObservation({ + store, + taskId: "FN-9", + runId: "run-9", + stage: "executor", + reason: "Minimal observation", + }), + ).not.toThrow(); + + const timeline = getPlannerInterventionTimeline(store, "FN-9"); + expect(timeline).toHaveLength(1); + }); + + it("enforces the single-writer contract: every emitter produces only overseer:intervention events via recordPlannerIntervention", () => { + const store = new FakeRunAuditStore(); + emitOverseerObservation({ store, taskId: "FN-10", stage: "executor", reason: "r1" }); + emitOverseerSteering({ store, taskId: "FN-10", stage: "executor", reason: "r2" }); + emitOverseerRecoveryAttempt({ store, taskId: "FN-10", stage: "reviewer", reason: "r3", attemptCount: 1, attemptLimit: 2 }); + emitOverseerRetry({ store, taskId: "FN-10", stage: "merger", reason: "r4", attemptCount: 1, attemptLimit: 2 }); + emitOverseerConfirmation({ store, taskId: "FN-10", stage: "pull-request", reason: "r5" }); + emitOverseerEscalation({ store, taskId: "FN-10", stage: "workflow-gate", reason: "r6" }); + + expect(store.events).toHaveLength(6); + for (const event of store.events) { + expect(event.mutationType).toBe(OVERSEER_INTERVENTION_MUTATION); + expect(event.mutationType).toBe("overseer:intervention"); + } + // No other overseer:* mutation types are introduced by this façade. + const mutationTypes = new Set(store.events.map((e) => e.mutationType)); + expect(mutationTypes.size).toBe(1); + }); +}); diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index aa3e6d7656..ea537ef253 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -32,6 +32,15 @@ export { isActiveNearDuplicateColumn, isNearDuplicateCanonicalInactive } from ". export type { NearDuplicateCanonicalState } from "./near-duplicate-canonical.js"; export { formatGitLabTrackedItemRef, isGitLabTrackingStale } from "./gitlab-tracking.js"; export * from "./planner-intervention.js"; +export { + emitOverseerObservation, + emitOverseerSteering, + emitOverseerRecoveryAttempt, + emitOverseerRetry, + emitOverseerConfirmation, + emitOverseerEscalation, +} from "./planner-overseer-events.js"; +export type { OverseerEventInput } from "./planner-overseer-events.js"; export * from "./frontend-ux-policy.js"; export * from "./file-scope-classification.js"; export { MAX_TASK_LIST_TEXT_CHARS, clampTaskListText, formatTaskListText } from "./task-list-format.js"; diff --git a/packages/core/src/planner-overseer-events.ts b/packages/core/src/planner-overseer-events.ts new file mode 100644 index 0000000000..d868432058 --- /dev/null +++ b/packages/core/src/planner-overseer-events.ts @@ -0,0 +1,128 @@ +import type { + PlannerInterventionAction, + PlannerInterventionOutcome, + PlannerInterventionSourceLink, + PlannerOversightStage, + RunAuditEvent, +} from "./types.js"; +import { type PlannerInterventionStore, recordPlannerIntervention } from "./planner-intervention.js"; + +/** + * FNXC:PlannerOversight 2026-07-04-19:30: + * FN-7520 canonical emission façade for planner-overseer decision points. + * Requirement: every overseer decision point — a passive observation, injected + * steering guidance, a bounded recovery attempt, a retry of a stuck/failed + * step, a merge/PR confirmation request, or an escalation to a human — is + * recorded as a run-audit/activity event through exactly ONE façade. Each + * exported emitter below fixes the `action` (and a sensible default + * `outcome`, overridable via input) for its category and delegates to + * FN-7519's `recordPlannerIntervention(...)`, which is the single canonical + * writer for the `overseer:intervention` run-audit mutation type. + * + * Single-writer contract: do NOT call `recordRunAuditEvent` directly from + * here and do NOT introduce a second `overseer:*` mutation type. FN-7511 / + * FN-7512 / FN-7513 (the monitoring loop, bounded-recovery engine, and + * confirmation/escalation producers) are expected to import and call these + * emitters rather than emit run-audit events inline. + */ + +/** Shared input for every `emitOverseer*` façade function. */ +export interface OverseerEventInput { + /** Store implementing the minimal `recordRunAuditEvent`/`getRunAuditEvents` seam (satisfied by `TaskStore`). */ + store: PlannerInterventionStore; + taskId: string; + /** Heartbeat run ID that produced this decision point. Defaults to a synthetic per-call ID when omitted (see `recordPlannerIntervention`). */ + runId?: string; + /** Agent ID that produced this decision point. Defaults to `"overseer"` when omitted. */ + agentId?: string; + stage: PlannerOversightStage; + reason: string; + /** Overrides the emitter's default outcome for this category. */ + outcome?: PlannerInterventionOutcome; + attemptCount?: number; + attemptLimit?: number; + sourceLinks?: PlannerInterventionSourceLink[]; + /** ISO-8601 timestamp override. Defaults to now. */ + timestamp?: string; +} + +/** + * Normalizes an `OverseerEventInput` into FN-7519's `RecordPlannerInterventionInput` + * shape for the given fixed `action`, applying `defaultOutcome` when the caller + * did not supply an explicit `outcome`, and delegates to `recordPlannerIntervention`. + * Non-throwing on optional-field absence — all optional fields are passed through + * as-is (`undefined` when absent) so FN-7519's own tolerant handling applies. + */ +function normalizeAndRecord( + input: OverseerEventInput, + action: PlannerInterventionAction, + defaultOutcome: PlannerInterventionOutcome, +): RunAuditEvent { + return recordPlannerIntervention(input.store, { + taskId: input.taskId, + runId: input.runId, + agentId: input.agentId, + stage: input.stage, + reason: input.reason, + action, + outcome: input.outcome ?? defaultOutcome, + attemptCount: input.attemptCount, + attemptLimit: input.attemptLimit, + sourceLinks: input.sourceLinks, + timestamp: input.timestamp, + }); +} + +/** + * Records a passive overseer observation (a watch signal with no corrective + * action taken). Default outcome: `"succeeded"` (the observation itself always + * "succeeds"; attempt fields are typically omitted for this category). + */ +export function emitOverseerObservation(input: OverseerEventInput): RunAuditEvent { + return normalizeAndRecord(input, "observe", "succeeded"); +} + +/** + * Records the overseer injecting steering guidance into a running task. + * Default outcome: `"pending"` (guidance has been injected; whether it lands + * successfully is determined by a later observation/retry). + */ +export function emitOverseerSteering(input: OverseerEventInput): RunAuditEvent { + return normalizeAndRecord(input, "inject-guidance", "pending"); +} + +/** + * Records a bounded recovery attempt (an overseer-issued fix request). + * Default outcome: `"pending"`. Callers should supply `attemptCount`/`attemptLimit` + * so the timeline can render bounded-recovery progress. + */ +export function emitOverseerRecoveryAttempt(input: OverseerEventInput): RunAuditEvent { + return normalizeAndRecord(input, "request-fix", "pending"); +} + +/** + * Records a retry of a stuck/failed step. Default outcome: `"pending"`. + * Callers should supply `attemptCount`/`attemptLimit` so the timeline can + * render bounded-retry progress. + */ +export function emitOverseerRetry(input: OverseerEventInput): RunAuditEvent { + return normalizeAndRecord(input, "retry", "pending"); +} + +/** + * Records a merge/PR confirmation request raised to a human. Default outcome: + * `"awaiting-confirmation"`. + */ +export function emitOverseerConfirmation(input: OverseerEventInput): RunAuditEvent { + return normalizeAndRecord(input, "request-confirmation", "awaiting-confirmation"); +} + +/** + * Records an escalation to a human (bounded recovery exhausted, or an + * unrecoverable condition). Default outcome: `"failed"`, overridable — for + * example a caller may escalate with outcome `"skipped"` when escalation is + * itself bypassed by a human-control guard. + */ +export function emitOverseerEscalation(input: OverseerEventInput): RunAuditEvent { + return normalizeAndRecord(input, "escalate", "failed"); +}