/cao-workflow

name: cao-workflow
description: Author and run CAO Python workflow scripts — multi-step, parameterized, fan-out
  orchestrations executed by `cao workflow run`. Use when the user wants a repeatable multi-step
  job (e.g. data analysis over many files, a review pipeline, a parameterized batch). Authoring
  ends at a validated script file; running it is a separate, user-approved step.

CAO Workflows

A CAO workflow is a **Python script** you write, validate, and — only after asking the user — run through `cao workflow run`. Each script drives one or more agent *steps* through CAO's shared substrate, so you can fan work out across agents, collect their results, and resume a run that was interrupted.

> Your job as an author ends at a **validated script file on disk**. Authoring does NOT run the > workflow. Never claim a workflow ran, or will run, when all you did was write it. Running is a > separate step the user must approve (see Lifecycle step c).

When to use

Reach for this skill when the user asks to **build or run a multi-step or parameterized workflow** — for example:

• "Analyze every file in `reports/` and summarize the findings."
• "Run a review pipeline: implement, then review, then verify."
• "Do the same batch job but with a different input directory each time."

If the work is a single one-off agent call, you don't need a workflow. Workflows earn their keep when there are multiple steps, fan-out, parameterization, or a need to resume.

The script API

Author scripts import from the `cao_workflow` package. This package runs **only in the script subprocess** and imports nothing from `cli_agent_orchestrator.*` — it talks to CAO over HTTP. Its public surface:

• `run_step(provider, agent, prompt, *, step_id=None, timeout=None, **opts) -> StepHandle` —

run one agent step. `StepHandle` has `.step_id`, `.terminal_id`, `.output`, `.status`.

• `get_inputs() -> dict` — the run's resolved inputs (see Parameterized workflows). Returns

`{}` when nothing was declared; never raises on absence.

• `emit_output(value)` — print the run-level `CAO_WORKFLOW_OUTPUT:` sentinel (the run's return).
• `ShimError` (and `ShimIdentityError`, `ShimTransportError`, `ShimHTTPError`) — the failure

hierarchy `run_step` raises. Failures surface **unchanged** — the shim never retries.

Lifecycle

Follow every step in order. **No step may be skipped** — validate is mandatory, and you must ask before running.

a. AUTHOR

Write a `.py` file to `~/.aws/cli-agent-orchestrator/workflows/<name>.py`. The workflow is **run by its stem** (`<name>`), so:

• The name must be a bare stem — **no path separators**, no directory prefix.
• Do **not** create a same-stem `.yaml` sibling — a `<name>.yaml` next to `<name>.py` collides

on the run surface.

b. VALIDATE (mandatory gate)

cao workflow validate ~/.aws/cli-agent-orchestrator/workflows/<name>.py

Fix **every** finding before proceeding — the lint findings are **load-bearing**, not style nits:

• **`import cli_agent_orchestrator` is banned.** The script runs in a separate subprocess and

must reach CAO only over HTTP (the `cao_workflow` shim). Importing the server package breaks that boundary.

• **`random` / `time` / `datetime` / `uuid` warnings.** Resume **re-executes the script

top-to-bottom** and replays journaled step results. Any nondeterministic value computed at the top level will differ on replay and raise `ReplayDivergenceError`. Keep the script deterministic: derive IDs from inputs, not from the clock or an RNG.

c. ASK the user — NEVER auto-run

The script tier executes generated Python. **Never run a workflow without the user's explicit approval.** Present the validated file and ask before doing anything in step d.

d. RUN with an explicit, pre-announced run-id

Announce the run-id before you start so the user can cancel it: "Starting run `kb-1` — cancel with `cao workflow cancel kb-1`."

Choose the invocation by how the run is triggered, because the two paths have very different client-side ceilings:

• **`cao workflow run` (CLI)** uses a client socket timeout of **~8820s (~2.45h)** — the CLI

itself won't give up early.

• **`workflow_run` MCP tool** is bounded by the **MCP host's own per-tool-call timeout** — a

host-dependent, much-shorter limit that can **drop a long blocking call and lose its return value even though the server run keeps going**.

So:

• **Short runs**: call the `workflow_run` MCP tool (blocking) and read the result directly.
• **Long runs**: background the run and poll, rather than blocking on it —

  cao workflow run <name> --run-id <id> --json &

Backgrounding keeps the run alive server-side without a short MCP host timeout silently dropping the return.

e. RESUME

cao workflow resume <run-id>

Resume replays completed steps from the journal and continues from the first incomplete one. Deterministic scripts (see step b) resume clean; nondeterministic ones diverge.

Parameterized workflows

Instead of editing a constant per run, declare inputs once and pass values at invocation time.

Add a **module-level `INPUTS` dict** and read the resolved values at runtime with `get_inputs()`:

from cao_workflow import get_inputs

INPUTS = {
    "target_dir": {"type": "path", "required": True},
    "max_files":  {"type": "int",  "required": False, "default": 20},
    "verbose":    {"type": "bool", "required": False, "default": False},
}

inputs = get_inputs()
target_dir = inputs["target_dir"]
max_files = inputs.get("max_files", 20)

Each entry declares `type` (`string` | `int` | `bool` | `path`), `required`, and an optional `default`. This makes one authored script reusable — "author once, invoke with inputs."

Operational discipline

These rules are load-bearing. Each is paired with the reason it exists.

R1 — Fan-out determinism

To run steps concurrently, use a `ThreadPoolExecutor` and give **every concurrent `run_step` an explicit, st