Flow builder

New to flows? Start with Flows from scratch, which introduces each method with a small example. For exact inputs and return values, use the API glossary.

defineJevFlow(name, build)#

Import from jev-devtools/flow in Node code. defineJevFlow<Input>() accepts a name and a synchronous builder callback, and returns a definition with run(input, options). The callback returns flow.output() or a terminal flow.branch().

import { choice } from "@typesafe-ai/sdk";
import { defineJevFlow } from "jev-devtools/flow";
import { inspector, jev } from "./server.js";

export const triage = defineJevFlow<{ message: string }>("triage", (flow) => {
  const classification = flow.jev("classify", {
    state: flow.input,
    questions: {
      team: choice("Which team can help?", {
        billing: "Charges and payments",
        other: "Everything else",
      }),
    },
  });

  return flow.branch("route", {
    when: classification.answer("team").choice.eq("billing"),
    then: (path) => path.output("billing-team"),
    else: (path) => path.output("support-team"),
  });
});

export async function routeMessage(message: string) {
  return triage.run({ message }, { client: jev, devtools: inspector });
}

Describe a flow once, then run it with different inputs. Defining it does not send requests.

defineJevFlow<Input>(name, build)

name: The flow name shown in DevTools.
build: Receives a builder. Return an output or a terminal branch.

Add a Jev request step. It returns references to answers you can use in later steps.

flow.jev(id, { state, questions }, display?)

id: A unique step ID within this flow.
state: Context sent to Jev for this request. Flow references are resolved when the step runs.
questions: Named questions to ask. Their keys identify the returned answers.
display?: Optional labels and display metadata for DevTools.

Context sent to Jev for this request. Flow references are resolved when the step runs.

state

A reference to the data passed to run(). Inside map/loop programs, it refers to the item or iteration input instead.

flow.input: FlowExpression<Input>

Named questions to ask. Their keys identify the returned answers.

questions

Finish the flow by choosing one of two paths. Each path must return an output or another branch.

flow.branch(id, { when, then, else }, display?)

id: A unique step ID within this flow.
when: A flow condition evaluated when the branch runs.
then: The path used when the condition is true.
else: The path used when the condition is false.
display?: Optional labels and display metadata for DevTools.

A flow condition evaluated when the branch runs.

when

Reference the result of a named question. The name must match a key in questions; answer fields can be used in later steps.

request.answer(name)

A reference to the label selected by a Choice question. Its allowed labels come from that question’s alternatives.

choice: FlowExpression<Label>

Build an equality condition. It compares matching scalar values when the flow runs; it does not compare values during definition.

expression.eq(value)

The path used when the condition is true.

then

Finish this path with a named outcome and, optionally, a value for your application.

flow.output(name, value?, display?)

name: The outcome string returned by run().
value?: Optional result data. Narrow its type in your application before using it.
display?: Optional labels and display metadata for DevTools.

The path used when the condition is false.

else

Execute the flow with new input. Resolves to { runId, outcome, value? }.

program.run(input, { client, devtools?, signal? })

input: Data matching the input type declared in defineJevFlow.
client: The TypeSafe-compatible client that executes Jev requests.
devtools?: An optional observer, such as the shared server adapter, that records execution.
signal?: An optional AbortSignal to cancel the run.

The TypeSafe-compatible client that executes Jev requests.

client

An optional observer, such as the shared server adapter, that records execution.

devtools

API reference: defineJevFlow() · flow.jev() · flow.input · flow.branch() · answer.choice · expression.eq() · flow.output() · program.run()

Example walkthrough: the classify Jev step asks which team should handle the message. .answer("team").choice.eq("billing") turns that answer into a branch condition. Each path returns a terminal team name. routeMessage() executes the recipe with the instrumented client and shared inspector.

Definition and run interfaces#

Value / member	Type	Description
`triage.definition`	`FlowDefinition`	The static recipe graph.
`triage.run(input, options)`	`Promise<FlowOutcome & { runId: string }>`	Executes the recipe and resolves to the fields below.
`runId`	`string`	Unique execution ID.
`outcome`	`string`	`"billing-team"` or `"support-team"` for this example.
`value?`	`unknown`	Optional output data; omitted by this example.

run accepts a required client with systemOne, optional devtools: FlowObserver, and optional signal: AbortSignal. It resolves to { runId, outcome, value? }. definition exposes the static graph for inspection. Definitions make no SDK calls.

Requests and function steps#

Method	Arguments and result
`jev(id, request, display?)`	A regular SDK request whose state can contain expressions. Returns `.answer(name)` field references and `.answers`.
`jev(id, { input, request }, display?)`	Resolve input, then synchronously construct an SDK request in the factory. Use a preceding task for async work.
`task(id, { input, run, timeoutMs? }, display?)`	Resolve nested input references and execute application code. Returns a typed expression for the result.

run on a task receives (resolvedInput, context). TaskContext contains signal, runId, stepId, and executionId. Respond to the signal for cooperative cancellation. Timeouts cannot undo side effects or forcibly stop non-cooperating code. Tasks do not retry automatically.

display is optional static { title?, description? } metadata. Keep it public-safe. Use .answers when question keys are generated dynamically; individual .answer(name) references require the named question to exist at execution.

Return interfaces#

Method / member	Type	Description
`jev(...).answer(name)`	Object of typed field expressions	References to the selected question's SDK result fields.
`jev(...).answers`	`FlowExpression` of the answer object	Reference to all answers. Both Jev overloads return this interface.
`task(...)`	`FlowExpression<Awaited<O>>`	Reference to the application callback's awaited result.

Decisions and outcomes#

Method	Behavior
`branch(id, { when, then, else }, display?)`	Terminal decision. Both paths must return `output()` or another terminal branch.
`if(id, { when, then, else }, display?)`	Return a typed value from one path, then continue shared work.
`output(name, value?, display?)`	Finish with a named outcome and optional resolved value. Does not execute a business action.

Path callbacks build both possible paths synchronously; only the selected path executes. Conditions use inspectable expressions. You cannot use a JavaScript if on an expression to select runtime work.

Return interfaces#

Method	Type	Description
`branch(...)`	`FlowExpression<FlowOutcome>`	Terminal result of the selected path.
`if(...)`	`FlowExpression<A \| B>`	Selected value, available to shared steps after the join.
`output(...)`	`FlowExpression<FlowOutcome>`	Terminal outcome name and optional resolved value.

Parallel lanes#

parallel(id, lanes, display?) takes a named object of lane callbacks. Each lane returns an expression. The result is an object of joined expressions with matching names. Work inside a lane is sequential; lanes are concurrent. A sibling lane's private intermediate references are not available to other lanes.

const results = flow.parallel("load", {
  left: (lane) => lane.task("left-value", { input: 2, run: (n) => n * 2 }),
  right: (lane) => lane.task("right-value", { input: 3, run: (n) => n * 3 }),
});
return flow.output("joined", { left: results.left, right: results.right });

Run named lanes concurrently and join their returned expressions. Each lane's work stays sequential. Lane names belong to your application.

flow.parallel(id, lanes, display?)

lanes: An object of builder callbacks, each returning an expression. Private sibling intermediates are not available across lanes.

Resolve input references, run application code, and return a reference to its result. Tasks do not retry automatically.

flow.task(id, { input, run, timeoutMs? }, display?)

input: Values and nested flow references resolved before run is called.
run: Your callback receives the resolved input and task context, including a cancellation signal.
timeoutMs?: Optional time limit. The callback must cooperate with cancellation; a timeout cannot undo side effects.

Values and nested flow references resolved before run is called.

input

Your callback receives the resolved input and task context, including a cancellation signal.

run

API reference: flow.parallel() · flow.task() · flow.output()

Example walkthrough: the left lane doubles 2 and the right lane triples 3. The joined references are passed into output(), producing { left: 4, right: 9 }. The lane computations are independent; neither reads the other's private steps.

Returned member	Type	Resolved value in this example
`results.left`	`FlowExpression<number>`	`4`.
`results.right`	`FlowExpression<number>`	`9`.

This fragment belongs inside a defineJevFlow() callback. For a complete request-based example, see the full flow guide.

Collections and loops#

map(id, { items, input, maxItems, concurrency? }, build, display?) executes one bounded item program per item and returns results in input order. Its builder input is { item, index, context }. The input option supplies explicit shared context.

loop(id, { initial, input, while, maxIterations }, build, display?) tests the state condition before each iteration. Its builder input is { state, index, context }; return the next state. It returns the final state after the condition becomes false. If the condition still holds at the iteration limit, it throws FlowLoopLimitError.

These complete definitions demonstrate application-owned tasks without inference:

import { defineJevFlow, all, not } from "jev-devtools/flow";

export const summarize = defineJevFlow<{ values: number[] }>(
  "summarize",
  (flow) => {
    const doubled = flow.map(
      "double",
      {
        items: flow.input.get("values"),
        input: {},
        maxItems: 20,
        concurrency: 2,
      },
      (item) =>
        item.task("multiply", {
          input: item.input.get("item"),
          run: (value) => value * 2,
        }),
    );

    const total = flow.task("total", {
      input: doubled,
      run: (values) => values.reduce((sum, value) => sum + value, 0),
    });

    const label = flow.if("label", {
      when: all(total.gt(10), not(total.eq(20))),
      then: (path) =>
        path.task("large", { input: total, run: (n) => `Large: ${n}` }),
      else: (path) =>
        path.task("regular", { input: total, run: (n) => `Regular: ${n}` }),
    });
    return flow.output("summary", { total, label });
  },
);

export const count = defineJevFlow<{ target: number }>("count", (flow) => {
  const initial = flow.task("initial", {
    input: flow.input.get("target"),
    run: (target) => ({ value: 0, target }),
  });
  const finalState = flow.loop(
    "advance",
    {
      initial,
      input: {},
      while: (state) => state.get("value").lt(state.get("target")),
      maxIterations: 10,
    },
    (iteration) =>
      iteration.task("increment", {
        input: iteration.input.get("state"),
        run: (state) => ({ ...state, value: state.value + 1 }),
      }),
  );
  return flow.output("counted", finalState);
});

Run a bounded item program for each item. Results preserve input order even when items execute concurrently.

flow.map(id, options, build, display?)

items: An array or a reference to one. Each item is passed to the item program.
input: Explicit shared context for each item program, available as input.context.
maxItems: Required bound on the number of items.
concurrency?: Maximum number of item programs allowed to run concurrently.
build: Receives a builder with input { item, index, context }. Return an expression for the item result.

An array or a reference to one. Each item is passed to the item program.

items

Select a field from a referenced value. Keys come from your data shape. A missing field is an execution error.

expression.get(key)

Explicit shared context for each item program, available as input.context.

input

Required bound on the number of items.

maxItems

Maximum number of item programs allowed to run concurrently.

concurrency

Choose one value-returning path, then continue with its result. Unlike branch(), this does not finish the flow.

flow.if(id, { when, then, else }, display?)

when: A flow condition evaluated during the run.
then: Build the path selected when the condition is true. Return an expression for its value.
else: Build the path selected when the condition is false. Return an expression for its value.

A flow condition evaluated during the run.

when

Combine Boolean flow conditions: every condition must be true. All operands are evaluated; this does not short-circuit missing references.

all(...conditions)

Build a strictly-greater-than condition. Compares finite numeric values when the flow runs.

expression.gt(value)

Invert a Boolean flow condition, evaluated when the flow runs.

not(condition)

Build the path selected when the condition is true. Return an expression for its value.

then

Build the path selected when the condition is false. Return an expression for its value.

else

Test a condition before each iteration and return the final state. Rejects if the condition still holds at the iteration limit.

flow.loop(id, options, build, display?)

initial: The starting JSON state, or a reference to it.
input: Explicit shared context, available to the iteration program as input.context.
while: Build a Boolean flow condition from the current state. Tested before each iteration.
maxIterations: Required bound. Exceeding it while the condition is still true rejects with FlowLoopLimitError.
build: Receives a builder with input { state, index, context }. Return an expression for the next state.

The starting JSON state, or a reference to it.

initial

Explicit shared context, available to the iteration program as input.context.

input

Build a Boolean flow condition from the current state. Tested before each iteration.

while

Build a strictly-less-than condition. Compares finite numeric values when the flow runs.

expression.lt(value)

Required bound. Exceeding it while the condition is still true rejects with FlowLoopLimitError.

maxIterations

API reference: defineJevFlow() · flow.map() · flow.input · expression.get() · flow.task() · flow.if() · all() · expression.gt() · not() · expression.eq() · flow.output() · flow.loop() · expression.lt()

Example walkthrough: summarize uses map() to double each number, a task to add them, and if() to choose a label before returning the summary. all() and not() make the label policy explicit. With [2, 4], the output is { total: 12, label: "Large: 12" }. The separate count flow carries { value, target } as loop state and increments value until it reaches the target.

Method	Return type	Resolved value
`map(...)`	`FlowExpression<Output[]>`	Item results in input order.
`loop(...)`	`FlowExpression<State>`	Final state when its pre-iteration condition becomes false.

A loop can contain a map. Nested maps, nested loops, and loops inside maps are unsupported. Shared context is explicit; use context inside map/loop item programs rather than reaching across scope boundaries.

Limits and errors#

IDs must be unique across the whole definition, including untaken branches and lanes: 1–80 letters, digits, _, or -, starting with a letter or digit. Output names are node IDs too. Definitions allow 256 nodes, 16 nested blocks, and a run-wide budget of 1,024 expanded nodes.

Structural errors fail during definition. Missing input values and invalid comparisons fail when execution reaches them. FlowLoopLimitError is exported for bounded-loop handling. For exact collection/concurrency/timeout limits, execution errors, and cancellation behavior, consult the detailed guide and installed declarations.

Replay captured answers (unreleased)#

replayClient(fixture) returns an SDK-compatible client for a flow run. Jev steps receive stored answers without a network call or API key. This is not in npm 0.4.1 yet.

// Unreleased: use the working-tree package, not npm 0.4.1.
import { choice } from "@typesafe-ai/sdk";
import {
  defineJevFlow,
  replayClient,
  type FlowFixture,
} from "jev-devtools/flow";

export async function replayExample() {
  const flow = defineJevFlow<{ message: string }>("saved-triage", (flow) => {
    const result = flow.jev("classify", {
      state: flow.input,
      questions: {
        team: choice("Which team?", { billing: null, other: null }),
      },
    });
    return flow.output("routed", result.answer("team").choice);
  });

  // A hand-written fixture; Copy fixture can produce this shape from a run.
  const fixture: FlowFixture = {
    version: 1,
    flow: "saved-triage",
    steps: {
      classify: {
        model: "jev-sample",
        answers: {
          team: {
            type: "choice",
            choice: "billing",
            probabilities: { billing: 0.8, other: 0.2 },
            confidence: 0.6,
          },
        },
      },
    },
  };
  return flow.run(
    { message: "I was charged twice." },
    { client: replayClient(fixture) },
  );
  // outcome: "routed", value: "billing". No key or inference call.
}

Saved responses for replay. Does not contain the original input or the flow implementation.

{ version: 1, flow, steps }

version: Fixture format version. Currently 1.
flow: The name of the flow this fixture can replay.
steps: Saved responses indexed by step ID. IDs inside map/loop programs include their execution path.

Fixture format version. Currently 1.

version

The name of the flow this fixture can replay.

flow

Saved responses indexed by step ID. IDs inside map/loop programs include their execution path.

steps

Return an SDK-compatible client that serves saved answers without a key or inference call. Missing responses reject. Tasks and request factories still execute again.

replayClient(fixture)

API reference: defineJevFlow() · flow.jev() · flow.input · flow.output() · answer.choice · FlowFixture · program.run() · replayClient()

Example walkthrough: the fixture stores the response for the classify step. replayClient() supplies that response to the same flow, so the Choice expression resolves to "billing" without inference. The run returns outcome "routed" and value "billing".

API	Return type	Description
`replayClient(fixture)`	`TypeSafeClient`	Client that supplies saved answers inside a flow run.
`replayExample()`	Promise of a flow run result	Resolves to `{ runId, outcome: "routed", value: "billing" }`.

A FlowFixture contains { version: 1, flow, steps }. It contains responses, not the original input or flow implementation. Replay checks the flow name, step IDs, and requested question names; it does not validate prompts or state. Missing responses reject rather than calling live inference. Map/loop steps use composed IDs such as rounds/0/batch/1/assess.

Tasks and request factories execute again. A task can still perform external calls or side effects, so use controlled tasks when replaying in tests. Failed or cancelled captures are partial fixtures, not replays of recorded failures. See the replay guide for fixture copying and the repository CLI.

Build a fixture (unreleased)#

flowFixture(definition, run, traces) is available from both jev-devtools and jev-devtools/flow. Pass the captured flow definition/run and a ReadonlyMap<string, Trace> indexed by trace ID. It returns a BuiltFlowFixture with { fixture, missing, warnings }.

Check missing for completed steps whose response was unavailable, and warnings for capture limitations before saving fixture as JSON. Server redaction and truncation remain in the fixture. The drawer's Copy fixture also applies its current display masking; calling this helper directly does not add display masking.

Return interface#

Property	Type	Description
`fixture`	`FlowFixture`	Captured response bodies indexed by step ID.
`missing`	`string[]`	Completed steps whose response captures were unavailable.
`warnings`	`string[]`	Capture limitations to inspect before saving.

FlowBuilder, FlowRunOptions, FlowOutcome, FlowTask, FlowRequestFactory, TaskContext, and Resolved describe builder and execution contracts. FlowDefinition, FlowRun, FlowStep, FlowDisplay, and FlowObserver describe inspection data and callbacks. FlowFixture and BuiltFlowFixture describe replay data (unreleased). See recorder and types.

Compare a fixture with a live run (unreleased)#

compareWithFixture runs the same recipe and input twice: first with saved answers, then with your supplied client. formatComparison turns the result into readable text. This comparison performs live inference; ordinary replayClient does not.

import { compareWithFixture, formatComparison } from "jev-devtools/flow";

// program, input, fixture, client, and inspector come from your app.
const comparison = await compareWithFixture(program, input, fixture, {
  client,
  devtools: inspector,
  tolerance: 0.01,
});
console.log(formatComparison(comparison));

Example walkthrough: the comparison replays the stored answers, then runs with the supplied client using the same input. The tolerance applies to numeric answer-field differences. formatComparison() returns text, and the application's console.log() prints it.

Value / member	Type	Description
`compareWithFixture(...)`	`Promise<FixtureComparison>`	Resolves to outcome, value, decisions, answers, step differences, and a live fixture.
`comparison.changed`	`boolean`	Outcome, value, decision, or executed Jev-step differences.
`comparison.answers`	`AnswerChange[]`	Individual scalar answer-field comparisons.
`comparison.liveFixture`	`FlowFixture`	Unredacted, untruncated live response bodies.
`formatComparison(comparison)`	`string`	Plain-text description of changes.

Tasks execute on both sides. Use controlled, repeatable tasks; neither replay nor comparison undoes side effects. Top-level changed tracks outcome, value, decisions, and executed Jev steps; answer drift alone does not set it. Inspect answers for individual field differences. Numeric tolerance only applies to those answer fields. The returned liveFixture contains unredacted live responses, so review it before saving or sharing.

See FixtureComparison for every output field and the comparison contract for errors and cancellation.

Jev DevTools is an independent, open-source community project. It is not affiliated with TypeSafe AI.

defineJevFlow(name, build)#

Definition and run interfaces#

Requests and function steps#

Return interfaces#

Decisions and outcomes#

Return interfaces#

Parallel lanes#

Collections and loops#

Limits and errors#

Replay captured answers (unreleased)#

Build a fixture (unreleased)#

Return interface#

Related types#

Compare a fixture with a live run (unreleased)#