Bicycle workflows: a guide for coding agents

Synced from bicycle-studio-api

Source: 05-workflows.md in the local directory context-collection (not yet on origin/platform), synced 2026-09-26. Do not edit this page here; change the source and run yarn sync:studio.

That source is itself generated (by arch/context-collection/build.py from bicycle-studio-api a78212f workflows/docs/WORKFLOWS.md via workflows/usage.py); change its source and re-run its build.

A workflow is a small graph of steps (nodes) that Studio runs in the background: on a schedule, when a person presses Run, or when an app, agent or MCP client calls it by ref. Steps read data, compute, ask a model, take a screenshot of an app and send the result. Each run keeps its outputs as versioned artifacts, and every send goes through a ledger (and, when the step asks for it, a person's approval).

Spec: bicycle.workflow/v1 (core workflow spec v6, runtime contract 1.5.0). The registry of node kinds is workflow_kinds (MCP) or GET /api/studio/v1/workflows/kinds (REST). REST paths below are under /api/studio/v1.

1. When to use a workflow

Use the least powerful thing that does the job.

You need to	Use	Not a workflow because
Show numbers, trends and breakdowns in an app	a semantic query in the app (declared query)	nothing runs later or sends
Compute one answer from one input, on demand (score, transform, look up)	a function (fn: ref, bda.fn.call)	one input, one output, no memory between calls
Investigate: pick the evidence, follow leads, explain why	an agent function (or bicycle:cause)	judgement over messy evidence; call it from a workflow step only when it has to run on a schedule
Email people a picture of an app view on a cadence	an app schedule (dataapp_schedule, or Subscribe in the app)	no steps: one saved view, delivered as is
Ad hoc "what changed, and why" on a panel	Detect and Explain (the app's analysis panel, chat, or analysis_run over MCP)	already built; the workflow detect / explain kinds validate but do not run yet
Several steps that run on a schedule, keep results between runs, or send something after a check	a workflow
Train a model on a cadence and use it later	a workflow with function steps and workflow blobs (section 7)
Change something in another system (email, Slack, a ticket)	a workflow send step, with approval where a person should look first	never an agent tool, never app code

A workflow is always asynchronous. A caller starts a run and follows it; the run outlives the caller.

2. Create one, change one

The loop (MCP; the REST route is in brackets)

1. workflow_kinds [GET /workflows/kinds]: the kinds, their slots and which ones this deployment runs.
2. workflow_create(title, document, files?, model?, timezone?) [POST /workflows]: always saved as revision 1, valid or not. model is the semantic model the queries block runs against. timezone (IANA) drives cron triggers and logical dates (empty = UTC).
3. workflow_get(workflow_id) [GET /workflows/{wf}]: the document, its files, its validation, and the draft revision number.
4. workflow_patch(workflow_id, expected_revision, ops, dry_run?) [POST /workflows/{wf}/patch]: JSON Patch ops plus typed ops (add_node, remove_node, rename, put_file, delete_file). Try dry_run: true first. The answer lists every error with a fix (one patch op you can send back as is) and the plan. workflow_put_file(workflow_id, expected_revision, path, content) [PUT /workflows/{wf}/files/{path}] writes one bundle file (sql/, prompts/, schemas/, data/, checks/).
5. workflow_validate [POST /workflows/{wf}/validate], then workflow_plan [POST /workflows/{wf}/plan]: what would run, what is cached, the estimated cost, the sends it would open, and the gated changes.
6. workflow_publish(workflow_id, revision) [POST /workflows/{wf}/publish]: manual and scheduled runs use the published revision.
7. workflow_run(workflow_id) [POST /workflows/{wf}/runs], then workflow_run_describe / workflow_runs [GET /workflows/{wf}/runs[/{run_id}]]; workflow_run_cancel [POST /workflows/{wf}/runs/{run_id}/cancel].

The working draft

There is one draft per workflow: the head of a chain of numbered revisions. Every save (patch or file) makes the next revision. Pass the draft number you read as expected_revision.

{"expected_revision": 7, "dry_run": true,
 "ops": [{"op": "replace", "path": "/nodes/write_digest/config/max_cost_usd", "value": 0.1}]}

If someone saved first you get 409 revision_conflict with details {current_revision, changed_since, edited_by, edited_by_name, edited_at}. Re-read with workflow_get, re-apply your ops on the new revision, and tell the person if their edit touched the same paths. Never retry blindly with the new number. (Functions call the same case draft_conflict.)

Gated changes need a person

workflow_publish refuses a revision with gated changes and lists them (409 human_publish_required or 403 gated_change_needs_person). Gated: a new or changed send step (its to, its approval), a new query, model or Use, anything upstream of a send that needs no approval, and AI-written text under approval: auto (ai_text_unreviewed). A person reviews and publishes those in Studio: send them to /{company}/apps/workflows/{workflow_id}. Do not try to get around a gate. Approving a send is never a tool.

Disable and enable (soft delete)

POST /workflows/{wf}:disable {"reason": "…"} and POST /workflows/{wf}:enable (the creator or a tenant admin; no MCP tool). Disable deletes the schedules, cancels pending approvals and queued runs (running ones finish), and new runs, patches and publishes answer 409 workflow_disabled. Apps and agents that pin it keep resolving the ref, but their calls are refused. Check GET /workflows/{wf}/dependents first and tell the person what uses it. There is no hard delete. POST /workflows/{wf}/unpublish takes it back to draft (schedules deleted, pending approvals cancelled).

Schedules and manual runs

A cron trigger is part of the document, so adding or changing one is a new revision and a publish:

"triggers": {
  "weekly": {"type": "cron", "cron": "0 8 * * 1", "targets": ["digest_receipt"]},
  "manual": {"type": "manual"}
}

• Cron fires in the workflow's zone (workflow_update(workflow_id, timezone="Asia/Kolkata")).
• Pause and resume without a publish: POST /workflows/{wf}/triggers/{trigger}:pause and …:resume.
• A manual run of the published revision is a real run: pointers move and sends go out (or to approval).
• A run of the draft ({"revision": <draft>}) is a try run: versions are written, nothing serves, and send steps only dry-run. Use try runs while you build.
• logical_date (YYYY-MM-DD) picks the period. A second run for the same date while one is active joins it.

Who may call it: PATCH /workflows/{wf} {"expose": {"apps": true, "agents": false, "mcp": false}}. Apps are on by default. Turning on agents or MCP takes a person in the Studio UI (403 person_required for a token or MCP caller).

3. Node kinds, one example each

Every node has kind, config, outputs (slot to artifact) and usually inputs (slot to artifact). when is a CEL guard (inputs.orders.rows > 0). Templates in params take ${run.logical_date}, with an offset such as ${run.logical_date - P7D}, and ${partition.<dim>}. Nothing else, and never the clock.

query: one declared query of the document's queries block, run against the workflow's model.

"fetch": {"kind": "query",
  "config": {"query": "failed_orders", "params": {"from": "${run.logical_date - P7D}", "to": "${run.logical_date + P1D}"}},
  "outputs": {"rows": "failed"}}

sql: DuckDB over the node's input tables (each input slot is a table name), in a locked connection: SELECT only, no clock, no files, no network. Pass dates in as params ($as_of).

"summarise": {"kind": "sql", "config": {"file": "sql/summary.sql", "params": {"as_of": "${run.logical_date}"}},
  "inputs": {"failed": "failed"}, "outputs": {"summary": "summary"}}

function: a published code (or llm, classify, agent) function, pinned fn:{tenant}/{name}@{n}. It must be shared with the tenant and exposed to workflows. It gets {inputs, params, run} and runs as the tenant's service identity through the functions broker. Output slot value is its whole answer (a json artifact); any other slot takes that key of the answer (a list of objects, for a table artifact). volatile: true re-runs it even when nothing changed.

"score": {"kind": "function",
  "config": {"ref": "fn:<tenant>/score_orders@3", "params": {"threshold": 0.8}},
  "inputs": {"failed": "failed"}, "outputs": {"value": "scores"}}

llm: one model call per input (or per rows_per_call rows), with a strict output schema and one repair. prompts/*.md templates read {{ inputs.<slot>.rows | tojson }} and {{ run.logical_date }}. builtin:message gives {subject, body} for a send. The output is untrusted text.

"write": {"kind": "llm",
  "config": {"prompt": "prompts/digest.md", "model": "claude-sonnet-5", "effort": "low", "max_output_tokens": 800,
             "max_cost_usd": 0.05, "output": {"schema": "builtin:message"}},
  "inputs": {"summary": "summary"}, "outputs": {"digest": "digest"}}

classify: sort rows into a fixed label set (typesafe.ai Jev), batched and consistent. Rows below threshold stay unlabelled for a person. The output is untrusted.

"triage": {"kind": "classify",
  "config": {"fields": ["reason"], "instructions": "Which team should fix this failed order?",
             "labels": {"payments": "Card declines, payment gateway errors", "stock": "Out of stock, allocation",
                        "other": "Anything else"},
             "threshold": 0.6, "keep": ["order_id"]},
  "inputs": {"rows": "failed"}, "outputs": {"rows": "triaged"}}

agent: there is no separate kind. Call an agent-kind function from a function node: {"kind": "function", "config": {"ref": "fn:<tenant>/investigate_drop@2"}}. It runs with the budgets and connections the agent function declares, and its trace is a child of the run. It is slow and costs the most, so guard it with when and use it only where judgement is the job.

snapshot: a PNG of one tab of a data app (tabs: ["default"] for an app without tabs), with filters as the link state's filters. It needs at least one input, which only orders it after the data it should show. The app must be shared with the whole workspace: the capture runs as the tenant's service identity, which holds no grant of its own. Validate and plan warn snapshot_app_not_shared on /nodes/<n>/config/app for any other app, and a run fails that step with the same code before any screenshot is taken. A page that renders but never loads (an error page, an error boundary, a blank page) fails the step as snapshot_render_failed.

A failed screenshot sends nothing. Whatever the reason (snapshot_app_not_shared, snapshot_render_failed, snapshot_timeout, snapshot_failed, …), the send that reads it is skipped (snapshot_failed): with auto or manual approval, with a required or an optional screenshot input, on the first run or later. It never falls back to the last good screenshot, it never goes out without the picture, and no approval request is opened. The run is failed or partial, and only the failure notice goes out, to the author by rule for a known code (Jev decides an unclear one). A snapshot guarded off with when is not a failure: the send goes without a picture.

"snap": {"kind": "snapshot", "config": {"app": "<app id>", "tabs": ["default"], "formats": ["png"]},
  "inputs": {"summary": "summary"}, "outputs": {"report": "screenshot"}}

action (Send): sends the payload input (a message) once per idempotency_key. Every other input is context: a report is attached and shown inline, a table is shown under the message (20 rows at most).

"send": {"kind": "action",
  "config": {"to": {"channel": "email", "recipients": ["<email>"]},
             "payload": "digest", "idempotency_key": "failed-orders-${run.logical_date}", "max_sends_per_run": 1,
             "approval": {"mode": "manual", "approvers": {"roles": ["workflow_owner"]}, "timeout": "P1D",
                          "on_timeout": "reject", "review": ["summary"]}},
  "inputs": {"digest": "digest", "summary": "summary", "screenshot": "screenshot"},
  "outputs": {"receipt": "digest_receipt"}}

• Where it goes. Put email addresses on the step, to: {channel: "email", recipients: [1..50]}, or run one of the tenant's actions (Slack, PagerDuty, Jira, webhook), to: {channel: "action", action_id, variables?} (workflow_actions_list shows each action's required variables; bind them as literals or ${payload.value.<field>}). destination: "dst_…" is a deprecated alias: do not add new ones.
• Approval. {"mode": "manual", …}: a person approves in the Studio inbox, bound to the exact payload. {"mode": "auto"}: no approval, it sends as the step runs (never on a try run). AI-written input under auto is the warning ai_text_unreviewed and a gated change a person acknowledges at publish. conditional runs as manual.
• Tenant action policy (GET /workflow-action-policy): an action send runs without a person (auto) only when a tenant admin marked the action auto_allowed AND the deployment lets it; otherwise auto is the save error action_requires_approval (its fix switches the step to manual approval), and a send is refused again at send time. The default is: every action send needs a person. workflow_actions_list shows auto_allowed per action.
• Tenant mail policy (GET /workflow-mail-policy): addresses outside the tenant's own and allowed domains are refused (recipient_domain_not_allowed), need a manual step (external_recipient_needs_approval, the default), or are allowed with a warning. Only a tenant admin changes the policy.
• Preview allowlist. On preview, a workflow mails only the addresses on the deployment allowlist (BSA_WORKFLOW_MAIL_ALLOWLIST). Any other address is an error at save (recipient_not_allowed) and is withheld at send. Never put an address outside the allowlist on a preview workflow, and never try to work around it.
• Per row. per: "row" (one send per payload row, key with ${row.<col>}) validates, but this build refuses it at plan (per_row_not_supported). Send one message that lists the rows instead.
• Idempotency. The key allows only ${run.logical_date}, ${partition.<dim>} (and ${row.<col>}). A key that was already sent is never sent again: a re-run for the same date joins it.
• Caps. max_sends_per_run (default 1, up to 500) fails the step with send_cap_run. A tenant sends to at most 500 recipients per UTC day (send_cap_daily).
• Outcome notifications are record settings, not document fields: PATCH /workflows/{wf} {"notify": {"on_success": "approve" | "send", "recipients": {"author": "…", "data_owner": "…", "bicycle_ops": "…"}, "routes": {"config": "author", "data": "data_owner", "platform": "bicycle_ops"}}}. on_success: send (a person sets it) sends a manual step without asking when every step upstream succeeded. A failed or partial production run mails one person: known failure classes route by rule, anything else is routed by Jev among the role names (low confidence goes to the author). Each decision is on the run (run.notify) and in its trace.

Kinds that validate and plan but do not run here: python (write a code function and call it), input, source, detect, explain.

4. What persists

What	Where	Survives between runs
Artifacts	every node output is a version (av_…); a production run that passes its checks moves the serving pointer. retention.keep_versions defaults to 5	yes. GET /workflows/{wf}/artifacts lists them, …/artifacts/{a}/versions/{v} has the rows, …/blob a snapshot's PNG
Cache	the fingerprint of a node's inputs, config, files and pinned refs. Classes: external (query: always runs; what depends on it is cached if the answer did not change), cacheable (sql, llm, function, classify: same fingerprint = cache hit, no cost), never (snapshot, action)	yes
Workflow blobs	files a function writes at the workflow's own address {tenant}/workflows/{wf}/store/{name}. Declare them in blobs; the function asks for blob.read / blob.write on those names	yes, until overwritten. Blobs are not versioned for you: put the version in the name (model-2026-09-26.skops) and keep a small pointer file (model-latest.json)
Send ledger	one entry per rendered idempotency_key: a key that was sent is never sent twice	yes
Run history	each run is an invocation: a root inv_… (kind: workflow), one child per node partition, and a function step's own invocation under its child, with the trace	yes. GET /invocations/{id}, …/events?after=&wait_s=, …/trace, GET /invocations?kind=workflow
Record settings	expose, notify, timezone, paused triggers	yes, with no new revision

What does not survive: a function's memory, temp files and globals (each call is a fresh sandbox), try-run outputs (they never serve), a failed version (the last good one keeps serving; held when a check fails), and anything a step returns but no output slot keeps.

5. Call it from an app, an agent, MCP or REST

A published workflow is callable by its pinned ref wf:{tenant}/{slug}@{n} (the published revision; @latest only for interactive callers, never in a manifest). The input is {logical_date?, targets?}, and the call is always a production run as the caller.

Data app. Declare it in bda.manifest.json. Adding or re-pinning a workflow import is reviewed by a person when the app version is published.

{
  "functions": {
    "my_workflow": {
      "ref": "wf:<tenant>/<slug>@<n>",
      "description": "What it does, in one line"
    }
  }
}

Start it, show progress, allow Cancel (the kit's bda.fn, src/studio/fn.ts):

// A workflow always runs in the background: start it, follow it, allow Cancel.
const started = await bda.fn.call("my_workflow", {}, { wait: false });
// input (optional): { logical_date: "YYYY-MM-DD", targets: ["<artifact>"] }
cancelButton.onclick = () => bda.fn.cancel(started.invocation_id);
const watch = bda.fn.watch(started.invocation_id, batch => showProgress(batch.items));
// batch.items: plain-word lines, e.g. "fetch succeeded · 812 rows · 1.2 s"
const final = await watch.done;     // the root invocation: status, output, error, usage
const out = bda.fn.outputOf(final) as {
  status: string; partial?: boolean; failed?: string[];
  artifacts: Record<string, { version_id: string; rows: number | null; data?: unknown }>;
  receipts: { node: string; state: string; approval_id?: string | null }[];
};                                  // outputOf throws a BdaError in words if the run failed or was cancelled
if (out.partial) showNote(`Some steps did not finish: ${(out.failed ?? []).join(", ")}`);

What comes back (outputOf(final)):

{
  "run_id": "run_…",
  "status": "succeeded",
  "logical_date": "2026-09-21",
  "artifacts": {
    "<artifact>": {"version_id": "av_…", "rows": 8, "published": true, "data": {"columns": [{"name": "…", "type": "…"}], "rows": [["…"]]}}
  },
  "receipts": [
    {"node": "<send step>", "destination": "email:…", "state": "awaiting_approval", "approval_id": "apr_…", "idempotency_key": "…"}
  ]
}
// "status": "partial" adds "partial": true and "failed": [steps] (the invocation itself says succeeded).
// artifacts[name].data is inlined for the run's targets while the output stays under 1 MiB; read the rest by
// version_id (GET /api/studio/v1/workflows/{wf}/artifacts/{name}/versions/{version_id}).
// A receipt's state follows the send: awaiting_approval, then sent (or rejected, expired, failed).

Rules for app code: call only the declared local name through bda.fn; start it from a button's onClick (the sandbox has no forms); render the output as text, never as HTML; reuse does not apply to workflow imports; a receipt is not delivered until its state is "sent".

Agent. Add wf:<tenant>/<slug>@<n> to the agent's functions (grants.functions) and switch on "Chat and agents" (expose.agents; a person does that in Studio). The agent then has a tool wf_<slug> that starts a run and streams its progress. A data app's chat lists the app's workflow imports as fn_<local>. An agent or app can start a workflow, never approve its sends.

MCP. With expose.mcp on, the workflow is a tool named wf_<slug> on /mcp/functions and /mcp; it starts a run and streams its progress. To build or edit workflows use the workflow_* tools above.

REST (a script, a backend):

# $STUDIO = https://app.bicycle.ai (preview: https://preview.bicycle.ai); $TOKEN = a Bicycle API token
curl -s -X POST "$STUDIO/api/studio/v1/functions/wf:<tenant>/<slug>@<n>:submit" \
  -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{"input": {"logical_date": "2026-09-21"}}'
# 202 {"invocation_id": "inv_…", "kind": "workflow", "status": "queued", ...} (the run's root invocation)
curl -s "$STUDIO/api/studio/v1/invocations/inv_…?wait_s=25" -H "Authorization: Bearer $TOKEN"    # poll: status, output
curl -s "$STUDIO/api/studio/v1/invocations/inv_…/events?after=0&wait_s=25" -H "Authorization: Bearer $TOKEN"
curl -s -X POST "$STUDIO/api/studio/v1/invocations/inv_…:cancel" -H "Authorization: Bearer $TOKEN"

:invoke answers 422 kind_requires_async. 404 function_not_found means an unknown or unpublished ref, or a surface it is not exposed to. 409 workflow_disabled means it is disabled.

6. Limits and safety

• Sends: at most 50 recipients a step, max_sends_per_run up to 500, 500 recipients per tenant per UTC day, and an approval waits at most 14 days (timeout).
• Cost: max_cost_usd on an llm step is at most $25, and the plan refuses a run estimated over $25 (cost_cap).
• Size: workflow.json up to 256 KiB, a bundle file up to 1 MiB, a json artifact up to 1 MiB, a blob up to 100 MiB (32 declared), a call's inline output data up to 1 MiB in total (the rest by version_id), at most 32 targets.
• Time: a function step defaults to 15 minutes, a snapshot to 240 s.
• People: approvers are roles (workflow_owner, workflow_editor, app_owner, app_editor, tenant_admin) or users (usr_…); groups alone are refused. Gated publishes, approvals, on_success: send, and exposing to agents or MCP always take a person.
• Preview: never email an address outside the deployment allowlist, never test a tenant action that would page or ticket real people, and name test workflows so they are easy to find and disable.

7. Worked examples

Train, then predict (ML with workflow blobs)

train fits a model and saves model-<date>.skops plus a model-latest.json pointer ({blob, sha256, features, metrics, logical_date}), or reuses the current model while it is recent. predict reads the pointer, checks the model's sha256 against it, loads it with skops and an explicit type allowlist (never pickle), and scores rows. Train on a weekly cron and predict daily, or run both in one workflow as here.

{
  "schema": "bicycle.workflow/v1",
  "title": "Train weekly, predict from the latest model",
  "blobs": [
    {"name": "model-*", "kind": "binary", "max_bytes": 20971520, "purpose": "one trained model per logical date"},
    {"name": "model-latest.json", "kind": "json", "max_bytes": 65536, "purpose": "which model is current"}
  ],
  "artifacts": {
    "trained": {"type": "json", "schema": "schemas/trained.json"},
    "predictions": {"type": "table"},
    "prediction_summary": {"type": "json", "schema": "schemas/prediction_summary.json"}
  },
  "nodes": {
    "train": {"kind": "function",
      "config": {"ref": "fn:<tenant>/ml_train@1",
                 "params": {"model_blob": "model-${run.logical_date}.skops", "latest_blob": "model-latest.json",
                            "max_age_days": 7}},
      "outputs": {"value": "trained"}},
    "predict": {"kind": "function",
      "config": {"ref": "fn:<tenant>/ml_predict@1", "params": {"latest_blob": "model-latest.json"}},
      "inputs": {"trained": "trained"},
      "outputs": {"predictions": "predictions", "value": "prediction_summary"}}
  },
  "triggers": {"daily": {"type": "cron", "cron": "0 6 * * *", "targets": ["prediction_summary"]},
               "manual": {"type": "manual"}}
}

{"type": "object", "required": ["blob", "sha256"],
 "properties": {"blob": {"type": "string"}, "sha256": {"type": "string"}, "logical_date": {"type": "string"}}}

{"type": "object", "properties": {"n": {"type": "integer"}, "model_blob": {"type": "string"}, "sha256": {"type": "string"}}}

The two functions (bda-python:3, resources.class: fn-small for training). Each declares its blob capabilities in function.json: train blob.write ["model-*", "model-latest.json"] and blob.read ["model-latest.json"], predict blob.read ["model-*", "model-latest.json"].

# train (main.py): reuse a recent model, else fit and save a new one by date
def handler(input, ctx):
    params, run = input["params"], input["run"]
    raw = ctx.blob.get(params["latest_blob"])
    if raw and days_between(run["logical_date"], json.loads(raw)["logical_date"]) < params["max_age_days"]:
        return {**json.loads(raw), "reused": True}
    model = fit(...)                                   # scikit-learn
    data = skops.io.dumps(model)
    latest = {"blob": params["model_blob"], "sha256": hashlib.sha256(data).hexdigest(),
              "logical_date": run["logical_date"], "features": FEATURES}
    ctx.blob.put(params["model_blob"], data, "application/octet-stream")
    ctx.blob.put(params["latest_blob"], json.dumps(latest), "application/json")
    return latest

# predict (main.py): load only what the pointer names, after the sha256 matches
def handler(input, ctx):
    latest = json.loads(ctx.blob.get(input["params"]["latest_blob"]))
    data = ctx.blob.get(latest["blob"])
    if hashlib.sha256(data).hexdigest() != latest["sha256"]:
        raise ValueError("the model does not match model-latest.json")
    if set(skops.io.get_untrusted_types(data=data)) - set(TRUSTED):
        raise ValueError("the model holds types outside the allowlist")
    model = skops.io.loads(data, trusted=TRUSTED)
    return {"predictions": score(model), "model_blob": latest["blob"], "sha256": latest["sha256"]}

Weekly failed-orders digest with a screenshot

A query, a sql summary, an llm digest, a screenshot of the team's app, and an email a person approves. A week with no failures sends nothing (the when guards). It runs Monday at 08:00 in the workflow's zone.

{
  "schema": "bicycle.workflow/v1",
  "title": "Weekly failed orders digest",
  "queries": {
    "failed_orders": {
      "sql": "SELECT day, reason, failed_orders FROM orders WHERE day >= $from AND day < $to ORDER BY day",
      "parameters": [{"name": "from", "type": "date"}, {"name": "to", "type": "date"}],
      "columns": [{"name": "day", "type": "date"}, {"name": "reason", "type": "string"},
                  {"name": "failed_orders", "type": "number"}],
      "maxLimit": 5000
    }
  },
  "artifacts": {
    "failed": {"type": "table"},
    "summary": {"type": "table"},
    "digest": {"type": "message", "channel": "email"},
    "screenshot": {"type": "report"},
    "digest_receipt": {"type": "receipt"}
  },
  "nodes": {
    "fetch": {"kind": "query",
      "config": {"query": "failed_orders",
                 "params": {"from": "${run.logical_date - P7D}", "to": "${run.logical_date + P1D}"}},
      "outputs": {"rows": "failed"}},
    "summarise": {"kind": "sql", "config": {"file": "sql/summary.sql"},
      "inputs": {"failed": "failed"}, "outputs": {"summary": "summary"}},
    "write": {"kind": "llm", "when": "inputs.summary.rows > 0",
      "config": {"prompt": "prompts/digest.md", "effort": "low", "max_output_tokens": 800, "max_cost_usd": 0.05,
                 "output": {"schema": "builtin:message"}},
      "inputs": {"summary": "summary"}, "outputs": {"digest": "digest"}},
    "snap": {"kind": "snapshot", "when": "inputs.summary.rows > 0",
      "config": {"app": "<app id shared with the workspace>", "tabs": ["default"]},
      "inputs": {"summary": "summary"}, "outputs": {"report": "screenshot"}},
    "send": {"kind": "action", "when": "inputs.summary.rows > 0",
      "config": {"to": {"channel": "email", "recipients": ["<email>"]},
                 "payload": "digest", "idempotency_key": "failed-orders-${run.logical_date}",
                 "approval": {"mode": "manual", "approvers": {"roles": ["workflow_owner"]}, "timeout": "P1D",
                              "on_timeout": "reject", "review": ["summary"]}},
      "inputs": {"digest": "digest", "summary": "summary", "screenshot": "screenshot"},
      "outputs": {"receipt": "digest_receipt"}}
  },
  "triggers": {"weekly": {"type": "cron", "cron": "0 8 * * 1", "targets": ["digest_receipt"]},
               "manual": {"type": "manual"}}
}

SELECT reason, sum(failed_orders) AS failed_orders
FROM failed
GROUP BY reason
ORDER BY failed_orders DESC

Write a short weekly digest of failed orders for the operations team, for the week ending {{ run.logical_date }}.

{{ inputs.summary.rows | tojson(indent=2) }}

- `subject`: under 80 characters, naming the biggest reason.
- `body`: Markdown, at most five bullets. Quote the numbers exactly; do not invent causes.

To build it: workflow_create with the document, the two files and your model, then workflow_plan, then a try run (workflow_run with the draft revision; the send only dry-runs), then ask a person to publish (the send step is a gated change). Replace the recipient with a real address the tenant's mail policy allows (on preview, one on the allowlist), and the query with one of your model's metrics (query_describe_model).