Agent logic flows (`@lacrew/flows`)

Flows are declarative pipelines of agent logic — model calls, LaCrew tools, policy-gated spends, and branches — that the orchestrator executes against its live runtime. The same JSON definition powers the cloud's visual Flow Builder (UX-first) and the code-first SDK path shown here, and the builder always exposes the definition as both JSON and this exact TypeScript.

Flows never hold keys and never touch the treasury: every onchain effect is policy-checked first and then either proposed as a spend intent or routed into governance, so policy stacks, escalation, and the audit trail apply exactly as they do for any other agent action.

Step kinds

Prompts and string args interpolate {{input}}, {{input.<key>}}, {{steps.<id>.text}}, {{steps.<id>.json}}, {{steps.<id>.verdict}}, and {{steps.<id>.json.<path>}} — a dotted path into an earlier step's result, so one tool call can hand a value to the next ({{steps.read.json.body.sha}}) without a model being asked to copy it. Unknown references render empty, and a path only reads own properties. Steps fall through in declaration order unless a step routes explicitly; null stops the flow. Cycles are rejected — recurrence belongs to the trigger layer instead:

Scope

A flow carries a scope that decides who can see and invoke it:

Scope is also a policy ceiling. A run always executes as its invoking principal — never as the scope — so effective authority is min(principal, scope): both policy stacks are read and the stricter verdict wins. An org-scoped flow invoked by a junior agent still only gets that agent's authority, and an agent-scoped flow invoked by a manager is capped at the scoped agent's limits.

How the ceiling is enforced

For spend value, the ceiling is enforced onchain. Before a scoped run proposes anything, the orchestrator issues that run a session key whose maxValue is min(principal cap, scope cap), read from SpendCapPolicy. EscalationRouter checks every propose against the key's limits, so an over-ceiling spend reverts with SessionValueExceeded — the chain refuses it, not the orchestrator. Sessions are cached per (agent, limits), so a wide key issued for an unscoped run is never reused for a tighter-scoped one.

Other policy dimensions — rate limits, time windows — are not carried by the session key. For those the ceiling remains an orchestrator-side check, and a compromised orchestrator could skip it. The principal's own stack is still enforced onchain in every case, which is the guarantee that protects the treasury.

Trust boundary. Session limits are only as strong as the issuer role. SessionRegistry.issue is root-or-issuer, and the orchestrator normally holds the issuer role, so a compromised orchestrator could mint itself a wider key. It cannot do so silently: every issue emits an onchain event, which Guardian can alert on and the root can revoke. Running the issuer as a separate key the orchestrator does not hold closes this gap.

Constitutional steps

org and budget steps do not write directly. Org structure and treasury grants are constitutional, and the orchestrator holds short-lived session keys only — letting it rewrite the org chart would be exactly the custody LaCrew refuses. So these steps always raise a governance proposal, and the policy verdict picks the tier:

Authority is read from SpendCapPolicy rather than the full stack: the target of an org action is a node, not a payee, so consulting WhitelistPolicy would deny every such action for a reason unrelated to authority.

budget: run-epoch is the exception and writes directly — the orchestrator is the EpochStreamer operator by design.

org distinguishes removal from suspension, because they are not the same decision:

• fire → OrgRegistry.removeNode. Permanent, and the node's children are rewired to its parent.
• deactivate / activate → OrgRegistry.setActive. Reversible; the node keeps its place in the chart and its reporting line.

Delegation

An agent step can hand work to another agent — a prompt, or a whole flow via flowId. The nested run gets its own principal, so the delegate acts under its own policy stack: a flow cannot borrow authority by invoking a more privileged agent.

Delegation is bounded. validateFlow rejects cycles between a flow's own edges, but a flowId is not an edge, so the runtime tracks the chain of flows on the stack: revisiting one fails with flow_delegation_cycle, and a chain deeper than four levels fails with flow_delegation_too_deep. A delegate that fails also fails the delegating step, rather than returning the failure as data for the parent to ignore.

Delegation and ask / gate

A delegated flow can stop on the same things any flow stops on: an ask-mode connector write waiting on a confirmation, or a blocking human gate. When it does, the delegating run parks too — status waiting, pause reason awaiting_child, and the child's run id as the pause token. Nothing below the agent step runs, and no second question is asked: the child's question is raised and answered where the write actually is, and the parent shows up in the stalled-run list pointing at it.

What holds it together:

• The link is durable, both ways. The child's run state records the parent run and the step it was delegated from; the parent's pause records the child. A restart hydrates both, and neither is picked up by boot recovery — a parked run is waiting on something, not stalled.
• The child ending is what releases the parent, whether it completed, failed, or was cancelled. The wake is a claim on the parent's waiting state, so a child ending in one replica and an operator pressing Resume in another cannot both continue the same run.
• A resumed parent re-enters the agent step and finds the run it already delegated rather than starting a second one — the delegate is not re-run, and the write behind the confirmation goes out exactly once. One child per (run, step) is a uniqueness constraint in the durable state, not a convention.
• The delegate's outcome is the step's outcome. A completed child returns its last summary as the step's text (branch on it like any other output); a child that failed fails the step with flow_delegate_failed; a cancelled one fails it with flow_delegate_cancelled. A rejected human gate inside the child is not a failure — the child takes the gate's reject port, finishes, and the parent continues from what it decided.
• Cancelling a parent cancels its delegates, recursively, and closes their asks and gates. The policy is cancel, not abandon: a delegated run left runnable behind a cancelled parent is a pipeline that spends on work nobody is waiting for.

Nesting still stops at four levels, and every level parks — a question three deep parks the whole chain and one answer releases it back up.

Triggers

Flows fire four ways, declared on the definition:

• manual (default) — run from the UI, SDK, or CLI.
• epoch — the orchestrator fires it on every payroll epoch, right after allowances stream (both the queue schedule and POST /epoch do this, in mock mode too), and the run is tagged trigger: "epoch" in the trace and the audit trail. The shipped treasury-pulse template is epoch-triggered out of the box.
• cron — a 5-field UTC schedule expression (*/5 * * * * style), minute resolution, fired at most once per matching minute by the orchestrator's provider-agnostic scheduler.
• webhook — a signed HTTP delivery from outside the org, see Webhook triggers.

{ "trigger": "cron", "schedule": "0 9 * * 1-5" }

A flow can also be run by a crew heartbeat without declaring anything: a heartbeat works through a checklist an operator wrote, so the flow's own trigger is irrelevant to it and a definition cannot opt into being on one. Those runs are tagged trigger: "heartbeat" — see Crew heartbeat for when to reach for one instead of a per-flow cron.

Persistence

Definitions and run traces persist to Postgres when DATABASE_URL is set (orchestrator_flows / orchestrator_flow_runs, plus orchestrator_webhook_triggers / orchestrator_webhook_deliveries — same @lacrew/db family as the audit trail) and hydrate back on boot; without a database everything still works in memory. /health reports which store is active under flows.store.

Run lifecycle — pause, resume, cancel

A run outlives the process that started it. After every completed step the orchestrator writes a checkpoint — the cursor, the outputs so far, the run's principal and input — to orchestrator_flow_checkpoints, and moves the run's cursor in orchestrator_flow_run_state, both in one transaction. A run can therefore stop and be picked back up later, by a different replica.

A run pauses in five ways: a wait step it declared, a human gate holding it until someone decides, an ask-mode connector write waiting on a human (connectors), a delegated flow that parked on any of those (see Delegation and ask / gate), or an operator asking for a pause. waiting.reason says which — awaiting_human, awaiting_webhook, human_gate, connector_ask, awaiting_child, or operator.

const flows = createFlowsClient({ baseUrl, token });

await flows.pauseRun(runId, "checking something"); // honoured at the next step
const finished = await flows.resumeRun(runId); // continues from the checkpoint
await flows.cancelRun(runId, "not doing it"); // terminal

Pause and cancel are requests, not mutations: the run may be moving inside another replica, and the only safe place to honour one is between two steps — never inside a write that is already in the air. A cancelled run asked to resume answers 409 run_cancelled.

No double write on resume

Before every side-effecting step (gate, org, budget, governance, agent, and any connector route) the orchestrator opens an attempt record, and closes it once the call returns. A process that dies in between leaves the attempt open, which is the one state that means "this write may already have happened".

On boot the orchestrator reads back every unfinished run and:

• resumes runs that stopped cleanly between two steps — the steps that already ran are not repeated;
• fails closed on a run with an open attempt, naming the step and the attempt key to reconcile against. It is not retried, because redoing it could pay twice and skipping it could skip a payment, and only a human knows which happened;
• leaves paused runs alone — they are waiting on something, not stalled.

A step whose repeat is genuinely harmless can opt into the retry:

flow("pr-triage", "PR triage")
  .tool("merge", "github.merge_pull_request", { number: "7" }, { idempotent: true })
  .build();

idempotent is off by default and is a claim about the far side of the call that LaCrew cannot check — untrue, it is a double spend.

Pausing an agent (POST /agents/pause) cancels that agent's parked runs with the reason attached: a paused agent should spend nothing, and a resumable run is authority waiting to be spent. Resuming a run always uses the run's original principal and scope, so a pause can never launder authority.

Blocking human gates

A human step is the one that says "a person decides before the rest of this runs". Entering it posts one question into the run's thread, parks the run on the durable state above, and goes no further. When someone answers with one of the offered options, the run resumes down that option's port.

flow("shortlist", "Publish the shortlist")
  .model("draft", { prompt: "Draft a shortlist from {{input}}", next: "signoff" })
  .human("signoff", {
    prompt: "Publish this shortlist?\n{{steps.draft.text}}",
    options: [
      { id: "yes", label: "Publish", port: "publish" },
      { id: "no", label: "Skip", port: "memo" },
    ],
    timeoutMs: 4 * 60 * 60 * 1000,
    timeoutPort: "memo",
  })
  .tool("publish", "typefully.create_draft", { content: "{{steps.draft.text}}" }, { next: null })
  .model("memo", { prompt: "Record that nothing was published.", next: null })
  .build();

The rules the step enforces:

• Exactly one question per run, whatever happens. The gate is keyed by the run and the step, so a resume — or a second replica picking the run up — finds the question that is already open instead of asking again.
• Only the listed options resolve it. "sure, go ahead" decides nothing: the question is re-posted and the run stays parked, because a guess here publishes something nobody chose.
• Only a human answers. The author is resolved server-side when the message is posted; an agent replying yes in its own thread leaves the gate open and lands on the audit trail (HumanGateUnresolved).
• Only the assignee, when the step names one. assignee is enforced, not decoration: another person's answer leaves the gate open and lands on the trail as assignee_mismatch, and the route that carried it answers 403 gate_assignee_mismatch rather than storing a reply that decides nothing. The question is deliberately not re-posted — it is still open and still theirs. Left empty, any human seat with access answers, which is what gates have always done. Assigning still grants nothing: it can only narrow who releases this run, never hand anyone access to the crew.
• A timeout fails closed. With a timeoutPort the run takes that branch; without one it stops. Nobody answering is never read as a yes. The deadline comes from timeoutMs (minimum 5 minutes) or LACREW_HUMAN_GATE_TTL_MS (default a day), and the sweep that expires it runs once a minute on the queue, so exactly one replica times each gate out.
• A cancelled run closes its gate. A late answer then lands on a closed question rather than restarting a run the operator ended.

A gate is control, not authority. It releases a pipeline the running principal was already allowed to execute — it approves no spend, changes no policy, and signs nothing onchain. A spend downstream of a gate still meets the policy stack and the escalation path exactly as it would have without one; if money moves, an onchain gate step and the Approvals path are still what decide it.

Open gates are readable at GET /flows/gates?status=pending (lacrew flows gates), and answered in the thread:

POST /messages {"thread":"agent:0x…","replyTo":"<questionId>","kind":"answer",
                "authorKind":"human","author":"Ada Lovelace","authorId":"seat_42",
                "body":"yes"}

authorId is the stable id of the seat the posting surface authenticated, when it has one — a hosted control plane passes its seat id here. An assigned gate is matched against it first and against author second, because a display name is renameable and a control that hinged on one would move with a rename. Both are set server-side by whatever posts the message, exactly like authorKind; nothing a message claims about itself is consulted. It is stored with the message, not only carried through the live post: the audit event names the seat that released a run, and a message that kept only the display name would stop agreeing with it the first time somebody is renamed.

What may go in assignee

One token: a seat id, a 0x address, or a single-word handle, optionally written seat:<id> (the spelling the dual-control reviewer spec uses). {{…}} is fine — it resolves per run.

Anything with a space in it is refused, both when the flow is saved and, for an interpolated value, when the gate is about to open (human_gate_invalid_assignee fails the step). This is a shape check, not a directory lookup: the runtime has no register of a workspace's people, so it cannot tell a live seat from a typo. What it can refuse is the input that is unmistakably not a reference to anybody — "whoever is on call", "ask the reviewer" — because a gate assigned to a sentence opens, looks decided, is answerable by nobody, and holds the run until the deadline fails it closed. A hosted control plane that does know its seats should offer them as a picker rather than a text box; validating a name against a directory is its job, not the runtime's.

There is deliberately no route that resolves a gate directly: a second way in would be one the conversation never gets to attribute to a seat. Every gate emits HumanGateOpened, then HumanGateResolved or HumanGateTimedOut, with the run and question ids — never the rendered prompt, which can name a private repo or a counterparty.

Code-first

import { flow, createFlowsClient } from "@lacrew/flows";

const budgetGuardedSpend = flow("budget-guarded-spend", "Budget-guarded spend")
  .gate("spend", {
    value: "75000000", // 75 USDC (6dp)
    onAllow: "confirm",
    onEscalate: "po-note",
  })
  .model("confirm", {
    prompt: "Spend allowed: {{steps.spend.json}}. Write a one-line receipt.",
    next: null,
  })
  .model("po-note", {
    prompt: "Spend escalated: {{steps.spend.json}}. Draft the purchase-order note.",
    next: null,
  })
  .build();

const flows = createFlowsClient({
  baseUrl: process.env.ORCH_URL ?? "http://127.0.0.1:8788",
  token: process.env.ORCH_TOKEN, // pairs with LACREW_ORCH_TOKEN
});

await flows.save(budgetGuardedSpend);
const run = await flows.run("budget-guarded-spend", { input: "manual run" });
console.log(
  run.status,
  run.steps.map((s) => s.summary),
);

runFlow(def, backend) executes a definition in-process against any FlowBackend; createMockFlowBackend() is the detached offline fallback the tests and demos use. Pass onStep to observe progress live:

await runFlow(def, backend, {
  input: "manual run",
  onStep: (t) => console.log(t.stepId, t.verdict ?? t.status, t.summary),
});

CLI

lacrew flows templates                  # built-in catalog (offline)
lacrew flows run treasury-pulse --local # offline mock run with live trace
lacrew flows run my-flow --input "hi"   # run on the orchestrator (ORCH_URL/ORCH_TOKEN)
lacrew flows save my-flow.json          # validate + persist
lacrew flows runs                       # recent traces, newest first
lacrew flows open                       # runs still going or parked on something
lacrew flows pause <runId>              # stop at the next step boundary
lacrew flows resume <runId>             # continue from the last checkpoint
lacrew flows cancel <runId>             # end it for good (never resumable)
lacrew flows code tpl-content-daily     # print the code-first snippet

Orchestrator HTTP surface

Every save and run lands in the audit trail as FlowSaved / FlowRun events, and a pause / resume / cancel as FlowRunLifecycle; trigger changes and accepted deliveries land as WebhookTriggerChanged / WebhookDelivery.

POST /hooks/:triggerId is the one route outside LACREW_ORCH_TOKEN, because its caller is an external producer holding the trigger's HMAC secret rather than the operator's bearer token. It is not an unauthenticated route — every request is verified against that signature and an unsigned one is refused.

LangChain

@lacrew/adapter-agents-langchain bridges both directions without a hard langchain dependency:

import {
  createLacrewLangChainTools,
  createLangChainFlowBackend,
} from "@lacrew/adapter-agents-langchain";

// LangChain agent → LaCrew: policy-checked tools for any LangChain agent.
// Each entry maps onto new DynamicStructuredTool({ name, description, schema, func: invoke }).
const tools = createLacrewLangChainTools({ backend: "http://127.0.0.1:8788" });

// LaCrew flow → LangChain: any runnable (chain, chat model, agent executor)
// becomes the model side of a FlowBackend; tool/gate steps stay policy-checked.
import { runFlow } from "@lacrew/flows";
const backend = createLangChainFlowBackend({ runnable: myChain });
const result = await runFlow(budgetGuardedSpend, backend);

Webhook triggers

A webhook flow is started by a signed HTTP delivery, so a crew can react to a PR opening or a post publishing instead of waiting for a human to press Run.

The declaration and the credential live in different places on purpose. The flow definition says "trigger": "webhook" — that is what makes it externally startable, and it travels with the definition through export and the marketplace. The trigger record — id, secret, principal, input mapping — is held by the orchestrator, because a secret inside a shareable definition would leak with every copy.

Registering a trigger

curl -X POST "$ORCH/flows/triggers" \
  -H "authorization: Bearer $LACREW_ORCH_TOKEN" \
  -H 'content-type: application/json' \
  -d '{
        "flowId": "pr-triage",
        "principal": "0xWorkerAgent",
        "scheme": "lacrew",
        "input": { "fields": { "pr": "pull_request.number", "title": "pull_request.title" } }
      }'

The response carries the secret once:

{
  "trigger": { "id": "wht_…", "secretVersion": 1 },
  "secret": "…",
  "secretShownOnce": true
}

There is no route that reads it back. Lose it and rotate (POST /flows/triggers/rotate), which invalidates the previous secret immediately.

Secrets are sealed at rest with the same AES-256-GCM envelope as session keys (LACREW_SESSION_KEY). With DATABASE_URL set and no sealing key configured, minting a trigger fails with webhook_sealing_unavailable rather than writing a cleartext secret to Postgres.

Event sources

scheme picks how a delivery proves it is genuine, which delivery it is, what happened, and where the payload lives. Providers differ on all four:

Signing a delivery

Two schemes, picked at registration:

The lacrew scheme signs the timestamp too, so a captured delivery cannot be replayed once it drifts past the tolerance window (300s; LACREW_WEBHOOK_TOLERANCE_SEC). The github scheme has no timestamp to sign, so replay protection there rests entirely on the X-GitHub-Delivery idempotency key — a property of the producer, not something the scheme asserts.

BODY='{"pull_request":{"number":7,"title":"Add hooks"}}'
TS=$(date +%s)
SIG=$(printf '%s.%s' "$TS" "$BODY" | openssl dgst -sha256 -hmac "$SECRET" -r | cut -d' ' -f1)

curl -X POST "$ORCH/hooks/$TRIGGER_ID" \
  -H 'content-type: application/json' \
  -H "X-Lacrew-Timestamp: $TS" \
  -H "X-Lacrew-Signature: sha256=$SIG" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d "$BODY"

A verified delivery answers 202 { "accepted": true, "runId": "run-wh-…" } and the run happens on a queue worker. The producer's socket is never what keeps a funded run alive — a flow that takes minutes of model time would otherwise be retried by every sane webhook sender while it was still working.

Filtering events

A trigger with no events runs on every delivery. Naming them subscribes:

{ "events": ["pull_request", "release.published"] }

Matching is by dotted prefix, so pull_request covers pull_request.opened without listing each action — but only in that direction: a filter for pull_request.opened is not satisfied by a bare pull_request.

An unsubscribed delivery answers 200 { "skipped": "event_not_selected" }, not a 4xx. GitHub disables a hook that keeps erroring, and "not interested" is not a delivery failure. The skip is recorded in the delivery log with the event type, so a quiet hook can be told apart from a broken one.

A delivery whose type the provider never declared always passes the filter: "I could not tell what this was" is not "you did not ask for this".

Wiring GitHub

Point a repository webhook (or a repository_dispatch) at $ORCH/hooks/<triggerId>, set its secret to the trigger's secret, choose application/json, and register the trigger with "scheme": "github". GitHub's X-GitHub-Delivery header is picked up as the idempotency key automatically, so its redeliveries answer 200 duplicate rather than starting a second run.

Google Pub/Sub push (Gmail, Calendar, Drive)

Pub/Sub push does not sign the body — it authenticates the sender, with a Google-signed OIDC token in Authorization: Bearer. There is no shared secret, so the trigger binds to what it will accept instead:

curl -X POST "$ORCH/flows/triggers" \
  -H "authorization: Bearer $LACREW_ORCH_TOKEN" \
  -H 'content-type: application/json' \
  -d '{
        "flowId": "inbox-triage",
        "scheme": "google-pubsub",
        "config": {
          "audience": "https://orch.example.com/hooks/PLACEHOLDER",
          "serviceAccountEmail": "pusher@my-project.iam.gserviceaccount.com"
        },
        "input": { "fields": { "mailbox": "emailAddress", "historyId": "historyId" } }
      }'

Both audience and serviceAccountEmail are required, and refused at creation rather than discovered on the first live event. This is the part that makes the source safe: anyone can have Google mint a valid token for their own service account and point their own subscription at your URL, so a signature proving only "Google signed this" authorizes nothing. The audience you configured plus the service account you expect are the binding.

The envelope is unwrapped before mapping. Pub/Sub delivers { "message": { "data": "<base64>", "messageId": "…" } } and the flow's input mapping sees the decoded message — mapping emailAddress against the envelope would read nothing and look like a typo rather than an unreachable path. messageId becomes the idempotency key.

Google's signing keys come from their JWKS endpoint and are cached. An unreachable key set answers 503 (retrying helps); a key absent from a current key set answers 401 (it will not).

Input mapping

The flow's input is a string. Without a mapping the whole body is passed through as JSON, which makes {{input}} and top-level {{input.action}} available. fields builds a flat object from dot paths — flat because {{input.x}} only reaches the top level, so a nested passthrough would render as an empty string and read like a missing field rather than an unreachable one. path lifts a single value:

{ "input": { "fields": { "pr": "pull_request.number" } } }
{ "input": { "path": "pull_request.title" } }

From the CLI

All of this is also lacrew flows triggers, so a self-hosted operator never has to hand-roll a signature:

lacrew flows triggers create --flow pr-triage --source github \
  --events pull_request --field pr=pull_request.number --field title=pull_request.title
lacrew flows triggers list
lacrew flows triggers curl <triggerId>       # a runnable signing example
lacrew flows triggers deliveries <triggerId>
lacrew flows triggers rotate <triggerId>

What a hook does not grant

A webhook decides who may start a flow. It never widens what the flow may do: the run executes as the trigger's principal down the same path as a manual run, so the principal's policy stack, its session key's onchain ceiling, and its pause state all still apply. A gate inside a webhook-started flow escalates to Approvals exactly as it would otherwise.

Failure modes

A paused principal is rejected rather than skipped: a webhook producer retries, and a silent skip would let a paused agent's events vanish behind a 2xx on every one of them.

Rejections are logged to the trigger's delivery log under their own key, never the producer's — a rejected delivery established no idempotency, and reusing its key would make the correctly-signed retry look like a replay of the failure and get dropped. The log records the reason code, the byte count and the run id; never the request body, which is attacker-supplied and routinely full of someone else's personal data.

Templates and the marketplace

flowTemplates ships first-party starters (treasury pulse, budget-guarded spend, escalation triage, content crew daily). The cloud's marketplace surface lists these today; third-party listings are mocked until the Phase 3 ecosystem work lands.

Scaffold a crew project

lacrew scaffold <template> turns any template into a standalone runnable project — package.json, crew.ts, the flow JSON, .env.example, README:

lacrew scaffold treasury-pulse --dir my-crew
cd my-crew && pnpm install && pnpm start   # offline mock run, full step trace
# set ORCH_URL (+ ORCH_TOKEN) to save + run on a live orchestrator instead

Inside a lacrew checkout the generated project links @lacrew/flows via file:; elsewhere it pins the npm name (publish pending).