diff --git a/packages/taskflow-core/src/compile.ts b/packages/taskflow-core/src/compile.ts index ed399488..fe5df1c4 100644 --- a/packages/taskflow-core/src/compile.ts +++ b/packages/taskflow-core/src/compile.ts @@ -20,7 +20,7 @@ import { LOOP_DEFAULT_MAX_ITERATIONS, TOURNAMENT_DEFAULT_VARIANTS, } from "./schema.ts"; -import { verifyTaskflow, type VerificationIssue, type VerificationResult } from "./verify.ts"; +import { verifyTaskflow, type TaskflowVerifier, type VerificationIssue, type VerificationResult } from "./verify.ts"; // --------------------------------------------------------------------------- // Types @@ -40,6 +40,9 @@ export interface CompileOptions { direction?: "TD" | "LR"; /** Document title (defaults to the flow name). */ title?: string; + /** Caller-supplied verifiers forwarded into `verifyTaskflow`; their issues + * overlay on the Mermaid diagram + report exactly like built-in issues. */ + verifiers?: TaskflowVerifier[]; } // --------------------------------------------------------------------------- @@ -308,11 +311,24 @@ function buildMermaid(flow: Taskflow, verification: VerificationResult, opts: Co // Final phases get a distinct border (unless they already carry an issue, // where the issue color wins — a final node that's broken should read red). - const finals = phases.filter((p) => p.final && !sev.has(p.id)).map((p) => idMap.get(p.id) ?? p.id); + // Overlays are applied ONLY to phase ids present in `idMap`: a plugin-supplied + // `phaseId` that names no real phase (or carries newlines/Mermaid syntax) must + // never reach a `class` statement, where it could inject extra directives. + // Such findings still appear as escaped text in the report below. + const finals = phases + .filter((p) => p.final && !sev.has(p.id)) + .map((p) => idMap.get(p.id)) + .filter((id): id is string => id !== undefined); if (finals.length) lines.push(`\tclass ${finals.join(",")} ${CLASS_FINAL};`); - const errNodes = [...sev].filter(([, s]) => s === "error").map(([id]) => idMap.get(id) ?? id); - const warnNodes = [...sev].filter(([, s]) => s === "warning").map(([id]) => idMap.get(id) ?? id); + const errNodes = [...sev] + .filter(([, s]) => s === "error") + .map(([id]) => idMap.get(id)) + .filter((id): id is string => id !== undefined); + const warnNodes = [...sev] + .filter(([, s]) => s === "warning") + .map(([id]) => idMap.get(id)) + .filter((id): id is string => id !== undefined); if (errNodes.length) lines.push(`\tclass ${errNodes.join(",")} ${CLASS_ERROR};`); if (warnNodes.length) lines.push(`\tclass ${warnNodes.join(",")} ${CLASS_WARN};`); @@ -341,13 +357,17 @@ function buildReport(flow: Taskflow, verification: VerificationResult): string { lines.push(""); lines.push(`### ❌ Errors (${errors.length})`); for (const e of errors) - lines.push(`- **${e.category}**${e.phaseId ? ` \`${mdInline(e.phaseId)}\`` : ""}: ${mdInline(e.message)}`); + lines.push( + `- **${e.category}**${e.phaseId ? ` \`${mdInline(e.phaseId)}\`` : ""}${e.source ? ` _(${mdInline(e.source)})_` : ""}: ${mdInline(e.message)}`, + ); } if (warnings.length) { lines.push(""); lines.push(`### ⚠️ Warnings (${warnings.length})`); for (const w of warnings) - lines.push(`- **${w.category}**${w.phaseId ? ` \`${mdInline(w.phaseId)}\`` : ""}: ${mdInline(w.message)}`); + lines.push( + `- **${w.category}**${w.phaseId ? ` \`${mdInline(w.phaseId)}\`` : ""}${w.source ? ` _(${mdInline(w.source)})_` : ""}: ${mdInline(w.message)}`, + ); } return lines.join("\n"); } @@ -361,12 +381,15 @@ function buildReport(flow: Taskflow, verification: VerificationResult): string { * report. Pure function — zero tokens, no LLM, no I/O. */ export function compileTaskflow(flow: Taskflow, opts: CompileOptions = {}): CompileResult { - const verification = verifyTaskflow({ - name: flow.name ?? "taskflow", - phases: flow.phases ?? [], - budget: flow.budget, - concurrency: flow.concurrency, - }); + const verification = verifyTaskflow( + { + name: flow.name ?? "taskflow", + phases: flow.phases ?? [], + budget: flow.budget, + concurrency: flow.concurrency, + }, + { verifiers: opts.verifiers }, + ); const mermaid = buildMermaid(flow, verification, opts); const report = buildReport(flow, verification); diff --git a/packages/taskflow-core/src/exec/driver.ts b/packages/taskflow-core/src/exec/driver.ts index 9168f402..9ea14063 100644 --- a/packages/taskflow-core/src/exec/driver.ts +++ b/packages/taskflow-core/src/exec/driver.ts @@ -27,6 +27,8 @@ import { foldEvents } from "./fold.ts"; import { EVENT_SCHEMA_VERSION, type Event } from "./events.ts"; import type { AgentConfig } from "../agents.ts"; import type { UsageStats } from "../usage.ts"; +import { pluginVerifierErrors } from "../verify.ts"; +import type { TaskflowVerifier } from "../verify.ts"; import { clampSubFlowBudget, containsInterpolationPlaceholder, @@ -49,6 +51,10 @@ export interface EventKernelDeps { eventKernel?: boolean; requestApproval?: (req: KernelApprovalRequest) => Promise; loadFlow?: (name: string) => Taskflow | undefined; + /** Caller-supplied zero-token verifiers (see verify.ts). Run in runNested's + * plugin-error preflight before every child-flow dispatch on this engine, and + * recursed via the ...deps spread. */ + verifiers?: TaskflowVerifier[]; _stack?: string[]; _dynamic?: boolean; } @@ -250,6 +256,37 @@ export async function runEventKernel(state: RunState, deps: EventKernelDeps): Pr blocked: false, }; } + // Plugin-error verifier preflight (no-spend gate, zero-token). runNested is + // the single funnel for BOTH inline-def and saved-use child dispatch on this + // engine, so centralizing here covers every child flow (review of #84, issue + // 1). It runs BEFORE childState is created, so a blocked child never + // dispatches. Verifiers see the DECLARED child def (opts.def), not the + // budget-clamped effectiveDef, matching what the imperative path's verify + // sites observe: a budget/concurrency policy verifier must see the same + // declaration on both engines, or the event kernel would silently let a child + // it blocks on the imperative path go on to spend. + // + // Built-in graph detectors (dead-ends, unreachable, …) are intentionally NOT + // run here: this is a plugin-only gate, matching the top-level preflight's + // advisory treatment of built-ins. It costs nothing reachable — the only + // error-severity built-ins are `unreachable` and self-dependency, and neither + // can reach a dispatching event-kernel child. A flow nesting an unreachable + // child is itself kernel-ineligible (concurrent DAG layers), so it runs on the + // imperative path where the built-in detectors still run; a self-dependency is + // rejected by validateTaskflow above. (Pinned in verify-pluggable.test.ts.) + const pluginErrors = pluginVerifierErrors( + { name: opts.def.name, phases: opts.def.phases as Phase[], budget: opts.def.budget, concurrency: opts.def.concurrency }, + deps.verifiers, + ); + if (pluginErrors) { + return { + finalOutput: `Nested flow '${effectiveDef.name}' failed verifier preflight: ${pluginErrors.join("; ")}`, + ok: false, + usage: emptyUsage(), + events: [], + blocked: false, + }; + } const childState: RunState = { // No `/` — validateRunId rejects path separators if ever persisted. runId: `${state.runId}-n-${effectiveDef.name.replace(/[^a-zA-Z0-9._-]/g, "_").slice(0, 40)}`, diff --git a/packages/taskflow-core/src/exec/step-kinds.ts b/packages/taskflow-core/src/exec/step-kinds.ts index 3fcdaadf..290368de 100644 --- a/packages/taskflow-core/src/exec/step-kinds.ts +++ b/packages/taskflow-core/src/exec/step-kinds.ts @@ -17,10 +17,9 @@ import { validateTaskflow, } from "../schema.ts"; import { parseGateVerdict, parseTournamentWinner } from "../deterministic.ts"; -import { verifyTaskflow } from "../verify.ts"; import { aggregateUsage, emptyUsage, type UsageStats } from "../usage.ts"; import { evaluateCondition, interpolate, interpolateValue, safeParse, tryEvaluateCondition, type InterpolationContext } from "../interpolate.ts"; -import { clampSubFlowBudget, kernelAttemptsOverBudget } from "./kernel-policy.ts"; +import { kernelAttemptsOverBudget } from "./kernel-policy.ts"; import { abortableDelay, isFailed as isFailedResult, isTransientError, mapWithConcurrencyLimit, PHASE_TIMEOUT_ABORT_GRACE_MS } from "../runner-core.ts"; import type { Event } from "./events.ts"; import { EVENT_SCHEMA_VERSION } from "./events.ts"; @@ -686,19 +685,12 @@ export async function executeFlowBody(phase: Phase, ctx: StepContext): Promise i.severity === "error"); - if (errs.length) { - return { midEvents: [], output: "", status: "done", usage: emptyUsage() }; - } - } - subDef = clampSubFlowBudget(wrapped, ctx.state.def.budget); + // Pass the DECLARED def (unclamped). runNested clamps to effectiveDef + // internally for dispatch, and its verifier preflight inspects opts.def — so + // this makes the event kernel verify the same declared budget/concurrency + // the imperative path verifies (wrapped), keeping the no-spend gate at parity + // across engines (review of #84, issue 1). + subDef = wrapped; recursionKey = `def:${subDef.name}`; } else { const useName = phase.use; diff --git a/packages/taskflow-core/src/runtime.ts b/packages/taskflow-core/src/runtime.ts index c81a5a04..73f7b848 100644 --- a/packages/taskflow-core/src/runtime.ts +++ b/packages/taskflow-core/src/runtime.ts @@ -37,7 +37,7 @@ const noRunnerInjected: RunTaskFn = async (_cwd, _agents, agentName, task) => ({ export { PHASE_TIMEOUT_ABORT_GRACE_MS } from "./runner-core.ts"; import { aggregateUsage, emptyUsage, type UsageStats } from "./usage.ts"; import { type Budget, type CacheScope, asArray, dependenciesOf, LOOP_DEFAULT_MAX_ITERATIONS, LOOP_HARD_MAX_ITERATIONS, MAX_DYNAMIC_MAP_ITEMS, MAX_DYNAMIC_NESTING, MAX_DYNAMIC_PHASES, parseTtlMs, type Phase, resolveArgs, type Taskflow, topoLayers, TOURNAMENT_DEFAULT_VARIANTS, TOURNAMENT_HARD_MAX_VARIANTS, type TournamentMode, validateInvocationArgs, validateTaskflow } from "./schema.ts"; -import { verifyTaskflow } from "./verify.ts"; +import { verifyTaskflow, pluginVerifierErrors, formatPluginIssueMessages, type TaskflowVerifier } from "./verify.ts"; import { combineScores, combineWithJudge, evaluatePureScorer, formatScorerReport, parseJudgeOutput, SCORE_DEFAULT_THRESHOLD, type ScoreConfig, scoreResultJSON, type ScorerResult, scorerShapeErrors } from "./scorers.ts"; import { parseGateVerdict, overBudget as overBudgetCheck, parseTournamentWinner, type BudgetCheckInput } from "./deterministic.ts"; export { parseTournamentWinner } from "./deterministic.ts"; @@ -117,6 +117,13 @@ export interface RuntimeDeps { * fallback. Default false; also set `PI_TASKFLOW_EVENT_KERNEL=1`. */ eventKernel?: boolean; + /** Caller-supplied zero-token verifiers (see verify.ts). Threaded into every + * `verifyTaskflow` preflight the runtime performs — dynamic subflows (spawn + + * inline `flow{def}`), the event-kernel path, and the top-level pre-execution + * gate (which blocks only on error-severity plugin issues, leaving built-in + * detectors advisory at the top level as before). A host embedding the engine + * registers the verifiers it trusts here. Undefined ⇒ built-in detectors only. */ + verifiers?: TaskflowVerifier[]; /** Internal: sub-flow call stack, for recursion detection. */ _stack?: string[]; /** Internal: pre-resolved Shared Context Tree dir for this run (sub-flows inherit the parent's). */ @@ -147,6 +154,11 @@ export interface RuntimeDeps { /** Internal: one immutable loader view per top-level execution. Saved-flow * definitions must not change between capability scan and execution. */ _flowLoaderSnapshot?: Map; + /** Internal: a child-flow dispatch site has already run the plugin-verifier + * preflight on the child's declared def, so executeTaskflow's top-level + * preflight must NOT re-run it on re-entry (avoids verifying each child 2-3x + * across the imperative path). Root/ctx_spawn entries leave this unset. */ + _verifierPreflightDone?: boolean; /** Internal phase-scoped W1a execution binding inherited only by descendants * that remain inside the selected cwd capability. */ _workspaceBinding?: ResolveOnlyPhaseBinding; @@ -870,7 +882,7 @@ async function runInlineSubflow( const error = `spawned subflow failed validation: ${v.errors.join("; ")}`; return { output: `(${error})`, usage: emptyUsage(), failed: true, error }; } - const ver = verifyTaskflow({ name: wrapped.name, phases: wrapped.phases as Phase[], budget: wrapped.budget, concurrency: wrapped.concurrency }); + const ver = verifyTaskflow({ name: wrapped.name, phases: wrapped.phases as Phase[], budget: wrapped.budget, concurrency: wrapped.concurrency }, { verifiers: deps.verifiers }); if (!ver.ok) { const errs = ver.issues.filter((i) => i.severity === "error").map((i) => i.message); const error = `spawned subflow failed verification: ${errs.join("; ")}`; @@ -903,6 +915,9 @@ async function runInlineSubflow( cwd: dynCwd, _cacheCwdIdentity: phase.cwd !== undefined || deps._cacheCwdIdentity !== undefined ? dynCwd : undefined, _dynamic: true, + // runInlineSubflow already ran the plugin-verifier preflight on this + // spawned child (above); don't re-run it on executeTaskflow re-entry. + _verifierPreflightDone: true, // The parent phase's isolated workspace (if any) applies only to the // parent — each spawned sub-phase resolves its own cwd. Clear the // override so the whole subflow doesn't inherit the parent's dir @@ -2813,8 +2828,21 @@ async function executePhaseInner( return defFailOpen(`inline def failed validation: ${v.errors.join("; ")}`); } // Static verification (dead-ends, unreachable, gate-exhaustion, budget, - // concurrency). Only error-severity issues block; warnings are advisory. - const ver = verifyTaskflow({ name: wrapped.name, phases: wrapped.phases as Phase[], budget: wrapped.budget, concurrency: wrapped.concurrency }); + // concurrency) + caller-supplied verifiers. Two contracts by issue origin: + // - Built-in error-severity issues FAIL-OPEN: a malformed LLM-authored def + // resolves as done/empty with a defError diagnostic and never aborts the + // run (authors who want a hard failure can gate downstream). Warnings + // are advisory. + // - Plugin-verifier errors FAIL-CLOSE: a host's trusted policy must block + // spend AND surface as a run failure — matching the saved-use path just + // below, runInlineSubflow, the event kernel (runNested), and the top-level + // preflight. Without this an inline-def child would silently bypass a + // plugin block with res.ok=true (review of #84). + const ver = verifyTaskflow({ name: wrapped.name, phases: wrapped.phases as Phase[], budget: wrapped.budget, concurrency: wrapped.concurrency }, { verifiers: deps.verifiers }); + const inlinePluginErrors = ver.issues.filter((i) => i.category === "plugin" && i.severity === "error"); + if (inlinePluginErrors.length) { + return failPhase(phase.id, `flow phase '${phase.id}': inline def '${wrapped.name}' failed verifier preflight: ${formatPluginIssueMessages(inlinePluginErrors).join("; ")}`); + } if (!ver.ok) { const errs = ver.issues.filter((i) => i.severity === "error").map((i) => i.message); return defFailOpen(`inline def failed verification: ${errs.join("; ")}`); @@ -2855,6 +2883,25 @@ async function executePhaseInner( // a bridge-bearing child must never be skipped by a cached parent result. const nestedBridgeTree = flowTreeUsesCwdBridge(subDef, deps.loadFlow); if (nestedBridgeTree) deps._disableCache = true; + // Plugin-error verifier preflight (no-spend gate) BEFORE cache/resume reuse, + // for SAVED-USE children only. Inline-def children are already gated by the + // verifyTaskflow in the hasDef branch above (plugin errors fail-close, + // built-in errors fail-open) — which also runs before this cache lookup — so + // re-verifying them here would be redundant. A cached saved-use child is + // returned just below WITHOUT re-entering executeTaskflow, and a saved-use + // child has no earlier verify site, so this is its single plugin gate and + // the cache-reuse guard (review + // of #84, issue 1). Blocks only on plugin error-severity issues; built-in + // detectors stay advisory. + if (!hasDef) { + const flowPluginErrors = pluginVerifierErrors( + { name: subDef.name, phases: subDef.phases as Phase[], budget: subDef.budget, concurrency: subDef.concurrency }, + deps.verifiers, + ); + if (flowPluginErrors) { + return failPhase(phase.id, `flow phase '${phase.id}': sub-flow '${subDef.name}' failed verifier preflight: ${flowPluginErrors.join("; ")}`); + } + } const flowCc: PhaseCacheCtx = nestedBridgeTree ? { ...cc, scope: "off" } : cc; // Every sub-flow cache identity includes the resolved definition. A saved // flow's name alone is insufficient: its contents can change without the @@ -2909,6 +2956,10 @@ async function executePhaseInner( cwd: effCwd, _cacheCwdIdentity: phase.cwd !== undefined || deps._cacheCwdIdentity !== undefined ? effCwd : undefined, _dynamic: hasDef || deps._dynamic === true ? true : undefined, + // The flow-phase handler already ran the plugin-verifier preflight on + // this child (inline-def via verifyTaskflow at the def branch, saved-use + // via pluginVerifierErrors just above); don't re-run it on re-entry. + _verifierPreflightDone: true, // The workspace override applies only to THIS flow phase, not to the // nested sub-phases (each resolves its own cwd). Clear it so the child // phases don't all inherit this phase's isolated dir as an override. @@ -4225,6 +4276,30 @@ export async function executeTaskflow(state: RunState, deps: RuntimeDeps): Promi "Use budget.maxTokens or a host with cost accounting.", ); } + // Top-level structural preflight (zero-token). Built-in detector findings + // (any severity) stay advisory at the top level — a flow may still run even + // with structural errors, exactly as before this seam existed. We block ONLY + // on error-severity issues from caller-supplied verifiers (deps.verifiers): + // category "plugin" is the canonical discriminator (a nameless verifier's + // fail-closed issue still carries it, whereas `source` would be undefined). + // Existing flows are unaffected and a host's trusted verifiers can gate spend + // before any agent is spawned. Plugin warnings never block here. SKIPPED on + // re-entry: a child-flow dispatch site (executePhaseInner / runInlineSubflow) + // already ran this gate on the child's declared def and sets + // _verifierPreflightDone, so we neither re-verify the same child nor re-run + // it against the budget-clamped child def that differs from what the + // dispatch site saw. + if (deps.verifiers?.length && !deps._verifierPreflightDone) { + const pluginErrors = pluginVerifierErrors( + { name: def.name, phases: def.phases as Phase[], budget: def.budget, concurrency: def.concurrency }, + deps.verifiers, + ); + if (pluginErrors) { + throw new Error( + `Taskflow '${def.name}' failed verifier preflight: ${pluginErrors.join("; ")}`, + ); + } + } // S2 strangler (default OFF): all phase kinds may use the event kernel when enabled. const { eventKernelEnabled, canUseEventKernel, runEventKernel } = await import("./exec/driver.ts"); // Existing phase state requires the imperative cache/inputHash machinery to @@ -4247,6 +4322,7 @@ export async function executeTaskflow(state: RunState, deps: RuntimeDeps): Promi persist: deps.persist, onProgress: deps.onProgress, eventKernel: deps.eventKernel, + verifiers: deps.verifiers, requestApproval: deps.requestApproval, loadFlow: deps.loadFlow, _stack: deps._stack, diff --git a/packages/taskflow-core/src/verify.ts b/packages/taskflow-core/src/verify.ts index df49f3db..eac85fe0 100644 --- a/packages/taskflow-core/src/verify.ts +++ b/packages/taskflow-core/src/verify.ts @@ -4,6 +4,9 @@ * Runs *before* any agent is spawned. Catches dead-end phases, unreachable * paths, gate exhaustion, budget overflow, and reference integrity issues * purely through graph algorithms on the DAG — no LLM required. + * + * Caller-supplied `TaskflowVerifier`s may also run, after the built-in + * detectors; they are pure by contract (no I/O, no LLM) — see TaskflowVerifier. */ import type { Phase } from "./schema.ts"; @@ -22,7 +25,8 @@ export type IssueCategory = | "concurrency" | "ref-integrity" | "guard-contradiction" - | "contract"; + | "contract" + | "plugin"; export interface VerificationIssue { /** Affected phase id, if applicable. */ @@ -30,6 +34,10 @@ export interface VerificationIssue { message: string; severity: "error" | "warning"; category: IssueCategory; + /** Name of the verifier that produced this issue. Undefined for the built-in + * structural detectors; set to the verifier's `name` on every issue emitted + * by a caller-supplied verifier (category "plugin"). */ + source?: string; } export interface VerificationResult { @@ -45,6 +53,196 @@ export interface VerifiableFlow { concurrency?: number; } +/** A single finding from a caller-supplied verifier. The engine stamps + * `category: "plugin"` and `source: `; a verifier only supplies + * what it actually knows — where, what, and how bad. */ +export interface VerifierIssue { + /** Affected phase id, if applicable. */ + phaseId?: string; + message: string; + severity: "error" | "warning"; +} + +/** A caller-supplied, zero-token static check plugged into `verifyTaskflow`. + * + * A verifier runs AFTER the built-in structural detectors, against the SAME + * sanitized flow, and its issues merge into the single `VerificationResult`. + * A verifier MUST be a pure function — no I/O, no LLM, zero tokens — matching + * this module's "no I/O" contract. A throwing or malformed verifier is + * fail-closed: normalized into a single `error`/`plugin` issue naming the + * verifier, and the remaining verifiers still run. */ +export interface TaskflowVerifier { + /** Stable, human-readable name. Attributes every issue the verifier emits + * (VerificationIssue.source) and appears in the fail-closed error message. */ + name: string; + /** Inspect the sanitized flow and return zero or more findings. Should not + * throw on well-formed input — if it cannot decide, return no issue. */ + verify: (flow: VerifiableFlow) => VerifierIssue[]; +} + +/** Options for {@link verifyTaskflow}. */ +export interface VerifyOptions { + /** Caller-supplied verifiers. Run after the built-in detectors, in array + * order, against the same sanitized flow; built-in issues always come first. */ + verifiers?: TaskflowVerifier[]; +} + +// --------------------------------------------------------------------------- +// Verifier isolation + fail-closed normalization +// --------------------------------------------------------------------------- + +/** Deeply freeze a plain-data object graph. Phases/budget are JSON-like data, + * so `Object.freeze` recurses cleanly with no non-freezable leaves. Used to + * hand each verifier an isolated snapshot it cannot mutate (protecting both the + * real execution plan and the data any later verifier observes). */ +function deepFreeze(value: T): T { + if (value && typeof value === "object") { + Object.freeze(value); + for (const v of Object.values(value as Record)) deepFreeze(v); + } + return value; +} + +/** Deep-clone a plain-data flow for verifier isolation. Unlike + * `structuredClone`, this never throws on a non-cloneable LEAF: a host may + * legitimately attach a function or Symbol to a phase (the schema permits + * `Record` fields such as `with`), and such a value must not + * block verification — it is dropped here. Phases/budget are JSON-like data, + * so no fidelity is lost on the fields a verifier inspects. A getter that + * throws while enumerating is NOT tolerated (the object is genuinely + * uninspectable); that case is fail-closed by the caller's try/catch. + * + * Cycles: a host may also attach a cyclic object graph (e.g. a node with a + * parent back-reference) to a phase field. Without a guard the clone — and + * then `deepFreeze` over its result — would recurse forever; the overflow is + * caught upstream and fail-closed, which blocks an otherwise-valid run. We + * instead track the ancestor chain on the current path and replace a back-edge + * with a plain `"[Circular]"` marker. Only genuine cycles are broken: a + * diamond/DAG shared subtree is no longer on the path once its first visit + * returns, so it clones normally on every reference. The snapshot therefore + * stays acyclic and a cyclic host value never blocks verification. */ +function cloneForVerifier(value: T): T { + return cloneForVerifierImpl(value, new WeakSet()) as T; +} + +function cloneForVerifierImpl(value: unknown, ancestors: WeakSet): unknown { + if (value === null || typeof value !== "object") return value; + // Back-edge to an ancestor still on the recursion stack ⇒ true cycle. Replace + // with a marker (NOT the original ref: that would leak the live plan into the + // snapshot and let a verifier mutate it). The marker is plain data, freezes + // cleanly, and keeps the clone acyclic so deepFreeze terminates. + if (ancestors.has(value as object)) return "[Circular]"; + ancestors.add(value as object); + try { + if (Array.isArray(value)) return value.map((v) => cloneForVerifierImpl(v, ancestors)); + const out: Record = {}; + for (const [k, v] of Object.entries(value as Record)) { + const t = typeof v; + if (t === "function" || t === "symbol") continue; + out[k] = cloneForVerifierImpl(v, ancestors); + } + return out; + } finally { + // Pop this object so a sibling reference to the same (non-cyclic) subtree + // clones in full — only an ancestor still above us on the stack is a cycle. + ancestors.delete(value as object); + } +} + +/** Produce an isolated, deeply-frozen snapshot of the sanitized flow for a + * verifier to inspect. The CLONE is the essential step, not the freeze: + * `safeFlow` shares its phase element references with the live execution plan + * (and, on the saved-use path, with the loader's run-wide cached snapshot), so + * freezing `safeFlow` in place would freeze those shared objects and break the + * runtime's later mutation of them (interpolation, clampSubFlowBudget). + * `cloneForVerifier` detaches a private graph first (tolerating non-cloneable + * host values); `deepFreeze` then makes that detached snapshot immutable, so the + * original definition and sibling verifiers see no mutation from any verifier. */ +function snapshotForVerifier(flow: VerifiableFlow): VerifiableFlow { + return deepFreeze(cloneForVerifier(flow)); +} + +/** Run caller-supplied verifiers against an isolated snapshot of `flow` and + * append their normalized issues to `issues`. Hardened (review of #84): + * - a non-array `verifiers` option is normalized to one fail-closed error; + * - if the isolated snapshot cannot be built (a throwing getter while cloning, + * since cloneForVerifier tolerates ordinary non-cloneable leaves), that is + * normalized to one fail-closed error and no verifier runs; + * - per entry, the name capture, shape check, AND invocation all run inside ONE + * try, so a malformed/Proxy entry that throws on `name`/`verify` access — or a + * verifier that throws — is normalized to one fail-closed plugin error using the + * name captured so far, and the loop continues (sibling verifiers still run); + * - each verifier sees a deep-frozen clone (issue #2: no plan mutation). */ +function runPluginVerifiers(flow: VerifiableFlow, verifiers: unknown, issues: VerificationIssue[]): void { + if (verifiers === undefined || verifiers === null) return; + if (!Array.isArray(verifiers)) { + issues.push({ + message: `verifiers option must be an array (got ${typeof verifiers})`, + severity: "error", + category: "plugin", + }); + return; + } + let snapshot: VerifiableFlow; + try { + snapshot = snapshotForVerifier(flow); + } catch (e) { + // cloneForVerifier tolerates non-cloneable leaves (functions/Symbols), + // so this only fires when the snapshot genuinely cannot be built — e.g. a + // getter that throws while enumerating. Fail closed once and skip verifiers. + issues.push({ + message: `verifier snapshot failed: ${e instanceof Error ? e.message : String(e)}`, + severity: "error", + category: "plugin", + }); + return; + } + for (const entry of verifiers) { + // Every property access below — name capture, shape check, invocation — is + // inside one try: a malformed/Proxy entry can throw on ANY access (not only + // inside verify()). The catch normalizes any throw to one fail-closed plugin + // issue using the name captured so far, and the loop continues so siblings run. + let name: string | undefined; + let label = ""; + try { + if (!entry || typeof entry !== "object") { + throw new Error("malformed (expected { name, verify })"); + } + if (typeof (entry as { name?: unknown }).name === "string") { + name = (entry as { name: string }).name; + label = name; + } + if (typeof (entry as { verify?: unknown }).verify !== "function") { + throw new Error("malformed (expected { name, verify })"); + } + const out = (entry as TaskflowVerifier).verify(snapshot); + if (!Array.isArray(out)) { + throw new Error(`returned a non-array (${typeof out})`); + } + for (const issue of out) { + issues.push({ + phaseId: issue?.phaseId, + message: + typeof issue?.message === "string" && issue.message + ? issue.message + : "(verifier emitted an issue with no message)", + // Default to "error" (fail-closed) when a verifier omits severity. + severity: issue?.severity === "warning" ? "warning" : "error", + category: "plugin", + source: name, + }); + } + } catch (e) { + issues.push({ + message: `verifier '${label}' failed: ${e instanceof Error ? e.message : String(e)}`, + severity: "error", + category: "plugin", + source: name, + }); + } + } +} + // --------------------------------------------------------------------------- // Graph helpers // --------------------------------------------------------------------------- @@ -420,9 +618,10 @@ function detectContractRefMismatches(phases: Phase[]): VerificationIssue[] { * Run all static verification passes against a parsed taskflow. * * Returns issues found; `ok === true` means no errors (warnings are ok). - * This is a pure function — no I/O, no LLM, zero tokens. + * The built-in detectors are pure (no I/O, no LLM, zero tokens); caller-supplied + * verifiers run after them and are pure by contract (see TaskflowVerifier). */ -export function verifyTaskflow(flow: VerifiableFlow): VerificationResult { +export function verifyTaskflow(flow: VerifiableFlow, options?: VerifyOptions): VerificationResult { // Tolerate malformed phase lists: null/non-object elements (validateTaskflow // reports them) would otherwise crash the graph helpers on `p.id`. Filter to // well-formed phase objects so verification degrades gracefully. @@ -440,6 +639,48 @@ export function verifyTaskflow(flow: VerifiableFlow): VerificationResult { issues.push(...detectGuardContradictions(phases)); issues.push(...detectContractRefMismatches(phases)); + // Caller-supplied verifiers run last, against an isolated deep-frozen snapshot + // of the sanitized flow (so a verifier cannot mutate the real execution plan + // or the data a later verifier sees). Fail-closed: a throwing, malformed, or + // shapeless verifier (incl. non-array entries / non-array return) is normalized + // to one error-severity "plugin" issue naming it, and the remaining verifiers + // still run. A verifier can never impersonate a built-in category — its + // findings are always stamped category "plugin" with source = its name. + runPluginVerifiers(safeFlow, options?.verifiers, issues); + const ok = !issues.some((i) => i.severity === "error"); return { ok, issues }; } + +/** Format verifier-origin issues into attributed message strings: the verifier + * `source` (its name) is prefixed when present, else the bare message. Shared by + * `pluginVerifierErrors` (the no-spend gate) and the imperative inline-def + * preflight in runtime.ts, so the attribution format lives in one place. */ +export function formatPluginIssueMessages(issues: VerificationIssue[]): string[] { + return issues.map((i) => (i.source ? `${i.source}: ${i.message}` : i.message)); +} + +/** Run only the caller-supplied verifiers against `flow` and return the messages + * of any error-severity plugin issues (prefixed with the verifier `source` when + * attributed), or `null` when there are none (or no verifiers registered). This + * is the no-spend gate called at every child-flow dispatch site so a host's + * trusted verifiers block spend uniformly across the imperative and event-kernel + * engines — see `runNested` (event kernel) and the top-level preflight in + * `executeTaskflow`. Pure, zero-token, fail-closed internally; it never throws. */ +export function pluginVerifierErrors( + flow: VerifiableFlow, + verifiers: TaskflowVerifier[] | undefined, +): string[] | null { + if (!verifiers || verifiers.length === 0) return null; + // Run ONLY the caller-supplied verifiers, not the built-in graph detectors. + // Every call site is a plugin-only no-spend gate where built-in findings are + // advisory or already enforced elsewhere, so re-running all 7 detectors per + // child dispatch is wasted work whose output is discarded. Sanitize the phase + // list exactly as verifyTaskflow does so verifiers see an equivalent view. + const phases = asArray(flow.phases).filter((p): p is Phase => !!p && typeof p === "object"); + const safeFlow: VerifiableFlow = { ...flow, phases }; + const issues: VerificationIssue[] = []; + runPluginVerifiers(safeFlow, verifiers, issues); + const errs = issues.filter((i) => i.category === "plugin" && i.severity === "error"); + return errs.length ? formatPluginIssueMessages(errs) : null; +} diff --git a/packages/taskflow-core/test/verify-pluggable.test.ts b/packages/taskflow-core/test/verify-pluggable.test.ts new file mode 100644 index 00000000..1e113cef --- /dev/null +++ b/packages/taskflow-core/test/verify-pluggable.test.ts @@ -0,0 +1,824 @@ +/** + * Pluggable verifier seam — `verifyTaskflow(flow, { verifiers })`. + * + * Covers: issue ordering (built-ins first, then verifiers in registration + * order), warning/error semantics, source/category attribution, fail-closed + * handling of throwing + malformed verifiers, back-compat (no verifiers ⇒ + * identical output), the compile Mermaid/report overlay, and no-spend runtime + * blocking on BOTH execution engines (imperative + event kernel). + */ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { verifyTaskflow, pluginVerifierErrors, type VerifiableFlow, type TaskflowVerifier } from "../src/verify.ts"; +import { compileTaskflow } from "../src/compile.ts"; +import type { CacheStore } from "../src/cache.ts"; +import type { AgentConfig } from "../src/agents.ts"; +import type { RunOptions, RunResult } from "../src/runner-core.ts"; +import { executeTaskflow, type RuntimeDeps } from "../src/runtime.ts"; +import { canUseEventKernel } from "../src/exec/driver.ts"; +import type { Phase, Taskflow } from "../src/schema.ts"; +import type { RunState } from "../src/store.ts"; +import { emptyUsage } from "../src/usage.ts"; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function agent(id: string, deps?: string[], overrides?: Partial): Phase { + return { id, type: "agent", task: "task for " + id, dependsOn: deps, ...overrides }; +} +/** A flow with one built-in dead-end warning on "a" (warning-only ⇒ ok:true). */ +function flowWithDeadEnd(): VerifiableFlow { + return { name: "test", phases: [agent("a"), agent("b"), agent("c", ["b"])] }; +} +/** A clean single-final-phase flow (zero built-in issues). */ +function cleanFlow(): VerifiableFlow { + return { name: "test", phases: [agent("a", undefined, { final: true })] }; +} + +// --------------------------------------------------------------------------- +// Ordering + attribution +// --------------------------------------------------------------------------- + +test("verifier: built-in issues precede verifier issues; verifier issues are stamped plugin + source", () => { + const flow = flowWithDeadEnd(); // built-in dead-end warning on "a" + const verifier: TaskflowVerifier = { + name: "lint", + verify: () => [{ phaseId: "b", message: "checker says no", severity: "warning" }], + }; + const r = verifyTaskflow(flow, { verifiers: [verifier] }); + + // Built-in issue comes first and carries no source. + assert.equal(r.issues[0].category, "dead-end"); + assert.equal(r.issues[0].source, undefined); + + // Verifier issue comes after, attributed and forced to the plugin category. + const vi = r.issues[r.issues.length - 1]; + assert.equal(vi.source, "lint"); + assert.equal(vi.category, "plugin"); + assert.equal(vi.severity, "warning"); + assert.equal(vi.phaseId, "b"); + assert.equal(vi.message, "checker says no"); +}); + +test("verifier: multiple verifiers run in registration order", () => { + const first: TaskflowVerifier = { name: "first", verify: () => [{ message: "one", severity: "warning" }] }; + const second: TaskflowVerifier = { name: "second", verify: () => [{ message: "two", severity: "warning" }] }; + const r = verifyTaskflow(cleanFlow(), { verifiers: [first, second] }); + assert.deepEqual( + r.issues.map((i) => i.source), + ["first", "second"], + ); +}); + +// --------------------------------------------------------------------------- +// Warning / error semantics +// --------------------------------------------------------------------------- + +test("verifier: a warning keeps ok true; an error flips ok false", () => { + assert.equal( + verifyTaskflow(cleanFlow(), { + verifiers: [{ name: "w", verify: () => [{ message: "meh", severity: "warning" }] }], + }).ok, + true, + ); + assert.equal( + verifyTaskflow(cleanFlow(), { + verifiers: [{ name: "e", verify: () => [{ message: "boom", severity: "error" }] }], + }).ok, + false, + ); +}); + +// --------------------------------------------------------------------------- +// Fail-closed: throwing + malformed verifiers +// --------------------------------------------------------------------------- + +test("verifier: a throwing verifier is normalized to one error/plugin issue and siblings still run", () => { + const r = verifyTaskflow(cleanFlow(), { + verifiers: [ + { name: "boom", verify: () => { throw new Error("exploded"); } }, + { name: "good", verify: () => [{ message: "hi", severity: "warning" }] }, + ], + }); + const fail = r.issues.find((i) => i.source === "boom"); + assert.ok(fail, "fail-closed issue emitted for the throwing verifier"); + assert.equal(fail!.severity, "error"); + assert.equal(fail!.category, "plugin"); + assert.match(fail!.message, /boom/); + assert.match(fail!.message, /exploded/); + // The sibling verifier still ran. + assert.ok(r.issues.some((i) => i.source === "good" && i.message === "hi")); + // The fail-closed issue is error-severity ⇒ the whole result is not ok. + assert.equal(r.ok, false); +}); + +test("verifier: a malformed (non-array) return is normalized to a fail-closed error issue", () => { + const r = verifyTaskflow(cleanFlow(), { + verifiers: [{ name: "bad", verify: () => "not-an-array" as unknown as [] }], + }); + const fail = r.issues.find((i) => i.source === "bad"); + assert.ok(fail); + assert.equal(fail!.severity, "error"); + assert.equal(fail!.category, "plugin"); + assert.match(fail!.message, /non-array/); + assert.equal(r.ok, false); +}); + +// --------------------------------------------------------------------------- +// Verifiers receive the sanitized flow (parity with built-in detectors) +// --------------------------------------------------------------------------- + +test("verifier: receives the sanitized safeFlow — null/non-object phases are filtered out", () => { + let seen: VerifiableFlow | undefined; + const spy: TaskflowVerifier = { + name: "spy", + verify: (f) => { seen = f; return []; }, + }; + assert.doesNotThrow(() => + verifyTaskflow({ name: "t", phases: [null as unknown as Phase, agent("a", undefined, { final: true })] }, { verifiers: [spy] }), + ); + assert.equal(seen?.phases.length, 1, "null phase filtered before the verifier sees the flow"); + assert.equal(seen?.phases[0].id, "a"); +}); + +// --------------------------------------------------------------------------- +// Back-compat: the seam is additive +// --------------------------------------------------------------------------- + +test("verifier: omitting verifiers, empty array, and undefined all behave identically to the pre-seam API", () => { + const flow = flowWithDeadEnd(); + const bare = verifyTaskflow(flow); + const withEmpty = verifyTaskflow(flow, { verifiers: [] }); + const withUndefined = verifyTaskflow(flow, { verifiers: undefined }); + assert.deepEqual(withEmpty.issues, bare.issues); + assert.deepEqual(withUndefined.issues, bare.issues); + assert.equal(withEmpty.ok, bare.ok); +}); + +// --------------------------------------------------------------------------- +// Compile overlay +// --------------------------------------------------------------------------- + +test("compile: plugin verifier issues overlay on the Mermaid diagram + report", () => { + const tf = { name: "t", phases: [agent("a", undefined, { final: true })] } as Taskflow; + const r = compileTaskflow(tf, { + verifiers: [{ name: "lint", verify: () => [{ phaseId: "a", message: "suspicious command", severity: "error" }] }], + }); + assert.equal(r.verification.ok, false); + // The report carries the attributed message + plugin category. + assert.match(r.markdown, /suspicious command/); + assert.match(r.markdown, /plugin/); + // The error-severity plugin issue paints node "a" with the error class. + assert.match(r.mermaid, /class a tfError/); +}); + +// --------------------------------------------------------------------------- +// Runtime: no-spend blocking on BOTH execution engines +// --------------------------------------------------------------------------- + +const AGENTS: AgentConfig[] = [{ name: "a", description: "test", systemPrompt: "", source: "user", filePath: "" }]; + +function mkState(def: Taskflow): RunState { + return { + runId: "v-run", + flowName: def.name, + def, + args: {}, + status: "running", + phases: {}, + createdAt: Date.now(), + updatedAt: Date.now(), + cwd: "/tmp", + }; +} +/** A runner that records every task it is asked to execute (for no-spend checks). */ +function recordingRunner(record: string[]): RuntimeDeps["runTask"] { + return async (_cwd, _agents, agentName, task, _o: RunOptions): Promise => { + record.push(task); + return { + agent: agentName, + task, + exitCode: 0, + output: "ok", + stderr: "", + usage: { ...emptyUsage(), output: 1 }, + stopReason: "end", + }; + }; +} +function baseDeps(runTask: RuntimeDeps["runTask"], extra: Partial = {}): RuntimeDeps { + return { cwd: "/tmp", agents: AGENTS, runTask, persist: () => {}, onProgress: () => {}, ...extra }; +} + +/** A clean linear chain eligible for BOTH engines (no built-in issues, kernel-admitted). */ +const chainDef: Taskflow = { + name: "vchain", + phases: [ + { id: "one", type: "agent", agent: "a", task: "start" }, + { id: "two", type: "agent", agent: "a", task: "end", dependsOn: ["one"], final: true }, + ], +}; +const blocker: TaskflowVerifier = { + name: "block", + verify: () => [{ phaseId: "one", message: "verifier says stop", severity: "error" }], +}; + +test("runtime: a blocking verifier aborts the imperative run before any spend", async () => { + const record: string[] = []; + const deps = baseDeps(recordingRunner(record), { verifiers: [blocker] }); + const res = await executeTaskflow(mkState(chainDef), deps); + assert.equal(res.ok, false); + assert.equal(record.length, 0, "runTask never invoked — zero spend"); + assert.match(res.finalOutput ?? "", /verifier preflight/); +}); + +test("runtime: a blocking verifier aborts the event-kernel run before any spend", async () => { + // Sanity: the flow is kernel-eligible, so this exercises the event-kernel + // dispatch branch (not just the imperative fallback). + assert.equal(canUseEventKernel(chainDef), true); + const record: string[] = []; + const deps = baseDeps(recordingRunner(record), { verifiers: [blocker], eventKernel: true }); + const res = await executeTaskflow(mkState(chainDef), deps); + assert.equal(res.ok, false); + assert.equal(record.length, 0, "runTask never invoked — zero spend (event-kernel path)"); + assert.match(res.finalOutput ?? "", /verifier preflight/); +}); + +test("runtime: a nameless verifier's fail-closed error still blocks the run (category, not source, discriminates)", async () => { + // Regression: the top-level preflight discriminates plugin issues by + // category === "plugin", not source !== undefined. A fail-closed error from a + // verifier without a name is stamped source: undefined but category "plugin"; + // gating on source would silently let it through. + const record: string[] = []; + const namelessThrower = { + name: undefined as unknown as string, + verify: () => { throw new Error("boom"); }, + } as TaskflowVerifier; + const deps = baseDeps(recordingRunner(record), { verifiers: [namelessThrower] }); + const res = await executeTaskflow(mkState(chainDef), deps); + assert.equal(res.ok, false); + assert.equal(record.length, 0, "zero spend — fail-closed plugin error gates the run even with no source attribution"); +}); + +test("runtime: a warning-only verifier does NOT block the run", async () => { + const record: string[] = []; + const warnOnly: TaskflowVerifier = { + name: "advisory", + verify: () => [{ phaseId: "one", message: "just a heads up", severity: "warning" }], + }; + const deps = baseDeps(recordingRunner(record), { verifiers: [warnOnly] }); + const res = await executeTaskflow(mkState(chainDef), deps); + assert.equal(res.ok, true, "warnings never block at the top level"); + assert.equal(record.length, 2, "both phases executed despite the warning"); +}); + +// --------------------------------------------------------------------------- +// Child-flow dispatch parity — the no-spend gate covers nested flows on BOTH +// engines (review of #84, issue 1: event-kernel saved-flow previously bypassed). +// --------------------------------------------------------------------------- + +/** A saved child flow the parent will `use`. Single agent phase ⇒ kernel-eligible. */ +const savedChild: Taskflow = { + name: "child", + phases: [{ id: "c1", type: "agent", agent: "a", task: "run inside child" }], +}; +/** A verifier that blocks ONLY when the flow being verified is the child. */ +const childOnlyBlocker: TaskflowVerifier = { + name: "childOnly", + verify: (f) => (f.name === "child" ? [{ phaseId: "c1", message: "child flow forbidden", severity: "error" }] : []), +}; +const parentUsingChild: Taskflow = { + name: "parent", + phases: [{ id: "run-child", type: "flow", use: "child", final: true }], +}; +function childLoader(name: string): Taskflow | undefined { + return name === "child" ? savedChild : undefined; +} + +test("runtime: a blocking verifier on a SAVED-flow child aborts the imperative run before child spend", async () => { + // Imperative path: the saved flow recurses through executeTaskflow, whose + // top-level preflight blocks on the child. Makes the plumbing explicit. + const record: string[] = []; + const deps = baseDeps(recordingRunner(record), { verifiers: [childOnlyBlocker], loadFlow: childLoader }); + const res = await executeTaskflow(mkState(parentUsingChild), deps); + assert.equal(res.ok, false); + assert.equal(record.length, 0, "child never dispatched — zero spend (imperative)"); + // The verifier message lands in the failed flow phase's error (the parent's + // finalOutput is "(no output)" for a failed final phase on the imperative path). + assert.match(res.state.phases["run-child"]?.error ?? "", /verifier preflight/); +}); + +test("runtime: a blocking verifier on a SAVED-flow child aborts the event-kernel run before child spend", async () => { + // THE reproduction from review of #84: previously the event kernel returned + // ok:true and ran the child (runNested had no plugin preflight). Now runNested + // is the centralized chokepoint for both inline-def and saved-use children. + assert.equal(canUseEventKernel(parentUsingChild, childLoader), true); + const record: string[] = []; + const deps = baseDeps(recordingRunner(record), { + verifiers: [childOnlyBlocker], + loadFlow: childLoader, + eventKernel: true, + }); + const res = await executeTaskflow(mkState(parentUsingChild), deps); + assert.equal(res.ok, false); + assert.equal(record.length, 0, "child never dispatched — zero spend (event-kernel path)"); + assert.match(res.finalOutput ?? "", /verifier preflight/); +}); + +// --------------------------------------------------------------------------- +// Cache/reuse parity — the preflight fires before a cached child is reused +// (review of #84, issue 1: "including before cache/resume reuse"). +// --------------------------------------------------------------------------- + +/** An in-memory CacheStore stand-in: isolated per instance (no disk, no + * project-root walk via findProjectFlowsDir), so a cross-run cache test is + * deterministic on every host regardless of tmpdir/home layout. */ +function memCacheStore(): CacheStore { + const map = new Map>(); + return { + get: (key: string) => map.get(key) ?? null, + put: (entry: Record & { key: string }) => { + map.set(entry.key, entry); + }, + clear: () => { + const n = map.size; + map.clear(); + return n; + }, + } as unknown as CacheStore; +} + +test("runtime: a blocking verifier on a child still blocks when the child result is cached (no cache-reuse bypass)", async () => { + const cacheableChild: Taskflow = { + name: "child", + phases: [{ id: "c1", type: "agent", agent: "a", task: "cached work" }], + }; + const parentCached: Taskflow = { + name: "parent", + phases: [{ id: "run-child", type: "flow", use: "child", final: true }], + }; + const loader = (n: string): Taskflow | undefined => (n === "child" ? cacheableChild : undefined); + const store = memCacheStore(); + + // Run 1 — no verifiers: primes the cross-run cache (parent flow phase + child). + const record1: string[] = []; + const r1 = await executeTaskflow( + mkState(parentCached), + baseDeps(recordingRunner(record1), { loadFlow: loader, cacheStore: store, cacheScopeDefault: "cross-run" }), + ); + assert.equal(r1.ok, true); + assert.equal(record1.length, 1, "run 1 executed the child's agent once"); + + // Run 2 — register a blocker on the child. Without the preflight-before-cache + // the cached parent-flow result would be reused and the verifier bypassed. + const record2: string[] = []; + const r2 = await executeTaskflow( + mkState(parentCached), + baseDeps(recordingRunner(record2), { + loadFlow: loader, + cacheStore: store, + cacheScopeDefault: "cross-run", + verifiers: [childOnlyBlocker], + }), + ); + assert.equal(r2.ok, false, "a cached child must not bypass the verifier"); + assert.equal(record2.length, 0, "zero new spend on run 2"); + assert.match(r2.state.phases["run-child"]?.error ?? "", /verifier preflight/); +}); + +test("runtime: a blocking verifier on an INLINE-DEF child aborts the event-kernel run before child spend", async () => { + // Inline def whose parsed name the verifier blocks on. After removing the + // redundant step-kinds verify, runNested is the single preflight chokepoint, + // so the inline-def branch is covered too (and verifiers run once, not twice). + const parentInline: Taskflow = { + name: "parentInline", + phases: [ + { + id: "run-inline", + type: "flow", + def: { name: "inline-child", phases: [{ id: "i1", type: "agent", agent: "a", task: "x" }] }, + final: true, + }, + ], + }; + const inlineBlocker: TaskflowVerifier = { + name: "inlineOnly", + verify: (f) => (f.name === "inline-child" ? [{ message: "inline child forbidden", severity: "error" }] : []), + }; + assert.equal(canUseEventKernel(parentInline), true); + const record: string[] = []; + const deps = baseDeps(recordingRunner(record), { verifiers: [inlineBlocker], eventKernel: true }); + const res = await executeTaskflow(mkState(parentInline), deps); + assert.equal(res.ok, false); + assert.equal(record.length, 0, "inline child never dispatched — zero spend (event-kernel path)"); + assert.match(res.finalOutput ?? "", /verifier preflight/); +}); + +test("runtime: a blocking verifier on an INLINE-DEF child aborts the IMPERATIVE run before child spend", async () => { + // Cross-engine parity with the test above. Previously the imperative inline-def + // path fail-opened (defFailOpen -> status:'done' -> res.ok=true), silently + // swallowing the host's policy block; only the event kernel fail-closed. Now + // plugin errors fail-close on both engines (built-in errors still fail-open). + const parentInline: Taskflow = { + name: "parentInline", + phases: [ + { + id: "run-inline", + type: "flow", + def: { name: "inline-child", phases: [{ id: "i1", type: "agent", agent: "a", task: "x" }] }, + final: true, + }, + ], + }; + const inlineBlocker: TaskflowVerifier = { + name: "inlineOnly", + verify: (f) => (f.name === "inline-child" ? [{ message: "inline child forbidden", severity: "error" }] : []), + }; + // No eventKernel flag -> imperative path (the hasDef branch in executePhaseInner). + const record: string[] = []; + const deps = baseDeps(recordingRunner(record), { verifiers: [inlineBlocker] }); + const res = await executeTaskflow(mkState(parentInline), deps); + assert.equal(res.ok, false, "imperative: a plugin block fail-closes (parity with the event kernel)"); + assert.equal(record.length, 0, "inline child never dispatched — zero spend (imperative path)"); + assert.match(res.state.phases["run-inline"]?.error ?? "", /verifier preflight/); +}); + +// --------------------------------------------------------------------------- +// Issue 2: a verifier cannot mutate the real execution plan. +// --------------------------------------------------------------------------- + +test("verifier: receives a deep-frozen snapshot — it cannot mutate the real plan or affect later verifiers", () => { + const flow: VerifiableFlow = { + name: "t", + phases: [{ id: "a", type: "agent", task: "SAFE", final: true }], + budget: { maxTokens: 100 }, + }; + const originalTask = flow.phases[0].task; + const originalBudget = flow.budget!.maxTokens; + + const mutator: TaskflowVerifier = { + name: "mutator", + verify: (f) => { + // Attempt to rewrite a phase task + raise the budget cap. On a frozen + // snapshot this is either silently ignored (sloppy) or throws (strict) — + // either way the real plan must come out unchanged. + try { + (f.phases[0] as { task: string }).task = "MUTATED"; + (f.budget as { maxTokens: number }).maxTokens = 9_999_999; + } catch { + /* strict-mode TypeError on a frozen object — expected */ + } + return []; + }, + }; + let spySawTask: unknown; + const spy: TaskflowVerifier = { + name: "spy", + verify: (f) => { + spySawTask = f.phases[0].task; + return []; + }, + }; + + verifyTaskflow(flow, { verifiers: [mutator, spy] }); + + // The real execution plan is unchanged. + assert.equal(flow.phases[0].task, originalTask); + assert.equal(flow.budget!.maxTokens, originalBudget); + // A sibling verifier observed the un-mutated value (snapshot is frozen, so a + // mutation by one verifier cannot leak into the snapshot another sees). + assert.equal(spySawTask, "SAFE"); +}); + +// --------------------------------------------------------------------------- +// Issue 3: malformed-verifier fail-closed handling is complete. +// --------------------------------------------------------------------------- + +test("verifier: a null entry in the registry is fail-closed and siblings still run", () => { + // Previously: [null, good] → try block threw on v.verify, catch re-read + // v.name (null) → second TypeError → good never ran. + const good: TaskflowVerifier = { name: "good", verify: () => [{ message: "hi", severity: "warning" }] }; + const r = verifyTaskflow(cleanFlow(), { + verifiers: [null as unknown as TaskflowVerifier, good], + }); + const fail = r.issues.find((i) => i.message.includes("malformed")); + assert.ok(fail, "null entry normalized to a fail-closed malformed/plugin error"); + assert.equal(fail!.severity, "error"); + assert.equal(fail!.category, "plugin"); + assert.ok(r.issues.some((i) => i.source === "good" && i.message === "hi"), "sibling verifier still ran"); + assert.equal(r.ok, false); +}); + +test("verifier: a non-array registry is normalized to a single fail-closed error", () => { + const r = verifyTaskflow(cleanFlow(), { + verifiers: "not-an-array" as unknown as TaskflowVerifier[], + }); + assert.equal(r.ok, false); + assert.ok(r.issues.some((i) => i.category === "plugin" && i.message.includes("must be an array"))); +}); + +test("verifier: an entry whose verify is not a function is fail-closed", () => { + const r = verifyTaskflow(cleanFlow(), { + verifiers: [{ name: "shapeless", verify: "nope" as unknown as () => [] }], + }); + const fail = r.issues.find((i) => i.source === "shapeless"); + assert.ok(fail); + assert.equal(fail!.severity, "error"); + assert.match(fail!.message, /malformed/); + assert.equal(r.ok, false); +}); + +// --------------------------------------------------------------------------- +// Issue 4: an unknown plugin phaseId cannot inject Mermaid syntax. +// --------------------------------------------------------------------------- + +test("compile: a plugin phaseId not present in the DAG cannot inject Mermaid syntax via the class statement", () => { + const tf = { name: "t", phases: [agent("a", undefined, { final: true })] } as Taskflow; + // A malicious plugin-supplied phaseId carrying a newline + a classDef + // directive. Previously emitted raw into `class tfError;`. + const evil = "a\nclassDef x fill:#fff\n"; + const r = compileTaskflow(tf, { + verifiers: [{ name: "evil", verify: () => [{ phaseId: evil, message: "x", severity: "error" }] }], + }); + assert.equal(r.verification.ok, false); + // The real node "a" is NOT painted (the plugin issue named an unknown id). + assert.doesNotMatch(r.mermaid, /class a tfError/); + // And the injected classDef directive did not leak into the diagram. + assert.doesNotMatch(r.mermaid, /classDef x/); + // The finding still appears (escaped) in the report, attributed to "evil". + assert.match(r.markdown, /evil/); +}); + +// --------------------------------------------------------------------------- +// Hardening pass (on top of #84): budget-clamp parity across engines, +// "never throws" fail-closed, and single-run (no redundant verification). +// --------------------------------------------------------------------------- + +// A saved-use child that DECLARES a budget larger than its parent caps. +const bigBudgetChild: Taskflow = { + name: "child", + budget: { maxUSD: 100 }, + phases: [{ id: "c1", type: "agent", agent: "a", task: "big-budget child work" }], +}; +const parentWithBudgetCap: Taskflow = { + name: "parent", + budget: { maxUSD: 30 }, + phases: [{ id: "rc", type: "flow", use: "child", final: true }], +}; +const bigBudgetLoader = (n: string): Taskflow | undefined => (n === "child" ? bigBudgetChild : undefined); + +test("runtime: a budget-policy verifier sees the DECLARED child budget on both engines (no clamp divergence)", async () => { + // The verifier must observe the child's declared budget (100), not the value + // clampSubFlowBudget tightens to the parent's cap (30). Before the fix the + // event kernel verified the clamped def and saw 30. + assert.equal(canUseEventKernel(parentWithBudgetCap, bigBudgetLoader), true, "flow is kernel-eligible"); + const observed: Record = {}; + const observer: TaskflowVerifier = { + name: "cap-observer", + verify: (f) => { + if (f.name === "child") observed.child = f.budget?.maxUSD; + return []; + }, + }; + for (const eventKernel of [false, true]) { + observed.child = undefined; + await executeTaskflow( + mkState(parentWithBudgetCap), + baseDeps(recordingRunner([]), { verifiers: [observer], loadFlow: bigBudgetLoader, eventKernel }), + ); + assert.equal(observed.child, 100, `${eventKernel ? "event-kernel" : "imperative"} saw the DECLARED budget, not the clamped 30`); + } +}); + +test("runtime: a blocking budget-policy verifier stops a child from spending on BOTH engines", async () => { + // The no-spend guarantee: a child whose declared budget violates policy is + // blocked before any agent runs, on both engines. (Before the fix the event + // kernel saw the clamped value and let the child spend.) + assert.equal(canUseEventKernel(parentWithBudgetCap, bigBudgetLoader), true); + const cap: TaskflowVerifier = { + name: "cap", + verify: (f) => (f.budget?.maxUSD ?? 0) > 50 ? [{ message: "declares too much budget", severity: "error" }] : [], + }; + for (const eventKernel of [false, true]) { + const record: string[] = []; + const res = await executeTaskflow( + mkState(parentWithBudgetCap), + baseDeps(recordingRunner(record), { verifiers: [cap], loadFlow: bigBudgetLoader, eventKernel }), + ); + assert.equal(record.length, 0, `${eventKernel ? "event-kernel" : "imperative"}: child never dispatched (zero spend)`); + assert.equal(res.ok, false, `${eventKernel ? "event-kernel" : "imperative"}: run fails (saved-use fail-closes)`); + } +}); + +test("runtime: an inline-def child's declared budget is also seen unclamped on the event kernel", async () => { + // Same parity check for the inline-def path (step-kinds passes the declared + // `wrapped`, not a pre-clamped def, to runNested). + const parentInlineBudget: Taskflow = { + name: "parent", + budget: { maxUSD: 30 }, + phases: [ + { + id: "rc", + type: "flow", + def: { name: "inline-child", budget: { maxUSD: 100 }, phases: [{ id: "i1", type: "agent", agent: "a", task: "inline big-budget" }] }, + final: true, + }, + ], + }; + assert.equal(canUseEventKernel(parentInlineBudget), true); + const observed: Record = {}; + const observer: TaskflowVerifier = { + name: "cap-observer", + verify: (f) => { + if (f.name === "inline-child") observed.c = f.budget?.maxUSD; + return []; + }, + }; + await executeTaskflow(mkState(parentInlineBudget), baseDeps(recordingRunner([]), { verifiers: [observer], eventKernel: true })); + assert.equal(observed.c, 100, "event-kernel inline-def saw the DECLARED budget, not the clamped 30"); +}); + +test("runtime: a plugin verifier runs exactly ONCE per child on each engine (no redundant preflight)", async () => { + // Regression guard: a non-cached flow-phase child must not be verified 2-3x. + const counts: Record = {}; + const counter: TaskflowVerifier = { + name: "counter", + verify: (f) => { + counts[f.name] = (counts[f.name] ?? 0) + 1; + return []; + }, + }; + // saved-use child, both engines + for (const eventKernel of [false, true]) { + counts.child = 0; + await executeTaskflow( + mkState(parentUsingChild), + baseDeps(recordingRunner([]), { verifiers: [counter], loadFlow: childLoader, eventKernel }), + ); + assert.equal(counts.child, 1, `${eventKernel ? "event-kernel" : "imperative"} saved-use: verified once`); + } + // inline-def child, both engines + const parentInline1: Taskflow = { + name: "parentInline", + phases: [{ id: "run-inline", type: "flow", def: { name: "inline-child", phases: [{ id: "i1", type: "agent", agent: "a", task: "x" }] }, final: true }], + }; + for (const eventKernel of [false, true]) { + counts["inline-child"] = 0; + await executeTaskflow(mkState(parentInline1), baseDeps(recordingRunner([]), { verifiers: [counter], eventKernel })); + assert.equal(counts["inline-child"], 1, `${eventKernel ? "event-kernel" : "imperative"} inline-def: verified once`); + } +}); + +test("verifier: a throwing name/verify getter is normalized (never throws, siblings run)", () => { + // A Proxy/getter entry that throws on property access must be fail-closed, + // not crash verifyTaskflow (pluginVerifierErrors documents "never throws"). + const good: TaskflowVerifier = { name: "good", verify: () => [{ message: "sibling ran", severity: "warning" }] }; + const evil: Record = { name: "evil" }; + Object.defineProperty(evil, "verify", { get() { throw new Error("boom-getter"); }, enumerable: true }); + const r = verifyTaskflow(cleanFlow(), { verifiers: [evil as unknown as TaskflowVerifier, good] }); + assert.equal(r.ok, false, "the throwing-getter entry is a fail-closed plugin error"); + assert.ok(r.issues.some((i) => i.category === "plugin" && i.severity === "error" && i.message.includes("boom-getter")), "throwing getter normalized to a plugin error"); + assert.ok(r.issues.some((i) => i.source === "good"), "sibling verifier still ran"); +}); + +test("verifier: a non-cloneable leaf (function/Symbol) is dropped, not fail-closed", () => { + // A host may attach a callback or Symbol to a phase (the schema permits + // Record fields like `with`). The isolation snapshot must + // tolerate it — dropping the non-cloneable leaf — rather than fail-closing + // verification, which would block the run whenever any verifier is registered. + let sawTask: unknown; + const flow: VerifiableFlow = { + name: "t", + phases: [{ id: "a", type: "agent", task: "x", final: true, with: { cb: () => 1, sym: Symbol("s") } } as unknown as Phase], + }; + const r = verifyTaskflow(flow, { + verifiers: [{ name: "spy", verify: (f) => { sawTask = f.phases[0].task; return []; } }], + }); + assert.equal(r.ok, true, "non-cloneable leaves do not block verification"); + assert.equal(sawTask, "x", "the verifier still ran and observed the cloneable phase data"); +}); + +test("verifier: a throwing getter while cloning the snapshot is fail-closed", () => { + // cloneForVerifier tolerates ordinary non-cloneable leaves, but a getter that + // throws while enumerating genuinely cannot be inspected — that stays + // fail-closed (rare, malformed host object). + const phase: Record = { id: "a", type: "agent", task: "x", final: true }; + Object.defineProperty(phase, "boom", { get() { throw new Error("getter-boom"); }, enumerable: true }); + const flow: VerifiableFlow = { name: "t", phases: [phase as unknown as Phase] }; + const r = verifyTaskflow(flow, { verifiers: [{ name: "spy", verify: () => [] }] }); + assert.equal(r.ok, false, "a throwing getter is a fail-closed plugin error"); + assert.ok(r.issues.some((i) => i.category === "plugin" && i.message.includes("snapshot")), "snapshot failure normalized to a plugin error"); +}); + +test("verifier: pluginVerifierErrors attributes the verifier source in its messages", () => { + // The runtime preflight surfaces these strings; source attribution lets a host + // tell WHICH verifier blocked. + const errs = pluginVerifierErrors(cleanFlow(), [ + { name: "named", verify: () => [{ message: "nope", severity: "error" }] }, + ]); + assert.deepEqual(errs, ["named: nope"]); + const errsUnnamed = pluginVerifierErrors(cleanFlow(), [ + { name: undefined as unknown as string, verify: () => [{ message: "anon", severity: "error" }] } as TaskflowVerifier, + ]); + assert.deepEqual(errsUnnamed, ["anon"], "no source prefix when the verifier is unnamed"); +}); + +// --------------------------------------------------------------------------- +// Regression: a cyclic host value must not block verification. Before the +// cycle guard, cloneForVerifier recursed forever on a back-edge → RangeError → +// fail-closed → a valid run was blocked whenever ANY verifier was registered +// (even one that never reads `with`). Cycles are now broken at the back-edge +// with a "[Circular]" marker; shared (DAG) subtrees still clone in full. +// --------------------------------------------------------------------------- + +test("verifier: a cyclic host value in a phase does not block verification (no snapshot overflow)", () => { + // A host attaches a cyclic graph to a phase `with` field (the schema permits + // Record). Before the guard this overflowed the isolation + // snapshot and blocked the run; now the back-edge is replaced with a marker. + const node: Record = { id: "child" }; + node.parent = { id: "root", children: [node] }; // cycle: node.parent.children[0] === node + const flow: VerifiableFlow = { + name: "t", + phases: [{ id: "a", type: "agent", task: "x", final: true, with: { node } } as unknown as Phase], + }; + + let ran = false; + let sawTask: unknown; + const benign: TaskflowVerifier = { + name: "ids-only", + verify: (f) => { + ran = true; + sawTask = f.phases[0].task; + return []; + }, + }; + + const r = verifyTaskflow(flow, { verifiers: [benign] }); + assert.equal(r.ok, true, "cyclic host data does not block verification"); + assert.equal(ran, true, "the verifier still ran"); + assert.equal(sawTask, "x", "the verifier observed the cloneable phase data"); + assert.ok(!r.issues.some((i) => /snapshot|Maximum call stack/i.test(i.message)), "no snapshot-overflow issue"); + // pluginVerifierErrors is the runtime no-spend gate — it must not report errors. + assert.equal(pluginVerifierErrors(flow, [benign]), null); + // The ORIGINAL flow's cyclic structure is intact (the marker lives on the clone only). + assert.equal((flow.phases[0] as unknown as { with: { node: { parent: { children: unknown[] } } } }).with.node.parent.children[0], node); +}); + +test("verifier: a shared (DAG) subtree is cloned fully — only true cycles are broken", () => { + // Two phases reference the SAME sub-object (a diamond/DAG, not a cycle). It + // must clone in full on both phases — NOT be replaced with "[Circular]". + const shared = { kind: "config", n: 7 }; + const flow: VerifiableFlow = { + name: "t", + phases: [ + { id: "a", type: "agent", task: "x", final: true, with: { c: shared } } as unknown as Phase, + { id: "b", type: "agent", task: "y", dependsOn: ["a"], with: { c: shared } } as unknown as Phase, + ], + }; + + const observed: unknown[] = []; + const spy: TaskflowVerifier = { + name: "spy", + verify: (f) => { + for (const p of f.phases) observed.push((p as { with?: { c?: unknown } }).with?.c); + return []; + }, + }; + + verifyTaskflow(flow, { verifiers: [spy] }); + assert.equal(observed.length, 2); + assert.deepEqual(observed[0], { kind: "config", n: 7 }, "first phase saw the full shared subtree"); + assert.deepEqual(observed[1], { kind: "config", n: 7 }, "second phase saw the full subtree, not a cycle marker"); + assert.ok(observed.every((o) => o !== "[Circular]"), "no false cycle marker on a shared (non-cyclic) subtree"); +}); + +// --------------------------------------------------------------------------- +// Pin: built-in graph detectors are advisory on the event-kernel child path +// (this PR centralized a PLUGIN-only preflight in runNested). The only +// error-severity built-ins — `unreachable` and self-dependency — cannot reach a +// dispatching event-kernel child, so no spend divergence with the imperative +// path is reachable. This locks that in so it can't silently regress. +// --------------------------------------------------------------------------- + +test("runtime: an event-kernel child with a built-in error-severity issue is screened out before dispatch (no spend divergence)", () => { + // Pins the absence of a cross-engine spend divergence. The only error-severity + // built-in detectors are `unreachable` and self-dependency; all others are + // warnings (advisory on BOTH engines, never blocked). Neither error-severity + // built-in can reach a DISPATCHING event-kernel child: + const unreachableChild: Taskflow = { + name: "child", + phases: [ + { id: "a", type: "agent", agent: "a", task: "child-a", final: true }, + { id: "b", type: "agent", agent: "a", task: "child-b", dependsOn: ["a"] }, + { id: "c", type: "agent", agent: "a", task: "child-c", dependsOn: ["d"] }, + { id: "d", type: "agent", agent: "a", task: "child-d" }, + ], + }; + assert.equal(verifyTaskflow(unreachableChild).ok, false, "the child carries an error-severity built-in (unreachable)"); + // canUseEventKernel recurses into nested inline defs: the unreachable child's + // concurrent DAG layers make the PARENT kernel-ineligible, so the whole run + // takes the imperative path — where the built-in detectors still run. + const parent: Taskflow = { name: "parent", phases: [{ id: "run-child", type: "flow", def: unreachableChild, final: true }] }; + assert.equal(canUseEventKernel(parent), false, "nesting an unreachable child makes the flow kernel-ineligible"); +});