Skip to content

feat(core): pluggable zero-token verifiers for verifyTaskflow #84

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
heggria merged 2 commits into from
Jul 20, 2026
Merged

feat(core): pluggable zero-token verifiers for verifyTaskflow #84

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
54f8397
feat(core): pluggable zero-token verifiers for verifyTaskflow
hirisov Jul 18, 2026
5b01256
fix(core): harden pluggable verifiers (review of #84)
hirisov Jul 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 35 additions & 12 deletions packages/taskflow-core/src/compile.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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[];
}

// ---------------------------------------------------------------------------
Expand Down Expand Up @@ -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};`);

Expand Down Expand Up @@ -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");
}
Expand All @@ -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);
Expand Down
37 changes: 37 additions & 0 deletions packages/taskflow-core/src/exec/driver.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -49,6 +51,10 @@ export interface EventKernelDeps {
eventKernel?: boolean;
requestApproval?: (req: KernelApprovalRequest) => Promise<KernelApprovalDecision>;
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;
}
Expand Down Expand Up @@ -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)}`,
Expand Down
22 changes: 7 additions & 15 deletions packages/taskflow-core/src/exec/step-kinds.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -686,19 +685,12 @@ export async function executeFlowBody(phase: Phase, ctx: StepContext): Promise<B
if (!v.ok) {
return { midEvents: [], output: "", status: "done", usage: emptyUsage() };
}
const ver = verifyTaskflow({
name: wrapped.name,
phases: wrapped.phases as Phase[],
budget: wrapped.budget,
concurrency: wrapped.concurrency,
});
if (!ver.ok) {
const errs = ver.issues.filter((i) => 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;
Expand Down
84 changes: 80 additions & 4 deletions packages/taskflow-core/src/runtime.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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). */
Expand Down Expand Up @@ -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<string, FlowLoaderSnapshotEntry>;
/** 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;
Expand Down Expand Up @@ -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("; ")}`;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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("; ")}`);
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand All @@ -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,
Expand Down
Loading