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

1b. Starting From an Existing SKILL.md

Most procedures worth moving into Lightflow already exist as a skill: a numbered list of steps an agent follows by hand. The translation is mechanical once each step is classified:

Worked example. A skill that reads:

1. Query last week's incidents from the tracker.
2. Group them by service and draft a summary.
3. Get the on-call lead to confirm the summary.
4. Post it to the channel.

maps to:

name: weekly_incident_digest
actions:
  - {id: fetch,  python_import: digest.actions.fetch_incidents}
  - {id: group,  python_import: digest.actions.group_by_service}
  - {id: post,   python_import: digest.actions.post_to_channel}
stages:
  - name: fetch_incidents                       # step 1: mechanical
    python_action: {action_id: fetch}
    retry_policy: {max_attempts: 3, initial_backoff_seconds: 5}
  - name: group_by_service                      # step 2a: mechanical
    run_after: [fetch_incidents]
    python_action: {action_id: group}
  - name: draft_summary                         # step 2b: agent judgment
    run_after: [group_by_service]
    operator_action:
      instructions: "'Draft a 5-line summary of ' + string(payload.outputs.group_by_service.count) + ' incidents grouped by service; resume with summary_md.'"
      json_schema: '{"type": "object", "required": ["summary_md"], "properties": {"summary_md": {"type": "string"}}}'
  - name: lead_confirms                         # step 3: a person decides
    run_after: [draft_summary]
    operator_action:
      instructions: "'Ask the on-call lead to confirm this summary before it is posted: ' + payload.outputs.draft_summary.summary_md"
  - name: post_to_channel                       # step 4: mechanical
    run_after: [lead_confirms]
    python_action: {action_id: post}

Keep the skill: replace its step list with lightflow start --lightflow=<dir> and the gate-handling rules from skills/run_lightflows/SKILL.md. The judgment and the decision stay where they were; only the mechanical steps and the ordering move into code. Run the create_lightflow meta-workflow above to scaffold and prove the result.

2. Workflow Manifest Reference (lightflow.yaml)

Define actions and stages in lightflow.yaml (or .json). 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.
# Optional: payload keys (any depth) shown as <redacted> in the visualizer,
# status --verbose and dry-run output. passport.json keeps the real values.
secret_keys: [api_token]

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. 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).
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 (0–86400; 0 disables the bound) 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; keep cleanup stages idempotent, since an ALL_DONE stage that ran on a failure path runs again after resume.

   A polling stage in YAML (the tick action goes under poll_tick_action; a stage may not also set python_action):

   - name: await_job
     run_after: [submit_job]
     polling_policy:
       condition: "payload.outputs.await_job.status == 'READY'"
       interval_seconds: 5
       timeout_seconds: 600        # or max_attempts: 120
       poll_tick_action:
         action_id: check_job      # (dict, message) per tick

   Each tick's returned dict is merged into payload.outputs.await_job before condition is evaluated, so a tick can read its own previous output.

5. Converging After a Conditional Branch: Two stages with complementary run_if conditions leave one of them SKIPPED. A stage that run_afters both must set trigger_rule: NONE_FAILED to run (the default ALL_SUCCESS skips it because an upstream was skipped; ALL_DONE would also run it after an upstream failure, which is only right for cleanup stages). dry_run points this out when it sees it.
6. Optional Keys with has(...): Under strict_expressions, read a key that may be absent (e.g. an optional gate field) as !has(payload.outputs.gate.note) || payload.outputs.gate.note != 'skip'. has() never raises.
7. Strict Expressions & Explicit Namespaces: Set top-level strict_expressions: true on new lightflows so a misspelled or not-yet-populated key in run_if / condition / instructions fails the stage instead of silently evaluating to null (guard optional keys with has(...)). Read upstream values as payload.outputs.<stage>.<key> rather than the top-level mirror, which is last-writer-wins across stages; the engine warns when a stage overwrites a top-level key that an expression reads. outputs is a reserved top-level key (--payload may not set it).

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 --output=/tmp/visualizer.html