# cereyan > Minimal, local-first orchestrator for Python data pipelines Cereyan is a minimal, local-first orchestrator for Python data pipelines: a Rust core (SQLite store, state machine, scheduler, HTTP server) behind a thin layer of Python decorators, shipped as one wheel with no runtime dependencies. Flows and tasks run offline as plain scripts and are recorded locally; `cereyan serve` adds the API, the UI, schedules, engines, rules, and a built-in MCP server on top of the same store. # Home # cereyan Cereyan is a minimal, local-first orchestrator for Python data pipelines. A Rust core (SQLite store, state machine, scheduler, HTTP server) sits behind a thin layer of Python decorators, and the whole thing ships as one wheel with no runtime dependencies. ``` from datetime import date from cereyan import flow, task @task def extract(day: date) -> list[int]: return [1, 2, 3] @flow(run_name="etl-{day}") def etl(day: date) -> int: return sum(extract(day)) assert etl(date(2026, 9, 6)) == 6 # recorded as a run in ~/.cereyan/db.sqlite ``` - **Offline first.** `python pipeline.py` records runs into a local SQLite file. Nothing else needs to run. - **One process to serve.** `cereyan serve dir/` hosts the API, the UI, the scheduler, a warm pool of engine processes, and a built-in MCP server for agents. - **Data-pipeline semantics.** Targets make reruns idempotent, backfills cover date ranges, resources are named semaphores, flows chain and fan in by key, and rules react to events or to their absence. ## Where to go - **[Quickstart](https://sercanatalik.github.io/cereyan/get-started/quickstart/index.md)** From `pip install` to a scheduled, retried, backfilled pipeline in ten minutes. - **[Concepts](https://sercanatalik.github.io/cereyan/concepts/app-and-projects/index.md)** The model: App, flow, task, run, state, schedule, target, resource, backfill, artifact, variable, event, rule. - **[Guides](https://sercanatalik.github.io/cereyan/guides/retries-timeouts-crashes/index.md)** One goal per page, from retries to running the server as a service. - **[Reference](https://sercanatalik.github.io/cereyan/reference/python-api/index.md)** Every option, command, route, event, state, and configuration key. Agents can read the whole site as one file: [llms.txt](https://sercanatalik.github.io/cereyan/llms.txt) is the index and [llms-full.txt](https://sercanatalik.github.io/cereyan/llms-full.txt) is the full text. # Get started # Install ``` pip install cereyan ``` Cereyan needs Python 3.11 or newer. The wheel is self-contained: it has no runtime dependencies and includes the Rust core and the web UI. Wheels are published for macOS (Apple silicon and Intel), Linux (x86_64 and aarch64, manylinux 2.28), and Windows (x86_64). ``` pip install cereyan ``` ``` uv add cereyan ``` Check the install: ``` cereyan --help python -c "import cereyan; print(cereyan.__version__)" ``` ## What gets created The first run creates the runtime home, `~/.cereyan` by default, holding the SQLite database and, later, `server.json` while a server runs, `secret.key` once a secret variable exists, and `storage/` for persisted results. Set `CEREYAN_HOME` or pass `--home` to put it elsewhere; see [Engines and the home directory](https://sercanatalik.github.io/cereyan/concepts/engines-and-home/index.md). ## Windows The Unix socket listener and engine niceness are Unix-only options, rejected or ignored with a message. Two behaviours are weaker on Windows, both of them signal-based: | Behaviour | On Windows | | ------------------------------------------------------------------------ | --------------------------------------------------------------------------------------------------- | | A flow's `timeout_seconds` offline (`python pipeline.py`, `cereyan run`) | Accepted and ignored; the flow runs unbounded. A served run is timed out by the server. | | Cancelling a flow blocked in a sleep or an I/O call | The engine is ended rather than interrupted inside the flow, so `on_cancellation` hooks do not run. | Everything else — the server, the UI, schedules, engines, rules, artifacts, variables, and the MCP server — works the same. Next: the [Quickstart](https://sercanatalik.github.io/cereyan/get-started/quickstart/index.md). # Quickstart In ten minutes you will write a pipeline, run it offline, serve it with a live UI, schedule it, make it retry, and backfill a date range. ## 1. Write a pipeline Decorate plain functions. Parameters come from the type hints and are coerced before the flow runs. ``` # pipeline.py from datetime import date from cereyan import flow, task, get_run_logger @task def extract(day: date) -> list[int]: get_run_logger().info("extracting %s", day) return [1, 2, 3] @task def load(rows: list[int]) -> int: return sum(rows) @flow(run_name="etl-{day}") def etl(day: date = date(2026, 9, 6)) -> int: return load(extract(day)) if __name__ == "__main__": print(etl()) ``` A flow is still a function. Calling it runs the tasks in order, records a run with its task runs and logs, and returns the result: ``` assert etl(day=date(2026, 1, 2)) == 6 ``` ## 2. Run it offline ``` python pipeline.py # records a run in ~/.cereyan/db.sqlite cereyan run pipeline.py:etl --param day=2026-01-02 # same, with parameters and a summary cereyan runs ls --state Failed --json # inspect history ``` Nothing else is running. Every run lands in the local SQLite store, and `cereyan run` exits 0 on success, 1 on failure, 2 when nothing ran, and 3 on a loading or parameter error. ## 3. Serve it ``` cereyan serve . # imports every module under the directory, opens http://127.0.0.1:4200 ``` One process now hosts the HTTP API, the web UI, the scheduler, and a warm pool of engine processes that execute runs. The dashboard shows runs and logs live. Scripts and `cereyan run` keep working while the server is up: they hand their run to the server and stream its logs back. ## 4. Schedule, retry, and write idempotent outputs ``` from datetime import date from cereyan import flow, task, Cron, LocalTarget, exponential @task(output=lambda day: LocalTarget(f"out/{day}.parquet"), retries=2, retry_delay=exponential(1)) def build(day: date): with LocalTarget(f"out/{day}.parquet").open("w") as fh: fh.write("...") @flow(schedule=Cron("0 9 * * *", timezone="Europe/Istanbul"), max_concurrent=1) def daily(day: date = date(2026, 9, 6)): build(day) ``` - `schedule=` fires the flow every morning at nine in Istanbul once a server is running. - `retries=2` with an exponential delay reruns `build` when it raises. - `output=` names a target. When the file exists the task is skipped, so rerunning a day is safe. - `max_concurrent=1` keeps two runs of `daily` from overlapping. Run it twice: the second call skips `build` because its target exists. ``` daily(date(2026, 1, 1)) daily(date(2026, 1, 1)) # build is Skipped: out/2026-01-01.parquet exists ``` ## 5. Backfill a date range With the server running: ``` cereyan backfill daily --param day --start 2026-06-01 --end 2026-08-30 --concurrency 4 ``` This creates one run per day, tagged `backfill:`, with at most four executing at once. The Backfill dialog on the flow page does the same. ## 6. React to events Rules run actions when events happen. A code rule is a decorated function: ``` from cereyan import App, emit_event app = App("warehouse") @app.rule(on="run.failed", flow="check_orders") def page_someone(event, run): print("failed:", run["name"]) @app.flow def check_orders(): emit_event("orders.table_empty", {"table": "orders"}) check_orders() ``` Rules can also be created in the UI with webhook, email, run-flow, cancel, set-state, and schedule actions, and a rule can fire when an expected event does *not* happen. ## Next - Take the [tour of the UI](https://sercanatalik.github.io/cereyan/get-started/tour/index.md). - Read the [concepts](https://sercanatalik.github.io/cereyan/concepts/app-and-projects/index.md) for the model behind what you just did. - Pick a guide: [retries and timeouts](https://sercanatalik.github.io/cereyan/guides/retries-timeouts-crashes/index.md), [schedules](https://sercanatalik.github.io/cereyan/guides/schedule-a-flow/index.md), [backfills](https://sercanatalik.github.io/cereyan/guides/backfill/index.md), [rules](https://sercanatalik.github.io/cereyan/guides/rules/index.md), or [using cereyan with an AI agent](https://sercanatalik.github.io/cereyan/guides/agents/index.md). # Tour of the UI `cereyan serve` opens the UI at `http://127.0.0.1:4200`. It is built for a desktop browser and updates live from the server's event stream. One top bar carries everything that is not a page: the eight sections (Dashboard, Runs, Flows, Events, Artifacts, Rules, Variables, and Settings), a **Project** switcher that scopes every list to one project, a search box (or **⌘K**, **Ctrl+K** elsewhere) that jumps to a section, a flow, a run by name, or an artifact by key, the connection indicator that reads **live** while the stream is connected, and the theme toggle. The palette is a warm neutral in light and dark. Ink is the only brand colour; every other colour on a page belongs to a run state, so a glance tells you what is running, failed, waiting, or late. ## Dashboard Counts for the selected range (Running, Completed, Failed, Crashed, Waiting for input, Late, Scheduled) with a proportion bar and an hourly histogram by state. **Needs attention** lists the runs waiting on you: paused runs with their question and an **Answer** button, failed runs with **Run again**, crashed and late runs with **Open**. **Running now** shows each active run with its elapsed time and how many of its tasks are done. **Upcoming** lists the next scheduled runs with a **Run now** shortcut. The range selector and the tag filter apply to the whole page. ## Runs Every run, newest first, in collapsible groups by the flow's group. Filters are popover buttons for state, project, flow, tags, and range, plus a name search and a sort. The **Tasks** column is a bar of the run's task runs by state. The **Task runs** tab lists task runs across runs the same way. Selecting rows raises a bar at the bottom of the window with **Cancel** and **Delete** for the selection; a selection may span groups. ## Run detail The header band shows the run's name, state, flow, tags, and a line with its start, elapsed or total time, attempt, what created it, and its parameters. **Run again** and **Cancel** sit on the right; **Delete** is in the overflow menu. A paused run shows its question and the **Resume** form in the band. The **tasks rail** on the left lists every task run with its state, duration, and, for a task waiting to retry, the attempt and a countdown. Click a task to focus it: the **Logs** tab then shows only that task run's lines, with a chip you can clear. A task run also has a page of its own, opened from the **Task runs** tab of the runs page or from a bar on the Timeline, carrying its logs, artifacts, and details. The tabs on the right: - **Logs**: log lines with a level filter, a search box, and **Follow** to keep the newest line in view while the run executes. - **Timeline**: the task runs on a time axis, and a **dependency** view of the same graph. - **Artifacts**: markdown, tables, progress bars, links, and images the run published. - **Parameters**: the values the run was called with. - **Details**: ids, timing, what created the run, its scheduled time, priority, attempt, the previous attempt, failure and crash counts, and the engine PID. ## Flows Every flow the server has registered, in collapsible groups. A flow's group is the one it declared with `group=`, else its project; a group of one project shows that project's source directory, and a group spanning several names them. Each row shows the schedule in words with the next fire time, the last ten runs as bars coloured by state and sized by duration, the last run's state, and tags. **Run** opens a form built from the flow's parameter schema. For a scheduled flow the row menu adds **Skip next run** (with the time it skips), **Skip runs…**, and **Reschedule…**, and the schedule cell counts skipped fires beside the next one. The **Dependencies** panel above the table draws each `after=` chain as nodes joined by arrows and each fan-in as its upstreams joined into the downstream flow with its key. A flow the running server did not register stays listed, dimmed, with its last-seen time and a **Delete** action. A group header is the same columns rolled up, so collapsing a group hides the detail without hiding what it says: the soonest next fire, the group's recent runs, its last-run states as a bar with counts, the union of its tags, and how many flows it holds. Flows the running server no longer has registered are counted separately as **stale**, so an all-green bar cannot hide them. A group of more than five rows starts collapsed and everything else starts open; a lone group is always open, and a search opens every group it matches. ## Flow detail The flow's description (rendered from its docstring), its schedule summary with the next fire that will run and how many are skipped, chips for priority, concurrency cap, and overlap policy, the last ten runs as dots, and **Run**, **Backfill**, **Skip next…**, and **Pause** actions, with **Reschedule** beside the schedule summary. Tabs list the runs; the upcoming runs, each with how far off it is and **Skip** or **Undo**, a skipped one saying who skipped it and when, several skippable at once or all from the header checkbox, and the fires past the look-ahead listed as projected, ten more at a time; the schedules with an editor and a preview of upcoming fire times; and the parameter schema. ## Events The event feed, newest first, filtered by name or prefix such as `run.*`, by resource kind, by flow, and by time range. Open an event to see its payload and related resources. New events appear as they happen. ## Artifacts Artifacts across all runs, filtered by kind, key, flow, and project. Open a key to see its history: every value published under that key across runs, newest first. ## Rules Every rule with its enabled switch, match clause, actions, last firing, and fire count. **New rule** opens the form: events, flows, tags, states, and project to match; the ordered actions; guards; and an **Unless** section for proactive rules. Code rules declared with `@app.rule` appear read-only with a `code` chip. Each rule's page lists its firings and open expectations and has a **Test** button that renders its templates against the most recent matching event without executing anything. ## Variables Named JSON values with tags. Secrets are stored encrypted and shown masked. Values are shared by every project on the machine. ## Settings The server's version, URL, PID, home, served directory, and start time; the database path, size, WAL size, and retention; resource totals you can add and edit; the retention and crash-retry defaults; the custom routes registered by the served Apps; and the engine pool with each engine's PID, module, runs done, and current run. Saving settings writes them back to `cereyan.toml`. Next: the [concepts](https://sercanatalik.github.io/cereyan/concepts/app-and-projects/index.md), or straight to the [guides](https://sercanatalik.github.io/cereyan/guides/retries-timeouts-crashes/index.md). # Concepts # App and projects ``` from cereyan import App app = App("warehouse") @app.flow def daily_etl(day: str = "2026-09-06") -> str: return day assert daily_etl.project == "warehouse" ``` An **App** is the registry a module's flows, custom routes, and code rules belong to. Its name is the **project** every one of its flows is identified by: a flow's identity is `(project, name)`, so two projects can each have a flow called `etl` without colliding. The project is also how flows and runs are listed in the UI, unless a flow declares a [group](#groups) of its own. ## The default App You do not have to create an App. `@flow` without one registers on the default App, which is created on first use and named after the directory of the module that defined the first flow, lowercased and sanitised to `[a-z0-9][a-z0-9_-]*`: ``` from cereyan import flow @flow def etl() -> int: return 1 print(etl.project) # the basename of the directory this module lives in ``` Two directories with the same basename that both rely on the default App collapse into one project. Give them an explicit `App(name=)` when that matters. ## Groups The UI lists flows and runs in collapsible groups. A flow's group is the one it declares with `group=`, else its project: ``` from cereyan import flow @flow(group="nightly") def daily_etl() -> int: return 1 assert daily_etl.group == "nightly" ``` Groups are a flat axis rather than a level inside the project: flows in different projects that declare the same group form one group, and a group named after a project merges with the flows defaulting to it, which is how a flow joins a group it does not live beside. A run is grouped by its flow's group, read through the flow rather than stored on the run, so renaming a group moves the run history with it. What a collapsed group header shows is on the [tour](https://sercanatalik.github.io/cereyan/get-started/tour/#flows). ## What an App holds | Registered with | Holds | | ------------------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------- | | `@app.flow` or `@flow` | Flows, keyed by name; a second flow with the same name raises `FlowRegistrationError` | | `@app.get`, `@app.post`, `@app.put`, `@app.delete`, `@app.patch`, `@app.route` | Custom HTTP routes served next to the API (see [Add custom HTTP routes](https://sercanatalik.github.io/cereyan/guides/custom-routes/index.md)) | | `@app.rule` | Code rules whose action calls the decorated function (see [Events and rules](https://sercanatalik.github.io/cereyan/concepts/events-and-rules/index.md)) | `app.serve()` serves the directory of the module that created the App, the same as `cereyan serve ` with the App's options; the CLI is the usual way to serve, and `serve` exists for scripts that want to set host, port, or token in code. ## Registration and liveness Flows are upserted by `(project, name)` every time a server registers them and are never deleted automatically. The server records each flow's module and source directory so it can start engines for it and so scripts in other projects can hand runs to it. A flow the running server did not register shows as *not live* in the UI with its last-seen time, and can be deleted there. ## What stays global Resources and variables are shared by every project on the machine, because they describe machine-level things: a resource `db = 4` shared by two projects is the intended reading. Prefix variable names when projects need isolation, for example `warehouse/api_token`. Rules can be scoped to a project with the `project` field of their match clause, and events that reference a flow carry `project` in their payload. Related: [Flows and parameters](https://sercanatalik.github.io/cereyan/concepts/flows-and-parameters/index.md), [Engines and the home directory](https://sercanatalik.github.io/cereyan/concepts/engines-and-home/index.md). # Artifacts ``` from cereyan import flow, task, artifacts @task def load(rows: list[dict]) -> int: artifacts.create_table(rows, key="loaded-rows") artifacts.create_progress(100, key="load", label="load") return len(rows) @flow def etl() -> int: n = load([{"day": "2026-09-06", "rows": 3}]) artifacts.create_markdown(f"Loaded **{n}** batch(es).") return n assert etl() == 1 ``` An **artifact** is a small record a run or task run publishes for people to look at: a markdown note, a table, a progress bar, a link, or an image. Artifacts show on the run page's Artifacts tab and, across runs, on the Artifacts page. ## Kinds | Function | Shows | | -------------------------------------------------------------------------- | --------------------------------------------------------- | | `create_markdown(text)` | Rendered markdown | | `create_table(rows, columns=None)` | A table from a list of dicts or a list of lists | | `create_progress(percent, label=None)` and `update_progress(key, percent)` | A progress bar; updates under the same key keep a history | | `create_link(url, text=None)` | A link | | `create_image(url_or_bytes, media_type="image/png")` | An image by URL or embedded bytes | Each call returns the artifact id. An artifact is limited to 1 MB and can only be created inside a run. ## Keys A `key` groups artifacts across runs: every artifact published under `loaded-rows` forms a history you can open from the Artifacts page, newest first. Use keys for values you want to track over time, such as row counts or data-quality scores; leave them off for one-off notes. ## Reading artifacts `GET /api/runs/{id}/artifacts` and `GET /api/task-runs/{id}/artifacts` return a run's artifacts; `GET /api/artifacts` lists them across runs with filters for kind, key, flow, and project and keyset pagination. The MCP `list_artifacts` tool and the `cereyan://runs/{id}/artifacts` resource expose the same to agents. Retention does not delete artifacts; deleting a run deletes its artifacts. Related: [Publish artifacts](https://sercanatalik.github.io/cereyan/guides/artifacts/index.md), [Variables](https://sercanatalik.github.io/cereyan/concepts/variables/index.md). # Backfills ``` from datetime import date from cereyan import flow, task, LocalTarget def already_done(values: list[date]) -> set[date]: return {d for d in values if LocalTarget(f"out/{d}.csv").exists()} @task(output=lambda day: LocalTarget(f"out/{day}.csv")) def build(day: date) -> None: with LocalTarget(f"out/{day}.csv").open("w") as fh: fh.write("...") @flow(bulk_complete=already_done) def daily(day: date) -> None: build(day) assert daily.options["has_bulk_complete"] ``` A **backfill** creates one run of a flow per step of a date or datetime parameter over a range, in one transaction, and runs them under its own concurrency limit. It is how you compute history for a new pipeline, or recompute a range after a bug fix. Backfills need a running server. ## Creating one ``` cereyan backfill daily --param day --start 2026-06-01 --end 2026-08-30 --interval 1d --concurrency 4 ``` The same is available from the flow page's **Backfill** dialog, `POST /api/flows/{id}/backfill`, `Client.backfill()`, and the MCP `backfill` tool (which dry-runs by default and reports the count). `--interval` accepts seconds or a duration such as `1d` or `12h`; `--reverse` creates the newest value first; `--extra name=value` fixes other parameters. ## What happens 1. The values from start to end (inclusive) are enumerated. 1. If the flow defines `bulk_complete(values) -> set`, it is called once and the values it returns are recorded as `Skipped` runs without dispatching them, so a rerun over a range that is half done only executes the missing half. 1. One run per remaining value is created, tagged `backfill:` and with `created_by = backfill:`, all in one transaction. Ten thousand runs take under a second. 1. The runs execute through a resource named after the backfill with total `concurrency`, so they never take more engines than allowed, and each run also respects the flow's own cap and resources. `GET /api/backfills/{id}` reports counts by state; `POST /api/backfills/{id}/cancel` cancels the remaining Scheduled and Running runs. The Runs page filters by the backfill's tag. ## Idempotency Backfills pair with [targets](https://sercanatalik.github.io/cereyan/concepts/targets-caching-results/index.md): a task with `output=` skips days whose file exists, and `bulk_complete` avoids even scheduling them. Together they let you rerun a backfill over the same range as often as you like. Related: [Backfill a date range](https://sercanatalik.github.io/cereyan/guides/backfill/index.md), [Resources and concurrency](https://sercanatalik.github.io/cereyan/concepts/resources-and-concurrency/index.md). # Dependencies ``` from datetime import date from cereyan import App app = App("reporting") @app.flow def sales(day: date) -> str: return f"sales {day}" @app.flow def inventory(day: date) -> str: return f"inventory {day}" @app.flow(after=["sales", "inventory"], batch_key="day") def report(day: date) -> str: return f"report {day}" @app.flow(after=("report", {"for_day": "{{ run.parameters.day }}"})) def notify(for_day: date) -> str: return f"notified {for_day}" assert report.after["flows"] == ["sales", "inventory"] assert report.after["key"] == "day" ``` A **dependency** makes one flow run after another. It is declared on the downstream flow with `after=` and evaluated by the server when upstream runs end: a run of the downstream is created with `created_by = run:` and a link to the triggering run in its details. Dependencies need a running server; offline, they are recorded with the flow. ## Single upstream `after="sales"` creates a downstream run whenever a `sales` run ends `Completed` or `Skipped`. A failed upstream creates nothing. Upstream parameters are copied to the downstream by name, and `after=("sales", {"for_day": "{{ run.parameters.day }}"})` renames or derives them with the same templates rules use. A run skipped by a person, a [skipped fire](https://sercanatalik.github.io/cereyan/concepts/schedules/#skipping-fires) of its schedule, is carried down instead: the downstream run is created already `Skipped`, with `details.reason = "upstream"` and `details.upstream_run` naming the skipped run, and the flows after it follow the same way. Every other `Skipped` run (`on_overlap="skip"`, a backfill value already done, a catch-up drop) means nothing needed doing, and triggers the downstream like a `Completed` one. ## Fan-in with a key `after=["sales", "inventory"], batch_key="day"` runs the downstream once per value of `day`, after *every* listed upstream has a Completed or Skipped run for that value. The rules: - The batch is identified by the value of `batch_key` in the upstream runs' parameters; different values never mix. - The downstream run is created when the last upstream completes the batch, with the key value and the usual copied, templated, and default parameters, and a `flow.fan_in` event records the key, the value, and the upstream run ids. - At most one downstream run exists per key value: an existing run with that value, however it was created, blocks another. Rerunning an upstream for a day that already has a report creates nothing. - A failed upstream blocks the batch until a rerun of it completes. - When any upstream's latest run for the value was skipped by a person, or skipped because its own upstream was, the downstream run is created `Skipped` once the batch is complete. `batch_key` is required when `after` lists more than one flow; registration rejects the flow otherwise. ## Visibility The flow page lists upstreams under *Triggered by* and downstreams under *Triggers*, the dependency graph draws one edge per upstream, and a run created by a dependency links to the run that triggered it. An `after=` naming a flow the server has not registered is reported as a flow error at start without stopping other flows. Related: [Chain flows](https://sercanatalik.github.io/cereyan/guides/chain-flows/index.md), [Events and rules](https://sercanatalik.github.io/cereyan/concepts/events-and-rules/index.md) for the more general way to run a flow when something happens. # Engines and the home directory ``` from cereyan import flow @flow(isolated=True) def heavy() -> str: return "fresh process every time" assert heavy.options["isolated"] ``` Cereyan has two execution paths that write the same store with the same rules. ``` offline served ─────── ────── python pipeline.py cereyan serve dir/ │ runs in-process │ imports every module under dir/ │ writes db.sqlite directly ├── HTTP API + UI + MCP │ under an OS advisory lock ├── scheduler (timer heap) ▼ ├── rules engine db.sqlite └── supervisor ──▶ engine processes ▲ (one module each, warm, └─── while a server holds the lock, a script hands ────── report over loopback HTTP) its run to the server and streams the logs back ``` ## The offline path `python pipeline.py` and `cereyan run` execute the flow in the current process and write runs, task runs, logs, events, and artifacts straight into `db.sqlite`, holding `db.lock` while they do. Nothing else needs to run. If a server holds the lock, the script cannot take it; it reads `server.json`, submits the run to the server with the module and directory to import, and streams the logs back to the terminal until the run ends. That handoff works across projects, so a script in one directory can run through a server started in another. ## The served path `cereyan serve dir/` is one process hosting the HTTP API, the UI, the scheduler, the rules engine, the MCP endpoint, and the **supervisor**, which keeps a warm pool of **engine** child processes. - Each engine is bound to one Python module, imports it once, and executes runs of its flows one at a time. Because the import happens once, module-level state — an HTTP client and its connection pool, a warmed cache — is shared by every run that engine serves; see [fetching from an HTTP API](https://sercanatalik.github.io/cereyan/guides/fetch-from-an-api/index.md). Engines are keyed by `(source_dir, module)` and pooled up to `max_engines` (default: CPU count). - An engine is recycled after `engine_max_runs` runs (default 100) or when its module file changes, so edits are picked up without restarting the server. `@flow(isolated=True)` gives every run of that flow a fresh process, terminated afterwards. - Engines report task-run transitions and logs in batches every 100 milliseconds (immediately on flow-level transitions) over one keep-alive connection, and heartbeat every five seconds per active run. Three missed heartbeats and a dead PID mark the run `Crashed`; it is rerun up to `crash_retries` times. - Cancelling a run moves it to `Cancelling`, tells the engine, waits `cancel_grace_secs`, sends SIGTERM, waits again, then SIGKILL. The engine raises `KeyboardInterrupt` in the flow so it can clean up; on Windows that interrupt cannot reach a blocking call, so a flow sleeping or waiting on I/O runs until the supervisor ends the process, which Windows does abruptly in place of SIGTERM. - Engines survive a server restart: on start the supervisor adopts runs whose engine PID is still alive and waits for the engine to reconnect; runs whose engine is gone are marked `Crashed` with the reason. - Stopping the server ends its engines. While shutting down it answers every waiting engine with an instruction to exit, then signals any that were between requests; an engine executing a run is left running so a restarted server can adopt it. An engine that loses its server without being told to stop — a kill, a crash — exits by itself: about thirty seconds when the server refuses connections outright, and within forty in the worst case, since it can only notice between requests and a request waits that long before giving up. One exception: an engine still reporting a run it has just finished keeps trying for up to ten minutes, so that a server restarted in the meantime records the outcome rather than seeing a crash. - An engine that fails to import its module fails every run assigned to it with the traceback, and the flow shows the error in the API and the UI. The Settings page lists the engines with their PID, module, runs done, and current run. ## The home directory One runtime home per machine holds everything: `db.sqlite`, `db.lock`, `server.json` while a server runs, `secret.key` once a secret exists, and `storage/` for persisted results. Only the account that created it can read it, which is what protects the store and the key alike. It resolves from `--home`, then `CEREYAN_HOME`, then `~/.cereyan`; `cereyan.toml` cannot move it, so a repository can never point the store elsewhere. Engine children inherit it through `CEREYAN_HOME`. Because the home is global, one server runs per machine and holds the advisory lock. History is a cache: losing `db.sqlite` loses the run list, not your data, which lives in your targets. A corrupted database is quarantined and a fresh one created. ## Performance shape The store is SQLite in WAL mode with one writer thread and group commit, a read pool, and an in-memory working set (active runs, counters, the schedule heap, resources, the rule index). Engine reports are serialised in Rust. The targets, such as 20k task transitions per second and a runs list under 10 ms at a million runs, are on the [Performance targets](https://sercanatalik.github.io/cereyan/design/performance/index.md) page. Related: [Run the server as a service](https://sercanatalik.github.io/cereyan/guides/run-as-a-service/index.md), [Secure the server](https://sercanatalik.github.io/cereyan/guides/secure-the-server/index.md), [Exit codes and the discovery file](https://sercanatalik.github.io/cereyan/reference/exit-codes/index.md). # Events and rules ``` from cereyan import App, emit_event, events, states app = App("shop") @app.rule(on="orders.*", flow="check_orders", once="per_run") def alert(event, run): print("empty table:", event["payload"]["table"], "in run", run["name"]) @app.rule(on=events.run.failed, states=[states.Failed]) def on_failure(event, run): print("failed:", run["name"]) @app.flow def check_orders() -> None: emit_event("orders.table_empty", {"table": "orders"}) spec = app.rules[0].spec() assert spec["when"]["events"] == ["orders.*"] assert spec["do"][0]["kind"] == "call" assert events.run.failed == "run.failed" ``` An **event** is a recorded fact: a name such as `run.failed`, a resource it is about, related resources, a payload, and a sequence number. The engine records one for every meaningful change (run and task-run transitions, schedule changes, rule firings, flow registration, resource exhaustion, expectations), and `emit_event` records custom ones. The [events catalogue](https://sercanatalik.github.io/cereyan/reference/events/index.md) lists them all. A **rule** is `when` plus `do`: a match clause over events and an ordered list of actions, with guards. Rules are how cereyan reacts: run a cleanup flow when an ETL completes, page someone when a nightly flow fails, cancel a run that was started by mistake. ## Matching `when` names event names or prefixes (`run.*`), flows, tags, states, and a project. A rule fires for an event that matches every clause it sets. A value in `states` matches the run's state *type* or its sub-state *name*, so `states=["Scheduled"]` covers a run that is `Late` or `AwaitingRetry` and `states=["Late"]` narrows to just that one. See [States and transitions](https://sercanatalik.github.io/cereyan/reference/states/index.md). ### Names are checked The engine owns the prefixes `run.`, `task_run.`, `flow.`, `schedule.`, `resource.`, `rule.`, and `expectation.`. A name under one of them that the engine never emits — `run.failure` for `run.failed` — is rejected when the rule is declared, rather than sitting silent forever: ``` import pytest from cereyan import App with pytest.raises(ValueError, match='did you mean "run.failed"'): @App("typo").rule(on="run.failure") def never(event, run): ... ``` Every other name is yours and is never checked, so `on="orders.table_empty"` needs no registration. `cereyan.events` and `cereyan.states` carry the catalogue if you would rather not type the strings: `events.run.failed` *is* `"run.failed"`, and `events.run.any` is `"run.*"`. ## Actions | Action | Effect | | ----------------------------------- | --------------------------------------------------------------- | | `run_flow` | Create a run of a flow with templated parameters | | `cancel_run` | Cancel the run the event is about | | `set_state` | Force the run into a state | | `pause_schedule`, `resume_schedule` | Stop or restart a schedule | | `webhook` | HTTP request with a templated body; three attempts with backoff | | `email` | SMTP through `[email]` in `cereyan.toml` | | `call` | A code rule's function, `fn(event, run)` | Templates use Jinja syntax rendered in Rust with `event`, `run`, `flow`, `state`, `payload`, and `parameters` in scope. An undefined variable is an error; a failing template fails only its action and is recorded as `rule.action.failed`. ## Guards Disabled rules never fire. `once=per_run` (the default) fires at most once per run and `once=never` drops that limit; `cooldown_seconds` and `max_per_minute` throttle; and a rule never fires on events of runs it created unless `allow_self` is set, so a rule that runs a flow cannot trigger itself forever. A keyword that is not one of these is rejected rather than ignored, so a misspelled guard cannot leave a rule running on defaults. ## Two kinds of rule **Data rules** are created on the Rules page or through `POST /api/rules`, stored in the database, and editable at runtime. **Code rules** come from `@app.rule(...)`, run their function as a `call` action, are re-registered on every server start, and show read-only in the UI. Once a server has registered them in the store they also fire on the offline path for a script's own events; a script whose rules have never been served records the events but does not fire the rules. ## Proactive rules A reactive rule fires when something happens. A proactive rule fires when something expected does *not* happen: add `unless` naming the expected event, and either `within` seconds of the arming `on` event (event-armed) or `at` a cron expression in timezone `tz` (clock-armed). A lapse is recorded as an `expectation.lapsed` event and the rule's actions run against it. See [Detect when something did not happen](https://sercanatalik.github.io/cereyan/guides/detect-missing-events/index.md). ## Where events go The Events page and `GET /api/events` filter by name prefix, resource, flow, run, and time with keyset pagination; the SSE stream delivers new ones as `event.created`; the MCP `list_events` tool reads them; and retention deletes events older than `retain_days`. Related: [React to events with rules](https://sercanatalik.github.io/cereyan/guides/rules/index.md), [Artifacts](https://sercanatalik.github.io/cereyan/concepts/artifacts/index.md). # Flows and parameters ``` from datetime import date from cereyan import flow @flow(run_name="etl-{day}", tags=["example"]) def etl(day: date, full: bool = False) -> str: return f"{day} full={full}" assert etl(day="2026-01-02", full="yes") == "2026-01-02 full=True" ``` A **flow** is a decorated function. Calling it starts a **run**: the tasks it calls are recorded as task runs, its logs are stored, and it ends in a terminal state. Offline the run is recorded in the local store. While a server holds the store, the call hands the run to the server and streams the logs back. A flow's identity is `(project, name)`; see [App and projects](https://sercanatalik.github.io/cereyan/concepts/app-and-projects/index.md). ## Parameters come from type hints Values arrive from the decorator call, the CLI (`--param day=2026-01-02`), the API and the UI run form (JSON), a schedule's defaults, or a backfill. Whatever the source, they are coerced to the declared types before the body runs, and a value that does not coerce raises `ParameterError` (exit code 3 on the CLI, 422 on the API) before anything is recorded as started. | Hint | Accepted values | | ------------------------- | ------------------------------------------------------------------------ | | `str`, `int`, `float` | The type, or a string that parses | | `bool` | `true`, `1`, `yes`, `on`, `y`, `t` and their negatives, case-insensitive | | `date`, `datetime` | ISO 8601 strings | | `timedelta` | Seconds, `MM:SS`, or `HH:MM:SS` | | `Optional[T]`, \`T | None\` | | `list[T]`, `dict[str, T]` | JSON on the CLI; elements coerced as `T` | | `Literal[...]` | One of the listed values | | `Enum` | The member's value, or its name | | dataclasses | A JSON object; fields coerced by their hints | Any other hint is opaque: the value passes through unchanged and the schema describes it as untyped. The schema derived from the hints drives the UI's run form and is exposed by the API as `parameter_schema`. ## Naming runs Runs get a random `adjective-animal` name unless the flow sets `run_name`: a `str.format` template over the parameters, or a callable receiving them as keyword arguments. Names need not be unique. ## What a flow can declare | Concern | Options | Where | | ------------ | ------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | When it runs | `schedule`, `schedules` | [Schedules](https://sercanatalik.github.io/cereyan/concepts/schedules/index.md) | | Reliability | `retries`, `retry_delay`, `timeout_seconds`, `crash_retries` | [Retry, time out and survive crashes](https://sercanatalik.github.io/cereyan/guides/retries-timeouts-crashes/index.md) | | Hooks | `on_completion`, `on_failure`, `on_crashed`, `on_cancellation` | [Run code on state changes](https://sercanatalik.github.io/cereyan/guides/state-hooks/index.md) | | Concurrency | `max_concurrent`, `on_overlap`, `resources`, `priority`, `disable_after` | [Resources and concurrency](https://sercanatalik.github.io/cereyan/concepts/resources-and-concurrency/index.md) | | Dependencies | `after`, `batch_key` | [Dependencies](https://sercanatalik.github.io/cereyan/concepts/dependencies/index.md) | | Backfills | `bulk_complete` | [Backfills](https://sercanatalik.github.io/cereyan/concepts/backfills/index.md) | | Execution | `runner`, `isolated`, `log_prints` | [Tasks](https://sercanatalik.github.io/cereyan/concepts/tasks/index.md), [Engines and the home directory](https://sercanatalik.github.io/cereyan/concepts/engines-and-home/index.md) | | Display | `name`, `description`, `tags`, `group` | [App and projects](https://sercanatalik.github.io/cereyan/concepts/app-and-projects/index.md) | The full option list with types and defaults is in the [Python API reference](https://sercanatalik.github.io/cereyan/reference/python-api/#cereyan.flow). ## The flow object `@flow` returns a `Flow`, not the function. It is still callable, and it exposes what was derived: `parameters`, `schema`, `project`, `name`, `tags`, and `options`. Use it in tests to call the flow directly or to read its schema. # Resources and concurrency ``` from cereyan import flow, task @task(resources={"db": 1}) def load(rows: int) -> int: return rows @flow(max_concurrent=1, on_overlap="skip", priority=5, resources={"gpu": 1}) def nightly(rows: int = 10) -> int: return load(rows) assert nightly() == 10 assert nightly.options["max_concurrent"] == 1 ``` A **resource** is a named counting semaphore with a total set in `cereyan.toml` (`[resources]`) or on the Settings page. Flows and tasks declare how much of a resource each run holds; a run whose resources are not available waits as `AwaitingResource` and is dispatched as soon as they are. Resources are released on every terminal state, on pause, and when an engine dies, so a crash cannot leak capacity. They are shared by every project on the machine. Offline, a resource is a local semaphore of size one, which serialises tasks holding the same name inside a single process. ## Per-flow cap and overlap `max_concurrent=N` is a resource named after the flow with total `N`: at most `N` runs of the flow are Pending or Running at once. The default is unlimited. `on_overlap` decides what a new run does when the cap is reached: | Value | Behaviour | | ------------------- | -------------------------------------------------------------------------------------------- | | `enqueue` (default) | Wait as `AwaitingResource` and start when a slot frees | | `skip` | End `Skipped` at once with the message `previous run still active` and a `run.skipped` event | | `cancel_new` | End `Cancelled` with the same message | ## Priority `priority` orders dispatch among runs waiting for an engine or a resource, higher first. It never preempts: a running run is never paused or killed to make room. A negative priority also raises the engine's OS niceness (`min(19, -priority)`) on Linux and macOS, so heavy background flows yield CPU to interactive work; such engines are pooled by their niceness and count toward `max_engines`. ## Disable windows `disable_after=(count, window_seconds, persist_seconds)` pauses every schedule of the flow for `persist_seconds` once it has failed `count` times within `window_seconds`, records `flow.disabled`, and resumes automatically with `flow.enabled`. Use it to stop a broken nightly flow from filling the run list until someone looks. The window survives a restart of the server: one still open when the server starts again ends on time, and one that ended while it was down resumes as it starts. Fires inside the window are never caught up; fires after it follow the schedule's `catchup` policy. ## Engines The number of runs executing at once is also bounded by the engine pool: `max_engines` (default: CPU count) engine processes, each running one run at a time. Resources and caps decide *which* runs may proceed; the pool decides *how many*. See [Engines and the home directory](https://sercanatalik.github.io/cereyan/concepts/engines-and-home/index.md). ## Backfills A backfill has its own resource, sized by its `concurrency`, so a large date range never floods the pool; see [Backfills](https://sercanatalik.github.io/cereyan/concepts/backfills/index.md). Related: [Limit concurrency and overlap](https://sercanatalik.github.io/cereyan/guides/resources-and-overlap/index.md), [Schedules](https://sercanatalik.github.io/cereyan/concepts/schedules/index.md). # Runs and states ``` from cereyan import flow, task attempts = [] @task(retries=1) def flaky() -> str: attempts.append(1) if len(attempts) == 1: raise RuntimeError("first try fails") return "ok" @flow def resilient() -> str: return flaky() assert resilient() == "ok" assert len(attempts) == 2 # task run: Running, AwaitingRetry, Retrying, Completed ``` A **run** is one execution of a flow. A **task run** is one execution of a task inside it. Both carry a **state** with a type, an optional sub-state name, a message, details, and a timestamp, and both keep the whole history of states they went through. ## State types ``` Scheduled ──▶ Pending ──▶ Running ──▶ Completed │ │ │ ▲ Failed │ │ │ │ Crashed │ │ ▼ │ │ │ Paused │ │ │ └────────────┴───────────┴──▶ Cancelling ──▶ Cancelled ``` | Type | Meaning | | --------------------------------------------- | ------------------------------------------------------------------- | | `Scheduled` | Created; waiting for its time, a retry delay, or a resource | | `Pending` | Dispatched to an engine | | `Running` | Executing | | `Paused` | Waiting for a person to answer `wait_for_input`; the engine is free | | `Cancelling` | Asked to stop | | `Completed`, `Failed`, `Cancelled`, `Crashed` | Terminal | Named sub-states refine a type: `Late` and `AwaitingRetry` and `AwaitingResource` (Scheduled), `Retrying` (Running), `TimedOut` (Failed), `Cached` and `Skipped` (Completed). The exact transition rules, shared by the offline and served paths, are on the [States and transitions](https://sercanatalik.github.io/cereyan/reference/states/index.md) page. ## What a run records - The flow, project, parameters, tags, and name. - `created_by`: `script` for a plain call, `client` for the API and CLI, `schedule:`, `backfill:`, `rule:`, `run:` for a dependency, `catchup`, `crash:` for a crash rerun, and `mcp:` for an agent. - Timing: created, scheduled, start, end, total run time; and counters for failures and crashes. - Every log line, task run, artifact, and event. Inside the run, `cereyan.runtime.run`, `cereyan.runtime.task_run`, and `cereyan.runtime.flow` read the run in progress — its id, name, and parameters — and are `None` outside one; see the [Python API](https://sercanatalik.github.io/cereyan/reference/python-api/index.md). ## Attempts Retries, crash reruns, and resumes after a pause are new attempts of the same run: the run keeps its id and history, `failure_count` or `crash_count` grows, and the task runs of the new attempt are recorded alongside the old ones. Tasks marked `cache=INPUTS` return their stored result on a replay instead of executing again. ## Finding runs The Runs page, `cereyan runs ls`, `GET /api/runs`, `Client.runs()`, and the MCP `list_runs` tool all filter by flow, project, state type and name, tags, name, and time, and page by keyset cursor. Task runs have the same across `GET /api/task-runs`. Related: [Tasks](https://sercanatalik.github.io/cereyan/concepts/tasks/index.md), [Events and rules](https://sercanatalik.github.io/cereyan/concepts/events-and-rules/index.md), [Retry, time out and survive crashes](https://sercanatalik.github.io/cereyan/guides/retries-timeouts-crashes/index.md). # Schedules ``` from datetime import datetime, timedelta from cereyan import flow, Cron, Interval, RRule @flow(schedule=Cron("0 9 * * 1-5", timezone="Europe/Istanbul")) def weekday_report(day: str = "today") -> str: return day @flow(schedules=[ Interval(timedelta(hours=1), anchor=datetime(2026, 1, 1, 0, 30)), RRule("DTSTART:20260101T090000\nRRULE:FREQ=MONTHLY;BYMONTHDAY=1", timezone="UTC", catchup="latest"), ]) def hourly_and_monthly() -> None: ... assert len(hourly_and_monthly.schedules) == 2 ``` A **schedule** tells the server when to create runs of a flow. It is declared in code as above, or created and edited on the flow page. A flow can have several, each pausable on its own. Schedules only do something while a server runs; offline, they are recorded with the flow and nothing fires. ## Kinds | Kind | Declaration | Notes | | -------- | ---------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------- | | Cron | `Cron("0 9 * * *", timezone=..., day_or=True)` | Five fields, evaluated by wall clock in the timezone. `day_or` keeps cron's rule that day-of-month and day-of-week are ORed. | | Interval | `Interval(seconds or timedelta, anchor=..., timezone=...)` | Fires every interval from the anchor. Intervals under a day are elapsed time; longer ones keep their local wall-clock time across DST changes. | | RRule | `RRule("DTSTART:...\nRRULE:...", timezone=...)` | An iCalendar recurrence rule set; must include `DTSTART`. | Timezones are IANA names and default to the machine's local zone. Each schedule may carry a `key` so a code declaration and its runtime edits stay matched; keys default to `code-0`, `code-1`, and so on. ## How the scheduler works - For every active schedule it keeps at least three future runs materialised (and at least one hour of coverage, at most 100 runs), so the Upcoming tab and the dashboard show what comes next. The runs sit in `Scheduled` until their time. - It wakes on a timer for the earliest due run, not by polling, and dispatches within 50 milliseconds of the scheduled time when an engine is free. - A run that has not started 15 seconds after its time is renamed `Late` and a `run.late` event is recorded; it still runs as soon as it can. - Pausing a schedule removes its not-yet-started runs; resuming materialises them again. ## Skipping fires Skip a fire when one run should not happen but the schedule should stay on: from the Flows page menu (**Skip next run**, **Skip runs…**), the flow page's **Upcoming** tab, or `POST /api/schedules/{id}/skips` with a list of `fires` or `{"next": N}`. A skip names one fire time of one schedule, at most 100 fires ahead, and lasts until that time passes: it survives a restart, a pause and resume, and an edit that still produces the time. At its time the fire's run ends `Skipped` with `details.reason = "user"` without starting, shows in Runs, and records `run.skipped`; until then **Undo** or `DELETE /api/schedules/{id}/skips/{fire}` takes it back. The look-ahead keeps three runs that will start, so it reaches past skipped fires, and the flows that run after this one are skipped for that fire too (see [Dependencies](https://sercanatalik.github.io/cereyan/concepts/dependencies/index.md)). ## Catch-up When the server starts after downtime, each schedule's `catchup` policy decides what happens to the fires it missed: `skip` (default) drops them, `latest` creates the most recent one, and `all` creates every one up to `catchup_max` (default 100). A skipped fire is never caught up. Catch-up runs carry `created_by = catchup` and the decision is recorded as a `schedule.catchup` event. ## Parameters and names A scheduled run gets the flow's default parameter values, and the schedule editor can set overrides. Use `run_name` templates over parameters, or the default scheduled name `flow-YYYYMMDDTHHMMSS`, to tell runs apart. ## Editing at runtime Schedules declared in code can be edited on the flow page or with **Reschedule…** (`PATCH /api/schedules/{id}`). The edit lasts until the server restarts, when the code declaration applies again; send `persist: true` to keep it and detach the schedule from its declaration for good. Skips whose fire time the edited schedule no longer produces are dropped and recorded as a `schedule.skips_dropped` event, as are those a restored declaration no longer produces. `POST /api/schedules/preview` returns the next fire times for a declaration, which the editors use to show them before saving. An agent can do the same through the [MCP tools](https://sercanatalik.github.io/cereyan/reference/mcp/index.md): `list_schedules` shows what is scheduled and `create_schedule`, `edit_schedule`, `delete_schedule`, `pause_schedule` and `resume_schedule` manage it. Two things differ from the flow page. An edit to a schedule declared in code lasts until the next restart, and the tool says so in its result, because an agent cannot see the note the flow page shows. Deleting one is refused outright, since the declaration would recreate it at the next restart; pause it instead, or remove the declaration from the flow. Related: [Schedule a flow](https://sercanatalik.github.io/cereyan/guides/schedule-a-flow/index.md), [Backfills](https://sercanatalik.github.io/cereyan/concepts/backfills/index.md), [Resources and concurrency](https://sercanatalik.github.io/cereyan/concepts/resources-and-concurrency/index.md). # Targets, caching and results ``` from datetime import date, timedelta from cereyan import flow, task, LocalTarget, INPUTS calls = [] @task(output=lambda day: LocalTarget(f"out/{day}.csv")) def export(day: date) -> None: calls.append("export") with LocalTarget(f"out/{day}.csv").open("w") as fh: fh.write("a,b\n1,2\n") @task(cache=INPUTS, persist_result=True, cache_expires=timedelta(hours=1)) def summarize(day: date) -> dict: calls.append("summarize") return {"day": str(day), "rows": 1} @flow def daily(day: date) -> dict: export(day) return summarize(day) first = daily(date(2026, 1, 1)) second = daily(date(2026, 1, 1)) assert first == second assert calls == ["export", "summarize"] # the second run skipped export and hit the cache ``` Cereyan treats your code and the files it writes as the source of truth, and its own database as a cache of history. Two mechanisms make reruns cheap and safe: **targets** for outputs that live outside cereyan, and **result caching** for return values. ## Targets A **Target** is anything with `exists()`. `LocalTarget(path)` is a file: it writes through a temporary path that is renamed onto the target on close, so a reader never sees a partial file and a crash leaves no half-written output. Declare a task's target with `output=`: a Target, or a callable over the task's arguments returning one. When the target exists the task run ends `Skipped` with the message `output exists` and its body does not execute. That is what makes rerunning a day, or a whole backfill, idempotent. Targets say nothing about *how* the file is written: the task body still has to write it, usually through the same `LocalTarget`'s `open("w")`. ## Result caching With `persist_result=True` a task's return value is stored under `/storage/` (pickle by default, `serializer="json"` for JSON) along with the Python version. `cache=` then reuses it: | Policy | Hit when | | ----------------- | ----------------------------------- | | `INPUTS` | The arguments are unchanged | | `SOURCE` | The task's source code is unchanged | | `INPUTS + SOURCE` | Both | A hit ends the task run `Cached` with the stored result and records `task_run.cached`. `cache_expires=timedelta(...)` makes entries stale after that long, and a result written by a different Python minor version counts as a miss. Caching requires `persist_result=True`; the decorator rejects the combination otherwise. ## Replay after a pause A run resumed after `wait_for_input` starts a new attempt from the top of the flow. Tasks marked `cache=INPUTS` return their stored results on the replay, so work done before the question is not repeated. See [Pause a run for approval](https://sercanatalik.github.io/cereyan/guides/human-approval/index.md). ## Where results live `storage/` is inside the runtime home, is never cleaned by retention, and is safe to delete: a missing entry is a cache miss. Results are not the way to pass data between flows; write a target and read it downstream. Related: [Make reruns idempotent with targets](https://sercanatalik.github.io/cereyan/guides/idempotent-reruns/index.md), [Cache task results](https://sercanatalik.github.io/cereyan/guides/cache-results/index.md), [Backfills](https://sercanatalik.github.io/cereyan/concepts/backfills/index.md). # Tasks ``` from cereyan import flow, task @task(retries=2) def fetch(n: int) -> list[int]: return list(range(n)) @task def total(rows: list[int]) -> int: return sum(rows) @flow def pipeline(n: int = 4) -> int: return total(fetch(n)) assert pipeline() == 6 ``` A **task** is a decorated function called inside a flow. Each call becomes a **task run** recorded with the run: its state, timing, retries, logs, and result. Outside a flow a task is an ordinary function call. ## Task runs and dynamic keys Every call gets a dynamic key made of the task name and a counter, such as `fetch-0` and `fetch-1`, so the same task called twice in one run is two task runs. Task runs record which earlier task runs produced their arguments, which is what the run page's timeline and dependency views draw. ## What a task can do | Option | Effect | Guide | | --------------------------------------------------------- | -------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------- | | `retries`, `retry_delay` | Rerun on failure, waiting a fixed, per-attempt, or exponential delay | [Retry, time out and survive crashes](https://sercanatalik.github.io/cereyan/guides/retries-timeouts-crashes/index.md) | | `timeout_seconds` | Fail the task run as `TimedOut` after the limit | same | | `output=` | Skip the task when its target already exists | [Make reruns idempotent with targets](https://sercanatalik.github.io/cereyan/guides/idempotent-reruns/index.md) | | `cache=`, `cache_expires`, `persist_result`, `serializer` | Reuse a persisted result when inputs or source are unchanged | [Cache task results](https://sercanatalik.github.io/cereyan/guides/cache-results/index.md) | | `resources` | Hold named resources while the task runs | [Limit concurrency and overlap](https://sercanatalik.github.io/cereyan/guides/resources-and-overlap/index.md) | | `on_completion`, `on_failure`, `on_cancellation` | Call hooks with `(task, run, state)` | [Run code on state changes](https://sercanatalik.github.io/cereyan/guides/state-hooks/index.md) | | `log_prints` | Tee `print` into the run log | [Tasks](#logging) below | | `name`, `description`, `tags` | What the UI shows | | ## Concurrency `task.submit(...)` returns a `Future` instead of waiting, and `task.map(iterable)` submits one task run per element. Futures passed as arguments to another task are resolved before it starts, and `wait_for=` adds ordering without passing data. The flow's `runner` decides where submitted tasks execute: a `ThreadRunner` by default, or a `ProcessRunner` for CPU-bound work. See [Run tasks concurrently](https://sercanatalik.github.io/cereyan/guides/concurrent-tasks/index.md). ``` from cereyan import flow, task @task def square(x: int) -> int: return x * x @flow def squares() -> list[int]: return [f.result() for f in square.map([1, 2, 3])] assert squares() == [1, 4, 9] ``` ## Logging Inside a task, `get_run_logger()` returns a logger whose records are stored with the run and tagged with the task run; standard `logging` calls from any logger are captured the same way. With `log_prints=True` on the task or the flow, `print` output is logged at INFO as well. ## Results A task's return value flows back to the caller as usual. With `persist_result=True` it is also written under `/storage`, which is what caching and replay after a pause read from. Related: [Flows and parameters](https://sercanatalik.github.io/cereyan/concepts/flows-and-parameters/index.md), [Runs and states](https://sercanatalik.github.io/cereyan/concepts/runs-and-states/index.md). # Variables ``` from cereyan import Variable, flow Variable.set("region", "eu", tags=["infra"]) Variable.set("warehouse/api_token", "s3cret", secret=True) @flow def where() -> str: return f"{Variable.get('region')}:{Variable.get('warehouse/api_token')}" assert where() == "eu:s3cret" assert Variable.get("missing", default="none") == "none" ``` A **variable** is a small named JSON value, up to 64 KB, that flows read at run time: a region, a feature flag, a threshold, a credential. Variables are shared by every project on the machine; prefix names with the project (`warehouse/api_token`) when they should not be. ## Names and values Names match `[a-z0-9][a-z0-9_./-]*`. Values are anything JSON-serialisable; dates and dataclasses are converted. Tags are free-form and only used for display. ## Secrets `secret=True` encrypts the value at rest with a key kept at `/secret.key` (created on first use, mode 0600) and masks it in the API, the UI, and the MCP `set_variable` tool. `Variable.get` decrypts it for the running process. Secrets stay on the machine: there is no remote secret store, and the key never leaves the home. ## Where they live and who can change them Variables are rows in the store. Offline, `Variable.set` writes to the store directly; while a server holds the store, the same call goes through `POST /api/variables`. The Variables page and the API create, edit, and delete them, and an agent can create or overwrite one through MCP. `overwrite=False` makes `set` fail instead of replacing an existing value. Related: [Store variables and secrets](https://sercanatalik.github.io/cereyan/guides/variables-and-secrets/index.md), [App and projects](https://sercanatalik.github.io/cereyan/concepts/app-and-projects/index.md) for what is global. # Guides # How to use cereyan with an AI agent Cereyan has a built-in [MCP](https://modelcontextprotocol.io) server, so an agent can list flows, start runs, follow and diagnose them, backfill, manage schedules, and answer a paused run's question. Nothing extra to install; it ships in the wheel and runs inside `cereyan serve`. ## Connect Claude Code or Claude Desktop Hosts that speak stdio start `cereyan mcp`, which proxies every message to the running server. Add to the host's MCP configuration: ``` { "mcpServers": { "cereyan": { "command": "cereyan", "args": ["mcp"], "env": {"CEREYAN_TOKEN": "your-token"} } } } ``` For Claude Code that is `claude mcp add cereyan -- cereyan mcp` or the same block in `.mcp.json`. `cereyan mcp` finds the server through `server.json` in the runtime home, and `--url` and `--socket` override it. The API token comes from `CEREYAN_TOKEN`, as in the block above; `--token` is a global flag, so on the command line it goes before the subcommand (`cereyan --token mcp`, not `cereyan mcp --token `). With a Unix socket and no token the proxy connects over the socket. Start `cereyan serve` first; the proxy reports an error to the host when no server is running. ## Connect over HTTP Web agents and custom clients use Streamable HTTP: `POST /mcp` with one JSON-RPC message per request and `Authorization: Bearer ` when the server has a token. `initialize` returns an `Mcp-Session-Id` header to send on later requests; `DELETE /mcp` ends the session. ``` import json, urllib.request session = {} def rpc(method, params=None, *, notification=False): msg = {"jsonrpc": "2.0", "method": method, "params": params or {}} if not notification: msg["id"] = 1 headers = {"content-type": "application/json", **session} req = urllib.request.Request(served.url + "/mcp", data=json.dumps(msg).encode(), headers=headers, method="POST") with urllib.request.urlopen(req) as resp: if resp.headers.get("Mcp-Session-Id"): session["Mcp-Session-Id"] = resp.headers["Mcp-Session-Id"] raw = resp.read() return json.loads(raw) if raw else None rpc("initialize", {"protocolVersion": "2025-06-18", "clientInfo": {"name": "docs-example", "version": "1"}, "capabilities": {}}) rpc("notifications/initialized", notification=True) tools = {t["name"] for t in rpc("tools/list")["result"]["tools"]} assert {"list_flows", "run_flow", "explain_failure", "resume_run"} <= tools ``` ## What the agent can do The tool set is curated: nine read-only tools (`list_flows`, `list_runs`, `get_run`, `run_logs`, `list_events`, `list_artifacts`, `list_rules`, `list_schedules`, `explain_failure`) and ten that change state (`run_flow`, `cancel_run`, `resume_run`, `backfill`, `create_schedule`, `edit_schedule`, `delete_schedule`, `pause_schedule`, `resume_schedule`, `set_variable`). Every description states its effect, `backfill` dry-runs unless told otherwise, and rule creation is not exposed. The full list with each argument's type, default, and range, and the keys every tool returns, is on the [MCP reference](https://sercanatalik.github.io/cereyan/reference/mcp/index.md) page. Schedules are the one place where an agent has less room than the flow page: editing a schedule that was declared in code detaches it from that declaration for good and the result says so, and deleting such a schedule is refused, because the declaration would recreate it at the next restart. See [Schedules](https://sercanatalik.github.io/cereyan/concepts/schedules/index.md). Two resource templates, `cereyan://runs/{id}/logs` and `cereyan://runs/{id}/artifacts`, expose a run's logs and artifacts as JSON — they are listed by `resources/templates/list`, not `resources/list` — and the `diagnose_run` prompt tells the model to call `explain_failure` and summarise the cause. ## Know what the agent did Runs created through MCP record `created_by = mcp:` from the `initialize` handshake, so the Runs page, the events feed, and `list_runs` show which agent started what. Filter the Runs page by that value to audit agent activity. ## Authentication and safety MCP uses the API token or the Unix socket; there is no second permission model. Anyone holding the token, or a local user on the socket, can do through MCP what the API allows, including starting work. Give an agent the socket rather than the token when it runs on the same machine, and see [Secure the server](https://sercanatalik.github.io/cereyan/guides/secure-the-server/index.md). ## Let the agent answer questions A flow paused with `wait_for_input` shows its question in `get_run`; the agent answers with `resume_run`. Combine this with a proactive rule on `run.paused` so a human is paged when neither an agent nor a person has answered in time. See [Pause a run for approval](https://sercanatalik.github.io/cereyan/guides/human-approval/index.md). ## Point the agent at the docs The whole site is available as one file at [llms-full.txt](https://sercanatalik.github.io/cereyan/llms-full.txt), with [llms.txt](https://sercanatalik.github.io/cereyan/llms.txt) as the index. Related: [MCP tools, resources and prompts](https://sercanatalik.github.io/cereyan/reference/mcp/index.md), the [agent diagnosis example](https://sercanatalik.github.io/cereyan/examples/agent_diagnosis/index.md). # How to publish artifacts Publish a table, a note, a progress bar, a link, or an image from a run so the result is visible on the run page and, under a key, tracked across runs. ## From a task or a flow ``` from cereyan import flow, task, artifacts @task def validate(rows: list[dict]) -> int: bad = [r for r in rows if r["amount"] < 0] artifacts.create_table(bad, key="bad-rows", columns=["id", "amount"]) artifacts.create_markdown(f"{len(bad)} of {len(rows)} rows rejected") return len(rows) - len(bad) @flow def load(rows: list[dict] | None = None) -> int: rows = rows or [{"id": 1, "amount": 10}, {"id": 2, "amount": -3}] artifacts.create_link("https://example.com/dashboards/loads", text="Load dashboard") return validate(rows) assert load() == 1 ``` Calls inside a task attach the artifact to the task run; calls in the flow body attach it to the run. Each returns the artifact id and raises `CereyanError` outside a run or over 1 MB. ## Report progress ``` from cereyan import flow, task, artifacts @task def process(batches: int) -> None: artifacts.create_progress(0, key="process", label="batches") for i in range(batches): artifacts.update_progress("process", 100 * (i + 1) / batches) @flow def big_load() -> None: process(4) big_load() ``` Each update is a new artifact under the key, so the run page shows the latest value and the Artifacts page keeps the history. ## Embed an image ``` from cereyan import flow, artifacts @flow def chart() -> None: png = b"\x89PNG\r\n\x1a\n" + b"\x00" * 16 # bytes from your plotting library artifacts.create_image(png, key="daily-chart", media_type="image/png") artifacts.create_image("https://example.com/chart.png") chart() ``` Bytes are embedded as a data URI, so keep images small; link large ones by URL. ## Track a value over time Give artifacts that recur a stable `key`. The Artifacts page filters by kind, key, flow, and project, and opening a key shows every value published under it across runs, newest first. Row counts, data-quality scores, and file sizes are good keys. ## Read them programmatically `GET /api/runs/{id}/artifacts`, `Client.artifacts(run_id)`, `GET /api/artifacts?key=...`, the MCP `list_artifacts` tool, and the `cereyan://runs/{id}/artifacts` resource return artifacts as JSON: ``` run = served.client.run("etl", day="2026-02-01") served.wait_run(run["id"]) items = served.client.artifacts(run["id"]) assert isinstance(items, list) ``` Related: [Artifacts](https://sercanatalik.github.io/cereyan/concepts/artifacts/index.md). # How to backfill a date range Create one run per day (or hour, or any step) between two dates, with a concurrency limit, and let targets and `bulk_complete` skip what is already done. Backfills need a running server. ## From the CLI ``` cereyan backfill daily_etl --param day --start 2026-06-01 --end 2026-08-30 --concurrency 4 cereyan backfill hourly --param at --start 2026-09-01T00:00 --end 2026-09-01T23:00 --interval 1h --reverse cereyan backfill daily_etl --param day --start 2026-06-01 --end 2026-06-30 --extra region=eu --json ``` The flow is named as `flow` or `project/flow` when the name exists in several projects. The command prints the backfill id, the number of runs, and the tag; `--json` prints the status object. ## From the UI, the API, and an agent The flow page's **Backfill** button opens a dialog with the same fields. `POST /api/flows/{id}/backfill` and `Client.backfill()` take `parameter`, `start`, `end`, `interval`, `concurrency`, `extra_parameters`, and `reverse`: ``` flow = next(f for f in served.client.flows() if f["name"] == "etl") status = served.client.backfill(flow["id"], "day", "2026-01-01", "2026-01-03", concurrency=2) assert status["total"] == 3 assert status["tag"].startswith("backfill:") runs = served.client.runs(tags=status["tag"])["items"] for run in runs: served.wait_run(run["id"]) assert served.client.backfill_status(status["id"])["total"] == 3 ``` The MCP `backfill` tool does the same and dry-runs by default, so an agent sees how many runs it would create before creating them. ## Skip work that is done Two mechanisms, used together: - A task with `output=` skips itself when its target exists, so a run over a finished day costs one process dispatch and no work. - A flow with `bulk_complete=` never creates runs for finished values: the backfill calls it once with every value and records the returned ones as `Skipped` without dispatching them. ``` from datetime import date from cereyan import flow, task, LocalTarget def finished(values: list[date]) -> set[date]: return {d for d in values if LocalTarget(f"out/{d}.csv").exists()} @task(output=lambda day: LocalTarget(f"out/{day}.csv")) def build(day: date) -> None: with LocalTarget(f"out/{day}.csv").open("w") as fh: fh.write("...") @flow(bulk_complete=finished) def daily_etl(day: date) -> None: build(day) ``` ## Watch and cancel Runs carry the tag `backfill:`; filter the Runs page by it. `GET /api/backfills/{id}` reports counts by state, and `POST /api/backfills/{id}/cancel` (or the Cancel button) cancels the remaining Scheduled and Running runs. The backfill's own resource keeps it to `concurrency` runs at once, and every run still respects the flow's `max_concurrent` and resources. Related: [Backfills](https://sercanatalik.github.io/cereyan/concepts/backfills/index.md), [Make reruns idempotent with targets](https://sercanatalik.github.io/cereyan/guides/idempotent-reruns/index.md). # How to cache task results Persist a task's return value and reuse it when the inputs, the code, or both are unchanged. Caching is for values that come back to the flow; for files, use [targets](https://sercanatalik.github.io/cereyan/guides/idempotent-reruns/index.md). ## Cache by inputs ``` from cereyan import flow, task, INPUTS calls = [] @task(cache=INPUTS, persist_result=True) def lookup(customer: str) -> dict: calls.append(customer) return {"customer": customer, "tier": "gold"} @flow def enrich(customer: str) -> dict: return lookup(customer) assert enrich("acme") == enrich("acme") assert calls == ["acme"] # the second call was Cached enrich("globex") assert calls == ["acme", "globex"] ``` A hit ends the task run `Cached` with the stored value, records `task_run.cached`, and shows in the UI as such. `persist_result=True` is required: the value has to be stored somewhere to be reused. ## Cache by source too `cache=INPUTS + SOURCE` also hashes the task's source code, so editing the function invalidates its entries: ``` from cereyan import flow, task, INPUTS, SOURCE @task(cache=INPUTS + SOURCE, persist_result=True) def transform(rows: int) -> int: return rows * 2 @flow def run(rows: int = 3) -> int: return transform(rows) assert run() == 6 ``` `cache=SOURCE` alone reuses one result for any inputs while the code is unchanged, which suits parameterless setup steps. ## Expire entries ``` from datetime import timedelta from cereyan import flow, task, INPUTS @task(cache=INPUTS, persist_result=True, cache_expires=timedelta(minutes=30)) def rates(currency: str) -> float: return 1.0 @flow def convert(currency: str = "EUR") -> float: return rates(currency) assert convert() == 1.0 ``` An entry older than `cache_expires` is a miss and the task runs again. ## Serialisation Results are pickled by default. Pass `serializer="json"` to store JSON instead when the value is plain data and you want to read it from other tools or the UI. Entries record the Python version that wrote them; a different minor version is a miss. ## Where entries live and how to clear them Entries are files under `/storage/`, keyed by a hash of the task key plus the inputs and source. Retention never touches them. Delete the directory, or the files for one task, to clear the cache; a missing entry is a miss. ## Replay after a pause Tasks with `cache=INPUTS` are what make [pausing for approval](https://sercanatalik.github.io/cereyan/guides/human-approval/index.md) cheap: the resumed attempt replays the flow from the top and the cached tasks return at once. Related: [Targets, caching and results](https://sercanatalik.github.io/cereyan/concepts/targets-caching-results/index.md). # How to chain flows Run one flow after another with `after=`. The server watches for upstream runs to end and creates the downstream run. Dependencies need a running server. ## One upstream ``` from datetime import date from cereyan import App app = App("sales") @app.flow def load_orders(day: date) -> int: return 42 @app.flow(after="load_orders") def build_report(day: date) -> str: return f"report for {day}" assert build_report.after == {"flow": "load_orders", "flows": ["load_orders"], "key": None, "parameters": {}} ``` Every time `load_orders` ends `Completed` or `Skipped`, a `build_report` run is created with the same `day`, `created_by = run:`, and a link to the upstream run in its details. A failed upstream creates nothing. ## Rename or derive parameters Parameters are copied by name. To map them, give a template for each downstream parameter; the context is the same as for rules, with `run`, `flow`, `state`, `payload`, and `parameters`: ``` from datetime import date from cereyan import App app = App("sales2") @app.flow def load_orders(day: date) -> int: return 1 @app.flow(after=("load_orders", {"for_day": "{{ run.parameters.day }}", "source": "'orders'"})) def notify(for_day: date, source: str = "unknown") -> str: return f"{source} {for_day}" assert notify.after["parameters"]["for_day"] == "{{ run.parameters.day }}" ``` ## Fan in: wait for several upstreams When the downstream needs every upstream to have finished the same batch, list them and name the parameter that identifies the batch: ``` from datetime import date from cereyan import App app = App("reporting") @app.flow def sales(day: date) -> None: ... @app.flow def inventory(day: date) -> None: ... @app.flow(after=["sales", "inventory"], batch_key="day") def report(day: date) -> str: return f"report {day}" assert report.after["flows"] == ["sales", "inventory"] and report.after["key"] == "day" ``` `report` runs once per `day`, after both `sales` and `inventory` have a Completed or Skipped run for that day. A day that already has a report never gets a second one, however the first was created, and a failed upstream blocks the batch until it is rerun. Each creation records a `flow.fan_in` event with the key value and the upstream run ids, and the flow page's dependency graph draws one edge per upstream. ## When to use a rule instead `after=` covers "run B when A finishes". For anything conditional, a [rule](https://sercanatalik.github.io/cereyan/guides/rules/index.md) with a `run_flow` action gives you the full match clause: only on failure, only for a tag, only in a project, with templated parameters and guards. ## Check the wiring At start, the server reports an `after=` naming a flow it has not registered as a flow error, visible on the Flows page and in `GET /api/flows`, without stopping other flows. The flow page lists *Triggered by* and *Triggers*. Related: [Dependencies](https://sercanatalik.github.io/cereyan/concepts/dependencies/index.md). # How to run tasks concurrently Call a task to run it now and wait. Submit it to run it in the background and get a `Future`. Map it to submit one task run per element. ## Submit and map ``` from cereyan import flow, task @task def fetch(source: str) -> int: return len(source) @task def combine(sizes: list[int]) -> int: return sum(sizes) @flow def gather() -> int: futures = fetch.map(["orders", "customers", "products"]) return combine([f.result() for f in futures]) assert gather() == 6 + 9 + 8 ``` `submit(*args, **kwargs)` returns a `Future` at once; `result()` blocks until the task run ends and re-raises its exception on failure, `done()` polls, `wait()` blocks without returning the value, and `exception()` returns what was raised. `map(iterable, **static)` submits one run per element and passes the static keyword arguments to each. ## Pass futures between tasks A future given as an argument, at any depth inside lists, tuples, sets, or dicts, is resolved before the receiving task starts, and the dependency is recorded for the timeline: ``` from cereyan import flow, task @task def extract() -> list[int]: return [1, 2, 3] @task def load(rows: list[int]) -> int: return sum(rows) @flow def pipeline() -> int: rows = extract.submit() return load.submit(rows).result() assert pipeline() == 6 ``` Use `wait_for=[future, ...]` on a call or submission to order tasks that share no data: ``` from cereyan import flow, task order = [] @task def first() -> None: order.append("first") @task def second() -> None: order.append("second") @flow def ordered() -> None: f = first.submit() second(wait_for=[f]) ordered() assert order == ["first", "second"] ``` ## Choose a runner The flow's `runner` decides where submitted tasks execute. | Runner | Use for | Notes | | ------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `ThreadRunner(max_workers)` (default) | I/O-bound work: HTTP ([fetching from an API](https://sercanatalik.github.io/cereyan/guides/fetch-from-an-api/index.md)), databases, files | Shares the process; the run context propagates. A task that waits on a child while every worker is busy gets a warning and a temporary extra worker, so nested waits cannot deadlock. | | `ProcessRunner(max_workers)` | CPU-bound work | Each task run is a spawned process; arguments and results must be picklable and the task must be importable from a module (not defined in `__main__` or a notebook). A timeout terminates the worker. | ``` from cereyan import flow, task, ThreadRunner @task def work(i: int) -> int: return i * 2 @flow(runner=ThreadRunner(max_workers=8)) def wide() -> list[int]: return [f.result() for f in work.map(range(10))] assert wide() == [i * 2 for i in range(10)] ``` `max_workers` defaults to the CPU count. Concurrency inside a run is separate from concurrency between runs, which [resources and caps](https://sercanatalik.github.io/cereyan/guides/resources-and-overlap/index.md) govern. ## Timeouts and retries still apply Each submitted task run has its own `retries` and `timeout_seconds`; a failure surfaces when you call `result()`. Related: [Tasks](https://sercanatalik.github.io/cereyan/concepts/tasks/index.md). # How to add custom HTTP routes Register handlers on an App and the server serves them next to its own API: a webhook that starts a run, a health check for your load balancer, a small JSON endpoint over your data. Handlers are FastAPI-shaped functions; routing and the HTTP server are in Rust. ## Register a route ``` from dataclasses import dataclass from cereyan import App, HTTPError, Request, Response import cereyan app = App("intake") @dataclass class Order: order_id: int amount: float @app.get("/api/ext/ping") def ping() -> dict: return {"ok": True} @app.get("/api/ext/orders/{order_id}") def order(order_id: int, verbose: bool = False) -> dict: if order_id < 1: raise HTTPError(404, "no such order") return {"order_id": order_id, "verbose": verbose} @app.post("/api/ext/orders") def receive(order: Order) -> tuple[dict, int]: run = cereyan.client.run("ingest_order", order_id=order.order_id, amount=order.amount) return {"run_id": run["id"]}, 201 @app.get("/api/ext/raw") def raw(request: Request) -> Response: return Response(request.headers.get("user-agent", ""), media_type="text/plain") assert [r.path for r in app.routes] == ["/api/ext/ping", "/api/ext/orders/{order_id}", "/api/ext/orders", "/api/ext/raw"] ``` `@app.get`, `@app.post`, `@app.put`, `@app.patch`, `@app.delete`, and `@app.route(method, path)` register a handler. The routes are served when the module is served with `cereyan serve` or `app.serve()`. ## How arguments bind | Parameter | Bound from | | ----------------------------------------- | ------------------------------------------------------------------------------------------------ | | Name in the path template (`{order_id}`) | The path segment, coerced through the type hint | | Other scalar parameter | The query string; `list[T]` collects repeated keys; missing ones use the default or answer 422 | | Dataclass, `TypedDict`, or pydantic model | The JSON body | | Annotated `Request` | The raw request: `method`, `path`, `path_params`, `query`, `headers`, `body`, `json()`, `text()` | Coercion follows the same rules as flow parameters; a value that does not coerce answers 422 with the reason. ## What to return | Return | Response | | --------------------------------------------- | -------------------------- | | `dict` or `list` | JSON, 200 | | `(value, status)` | The value with that status | | `str` | `text/plain` | | `bytes` | `application/octet-stream` | | `Response(body, status, headers, media_type)` | As given | | `None` | JSON `null` | Raise `HTTPError(status, message)` for an error; an unhandled exception answers 500 with the traceback in the server log. ## Call it ``` import json, urllib.request with urllib.request.urlopen(served.url + "/api/ext/ping") as resp: assert json.loads(resp.read()) == {"ok": True} body = json.dumps({"order_id": 7, "amount": 99.5}).encode() req = urllib.request.Request(served.url + "/api/ext/orders", data=body, method="POST", headers={"content-type": "application/json"}) with urllib.request.urlopen(req) as resp: assert resp.status == 201 run_id = json.loads(resp.read())["run_id"] assert served.wait_run(run_id)["state"]["type"] == "Completed" ``` ## Async handlers `async def` handlers are supported and run on one shared event loop, so a blocking call inside one blocks every other async handler. Keep them non-blocking, or use a sync handler, which runs on its own thread. An async handler that takes longer than 30 seconds answers 504. ``` from cereyan import App app = App("async-intake") @app.get("/api/ext/status") async def status() -> dict: return {"ok": True} ``` ## Where routes appear The Settings page lists every registered route with its source. Routes under `/api/` are protected by the API token like the built-in API; routes elsewhere are open. Starting runs from a route goes through `cereyan.client`, so the run records `created_by = client`. Related: [Secure the server](https://sercanatalik.github.io/cereyan/guides/secure-the-server/index.md), the [webhook route example](https://sercanatalik.github.io/cereyan/examples/webhook_route/index.md). # How to detect when something did not happen A reactive rule fires when an event arrives. A proactive rule fires when an expected event does *not* arrive: the nightly load that never finished, the run that has been going for two hours, the daily file that was not produced by nine. Add `unless` to a rule and arm it from an event or from the clock. ## Event-armed: a deadline after something starts ``` from cereyan import App app = App("loads") @app.rule(on="run.running", flow="etl", unless="run.completed", within=2 * 3600) def etl_overran(event, run): print(f"{run['name']} did not complete within two hours") spec = app.rules[0].spec() assert spec["unless"]["events"] == ["run.completed"] and spec["within"] == 7200.0 ``` `on` arms an expectation when the run starts, `unless` names the event that disarms it, and `within` is the deadline in seconds. The expectation is keyed by the run (or by the flow, for events without a run). If the run completes in time the expectation is met; if it fails, or is still running at the deadline, the rule fires. Expectations are stored, so one whose deadline passed while the server was down fires once on start. ## Clock-armed: something should have happened by now ``` from cereyan import App app = App("loads2") @app.rule(at="0 9 * * *", tz="Europe/Istanbul", flow="daily_load", unless="run.completed") def daily_load_missing(event, run): print("no daily_load completed before nine") spec = app.rules[0].spec() assert spec["at"] == {"cron": "0 9 * * *", "tz": "Europe/Istanbul"} ``` At each tick of `at` the rule fires unless a matching event occurred in the look-back window: `within` seconds when given, otherwise the time since the previous tick. Ticks missed while the server was down are skipped with a log line. Clock-armed rules need a running server. A rule does not fire until it has been watching for a whole window: until then its look-back would reach past its own creation, where nothing had happened yet, and the first tick after you save a rule would page you about it. ## What the rule sees A lapse is recorded as an `expectation.lapsed` event with resource `rule/`, the related run and flow, and a payload carrying `rule`, `flow`, `project`, `run`, `run_name`, `expected`, `deadline`, and `armed_at`. The rule's actions execute against that event, so templates can say: ``` {{ event.payload.expected[0] }} did not happen for {{ flow.name }} by {{ event.payload.deadline }} ``` All guards apply: `once="per_run"` keys on the arming run, and runs created by a lapse's `run_flow` action never re-trigger the same rule unless `allow_self` is set. ## In the UI and the API The rule form's **Unless** section has the same fields (`unless`, `within`, `at`, `tz`); validation rejects `unless` without either `within` or `at`. The rule page lists open expectations, `GET /api/rules/{id}/expectations` returns them (`?open=false` for history), and **Test** renders a synthetic lapse for a proactive rule. ## Offline Event-armed expectations are evaluated when the run ends: a run that overruns `within` fires the lapse at completion, provided the rule was registered by a server earlier. Clock-armed rules only work with a server. Related: [React to events with rules](https://sercanatalik.github.io/cereyan/guides/rules/index.md), [Events](https://sercanatalik.github.io/cereyan/reference/events/index.md). # How to fetch from an HTTP API Build the client once, submit one task run per request, and let the runner fetch them concurrently. ## Build the client once An engine imports your module once and serves many runs from that import, so a client built at module level keeps its connection pool and TLS sessions across every run that engine serves. Building one inside the task body pays a fresh handshake on every call. The blocks below fetch from a small local server so the page can be tested; in your own pipeline `BASE` is the API you call. ``` import http.server, json, threading, urllib.request from cereyan import flow, task class Handler(http.server.BaseHTTPRequestHandler): def do_GET(self): body = json.dumps({"path": self.path}).encode() self.send_response(200) self.send_header("content-length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, *args): pass api = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) threading.Thread(target=api.serve_forever, daemon=True).start() BASE = f"http://127.0.0.1:{api.server_port}" opener = urllib.request.build_opener() # built once per import, not once per run @task(retries=2, timeout_seconds=10) def fetch(path: str) -> dict: with opener.open(BASE + path, timeout=10) as resp: return json.loads(resp.read()) @flow def one() -> dict: return fetch("/orders") assert one() == {"path": "/orders"} ``` `httpx` and `requests` work the same way: build the `Client` or `Session` at module level and call it from the task body. Both are safe to share across the threads a `ThreadRunner` uses. ## Fetch concurrently `map` submits one task run per element. Each one is recorded, retried and timed on its own, and the timeline shows them side by side. ``` from cereyan import ThreadRunner @flow(runner=ThreadRunner(max_workers=4)) def collect() -> list[str]: futures = fetch.map(["/orders", "/customers", "/products"]) return [f.result()["path"] for f in futures] assert collect() == ["/orders", "/customers", "/products"] ``` `ThreadRunner` is the default and suits HTTP, because the calls release the GIL while they wait. Raise `max_workers` for more requests in flight; it defaults to the CPU count. [Run tasks concurrently](https://sercanatalik.github.io/cereyan/guides/concurrent-tasks/index.md) covers futures, `wait_for` and runners in full. ## Retry and time out `retries` and `timeout_seconds` on `@task` apply per request, so one slow endpoint does not fail the whole run. A task run that passes `timeout_seconds` ends Failed with sub-state `TimedOut`; under a `ThreadRunner` the request itself may still be in flight, so set a timeout on the client as well, as the blocks above do. ## Async clients are not supported `@flow` and `@task` reject `async def`, with or without `yield`. Bodies run synchronously, so the coroutine would never be awaited and the run would be recorded as having succeeded without fetching anything. Drive the loop yourself: ``` import asyncio from cereyan import task async def _fetch(path: str) -> str: await asyncio.sleep(0) return path @task def fetch_one(path: str) -> str: return asyncio.run(_fetch(path)) assert fetch_one("/orders") == "/orders" ``` Each `asyncio.run` gets its own event loop, so an async client cannot be shared between task runs the way a synchronous one can. `async def` route handlers are unaffected; see [custom routes](https://sercanatalik.github.io/cereyan/guides/custom-routes/index.md). ## When the client is not reused | Situation | What happens | | ---------------------- | ------------------------------------------------------- | | `@flow(isolated=True)` | A fresh process and a fresh import for every run | | `ProcessRunner` | Every task run is a spawned process with its own import | | `python pipeline.py` | One process, one run: reuse within the run only | Related: [Run tasks concurrently](https://sercanatalik.github.io/cereyan/guides/concurrent-tasks/index.md), [Engines and the home directory](https://sercanatalik.github.io/cereyan/concepts/engines-and-home/index.md). # How to pause a run for approval Call `wait_for_input` where the flow needs a decision. The run pauses with the question visible in the UI and the API, the engine is released, and the answer resumes it. Pausing needs a running server; offline, the question is asked on the terminal. ## Ask the question ``` from datetime import date from cereyan import flow, task, wait_for_input, INPUTS @task(cache=INPUTS, persist_result=True) def prepare(day: date) -> int: return 1200 @task def release(rows: int) -> str: return f"released {rows}" @flow def publish(day: date) -> str: rows = prepare(day) decision = wait_for_input( f"Release {rows} rows for {day}?", schema={"type": "object", "properties": {"approve": {"type": "boolean"}}, "required": ["approve"]}, ) return release(rows) if decision["approve"] else "held" ``` The first call to `wait_for_input` moves the run to `Paused` with the prompt and schema in its state details, records `run.paused`, and ends the attempt. Mark the tasks before the question with `cache=INPUTS` so the resumed attempt does not redo their work. ## Answer it On the run page, the Details tab shows the question with a form built from the schema and a **Resume** button. Programmatically: ``` run = served.client.run("publish", day="2026-03-01") paused = served.wait_run(run["id"], until=lambda r: r["state"]["type"] == "Paused") assert "Release" in paused["state"]["details"]["prompt"] served.client.resume(run["id"], {"approve": True}) done = served.wait_run(run["id"]) assert done["state"]["type"] == "Completed" ``` `POST /api/runs/{id}/resume` with `{"input": ...}`, `Client.resume`, and the MCP `resume_run` tool do the same. `GET /api/runs/{id}/input` returns the stored answer. Resuming a run that is not paused answers 409. ## What happens on resume The answer is stored, `run.resumed` is recorded, and a new attempt of the same run is scheduled. It reruns the flow from the top: tasks with `cache=INPUTS` return their cached results as `Cached` task runs, `wait_for_input` returns the answer instead of pausing, and the rest of the flow executes. Any code between the top of the flow and the question that is not in a cached task runs again, so keep side effects inside tasks. ## While it waits A paused run counts as active, appears on the dashboard, can be cancelled, and holds no engine and no resources. Use a proactive rule to notice a run that waits too long: ``` from cereyan import App app = App("approvals") @app.rule(on="run.paused", flow="publish", unless="run.resumed", within=4 * 3600) def nobody_answered(event, run): print(f"{run['name']} has been waiting for four hours") ``` ## Several questions Each `wait_for_input` call in a flow pauses in turn; the stored answer belongs to the attempt that asked, so a flow can ask, resume, and ask again. ## Offline Without a server, `wait_for_input` reads the answer from the terminal (as JSON when a schema is given) and raises `CereyanError` when there is no interactive terminal, so scripts under cron fail fast instead of hanging. Related: [Use cereyan with an AI agent](https://sercanatalik.github.io/cereyan/guides/agents/index.md) for answering questions from an agent, [Cache task results](https://sercanatalik.github.io/cereyan/guides/cache-results/index.md). # How to make reruns idempotent with targets Declare where a task writes, and cereyan skips the task when that output already exists. Rerunning a day, a failed run, or a whole backfill then only does the missing work. ## Declare a target ``` from datetime import date from cereyan import flow, task, LocalTarget writes = [] @task(output=lambda day: LocalTarget(f"data/{day}.csv")) def export(day: date) -> None: writes.append(day) with LocalTarget(f"data/{day}.csv").open("w") as fh: fh.write("id,amount\n1,10\n") @flow def daily(day: date) -> None: export(day) daily(date(2026, 3, 1)) daily(date(2026, 3, 1)) assert writes == [date(2026, 3, 1)] # the second run skipped export ``` `output=` takes a Target, or a callable over the task's arguments returning one. Only the parameters named in the callable's signature are passed, so `lambda day: ...` works for a task with more arguments. When `exists()` is true the task run ends `Skipped` with the message `output exists`, records `task_run.skipped`, and the body never runs. ## Write atomically `LocalTarget.open("w")` writes to a temporary file next to the target and renames it into place on close. A crash mid-write leaves no partial file, so the next run does not mistake half an output for a finished one. Use `temporary_path()` when a library insists on writing a path itself: ``` from cereyan import LocalTarget target = LocalTarget("data/report.txt") with target.temporary_path() as tmp: with open(tmp, "w") as fh: fh.write("done") assert target.exists() ``` ## Custom targets Anything with an `exists()` method is a target: a row in a table, a key in an object store, a flag in an API. ``` from datetime import date from cereyan import flow, task class TableTarget: done: set = set() def __init__(self, name: str) -> None: self.name = name def exists(self) -> bool: return self.name in TableTarget.done @task(output=lambda day: TableTarget(f"sales_{day}")) def build(day: date) -> None: TableTarget.done.add(f"sales_{day}") @flow def nightly(day: date) -> None: build(day) nightly(date(2026, 3, 2)) nightly(date(2026, 3, 2)) ``` ## Skip whole runs Targets skip tasks, not runs. To avoid scheduling runs for work already done, give the flow `bulk_complete=` and let [backfills](https://sercanatalik.github.io/cereyan/guides/backfill/index.md) skip those values before they are created. ## Rerun on purpose To recompute, delete the target (`LocalTarget.remove()`) and run again; there is no force flag, because the file is the truth. Related: [Targets, caching and results](https://sercanatalik.github.io/cereyan/concepts/targets-caching-results/index.md), [Cache task results](https://sercanatalik.github.io/cereyan/guides/cache-results/index.md). # How to migrate from Prefect or Luigi Cereyan borrows its vocabulary from Prefect and its file-oriented idempotency from Luigi, so most concepts map directly. The tables below say what to reach for and what has no equivalent, with the reasons on the [Design and limitations](https://sercanatalik.github.io/cereyan/design/limitations/index.md) page. ## From Prefect | Prefect | cereyan | Notes | | --------------------------------------- | -------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------- | | `@flow`, `@task` | `@flow`, `@task` | Same shape. Parameters are typed by hints; pydantic is not used. | | Deployment | A served flow with `schedule=` | There is no deployment object. `cereyan serve dir/` registers every flow in the directory; schedules live on the flow. | | Work pool, worker | Engine pool | The server runs engine processes itself; there are no remote workers. | | `prefect.yaml`, profiles, settings | `cereyan.toml`, `CEREYAN_*` | One file next to the flows. | | Blocks, secrets | Variables with `secret=True` | Small JSON values, encrypted locally. No typed blocks, no cloud secret stores. | | Automations, triggers | Rules (`when` / `do`) and `unless` | Same reactive model plus proactive rules; templates use the same Jinja syntax. | | Artifacts | Artifacts | Markdown, table, progress, link, image; keyed history across runs. | | Results, caching | `persist_result`, `cache=INPUTS`, `SOURCE` | Local storage only. | | Task runners | `ThreadRunner`, `ProcessRunner` | No Dask or Ray. | | `pause_flow_run`, `wait_for_input` | `wait_for_input` | Same idea; the resumed attempt replays cached tasks. | | States | Same names, plus `Cancelling` and named sub-states | The transition rules are on the [states page](https://sercanatalik.github.io/cereyan/reference/states/index.md). | | Events | Events | Names are `run.completed` rather than `prefect.flow-run.Completed`. | | Flow run retries, `retry_delay_seconds` | `retries`, `retry_delay` | `exponential(...)` replaces `exponential_backoff`. | | Global concurrency limits, tag limits | Resources, `max_concurrent` | Named semaphores declared in configuration. | | Prefect Cloud, workspaces, RBAC, SSO | none | One machine, one user level, one token. | | Assets, SLAs, incident management | none | Not planned. | | `prefect-*` integration packages | none | Use the library directly inside tasks. | | Prefect MCP server | Built-in MCP server | `cereyan mcp` for stdio hosts, `POST /mcp` over HTTP. | What changes in practice: delete the deployment step and the worker, keep the decorators, move settings into `cereyan.toml`, and replace `Secret.load` with `Variable.get`. Flows that used `prefect.runtime` read the run through `get_run_logger()` and the hooks' `run` argument. ## From Luigi | Luigi | cereyan | Notes | | --------------------------------------------- | ------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------ | | `luigi.Task` with `requires`, `output`, `run` | `@task(output=...)` called from a `@flow` | Ordering is the order of calls in the flow body instead of a `requires` graph; the target still decides whether the task runs. | | `luigi.Target`, `LocalTarget` | `Target`, `LocalTarget` | Same protocol: anything with `exists()`. Atomic writes are built in. No HDFS, S3, or database targets. | | `luigi.Parameter`, `DateParameter` | Type hints | `day: date` instead of `luigi.DateParameter()`. | | `luigi --module x Task --param` | `cereyan run x.py:flow --param name=value` | | | `luigi.build([...])` | Call the flow | A flow is a function. | | Central scheduler (`luigid`) | `cereyan serve` | Also the UI, the API, the scheduler, rules, and MCP. | | Workers, `--workers N` | Engine pool, `max_engines` | Managed by the server. | | `RangeDaily`, `RangeHourly` | Backfills | `cereyan backfill flow --param day --start ... --end ...` | | `bulk_complete` | `bulk_complete=` on the flow | Same purpose: skip values already done before scheduling. | | `resources` | `resources=` | Named semaphores; totals in `cereyan.toml` instead of `luigi.cfg`. | | `retry_count`, `retry_delay` | `retries`, `retry_delay` | Per task or per flow. | | `luigi.cfg` | `cereyan.toml` | | | Event handlers (`@Task.event_handler`) | Hooks (`on_failure=`) and rules | Hooks for in-process reactions, rules for everything else. | | Visualiser | The UI | Runs, flows, timeline, events, artifacts. | | Cron to trigger recurring work | Schedules | Built in; catch-up policies replace "run the task for every missed date". | | Task history database | The store | SQLite, with retention for logs and events. | What changes in practice: turn each `requires` chain into a flow that calls the tasks in order (or several flows chained with `after=`), keep the `output` targets, replace parameters with type hints, and move `luigi.cfg` resources and retry settings into `cereyan.toml` and the decorators. Related: [Design and limitations](https://sercanatalik.github.io/cereyan/design/limitations/index.md), [Concepts](https://sercanatalik.github.io/cereyan/concepts/app-and-projects/index.md). # How to limit concurrency and overlap Three controls, from narrow to wide: a cap on one flow, named resources shared by many, and the engine pool. ## Cap one flow ``` from cereyan import flow @flow(max_concurrent=1, on_overlap="skip") def every_ten_minutes() -> None: ... @flow(max_concurrent=2) def exports() -> None: ... assert every_ten_minutes.options["on_overlap"] == "skip" ``` `max_concurrent` limits how many runs of the flow are Pending or Running at once. When the cap is reached, `on_overlap` decides: `enqueue` (default) waits as `AwaitingResource`, `skip` ends the new run `Skipped`, `cancel_new` ends it `Cancelled`. Pick `skip` for polling flows where a missed tick does not matter and `enqueue` for flows that must not lose work. A run that finds no slot is still created for its tick, so every scheduled time appears in history. Under `skip` and `cancel_new` it ends at once with the message "previous run still active" (and a `run.skipped` event for `skip`); under `enqueue` it stays Scheduled, turns Late once its time passes, and starts in scheduled-time order as soon as a run of the flow ends. The reference for this behaviour is the overlap soak, `just soak` in [Contributing](https://sercanatalik.github.io/cereyan/contributing/index.md), which runs fifteen such flows for an hour and checks that caps hold, controls never skip, and queues drain in order. ## Share a resource Declare totals in `cereyan.toml` or on the Settings page, and claim units on flows or tasks: ``` [resources] db = 4 gpu = 1 ``` ``` from cereyan import flow, task @task(resources={"db": 1}) def query(sql: str) -> str: return sql @flow(resources={"gpu": 1}) def train() -> str: return query("select 1") assert train() == "select 1" ``` A run or task run waits as `AwaitingResource` until every resource it declares has capacity, and a `resource.exhausted` event records the first wait. Units are released on every terminal state, on pause, and when an engine dies. Resources are global to the machine, so two projects declaring `db` share the same four units. A resource that is not declared in configuration has no limit. ## Order the queue `priority` (higher first) decides which waiting run gets the next free engine or unit. It never preempts a running run. A negative priority also lowers the engine's OS scheduling priority on Linux and macOS, useful for background reprocessing: ``` from cereyan import flow @flow(priority=-10) def reprocess_history() -> None: ... ``` ## Bound the pool `max_engines` in `[server]` or `--max-engines` caps engine processes, and therefore runs executing at once, machine-wide. The default is the CPU count. ## Stop a failing flow ``` from cereyan import flow @flow(disable_after=(3, 3600, 86400)) def nightly() -> None: ... ``` Three failures within an hour pause the flow's schedules for a day, with `flow.disabled` and `flow.enabled` events marking the window. Manual runs still work. ## Check what is waiting The Runs page filter `AwaitingResource`, the run's Details tab (which names the resource), and `GET /api/counts` show what is blocked. The Settings page shows each resource's total and in-use count. Related: [Resources and concurrency](https://sercanatalik.github.io/cereyan/concepts/resources-and-concurrency/index.md), [Schedule a flow](https://sercanatalik.github.io/cereyan/guides/schedule-a-flow/index.md). # How to retry, time out, and survive crashes Failures come in three shapes: the code raised, the code ran too long, or the process died. Each has its own option. ## Retry a task or a flow Set `retries` on the task. The task run goes through `AwaitingRetry`, waits, then executes again as `Retrying`; only after the last retry does it fail. ``` from cereyan import flow, task, exponential calls = [] @task(retries=3, retry_delay=exponential(base=0.01, jitter=0.01, maximum=1)) def fetch() -> str: calls.append(1) if len(calls) < 3: raise ConnectionError("try again") return "data" @flow def sync() -> str: return fetch() assert sync() == "data" assert len(calls) == 3 ``` `retry_delay` takes seconds, a list of per-attempt seconds (`[1, 5, 30]`, the last value repeats), or `exponential(base, jitter, maximum)` for `base * 2**attempt` plus jitter, capped at `maximum`. `retries` on a flow retries the whole run: a new attempt of the same run, with `failure_count` increased. Tasks that already completed run again unless they are cached or have a target. ``` from cereyan import flow, task seen = [] @flow(retries=1, retry_delay=0) def whole_run() -> int: seen.append(1) if len(seen) == 1: raise RuntimeError("transient") return len(seen) assert whole_run() == 2 ``` ## Time out `timeout_seconds` on a task fails the task run as `TimedOut` after the limit; on a flow it fails the run. A timed-out task still counts toward `retries`, so `retries=2, timeout_seconds=30` gives three attempts of thirty seconds each. ``` import time from cereyan import flow, task @task(timeout_seconds=0.2) def slow() -> None: time.sleep(5) @flow def limited() -> None: slow() try: limited() except TimeoutError as exc: assert "exceeded" in str(exc) ``` With the default thread runner a timed-out task's thread is abandoned; with a `ProcessRunner` the worker process is terminated. A flow's timeout is enforced by the server, which records `TimedOut` and ends the engine, and offline by an alarm in the running process. That alarm does not exist on Windows: a flow run by `python pipeline.py` or `cereyan run` there is not timed out at all, and `timeout_seconds` on it is accepted and ignored. Run the flow through `cereyan serve` for a timeout that holds on every platform. Task timeouts are unaffected. ## Survive a crash A **crash** is the engine process dying while a run executes: an out-of-memory kill, a segfault in a C extension, a machine reboot. The server notices through missed heartbeats and a dead PID, marks the run `Crashed`, and reruns it up to `crash_retries` times (the decorator, then `[defaults] crash_retries` in `cereyan.toml`, then 5) before it stays `Failed`. Crash reruns carry `created_by = crash:`. ``` from cereyan import flow @flow(crash_retries=2) def fragile() -> None: ... ``` A crash is not a failure of your code, so it does not consume `retries`. Design flows so a rerun is safe: write outputs through [targets](https://sercanatalik.github.io/cereyan/concepts/targets-caching-results/index.md) and cache expensive steps, and a crashed run picks up where the files say it left off. ## See what happened The run page's Details tab shows every state with its message; the failing exception's type, message, and traceback are in the state details and the log. The MCP `explain_failure` tool and the `diagnose_run` prompt collect the same for an agent. Related: [Run code on state changes](https://sercanatalik.github.io/cereyan/guides/state-hooks/index.md), [Runs and states](https://sercanatalik.github.io/cereyan/concepts/runs-and-states/index.md). # How to react to events with rules A rule matches events and runs actions: start a flow, cancel a run, force a state, pause a schedule, call a webhook, send an email, or call your function. Data rules are created in the UI or through the API and edited at runtime; code rules live next to the flows. ## Create a data rule On the Rules page, **New rule** opens the form. The same body goes to `POST /api/rules`: ``` rule = served.client.create_rule({ "name": "cleanup after etl", "when": {"events": ["run.*"], "flows": ["etl"], "states": ["Completed"]}, "do": [{"kind": "run_flow", "flow": "etl", "parameters": {"day": "{{ run.parameters.day }}"}}], "once": "per_run", "cooldown_seconds": 0, "max_per_minute": 60, }) assert rule["name"] == "cleanup after etl" assert any(r["id"] == rule["id"] for r in served.client.rules()) ``` `when` takes `events` (names or prefixes such as `run.*`), `flows`, `tags`, `states`, and `project`; a rule fires for an event that matches every clause it sets. A `states` value matches the run's state type or its sub-state name, so `["Completed"]` above also covers a run that ended `Skipped`, and `["Skipped"]` would narrow to just those. Event and state names are checked against the [catalogue](https://sercanatalik.github.io/cereyan/reference/events/index.md) before the rule is stored. A name under one of the engine's prefixes (`run.`, `task_run.`, `flow.`, `schedule.`, `resource.`, `rule.`, `expectation.`) that nothing emits is refused with the nearest match, so a rule cannot be saved dead: ``` from cereyan.client import ApiError try: served.client.create_rule({ "name": "typo", "when": {"events": ["run.failure"]}, "do": [{"kind": "cancel_run"}], }) except ApiError as exc: assert "run.failed" in str(exc) ``` Any name outside those prefixes is a custom event of yours and is accepted as typed. ## Actions | Kind | Fields | Effect | | ----------------------------------- | ---------------------------------------------- | ------------------------------------------------- | | `run_flow` | `flow`, `parameters` (templated) | Create a run; it records `created_by = rule:` | | `cancel_run` | | Cancel the run the event is about | | `set_state` | `state_type`, `message` | Force the run into a state | | `pause_schedule`, `resume_schedule` | `schedule_id` | Stop or restart a schedule | | `webhook` | `url`, `method`, `headers`, `body` (templated) | HTTP request; three attempts with backoff | | `email` | `to`, `subject`, `body` (templated) | Sent through `[email]` in `cereyan.toml` | | `call` | `callable` | A code rule's function | Actions run in order; a failing action is recorded as `rule.action.failed` and the rest still run. ## Templates Fields marked templated use Jinja syntax, rendered in Rust, with `event`, `run`, `flow`, `state`, `payload`, and `parameters` in scope: ``` {{ flow.name }} run {{ run.name }} ended {{ state.type }}: {{ state.message }} {{ run.parameters.day }} {{ payload.table }} ``` An undefined variable is an error for that action only. **Test** on the rule page (`POST /api/rules/{id}/test`) renders the templates against the most recent matching event without executing anything. ## Notify by webhook or email ``` { "name": "page on nightly failure", "when": {"events": ["run.failed"], "flows": ["nightly"]}, "do": [ {"kind": "webhook", "url": "https://hooks.example.com/pager", "method": "POST", "body": "{\"text\": \"{{ flow.name }} failed: {{ state.message }}\"}"}, {"kind": "email", "to": "oncall@example.com", "subject": "{{ flow.name }} failed", "body": "Run {{ run.name }} failed with {{ state.message }}."} ], "once": "per_run" } ``` Email needs `[email]` configured; see [Configuration](https://sercanatalik.github.io/cereyan/reference/configuration/index.md). ## Write a code rule ``` from cereyan import App, events app = App("ops") @app.rule(on=events.run.failed, flow="nightly", once="per_run", cooldown_seconds=60) def on_nightly_failure(event, run): print(f"{run['name']} failed: {event['payload'].get('message')}") assert app.rules[0].spec()["when"]["flows"] == ["nightly"] assert app.rules[0].spec()["when"]["events"] == ["run.failed"] ``` `events.run.failed` is the string `"run.failed"`, so the constants and the literals are interchangeable; the constants just autocomplete and catch a typo at the point you write it. `events.run.any` is `"run.*"`, and `cereyan.states` does the same for `states=`. The function receives the event and the run as dicts and its return value is recorded with the firing. Code rules are re-registered on every server start, show read-only on the Rules page with a `code` badge, and fire on the offline path for a script's own events once a server has registered them. Guards are keyword arguments: `once`, `cooldown_seconds`, `max_per_minute`, `allow_self`, and `name` — anything else raises, so a misspelled guard cannot silently leave the rule on its defaults. ## Guards - `once="per_run"` (the default) fires at most once per run, so a run that retries three times alerts once. `once="never"` lifts the limit and fires on every matching event. - `cooldown_seconds` and `max_per_minute` throttle noisy rules. - A rule never fires on events of runs it created, unless `allow_self` is set; this is what stops a `run_flow` rule looping. - Disabled rules never fire. Toggle them on the Rules page or with `PATCH /api/rules/{id}`. A rule whose names are all valid can still match nothing — the wrong flow, a tag that is never set. The Rules page marks any rule that has never fired, which is the one thing name checking cannot tell you. ## See what fired The rule page lists firings with each action's outcome; `GET /api/rules/{id}/firings` returns them. Every firing also records `rule.fired` and per-action `rule.action.completed` or `rule.action.failed` events. Related: [Events and rules](https://sercanatalik.github.io/cereyan/concepts/events-and-rules/index.md), [Detect when something did not happen](https://sercanatalik.github.io/cereyan/guides/detect-missing-events/index.md). # How to run the server as a service Run `cereyan serve` under the operating system's service manager so schedules fire after a reboot and the server restarts if it dies. Engines survive a server restart: on start the supervisor adopts runs whose engine is still alive and marks the rest crashed and reruns them. ## What the service needs - A working directory containing the flows (and `cereyan.toml`), passed to `cereyan serve`. - `CEREYAN_HOME` set explicitly, so the home does not depend on which user's `~` is in effect. - `CEREYAN_NO_BROWSER=1` or `--no-open`, since there is no display. - The token in the environment if the API is protected. - `Restart=always` or the equivalent; the server is safe to restart at any time. ## systemd (Linux) `/etc/systemd/system/cereyan.service`: ``` [Unit] Description=cereyan pipeline server After=network.target [Service] User=pipelines WorkingDirectory=/srv/pipelines Environment=CEREYAN_HOME=/var/lib/cereyan Environment=CEREYAN_NO_BROWSER=1 EnvironmentFile=-/etc/cereyan/env ExecStart=/srv/pipelines/.venv/bin/cereyan serve /srv/pipelines --host 127.0.0.1 --port 4200 Restart=always RestartSec=2 KillSignal=SIGTERM TimeoutStopSec=60 [Install] WantedBy=multi-user.target ``` ``` sudo systemctl daemon-reload sudo systemctl enable --now cereyan journalctl -u cereyan -f ``` Put `CEREYAN_TOKEN=...` in `/etc/cereyan/env` (mode 0600) rather than in the unit file. `TimeoutStopSec` should exceed `cancel_grace_secs` twice over, so a stop lets running work finish or be cancelled cleanly. ## launchd (macOS) `~/Library/LaunchAgents/xyz.helixio.cereyan.plist`: ``` Labelxyz.helixio.cereyan ProgramArguments /Users/me/pipelines/.venv/bin/cereyan serve /Users/me/pipelines --no-open WorkingDirectory/Users/me/pipelines EnvironmentVariables CEREYAN_HOME/Users/me/.cereyan RunAtLoad KeepAlive StandardOutPath/Users/me/Library/Logs/cereyan.log StandardErrorPath/Users/me/Library/Logs/cereyan.log ``` ``` launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/xyz.helixio.cereyan.plist launchctl kickstart -k gui/$(id -u)/xyz.helixio.cereyan # restart after editing flows' dependencies tail -f ~/Library/Logs/cereyan.log ``` ## Windows Use Task Scheduler with a task that runs at logon or at startup, action `cereyan.exe serve C:\pipelines --no-open`, and "restart if the task fails". Set `CEREYAN_HOME` in the task's environment or system-wide. ## Logs The server logs to standard error: one line per start with the address and auth state, warnings for unknown configuration keys, and errors from engines that fail to import. Run logs are in the database, not in the service log; read them in the UI, with `cereyan runs ls`, or through the API. ## Upgrading Stop the service, install the new wheel into the same environment, start it. Migrations run on the first open; downgrades are noted in the [changelog](https://sercanatalik.github.io/cereyan/changelog/index.md) when they need care. ## Picking up code changes Engines are recycled when their module file changes, so editing a flow's module takes effect on the next run without a restart. Adding a new module, changing `cereyan.toml`, or changing a flow's schedule declaration needs a restart. Related: [Engines and the home directory](https://sercanatalik.github.io/cereyan/concepts/engines-and-home/index.md), [Secure the server](https://sercanatalik.github.io/cereyan/guides/secure-the-server/index.md). # How to schedule a flow Add `schedule=` to the flow and serve it. The server creates runs at each fire time; nothing fires offline. ## Declare the schedule in code ``` from datetime import date, timedelta from cereyan import flow, Cron, Interval @flow(schedule=Cron("30 6 * * *", timezone="Europe/Istanbul")) def morning_load(day: date = date.today()) -> str: return str(day) @flow(schedule=Interval(timedelta(minutes=15)), max_concurrent=1, on_overlap="skip") def poll_queue() -> None: ... assert morning_load.schedules[0]["kind"] == "cron" assert poll_queue.schedules[0]["interval"] == 900 ``` Then: ``` cereyan serve pipelines/ ``` The flow page shows the schedule summary and next fire time, the **Upcoming** tab lists the runs materialised ahead, and the dashboard's Upcoming panel shows the next fires across flows. ## Pick the kind | Want | Declare | | -------------------------- | --------------------------------------------------------------------------------- | | Wall-clock times | `Cron("0 9 * * 1-5", timezone="Europe/Istanbul")` | | Every N seconds or minutes | `Interval(timedelta(minutes=15))`; add `anchor=datetime(...)` to align the grid | | Calendar rules | `RRule("DTSTART:20260101T090000\nRRULE:FREQ=MONTHLY;BYDAY=-1FR", timezone="UTC")` | Several schedules can be given as `schedules=[...]`. Each takes `catchup` (`skip`, `latest`, `all`) and `catchup_max` for what happens to fires missed while the server was down. ## Keep runs from piling up A slow flow on a fast schedule needs a policy. `max_concurrent=1` with `on_overlap="skip"` drops a fire while the previous run is still going; `"enqueue"` (the default) lets it wait; `"cancel_new"` records it as cancelled. See [Limit concurrency and overlap](https://sercanatalik.github.io/cereyan/guides/resources-and-overlap/index.md). ## Pause and resume Pause from the flow page, `POST /api/schedules/{id}/pause`, the MCP `pause_schedule` tool, or a rule's `pause_schedule` action. Pausing removes the schedule's not-yet-started runs; resuming materialises them again. `disable_after=(count, window, persist)` pauses automatically after repeated failures. ## Skip a run To stop one run without pausing the schedule, open the flow's menu on the Flows page: **Skip next run** skips the next fire, and **Skip runs…** opens a checklist of upcoming fires filled by a *Skip the next N* stepper or an *Until a time* field, with the time the schedule resumes and the flows after this one that are skipped too. The flow page's **Upcoming** tab does the same row by row, for several rows at once, or for all of them with the header checkbox, and lists fires past the materialised runs as projected so you can skip one days ahead. A skipped fire shows who skipped it and when; at its time it ends `Skipped` without starting, and so do the runs of the flows declared `after=` it. **Undo** takes a skip back until its time. See [Skipping fires](https://sercanatalik.github.io/cereyan/concepts/schedules/#skipping-fires). ## Inspect through the API ``` flow = next(f for f in served.client.flows() if f["name"] == "daily_etl") schedules = served.client.schedules(flow["id"]) assert schedules and schedules[0]["schedule"]["kind"] == "cron" upcoming = served.client.upcoming(flow["id"]) assert len(upcoming) >= 1 ``` ## Edit at runtime **Reschedule…**, in the Flows page menu and beside the schedule summary on the flow page, edits a cron schedule as a daily, weekly, or monthly time in a timezone, or as raw cron, with its catch-up options, and shows the coming week before and after the change with what saving replaces and the skips it would drop. Interval and RRule schedules open in the Schedules tab's editor, which previews the next fire times (`POST /api/schedules/preview`). An edit to a code-declared schedule lasts until the next restart re-applies the declaration, unless the request sets `persist: true`. Related: [Schedules](https://sercanatalik.github.io/cereyan/concepts/schedules/index.md), [Backfill a date range](https://sercanatalik.github.io/cereyan/guides/backfill/index.md). # How to secure the server By default the server listens on loopback with no authentication: anyone on the machine can use it, and nobody off the machine can reach it. Add a token before binding to another address or exposing MCP, and use the Unix socket to let trusted local processes in without the token. ## Require a token ``` [server] token = "change-me" ``` Or `cereyan serve --token change-me`, `CEREYAN_TOKEN`, or `app.serve(token=...)`, in that precedence. With a token set, every `/api/*` route except `/api/health`, and the `/mcp` endpoint, require `Authorization: Bearer `. Requests without it get 401. Clients pick the token up from `CEREYAN_TOKEN` or `cereyan --token`; engine children receive it in their environment; the UI prompts for it once and stores it in a `cereyan_token` cookie scoped to `/api`. `server.json` records `auth: true` and never the token itself. ``` from cereyan import client api = client.Client("http://127.0.0.1:4200", token="change-me") assert api.token == "change-me" ``` Generate the token with something like `python -c "import secrets; print(secrets.token_urlsafe(32))"` and keep it out of the repository: `cereyan.toml` is usually committed, so prefer the environment variable on shared machines. ## Bind beyond loopback ``` [server] host = "0.0.0.0" port = 4200 token = "..." ``` The server warns at start when bound to a non-loopback address without a token. There is no TLS: put a reverse proxy in front if the network is not trusted. ## Trust local processes through the socket ``` [server] socket = "/tmp/cereyan.sock" ``` Or `--socket`, `CEREYAN_SOCKET`, or `app.serve(socket=...)`. The server also listens on the Unix socket, created with mode 0600 and removed on shutdown; a stale file from a crashed server is replaced. Requests over the socket skip the token check, because the file permissions are the authentication. `server.json` records the path, the Python client uses it when the server needs a token the client does not have (`Client(socket_path=...)` selects it explicitly), and `cereyan mcp --socket` connects through it. Keep the path short; the OS limits socket paths to about 100 bytes. Not available on Windows. ## What a token holder can do Everything the API allows, including starting runs, backfilling, setting variables, and, through MCP, the same for an agent. There is one permission level; there are no read-only tokens. ## Custom routes Routes registered with `@app.get` and friends under `/api/` are behind the token like the built-in API; routes outside `/api/` are open. Put anything that changes state under `/api/`. Related: [Configuration](https://sercanatalik.github.io/cereyan/reference/configuration/index.md), [Use cereyan with an AI agent](https://sercanatalik.github.io/cereyan/guides/agents/index.md). # How to run code on state changes Hooks are plain functions called in the engine when a run or task run reaches a state. Use them for in-process reactions such as closing a connection, writing a marker file, or posting a metric. For reactions that should happen even when the process is gone, or that involve other flows, use [rules](https://sercanatalik.github.io/cereyan/guides/rules/index.md) instead. ## Flow hooks ``` from cereyan import flow events = [] def note(flow, run, state): events.append((flow.name, state["type"])) @flow(on_completion=[note], on_failure=[note], on_crashed=[note], on_cancellation=[note]) def pipeline() -> str: return "done" pipeline() assert events == [("pipeline", "Completed")] ``` Each hook receives the `Flow`, the run as a dict (`id`, `name`, `parameters`, `state`, ...), and the state dict that triggered it. Hooks run after the state is recorded, in the order given. An exception in a hook is logged with the run and does not change the run's state. | Option | Called when | | ----------------- | ------------------------------------------------------------------------------------ | | `on_completion` | The run ends `Completed` | | `on_failure` | The run ends `Failed`, including `TimedOut` | | `on_crashed` | The server marks the run `Crashed`; runs on the server side since the engine is gone | | `on_cancellation` | The run ends `Cancelled` | ## Task hooks Tasks accept `on_completion`, `on_failure`, and `on_cancellation` with the same signature, receiving the `Task` instead of the flow: ``` from cereyan import flow, task seen = [] def audit(task, run, state): seen.append((task.name, state["type"], state.get("message"))) @task(on_failure=[audit]) def load() -> None: raise ValueError("bad rows") @flow def etl() -> None: load() try: etl() except ValueError: pass assert seen[0][:2] == ("load", "Failed") ``` ## Choosing between hooks and rules | | Hooks | Rules | | ----------------------- | ----------------------------- | --------------------------- | | Run where | In the engine, with your code | In the server | | Survive a crash | No, except `on_crashed` | Yes | | Can start other flows | Through the client | Yes, `run_flow` | | Configurable at runtime | No | Data rules yes | | Work offline | Yes | Code rules, once registered | Related: [React to events with rules](https://sercanatalik.github.io/cereyan/guides/rules/index.md), [Retry, time out and survive crashes](https://sercanatalik.github.io/cereyan/guides/retries-timeouts-crashes/index.md). # How to test a pipeline Flows are functions and every run is recorded in whatever home you point cereyan at, so tests need no server and no mocks: give each test a temporary home, call the flow, and assert on the return value, the files it wrote, or the recorded run. ## Isolate the home Set `CEREYAN_HOME` to a temporary directory before the flow runs, so tests never touch `~/.cereyan`: ``` # conftest.py import pytest @pytest.fixture(autouse=True) def cereyan_home(tmp_path, monkeypatch): monkeypatch.setenv("CEREYAN_HOME", str(tmp_path / "home")) monkeypatch.chdir(tmp_path) ``` ## Call the flow ``` from datetime import date from cereyan import flow, task, LocalTarget @task(output=lambda day: LocalTarget(f"out/{day}.csv")) def build(day: date) -> None: with LocalTarget(f"out/{day}.csv").open("w") as fh: fh.write("id\n1\n") @flow def daily(day: date) -> str: build(day) return f"out/{day}.csv" def test_daily_writes_the_file(): path = daily(date(2026, 1, 1)) assert LocalTarget(path).exists() test_daily_writes_the_file() ``` A flow that raises propagates the exception after recording the run as Failed, so `pytest.raises` works as usual. Parameters are coerced on the way in, so passing strings tests the same path the CLI and the API use. ## Assert on the recorded run `cereyan runs ls --json` reads the home's store, so a test that drives the CLI can check what was recorded. Run the flow in a subprocess too, since a process that is executing a flow holds the store lock: ``` import json, subprocess, sys, textwrap open("pipeline.py", "w").write(textwrap.dedent(""" from cereyan import flow @flow(tags=["nightly"]) def tagged() -> int: return 1 """)) subprocess.run([sys.executable, "-m", "cereyan", "run", "pipeline.py:tagged", "--quiet"], check=True) out = subprocess.run([sys.executable, "-m", "cereyan", "runs", "ls", "--flow", "tagged", "--json"], capture_output=True, text=True, check=True) runs = json.loads(out.stdout) assert runs[0]["state"]["type"] == "Completed" and "nightly" in runs[0]["tags"] ``` ## Test hooks and rules without a server Hooks are plain functions; call the flow and check what they captured. Code rules do not fire offline until a server has registered them, so test the rule's function directly with a fake event and run: ``` from cereyan import App app = App("tests") alerts = [] @app.rule(on="run.failed", flow="nightly") def page(event, run): alerts.append(run["name"]) page({"name": "run.failed"}, {"name": "nightly-1"}) assert alerts == ["nightly-1"] assert app.rules[0].spec()["when"]["events"] == ["run.failed"] ``` ## Test against a server For schedules, backfills, dependencies, rules, routes, and pauses, start a server on a temporary home in a session fixture and drive it through the client. The pattern used by cereyan's own suite and its documentation tests is in `tests/server_helpers.py`: start `cereyan serve --port 0 --no-open` with `CEREYAN_HOME` set, wait for `server.json` and `/api/health`, and stop it with SIGTERM. ``` run = served.client.run("etl", day="2026-04-01") final = served.wait_run(run["id"]) assert final["state"]["type"] == "Completed" assert any(t["name"] == "load" for t in served.client.task_runs(run["id"])) ``` ## Speed Offline runs cost a few milliseconds each. Keep the home per test (a fresh SQLite file is cheap) rather than per session when tests assert on run lists, so counts do not leak between tests. Related: [Engines and the home directory](https://sercanatalik.github.io/cereyan/concepts/engines-and-home/index.md), [Run code on state changes](https://sercanatalik.github.io/cereyan/guides/state-hooks/index.md). # How to store variables and secrets Keep configuration and credentials out of the code: set them once, read them at run time, and mark the sensitive ones secret. ## Set and read ``` from cereyan import Variable, flow Variable.set("batch_size", 500, tags=["tuning"]) Variable.set("regions", ["eu", "us"]) @flow def load() -> int: size = Variable.get("batch_size", default=100) return size * len(Variable.get("regions")) assert load() == 1000 ``` Values are JSON: numbers, strings, lists, and objects. Names use lowercase letters, digits, `_`, `-`, `/`, and `.`, and values are limited to 64 KB. `Variable.unset(name)` deletes; `set(..., overwrite=False)` refuses to replace an existing value. ## Store a secret ``` from cereyan import Variable Variable.set("warehouse/password", "hunter2", secret=True) assert Variable.get("warehouse/password") == "hunter2" ``` A secret is encrypted at rest with the key in `/secret.key`, created on first use with mode 0600, and shown masked in the UI, the API, and the MCP tools. `Variable.get` decrypts it inside your process. Back up `secret.key` with the database, and treat access to the home directory as access to the secrets. ## From the UI, the API, and an agent The Variables page adds, edits, and deletes variables; `POST /api/variables`, `PATCH /api/variables/{name}`, and `DELETE /api/variables/{name}` do the same; `Client.variables()` lists them with secrets masked; and the MCP `set_variable` tool lets an agent create or overwrite one: ``` names = {v["name"] for v in served.client.variables()} assert isinstance(names, set) ``` ## Scope Variables are global to the machine, not per project. Prefix names by project when two projects would otherwise collide, as in `warehouse/password` above. ## Offline and served Offline, `Variable.set` and `get` use the store directly. While a server holds the store, the same calls go through its API and the running server sees the change at once, so a script can set a value that the next scheduled run reads. Related: [Variables](https://sercanatalik.github.io/cereyan/concepts/variables/index.md), [Configuration](https://sercanatalik.github.io/cereyan/reference/configuration/index.md) for settings that belong in `cereyan.toml` instead. # Reference # CLI The `cereyan` command. Every subcommand accepts the global options first, for example `cereyan --home /tmp/h run pipeline.py:etl`. ``` cereyan [-h] [--home HOME] [--token CLIENT_TOKEN] {run,serve,backfill,mcp,runs} ... ``` ## Global options | Option | Meaning | | ------------------------ | -------------------------------------------------------------- | | `--home` `HOME` | runtime home directory (default: $CEREYAN_HOME or ~/.cereyan). | | `--token` `CLIENT_TOKEN` | API token for a running server (also CEREYAN_TOKEN). | ## `cereyan run` ``` cereyan run [-h] [--param NAME=VALUE] [--quiet] target ``` | Argument | Meaning | | -------- | ------------------------------------------ | | `target` | module_or_file:flow, e.g. pipeline.py:etl. | | Option | Meaning | | ---------------------------- | -------------------------------------------------------------------------------- | | `--param`, `-p` `NAME=VALUE` | set a flow parameter; repeatable, values are coerced from the flow's type hints. | | `--quiet`, `-q` | do not echo run logs. | ## `cereyan serve` ``` cereyan serve [-h] [--host HOST] [--port PORT] [--max-engines MAX_ENGINES] [--engine-max-runs ENGINE_MAX_RUNS] [--no-open] [--crash-retries CRASH_RETRIES] [--token TOKEN] [--socket SOCKET] [dir] ``` | Argument | Meaning | | ---------------- | -------------------------------------------------- | | `dir` (optional) | directory to import flows from (default: current). | | Option | Meaning | | ------------------------------------- | --------------------------------------------------------------------------------- | | `--host` `HOST` | bind address (also CEREYAN_HOST or [server] host; default 127.0.0.1). | | `--port` `PORT` | TCP port (also CEREYAN_PORT or [server] port; default 4200, 0 picks a free port). | | `--max-engines` `MAX_ENGINES` | size of the warm engine pool (also [server] max_engines). | | `--engine-max-runs` `ENGINE_MAX_RUNS` | recycle an engine after this many runs (also [server] engine_max_runs). | | `--no-open` | do not open the browser. | | `--crash-retries` `CRASH_RETRIES` | default crash retry limit (flow decorators override). | | `--token` `TOKEN` | require this API token (also CEREYAN_TOKEN or [server] token). | | `--socket` `SOCKET` | also listen on this Unix socket path (also CEREYAN_SOCKET or [server] socket). | ## `cereyan backfill` ``` cereyan backfill [-h] --param PARAM --start START --end END [--interval INTERVAL] [--concurrency CONCURRENCY] [--reverse] [--extra NAME=VALUE] [--json] flow ``` | Argument | Meaning | | -------- | ----------------------------------- | | `flow` | flow name, optionally project/flow. | | Option | Meaning | | ----------------------------- | ----------------------------------------------------------------- | | `--param` `PARAM` | date or datetime parameter name. | | `--start` `START` | first value of the parameter, a date or datetime. | | `--end` `END` | last value of the parameter, inclusive. | | `--interval` `INTERVAL` | seconds or a duration like 1d, 12h (default 1d). Default `1d`. | | `--concurrency` `CONCURRENCY` | how many of the backfill's runs may execute at once. Default `1`. | | `--reverse` | create the newest value first. | | `--extra` `NAME=VALUE` | fixed value for another flow parameter; repeatable. | | `--json` | print the backfill status as JSON. | ## `cereyan mcp` ``` cereyan mcp [-h] [--url URL] [--socket SOCKET] ``` | Option | Meaning | | ------------------- | --------------------------------------- | | `--url` `URL` | server URL (default: from server.json). | | `--socket` `SOCKET` | Unix socket path of the server. | ## `cereyan runs` ``` cereyan runs [-h] {ls} ... ``` ### `cereyan runs ls` ``` cereyan runs ls [-h] [--flow FLOW] [--project PROJECT] [--state STATE] [--limit LIMIT] [--json] ``` | Option | Meaning | | --------------------- | ------------------------------------- | | `--flow` `FLOW` | only runs of this flow name. | | `--project` `PROJECT` | only runs of flows in this project. | | `--state` `STATE` | state type, e.g. Failed. | | `--limit` `LIMIT` | number of runs to show. Default `20`. | | `--json` | print the runs as JSON. | ## Exit codes Every command exits 0 on success and 3 on an error, printing the reason to standard error. `cereyan run` also uses 1 and 2 to report the outcome of the run. | Code | Constant | Meaning | | ---- | ------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------ | | 0 | `EXIT_OK` | The run completed. | | 1 | `EXIT_FAILED` | The run failed. | | 2 | `EXIT_NOTHING_RAN` | Nothing ran: the store was locked and no server took the run. | | 3 | `EXIT_SCHEDULING` | The flow could not be loaded, a parameter did not coerce, a server was needed but unreachable or required a token, or a backfill argument was invalid. | # Configuration Cereyan reads `cereyan.toml` from the served directory, a handful of environment variables, and the CLI flags. The keys are validated in `python/cereyan/config.py`; unknown keys produce a warning at startup naming the key. A `[store]` table is an error: the home is set by `--home` or `CEREYAN_HOME` only. ## `cereyan.toml` ``` [server] host = "127.0.0.1" port = 4200 token = "change-me" # optional: require Authorization: Bearer on the API socket = "/tmp/cereyan.sock" # optional: also listen on a Unix socket (Unix only) max_engines = 8 engine_max_runs = 100 cancel_grace_secs = 10 open_browser = true [defaults] catchup = "skip" crash_retries = 5 retain_days = 30 [resources] db = 4 gpu = 1 [email] host = "smtp.example.com" port = 587 tls = "starttls" # none, starttls, tls username = "..." password = "..." from = "cereyan@example.com" ``` ### `[server]` | Key | Type | Default | Meaning | | ------------------- | ------- | ----------- | ------------------------------------------------------------------------------------ | | `host` | string | `127.0.0.1` | Bind address. The server warns when bound to a non-loopback address without a token. | | `port` | integer | `4200` | TCP port; `0` picks a free port. | | `token` | string | unset | API token required on every `/api/*` route except `/api/health`. | | `socket` | string | unset | Unix socket path served next to the TCP port; not available on Windows. | | `max_engines` | integer | CPU count | Size of the warm engine pool. | | `engine_max_runs` | integer | `100` | Runs an engine executes before it is recycled. | | `cancel_grace_secs` | integer | `10` | Seconds between SIGTERM and SIGKILL when cancelling a run. | | `open_browser` | boolean | `true` | Open the UI when the server starts. | ### `[defaults]` | Key | Type | Default | Meaning | | -------------------------------- | ------- | ------------- | ------------------------------------------------------------------------------ | | `catchup` | string | `skip` | Catch-up policy for schedules that do not set one: `skip`, `latest`, or `all`. | | `crash_retries` | integer | `5` | Reruns of a crashed run before it is marked Failed. | | `retain_days` | integer | `30` | Days of logs and events kept by retention. | | `max_engines`, `engine_max_runs` | integer | as `[server]` | Accepted here for compatibility; `[server]` takes precedence. | ### `[resources]` Each key is a resource name and its value the total, for example `db = 4`. Resources are shared by every project on the machine. ### `[email]` | Key | Type | Meaning | | ---------------------- | --------------- | -------------------------------- | | `host`, `port` | string, integer | SMTP server; `host` is required. | | `tls` | string | `none`, `starttls`, or `tls`. | | `username`, `password` | string | SMTP credentials. | | `from` | string | Sender address; required. | The Settings page (`PATCH /api/settings`) updates resource totals, retention, and the crash retry default and writes them back to `cereyan.toml`. ## Precedence | Setting | Order, highest first | | --------------- | ------------------------------------------------------------------------------------------------------- | | Home | `--home`, `CEREYAN_HOME`, `~/.cereyan` | | Host and port | `--host`/`--port`, `CEREYAN_HOST`/`CEREYAN_PORT`, `app.serve(host, port)`, `[server]`, `127.0.0.1:4200` | | Token | `--token`, `CEREYAN_TOKEN`, `app.serve(token=)`, `[server] token` | | Socket | `--socket`, `CEREYAN_SOCKET`, `app.serve(socket=)`, `[server] socket` | | `crash_retries` | the flow decorator, `[defaults]`, `--crash-retries`, `5` | ## Environment variables | Variable | Meaning | | ------------------------------ | ------------------------------------------------------------------------------------------------------------ | | `CEREYAN_HOME` | Runtime home directory. Engine children inherit it. | | `CEREYAN_HOST`, `CEREYAN_PORT` | Server bind address. | | `CEREYAN_TOKEN` | API token required by the server and sent by the CLI, the Python client, `cereyan mcp`, and engine children. | | `CEREYAN_SOCKET` | Unix socket path served next to the TCP port. | | `CEREYAN_NO_BROWSER` | Do not open the UI on `serve`. | Set by the server for its engine children, not for users: `CEREYAN_SERVER` (the server URL) and `CEREYAN_ENGINE_ID`. Two knobs exist for the test suite and benchmarks only: `CEREYAN_RETENTION_INTERVAL` (seconds between retention passes, default hourly) and `CEREYAN_FAST_CRASH_RERUN`. # Events Every meaningful change is recorded as an event with a name, a sequence number, an occurred time, a resource (`run`, `task_run`, `flow`, `schedule`, `rule`, or `custom`), related resources, and a JSON payload. Rules match on events, the Events page and `GET /api/events` list them, and new ones arrive on the SSE stream as `event.created`. The names below are the whole catalogue. It is declared once, in `crates/core/src/events.rs`, and every site that records an event names it from there, so this page is generated rather than maintained. `GET /api/vocabulary` serves the same list to the UI, and `cereyan.events` exposes it to Python: `events.run.failed` is the string `"run.failed"`. ## Run events Resource `run`; related `flow` and the run's tags. Offline the payload is `state` and `message` only. | Event | Payload | When | | --------------- | --------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `run.scheduled` | `state`, `state_type`, `message`, `flow`, `project`, `parameters`, `created_by` | The run is created, or re-enters Scheduled for a crash rerun | | `run.pending` | `state`, `state_type`, `message`, `flow`, `project`, `parameters`, `created_by` | An engine accepted the run | | `run.running` | `state`, `state_type`, `message`, `flow`, `project`, `parameters`, `created_by` | User code started | | `run.completed` | `state`, `state_type`, `message`, `flow`, `project`, `parameters`, `created_by` | The run finished without error | | `run.failed` | `state`, `state_type`, `message`, `flow`, `project`, `parameters`, `created_by` | The run raised, timed out, or was set failed | | `run.crashed` | `state`, `state_type`, `message`, `flow`, `project`, `parameters`, `created_by` | The engine died while the run was executing | | `run.cancelled` | `state`, `state_type`, `message`, `flow`, `project`, `parameters`, `created_by` | The run was cancelled | | `run.late` | `state`, `state_type`, `message`, `flow`, `project`, `parameters`, `created_by`, `scheduled_time`, `name` | The scheduled time passed 15 seconds ago and the run has not started | | `run.retrying` | `state`, `state_type`, `message`, `flow`, `project`, `parameters`, `created_by` | A retry attempt started | | `run.skipped` | `state`, `state_type`, `message`, `flow`, `project`, `parameters`, `created_by`, `reason` | The run ended Skipped: `on_overlap="skip"`, a backfill value already done, a catch-up drop, a fire a person skipped (`reason` `user`), or an upstream run skipped that way (`reason` `upstream`) | | `run.paused` | `state`, `state_type`, `message`, `flow`, `project`, `parameters`, `created_by` | The run is waiting on `wait_for_input` | | `run.resumed` | `state`, `state_type`, `message`, `flow`, `project`, `parameters`, `created_by` | The run was answered and its next attempt scheduled | `AwaitingRetry`, `AwaitingResource`, and `Cancelling` record no event. ## Task run events Resource `task_run` (id is the task run's external id, name its dynamic key); related `run` and `flow`. Offline the payload is `state` and `task_run`. | Event | Payload | When | | -------------------- | ------------------------------------------------------------ | ------------------------------------------------ | | `task_run.running` | `task`, `dynamic_key`, `state`, `message`, `flow`, `project` | The task started | | `task_run.completed` | `task`, `dynamic_key`, `state`, `message`, `flow`, `project` | The task returned | | `task_run.failed` | `task`, `dynamic_key`, `state`, `message`, `flow`, `project` | The task raised or timed out (after its retries) | | `task_run.cancelled` | `task`, `dynamic_key`, `state`, `message`, `flow`, `project` | The task was cancelled with its run | | `task_run.skipped` | `task`, `dynamic_key`, `state`, `message`, `flow`, `project` | The task's `output=` target already existed | | `task_run.cached` | `task`, `dynamic_key`, `state`, `message`, `flow`, `project` | The task returned a persisted result | ## Flow events Resource `flow`. | Event | Payload | When | | ----------------- | ------------------------------------- | ----------------------------------------------------------------------------------- | | `flow.registered` | `flow`, `project`, `module` | The server registered the flow at start or on handoff | | `flow.disabled` | `failures`, `window_seconds`, `until` | `disable_after` tripped; the flow's schedules are paused until `until` | | `flow.enabled` | empty | The disable window ended and the schedules resumed | | `flow.fan_in` | `key`, `value`, `upstream`, `run_id` | Every upstream completed a run for the key value and the downstream run was created | ## Schedule events Resource `schedule`, related `flow`. | Event | Payload | When | | ------------------------ | ------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------- | | `schedule.paused` | `schedule_id`, `reason` | A schedule was paused from the UI, the API, an MCP tool, a rule, or a disable window | | `schedule.resumed` | `schedule_id` | A schedule was resumed | | `schedule.catchup` | `schedule_id`, `policy`, `missed`, `created`, `dropped` | The server started and applied the catch-up policy to fires missed while it was down | | `schedule.skips_dropped` | `schedule_id`, `dropped` | An edit, or a restart that restored a code declaration, left skipped fires the schedule no longer produces, and they were forgotten | ## Resource events | Event | Payload | When | | -------------------- | ---------- | ------------------------------------------------------------------------ | | `resource.exhausted` | `resource` | A run waited for a resource that had no capacity; recorded once per wait | ## Rule events Resource `rule`, related the run and flow of the triggering event. | Event | Payload | When | | ----------------------- | ------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------- | | `rule.fired` | `rule_id`, `rule`, `event`, `event_id` | A rule matched an event and its actions started | | `rule.action.completed` | `rule_id`, `action`, `index`, `detail` | One action finished | | `rule.action.failed` | `rule_id`, `action`, `index`, `error` | One action failed, including a template that did not render | | `expectation.armed` | `id`, `rule_id`, `key`, `run_id`, `deadline` | A proactive rule's `when` event armed an expectation | | `expectation.met` | `id`, `rule_id`, `key` | The expected event arrived before the deadline | | `expectation.lapsed` | `rule_id`, `rule`, `flow`, `project`, `run`, `run_name`, `expected`, `deadline`, `armed_at`, `expectation_id` | The deadline passed, or a clock-armed rule's tick found no matching event; the rule's actions run against this event | Runs created by a rule record `created_by = rule:`, and a rule never fires on events of runs it created unless `allow_self` is set. ## Custom events `cereyan.emit_event(name, payload=None, resource=None)` records any name of your own. Inside a run the resource defaults to that run and the task run is recorded; outside a run the event goes to the store, or to the server's `POST /api/events` when one holds the store. Names with a dot-separated prefix, such as `orders.table_empty`, match rules with `events: ["orders.*"]`. The prefixes `run.`, `task_run.`, `flow.`, `schedule.`, `resource.`, `rule.`, `expectation.` are the engine's. A name under one of them that is not in the catalogue above is rejected by `emit_event`, by `@app.rule`, and by the rules API, because nothing would ever emit it — a rule matching one could only ever sit silent. Every other name is yours and is never checked. ## Stream messages The SSE stream at `GET /api/stream` carries live notifications for the UI. They are not stored events and rules cannot match them, even where a name is shared with one: `run.updated` is a stream message, so a rule naming it is rejected. | Message | Data | Sent when | | ------------------------------------------------------------ | ----------------------------------------- | --------------------------------------------------------------------------------------- | | `hello` | `latest`, `since` | The connection opens, before any backlog | | `run.updated` | The run | A run was created or changed state | | `task_run.updated` | The task run | A task run was created or changed state | | `log.appended` | `run_id`, `last_id`, `count` | New log lines arrived | | `event.created` | The event | Any event was recorded | | `flow.registered` | The flow | A flow was registered or re-registered | | `rule.updated` | The rule, or `id` and `deleted` | A rule was created, edited, fired, or deleted | | `variable.updated` | The variable, or `name` and `deleted` | A variable was set or removed | | `artifact.updated` | The artifact | A run published or updated an artifact | | `schedule.updated` | The schedule, or `id` and `deleted` | A schedule was created, edited, paused, resumed, or deleted | | `backfill.created` | `backfill_id`, `flow_id`, `count` | A backfill created its runs | | `backfill.updated` | `backfill_id`, `cancelled` | A backfill was cancelled | | `expectation.armed`, `expectation.met`, `expectation.lapsed` | `id`, `rule_id`, and the key or the event | A proactive rule armed, disarmed, or lapsed an expectation | | `resync` | `latest` | The client asked for a sequence number older than the replay buffer; refetch everything | Each message carries the sequence number as its SSE id. Reconnect with `?since=` to replay what was missed, or receive `resync` when the buffer no longer reaches back that far. # Exit codes and the discovery file ## CLI exit codes | Code | Meaning | | ---- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | 0 | Success. For `cereyan run`, the run completed. | | 1 | `cereyan run` only: the run failed. | | 2 | `cereyan run` only: nothing ran, because the store was locked and no server took the run. | | 3 | An error before or outside the run: the flow could not be loaded, a parameter did not coerce, a server was needed but unreachable or required a token, or a backfill argument was invalid. | The full command reference, generated from the parser, is on the [CLI](https://sercanatalik.github.io/cereyan/reference/cli/index.md) page. ## The runtime home One home per machine, resolved from `--home`, then `CEREYAN_HOME`, then `~/.cereyan`. It contains: | Entry | Meaning | | ------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `db.sqlite` | The store: flows, runs, task runs, logs, events, rules, artifacts, variables, settings. WAL mode, so `db.sqlite-wal` and `db.sqlite-shm` appear while it is open. | | `db.lock` | OS advisory lock held by the process that owns the store: a running server, or an offline script while it writes. | | `server.json` | Written by a running server and removed on shutdown; see below. | | `secret.key` | Created when the first secret variable is set; encrypts secrets at rest. | | `storage/` | Persisted task results and cache entries. | A corrupted `db.sqlite` is moved aside on open and a fresh store is created; history is a cache and the code plus targets are the source of truth. The home is created readable only by the account that creates it, and everything in the table above is protected by that rather than by permissions set on each file. A home from an earlier version that is broader is narrowed when it is opened, and cereyan says so once. On Windows a home under your user profile inherits the same protection; a `CEREYAN_HOME` pointed somewhere else is yours to protect. ## `server.json` A running server records how to reach it so scripts, the CLI, `cereyan mcp`, and engines find it: ``` { "host": "127.0.0.1", "port": 4200, "url": "http://127.0.0.1:4200", "pid": 12345, "started_at": 1788998400000000, "version": "1.5.0", "auth": true, "socket": "/tmp/cereyan.sock" } ``` | Field | Meaning | | -------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `host`, `port` | What the listener bound. `host` is `0.0.0.0` when the server was started on every interface. | | `url` | Where to reach it from this machine. When the server bound an unspecified address, this is loopback rather than the bind address, because `0.0.0.0` is not somewhere a client can connect. | | `pid` | The server process; clients check it is alive before trusting a stale file. | | `started_at` | Microseconds since the Unix epoch. | | `version` | The cereyan version. | | `auth` | Whether a token is required. The token itself is never written. | | `socket` | The Unix socket path when one is served, else `null`. | `cereyan.client.read_discovery()` returns it and `cereyan.client.find_server()` turns it into a client after checking `/api/health`. # HTTP API The server's OpenAPI document (version 1.13.0) is served at `GET /api/openapi.json`; this page is rendered from the checked-in snapshot that the UI client and the test suite are generated from. Every `/api/*` route except `/api/health` requires `Authorization: Bearer ` when a token is set (see [Secure the server](https://sercanatalik.github.io/cereyan/guides/secure-the-server/index.md)); requests over the Unix socket skip the check. Responses are JSON; errors carry an `ErrorBody`. Times are microseconds since the Unix epoch in UTC. List endpoints page by keyset cursor. The MCP endpoint (`POST /mcp`) is not part of the OpenAPI document; see [MCP tools, resources and prompts](https://sercanatalik.github.io/cereyan/reference/mcp/index.md). ## Health and server ### `GET /api/counts` | Parameter | In | Type | Required | Description | | --------- | ----- | -------------- | -------- | ----------- | | `project` | query | string or null | no | | | Status | Body | | ------ | -------------------------------------- | | 200 | [`Counts`](#counts) (application/json) | ### `GET /api/health` Note Exempt from the API token. | Status | Body | | ------ | ------------ | | 200 | Server is up | ### `GET /api/server` | Status | Body | | ------ | ---------------------------------------------- | | 200 | [`ServerInfo`](#serverinfo) (application/json) | ### `GET /api/settings` | Status | Body | | ------ | ------------------------------------------ | | 200 | [`Settings`](#settings) (application/json) | ### `PATCH /api/settings` **Request body** (application/json): [`SettingsPatch`](#settingspatch) | Status | Body | | ------ | ------------------------------------------ | | 200 | [`Settings`](#settings) (application/json) | ### `GET /api/stream` | Parameter | In | Type | Required | Description | | --------- | ----- | ----------------------- | -------- | ----------------------------------------- | | `since` | query | integer or null (int64) | no | Last sequence number the client has seen. | | Status | Body | | ------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | 200 | text/event-stream of hello, run.updated, task_run.updated, log.appended, event.created, flow.registered, rule.updated, variable.updated, artifact.updated, schedule.updated, backfill.created, backfill.updated, expectation.armed, expectation.met, expectation.lapsed, and resync | ### `GET /api/vocabulary` | Status | Body | | ------ | ---------------------------------------------- | | 200 | [`Vocabulary`](#vocabulary) (application/json) | ## Flows ### `GET /api/flows` | Parameter | In | Type | Required | Description | | --------- | ----- | -------------- | -------- | ----------- | | `project` | query | string or null | no | | | Status | Body | | ------ | -------------------------------------------------- | | 200 | [`FlowSummary`](#flowsummary)[] (application/json) | ### `GET /api/flows/{id}` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | ------------------------------------------------ | | 200 | [`FlowSummary`](#flowsummary) (application/json) | | 404 | no body | ### `DELETE /api/flows/{id}` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | ------------ | | 204 | no body | | 404 | no body | | 409 | Flow is live | ### `POST /api/flows/{id}/backfill` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | **Request body** (application/json): [`BackfillBody`](#backfillbody) | Status | Body | | ------ | ------------------------------------------------------ | | 201 | [`BackfillStatus`](#backfillstatus) (application/json) | | 422 | no body | ### `GET /api/flows/{id}/backfills` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | -------------------------------------------------------- | | 200 | [`BackfillStatus`](#backfillstatus)[] (application/json) | ### `POST /api/flows/{id}/pause` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | -------------------- | | 200 | All schedules paused | ### `POST /api/flows/{id}/resume` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | --------------------- | | 200 | All schedules resumed | ### `POST /api/flows/{id}/runs` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | **Request body** (application/json): [`CreateRunForFlowBody`](#createrunforflowbody) | Status | Body | | ------ | -------------------------------- | | 201 | [`Run`](#run) (application/json) | | 422 | Invalid parameters | ### `GET /api/flows/{id}/schedules` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | -------------------------------------------------- | | 200 | [`ScheduleRow`](#schedulerow)[] (application/json) | ### `POST /api/flows/{id}/schedules` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | **Request body** (application/json): [`ScheduleBody`](#schedulebody) | Status | Body | | ------ | ------------------------------------------------ | | 201 | [`ScheduleRow`](#schedulerow) (application/json) | | 422 | no body | ### `GET /api/flows/{id}/upcoming` | Parameter | In | Type | Required | Description | | ----------- | ----- | --------------- | -------- | --------------------------------------------------------------------------------------------------------------------------- | | `id` | path | integer (int64) | yes | | | `projected` | query | integer or null | no | Also list this many fires of each active schedule past its materialized runs, computed without creating runs (at most 100). | | Status | Body | | ------ | ---------------------------------------------------- | | 200 | [`UpcomingItem`](#upcomingitem)[] (application/json) | ## Runs ### `GET /api/runs` | Parameter | In | Type | Required | Description | | ----------------- | ----- | ----------------------- | -------- | ---------------------------------------------------------------------------------------------------------------- | | `project` | query | string or null | no | | | `flow` | query | string or null | no | | | `flow_id` | query | integer or null (int64) | no | | | `state_type` | query | string or null | no | | | `state_name` | query | string or null | no | | | `name` | query | string or null | no | | | `tags` | query | string or null | no | Tags the run must carry; a query string may pass them comma-separated. | | `start_after` | query | integer or null (int64) | no | Inclusive lower bound on the run's start time (or creation when never started), microseconds. | | `start_before` | query | integer or null (int64) | no | | | `sort` | query | string or null | no | `created_desc` (default), `created_asc`, `start_desc`, `start_asc`, `duration_desc`, `duration_asc`, `name_asc`. | | `limit` | query | integer or null | no | | | `cursor` | query | integer or null (int64) | no | Keyset cursor: the id of the last run of the previous page (id-ordered sorts only). | | `schedule_id` | query | integer or null (int64) | no | | | `backfill_id` | query | integer or null (int64) | no | | | `scheduled_after` | query | integer or null (int64) | no | Only runs with a scheduled time after this (upcoming lists). | | Status | Body | | ------ | ------------------------------------------ | | 200 | [`RunsPage`](#runspage) (application/json) | ### `POST /api/runs` **Request body** (application/json): [`CreateRunBody`](#createrunbody) | Status | Body | | ------ | -------------------------------- | | 201 | [`Run`](#run) (application/json) | | 404 | Unknown flow and no module given | | 422 | no body | ### `GET /api/runs/{id}` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | -------------------------------- | | 200 | [`Run`](#run) (application/json) | | 404 | no body | ### `DELETE /api/runs/{id}` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | ------- | | 204 | no body | | 404 | no body | ### `GET /api/runs/{id}/artifacts` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | -------------------------------------------------- | | 200 | [`ArtifactRow`](#artifactrow)[] (application/json) | ### `POST /api/runs/{id}/cancel` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | -------------------------------- | | 200 | [`Run`](#run) (application/json) | | 404 | no body | ### `GET /api/runs/{id}/graph` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | ------------------------------------------ | | 200 | [`RunGraph`](#rungraph) (application/json) | ### `GET /api/runs/{id}/input` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | -------------------------------------------------- | | 200 | {input: } or {input: null} when nothing was stored | | 404 | no body | ### `GET /api/runs/{id}/logs` | Parameter | In | Type | Required | Description | | --------- | ----- | ----------------------- | -------- | ------------------------------------------------------------------------------- | | `id` | path | integer (int64) | yes | | | `after` | query | integer or null (int64) | no | Return rows with id greater than this (keyset cursor). | | `level` | query | string or null | no | Minimum level: a Python numeric level or DEBUG, INFO, WARNING, ERROR, CRITICAL. | | `search` | query | string or null | no | | | `limit` | query | integer or null | no | | | Status | Body | | ------ | ------------------------------------------ | | 200 | [`LogsPage`](#logspage) (application/json) | ### `POST /api/runs/{id}/resume` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | **Request body** (application/json): [`ResumeBody`](#resumebody) | Status | Body | | ------ | -------------------------------- | | 200 | [`Run`](#run) (application/json) | | 404 | no body | | 409 | no body | ### `GET /api/runs/{id}/tasks` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | ------------------------------------------ | | 200 | [`TaskRun`](#taskrun)[] (application/json) | ### `POST /api/runs/{id}/transition` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | **Request body** (application/json): [`TransitionBody`](#transitionbody) | Status | Body | | ------ | -------------------------------------------------------------- | | 200 | [`Run`](#run) (application/json) | | 404 | no body | | 409 | [`TransitionRejected`](#transitionrejected) (application/json) | ## Task runs ### `GET /api/task-runs` | Parameter | In | Type | Required | Description | | -------------- | ----- | ----------------------- | -------- | ----------- | | `run_id` | query | integer or null (int64) | no | | | `project` | query | string or null | no | | | `flow` | query | string or null | no | | | `state_type` | query | string or null | no | | | `state_name` | query | string or null | no | | | `name` | query | string or null | no | | | `start_after` | query | integer or null (int64) | no | | | `start_before` | query | integer or null (int64) | no | | | `limit` | query | integer or null | no | | | `cursor` | query | integer or null (int64) | no | | | Status | Body | | ------ | -------------------------------------------------- | | 200 | [`TaskRunsPage`](#taskrunspage) (application/json) | ### `GET /api/task-runs/{id}` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | ---------------------------------------- | | 200 | [`TaskRun`](#taskrun) (application/json) | | 404 | no body | ### `GET /api/task-runs/{id}/artifacts` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | -------------------------------------------------- | | 200 | [`ArtifactRow`](#artifactrow)[] (application/json) | ### `GET /api/task-runs/{id}/logs` | Parameter | In | Type | Required | Description | | --------- | ----- | ----------------------- | -------- | ------------------------------------------------------------------------------- | | `id` | path | integer (int64) | yes | | | `after` | query | integer or null (int64) | no | Return rows with id greater than this (keyset cursor). | | `level` | query | string or null | no | Minimum level: a Python numeric level or DEBUG, INFO, WARNING, ERROR, CRITICAL. | | `search` | query | string or null | no | | | `limit` | query | integer or null | no | | | Status | Body | | ------ | ------------------------------------------ | | 200 | [`LogsPage`](#logspage) (application/json) | ## Schedules ### `POST /api/schedules/preview` **Request body** (application/json): [`PreviewBody`](#previewbody) | Status | Body | | ------ | -------------------------------------------------------- | | 200 | [`PreviewResponse`](#previewresponse) (application/json) | | 422 | no body | ### `PATCH /api/schedules/{sid}` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `sid` | path | integer (int64) | yes | | **Request body** (application/json): [`SchedulePatchBody`](#schedulepatchbody) | Status | Body | | ------ | ------------------------------------------------ | | 200 | [`ScheduleRow`](#schedulerow) (application/json) | | 422 | no body | ### `DELETE /api/schedules/{sid}` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `sid` | path | integer (int64) | yes | | | Status | Body | | ------ | ------- | | 204 | no body | | 404 | no body | ### `POST /api/schedules/{sid}/pause` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `sid` | path | integer (int64) | yes | | | Status | Body | | ------ | ------------------------------------------------ | | 200 | [`ScheduleRow`](#schedulerow) (application/json) | ### `POST /api/schedules/{sid}/resume` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `sid` | path | integer (int64) | yes | | | Status | Body | | ------ | ------------------------------------------------ | | 200 | [`ScheduleRow`](#schedulerow) (application/json) | ### `POST /api/schedules/{sid}/skips` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `sid` | path | integer (int64) | yes | | **Request body** (application/json): [`SkipBody`](#skipbody) | Status | Body | | ------ | -------------------------------------------------- | | 200 | [`SkipResponse`](#skipresponse) (application/json) | | 404 | no body | | 422 | no body | ### `DELETE /api/schedules/{sid}/skips/{fire}` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | -------------------------------------- | | `sid` | path | integer (int64) | yes | | | `fire` | path | integer (int64) | yes | The skipped fire time, in microseconds | | Status | Body | | ------ | ------------------------------------------------ | | 200 | [`ScheduleRow`](#schedulerow) (application/json) | | 404 | no body | | 422 | no body | ## Backfills ### `GET /api/backfills/{id}` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | ------------------------------------------------------ | | 200 | [`BackfillStatus`](#backfillstatus) (application/json) | | 404 | no body | ### `POST /api/backfills/{id}/cancel` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | ------------------------------------------------------ | | 200 | [`BackfillStatus`](#backfillstatus) (application/json) | ### `POST /api/backfills/{id}/prefilter` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | **Request body** (application/json): [`PrefilterBody`](#prefilterbody) | Status | Body | | ------ | ------- | | 200 | no body | ## Events ### `GET /api/events` | Parameter | In | Type | Required | Description | | --------------- | ----- | ----------------------- | -------- | ------------------------------------------------------------- | | `name` | query | string or null | no | Exact name, or a prefix ending in `*` (e.g. `run.*`). | | `resource_kind` | query | string or null | no | | | `resource_id` | query | string or null | no | | | `run_id` | query | integer or null (int64) | no | | | `flow_id` | query | integer or null (int64) | no | | | `after` | query | integer or null (int64) | no | | | `before` | query | integer or null (int64) | no | | | `limit` | query | integer or null | no | | | `cursor` | query | integer or null (int64) | no | Keyset cursor: the id of the last event of the previous page. | | `ascending` | query | boolean | no | Ascending order when true (defaults to newest first). | | Status | Body | | ------ | ---------------------------------------------- | | 200 | [`EventsPage`](#eventspage) (application/json) | ### `POST /api/events` **Request body** (application/json): [`EmitEventBody`](#emiteventbody) | Status | Body | | ------ | ------------------------------------ | | 201 | [`Event`](#event) (application/json) | ### `GET /api/events/{id}` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | ------------------------------------ | | 200 | [`Event`](#event) (application/json) | | 404 | no body | ## Rules ### `GET /api/rules` | Status | Body | | ------ | ------------------------------------------ | | 200 | [`RuleRow`](#rulerow)[] (application/json) | ### `POST /api/rules` **Request body** (application/json): [`RuleBody`](#rulebody) | Status | Body | | ------ | ---------------------------------------- | | 201 | [`RuleRow`](#rulerow) (application/json) | | 422 | no body | ### `GET /api/rules/{id}` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | ---------------------------------------- | | 200 | [`RuleRow`](#rulerow) (application/json) | | 404 | no body | ### `PATCH /api/rules/{id}` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | **Request body** (application/json): [`RulePatch`](#rulepatch) | Status | Body | | ------ | ---------------------------------------- | | 200 | [`RuleRow`](#rulerow) (application/json) | | 404 | no body | | 422 | no body | ### `DELETE /api/rules/{id}` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | ------- | | 204 | no body | | 404 | no body | | 409 | no body | ### `GET /api/rules/{id}/expectations` | Parameter | In | Type | Required | Description | | --------- | ----- | --------------- | -------- | --------------------------------------------- | | `id` | path | integer (int64) | yes | | | `open` | query | boolean or null | no | Only open (armed) expectations; default true. | | `limit` | query | integer or null | no | | | Status | Body | | ------ | -------------------------------------------------- | | 200 | [`Expectation`](#expectation)[] (application/json) | | 404 | no body | ### `GET /api/rules/{id}/firings` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | ------------------------------------------------ | | 200 | [`RuleFiring`](#rulefiring)[] (application/json) | ### `POST /api/rules/{id}/test` | Parameter | In | Type | Required | Description | | --------- | ---- | --------------- | -------- | ----------- | | `id` | path | integer (int64) | yes | | | Status | Body | | ------ | -------------------------------------------------- | | 200 | Rendered actions against the latest matching event | ## Artifacts ### `GET /api/artifacts` | Parameter | In | Type | Required | Description | | --------- | ----- | ----------------------- | -------- | ---------------------------------------------------------------- | | `kind` | query | string or null | no | | | `key` | query | string or null | no | | | `flow` | query | string or null | no | | | `project` | query | string or null | no | | | `run_id` | query | integer or null (int64) | no | | | `limit` | query | integer or null | no | | | `after` | query | integer or null (int64) | no | Keyset cursor: the id of the last artifact of the previous page. | | Status | Body | | ------ | ---------------------------------------------------- | | 200 | [`ArtifactsPage`](#artifactspage) (application/json) | ### `POST /api/artifacts` **Request body** (application/json): [`ArtifactBody`](#artifactbody) | Status | Body | | ------ | ------------------------------------------------ | | 201 | [`ArtifactRow`](#artifactrow) (application/json) | | 422 | no body | ## Variables ### `GET /api/variables` | Status | Body | | ------ | -------------------------------------------------- | | 200 | [`VariableRow`](#variablerow)[] (application/json) | ### `POST /api/variables` **Request body** (application/json): [`VariableBody`](#variablebody) | Status | Body | | ------ | ------------------------------------------------ | | 201 | [`VariableRow`](#variablerow) (application/json) | | 409 | no body | | 422 | no body | ### `GET /api/variables/{name}` | Parameter | In | Type | Required | Description | | --------- | ----- | --------------- | -------- | -------------------------------------------------------------------------- | | `name` | path | string | yes | | | `raw` | query | boolean or null | no | Return the stored ciphertext of a secret so a local client can decrypt it. | | Status | Body | | ------ | -------------------------------------------------------- | | 200 | [`VariableWithRaw`](#variablewithraw) (application/json) | | 404 | no body | ### `PATCH /api/variables/{name}` | Parameter | In | Type | Required | Description | | --------- | ---- | ------ | -------- | ----------- | | `name` | path | string | yes | | **Request body** (application/json): [`VariablePatch`](#variablepatch) | Status | Body | | ------ | ------------------------------------------------ | | 200 | [`VariableRow`](#variablerow) (application/json) | | 404 | no body | ### `DELETE /api/variables/{name}` | Parameter | In | Type | Required | Description | | --------- | ---- | ------ | -------- | ----------- | | `name` | path | string | yes | | | Status | Body | | ------ | ------- | | 204 | no body | | 404 | no body | ## Resources ### `POST /api/resources/acquire` **Request body** (application/json): [`AcquireRequest`](#acquirerequest) | Status | Body | | ------ | -------------- | | 200 | {lease: id} or | ### `POST /api/resources/release` **Request body** (application/json): [`ReleaseRequest`](#releaserequest) | Status | Body | | ------ | ------- | | 200 | no body | ## Engine (internal) ### `POST /api/engine/failed` Note Used by engine processes to report to the server; not intended for clients and not covered by compatibility promises. **Request body** (application/json): [`FailedRequest`](#failedrequest) | Status | Body | | ------ | -------------------------------- | | 200 | Queued runs of the module failed | ### `POST /api/engine/heartbeat` Note Used by engine processes to report to the server; not intended for clients and not covered by compatibility promises. **Request body** (application/json): [`HeartbeatRequest`](#heartbeatrequest) | Status | Body | | ------ | ---------------------------- | | 200 | {cancel: bool, active: bool} | ### `POST /api/engine/report` Note Used by engine processes to report to the server; not intended for clients and not covered by compatibility promises. **Request body** (application/json): [`ReportRequest`](#reportrequest) | Status | Body | | ------ | ------------------------------------------------------ | | 200 | [`ReportResponse`](#reportresponse) (application/json) | | 404 | no body | ### `POST /api/engine/work` Note Used by engine processes to report to the server; not intended for clients and not covered by compatibility promises. **Request body** (application/json): [`WorkRequest`](#workrequest) | Status | Body | | ------ | -------------------------------------------------- | | 200 | [`WorkResponse`](#workresponse) (application/json) | ## Schemas ### `AcquireRequest` | Field | Type | Required | Description | | ----------- | --------------- | -------- | ----------- | | `resources` | object | yes | | | `run_id` | integer (int64) | yes | | | `wait_ms` | integer (int64) | no | | ### `AfterSpec` | Field | Type | Required | Description | | ------------ | --------------- | -------- | ---------------------------------------------------------------------- | | `flow` | string | yes | The first upstream (kept for 1.0 payloads and the run details link). | | `flows` | array of string | no | Every upstream; empty for rows written before fan-in existed. | | `key` | string or null | no | Parameter that identifies a batch: the downstream runs once per value. | | `parameters` | object | no | | ### `ArtifactBody` | Field | Type | Required | Description | | ------------- | ----------------------- | -------- | ----------- | | `data` | object | yes | | | `key` | string or null | no | | | `kind` | string | yes | | | `run_id` | integer (int64) | yes | | | `task_run_id` | integer or null (int64) | no | | ### `ArtifactListItem` One artifact in the cross-run listing, with its run's identity. Type: any. ### `ArtifactRow` A stored artifact attached to a run and optionally a task run. | Field | Type | Required | Description | | ------------- | ----------------------- | -------- | ---------------------------------------------------- | | `created_at` | [`i64`](#i64) | yes | | | `data` | object | yes | | | `external_id` | [`Id`](#id) | yes | | | `id` | integer (int64) | yes | | | `key` | string or null | no | | | `kind` | string | yes | `markdown`, `table`, `progress`, `link`, or `image`. | | `run_id` | integer (int64) | yes | | | `task_run_id` | integer or null (int64) | no | | | `updated_at` | [`i64`](#i64) | yes | | ### `ArtifactsPage` | Field | Type | Required | Description | | ------------- | ----------------------------------------- | -------- | ----------- | | `items` | [`ArtifactListItem`](#artifactlistitem)[] | yes | | | `next_cursor` | integer or null (int64) | no | | ### `Backfill` | Field | Type | Required | Description | | ------------------ | --------------- | -------- | ----------- | | `cancelled` | boolean | yes | | | `concurrency` | integer (int64) | yes | | | `created_at` | [`i64`](#i64) | yes | | | `end_value` | string | yes | | | `external_id` | [`Id`](#id) | yes | | | `extra_parameters` | object | no | | | `flow_id` | integer (int64) | yes | | | `id` | integer (int64) | yes | | | `interval_secs` | number (double) | yes | | | `parameter` | string | yes | | | `start_value` | string | yes | | | `total` | integer (int64) | yes | | ### `BackfillBody` | Field | Type | Required | Description | | ------------------ | ----------------------- | -------- | ----------------------------------------------------------------- | | `concurrency` | integer or null (int64) | no | | | `end` | string | yes | | | `extra_parameters` | object | no | | | `interval` | any | no | Seconds, or a shorthand like `1d`, `12h`, `30m`. Default one day. | | `parameter` | string | yes | | | `reverse` | boolean | no | | | `start` | string | yes | | ### `BackfillStatus` Type: any. ### `CatchupPolicy` One of: `skip`, `latest`, `all`. ### `Counts` | Field | Type | Required | Description | | ----------- | --------------- | -------- | ------------------------------------------- | | `active` | integer (int64) | yes | Number of non-terminal runs held in memory. | | `flows` | object | yes | Run counts by flow id and state type. | | `runs` | object | yes | Run counts by state type. | | `task_runs` | object | yes | Task run counts by state type. | ### `CreateRunBody` Create a run by flow key, registering the flow when the server does not know it yet (offline handoff from another project). | Field | Type | Required | Description | | ------------------ | --------------- | -------- | ----------------------------------------------------------------------- | | `created_by` | string or null | no | | | `description` | string or null | no | | | `flow` | string | yes | | | `flow_group` | string or null | no | The flow's declared group; absent leaves the registered group as it is. | | `flow_tags` | array of string | no | | | `module` | string or null | no | | | `name` | string or null | no | | | `options` | object | no | | | `parameter_schema` | object | no | | | `parameters` | object | no | | | `project` | string | yes | | | `source_dir` | string or null | no | | | `tags` | array of string | no | | ### `CreateRunForFlowBody` | Field | Type | Required | Description | | ------------ | --------------- | -------- | ----------- | | `name` | string or null | no | | | `parameters` | object | no | | | `tags` | array of string | no | | ### `DownstreamSkip` A flow that runs after the skipped one, directly or further down its chain. | Field | Type | Required | Description | | --------- | ------------------------ | -------- | ------------------------------------------------------------------ | | `fires` | array of integer (int64) | yes | The skipped fires whose runs of this flow will be created Skipped. | | `flow` | string | yes | | | `project` | string | yes | | ### `EmitEventBody` | Field | Type | Required | Description | | ---------- | ------------------------------- | -------- | ----------- | | `flow_id` | integer or null (int64) | no | | | `name` | string | yes | | | `payload` | object | no | | | `resource` | null or [`Resource`](#resource) | no | | | `run_id` | integer or null (int64) | no | | ### `ErrorBody` | Field | Type | Required | Description | | ------- | ------ | -------- | ----------- | | `error` | string | yes | | ### `Event` | Field | Type | Required | Description | | ------------- | ------------------------- | -------- | -------------------------------------------------------- | | `external_id` | [`Id`](#id) | yes | | | `flow_id` | integer or null (int64) | no | | | `id` | integer (int64) | yes | | | `name` | string | yes | | | `occurred` | [`i64`](#i64) | yes | | | `payload` | object | no | | | `related` | [`Resource`](#resource)[] | no | | | `resource` | [`Resource`](#resource) | yes | | | `run_id` | integer or null (int64) | no | | | `seq` | integer (int64) | yes | Sequence number: the writer assigns ids in commit order. | ### `EventEntry` | Field | Type | Required | Description | | ---------------- | --------------- | -------- | ---------------------------------------------- | | `name` | string | yes | The name a rule matches on, e.g. `run.failed`. | | `payload_fields` | array of string | yes | The payload keys the emit site sets. | | `resource` | string | yes | The resource kind the event hangs off. | | `when` | string | yes | One sentence on when the engine records it. | ### `EventsPage` | Field | Type | Required | Description | | ------------- | ----------------------- | -------- | ----------- | | `items` | [`Event`](#event)[] | yes | | | `next_cursor` | integer or null (int64) | no | | ### `Expectation` An armed expectation of a proactive rule. | Field | Type | Required | Description | | ---------- | ----------------------- | -------- | ---------------------------------------- | | `armed_at` | [`i64`](#i64) | yes | | | `deadline` | [`i64`](#i64) | yes | | | `flow_id` | integer or null (int64) | no | | | `id` | integer (int64) | yes | | | `key` | string | yes | `run:` or `flow:`. | | `rule_id` | integer (int64) | yes | | | `run_id` | integer or null (int64) | no | | | `status` | string | yes | `open`, `met`, `lapsed`, or `cancelled`. | ### `FailedRequest` | Field | Type | Required | Description | | ------------ | --------------- | -------- | ----------- | | `engine_id` | string | yes | | | `isolated` | boolean | no | | | `module` | string | yes | | | `nice` | integer (int32) | no | | | `source_dir` | string | yes | | | `traceback` | string | yes | | ### `Flow` A registered flow. Identity is `(project, name)`. | Field | Type | Required | Description | | ------------------ | --------------- | -------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `created_at` | [`i64`](#i64) | yes | | | `description` | string or null | no | | | `error` | string or null | no | | | `external_id` | [`Id`](#id) | yes | | | `group` | string or null | no | The group the flow is listed under, declared in Python with `group=`. Null means none was declared and the flow is grouped under its project; API responses carry the resolved value, so clients never apply that fallback themselves. | | `id` | integer (int64) | yes | | | `last_seen_at` | [`i64`](#i64) | yes | | | `live` | boolean | no | Whether the running server has this flow registered from code. | | `module` | string | yes | | | `name` | string | yes | | | `options` | object | no | | | `parameter_schema` | object | no | | | `project` | string | yes | | | `source_dir` | string | yes | | | `tags` | array of string | no | | ### `FlowOptions` Flow-level options declared in Python and stored as JSON on the flow row. | Field | Type | Required | Description | | ------------------- | --------------------------------- | -------- | ---------------------------------------- | | `after` | null or [`AfterSpec`](#afterspec) | no | | | `crash_retries` | integer or null (int64) | no | | | `disable_after` | array or null | no | (count, window_seconds, persist_seconds) | | `has_bulk_complete` | boolean | no | | | `has_crash_hooks` | boolean | no | | | `isolated` | boolean | no | | | `log_prints` | boolean | no | | | `max_concurrent` | integer or null (int64) | no | | | `on_overlap` | string | no | `enqueue`, `skip`, or `cancel_new`. | | `priority` | integer (int64) | no | | | `resources` | object | no | | | `retries` | integer (int64) | no | | | `schedules` | [`ScheduleDecl`](#scheduledecl)[] | no | | | `timeout_seconds` | number or null (double) | no | | ### `FlowSummary` Type: any. ### `GraphEdge` | Field | Type | Required | Description | | ------ | --------------- | -------- | ----------- | | `from` | integer (int64) | yes | | | `to` | integer (int64) | yes | | ### `GraphNode` | Field | Type | Required | Description | | ------------- | ----------------------- | -------- | ----------- | | `created_at` | integer (int64) | yes | | | `dynamic_key` | string | yes | | | `end_time` | integer or null (int64) | no | | | `external_id` | string | yes | | | `id` | integer (int64) | yes | | | `name` | string | yes | | | `start_time` | integer or null (int64) | no | | | `state` | [`State`](#state) | yes | | ### `HeartbeatRequest` | Field | Type | Required | Description | | ----------- | --------------- | -------- | ----------- | | `engine_id` | string | yes | | | `run_id` | integer (int64) | yes | | ### `Id` Type: string. ### `Log` | Field | Type | Required | Description | | ------------- | ----------------------- | -------- | ----------- | | `id` | integer (int64) | yes | | | `level` | integer (int32) | yes | | | `logger` | string | yes | | | `message` | string | yes | | | `run_id` | integer (int64) | yes | | | `task_run_id` | integer or null (int64) | no | | | `timestamp` | [`i64`](#i64) | yes | | ### `LogsPage` | Field | Type | Required | Description | | ------------- | ----------------------- | -------- | ----------- | | `items` | [`Log`](#log)[] | yes | | | `next_cursor` | integer or null (int64) | no | | ### `PrefilterBody` | Field | Type | Required | Description | | ------ | --------------- | -------- | ----------- | | `skip` | array of string | yes | | ### `PreviewBody` Type: any. ### `PreviewResponse` | Field | Type | Required | Description | | ---------- | ------------------------ | -------- | ----------- | | `next` | array of integer (int64) | yes | | | `timezone` | string | yes | | ### `ProjectedFire` A fire past the look-ahead, computed from the schedule; no run exists for it yet. | Field | Type | Required | Description | | ---------------- | ----------------------- | -------- | ------------ | | `projected` | boolean | yes | Always true. | | `schedule_id` | integer (int64) | yes | | | `scheduled_time` | integer (int64) | yes | | | `skipped` | boolean | yes | | | `skipped_at` | integer or null (int64) | no | | | `skipped_by` | string or null | no | | ### `ReleaseRequest` | Field | Type | Required | Description | | -------- | --------------- | -------- | ----------- | | `lease` | integer (int64) | yes | | | `run_id` | integer (int64) | yes | | ### `ReportRequest` | Field | Type | Required | Description | | ----------- | --------------- | -------- | ----------- | | `engine_id` | string | yes | | | `events` | array of object | yes | | | `run_id` | integer (int64) | yes | | ### `ReportResponse` | Field | Type | Required | Description | | ---------- | --------------- | -------- | ----------- | | `applied` | integer | yes | | | `cancel` | boolean | yes | | | `last_seq` | integer (int64) | yes | | | `skipped` | integer | yes | | ### `Resource` What an event is about. | Field | Type | Required | Description | | ------ | ------ | -------- | ----------------------------------------------------------- | | `id` | string | yes | | | `kind` | string | yes | `run`, `task_run`, `flow`, `schedule`, `rule`, or `custom`. | | `name` | string | no | | ### `ResumeBody` | Field | Type | Required | Description | | ------- | ---- | -------- | ------------------------------------------------------------------------ | | `input` | any | yes | The answer handed to `wait_for_input` on the resumed attempt (any JSON). | ### `RouteSpec` | Field | Type | Required | Description | | -------- | -------------- | -------- | -------------------------------------------------------------------------- | | `id` | integer | yes | | | `method` | string | yes | | | `path` | string | yes | | | `source` | string or null | no | Where the handler is defined (module and function), for the settings page. | ### `RuleAction` One rule action. `kind` selects the variant; other fields are optional. | Field | Type | Required | Description | | ------------ | --------------- | -------- | ------------------------------------------------------------------------------------------------------- | | `body` | string or null | no | | | `callable` | string or null | no | Name of the Python callable for `call` actions (code rules). | | `flow` | string or null | no | | | `headers` | object | no | | | `kind` | string | no | `run_flow`, `cancel_run`, `set_state`, `pause_schedule`, `resume_schedule`, `webhook`, `email`, `call`. | | `message` | string or null | no | | | `method` | string or null | no | | | `parameters` | object | no | | | `state_type` | string or null | no | | | `subject` | string or null | no | | | `to` | array of string | no | | | `url` | string or null | no | | ### `RuleBody` Type: any. ### `RuleClock` Clock of a clock-armed proactive rule: a cron expression in a timezone. | Field | Type | Required | Description | | ------ | -------------- | -------- | ----------- | | `cron` | string | no | | | `tz` | string or null | no | | ### `RuleFiring` | Field | Type | Required | Description | | ----------- | ----------------------- | -------- | ----------- | | `event_id` | integer or null (int64) | no | | | `id` | integer (int64) | yes | | | `outcomes` | array of object | yes | | | `rule_id` | integer (int64) | yes | | | `run_id` | integer or null (int64) | no | | | `timestamp` | [`i64`](#i64) | yes | | ### `RuleMatch` Match clause of a rule. | Field | Type | Required | Description | | --------- | --------------- | -------- | --------------------------------------- | | `events` | array of string | no | Event names, or prefixes ending in `*`. | | `flows` | array of string | no | | | `project` | string or null | no | | | `states` | array of string | no | | | `tags` | array of string | no | | ### `RulePatch` | Field | Type | Required | Description | | ------------------ | ----------------------- | -------- | ----------- | | `allow_self` | boolean or null | no | | | `at` | object | no | | | `cooldown_seconds` | number or null (double) | no | | | `do` | array or null | no | | | `enabled` | boolean or null | no | | | `max_per_minute` | integer or null (int64) | no | | | `name` | string or null | no | | | `once` | string or null | no | | | `unless` | object | no | | | `when` | object | no | | | `within` | number or null (double) | no | | ### `RuleRow` Type: any. ### `RuleSpec` | Field | Type | Required | Description | | ------------------ | --------------------------------- | -------- | ------------------------------------------------------------------ | | `allow_self` | boolean | no | | | `at` | null or [`RuleClock`](#ruleclock) | no | | | `cooldown_seconds` | number (double) | no | | | `do` | [`RuleAction`](#ruleaction)[] | no | | | `max_per_minute` | integer (int64) | no | | | `once` | string | no | `per_run` or `never`. | | `unless` | null or [`RuleMatch`](#rulematch) | no | | | `when` | [`RuleMatch`](#rulematch) | no | | | `within` | number or null (double) | no | Seconds after the arming event (or look-back before a clock tick). | ### `Run` | Field | Type | Required | Description | | ---------------- | ----------------------- | -------- | --------------------------------------------------------------------- | | `attempt` | integer (int64) | no | | | `backfill_id` | integer or null (int64) | no | | | `crash_count` | integer (int32) | yes | | | `created_at` | [`i64`](#i64) | yes | | | `created_by` | string | no | | | `end_time` | null or [`i64`](#i64) | no | | | `engine_id` | string or null | no | | | `engine_pid` | integer or null (int64) | no | | | `external_id` | [`Id`](#id) | yes | | | `failure_count` | integer (int32) | yes | | | `flow_id` | integer (int64) | yes | | | `flow_name` | string | yes | | | `group` | string | no | The flow's group, read through the flow: never stored on the run. | | `id` | integer (int64) | yes | | | `name` | string | yes | | | `parameters` | object | no | | | `parent_run_id` | integer or null (int64) | no | | | `priority` | integer (int64) | no | | | `project` | string | yes | | | `report_seq` | integer (int64) | no | | | `schedule_id` | integer or null (int64) | no | | | `scheduled_time` | null or [`i64`](#i64) | no | | | `start_time` | null or [`i64`](#i64) | no | | | `state` | [`State`](#state) | yes | | | `tags` | array of string | no | | | `task_counts` | object | no | Task runs of this run counted by state type; empty until tasks exist. | | `total_run_time` | null or [`i64`](#i64) | no | | ### `RunGraph` | Field | Type | Required | Description | | -------- | --------------------------- | -------- | ----------- | | `edges` | [`GraphEdge`](#graphedge)[] | yes | | | `nodes` | [`GraphNode`](#graphnode)[] | yes | | | `run_id` | integer (int64) | yes | | ### `RunsPage` | Field | Type | Required | Description | | ------------- | ----------------------- | -------- | ----------- | | `items` | [`Run`](#run)[] | yes | | | `next_cursor` | integer or null (int64) | no | | ### `Schedule` One of: object, object, object. ### `ScheduleBody` Type: any. ### `ScheduleDecl` Type: any. ### `SchedulePatchBody` | Field | Type | Required | Description | | ------------- | ----------------------------------------- | -------- | ----------- | | `anchor` | integer or null (int64) | no | | | `catchup` | null or [`CatchupPolicy`](#catchuppolicy) | no | | | `catchup_max` | integer or null (int64) | no | | | `cron` | string or null | no | | | `day_or` | boolean or null | no | | | `interval` | number or null (double) | no | | | `persist` | boolean or null | no | | | `rrule` | string or null | no | | | `timezone` | string or null | no | | ### `ScheduleRow` Placeholders for later phases; defined now so the schema and API types are stable from the start. A stored schedule row: the schedule itself plus policy and bookkeeping. | Field | Type | Required | Description | | --------------- | --------------------------------- | -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `active` | boolean | yes | | | `catchup` | [`CatchupPolicy`](#catchuppolicy) | yes | | | `catchup_max` | integer (int64) | yes | | | `code_key` | string or null | no | | | `created_at` | [`i64`](#i64) | yes | | | `external_id` | [`Id`](#id) | yes | | | `flow_id` | integer (int64) | yes | | | `id` | integer (int64) | yes | | | `next_fire` | null or [`i64`](#i64) | no | | | `paused_reason` | string or null | no | | | `paused_until` | null or [`i64`](#i64) | no | | | `persist` | boolean | yes | | | `schedule` | [`Schedule`](#schedule) | yes | | | `skipped` | integer (int64) | no | Future fires skipped by a person, counted by the scheduler (not stored). | | `source` | string | yes | `code` for schedules declared on the flow, `ui` for ones created in the interface, `mcp` for ones created by an agent. Startup reconciliation singles out `code` alone; every other value is left as it is. | | `updated_at` | [`i64`](#i64) | yes | | ### `ServerInfo` | Field | Type | Required | Description | | ------------ | --------------- | -------- | ----------- | | `engines` | array of any | yes | | | `home` | string | yes | | | `pid` | integer (int32) | yes | | | `queued` | integer | yes | | | `served_dir` | string or null | no | | | `started_at` | integer (int64) | yes | | | `stream_seq` | integer (int64) | yes | | | `url` | string | yes | | | `version` | string | yes | | ### `Settings` | Field | Type | Required | Description | | ------------------------ | --------------------------- | -------- | ----------- | | `catchup_default` | string | yes | | | `crash_retries_default` | integer (int64) | yes | | | `custom_routes` | [`RouteSpec`](#routespec)[] | yes | | | `database_bytes` | integer (int64) | yes | | | `database_path` | string | yes | | | `email_configured` | boolean | yes | | | `engine_max_runs` | integer (int32) | yes | | | `engine_saturation_risk` | boolean | yes | | | `home` | string | yes | | | `host` | string | yes | | | `max_engines` | integer | yes | | | `pid` | integer (int32) | yes | | | `port` | integer (int32) | yes | | | `resources` | object | yes | | | `retain_days` | integer (int64) | yes | | | `saturation_flows` | array of string | yes | | | `saturation_reason` | string or null | no | | | `secret_key_missing` | boolean | yes | | | `secret_key_present` | boolean | yes | | | `served_dir` | string or null | no | | | `version` | string | yes | | | `wal_bytes` | integer (int64) | yes | | ### `SettingsPatch` | Field | Type | Required | Description | | --------------- | ----------------------- | -------- | ----------- | | `crash_retries` | integer or null (int64) | no | | | `resources` | object or null | no | | | `retain_days` | integer or null (int64) | no | | ### `SkipBody` | Field | Type | Required | Description | | ------- | ------------------------ | -------- | --------------------------------------------------------------------- | | `by` | string or null | no | Who asked: `ui` or `api` (the default). | | `fires` | array of integer (int64) | no | Fire times to skip, in microseconds, as the upcoming list gives them. | | `next` | integer or null | no | Skip the next N fires not already skipped instead. | ### `SkipResponse` | Field | Type | Required | Description | | ------------ | ------------------------------------- | -------- | ------------------------------- | | `downstream` | [`DownstreamSkip`](#downstreamskip)[] | yes | | | `schedule` | [`ScheduleRow`](#schedulerow) | yes | | | `skipped` | array of integer (int64) | yes | The fires this request skipped. | ### `State` | Field | Type | Required | Description | | ----------- | ------------------------- | -------- | ----------- | | `details` | object | no | | | `message` | string or null | no | | | `name` | string | yes | | | `timestamp` | [`i64`](#i64) | yes | | | `type` | [`StateType`](#statetype) | yes | | ### `StateEntry` | Field | Type | Required | Description | | -------------- | ------- | -------- | ------------------------------------------------------------ | | `is_sub_state` | boolean | yes | | | `name` | string | yes | | | `state_type` | string | yes | For a sub-state, the type it belongs to; for a type, itself. | ### `StateType` One of: `Scheduled`, `Pending`, `Running`, `Completed`, `Failed`, `Cancelled`, `Crashed`, `Paused`, `Cancelling`. ### `TaskRun` | Field | Type | Required | Description | | ---------------- | --------------------- | -------- | -------------------------------------------------------------------- | | `crash_count` | integer (int32) | yes | | | `created_at` | [`i64`](#i64) | yes | | | `dynamic_key` | string | yes | | | `end_time` | null or [`i64`](#i64) | no | | | `external_id` | [`Id`](#id) | yes | | | `failure_count` | integer (int32) | yes | | | `flow_id` | integer (int64) | no | | | `flow_name` | string | no | | | `id` | integer (int64) | yes | | | `name` | string | yes | | | `parents` | [`Id`](#id)[] | no | External ids of task runs this one waited on (futures and wait_for). | | `project` | string | no | | | `run_id` | integer (int64) | yes | | | `run_name` | string | no | | | `start_time` | null or [`i64`](#i64) | no | | | `state` | [`State`](#state) | yes | | | `task_key` | string | yes | | | `total_run_time` | null or [`i64`](#i64) | no | | ### `TaskRunsPage` | Field | Type | Required | Description | | ------------- | ----------------------- | -------- | ----------- | | `items` | [`TaskRun`](#taskrun)[] | yes | | | `next_cursor` | integer or null (int64) | no | | ### `TransitionBody` | Field | Type | Required | Description | | --------- | ------------------------- | -------- | ----------- | | `details` | object | no | | | `force` | boolean | no | | | `message` | string or null | no | | | `name` | string or null | no | | | `type` | [`StateType`](#statetype) | yes | | ### `TransitionRejected` | Field | Type | Required | Description | | --------- | ------------------------- | -------- | ----------- | | `current` | null or [`State`](#state) | no | | | `error` | string | yes | | | `reason` | string | yes | | ### `UpcomingItem` One entry of the upcoming list: a run, or with `projected=N` a fire without one. One of: [`UpcomingRun`](#upcomingrun), [`ProjectedFire`](#projectedfire). ### `UpcomingRun` A materialized run in the upcoming list. Type: any. ### `VariableBody` | Field | Type | Required | Description | | ----------- | --------------- | -------- | ----------- | | `name` | string | yes | | | `overwrite` | boolean | no | | | `secret` | boolean | no | | | `tags` | array of string | no | | | `value` | object | yes | | ### `VariablePatch` | Field | Type | Required | Description | | -------- | --------------- | -------- | ----------- | | `secret` | boolean or null | no | | | `tags` | array or null | no | | | `value` | object | no | | ### `VariableRow` | Field | Type | Required | Description | | ------------ | --------------- | -------- | --------------------------------------------------------------- | | `created_at` | [`i64`](#i64) | yes | | | `name` | string | yes | | | `secret` | boolean | yes | | | `tags` | array of string | no | | | `updated_at` | [`i64`](#i64) | yes | | | `value` | object | yes | Plain JSON value, or `"********"` for secrets in API responses. | ### `VariableWithRaw` Type: any. ### `Vocabulary` | Field | Type | Required | Description | | ------------------- | ----------------------------- | -------- | ---------------------------------------------------------------------------------------------------------------------------------------- | | `events` | [`EventEntry`](#evententry)[] | yes | Every engine-emitted event, in catalogue order. | | `reserved_prefixes` | array of string | yes | Prefixes the engine owns. A name under one of these must be an entry in `events`; any other name is a custom event and is never checked. | | `states` | [`StateEntry`](#stateentry)[] | yes | Every value a rule's `states` accepts: the types then the sub-states. A rule naming a type also matches that type's sub-states. | ### `WorkItem` Work handed to an engine. | Field | Type | Required | Description | | ------------------ | --------------- | -------- | ----------------------------------- | | `cancel_requested` | boolean | yes | | | `external_id` | string | yes | | | `flow` | string | yes | | | `kind` | string | no | `run`, `hooks`, or `bulk_complete`. | | `options` | object | yes | | | `parameters` | object | yes | | | `payload` | object | no | | | `project` | string | yes | | | `run_id` | integer (int64) | yes | | | `run_name` | string | yes | | ### `WorkRequest` | Field | Type | Required | Description | | ------------ | --------------- | -------- | ------------------------------------------- | | `engine_id` | string | yes | | | `isolated` | boolean | no | | | `module` | string | yes | | | `nice` | integer (int32) | no | | | `pid` | integer (int32) | yes | | | `source_dir` | string | yes | | | `wait_ms` | integer (int64) | no | Long-poll wait in milliseconds (max 30000). | ### `WorkResponse` | Field | Type | Required | Description | | ------ | ------------------------------- | -------- | ----------- | | `exit` | boolean | no | | | `run` | null or [`WorkItem`](#workitem) | no | | ### `i64` Type: integer (int64). # MCP tools, resources and prompts The server implements the [Model Context Protocol](https://modelcontextprotocol.io) (JSON-RPC 2.0) so an agent can operate cereyan. The tool set is curated rather than a mirror of the HTTP API. Definitions live in `crates/server/src/mcp.rs`; this page is generated from a snapshot of what a client actually receives. Setup for Claude Code, Claude Desktop, and HTTP clients is in [Use cereyan with an AI agent](https://sercanatalik.github.io/cereyan/guides/agents/index.md). ## Transports | Transport | How | Authentication | | --------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | | stdio | `cereyan mcp`, started by the host; proxies every message to the running server found through `server.json` (`--url` and `--socket` override) | `CEREYAN_TOKEN`, or `--token` before the subcommand (`cereyan --token mcp`); with a Unix socket and no token it connects over the socket | | Streamable HTTP | `POST /mcp` with one JSON-RPC message per request; requests get a JSON reply, notifications get 202; `initialize` returns `Mcp-Session-Id` to send on later requests; `DELETE /mcp` ends the session; `GET /mcp` answers 405 | `Authorization: Bearer ` when the server has a token | Runs created through MCP record `created_by = mcp:` from the `initialize` handshake. ## Handshake `initialize` answers with protocol version `2025-06-18`, server name `cereyan` and its release version, and the capabilities `prompts`, `resources`, `tools`. It also carries the instructions the server gives the model: > cereyan runs Python pipelines on this machine. Use list_flows to see what can run, run_flow to start work, get_run and run_logs to follow it, and explain_failure when a run fails. Flows run on demand: a flow needs no schedule, and run_flow is how work usually starts. For work that should recur, list_schedules shows what is scheduled and the create, edit, delete, pause and resume schedule tools manage it. Writes take effect immediately. ## Tools Every tool description states its effect so a model can decide before calling. A tool returns one text content block holding the JSON whose top-level keys are listed as its response; a failure returns `isError: true` and a message instead. Rule creation is not exposed, and a schedule declared in a flow's code cannot be deleted through MCP: the next restart recreates it from the declaration, so pausing it is what lasts. ### Read-only tools (9) | Tool | Arguments | Returns | | ----------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `list_flows` | `project` (string) — Only flows of this project | List the registered flows with their project, parameters schema, tags, and any registration error. Response keys: `flows` | | `list_runs` | `flow` (string) `limit` (integer, 1 to 200, default `20`) `name` (string) — Exact run name `project` (string) `state_name` (string) — A named sub-state such as Late or AwaitingRetry `state_type` (string) — Scheduled, Pending, Running, Completed, Failed, Cancelled, Crashed, Paused, Cancelling | List runs, newest first, with optional filters. Response keys: `next_cursor`, `runs` | | `get_run` | `run_id` (integer, required) | One run with its state, parameters, timing, and task runs. Response keys: `run`, `task_runs` | | `run_logs` | `limit` (integer, 1 to 1000, default `200`) `min_level` (integer) `run_id` (integer, required) `search` (string) | Log lines of a run, oldest first. Filter by minimum level (10 debug, 20 info, 30 warning, 40 error) or a search string. Response keys: `logs`, `next_cursor` | | `list_events` | `flow_id` (integer) `limit` (integer, 1 to 500, default `50`) `name` (string) — Exact name or a prefix ending in * such as run.\* `run_id` (integer) | Recent events (run and task transitions, schedule changes, rule firings, custom events), newest first. Response keys: `events`, `next_cursor` | | `list_artifacts` | `flow` (string) `key` (string) `kind` (string) `limit` (integer, 1 to 200, default `50`) `project` (string) `run_id` (integer) | Artifacts across runs, newest first, with their run, flow, and project. Response keys: `artifacts`, `next_cursor` | | `list_rules` | none | The rules (reactive and proactive) with their match, actions, guards, and fire counts. Response keys: `rules` | | `list_schedules` | `flow` (string) — Flow name, or project/flow when the name exists in several projects `project` (string) — Only schedules of this project | The schedules of one flow or of every flow: the spec, whether it is active, when it next fires, and whether it was declared in the flow's code, created in the interface, or created by an agent. Response keys: `schedules` | | `explain_failure` | `run_id` (integer, required) | Everything needed to diagnose a run in one call: the run, its failed or crashed task runs, the last warning-or-above log lines, and the run's events. Response keys: `error_logs`, `events`, `failed_task_runs`, `run`, `verdict` | ### Tools that change state (10) | Tool | Arguments | Returns | | ----------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `run_flow` | `flow` (string, required) — Flow name, or project/flow when the name exists in several projects `name` (string) — Optional run name `parameters` (object) — Flow parameters as JSON `tags` (array of string) | Start a run of a flow now. Creates the run immediately and returns without waiting; follow it with get_run. Parameters are validated against the flow's schema. Response keys: `note`, `run` | | `cancel_run` | `run_id` (integer, required) | Cancel a run. A queued run is cancelled at once; a running run is asked to stop and killed after the grace period. Response keys: `run` | | `resume_run` | `input` (any JSON, required) — The answer, any JSON `run_id` (integer, required) | Answer a Paused run's wait_for_input question and schedule its next attempt. The answer can be any JSON. Response keys: `run` | | `backfill` | `concurrency` (integer, default `1`) `dry_run` (boolean, default `true`) `end` (string, required) `extra_parameters` (object) `flow` (string, required) — Flow name, or project/flow when the name exists in several projects `interval` (string) — Seconds or a duration such as 1d or 12h (default 1d) `parameter` (string, required) `reverse` (boolean) `start` (string, required) — YYYY-MM-DD or RFC 3339 | Create one run per value of a date or datetime parameter between start and end. Defaults to a dry run that only reports how many runs would be created; pass dry_run false to create them. Can create thousands of runs. Response keys: dry run: `dry_run`, `first`, `flow`, `interval_seconds`, `last`, `note`, `parameter`, `runs` dry_run false: `backfill`, `dry_run` | | `create_schedule` | `anchor` (integer) — Microseconds UTC the interval counts from; defaults to now `catchup` (string, default `"skip"`) `catchup_max` (integer, default `100`) `cron` (string) — Five-field cron expression, for kind cron `day_or` (boolean) — For cron, OR day-of-month with day-of-week (default true) `flow` (string, required) — Flow name, or project/flow when the name exists in several projects `interval` (number) — Seconds between fires, for kind interval `kind` (string, required) — Which kind of schedule `project` (string) `rrule` (string) — RFC 5545 RRULE, for kind rrule `timezone` (string) — IANA name such as Europe/Istanbul; UTC when unset | Make a flow run repeatedly. To run a flow once, now, use run_flow instead: a flow needs no schedule, and running on demand is the normal case. Returns the schedule and the next few times it will fire. Response keys: `next_fires`, `schedule` | | `edit_schedule` | `anchor` (integer) `catchup` (string) `catchup_max` (integer) `cron` (string) `day_or` (boolean) `interval` (number) `rrule` (string) `schedule_id` (integer, required) `timezone` (string) | Retime an existing schedule. Editing one that was declared in the flow's code lasts until the server restarts, when the declaration in the Python source applies again, and the result says so. Returns the schedule and the next few times it will fire. Response keys: `next_fires`, `schedule` | | `delete_schedule` | `schedule_id` (integer, required) | Remove a schedule that was created in the interface or by an agent. A schedule declared in the flow's code cannot be removed this way, because the next restart recreates it from the declaration; pause_schedule stops that one durably. Response keys: `deleted`, `schedule_id` | | `pause_schedule` | `schedule_id` (integer, required) | Pause a schedule so it stops creating runs until resumed. Response keys: `schedule` | | `resume_schedule` | `schedule_id` (integer, required) | Resume a paused schedule. Response keys: `schedule` | | `set_variable` | `name` (string, required) `secret` (boolean, default `false`) `tags` (array of string) `value` (any JSON, required) — Any JSON | Create or overwrite a variable. Secrets are encrypted at rest and never returned in plain text. Response keys: `variable` | ### Fields of the lists a tool returns Recorded from real responses, so a model knows what it gets without a second call. | Tool | Key | Item fields | | ----------------- | ------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `explain_failure` | `error_logs` | `id`, `level`, `logger`, `message`, `run_id`, `task_run_id`, `timestamp` | | `explain_failure` | `events` | `external_id`, `flow_id`, `id`, `name`, `occurred`, `payload`, `related`, `resource`, `run_id`, `seq` | | `get_run` | `task_runs` | `crash_count`, `created_at`, `dynamic_key`, `end_time`, `external_id`, `failure_count`, `flow_id`, `flow_name`, `id`, `name`, `parents`, `project`, `run_id`, `run_name`, `start_time`, `state`, `task_key`, `total_run_time` | | `list_artifacts` | `artifacts` | `created_at`, `data`, `external_id`, `flow_name`, `id`, `key`, `kind`, `project`, `run_id`, `run_name`, `task_run_id`, `updated_at` | | `list_events` | `events` | `external_id`, `flow_id`, `id`, `name`, `occurred`, `payload`, `related`, `resource`, `run_id`, `seq` | | `list_flows` | `flows` | `description`, `error`, `id`, `live`, `name`, `options`, `parameter_schema`, `project`, `tags` | | `list_runs` | `runs` | `attempt`, `backfill_id`, `crash_count`, `created_at`, `created_by`, `end_time`, `engine_id`, `engine_pid`, `external_id`, `failure_count`, `flow_id`, `flow_name`, `group`, `id`, `name`, `parameters`, `parent_run_id`, `priority`, `project`, `report_seq`, `schedule_id`, `scheduled_time`, `start_time`, `state`, `tags`, `task_counts`, `total_run_time` | | `list_schedules` | `schedules` | `active`, `catchup`, `catchup_max`, `flow`, `id`, `next_fire`, `paused_reason`, `paused_until`, `project`, `schedule`, `source` | | `run_logs` | `logs` | `id`, `level`, `logger`, `message`, `run_id`, `task_run_id`, `timestamp` | ## Resources `resources/list` returns none: both resources are templates, listed by `resources/templates/list` and fetched with `resources/read`. | URI template | Name | Content | MIME type | | ------------------------------- | ------------- | -------------------------- | ------------------ | | `cereyan://runs/{id}/logs` | Run logs | Log lines of a run as JSON | `application/json` | | `cereyan://runs/{id}/artifacts` | Run artifacts | Artifacts of a run as JSON | `application/json` | ## Prompts | Prompt | Arguments | Purpose | | -------------- | ------------------------------- | ------------------------------------------------ | | `diagnose_run` | `run_id` (The run id, required) | Explain why a run failed and what to do about it | `diagnose_run` renders one message: > **user**: Run of a cereyan pipeline needs a diagnosis. Call the explain_failure tool with run_id , read the failed task runs, the error log lines, and the events, then summarise the root cause in two sentences and suggest one concrete next step (rerun with run_flow, fix the code, or adjust a schedule). # Python API Everything below is importable from `cereyan` unless a module path is shown. This page is rendered from the docstrings in `python/cereyan/` by mkdocstrings; edit the docstrings, not this page. ## Decorators and the App ### flow ``` flow(fn: Callable | None = None, *, name: str | None = None, description: str | None = None, tags: Iterable[str] = (), group: str | None = None, run_name: str | Callable[..., str] | None = None, isolated: bool = False, log_prints: bool = False, app=None, **options) ``` Register a function as a flow on `app` or on the default App. Use as `@flow` or `@flow(...)`. Calling the returned `Flow` runs it: offline, the run is recorded in the local store; while a server is up, the run is handed to the server and its logs stream back. Parameters come from the function's type hints and are coerced before the body runs. Parameters: | Name | Type | Description | Default | | ----------------- | ---------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `fn` | `Callable` | The function to wrap; supplied by the decorator syntax. | `None` | | `name` | \`str | None\` | Flow name; defaults to the function name. A flow's identity is (project, name). | | `description` | \`str | None\` | Shown in the UI; defaults to the function's docstring. | | `tags` | `Iterable[str]` | Tags copied onto every run. | `()` | | `group` | \`str | None\` | The group this flow is listed under in the UI; defaults to the flow's project. Groups are a flat axis, not a level inside the project: flows in different projects declaring the same group form one group, and a group named after a project merges with the flows that default to it. | | `run_name` | \`str | Callable[..., str] | None\` | | `isolated` | `bool` | Run every run in a fresh engine process instead of the warm pool. | `False` | | `log_prints` | `bool` | Tee print output into the run log at INFO. | `False` | | `app` | \`App | None\` | The App to register on; defaults to the default App of the defining module. | | `schedule` | \`Cron | Interval | RRule | | `schedules` | \`Cron | Interval | RRule | | `retries` | `int` | How many times a failed run is retried. | *required* | | `retry_delay` | \`float | list[float] | exponential\` | | `timeout_seconds` | \`float | None\` | End the run as Failed with sub-state TimedOut after this many seconds. | | `crash_retries` | \`int | None\` | How many times a run whose engine died is rerun before it is marked Failed; defaults to [defaults] crash_retries in cereyan.toml, then 5. | | `priority` | `int` | Dispatch order among queued runs, higher first; never preempts a running run. A negative value also raises the engine's OS niceness on Linux and macOS. | *required* | | `max_concurrent` | \`int | None\` | Cap on runs of this flow in Pending or Running at once, implemented as a resource named after the flow; unlimited by default. | | `on_overlap` | `str` | What a new run does when max_concurrent is reached: "enqueue" (wait as AwaitingResource, the default), "skip" (end Skipped), or "cancel_new" (end Cancelled). | *required* | | `resources` | \`dict[str, float] | None\` | Named resources and the amount each run holds, for example {"db": 1}; a run waits as AwaitingResource until they are free. | | `after` | \`str | tuple | list[str] | | `batch_key` | \`str | None\` | With a list in after, the parameter whose value identifies a batch; this flow runs once per value after every upstream has a completed run for it. | | `disable_after` | \`tuple[int, int, int] | None\` | (count, window_seconds, persist_seconds): after count failures within window_seconds the flow's schedules pause for persist_seconds and resume automatically. | | `runner` | \`ThreadRunner | ProcessRunner | None\` | | `bulk_complete` | \`Callable | None\` | bulk_complete(values) -> set called once by a backfill; runs for the returned values are recorded as Skipped. | | `on_completion` | `Iterable[Callable]` | Hooks hook(flow, run, state) called after a run completes. | *required* | | `on_failure` | `Iterable[Callable]` | Hooks called after a run fails. | *required* | | `on_crashed` | `Iterable[Callable]` | Hooks called after a run crashes. | *required* | | `on_cancellation` | `Iterable[Callable]` | Hooks called after a run is cancelled. | *required* | Returns: | Type | Description | | ------ | --------------------------------------------------------- | | `Flow` | The flow wrapping fn; call it like the original function. | ### Flow ``` Flow(fn: Callable, *, name: str | None = None, description: str | None = None, tags: Iterable[str] = (), group: str | None = None, run_name: str | Callable[..., str] | None = None, isolated: bool = False, log_prints: bool = False, schedule: Any = None, schedules: Any = None, retries: int = 0, retry_delay: Any = 0, timeout_seconds: float | None = None, crash_retries: int | None = None, priority: int = 0, max_concurrent: int | None = None, on_overlap: str = 'enqueue', resources: dict[str, float] | None = None, after: Any = None, batch_key: str | None = None, disable_after: tuple | None = None, runner: Any = None, bulk_complete: Callable | None = None, on_completion: Iterable[Callable] = (), on_failure: Iterable[Callable] = (), on_crashed: Iterable[Callable] = (), on_cancellation: Iterable[Callable] = ()) ``` A registered flow: the wrapped function plus its options, parameters, and identity. Instances are created by `flow`; call one to start a run. Attributes mirror the decorator options, and `parameters` and `schema` describe the parameters derived from the function's type hints. ### task ``` task(fn: Callable | None = None, **options) ``` Register a function as a task: each call inside a flow becomes a task run. Use as `@task` or `@task(...)`. Outside a flow the function runs as-is. Inside a flow each call is recorded as a task run with a dynamic key such as `load-0`, and `submit` and `map` run calls concurrently on the flow's runner. Parameters: | Name | Type | Description | Default | | ----------------- | -------------------- | ------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------- | | `fn` | `Callable` | The function to wrap; supplied by the decorator syntax. | `None` | | `name` | \`str | None\` | Task name; defaults to the function name. | | `description` | \`str | None\` | Shown in the UI; defaults to the function's docstring. | | `tags` | `Iterable[str]` | Tags recorded on every task run. | *required* | | `log_prints` | `bool` | Tee print output into the run log at INFO. | *required* | | `retries` | `int` | How many times a failed task run is retried. | *required* | | `retry_delay` | \`float | list[float] | exponential\` | | `timeout_seconds` | \`float | None\` | Fail the task run with sub-state TimedOut after this many seconds. | | `output` | \`Target | Callable[..., Target] | None\` | | `cache` | \`CachePolicy | None\` | INPUTS, SOURCE, or INPUTS + SOURCE: reuse a persisted result when the arguments, the source, or both are unchanged. Requires persist_result=True. | | `cache_expires` | \`timedelta | None\` | A timedelta after which a cached result is stale. | | `persist_result` | `bool` | Store the return value under /storage so cached and replayed runs can read it. | *required* | | `serializer` | `str` | "pickle" (default) or "json" for persisted results. | *required* | | `resources` | \`dict[str, float] | None\` | Named resources and the amount each task run holds, for example {"gpu": 1}. | | `on_completion` | `Iterable[Callable]` | Hooks hook(task, run, state) called after a task run completes. | *required* | | `on_failure` | `Iterable[Callable]` | Hooks called after a task run fails. | *required* | | `on_cancellation` | `Iterable[Callable]` | Hooks called after a task run is cancelled. | *required* | Returns: | Type | Description | | ------ | --------------------- | | `Task` | The task wrapping fn. | ### Task ``` Task(fn: Callable, *, name: str | None = None, description: str | None = None, tags: Iterable[str] = (), log_prints: bool = False, retries: int = 0, retry_delay: Any = 0, timeout_seconds: float | None = None, output: Any = None, cache: CachePolicy | None = None, cache_expires: timedelta | None = None, persist_result: bool = False, serializer: str = 'pickle', resources: dict[str, float] | None = None, on_completion: Iterable[Callable] = (), on_failure: Iterable[Callable] = (), on_cancellation: Iterable[Callable] = ()) ``` A registered task: the wrapped function plus its options. Instances are created by `task`. Calling one inside a flow records a task run; `submit` and `map` return `Future` objects instead of waiting. #### submit ``` submit(*args, wait_for=None, **kwargs) ``` Run the task concurrently; returns a Future. #### map ``` map(iterable, *, wait_for=None, **static) ``` Submit one task run per element; returns a list of Futures. ### App ``` App(name: str | None = None, *, source_file: str | None = None) ``` Registry of flows and routes. Its name is the project every flow belongs to. #### flow ``` flow(fn: Callable | None = None, **options) ``` Decorator equivalent to `@flow` bound to this App. #### task ``` task(fn: Callable | None = None, **options) ``` Decorator equivalent to `@task`; tasks are not tied to an App, this exists for symmetry. #### register ``` register(flow: Flow) -> None ``` Add a flow to this App; raises `FlowRegistrationError` when another flow of the same name is registered. #### route ``` route(method: str, path: str) ``` Register a custom HTTP route handled by the decorated function. Parameters: | Name | Type | Description | Default | | -------- | ----- | --------------------------------------------------------------------------- | ---------- | | `method` | `str` | HTTP method. | *required* | | `path` | `str` | Path template with {name} placeholders, for example "/api/ext/orders/{id}". | *required* | #### get ``` get(path: str) ``` Register a `GET` route; see `route`. #### post ``` post(path: str) ``` Register a `POST` route; see `route`. #### put ``` put(path: str) ``` Register a `PUT` route; see `route`. #### delete ``` delete(path: str) ``` Register a `DELETE` route; see `route`. #### patch ``` patch(path: str) ``` Register a `PATCH` route; see `route`. #### rule ``` rule(on=None, flow=None, tags=None, states=None, unless=None, within=None, at=None, tz=None, **guards) ``` Register a code rule whose action calls the decorated `fn(event, run)`. A reactive rule fires when an event matching `on` happens. A proactive rule fires when the event named by `unless` does *not* happen: either `within` seconds of the `on` event (event-armed), or by each tick of the cron expression `at` in timezone `tz` (clock-armed, no `on`). Event and state names are checked here, so a rule that could never fire raises at import rather than sitting silent. Use the constants in `cereyan.events` and `cereyan.states` to get them right the first time; a custom event name outside the reserved prefixes is always accepted. A value in `states` matches the run's state type or its sub-state name, so `states=["Scheduled"]` also covers Late and AwaitingRetry. See `cereyan.rules.register` for the full argument list. #### serve ``` serve(host: str | None = None, port: int | None = None, **options) -> int ``` Serve the flows and routes registered so far, blocking until stopped. Host and port given here rank below the CLI flags and environment and above `cereyan.toml`. ### get_default_app ``` get_default_app(source_file: str | None = None) -> App ``` The default App, created on first use and named after the directory of the module that first needed it. ## The current run `cereyan.runtime` is a live view of the run executing in this thread. Each attribute is `None` outside a run, so a task called directly from a test reads `None` rather than raising. | Attribute | Type | Carries | | -------------------------- | ---------------- | -------------------------------------------------------------------------------- | | `cereyan.runtime.run` | `RunContext` | `id`, `external_id`, `name`, `flow`, `parameters`, and `flow_name` and `project` | | `cereyan.runtime.task_run` | `TaskRunContext` | `id` (the external UUID), `name`, `task_key`, `dynamic_key` | | `cereyan.runtime.flow` | `Flow` | The flow of the current run | ``` from cereyan import flow, task, runtime @task def record() -> str: return f"{runtime.run.name}/{runtime.task_run.dynamic_key}" @flow def report() -> str: return f"{runtime.flow.name}#{runtime.run.id}: {record()}" assert runtime.run is None assert report().endswith("/record-0") # "report#1: /record-0" ``` ## Schedules and retry delays ### Cron ``` Cron(cron: str, timezone: str | None = None, day_or: bool = True, catchup: str = 'skip', catchup_max: int = 100, key: str | None = None) ``` A cron schedule evaluated by wall clock in `timezone`. #### to_json ``` to_json() -> dict[str, Any] ``` The schedule as the server stores it. ### Interval ``` Interval(interval: float | timedelta, anchor: datetime | None = None, timezone: str | None = None, catchup: str = 'skip', catchup_max: int = 100, key: str | None = None) ``` Fire every `interval` (seconds or timedelta) from `anchor`. Intervals under a day are elapsed time; longer intervals keep their local wall-clock time across DST changes. #### to_json ``` to_json() -> dict[str, Any] ``` The schedule as the server stores it; raises `ValueError` for a non-positive interval. ### RRule ``` RRule(rrule: str, timezone: str | None = None, catchup: str = 'skip', catchup_max: int = 100, key: str | None = None) ``` An iCalendar recurrence rule set; must include DTSTART. #### to_json ``` to_json() -> dict[str, Any] ``` The schedule as the server stores it. ### exponential ``` exponential(base: float = 1.0, jitter: float = 0.0, maximum: float = 3600.0) ``` Retry delay growing as `base * 2**attempt` with optional jitter. #### delay ``` delay(attempt: int) -> float ``` Seconds to wait before retry number `attempt` (0-based), capped at `maximum`. ## Targets, caching and results ### Target Bases: `Protocol` Anything with `exists()`: the protocol a task's `output=` must satisfy. #### exists ``` exists() -> bool ``` `True` when the output this target stands for is already present. ### LocalTarget ``` LocalTarget(path: str | PathLike, *, mkdir: bool = True) ``` A file on the local filesystem written atomically. #### exists ``` exists() -> bool ``` `True` when the file exists. #### open ``` open(mode: str = 'r', **kwargs: Any) ``` Open the file; write modes write to a temporary path that replaces the target on close, so readers never see a partial file. #### remove ``` remove() -> None ``` Delete the file if it exists. #### temporary_path ``` temporary_path() ``` Yield a temporary path; on success it is renamed onto the target. ### CachePolicy Bases: `Flag` `INPUTS` and `SOURCE` are the members of `CachePolicy`, exported at the top level so `cache=INPUTS + SOURCE` reads naturally. ## Concurrency ### Future ``` Future(task, external_id: str, dynamic_key: str) ``` Handle to a submitted task run. #### done ``` done() -> bool ``` `True` once the task run has finished, in any state. #### wait ``` wait(timeout: float | None = None) -> None ``` Block until the task run finishes or `timeout` seconds pass. #### result ``` result(timeout: float | None = None) -> Any ``` The task's return value, blocking until it is available; re-raises the task's exception on failure. #### exception ``` exception(timeout: float | None = None) ``` The exception the task raised, or `None`; blocks like `result`. ### ThreadRunner ``` ThreadRunner(max_workers: int | None = None) ``` Bases: `BaseRunner` Runs submitted tasks on a bounded set of threads with the run context propagated. A task that waits on a child while every worker is busy gets a warning and a temporary extra worker so the wait cannot deadlock. ### ProcessRunner ``` ProcessRunner(max_workers: int | None = None) ``` Bases: `BaseRunner` Runs each submitted task in a spawned process; arguments and results must be picklable and the task function importable. Timeouts terminate the worker. ## Logging, events and artifacts ### get_run_logger ``` get_run_logger() -> Logger ``` A logger that carries the current run and task-run in its records. ### emit_event ``` emit_event(name: str, payload: dict | None = None, resource: dict | None = None) -> None ``` Record an event. Inside a run the resource defaults to that run; outside a run the event goes straight to the store or, when a server is up, to its API. Any name is allowed except one under a reserved prefix (see `cereyan.events.RESERVED_PREFIXES`) that the engine does not emit, which is always a mistake rather than a custom event. Raises: | Type | Description | | ------------ | ---------------------------------------------------------------------------- | | `ValueError` | When the name is empty, not a string, or shadows the engine's own namespace. | ### EventName Bases: `str` An engine event name; a `str` carrying the exact wire name. Attributes: | Name | Type | Description | | ---------------- | ----------------- | -------------------------------------------------------------------------------------- | | `resource` | `str` | The resource kind the event hangs off (run, task_run, flow, schedule, resource, rule). | | `when` | `str` | One sentence on when the engine records it. | | `payload_fields` | `tuple[str, ...]` | The payload keys the emit site sets. | ### EventGroup ``` EventGroup(prefix: str, names: 'list[EventName]') ``` The events under one reserved prefix, as attributes. `events.run.failed` is `"run.failed"` and `events.run.any` is `"run.*"`, which matches every event under the prefix. ### StateName Bases: `str` A state type or named sub-state; a `str` carrying the wire name. Attributes: | Name | Type | Description | | -------------- | ------ | ------------------------------------------------------------------ | | `state_type` | `str` | For a sub-state, the type it belongs to; for a state type, itself. | | `is_sub_state` | `bool` | True for a named sub-state such as Late. | ### create_markdown ``` create_markdown(text: str, key: str | None = None) -> str ``` Attach a markdown artifact to the current run or task run. Parameters: | Name | Type | Description | Default | | ------ | ----- | ------------------------------------------ | ------------------------------------------------------------------------ | | `text` | `str` | Markdown source, rendered on the run page. | *required* | | `key` | \`str | None\` | Optional stable key; artifacts sharing a key form a history across runs. | Returns: | Type | Description | | ----- | ---------------- | | `str` | The artifact id. | Raises: | Type | Description | | -------------- | ------------------------------------------------- | | `CereyanError` | Outside a run, or when the artifact exceeds 1 MB. | ### create_table ``` create_table(rows: list[dict] | list[list], key: str | None = None, columns: list[str] | None = None) -> str ``` Attach a table artifact. Parameters: | Name | Type | Description | Default | | --------- | ------------ | ------------ | ------------------------------------------------------------------------------------------------------------- | | `rows` | \`list[dict] | list[list]\` | A list of dicts (columns inferred from the keys, in first-seen order) or a list of lists (then pass columns). | | `key` | \`str | None\` | Optional stable key. | | `columns` | \`list[str] | None\` | Column names; required for list-of-lists rows. | Returns: | Type | Description | | ----- | ---------------- | | `str` | The artifact id. | ### create_progress ``` create_progress(percent: float, key: str | None = None, label: str | None = None) -> str ``` Attach a progress bar artifact. Parameters: | Name | Type | Description | Default | | --------- | ------- | ----------------------------------------------------- | ---------------------------------------------------------------- | | `percent` | `float` | Completion from 0 to 100; values outside are clamped. | *required* | | `key` | \`str | None\` | Optional stable key; pass the same key to update_progress later. | | `label` | \`str | None\` | Text shown next to the bar. | Returns: | Type | Description | | ----- | ---------------- | | `str` | The artifact id. | ### update_progress ``` update_progress(key: str, percent: float, label: str | None = None) -> str ``` Record a new value for the progress artifact with `key`. Each update is a new artifact row under the same key, so the history is kept. Parameters: | Name | Type | Description | Default | | --------- | ------- | ----------------------------------------------------- | --------------------------- | | `key` | `str` | The key given when the progress artifact was created. | *required* | | `percent` | `float` | Completion from 0 to 100. | *required* | | `label` | \`str | None\` | Text shown next to the bar. | Returns: | Type | Description | | ----- | ---------------- | | `str` | The artifact id. | ### create_link ``` create_link(url: str, text: str | None = None, key: str | None = None) -> str ``` Attach a link artifact. Parameters: | Name | Type | Description | Default | | ------ | ----- | ---------------- | ------------------------------- | | `url` | `str` | The link target. | *required* | | `text` | \`str | None\` | Link text; defaults to the URL. | | `key` | \`str | None\` | Optional stable key. | Returns: | Type | Description | | ----- | ---------------- | | `str` | The artifact id. | ### create_image ``` create_image(url_or_bytes: str | bytes, key: str | None = None, media_type: str = 'image/png') -> str ``` Attach an image artifact from a URL or raw bytes. Parameters: | Name | Type | Description | Default | | -------------- | ----- | ------------------------------------ | -------------------------------------------------------- | | `url_or_bytes` | \`str | bytes\` | An image URL, or the image bytes to embed as a data URI. | | `key` | \`str | None\` | Optional stable key. | | `media_type` | `str` | MIME type used when bytes are given. | `'image/png'` | Returns: | Type | Description | | ----- | ---------------- | | `str` | The artifact id. | ## Variables ### Variable Named JSON values shared by every project on the machine, optionally encrypted. Names match `[a-z0-9][a-z0-9_./-]*` and values are limited to 64 KB. Offline, values go straight to the store; while a server holds the store, they go through its API. Secrets are encrypted with the key kept at `/secret.key` and are masked in the API and the UI. #### get ``` get(name: str, default: Any = None) -> Any ``` Read a variable. Parameters: | Name | Type | Description | Default | | --------- | ----- | ------------------------------------------ | ---------- | | `name` | `str` | The variable name. | *required* | | `default` | `Any` | Returned when the variable does not exist. | `None` | Returns: | Type | Description | | ----- | ---------------------------------------- | | `Any` | The stored value, decrypted for secrets. | #### set ``` set(name: str, value: Any, tags: list[str] | None = None, secret: bool = False, overwrite: bool = True) -> None ``` Create or update a variable. Parameters: | Name | Type | Description | Default | | ----------- | ----------- | ------------------------------------------------------------------ | --------------------------------- | | `name` | `str` | The variable name. | *required* | | `value` | `Any` | Any JSON-serialisable value (dates and dataclasses are converted). | *required* | | `tags` | \`list[str] | None\` | Tags shown on the Variables page. | | `secret` | `bool` | Encrypt the value at rest and mask it in the API and the UI. | `False` | | `overwrite` | `bool` | When False, raise if the variable already exists. | `True` | Raises: | Type | Description | | -------------- | ------------------------------------------------ | | `ValueError` | On an invalid name or a value over 64 KB. | | `CereyanError` | When overwrite is False and the variable exists. | #### unset ``` unset(name: str) -> bool ``` Delete a variable. Parameters: | Name | Type | Description | Default | | ------ | ----- | ------------------ | ---------- | | `name` | `str` | The variable name. | *required* | Returns: | Type | Description | | ------ | ---------------------------------------------------------- | | `bool` | True when a variable was deleted, False when none existed. | ## Human input ### wait_for_input ``` wait_for_input(prompt: str, schema: dict | None = None) -> Any ``` Return the answer given to this run, pausing it when there is none yet. Inside a served run the first call transitions the run to `Paused` with the prompt and ends the attempt; the engine is free while the run waits. `POST /api/runs/{id}/resume` (or the Resume button, or the MCP `resume_run` tool) stores the answer and schedules a new attempt, which reruns the flow from the top and gets the answer from this call. Tasks marked `cache=INPUTS` are skipped on the replay. Outside a served run the answer is read from the terminal, or an error is raised when stdin is not interactive. ## Client The module-level functions use the server recorded in `server.json` of the runtime home; `Client` targets any server explicitly. ### run ``` run(flow: str, project: str | None = None, **parameters: Any) -> dict ``` Create a run of a registered flow on the live server and return it. Parameters: | Name | Type | Description | Default | | -------------- | ----- | ---------------- | ---------------------------------------------------------------- | | `flow` | `str` | Flow name. | *required* | | `project` | \`str | None\` | Project name; required when the name exists in several projects. | | `**parameters` | `Any` | Flow parameters. | `{}` | Raises: | Type | Description | | ------------------- | --------------------------------------- | | `ServerUnavailable` | When no server is running for the home. | | `CereyanError` | When the flow is unknown or ambiguous. | ### get_run ``` get_run(run_id: int) -> dict ``` `GET /api/runs/{id}` on the live server. ### list_runs ``` list_runs(**filters: Any) -> list[dict] ``` The items of `GET /api/runs` on the live server with the given filters. ### list_flows ``` list_flows(project: str | None = None) -> list[dict] ``` `GET /api/flows` on the live server, optionally filtered by `project`. ### cancel ``` cancel(run_id: int) -> dict ``` `POST /api/runs/{id}/cancel` on the live server. ### Client ``` Client(base_url: str = '', timeout: float = 30.0, token: str | None = None, socket_path: str | None = None) ``` HTTP client for a running server, on the standard library only. Parameters: | Name | Type | Description | Default | | ------------- | ------- | -------------------------------------------------- | --------------------------------------------- | | `base_url` | `str` | The server URL, for example http://127.0.0.1:4200. | `''` | | `timeout` | `float` | Seconds to wait for a response. | `30.0` | | `token` | \`str | None\` | API token; defaults to CEREYAN_TOKEN. | | `socket_path` | \`str | None\` | Connect over this Unix socket instead of TCP. | Most methods return the decoded JSON of the matching endpoint (see the HTTP API reference) and raise `ApiError` on an error status or `ServerUnavailable` when the server cannot be reached. #### health ``` health() -> bool ``` `True` when `GET /api/health` answers `ok`; never raises. #### server ``` server() -> dict ``` `GET /api/server`: version, home, engines, auth state, and listeners. #### flows ``` flows(project: str | None = None) -> list[dict] ``` `GET /api/flows`, optionally filtered by `project`. #### flow ``` flow(flow_id: int) -> dict ``` `GET /api/flows/{id}`: one flow with its schedules and summary. #### runs ``` runs(**filters: Any) -> dict ``` `GET /api/runs` with the given filters (`flow`, `project`, `state_type`, `tags`, `limit`, `cursor`, ...); returns the page. #### get_run ``` get_run(run_id: int) -> dict ``` `GET /api/runs/{id}`: one run with its state and parameters. #### task_runs ``` task_runs(run_id: int) -> list[dict] ``` `GET /api/runs/{id}/tasks`: the run's task runs. #### logs ``` logs(run_id: int, after: int = 0, level: str | None = None, search: str | None = None, limit: int = 1000) -> dict ``` `GET /api/runs/{id}/logs` after sequence `after`, optionally filtered by `level` and `search`. #### resume ``` resume(run_id: int, input: Any) -> dict ``` `POST /api/runs/{id}/resume`: answer a Paused run with `input` and schedule its next attempt. #### cancel ``` cancel(run_id: int) -> dict ``` `POST /api/runs/{id}/cancel`: cancel a queued run at once or ask a running one to stop. #### delete_run ``` delete_run(run_id: int) -> None ``` `DELETE /api/runs/{id}`: remove a finished run and its task runs, logs, and artifacts. #### counts ``` counts(project: str | None = None) -> dict ``` `GET /api/counts`: run counts by state for the dashboard, optionally per `project`. #### backfill ``` backfill(flow_id: int, parameter: str, start: str, end: str, *, interval: Any = '1d', concurrency: int = 1, extra_parameters: dict | None = None, reverse: bool = False) -> dict ``` `POST /api/flows/{id}/backfill`: create one run per step of `parameter` from `start` to `end`. Parameters: | Name | Type | Description | Default | | ------------------ | ------ | ---------------------------------------------------- | ---------------------------------- | | `flow_id` | `int` | The flow's id. | *required* | | `parameter` | `str` | The date or datetime parameter to step. | *required* | | `start` | `str` | First value, ISO formatted. | *required* | | `end` | `str` | Last value, inclusive. | *required* | | `interval` | `Any` | Step as seconds or a duration such as "1d" or "12h". | `'1d'` | | `concurrency` | `int` | How many of the backfill's runs may execute at once. | `1` | | `extra_parameters` | \`dict | None\` | Fixed values for other parameters. | | `reverse` | `bool` | Create the newest value first. | `False` | Returns: | Type | Description | | ------ | ------------------------------------------------- | | `dict` | The backfill status with its id, tag, and counts. | #### backfill_status ``` backfill_status(backfill_id: int) -> dict ``` `GET /api/backfills/{id}`: counts of the backfill's runs by state. #### cancel_backfill ``` cancel_backfill(backfill_id: int) -> dict ``` `POST /api/backfills/{id}/cancel`: cancel the backfill's remaining runs. #### schedules ``` schedules(flow_id: int) -> list[dict] ``` `GET /api/flows/{id}/schedules`: the flow's schedules with their next fire times. #### upcoming ``` upcoming(flow_id: int) -> list[dict] ``` `GET /api/flows/{id}/upcoming`: the runs materialised ahead for the flow. #### settings ``` settings() -> dict ``` `GET /api/settings`: resource totals, retention, and defaults. #### events ``` events(kind: str | None = None, after: int = 0, limit: int = 100, **filters: Any) -> list[dict] ``` `GET /api/events` newest first, filtered by name prefix `kind` and other filters; with `after` returns events following that sequence number in ascending order. #### rules ``` rules() -> list[dict] ``` `GET /api/rules`: every rule with its match clause, actions, guards, and fire counts. #### create_rule ``` create_rule(body: dict) -> dict ``` `POST /api/rules`: create a data rule from a rule body. #### variables ``` variables() -> list[dict] ``` `GET /api/variables`: every variable with secrets masked. #### artifacts ``` artifacts(run_id: int) -> list[dict] ``` `GET /api/runs/{id}/artifacts`: the run's artifacts. #### submit ``` submit(project: str, flow: str, parameters: dict | None = None, *, name: str | None = None, tags: list[str] | None = None, module: str | None = None, source_dir: str | None = None, description: str | None = None, parameter_schema: dict | None = None, options: dict | None = None, flow_tags: list[str] | None = None, flow_group: str | None = None, created_by: str = 'client') -> dict ``` `POST /api/runs`: create a run with explicit project and flow, registering the flow when needed. This is the low-level call used by the offline handoff and by `run`; pass `module` and `source_dir` so the server can import a flow it has not seen. #### run ``` run(flow: str, project: str | None = None, *, name: str | None = None, tags: list[str] | None = None, **parameters: Any) -> dict ``` Create a run of a registered flow by name. Returns the run. ### find_server ``` find_server(home: str | None = None) -> Client | None ``` A client for the live server of `home`, or `None` when there is none; raises `AuthRequired` when a token is needed. ## Custom routes ### Request ``` Request(method: str, path: str, path_params: dict[str, str], query: dict[str, list[str]], headers: dict[str, str], body: bytes) ``` The raw request handed to a handler parameter annotated `Request`. #### json ``` json() -> Any ``` The request body decoded as JSON, or `None` when empty. #### text ``` text() -> str ``` The request body decoded as UTF-8 text. ### Response ``` Response(body: bytes | str = b'', status: int = 200, headers: dict[str, str] | None = None, media_type: str | None = None) ``` An explicit response: body bytes or str, status, headers. ### HTTPError ``` HTTPError(status: int, message: str = '') ``` Bases: `CereyanError` Raise inside a handler to send a status with a JSON error body. ## Errors ### CereyanError Bases: `Exception` Base class for errors raised by cereyan. ### ParameterError ``` ParameterError(name: str, expected: str, value: object) ``` Bases: `CereyanError`, `ValueError` A parameter value could not be coerced to its declared type. ### StoreLocked Bases: `RuntimeError` The store is locked by another process. ### TransitionRejected Bases: `RuntimeError` A state transition was rejected by the rules. ### ServerUnavailable Bases: `CereyanError` No live server was found for the home directory. ### ApiError ``` ApiError(status: int, body: Any) ``` Bases: `CereyanError` The server answered with an error status; `status` and the decoded `body` are kept. ### AuthRequired Bases: `CereyanError` The server requires a token the client does not have (or a wrong one). # States and transitions Runs and task runs move through the same states, and the same rules accept or reject every transition on both the offline path and the served path. The rules live in `crates/core/src/rules.rs` (`propose`) and the names in `crates/core/src/state.rs`; this page mirrors them. ## State types | Type | Meaning | Terminal | | ------------ | ---------------------------------------------------------------------------------- | -------- | | `Scheduled` | Created and waiting for its time, a retry, or a resource | no | | `Pending` | Dispatched to an engine, not yet executing | no | | `Running` | Executing user code | no | | `Paused` | Waiting for an answer to `wait_for_input`; the engine is released | no | | `Cancelling` | Asked to stop; becomes Cancelled when the process ends or the grace period expires | no | | `Completed` | Finished without error | yes | | `Failed` | Raised, timed out, or was set failed | yes | | `Cancelled` | Stopped on request | yes | | `Crashed` | The engine died; rerun up to `crash_retries` times, then Failed | yes | A state carries a `type`, a `name` (equal to the type unless it is a sub-state), an optional `message`, a `details` object, and a `timestamp` in microseconds. A rule's `states` clause matches either: `states=["Scheduled"]` covers every scheduled run including `Late` and `AwaitingRetry`, and `states=["Late"]` narrows to that sub-state. `cereyan.states` exposes both sets as constants. ## Named sub-states | Name | Type | When | | ------------------ | --------- | -------------------------------------------------------------------------------------------------------------------------- | | `Late` | Scheduled | The scheduled time passed more than 15 seconds ago and the run has not started | | `AwaitingRetry` | Scheduled | Waiting out `retry_delay` after a failure; increments `failure_count` | | `AwaitingResource` | Scheduled | Waiting for a resource or the flow's concurrency cap | | `Retrying` | Running | A retry attempt is executing | | `TimedOut` | Failed | `timeout_seconds` elapsed | | `Cached` | Completed | A task run returned a persisted result instead of executing | | `Skipped` | Completed | A task run's target existed, a run hit `on_overlap="skip"`, a backfill value was already done, or catch-up dropped the run | A resumed run re-enters `Scheduled` with the name `Resuming` before its next attempt. ## Transition rules In order of evaluation: 1. A **forced** transition (`POST /api/runs/{id}/transition` with `force`, or a rule's `set_state` action) is always accepted and records `forced: true` in the state's details. 1. From a **terminal** state nothing is accepted. 1. The **same type and name within one second** of the current state is rejected as a duplicate. 1. Entry rules by proposed type: | Proposed | Accepted from | | ----------------------------------------------------------- | ----------------------------------------------------------------------------- | | `Pending` | no state yet, `Scheduled` | | `Running` | `Pending`, `Scheduled`, `Paused` | | `Paused` | `Running` | | `Cancelled` | `Cancelling`, `Scheduled`, `Pending`, `Running`, `Paused` | | `Scheduled`, `Completed`, `Failed`, `Crashed`, `Cancelling` | any non-terminal state except `Cancelling`, which may only become `Cancelled` | The API answers 409 with the reason (`terminal`, `duplicate`, `invalid-entry`, or `cancelling`) when a transition is rejected; the Python side raises `TransitionRejected`. ## Counters and timing Each accepted transition updates the run's counters: `failure_count` on `Failed` and on `AwaitingRetry`, `crash_count` on `Crashed`, `start_time` on the first `Running`, and `end_time` and `total_run_time` on a terminal state. ## Events Each transition that has a meaning outside the run records an event; see the [events catalogue](https://sercanatalik.github.io/cereyan/reference/events/index.md). `AwaitingRetry`, `AwaitingResource`, and `Cancelling` record none. # Examples # Agent diagnosis Start a failing run through the MCP endpoint and ask the server to explain it, the way an agent would. Source: [`examples/agent_diagnosis.py`](https://github.com/sercanatalik/cereyan/blob/main/examples/agent_diagnosis.py). Run it with `python examples/agent_diagnosis.py` while `cereyan serve examples/` is running. The server's MCP endpoint gives an agent a curated set of tools. This script does what an agent host does over Streamable HTTP: initialise a session, start a run with `run_flow`, wait for it, and call `explain_failure` to get the run, its failed task runs, the last error logs, and its events in one answer. Run it against `cereyan serve examples/`. ``` import json import time import urllib.request from cereyan import client, flow, get_run_logger, task @task def load(rows: int) -> int: if rows > 3: raise ValueError(f"too many rows: {rows} (limit 3)") return rows ``` ## A flow that fails for some inputs ``` @flow def flaky_load(rows: int = 1) -> int: get_run_logger().info("loading %d rows", rows) return load(rows) ``` ## A minimal MCP client One JSON-RPC message per POST. The `initialize` reply carries an `Mcp-Session-Id` header that later requests send back; the client name from the handshake is what runs record as `created_by`. ``` class Mcp: def __init__(self, url: str) -> None: self.url = url + "/mcp" self.session = None self.counter = 0 def rpc(self, method: str, params: dict | None = None, *, notification: bool = False): self.counter += 1 msg = {"jsonrpc": "2.0", "method": method, "params": params or {}} if not notification: msg["id"] = self.counter headers = {"content-type": "application/json"} if self.session: headers["Mcp-Session-Id"] = self.session req = urllib.request.Request(self.url, data=json.dumps(msg).encode(), headers=headers, method="POST") with urllib.request.urlopen(req, timeout=10) as resp: self.session = resp.headers.get("Mcp-Session-Id") or self.session raw = resp.read() return json.loads(raw) if raw else None def call(self, tool: str, **arguments): result = self.rpc("tools/call", {"name": tool, "arguments": arguments})["result"] text = result["content"][0]["text"] return json.loads(text) if not result.get("isError") else {"error": text} ``` ## Diagnose a failure ``` if __name__ == "__main__": mcp = Mcp(client.read_discovery()["url"]) mcp.rpc("initialize", {"protocolVersion": "2025-06-18", "clientInfo": {"name": "example-agent", "version": "1"}, "capabilities": {}}) mcp.rpc("notifications/initialized", notification=True) run = mcp.call("run_flow", flow="flaky_load", parameters={"rows": 5})["run"] assert run["created_by"] == "mcp:example-agent" deadline = time.time() + 30 while time.time() < deadline and client.get_run(run["id"])["state"]["type"] not in ("Completed", "Failed"): time.sleep(0.1) explained = mcp.call("explain_failure", run_id=run["id"]) print("verdict:", explained["verdict"]) assert "too many rows" in explained["verdict"] assert any(e["name"] == "run.failed" for e in explained["events"]) ``` # Alerting A custom event, a code rule that reacts to it, and a proactive rule for a run that overruns. Source: [`examples/alerting.py`](https://github.com/sercanatalik/cereyan/blob/main/examples/alerting.py). Run it with `python examples/alerting.py` while `cereyan serve examples/` is running. Rules turn events into actions. This module declares a flow that emits a custom event when a table looks empty, a code rule that reacts to that event, and a proactive rule that fires when a run of the flow does not complete within ten minutes of starting. Rules fire in the server, so run this against `cereyan serve examples/`. ``` import time from cereyan import App, client, emit_event, get_run_logger app = App("shop") ``` ## A flow that emits an event ``` @app.flow def check_orders(count: int = 0) -> int: if count == 0: emit_event("orders.table_empty", {"table": "orders"}) get_run_logger().info("orders: %d", count) return count ``` ## A reactive code rule `on` accepts a prefix; the function receives the event and the run as dicts and runs as the rule's `call` action. ``` @app.rule(on="orders.*", flow="check_orders", once="per_run") def alert_empty(event, run): print(f"ALERT {event['payload']['table']} is empty (run {run['name']})") return {"alerted": True} ``` ## A proactive rule Armed when a `check_orders` run starts, disarmed by its completion, fired by the server if two minutes pass first. ``` @app.rule(on="run.running", flow="check_orders", unless="run.completed", within=120) def check_orders_overran(event, run): print(f"ALERT {run['name']} has been running for two minutes") ``` ## Run it and read the firing ``` if __name__ == "__main__": run = client.run("check_orders", count=0) deadline = time.time() + 30 while time.time() < deadline and client.get_run(run["id"])["state"]["type"] not in ("Completed", "Failed"): time.sleep(0.1) api = client.default_client() events = api.events(kind="orders.*", run_id=run["id"]) assert events and events[0]["name"] == "orders.table_empty" # Rules fire after the run that emitted the event has finished, not with it, # so wait for the firing rather than reading the count straight away. deadline = time.time() + 30 while time.time() < deadline: fired = [r for r in api.rules() if r["name"] == "alert_empty"] if fired and fired[0]["fire_count"] >= 1: break time.sleep(0.1) assert fired and fired[0]["fire_count"] >= 1, fired print("rule fired", fired[0]["fire_count"], "time(s)") ``` # Approval A flow that pauses for a human decision and resumes with the answer. Source: [`examples/approval.py`](https://github.com/sercanatalik/cereyan/blob/main/examples/approval.py). Run it with `python examples/approval.py` while `cereyan serve examples/` is running. `wait_for_input` parks the run in `Paused` with a question. The engine is released while the run waits; answering through the UI, the API, the client, or the MCP `resume_run` tool schedules a new attempt that replays the flow from the top, where tasks marked `cache=INPUTS` return their stored results and `wait_for_input` returns the answer. This example runs against a server: start `cereyan serve examples/` first. ``` import time from datetime import date from cereyan import INPUTS, flow, get_run_logger, task, wait_for_input from cereyan import client @task(cache=INPUTS, persist_result=True) def prepare(day: date) -> int: get_run_logger().info("preparing %s", day) return 1200 @task def release(rows: int) -> str: return f"released {rows} rows" ``` ## The flow The question carries a JSON schema, which the run page turns into a form. ``` @flow def publish(day: date) -> str: rows = prepare(day) decision = wait_for_input( f"Release {rows} rows for {day}?", schema={"type": "object", "properties": {"approve": {"type": "boolean"}}, "required": ["approve"]}, ) if not decision["approve"]: return "held" return release(rows) ``` ## Driving it from a script Start the run, wait until it pauses, read the question, answer it, and wait for the result. ``` def wait_for(run_id: int, predicate, timeout: float = 30.0) -> dict: deadline = time.time() + timeout while time.time() < deadline: run = client.get_run(run_id) if predicate(run): return run time.sleep(0.1) raise TimeoutError(f"run {run_id} did not reach the expected state") if __name__ == "__main__": run = client.run("publish", day="2026-03-01") paused = wait_for(run["id"], lambda r: r["state"]["type"] == "Paused") print("question:", paused["state"]["details"]["prompt"]) client.default_client().resume(run["id"], {"approve": True}) done = wait_for(run["id"], lambda r: r["state"]["type"] in ("Completed", "Failed", "Crashed", "Cancelled")) assert done["state"]["type"] == "Completed", done["state"] print("state:", done["state"]["type"]) ``` # Daily ETL A scheduled daily flow whose reruns and backfills are idempotent through targets. Source: [`examples/daily_etl.py`](https://github.com/sercanatalik/cereyan/blob/main/examples/daily_etl.py). Run it with `python examples/daily_etl.py` (no server needed). A daily flow: extract a day's data, write it to a file, and never do the same day twice. The target on `build` makes reruns skip finished days, `bulk_complete` lets a backfill skip them before runs are even created, and the schedule fires the flow every morning once a server is running. ``` from datetime import date, timedelta from cereyan import Cron, LocalTarget, exponential, flow, get_run_logger, task def output_for(day: date) -> LocalTarget: return LocalTarget(f"out/daily/{day}.csv") ``` ## Skipping finished days `bulk_complete` receives every value a backfill is about to create and returns the ones already done; those runs are recorded as Skipped without dispatching. ``` def already_built(values: list[date]) -> set[date]: return {d for d in values if output_for(d).exists()} ``` ## The task `output=` names the target; when it exists the task run ends Skipped and the body does not execute. Retries with an exponential delay cover transient failures. ``` @task(output=output_for, retries=2, retry_delay=exponential(base=0.5, maximum=30)) def build(day: date) -> str: target = output_for(day) get_run_logger().info("building %s", target.path) with target.open("w") as fh: fh.write("id,amount\n") for i in range(3): fh.write(f"{i},{(i + 1) * 10}\n") return target.path ``` ## The flow Fires at 06:30 Istanbul time every day when served, never overlaps itself, and defaults to yesterday so a manual run does the most recent complete day. ``` @flow( schedule=Cron("30 6 * * *", timezone="Europe/Istanbul"), max_concurrent=1, on_overlap="skip", bulk_complete=already_built, run_name="daily-{day}", ) def daily_etl(day: date = date.today() - timedelta(days=1)) -> str: build(day) return output_for(day).path ``` ## Run it twice The second call for the same day skips `build` because its file exists; a skipped task returns `None`, so the flow returns the path itself. ``` if __name__ == "__main__": day = date(2026, 1, 15) first = daily_etl(day) second = daily_etl(day) assert first == second assert output_for(day).exists() print("built", first) ``` # Fan-in A report that runs once per day after both of its upstream flows finished that day. Source: [`examples/fan_in.py`](https://github.com/sercanatalik/cereyan/blob/main/examples/fan_in.py). Run it with `python examples/fan_in.py` (no server needed). Two independent loads and a report that needs both. With `after=[...]` and `batch_key="day"`, the server creates one `report` run per day as soon as the last of `sales` and `inventory` has a completed run for that day. Offline, the three flows are ordinary functions you call in order. ``` from datetime import date from cereyan import App, get_run_logger, task app = App("reporting") @task def load(source: str, day: date) -> int: get_run_logger().info("loading %s for %s", source, day) return 100 ``` ## Upstream flows ``` @app.flow def sales(day: date) -> int: return load("sales", day) @app.flow def inventory(day: date) -> int: return load("inventory", day) ``` ## The downstream flow `day` is both the flow's parameter and the batch key. A day that already has a report never gets a second one, and a failed upstream blocks the day until it is rerun successfully. ``` @app.flow(after=["sales", "inventory"], batch_key="day", run_name="report-{day}") def report(day: date) -> str: return f"report for {day}" ``` ## Offline Without a server nothing is triggered automatically; call the flows yourself. ``` if __name__ == "__main__": day = date(2026, 2, 1) sales(day) inventory(day) print(report(day)) ``` # Quickstart pipeline Three tasks and one flow, run offline as a plain script. Source: [`examples/pipeline.py`](https://github.com/sercanatalik/cereyan/blob/main/examples/pipeline.py). Run it with `python examples/pipeline.py` (no server needed). The smallest useful pipeline. Run it with `python examples/pipeline.py` and a run is recorded in the runtime home; run it while `cereyan serve examples/` is up and the run is handed to the server instead. Parameters come from the type hints, so `cereyan run examples/pipeline.py:etl --param day=2026-01-02` coerces the string to a `date` before the flow starts. ``` from datetime import date from cereyan import flow, get_run_logger, task ``` ## Tasks A task is a function whose calls inside a flow are recorded as task runs. The run logger writes lines that are stored with the run and shown live in the UI. ``` @task def extract(day: date) -> list[int]: get_run_logger().info("extracting %s", day) return [1, 2, 3] @task def transform(rows: list[int]) -> list[int]: return [r * 2 for r in rows] @task def load(rows: list[int]) -> int: get_run_logger().info("loading %d rows", len(rows)) return sum(rows) ``` ## The flow The flow calls the tasks in order; `run_name` names each run after its parameter. ``` @flow(run_name="etl-{day}", tags=["example"]) def etl(day: date = date(2026, 9, 6)) -> int: return load(transform(extract(day))) ``` ## Run it A flow is a function: calling it runs the tasks and returns the result. ``` if __name__ == "__main__": total = etl(date(2026, 9, 6)) print("total:", total) assert total == 12 ``` # Webhook route A custom HTTP route that receives an order and starts a flow for it. Source: [`examples/webhook_route.py`](https://github.com/sercanatalik/cereyan/blob/main/examples/webhook_route.py). Run it with `python examples/webhook_route.py` while `cereyan serve examples/` is running. Custom routes are plain functions registered on an App and served next to the API. Path and query parameters bind by name and are coerced through the type hints; a dataclass parameter receives the JSON body. This route starts a run for each order it receives. Run it against `cereyan serve examples/`. ``` import json import time import urllib.request from dataclasses import dataclass from cereyan import App, client, get_run_logger app = App("intake") @dataclass class Order: order_id: int amount: float ``` ## The flow the route starts ``` @app.flow def ingest_order(order_id: int, amount: float) -> str: get_run_logger().info("order %d for %.2f", order_id, amount) return f"order {order_id} ingested" ``` ## The routes Return a dict for JSON, a `(value, status)` tuple to set the status, a str for text, or a `Response` for full control. Raise `HTTPError` for an error status. ``` @app.get("/api/ext/ping") def ping() -> dict: return {"ok": True} @app.post("/api/ext/orders") def receive_order(order: Order) -> tuple[dict, int]: run = client.run("ingest_order", order_id=order.order_id, amount=order.amount) return {"run_id": run["id"]}, 201 ``` ## Call it ``` if __name__ == "__main__": url = client.read_discovery()["url"] body = json.dumps({"order_id": 7, "amount": 99.5}).encode() req = urllib.request.Request(url + "/api/ext/orders", data=body, method="POST", headers={"content-type": "application/json"}) with urllib.request.urlopen(req) as resp: assert resp.status == 201 run_id = json.loads(resp.read())["run_id"] deadline = time.time() + 30 while time.time() < deadline and client.get_run(run_id)["state"]["type"] not in ("Completed", "Failed"): time.sleep(0.1) print("run", run_id, client.get_run(run_id)["state"]["type"]) ``` # Design # Architecture ``` ┌──────────────────────────────── cereyan serve dir/ ────────────────────────────────┐ │ Python process │ │ ├── imports every module under dir/: flows, routes, rules │ │ └── cereyan._core (pyo3) ──────────────────────────────────────────┐ │ │ Rust, tokio runtime │ │ │ ┌─────────────┐ ┌───────────┐ ┌────────────┐ ┌───────────┐ │ │ │ │ axum HTTP │ │ scheduler │ │ rules │ │ supervisor│ │ │ │ │ /api, /mcp, │ │ timer heap│ │ match + │ │ engine │ │ │ │ │ UI, SSE, │ │ look-ahead│ │ templates │ │ pool │ │ │ │ │ custom │ │ catch-up │ │ expectations│ │ heartbeats│ │ │ │ │ routes ─────┼──┼──▶ Python │ └────────────┘ └─────┬─────┘ │ │ │ └──────┬──────┘ └─────┬─────┘ │ │ │ │ │ └───────────────┴──────────────┴───────────────┘ │ │ │ │ in-memory working set: │ │ │ │ active runs, counters, │ │ │ │ schedule heap, resources, │ │ │ │ rule index │ │ │ ▼ │ │ │ ┌─────────────────┐ │ │ │ │ store (SQLite) │ one writer thread, │ │ │ │ WAL, group │ read pool, │ │ │ │ commit │ migrations │ │ │ └─────────────────┘ │ │ └─────────────────────────────────────────┬───────────────────────────────────────────┘ │ loopback HTTP: work, report, heartbeat ┌────────────────┴───────────────┐ │ engine processes (Python) │ one module each, warm, │ import module once, run flows │ recycled on change or │ buffer transitions + logs in │ after N runs │ embedded core, flush 100 ms │ └────────────────────────────────┘ ``` ## Crates | Crate | Role | | ---------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `cereyan-core` | The model (flows, runs, task runs, states, schedules, events, rules), the transition rules, id and time types. No I/O, no async, no dependency on SQLite or Python, so both execution paths share it. | | `cereyan-store` | SQLite: opening and quarantine, embedded migrations, a single writer thread with group commit, a read pool, secrets encryption, the home directory and its lock. | | `cereyan-rules` | Matching a rule's `when` and `unless` clauses against events and rendering action templates with minijinja. | | `cereyan-server` | axum on tokio: the API with an OpenAPI document from utoipa, the SSE stream, the embedded UI, custom-route dispatch into Python, the scheduler, the rules engine and expectations, the engine supervisor, MCP, auth, retention. | | `cereyan-py` | The pyo3 module `cereyan._core`: the store for the offline path, the server entry point, and the engine's reporting client. | The Python package is thin: decorators and parameter coercion, targets and results, the CLI, the client, the engine child that executes runs, the route dispatcher, and the MCP stdio proxy. ## The offline path `python pipeline.py` opens the store through `_core`, takes the advisory lock, and executes the flow in-process. Transitions go through the same `propose` rules as on the server, and the process appends runs, task runs, logs, events, and artifacts directly. If the lock is held by a server, the script submits the run to it instead and streams the logs back. ## The served path 1. The server imports every module under the directory, registers flows (upsert by project and name), routes, and code rules, emits `flow.registered`, and reconciles non-terminal runs from the previous life: engines still alive are adopted, the rest are marked crashed. 1. The scheduler materialises upcoming runs and wakes on a timer heap; rules and expectations use the same heap. 1. A due run is dispatched to an engine keyed by its module and source directory, spawning one when the pool has room. The engine proposes Pending and Running through the API, executes the flow, and buffers task-run transitions and logs in its embedded core, flushing every 100 milliseconds or on flow-level transitions, idempotently by sequence. 1. Every accepted transition updates the in-memory index (active runs, counts, resources), records an event, is pushed to the SSE stream, and is checked against the rule index. 1. Terminal states release resources and engines; crashes are detected by missed heartbeats and a dead PID. ## Identity and time Rows have integer ids for joins and UUIDv7 external ids for the API. Timestamps are microseconds since the Unix epoch in UTC; schedules evaluate in their IANA timezone. ## The UI React with Tanstack Router, Query, and Table and shadcn components, built with Vite and embedded in the wheel. It talks to the API through a client generated from the OpenAPI snapshot and updates from the SSE stream; the timeline graph is inline SVG. Related: [Engines and the home directory](https://sercanatalik.github.io/cereyan/concepts/engines-and-home/index.md), [Performance targets](https://sercanatalik.github.io/cereyan/design/performance/index.md). # Design and limitations Cereyan is built for one machine, one wheel, and pipelines that can always be rerun. Those three choices explain most of what it does and everything it refuses to do. This page says what those choices mean for you. ## What it is for - A data engineer with pipelines on a laptop, a workstation, or one server, who wants schedules, retries, backfills, a UI, and alerts without running a platform. - Batch work measured in seconds to hours, tens of thousands of runs a day at most, where files and tables written by the flow are the real output. - Teams that keep flows in one repository and prefer code over configuration. ## Decisions **One wheel, no runtime dependencies.** `pip install cereyan` is the whole install: the Rust core, the server, the UI, and the MCP server are inside the wheel. Nothing is pulled in at run time, so a pipeline environment gains no transitive dependencies from its orchestrator. Docs tooling and tests use dependency groups that never reach the wheel. **One process serves everything.** `cereyan serve` is the API, the UI, the scheduler, the rules engine, the MCP endpoint, and the engine supervisor. There are no separate agents, workers, or queues to run. **Two execution paths, one store.** A plain `python pipeline.py` writes the same SQLite store the server uses, under the same state rules, so a script and a served run look identical in history. When a server holds the store, scripts hand their runs to it. **History is a cache.** The database records what happened; your code and your targets are the source of truth. Losing `db.sqlite` loses the run list, not your data, and a corrupted file is quarantined rather than blocking startup. Crashed runs are rerun, up to a limit, on the assumption that a rerun is safe because targets make it so. **A fixed vocabulary.** App, flow, task, run, task run, state, schedule, parameter, target, resource, backfill, artifact, variable, event, rule. There is no "deployment": a served flow with a schedule is the deployed thing. **Performance is a feature.** The store is SQLite in WAL mode with one writer, group commit, a read pool, and an in-memory working set; engines report in batches serialised in Rust. The [targets](https://sercanatalik.github.io/cereyan/design/performance/index.md) are enforced by benchmarks in CI. ## What it does not do, and why | Not provided | Reason | Instead | | ---------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------- | | Remote workers, work pools, Kubernetes or Docker execution | Cereyan runs on the machine it is installed on; distributing execution would need a broker, a scheduler that knows about hosts, and a rollout story, which is the platform it set out not to be. | Run a server per machine, or call remote systems from inside tasks. | | Postgres or any other database | SQLite gives one file, no service, and the performance the targets need; a second backend would double the store and change the operational shape. | Retention keeps the file bounded; back it up like any file. | | Remote or object-store targets (S3, GCS, HDFS) | A `Target` is anything with `exists()`, so you can write one in a few lines with the client library you already use; shipping them would add dependencies to the wheel. | Write a small class with `exists()`. | | Integrations and connector packages | Tasks are plain Python; the library for your warehouse or API works unchanged inside one. | Import it in the task. | | Multi-user accounts, roles, SSO, audit trails | One token, one permission level, one machine. | Put a reverse proxy in front for network access; use the token or the socket. | | Cloud-style extras: assets, SLAs, incident management, metric triggers, incoming webhooks as event sources | Each is a product on its own; rules with `unless` and custom routes cover the local versions of the common cases. | A proactive rule for "did not happen by"; a custom route that calls `emit_event` for incoming webhooks. | | TLS | The server is loopback-only by default and a reverse proxy does TLS better. | nginx, Caddy, or an SSH tunnel. | | Versioned flows, code storage, image builds | The code in the served directory is the version; engines reload it when it changes. | Git. | | Real-time or streaming pipelines | Runs are units of work with a beginning and an end; the scheduler is a timer heap, not an event loop over streams. | A long-running flow per stream partition, or another tool. | | Preemption, priority-based killing | A running run is never interrupted to make room; priority only orders the queue. | Resources and caps to keep heavy work from starting. | ## Limits worth knowing - One server per machine, because the home and its advisory lock are global. - The engine pool bounds concurrent runs at `max_engines`, default the CPU count; a run occupies an engine for its whole duration, including time spent waiting on I/O. - Artifacts are limited to 1 MB and variables to 64 KB. - Retention deletes logs and events older than `retain_days`; runs and task runs stay. - Unix sockets and engine niceness do not exist on Windows. Two behaviours are also weaker there: a flow's `timeout_seconds` does nothing on the offline path, and cancelling a flow blocked in a call ends the engine instead of raising inside the flow. - Schedules, dependencies, backfills, data rules, clock-armed proactive rules, and pausing need a running server; offline scripts record runs and fire code rules that a server has registered. Related: [Architecture](https://sercanatalik.github.io/cereyan/design/architecture/index.md), [Migrate from Prefect or Luigi](https://sercanatalik.github.io/cereyan/guides/migrate/index.md). # Performance targets These targets shaped the design (a Rust core, batched reporting, an in-memory working set) and are enforced by the benchmarks in CI. They are measured on a developer laptop; a small server does better. | Operation | Target | | ------------------------------------------ | --------------- | | Server start with 1M historical runs | under 200 ms | | Task transitions ingested | 20k per second | | Log lines ingested | 100k per second | | Runs list query at 1M runs | under 10 ms | | Schedule wake-up drift | under 50 ms | | Backfill create of 10k runs | under 1 s | | Warm-pool overhead, Scheduled to user code | under 5 ms | | Counts endpoint | under 5 ms | | Orchestration cost per completed task run | under 200 µs | ## How they are checked `just bench` runs two suites: - **Criterion benchmarks** in the Rust crates: the store's hot path (a transition on either table, and applying an engine report at three batch widths), bulk run creation and log append, the transition rules, matching, and the stream payload built for every transition. - **The end-to-end benchmark** `benches/e2e.py`, which starts a server on a temporary home, generates its own fixtures (a million-run history, wide and log-heavy flows, a long backfill, an interval schedule), and measures each operation through the public API. The end-to-end benchmark fails when a target is missed or when a number regresses more than 20 percent against the checked-in baseline for the platform, `benches/baseline..json`. CI runs it on every push with `--quick`. When a regression is intentional, refresh the baseline in the same change with `just bench-baseline`. That rewrites every key, so read the table it prints before committing the result: a number that moved for a reason you have not established is not one to bless. The two suites do different jobs. Only the end-to-end benchmark gates: it is the one with a checked-in baseline. Criterion keeps its baselines in `target/criterion/`, which is not committed, so it compares a run against the previous run on the same machine — what you want while you are changing the store, and no help at all in CI. A store regression small enough to hide inside end-to-end noise will not fail a build. ## What makes the numbers - SQLite in WAL mode with `synchronous=NORMAL`, one writer thread with group commit, and a read pool, so reads never wait for writes. - Integer rowids for joins and keyset pagination for every list endpoint, so the runs list costs the same at a million rows as at ten. - An in-memory index of active runs, counters, the schedule heap, resources, and the rule index, so dispatch, counts, and matching do not query the database. - Engines buffer transitions and logs in their embedded Rust core and report in batches serialised there; reports are accepted up to 64 MB and appended in one write per report. - Opening a large database after a clean shutdown does not read the whole file. Related: [Architecture](https://sercanatalik.github.io/cereyan/design/architecture/index.md). # Project # Changelog Versions follow the wheel on PyPI. Each entry lists what changed and, where a downgrade needs care, how to do it. # Changelog ## Unreleased ## 1.13.0 (2026-09-13) - **A flow disabled by `disable_after` now resumes after a restart.** The resume at the end of the window was a timer held only in the server's memory, and start never read the `paused_until` it had stored beside it, so a server restarted inside the window left the flow's schedules paused until someone resumed them by hand. Start now derives the window from `paused_until`: one still open is armed again and ends on time, and one that ended while the server was down resumes as start completes, recording `flow.enabled` either way. Fires inside the window are still never caught up; fires after it, missed only because the server was down, follow the schedule's `catchup` policy. - **Skip upcoming runs, and reschedule from the Flows page.** Pausing was the only way to stop a scheduled run, and it dropped every upcoming run until someone remembered to resume. A fire can now be skipped on its own: from the Flows row menu (**Skip next run**, **Skip runs…**), the flow page's **Skip next…** and **Upcoming** tab, or `POST /api/schedules/{id}/skips` with `fires` or `next`. A skip is stored against the schedule and the fire time, so it survives a restart, a pause and resume, and an edit that keeps the time; catch-up never recreates a skipped fire, and the look-ahead keeps three runs that will start past the skipped ones. At its time the run ends `Skipped` with `reason` `user` without starting, and the runs of flows declared `after=` it, keyed fan-in included, are created `Skipped` with `reason` `upstream`; every other Skipped run still triggers its downstream as before. **Undo**, or `DELETE /api/schedules/{id}/skips/{fire}`, takes a skip back until its time. `GET /api/flows/{id}/upcoming` marks each run `skipped`, with `skipped_by` and `skipped_at`, and, with `projected=N`, also lists fires past the look-ahead; each schedule reports `skipped`, and its `next_fire` is the next fire that will run. **Reschedule…** edits a cron schedule as a daily, weekly or monthly time in a timezone, or as raw cron, with its catch-up options and the coming week before and after the change; skips an edit no longer produces are dropped and recorded as `schedule.skips_dropped`. `run.skipped` now carries `reason`. - **Breaking: a schedule edit lasts until the next restart unless it asks to persist.** `PATCH /api/schedules/{id}` used to mark the row `persist`, which detached a code-declared schedule from its declaration for good and said so only afterwards, in small grey text. It now leaves `persist` as it was unless the body sets it, so an edit from the flow page, the Reschedule dialog, or MCP's `edit_schedule` holds until the server restarts and the declaration applies again. Send `persist: true` for the old behaviour. - CI runs the documentation tests on Windows, so every suite now runs on Linux, macOS and Windows. The Python suite has run there since 1.5.0, once two things were fixed: a test liveness probe that killed the process it was asking about, and a stop path that hard-killed the server before it could shut down (the 1.5.0 graceful-stop entry). The documentation step was the last one conditional on the platform. No product behaviour changed. ## 1.12.0 (2026-09-10) - **A rule that could never fire is now refused instead of stored.** `@app.rule(on="run.failure")` registered cleanly, stored cleanly, and never fired — no error, no warning, no log line — because nothing on either path checked an event name. The names themselves lived in four places that drifted apart: literals across the server crate, a smaller set on the offline path, a hand-written reference table that claimed to mirror them, and a free-text box in the rule form. There is now one catalogue, in `crates/core/src/events.rs`, that the emit sites are compiled against, so renaming an event breaks the build. The engine owns the prefixes `run.`, `task_run.`, `flow.`, `schedule.`, `resource.`, `rule.` and `expectation.`: a name under one of them that nothing emits is rejected at import by `@app.rule`, with 422 by the rules API, and by `emit_event`, each with the nearest real name — `run.failure` answers "did you mean `run.failed`?". Every name outside those prefixes is a custom event and is not checked, so `on="orders.table_empty"` still needs no registration. `cereyan.events` and `cereyan.states` expose the catalogue as `str` constants that are the wire names, so `events.run.failed` and `"run.failed"` are interchangeable and nothing downstream can tell them apart; `events.run.any` is `"run.*"`. `GET /api/vocabulary` serves the same list, the rule form offers it as suggestions while still taking a typed custom name, and `docs/reference/events.md` is generated from it. - **A rule naming a state type now matches that type's sub-states.** `states=` was documented as "state types the run must be in" and matched the sub-state name, so `states=["Scheduled"]` matched no run that was `Late`, `AwaitingRetry` or `AwaitingResource`, and expressing "any scheduled run" meant listing all four. A value now matches the run's state type or its sub-state name: `["Scheduled"]` covers the sub-states, `["Late"]` still narrows to one. This widens what an existing rule matches — it can only match more, never less, since no type and sub-state name collide — so a rule written to exploit the old narrow reading fires more often. The guards (`once`, `cooldown_seconds`, `max_per_minute`) bound the effect. - `@app.rule` rejects an unrecognised keyword instead of dropping it, so `cooldownseconds=60` no longer leaves a rule running on defaults nobody asked for. `once="never"`, valid since rules shipped and documented nowhere, is on Events and rules and the rules guide. The Rules page marks a rule that has never fired, which is the one failure that name checking cannot catch: a rule spelled plausibly that matches nothing. `CodeRule.extra_actions`, which nothing set and no decorator accepted, is gone. - The schedule drift test no longer gates a release on a wall-clock ceiling. `test_schedule_fires_on_time_with_low_drift` asserted that a run starts within 250 ms of its scheduled time; a Windows runner measured 294 ms and failed the 1.11.0 build, which is the flake the repository already has a rule against — the correctness gate excludes wall-clock ceilings, because a bound calibrated on a developer machine says nothing about correctness on a shared runner. The 250 ms was also a loosened stand-in for a target that already has a home: `benches/e2e.py` measures `schedule_drift_ms` against the documented 50 ms, with a per-platform baseline and regression detection. The test is now `test_schedule_fires_at_its_scheduled_time` and asserts what does not depend on the hardware — the run is materialised at the anchor, it completes, and it does not fire early. No product behaviour changed. ## 1.11.0 (2026-09-09) - Documentation: `@flow(group=...)` shipped in 1.10.0 with a paragraph on App and projects and nothing a reader could copy. It now has a section of its own there with a snippet, and Flows and parameters lists it among what a flow declares, so the option is findable from the page a reader reaches for when choosing one. The README's screenshot captions describe the collapsible groups the recaptured Flows and Runs shots already show. ## 1.10.0 (2026-09-09) - **Flows can declare a group, and the UI collapses by it.** `@flow(group="nightly")` names the group a flow is listed under; without it a flow is grouped under its project, as before. Groups are a flat axis rather than a level inside the project, so flows in different projects declaring the same group form one group and a group named after a project merges with the flows defaulting to it — which is how a flow joins a group it does not live beside. A run carries its flow's group, read through the flow rather than stored on the run, so renaming a group moves the run history with it instead of cleaving it in two. Both the Flows and the Runs page now draw their groups as collapsible sections whose header is the same columns rolled up: the soonest next fire, the group's recent runs, its states as a bar with counts, the tag union, and the row count. Flows the running server no longer has registered are counted separately as stale, because a group whose flows all last succeeded but have since deregistered would otherwise read as healthy behind a folded header. A group of more than five rows starts collapsed, a lone group is always open, a search opens every group it matches, and the default is fixed when a group is first seen so a live update never shuts one under the cursor. - An engine still reporting a run it has just finished keeps retrying for up to ten minutes when its server disappears, so a restarted server records the outcome instead of seeing a crash. That is deliberate, and it is now on Engines and the home directory, where the thirty to forty second figure for an idle engine had been the only number given. Three CI failures came from tests that did not know it: `wait_run` returns when the server records the terminal state, while the engine is still flushing its last events and log lines. Two shutdown tests killed the server inside that window and measured the ten minute path; the retention test rewrote the timestamps of log rows the engine was still adding to, so a line landing afterwards carried a fresh one and was rightly never swept. `ServerProcess.wait_idle` gives all three the wait they were missing. ## 1.9.2 (2026-09-09) - **A clock-armed proactive rule no longer cries wolf on its first tick.** `unless` with `at` fires when an expected event did not arrive in the look-back window, and an early tick's window reaches back past the rule's own creation — where the store is empty because nothing had happened yet, not because anything was missed. A rule with `within=3` fired 0.77 seconds after the server started, and one saved from the UI could page someone about the hour before it existed. A tick now evaluates only once the rule has been watching for a whole window; the schedule keeps running meanwhile, so the first honest lapse is delayed by at most `within`. This was recorded in `WINDOWS.md` as an unexplained Windows-only flake; it is neither. Windows lost the coin flip more often because a slower start delays the first event, and CI caught it on Linux at 1.9.1. The `xfail` marker that blamed Windows is gone, and `test_clock_armed_rule_waits_until_it_has_watched_a_whole_window` pins the behaviour. ## 1.9.1 (2026-09-09) - The engine give-up test no longer gates a release on a wall-clock ceiling. `test_idle_engine_gives_up_when_the_server_is_killed` asserted that a SIGKILLed server's engine exits within 60 seconds, against a bound of 40; on a loaded macOS runner the same commit that passed on `main` failed on the tag, which is the flake the repository already has a rule against. It now asserts only that the engine gives up, and the 40 second bound moved to a `performance` test that `just bench` runs on calibrated hardware, where a number like that means something. No product behaviour changed. ## 1.9.0 (2026-09-09) - Documentation: features that shipped without a page. The MCP guide still described fifteen tools and named none of the four schedule tools added in 1.8.0. `cereyan.runtime`, the public view of the run in progress (`run`, `task_run`, `flow`), was exported from the package and mentioned nowhere; it now has a section on the Python API page and a pointer from Runs and states. The events reference listed five of the fifteen kinds the SSE stream carries and the OpenAPI description behind `GET /api/stream` listed four; both now list all of them with their payloads. The tour still said the run page's Details tab shows an exit code, which was removed in 1.4.0, and did not mention the task run page. - Documentation: the three Windows behaviours tracked in `WINDOWS.md` are now stated where a reader meets them rather than only in a file that ships with the repository. A flow's `timeout_seconds` does nothing on the offline path there, so Retry, time out and survive crashes says so; a cancel cannot interrupt a flow blocked in a call, so Engines and the home directory says so; a clock-armed proactive rule can lapse while its events are still arriving, so Detect when something did not happen says so. Install carries the three as a table and Design and limitations names them. ## 1.8.0 (2026-09-09) - **An agent can manage schedules.** `pause_schedule` and `resume_schedule` both took a `schedule_id` that no MCP tool returned, so reaching them meant already knowing the id: a schedule that had never fired was invisible while the tool to pause it sat in the list. `list_schedules` closes that, filtered by flow or project, reporting each schedule's spec, whether it is active, when it next fires, and whether it came from code, the interface or an agent. `create_schedule`, `edit_schedule` and `delete_schedule` complete the lifecycle; create and edit return the next few fire times so a spec can be checked before it fires unattended, and schedules an agent creates record `source: "mcp"`. Two guards: editing a schedule declared in a flow's code returns a note that the declaration no longer governs it, and deleting one is refused, because startup recreates it from the declaration and a success would be untrue. Flows still run on demand by default; a flow needs no schedule. ## 1.7.0 (2026-09-09) - **Cereyan is production ready.** The Python API, the HTTP API, the CLI, the MCP surface and the database schema are settled; the under-development warning is gone from the README and the documentation, and the PyPI classifier moves from Beta to Production/Stable. The README also said the package was not yet published, which stopped being true at 1.4.0. - Benchmarks for the write path a served task run actually travels. `transition` on both tables and `apply_report` over the event sequence an engine child sends had no benchmark at all, so a change to the store showed up only once it was large enough to move an end-to-end number that also contains subprocess spawn, HTTP and the client's flush interval. `cargo bench -p cereyan-store --bench hot_path` now measures both directly, at three batch widths so per-batch cost that grows faster than the batch is visible as a shape. A second bench sizes the stream payload built for every transition. `just bench` gains `task_run_cost_us`, the orchestration cost of one completed task run, with a 200 µs target — the unit a fan-out workload is actually counted in. - The performance baseline is refreshed. It was last written at 1.3.0 and three of its numbers had drifted, `schedule_drift_ms` past the 20 percent gate. No code explains it: every file on the ingestion and dispatch paths — `writer.rs`, `read.rs`, `dispatch.rs`, `api/engine.rs`, `index.rs`, `scheduler.rs`, `timer.rs`, and the core state rules — is byte-identical to that commit, and `Cargo.lock` differs only in the version string. The repository pins no Rust toolchain, so a baseline does not survive a compiler upgrade; the numbers here were recorded on rustc 1.92.0. Every target is still met with room to spare. ## 1.6.1 (2026-09-09) - Internal refactoring. ## 1.6.0 (2026-09-09) - **`@flow` and `@task` reject `async def`.** An async body was never awaited: the call returned a coroutine, the body never ran, and the run was recorded `Completed` — a pipeline that fetched nothing and reported success, its only trace a `RuntimeWarning` on stderr after the fact. Both decorators now raise `TypeError` at decoration, naming the function and showing the synchronous wrapper. `async def` with `yield` is rejected too: `inspect.iscoroutinefunction` is false for an async generator function, which failed the same way. Async bodies remain unsupported; `async def` route handlers are unaffected. Anything this breaks was already reporting success without running. - A guide for fetching from an HTTP API: building the client once so a warm engine reuses its connection pool, `map` over a `ThreadRunner` for concurrency, and where the reuse stops (`isolated=True`, `ProcessRunner`, offline runs). That an engine imports its module once, and so shares module-level state across the runs it serves, is now stated on Engines and the home directory and covered by a test, rather than being an undocumented accident of the implementation. ## 1.5.0 (2026-09-09) - The runtime home is readable only by the account that created it. It was created with whatever the umask gave — `0755` on a typical machine — so `db.sqlite`, with every run, log, event and variable, was readable by any other account, and `secret.key` was too: it was narrowed to `0600` after being written, leaving a window in which the key that decrypts every secret was world-readable, and that narrowing never ran on Windows at all. Protecting the directory covers everything in it, closes the window, and needs nothing platform-specific for a home under your user profile. A home from an earlier version is narrowed when opened, with a message saying so. - An engine whose server disappeared could take up to ninety seconds to exit rather than the thirty its documentation promised. It notices only between requests, so the wait was bounded by the client's long-poll timeout — ninety seconds — and not by the thirty second idle threshold that appeared to govern it. The timeout is now forty seconds, ten more than the server ever holds a request, so the worst case is forty rather than ninety. The regression test used to wait exactly ninety seconds and so raced the very timeout that defeated the bound; it now waits sixty against a forty second bound. - Windows: stopping the server no longer kills the engine executing a run. Engine children were spawned in their own process group on Unix — so a Ctrl-C or a stop aimed at the server never reached them — but there was no Windows equivalent, so a console control event swept up every engine, including the one mid-run that a restarted server is supposed to adopt. A run in flight when the server stopped was lost rather than resumed. Engines now get their own group on both platforms. - `server.json` records a URL clients can actually use. It carried the address the listener bound, so starting the server on every interface wrote `http://0.0.0.0:` — a bind address, not a destination. Linux and macOS route that to loopback, so it worked by accident; Windows refuses it, and nothing that reads the discovery file could find the server there. `url` now names loopback when the bind address is unspecified, and `host` still records what was bound. - **Windows: `cereyan serve` can be stopped gracefully.** It handled `SIGTERM`, which Windows never delivers, so anything stopping the server other than an interactive Ctrl-C — a service manager, a script, a supervisor — ended the process before it could shut down: the discovery file was left behind, pending writes were not flushed, the WAL was not checkpointed, and engine processes were orphaned. A console control event arrives as `SIGBREAK` on Windows and is now handled the same way `SIGTERM` is on Unix. Unix behaviour is unchanged. Found by running the Python test suite on Windows for the first time. ## 1.4.0 (2026-09-08) - The Intel macOS wheel is cross-built from the arm64 runner. GitHub retired the `macos-13` label, so that build queued forever and, because the smoke test and the PyPI upload wait for every wheel, no release could complete at all. - `cargo bench` runs again. Criterion's flags were being handed to the auto libtest harness of every lib target, which rejects them, so the benchmarks stopped at the first target reached — through `just bench` as much as in CI. Each crate's lib now sets `bench = false`. - Wall-clock ceilings moved out of the test suite that gates a release. Two Rust timing tests are `#[ignore]` and the Python `performance` marker is deselected there; both run in `just bench` and in CI's benchmark job, which is where the per-platform baselines live. A ceiling calibrated on a developer machine failing on a shared CI runner said nothing about correctness. `benches/e2e.py --no-ceilings` reports an absolute miss instead of failing, while a regression against a baseline still fails. - Windows: `cereyan-server` now compiles. The Unix socket listener's serve path was never excluded on platforms without Unix sockets, so the crate failed to build and the Windows wheel and test job have been broken since the socket landed in 1.1. Windows had no working wheel for 1.1, 1.2, or 1.3 despite being listed as a supported platform. - Engines now end with the server. A graceful stop answers every waiting engine with an instruction to exit and signals any that were idle between requests, so `cereyan serve` leaves no engine child behind; an engine executing a run is deliberately left alone, so a restarted server still adopts its run. An engine that loses its server without being told to stop — a kill, a crash — exits by itself after thirty seconds of failing to reach it. **This reverses a documented guarantee:** the server previously exited without terminating any engine process, so anything that relied on the warm pool outliving a graceful stop or restart now sees a cold pool instead. - Removed `exit_code` from runs. The field was in the `Run` model, the OpenAPI and MCP surfaces, and the run page, but nothing ever wrote it, so it was always null; the store's `set_run_exit` write path had no callers. Clients reading `run.exit_code` should drop it. The SQLite column stays, unread, because migrations are append-only. - Packaging: the wheel and sdist now carry `LICENSE` and `NOTICE`, project URLs, keywords, an author, and a fuller classifier set, so the PyPI page renders its links, licence, and images. Tagging a release now publishes to PyPI from CI through trusted publishing, which the release checklist already promised. - UI redesign: a top bar replaces the sidebar and breadcrumb bar, with the eight sections as tabs, a project switcher that scopes every list, a ⌘K palette that jumps to sections, flows, runs, and artifacts, and a warm neutral palette in light and dark with Geist bundled. Controls are shadcn components. The dashboard leads with a Needs attention list (paused, failed, crashed, and late runs with inline actions), Running now with task progress, and the histogram with an axis; the runs page has popover filters, a task-state bar per run, and a floating selection bar; the run page is a workbench with a tasks rail that filters the logs to one task run and shows retry countdowns; the flows page groups by project with the schedule in words, a run-history sparkline, and a Dependencies panel of `after=` chains and fan-in groups. - API: run list items and `GET /api/runs/{id}` carry `task_counts` by task state; `recent_runs` entries on flows gain the run's duration; `AwaitingRetry` details record `retries`. - Documentation authoring guide: `docs/AGENTS.md` now exists, covering the layout, the generated pages and their generators, the page kinds and section budgets, the vocabulary, the style rules, the tested-block markers and fixtures, the redirect rule, and the commands. `just lint` fails when it is missing or stops naming a generated page, which is what let the documentation-restructure entry below promise a file that was never written. - The MCP reference is generated: `scripts/gen_mcp_reference.py` renders `docs/reference/mcp.md` from `tests/mcp_snapshot.json`, a snapshot of the server's own handshake, tools, resource templates, prompts, and response keys, checked in `just lint` like the CLI and HTTP references. The page now carries the protocol version, the instructions the server gives the model, each argument's type, default, and range, the keys every tool returns (including `next_cursor` and `verdict`), the fields of the lists they hold, and the prompt text. - Documentation: `cereyan mcp --token` never worked; `--token` is global and goes before the subcommand. The agent guide says so, and calls the two resources templates, which is what they are. - Overlap soak (`just soak`, `benches/soak_overlap.py`): fifteen scheduled flows across `enqueue`, `skip`, and `cancel_new` run for an hour against a real server and are checked against nine overlap invariants; `--quick` for twelve minutes, `--keep` to browse the UI afterwards, manual `workflow_dispatch` job in CI. - Documentation restructured into Get started, Concepts, Guides, Reference, Examples, and Design on a Material for MkDocs site with `llms.txt` output; every Python block in the docs and every file under `examples/` is executed by the test suite; the CLI and HTTP API references are generated from the parser and the OpenAPI snapshot; a docstring check covers the public API; `docs/AGENTS.md` and an OpenSpec rule keep future changes documented. Old page URLs redirect. - Docstrings for every public class, function, and method, and help text for every CLI option (`cereyan --help`). No behaviour changes. - Seven literate examples under `examples/`: quickstart pipeline, daily ETL, fan-in, approval, alerting, webhook route, agent diagnosis. ## 1.3.0 (2026-09-06) - Fan-in dependencies: `@flow(after=["a", "b"], batch_key="day")` runs the downstream once per key value after every upstream completed it; `flow.fan_in` events, upstream lists in the flow summary and graph. ## 1.2.0 (2026-09-06) - Built-in MCP server: `POST /mcp` (Streamable HTTP, JSON replies) and `cereyan mcp` (stdio proxy) with fifteen curated tools, two resources, and a `diagnose_run` prompt; runs started by agents record `created_by = mcp:`. - Human-in-the-loop: `wait_for_input(prompt, schema=None)` pauses a run, `POST /api/runs/{id}/resume` answers it, the run page shows the question with a form; the engine and resources are released while a run waits. `run.paused` and `run.resumed` events. - `Paused` is now entered only from `Running`. ## 1.1.0 (2026-09-06) - Proactive rules: `unless` with `within` (event-armed) or `at` cron with `tz` (clock-armed) fires when an expected event does not happen; lapses are `expectation.lapsed` events, expectations survive restarts, `GET /api/rules/{id}/expectations`, and an "Unless" section in the rule form. - API token: `--token`, `CEREYAN_TOKEN`, `app.serve(token=)`, or `[server] token` protects every API route except health; the UI prompts for it, clients and engines send it, and `server.json` records `auth`. - Unix socket listener (`--socket`, `CEREYAN_SOCKET`, `[server] socket`) trusted by file permission, recorded in `server.json`, and used by the Python client. - Async custom route handlers on one shared event loop. - Artifacts page and `GET /api/artifacts` with filters, keyset pagination, and per-key history. - Negative `priority` lowers engine niceness on Unix. - Downgrading to 1.0.x: delete rules that use `unless` first; 1.0 does not load them. ## 1.0.0 (2026-09-06) First release. One wheel, no runtime dependencies, Python 3.11 or newer. - Flows and tasks with parameters from type hints, offline execution into a local SQLite store, and the `cereyan run` and `cereyan runs ls` commands. - `cereyan serve`: HTTP API with an OpenAPI document, server-sent events, a warm pool of engine processes that survive server restarts, custom routes, and the embedded React UI. - One runtime home per machine (`--home`, `CEREYAN_HOME`, `~/.cereyan`); flows identified by project and name; cross-project handoff from scripts to a running server. - Schedules (cron, interval, RRule) with timezones and catch-up policies; retries, timeouts, hooks, and crash chains; Targets with atomic writes; input and source caching; backfills; resources, priority, overlap policies, and disable windows; concurrent tasks with futures, map, thread and process runners; single-upstream flow dependencies; timeline graph. - Events, rules with seven actions and templating, artifacts, variables with encrypted secrets, settings and retention, benchmarks against the performance targets, docs, and the release wheel matrix. - Performance: engine reports are chunked and accepted up to 64 MB, task-run events are appended in one batch per report, and opening a large database no longer reads the whole file when the previous shutdown was clean. # Contributing Cereyan is a Cargo workspace (`crates/`) with a Python package (`python/cereyan`), a React UI (`ui/`), and this documentation (`docs/`), built as one wheel by maturin. Design decisions and the phased history live in `roadmap.md`. ## Set up You need Rust (stable), Python 3.11 or newer with [uv](https://docs.astral.sh/uv/), Node 22, and [just](https://github.com/casey/just). ``` git clone https://github.com/sercanatalik/cereyan cd cereyan just ui # install UI dependencies and build ui/dist, which the server crate embeds just dev # uv sync and maturin develop: builds the extension in place ``` ## Everyday commands | Command | Does | | ------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `just test` | Rust tests, Python tests, documentation tests, UI tests. CI runs every suite on Linux, macOS, and Windows; control a server process in a test through `tests/server_helpers.py`, which knows the platforms' differences. Correctness only: tests that assert a wall-clock ceiling are marked `#[ignore]` or `@pytest.mark.performance` and run in `just bench` instead, because a ceiling calibrated on a developer machine fails on a shared CI runner and says nothing about correctness | | `just lint` | rustfmt and clippy, Python compile check, docstring check, generated-page checks, UI lint and client drift check | | `just docs` | Regenerate the reference pages and example pages, check docstrings, build the site into `site/` in strict mode, print the tested-block summary | | `just docs-serve` | Serve the docs locally with live reload | | `just docs-test` | Execute every Python block under `docs/` and every file under `examples/` | | `just demo` | Serve `examples/` on a temporary home at http://127.0.0.1:4200 | | `just service-sync` | Regenerate the UI's typed client from `ui/openapi.snapshot.json` | | `just bench` | Criterion benchmarks, the wall-clock tests `just test` skips, and the end-to-end benchmark against the checked-in baseline. Only the end-to-end benchmark gates; criterion's baselines are not committed, so it compares against the previous run on the same machine and is what you use while changing the store. Run it on calibrated hardware: CI runs the criterion benchmarks and the end-to-end targets advisorily, and does not run the wall-clock tests at all | | `just soak` | The one-hour overlap soak: fifteen scheduled flows whose runs outlast their interval, checked against the overlap invariants at the end. `just soak --quick` takes twelve minutes; `--keep` leaves the UI up. Not a regression gate and not on the default CI path | | `just build` | Build a release wheel into `dist/` | ## Where things are | Path | Contents | | -------------------------------- | ------------------------------------------------------------------------------------------------------- | | `crates/core` | The model and the state rules: no I/O, no async | | `crates/store` | SQLite: writer thread, read pool, migrations, quarantine | | `crates/rules` | Rule matching and templating | | `crates/server` | axum: API, UI serving, scheduler, supervisor, rules engine, MCP, retention | | `crates/py` | The pyo3 module `cereyan._core` | | `python/cereyan` | Decorators, parameters, targets, results, client, CLI, engine child, MCP stdio proxy | | `ui/` | React, Tanstack, shadcn; embedded in the wheel at build time | | `docs/`, `examples/`, `scripts/` | This site, the literate examples, the generators; `docs/AGENTS.md` is the authoring guide | | `tests/` | Python tests; `tests/docs/` holds the documentation fixtures, `tests/mcp_snapshot.json` the MCP surface | | `benches/` | Benchmarks, per-platform baselines, and the overlap soak (`soak_overlap.py`) | ## The snapshots `ui/openapi.snapshot.json` is the contract between the server, the UI client, and the HTTP API reference. The Python suite asserts the live server's document matches it, the UI build fails when the generated client drifts, and `just lint` fails when the reference page is stale. After changing a route: ``` CEREYAN_UPDATE_SNAPSHOTS=1 uv run pytest tests/test_server_api.py -k openapi just service-sync just docs ``` `tests/mcp_snapshot.json` does the same for the MCP surface: the `initialize` handshake, every tool with its schema, the resource templates, the prompts, and the keys each tool returns. `scripts/gen_mcp_reference.py` renders `docs/reference/mcp.md` from it and `scripts/mcp_reference_template.md`. After changing `crates/server/src/mcp.rs`: ``` CEREYAN_UPDATE_SNAPSHOTS=1 uv run pytest tests/test_agent_mcp.py -k snapshot just docs ``` ## Writing documentation `docs/AGENTS.md` is the authoring guide: page kinds, the section budgets, the vocabulary, the style, how code blocks are tested and marked, which pages are generated, and how to redirect a moved page. In short: every Python block runs in tests, every feature is documented once in the section its kind of content belongs to, and every change's tasks include a documentation task. ## Release checklist 1. `just lint` and `just test` are green on the release commit. 1. `just bench` shows no target regressing more than 20 percent against `benches/baseline..json` (`just bench-baseline` writes it for the current platform); update the baseline in the same change when a regression is intentional. Run this on calibrated hardware: CI's benchmark job passes `--no-ceilings`, because a shared runner has no baseline and cannot meet absolute targets, so it reports rather than gates. 1. Bump the version in `pyproject.toml`, `Cargo.toml` (workspace), `python/cereyan/__init__.py`, and `ui/package.json`, and regenerate `Cargo.lock`. One version, four files: `test_the_four_version_strings_agree` fails when one of them is missed. The version is also in the OpenAPI document's `info` block, so refresh the snapshot and the reference page it feeds: ``` CEREYAN_UPDATE_SNAPSHOTS=1 uv run pytest tests/test_server_api.py -k openapi just service-sync just docs ``` 4. Move the `CHANGELOG.md` entries under a heading for the new version. Every change to the Python API, the HTTP API, the CLI, the MCP surface, the UI, or the behaviour of a running server needs an entry, and an entry that removes or reverses documented behaviour says what a reader relying on it must do. 1. CI builds the UI, then wheels for macOS arm64 and x86_64 (the Intel one cross-built from the arm64 runner), Linux x86_64 and aarch64 (manylinux 2.28), and Windows x86_64, plus the sdist. 1. The smoke stage installs each wheel into a fresh virtual environment on its platform and runs `scripts/smoke.sh`: import, offline run, `cereyan runs ls`. 1. Tag the release as `v`. The `docs` job uploads the built site as a Pages artifact, `deploy-docs` publishes it, and only then does `publish` upload the wheels and the sdist to PyPI — so the package page never goes live linking to a site that does not yet exist. ### One-time setup Two things live in GitHub rather than in the repository, and a fork or a restored repository needs both before its first tagged release. **GitHub Pages.** The site is deployed from a build artifact, not from a `gh-pages` branch, so Pages must be set to the workflow build type. This works on a repository that has never deployed anything: ``` gh api -X POST repos///pages -f build_type=workflow ``` The `github-pages` environment GitHub creates alongside it allows deployments only from the default branch, which would reject every tag-triggered deploy. Add a tag policy: ``` gh api -X POST repos///environments/github-pages/deployment-branch-policies \ -f name='v*' -f type=tag ``` **PyPI.** Publishing uses [trusted publishing](https://docs.pypi.org/trusted-publishers/): PyPI holds a publisher for this repository, workflow `ci.yml`, environment `pypi`, and CI mints a short-lived token for it, so no API token is stored in the repository. Set the publisher up on PyPI and create the `pypi` environment in the repository settings. Nothing else in CI needs credentials. ## License MIT. UI components adapted from Prefect are listed in `NOTICE` and remain under the Apache License 2.0 of their origin.