version: "1.0.0" name: create_lightflows description: Guides the design, authoring, compilation, and verification of new Python DAG workflows using Lightflow (YAML, JSON, or Textproto). Use when building a new multi-stage pipeline, authoring actions.py and lightflow.yaml, or scaffolding a workflow with the create_lightflow meta-workflow. For merely running or resuming an existing workflow, use run_lightflows instead.

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:

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>.

1. `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.

1. `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:

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

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).

1. 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.

1. 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. Validating Your New Workflow

Before handing off or running in production, always verify compilation, CEL-like expression evaluation, and gate schemas:

lightflow dry_run --lightflow=lightflow.yaml
lightflow render --lightflow=lightflow.yaml
lightflow visualize --lightflow=lightflow.yaml --out=/tmp/visualizer.html