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

fix(review): apply autofix feedback

Browse Source
This commit is contained in:
gsxdsm
2026-06-04 02:50:54 -07:00
parent eee715218e
commit 30e2fd7b09
18 changed files with 248 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
packages/engine/src/hold-release.ts
View File

@@ -41,6 +41,7 @@ import {
resolveColumnCapacity,
resolveColumnFlags,
resolveColumnAdjacency,
DEFAULT_WORKFLOW_POOL_ID,
TransitionRejectionError,
BUILTIN_CODING_WORKFLOW_IR,
getBuiltinWorkflow,
@@ -54,8 +55,6 @@ import {
} from "@fusion/core";
import { schedulerLog } from "./logger.js";
const DEFAULT_WORKFLOW_POOL_ID = "__default-workflow__";
/** A reservation handle returned by {@link HoldReleaseDeps.reserveSlot}. The
* sweep calls `release()` if the subsequent move rejects on capacity. */
export interface SlotReservation {
@@ -95,7 +94,13 @@ export interface HoldReleaseResult {
// ── Workflow IR resolution (read-only, mirrors store + merge-trait) ───────────
async function resolveTaskWorkflowIr(store: TaskStore, taskId: string): Promise<WorkflowIr> {
async function resolveTaskWorkflowIr(
store: TaskStore,
taskId: string,
// Optional per-sweep cache keyed by workflowId so each distinct workflow's IR
// is resolved (and its definition fetched) at most once per sweep.
irCache?: Map<string, WorkflowIr>,
): Promise<WorkflowIr> {
let workflowId: string | undefined;
try {
workflowId = store.getTaskWorkflowSelection(taskId)?.workflowId;
@@ -103,14 +108,20 @@ async function resolveTaskWorkflowIr(store: TaskStore, taskId: string): Promise<
workflowId = undefined;
}
if (!workflowId) return BUILTIN_CODING_WORKFLOW_IR;
const cached = irCache?.get(workflowId);
if (cached) return cached;
if (isBuiltinWorkflowId(workflowId)) {
const builtin = getBuiltinWorkflow(workflowId);
return builtin?.ir ?? BUILTIN_CODING_WORKFLOW_IR;
const ir = builtin?.ir ?? BUILTIN_CODING_WORKFLOW_IR;
irCache?.set(workflowId, ir);
return ir;
}
try {
const def = await store.getWorkflowDefinition(workflowId);
if (!def) return BUILTIN_CODING_WORKFLOW_IR;
return typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir;
const ir = typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir;
irCache?.set(workflowId, ir);
return ir;
} catch {
return BUILTIN_CODING_WORKFLOW_IR;
}
@@ -287,15 +298,17 @@ function resolveTimerDeadline(holdConfig: Record<string, unknown>, task: Task):
* arbitration is still the in-txn check, which rejects a losing racer.
*/
function countCapacitySlot(
store: TaskStore,
allTasks: Task[],
// Pre-built taskId → effective workflowId map (one pass per sweep) so this
// counting loop avoids a per-task `effectiveWorkflowId` DB call.
effectiveWorkflowIdByTask: Map<string, string>,
targetColumn: string,
workflowId: string,
countPending: boolean,
): number {
let count = 0;
for (const t of allTasks) {
if (effectiveWorkflowId(store, t.id) !== workflowId) continue;
if ((effectiveWorkflowIdByTask.get(t.id) ?? DEFAULT_WORKFLOW_POOL_ID) !== workflowId) continue;
if (t.column === targetColumn) {
count += 1;
continue;
@@ -325,6 +338,17 @@ export async function runHoldReleaseSweep(
const allTasks = await store.listTasks({ includeArchived: false });
// Per-sweep caches. `allTasks` is a snapshot-stable read within a sweep, so we
// resolve each workflow's IR at most once (irCache) and pre-build the
// taskId → effective-workflowId map a single time rather than per-task DB
// calls inside the capacity counting loop. The authoritative in-txn capacity
// check is unaffected — this only trims the sweep pre-check cost.
const irCache = new Map<string, WorkflowIr>();
const effectiveWorkflowIdByTask = new Map<string, string>();
for (const t of allTasks) {
effectiveWorkflowIdByTask.set(t.id, effectiveWorkflowId(store, t.id));
}
for (const task of allTasks) {
// Skip paused / recovery-backoff tasks exactly as the legacy scheduler does.
if (task.paused || task.userPaused) {
@@ -334,7 +358,7 @@ export async function runHoldReleaseSweep(
continue;
}
const ir = await resolveTaskWorkflowIr(store, task.id);
const ir = await resolveTaskWorkflowIr(store, task.id, irCache);
if (!isHeldTask(ir, task)) continue;
const column = findColumn(ir, task.column);
@@ -372,8 +396,8 @@ export async function runHoldReleaseSweep(
}
const capacity = resolveColumnCapacity(ir, target, settings);
if (capacity.hasCapacity && Number.isFinite(capacity.limit)) {
const workflowId = effectiveWorkflowId(store, task.id);
const occupants = countCapacitySlot(store, allTasks, target, workflowId, capacity.countPending);
const workflowId = effectiveWorkflowIdByTask.get(task.id) ?? DEFAULT_WORKFLOW_POOL_ID;
const occupants = countCapacitySlot(allTasks, effectiveWorkflowIdByTask, target, workflowId, capacity.countPending);
if (occupants >= capacity.limit) {
result.held.push({ taskId: task.id, reason: "downstream-full" });
continue;

3
packages/engine/src/merge-trait.ts
View File

@@ -51,6 +51,7 @@ import {
type WorkflowIr,
type WorkflowIrColumn,
} from "@fusion/core";
import { mergerLog } from "./logger.js";
// ── Resolved merge policy ────────────────────────────────────────────────────
@@ -242,7 +243,7 @@ async function mergeOnEnter(store: TaskStore, task: Pick<Task, "id" | "priority"
// the card is never stranded and the queue is never corrupted. The store
// already audits the rejection.
const message = err instanceof Error ? err.message : String(err);
void message;
mergerLog.warn(`merge enqueue skipped for task ${task.id}: ${message}`);
}
}

4
packages/engine/src/scheduler.ts
View File

@@ -32,7 +32,7 @@ import type { AutoClaimSnapshotManager } from "./auto-claim-snapshot.js";
import { StaleTaskReporter } from "./stale-task-reporter.js";
import { BacklogPressureReporter } from "./backlog-pressure-reporter.js";
import { createRunAuditor, generateSyntheticRunId } from "./run-audit.js";
import { isWorkflowColumnsEnabled } from "@fusion/core";
import { isWorkflowColumnsEnabled, DEFAULT_WORKFLOW_POOL_ID } from "@fusion/core";
import { runHoldReleaseSweep, type SlotReservation } from "./hold-release.js";
/**
@@ -1251,7 +1251,7 @@ export class Scheduler {
// Additive: omitted flag-OFF so the three-gate report shape is unchanged.
const perColumnGates = isWorkflowColumnsEnabled(settings)
? [{
workflowId: "__default-workflow__",
workflowId: DEFAULT_WORKFLOW_POOL_ID,
columnId: "in-progress",
used: agentSlots,
limit: maxConcurrent,

27
packages/engine/src/workflow-graph-branches.ts
View File

@@ -2,6 +2,7 @@ import type { Settings, TaskDetail, WorkflowIrEdge, WorkflowIrNode } from "@fusi
import { WorkflowIrError } from "@fusion/core";
import type { WorkflowNodeOutcome, WorkflowNodeResult } from "./workflow-graph-executor.js";
import { schedulerLog } from "./logger.js";
/**
* Concurrent fan-out/join branch execution (U13, KTD-11, R21).
@@ -40,6 +41,26 @@ export interface WorkflowBranchPersistence {
loadBranchStates?(taskId: string, runId: string): WorkflowBranchRunState[] | Promise<WorkflowBranchRunState[]>;
}
/**
* Await a `saveBranchState` call inside a guard so a Promise-returning impl
* cannot escape as an unhandled rejection, and so a persistence failure never
* kills branch execution (log-and-continue). For a synchronous impl this
* preserves the prior behavior (the write completes before the caller proceeds).
*/
async function persistBranchState(
persistence: WorkflowBranchPersistence | undefined,
state: WorkflowBranchRunState,
): Promise<void> {
try {
await persistence?.saveBranchState?.(state);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
schedulerLog.warn(
`saveBranchState failed for task ${state.taskId} run ${state.runId} branch ${state.branchId}: ${message}`,
);
}
}
/** Minimal semaphore shape — structurally compatible with AgentSemaphore. */
export interface WorkflowBranchSemaphore {
run<T>(fn: () => Promise<T>): Promise<T>;
@@ -251,7 +272,7 @@ async function walkBranch(
} else {
const exec = async (): Promise<WorkflowNodeResult> => env.runBranchNode(node, signal);
lastResult = env.semaphore ? await env.semaphore.run(exec) : await exec();
env.persistence?.saveBranchState?.({
await persistBranchState(env.persistence, {
taskId: env.task.id,
runId: env.runId,
branchId: startNodeId,
@@ -266,7 +287,7 @@ async function walkBranch(
}
if (lastResult.outcome === "failure") {
env.persistence?.saveBranchState?.({
await persistBranchState(env.persistence, {
taskId: env.task.id,
runId: env.runId,
branchId: startNodeId,
@@ -282,7 +303,7 @@ async function walkBranch(
return { outcome: lastResult.outcome, lastNodeId: currentId };
}
if (next === joinId) {
env.persistence?.saveBranchState?.({
await persistBranchState(env.persistence, {
taskId: env.task.id,
runId: env.runId,
branchId: startNodeId,