root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

FN-8768: recover planning handoffs after dependency reseeds

Browse Source 
Prevent dependency reseeds from leaving completed planning work without a dispatchable continuation.

- Serialize dependency invalidation with workflow claims and retire only pending continuations.
- Persist dispatch-deduplication state and recover legacy reseeded planning handoffs safely.
- Add PostgreSQL migration, integration coverage, architecture guidance, and a patch changeset.

Files changed:
 .changeset/fn-8768-planning-reseed.md              |   7 ++
 docs/architecture.md                               |  10 +-
 .../core/src/__test-utils__/pg-test-harness.ts     |   8 ++
 .../__tests__/postgres/backend-resolver.test.ts    |   5 +
 .../src/__tests__/postgres/schema-applier.test.ts  |  17 ++-
 .../postgres/task-dependency-mutation.pg.test.ts   |  42 ++++++-
 .../__tests__/task-update-lanes-resolved.test.ts   |  20 +++-
 packages/core/src/postgres/advisory-locks.ts       | 122 +++++++++++++++++++++
 packages/core/src/postgres/backend-resolver.ts     |  13 ++-
 packages/core/src/postgres/data-layer.ts           |   4 +
 packages/core/src/postgres/embedded-lifecycle.ts   |   6 +
 .../migrations/0043_fn8768_dispatch_dedupe.sql     |  17 +++
 packages/core/src/postgres/schema-applier.ts       |  13 ++-
 packages/core/src/postgres/schema/project.ts       |  15 +++
 packages/core/src/store.ts                         |  56 +++++++++-
 packages/core/src/task-store/audit-ops.ts          |  49 +++++++++
 .../core/src/task-store/branch-and-pr-entities.ts  |  30 +++++
 packages/core/src/task-store/project-store-ops.ts  |  55 ++++++++--
 packages/core/src/task-store/task-update.ts        |  30 ++++-
 packages/core/src/task-store/update-task-deps.ts   |  30 ++++-
 packages/engine/src/__tests__/triage.test.ts       |  66 ++++++++++-
 packages/engine/src/execution/hold-release.ts      |  45 ++++++++
 packages/engine/src/scheduler.ts                   |   3 +-
 packages/engine/src/triage.ts                      |  87 ++++++++++++++-
 24 files changed, 713 insertions(+), 37 deletions(-)

Fusion-Task-Id: FN-8768

Fusion-Task-Lineage: 539ef649-5a13-4eaa-a695-bc68370fed22

Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
2026-08-03 19:31:03 -07:00
parent 9dc7b94c17
commit bb17baaacf
24 changed files with 713 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
packages/engine/src/execution/hold-release.ts
View File

@@ -59,6 +59,7 @@ import {
type WorkflowIrV2,
type WorkflowIrColumn,
} from "@fusion/core";
import { createHash } from "node:crypto";
import { readFile } from "node:fs/promises";
import { schedulerLog } from "../logger.js";
import { getPromptPath } from "./spec-staleness.js";
@@ -174,6 +175,48 @@ export function resolvePreReleasePlanReviewNode(ir: WorkflowIr): WorkflowIrNode
return planReviewNode;
}
/*
FNXC:PlanningDependencyReseed 2026-08-04-02:14:
A release refusal is otherwise invisible after scheduler dispatch returns early.
Hash only durable state that changes the planning/Plan-Review episode; the core
store atomically claims this project/task episode and appends one task-log entry.
*/
export async function checkAndRecordUnplannedExecutionBlock(
store: TaskStore,
task: Task,
ir: WorkflowIr,
): Promise<void> {
const recorder = (store as Partial<Pick<TaskStore, "checkAndRecordUnplannedExecutionBlock">>).checkAndRecordUnplannedExecutionBlock;
if (!recorder) return;
const planReviewNode = resolvePreReleasePlanReviewNode(ir)?.id ?? "none";
let promptContent = typeof task.prompt === "string" ? task.prompt : "";
const tasksDir = typeof store.getTasksDir === "function" ? store.getTasksDir() : undefined;
if (tasksDir) {
try {
promptContent = await readFile(getPromptPath(tasksDir, task.id), "utf8");
} catch {
promptContent = "";
}
}
const promptMarker = promptContent.length > 0
? createHash("sha256").update(promptContent).digest("hex")
: "missing";
const dependencies = [...(task.dependencies ?? [])].sort();
const episode = createHash("sha256").update(JSON.stringify({
planReviewNode,
promptMarker,
dependencies,
status: task.status ?? null,
handoffFingerprint: task.approvedPlanFingerprint ?? null,
})).digest("hex");
try {
await recorder.call(store, task.id, episode);
} catch (error) {
// The gate is safety-critical; its diagnostic must not turn an otherwise-safe refusal into a dispatch failure.
schedulerLog.warn(`Could not persist unplanned dispatch refusal for ${task.id}: ${error instanceof Error ? error.message : String(error)}`);
}
}
export async function isUnplannedForExecution(store: TaskStore, task: Task, ir: WorkflowIr): Promise<boolean> {
/*
FNXC:PlanReview 2026-07-19-00:40 (U3):
@@ -722,6 +765,7 @@ async function issueRelease(
}
if (targetIsProcessing && !options.allowUnplanned && (await isUnplannedForExecution(store, task, ir))) {
await checkAndRecordUnplannedExecutionBlock(store, task, ir);
/*
FNXC:StrandedHoldContinuation 2026-07-26-14:15:
Before FN-8592 this was an undeduplicated `schedulerLog.log`, not debug.
@@ -883,6 +927,7 @@ export async function promoteHeldTask(
: false;
const unplanned = targetIsProcessing && (await isUnplannedForExecution(store, task, ir));
if (unplanned && options.force !== true) {
await checkAndRecordUnplannedExecutionBlock(store, task, ir);
return { released: false, rejection: "unplanned-for-execution", toColumn: target };
}