create-lightflows

# Creating Lightflow Workflows (`create_lightflows`) Lightflow coordinates multi-step operations between AI agents and human operators using an append-only state ledger (`passport.json`), safe AST-based CEL-like expressions, and selective subgraph re-arming. > **Note**: If you only need to execute, inspect, or resume an *existing* > Lightflow workflow, read the `run_lightflows` skill instead. ## 1. Scaffolding a New Workflow via `create_lightflow` (3-Phase Meta-Workflow) Instead of authoring a workflow from scratch in one shot, run the built-in 3-phase meta-workflow, which gates progress on deterministic AST signature checks, `sample_payload` return-contract smoke tests, unit test execution, and `payload.outputs.<stage>` DAG cross-checks. The meta-workflow ships in the Lightflow repository under `examples/create_lightflow/`: run it from a clone of `github.com/google/lightflow`, or pass the absolute path to `examples/create_lightflow` in your clone: ```bash lightflow start \ --lightflow=/path/to/lightflow/examples/create_lightflow \ --log_id=scaffold_<workflow_name> ``` (`python3 -m lightflow` is equivalent to `lightflow` whenever the console script is not on `PATH`.) 1. **`align_on_design`** (`operator_action`): Align with the user on stage boundaries (separate read/compute stages from external write/mutate stages), `operator_action` gates, and the output keys each stage produces under `payload.outputs.<stage>`. 2. **`implement_and_test_actions` $\rightarrow$ `verify_actions`**: Write `actions.py` + unit tests (or pass `sample_payload`); `verify_actions` parses the AST, smoke-tests that every action returns a JSON-serializable `(dict, str)` tuple, and runs `unittest`. 3. **`draft_and_explain_manifest` $\rightarrow$ `prove_manifest_dry_run`**: Write `lightflow.yaml` and explain the DAG to the user; `prove_manifest_dry_run` compiles the DAG, verifies that every `payload.outputs.<stage>` reference points to an upstream dependency in `run_after`, runs `dry_run`, and emits `visualizer.html`. ## 2. Workflow Manifest Reference (`lightflow.yaml`) Define actions and stages in `lightflow.yaml` (or `.json` / `.textproto`). `python_import` resolves relative to the workflow's directory (e.g. `actions.fetch_metrics` for `actions.py` in the same folder) as well as package paths: ```yaml name: release_verification_pipeline description: Fetch metrics, pause for operator review, and publish report. actions: - id: fetch_metrics python_import: actions.fetch_metrics - id: publish_report python_import: actions.publish_report - id: cleanup_temp python_import: actions.cleanup_temp stages: - name: fetch description: Fetch latest metrics timeout_seconds: 60 python_action: action_id: fetch_metrics static_kwargs: limit: 500 retry_policy: max_attempts: 3 initial_backoff_seconds: 5 backoff_multiplier: 2.0 rollback_action: action_id: cleanup_temp - name: approve description: Operator approval gate run_after: [fetch] operator_action: # CEL-like expression: string literals MUST be quoted inside the string. instructions: "'Ready to publish: ' + payload.outputs.fetch.summary" json_schema: type: object required: [approved] properties: approved: type: boolean - name: publish description: Publish verified report run_after: [approve] run_if: "payload.outputs.approve.approved == true" python_action: action_id: publish_report - name: always_cleanup run_after: [publish] trigger_rule: ALL_DONE python_action: action_id: cleanup_temp ``` ## 3. Three Rules for Crash-Free Python Actions ```python from typing import Any def fetch_metrics( payload: dict[str, Any], **static_kwargs: Any ) -> tuple[dict[str, Any], str]: limit = static_kwargs.get("limit", 100) return {"row_count": limit, "summary": f"{limit} rows"}, "Fetched metrics" ``` 1. **One Side Effect per Stage**: Separate read/compute stages from external write/mutate stages. Lightflow's `Passport` never re-runs `COMPLETED` stages on `resume`; make write stages safe to retry if they fail mid-execution (overwrite files instead of appending, or attach a `rollback_action`). 2. **Consistent Output Keys (`payload.outputs.<stage_name>`)**: Every return path of an action (including any `offline` test mode for external APIs) must return a JSON-serializable `(dict, str)` with the **exact same dictionary keys** so downstream `run_if` and `instructions` expressions never crash. 3. **Cross-Check `payload.outputs.<stage>` Against `run_after`**: Any stage whose `run_if` or `instructions` reads `payload.outputs.<upstream>.<key>` must list `<upstream>` (directly or transitively) in `run_after`. 4. **Timeouts, Polling & Resilience**: Set stage-level `timeout_seconds` (`1`–`86400`) on a stage to bound a `python_action` (or individual poll tick); for long-running external jobs, prefer `polling_policy` (`condition`, `interval_seconds`, `timeout_seconds`, `max_attempts`, optional `poll_tick_action`), which persists each tick's progress message to `passport.json` so `lightflow status` shows live progress. Use `retry_policy` for transient errors and `rollback_action` / `trigger_rule: ALL_DONE` for cleanup. ## 4. Validating Your New Workflow Before handing off or running in production, always verify compilation, CEL-like expression evaluation, and gate schemas: ```bash lightflow dry_run --lightflow=lightflow.yaml lightflow render --lightflow=lightflow.yaml lightflow visualize --lightflow=lightflow.yaml --out=/tmp/visualizer.html ```