# @kindgi/flow > Flow model + predicate DSL for Kindgi™. Defines the task-flow shape (nodes, edges, conditional edges, data-flow mappings) executed by the kernel. Pure — no I/O, no database, no runtime. Ships types, a JSON-schema-validated loader (loadFlow / defineFlow), a predicate evaluator and mapping resolver (evaluateExpr / resolveMapping), and the scheduler tick (schedulerTick) the kernel uses to pick the nodes ready for concurrent execution. `npm install @kindgi/flow` · [source](https://github.com/kindgi/kindgi-sdk/tree/@kindgi/sdk@0.1.9/packages/flow) Flow type primitives, JSON-Schema validation, edge predicate evaluator, and scheduler-tick primitive for Kindgi. Consumed by the Kindgi runtime's kernel and any tool that authors, validates, or reasons about durable task flows. ## Exports - **`defineFlow(spec)`** — authoring surface. Thin wrapper over `loadFlow` that mirrors `defineTool` / `defineCheck` / `defineAgent`; the recommended entry point for pack authors declaring flows in TypeScript. Returns `Result`. - **`loadFlow(input)`** — validates against `flow.schema.json` + runs cross-cutting checks (reserved and duplicate ids, edge references, predicates, edge policy bounds, loop / fanout / subgraph rules, data-flow mappings, start edge, cycle detection, loop-body integrity). Returns `Result`. Kept as the wire-form entry point for JSON coming off disk / network. `GRAPH_SCHEMA_URI` is the `$id` it validates against. - **`evaluateExpr(expr, env)`** — deterministic boolean evaluator over the closed Expr DSL. - **`resolveMapping(mapping, env)`** — resolves a `Mapping` (leaf or subgraph `inputMapping`, flow `output.mapping`) against an `EvalEnv`; a path that doesn't resolve omits its key. `resolvePath(path, env)` resolves one path; `MISSING_PATH` is the marker it returns when the path doesn't resolve. - **`schedulerTick(flow, state)`** — pure per-tick output (`{ ready, edgeEvals, done }`) from a `SchedulerState`. - Type exports: `Flow`, `FlowSpec`, `FlowEdge`, `FlowNode`, `FlowOutputSpec`, `LeafNode`, `WhileLoopNode`, `ForeachLoopNode`, `LoopNode`, `LoopBody`, `LoopEdge`, `FanoutNode`, `FanoutBranch`, `ConvergenceMode`, `SubflowNode`, `SubflowRef`, `SubflowConvergence`, `Mapping`, `Expr`, `Operand`, `LiteralOperand`, `PathOperand`, `Path`, `PathRoot`, `OperatorName`, `LoopOutputSchema`, `EdgePolicy`, `RetryPolicy`, `BackoffShape`, `EvalEnv`, `SchedulerState`, `SchedulerTick`, `PendingEdgeEval`, and `FlowError` with one interface per error code. - Value exports: `EDGE_POLICY_PRIORITY_DEFAULT` (0), `EDGE_POLICY_PRIORITY_MIN` (-100), `EDGE_POLICY_PRIORITY_MAX` (100), `getEffectivePriority(edgesByTo, nodeId)` (resolves the effective scheduler priority for a node, applying the fan-in default rule), `OPERATORS`, `CONVERGENCE_MODES`, `SUBGRAPH_CONVERGENCE_MODES`, the sentinels `START_NODE` / `END_NODE` / `LOOP_START_NODE` / `LOOP_END_NODE` / `SENTINEL_IDS`, and the type guards `isLeafNode` / `isLoopNode` / `isFanoutNode` / `isSubflowNode`. ## Authoring Two entry points for building a `Flow`, both returning the same `Result` after the same validation gauntlet: ```ts excerpt import { defineFlow, loadFlow } from '@kindgi/flow'; import type { FlowSpec } from '@kindgi/flow'; // Recommended for pack authors — mirrors defineTool / defineCheck / defineAgent. const flow = defineFlow({ id: 'my.flow', version: '1.0.0', nodes: [{ id: 'step-a', kind: 'tool', ref: 'noop' }], edges: [ { id: 'e1', from: '$start', to: 'step-a' }, { id: 'e2', from: 'step-a', to: '$end' }, ], } satisfies FlowSpec); // Same validation surface — use loadFlow when the input is untrusted wire JSON. const loaded = loadFlow(JSON.parse(bytesFromDisk)); ``` `defineFlow` is byte-for-byte equivalent to `loadFlow` on the same input; the two share every check. Pick `defineFlow` when authoring in TypeScript to align with the other `define*` primitives; pick `loadFlow` when consuming flow JSON from disk, network, or an untrusted caller. ## Retry policy (schema-version 1.3.0+) Per-edge retry is declared via `FlowEdge.policy.retry`. The kernel honors this policy only when the destination node has EXACTLY ONE incoming edge — fan-in destinations (2+ incoming edges) ignore retry policies (there is no rule for whose policy would win). ```ts excerpt import type { FlowEdge, RetryPolicy } from '@kindgi/flow'; const retry: RetryPolicy = { maxAttempts: 5, // 1..10; 1 = no retry delayMs: 100, // ≥ 0; default 0 backoff: 'exponential', // 'fixed' | 'linear' | 'exponential'; default 'fixed' maxDelayMs: 5_000, // required when backoff = 'exponential' }; const edge: FlowEdge = { id: 'e1', from: 'produce', to: 'consume', policy: { retry }, }; ``` ### Valid ranges (enforced by `loadFlow`) | Field | Range | Notes | | ------------- | -------------------------------------- | ------------------------------------------------------------------------------------------ | | `maxAttempts` | `1..10` (integer) | Larger values are a design smell — retry-heavy control flow belongs in a supervising flow. | | `delayMs` | `≥ 0` (integer) | Default `0`. | | `backoff` | `'fixed' \| 'linear' \| 'exponential'` | Default `'fixed'`. | | `maxDelayMs` | `≥ 0` (integer) | REQUIRED when `backoff = 'exponential'`. Defaults to `60_000` for `linear`. | Loader errors: `invalid-edge-policy` (with `field: 'retry.maxAttempts' | 'retry.delayMs' | 'retry.backoff' | 'retry.maxDelayMs'` and a `reason` string) for semantic violations; `schema-validation-failed` for shape violations. ### Delay formula For the `attempt`-th retry (1-based; `attempt = 1` is the first retry after the initial failure): | Shape | Formula | | ------------- | ---------------------------------------------- | | `fixed` | `delayMs` | | `linear` | `min(delayMs * attempt, maxDelayMs ?? 60_000)` | | `exponential` | `min(delayMs * 2^(attempt - 1), maxDelayMs)` | Kernel runtime semantics + journal shape (`step.retry-scheduled`, resume behaviour, cancellation) are implemented by the Kindgi runtime's kernel. ## Timeout policy (schema-version 1.3.0+) Per-edge handler-invocation timeout is declared via `FlowEdge.policy.timeoutMs`. Same fan-in rule as retry: applies only to a destination node with EXACTLY ONE incoming edge. ```ts excerpt const edge: FlowEdge = { id: 'e1', from: 'produce', to: 'consume', policy: { timeoutMs: 5_000 }, }; ``` ### Valid ranges (enforced by `loadFlow`) | Field | Range | Notes | | ----------- | ------------------------ | ----------------------------------------------------------------------------------------------- | | `timeoutMs` | `1..3_600_000` (integer) | 1 hour cap. Longer work belongs behind `ctx.waitForToken` or in a `foreach` with `concurrency`. | Loader errors: `invalid-edge-policy` with `field: 'timeoutMs'` for out-of-band values; `schema-validation-failed` for shape violations. Kernel runtime semantics (timer wiring, `AbortSignal` composition, timeout+retry interaction, cancel+timeout attribution, loop-node bounding) are implemented by the Kindgi runtime's kernel. ## Concurrency policy (schema-version 1.4.0+) Per-edge kernel semaphore is declared via `FlowEdge.policy.concurrencyKey`. When set on the sole incoming edge of a destination node, the kernel serializes dispatch across every node (across every run, across every flow, within a single tenant) that shares the same key: at most one such node runs at a time. ```ts excerpt const edge: FlowEdge = { id: 'e-model', from: '$start', to: 'call-model', policy: { concurrencyKey: 'per-tenant:model-api' }, }; ``` ### Valid ranges (enforced by `loadFlow`) | Field | Range | Notes | | ---------------- | ------------------------------------------ | ----------------------------------------------------------------------------- | | `concurrencyKey` | 1..256 chars, printable ASCII (0x20..0x7E) | Coordination handle, not a display label. Non-printable / non-ASCII rejected. | Loader errors: `invalid-edge-policy` with `field: 'concurrencyKey'` for length / character-set violations; `schema-validation-failed` for shape violations. Kernel runtime semantics (lease acquire/release lifecycle, `step.concurrency-deferred` journal kind, waitForToken suspend-release / resume-reacquire, retry re-acquire, sweep on resume) are implemented by the Kindgi runtime's kernel. ## Priority policy (schema-version 1.5.0+) Per-edge scheduler priority is declared via `FlowEdge.policy.priority`. Same fan-in rule as retry / timeoutMs / concurrencyKey: applies only to a destination node with EXACTLY ONE incoming edge. Fan-in nodes inherit the default (`0`). ```ts excerpt const highPriority: FlowEdge = { id: 'e-urgent', from: '$start', to: 'expedite', policy: { priority: 90 }, }; const background: FlowEdge = { id: 'e-lazy', from: '$start', to: 'cleanup', policy: { priority: -20 }, }; ``` ### Valid ranges (enforced by `loadFlow`) | Field | Range | Notes | | ---------- | --------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `priority` | `-100..100` (integer) | Default `0`. Integer-only: floats introduce IEEE-754 comparison ambiguity in a replay-critical sort key. Bounded range because the primitive expresses tiers ("urgent" / "normal" / "background") plus finer-grained ordering — wider ranges add no expressive power. | Loader errors: `invalid-edge-policy` with `field: 'priority'` for range / integer / type violations; `schema-validation-failed` for shape violations. Kernel runtime semantics (dispatch ordering, maxParallelism interaction, concurrencyKey interaction, determinism) are implemented by the Kindgi runtime's kernel. ### `EdgePolicy` is a closed surface `retry` / `timeoutMs` / `concurrencyKey` / `priority` are the complete honored set. There is no `suspend` field — declarative suspension uses `ctx.waitForToken` at handler level. The schema enforces `additionalProperties: false`, so unknown edge-policy keys are rejected at load time. ## Loop primitive Loop node types + `outputSchema` well-formedness live here; the runtime lives in the Kindgi runtime's kernel. ## Fanout primitive (schema-version 1.6.0+) `FanoutNode` is a first-class node kind that runs N different handler bodies in parallel from a single dispatch, converging by a declared mode. Complementary to `foreach` (N iterations of the SAME body), fanout runs N DIFFERENT bodies concurrently. Shape: ```jsonc { "id": "the-fanout", "kind": "fanout", "branches": [ // >= 2, unique branchIds { "branchId": "a", "handler": "h-a", "outputSchema": { /* draft 2020-12 */ } }, { "branchId": "b", "handler": "h-b", "outputSchema": { /* draft 2020-12 */ } } ], "concurrency": 2, // optional, default = branches.length; 1..branches.length "convergence": "settle-all" // 'all-succeed' | 'any-succeed' | 'settle-all' } ``` Loader-enforced guardrails (`invalid-fanout-node`): - `branches.length >= 2` (a single-branch fanout is just a leaf node). - `branchId` non-empty + unique within a single fanout node. - `handler` non-empty (opaque to the loader — the kernel resolves it against the HandlerRegistry). - Per-branch `outputSchema` present + compiles against the draft-2020-12 meta-schema. - `concurrency` (when set) integer in `[1, branches.length]`. - `convergence` in the closed enum `{ 'all-succeed', 'any-succeed', 'settle-all' }`. Kernel runtime semantics (dispatch, sibling cancellation, replay, fan-in shape per convergence) are implemented by the Kindgi runtime's kernel. ## Subgraph primitive (schema-version 1.7.0+) `SubflowNode` is a first-class node kind that dispatches another registered flow as a full kernel sub-run. Complementary to loop + fanout: `loop.foreach` iterates the SAME body over N inputs; `fanout` runs N DIFFERENT bodies in parallel; `subgraph` runs a whole OTHER flow inline as a node. Shape: ```jsonc { "id": "the-subgraph", "kind": "subgraph", "flowRef": { "flowId": "child.flow.id", "version": "1.2.3" // EXACT semver (no ranges) }, "inputMapping": { // Record "target": { "path": "runInput.target" }, "hint": { "literal": "expedite" } }, "outputSchema": { /* draft 2020-12 */ }, // typed escape contract "convergence": "settle-all" // 'success-only' | 'settle-all' } ``` Loader-enforced guardrails (`invalid-subgraph-node`): - `flowRef.flowId` non-empty. - `flowRef.version` is EXACT semver (`^` / `~` / ranges rejected — deterministic replay requires exact resolution). - `inputMapping` is a plain object whose every value is a valid `Operand` (`{ path }` or `{ literal }`). - `outputSchema` present + compiles against the draft-2020-12 meta-schema (same discipline as loop / fanout output schemas). - `convergence` in the closed enum `{ 'success-only', 'settle-all' }`. Kernel runtime semantics (child-run dispatch, journal shape, replay via prior-dispatch reattach, cross-tenant guard, depth-limit enforcement) are implemented by the Kindgi runtime's kernel. ## Data flow (schema-version 1.8.0+) A tool or agent node can declare an `inputMapping`, and the flow can declare an `output`. Both are `Mapping`s: each key maps to a `{ literal }` or a `{ path }` rooted at `runInput`, `state` or `nodeOutputs.`, resolved with `resolveMapping`. A path that doesn't resolve omits its key. ```jsonc { "nodes": [ { "id": "parse", "kind": "tool", "ref": "acme.parse-profile" }, { "id": "rank", "kind": "tool", "ref": "acme.rank", "inputMapping": { // absent → the single upstream node's output "profile": { "path": "nodeOutputs.parse" }, "orgId": { "path": "runInput.orgId" } } } ], "output": { // absent → the output of the node feeding $end "mapping": { "ranking": { "path": "nodeOutputs.rank" } }, "schema": { "type": "object" } // optional; must compile } } ``` Loader errors: `invalid-mapping` when a mapping names a node the flow doesn't have, maps a node's own output, or uses a loop-only root (`iterationIndex` / `iterationOutput`); `invalid-flow-output` when `output.schema` doesn't compile; `schema-validation-failed` for a path with an unknown root. A conditional branch can rejoin: `schedulerTick` treats a node whose every incoming edge is not taken, or comes from such a skipped node, as dead. A join after a skipped arm becomes ready once its taken arms resolve, and a join whose arms are all skipped is skipped too. ## Schema drift `src/flow.schema.json` is bundled into the package for offline validation. It must match `@kindgi/specs/flow.schema.json` (a test compares the parsed JSON); the two change together.