Cao Workflow

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:

• step(provider, agent, prompt, *, recovery, step_id=None, timeout=None, **opts) -> StepHandle — run one agent step and declare what re-running it would mean. recovery is keyword-only with no default, so omitting it is a TypeError at the call. See "Declaring a recovery policy" below before you pick a value.
• run_step(provider, agent, prompt, *, step_id=None, timeout=None, **opts) -> StepHandle — the same call, declaring no policy. That is the only difference between the two. A recovery= passed to run_step lands in **opts; the server validates it, the shim does not — see below.
• StepHandle has five fields: .step_id, .terminal_id, .output, .status, and .replayed. .replayed qualifies .terminal_id. When it is True the server returned a stored result and ran nothing, and .terminal_id is the ORIGINAL id — it names a terminal that no longer exists. That flag is the only thing standing between you and reading, writing to, or waiting on a dead id, so check it before you touch .terminal_id.
• 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 step and run_step raise. Failures surface unchanged — the shim never retries.

Declaring a recovery policy

recovery= is the author's claim about the step, and nothing more. CAO has no mechanism to prove what a step does to the outside world, so it cannot and does not verify the claim. A recovery policy DECLARES what re-running this step would mean; it never grants permission.

The three values, all of which are statements you are making, not protections you are getting:

"idempotent" grants nothing and protects nothing. It does not make a step safe to re-run; it tells the resume gate that you believe it already is — and wherever the gate would otherwise stop and ask a human, it re-executes the step on your word instead. Declare it on a step that charges a card, sends mail, or files a ticket and CAO will charge the card again, exactly as instructed. If you cannot show the step is safe to repeat, "manual" is the honest declaration.

Omitting a policy is a fourth, distinct state — it is never silently read as "manual". Use run_step for it deliberately: an undeclared step still replays (replay executes nothing), but where the alternative is re-execution it halts for a human.

recovery= on run_step is checked late, not never. run_step has no recovery parameter, so the value rides **opts to the server, which stores it, lets the resume gate honour it, and rejects an unknown value with a 422 — the route types that field as the closed policy enum. What run_step lacks is step()'s client-side check, which refuses a bad value before any HTTP attempt; on run_step a typo instead fails that step mid-run. Neither surface has its value checked by validate (the linter sees the keyword, not its contents), which is why validate reports the run_step form as unenforced-recovery-policy. Use step() to declare, and run_step only to declare nothing.

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.
• missing-recovery-policy is a blocking ERROR. A step() call with no recovery= keyword fails validation — the signature requires one and so does the linter. Two related warnings fire without blocking: unverifiable-recovery-policy (a step() call passing **kwargs, so the linter cannot see whether a policy is in there) and unenforced-recovery-policy (a recovery= on run_step, which is honoured at resume and validated by the server with a 422, but is not checked client-side before it is sent). See "Declaring a recovery policy" above.

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 re-executes the script top-to-bottom — that is what step b's determinism warning is about — and the server decides each step call as it arrives. Never assume your top-level code does not re-run. Each step lands on one of three outcomes:

• replayed — the stored result is returned and nothing runs. StepHandle.replayed is True, and its .terminal_id names a terminal that no longer exists.
• executed — the step runs again for real.
• halted — CAO will not decide this one alone, so the run stops there and waits for a human.

A fourth outcome ends the whole run rather than one step: if the script changed at a step's key, that step diverges and the run fails with ReplayDivergenceError. Deterministic scripts (see step b) resume clean; nondeterministic ones diverge.

Resolving a halt

A halt reaches your script as a ShimHTTPError whose .status is 409 and whose .body names kind: "decision_required", the step_id, and which condition halted it. A step halts when its outcome is genuinely unknown or unverifiable: it was dispatched and never settled and no declared policy permits re-execution; its stored result is unreadable; its recorded provenance cannot be verified under the current scheme; or its author declared recovery="manual" and asked to see it.

Resolve it by naming a decision per halted step and resuming again:

cao workflow resume <run-id> --decide <step_id>=rerun   # re-execute that step
cao workflow resume <run-id> --decide <step_id>=skip    # accept its stored result

--decide is repeatable, one per halted step.

A decision authorises exactly ONE attempt. If that attempt crashes before it settles, the next resume asks again rather than re-executing on the old consent. Consent does not carry forward — never present one rerun to a user as standing authorisation for later resumes.

Do not let a blanket except ShimError swallow a halt (see R4): ShimHTTPError is a ShimError, so a catch-all around a step absorbs the 409 and the run finishes with a sentinel where a human decision was required. Re-raise when .status == 409.

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, stable step_id. The sequential call-N counter fallback is race-free but not deterministic across runs under concurrent scheduling — so resume would replay the wrong results. Iterate over sorted() inputs so the mapping from item → step_id is stable.

Default max_workers=2 for claude_code (measured: 4 starved the heaviest lens). Expose it as a tunable input; higher values are fine when steps are light.

R2 — Secrets as references, never literals

Inputs are journaled in plaintext and replayed on resume. Never pass a literal secret (token, key, password) as an input. Pass a name/reference and resolve the actual secret at step time (env var, secrets manager) inside the step.

R3 — Role-capability matching

Only write-capable roles (e.g. developer) should be told to write files. A read-only role (e.g. reviewer) instructed to write will hang the full step budget waiting on a permission it can't get. Read-only steps must READ their inputs and RETURN findings inline.

R4 — Per-unit fault tolerance

Catch ShimError inside each fan-out unit so one step's timeout degrades to a survivor set rather than failing the whole run with a 504. Return a sentinel/None for the failed unit and let the aggregate proceed.

But do not swallow a halt or a divergence. ShimHTTPError is a ShimError, so the same catch also absorbs the 409 a resume raises when a step halts or diverges — and the run then completes with a sentinel in place of a result a human was supposed to decide on. Re-raise when .status == 409 (see Resolving a halt).

Big-outputs discipline

For large results, have the step write to a file and return the path — don't return megabytes inline. Per-step output is null for schema-less steps; the files (and the aggregate you build) are the source of truth.

R5 (INTERIM) — Prefer a headless provider

Prefer claude_code as the step provider. kiro_cli currently launches an interactive TUI that hangs run_step. This is interim guidance — a kiro mitigation is a tracked follow-up, not a permanent verdict — but until it lands, use a headless provider.

Projection ranking

The runtime journal is the primary truth for progress and UI — it reflects what actually ran. A static script→YAML preview is optional and lossy; never treat it as the truth source and never author against it.

Handoff when you're read-only

If you lack write permission (you can't create the .py file), hand off authoring to a developer agent, and pass this skill's name (cao-workflow) in the handoff message so the developer follows the same lifecycle.

Honesty discipline

• Never claim a workflow ran that didn't.
• Authoring ends at a validated file; running is a separate, user-approved step.
• Be honest about failures — surface ShimErrors and non-zero validate findings; don't paper over them.

Worked example — parameterized fan-out

A script that summarizes each file in a directory concurrently, with a stable step_id per file, per-unit fault tolerance, and results written to disk:

"""summarize_dir — fan out a summary step over every file in target_dir."""
import os
from concurrent.futures import ThreadPoolExecutor

from cao_workflow import run_step, emit_output, get_inputs, ShimError

# Parameterized: author once, invoke with different inputs.
INPUTS = {
    "target_dir":  {"type": "path", "required": True},
    "max_workers": {"type": "int",  "required": False, "default": 2},
}

inputs = get_inputs()
target_dir = inputs["target_dir"]
max_workers = inputs.get("max_workers", 2)

# sorted() → the item→step_id mapping is stable across runs (R1 determinism).
files = sorted(
    name for name in os.listdir(target_dir)
    if os.path.isfile(os.path.join(target_dir, name))
)


def summarize(filename: str):
    path = os.path.join(target_dir, filename)
    try:
        # Explicit, STABLE step_id per concurrent call (R1). Read-only role
        # RETURNS its summary inline (R3) — it does not write files.
        handle = run_step(
            provider="claude_code",          # headless (R5)
            agent="reviewer",
            prompt=f"Summarize the file at {path} in 3 bullet points. Return the summary only.",
            step_id=f"summarize:{filename}",
        )
        return filename, handle.output
    except ShimError as exc:
        # Per-unit tolerance (R4): one timeout degrades to a survivor, not a 504.
        return filename, f"ERROR: {exc}"


with ThreadPoolExecutor(max_workers=max_workers) as pool:
    results = dict(pool.map(summarize, files))

# Big output → write to a file, return the path (big-outputs discipline).
out_path = os.path.join(target_dir, "_summaries.json")
with open(out_path, "w") as fh:
    import json
    json.dump(results, fh, indent=2)

emit_output({"summarized": len(results), "output_file": out_path})

Validate it, ask the user, then run with a pre-announced run-id:

cao workflow validate ~/.aws/cli-agent-orchestrator/workflows/summarize_dir.py
# fix findings, then — after the user approves:
cao workflow run summarize_dir --run-id sum-1 --json &