feat(workflow): script-tier run surface + author shim — U5+U6 (#312) (#399)

* feat(workflow): script-tier run surface + author shim — U5+U6 (#312)

Tier dispatch on run/validate/resume/cancel/status, ScriptSpec + .py indexing with collision reject, typed-error→HTTP mapping, stdlib cao_workflow shim, authoring guide + examples gallery.

* fix(workflow): close CodeQL path-injection alerts + journal schema self-heal

_safe_spec_path now uses resolve()+is_relative_to (recognized sanitizer, covers both flagged sinks); workflow_journal._connect runs idempotent migrators so test_cancel_run_unknown_404 passes in isolation. Adds traversal/symlink rejection tests.

* Potential fix for pull request finding 'CodeQL / Uncontrolled data used in path expression'

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* fix(workflow): validate every spec sink + bound spec reads

_read_script_spec self-validates via _safe_spec_path (closes .py-by-name index bypass); rebuild/upsert store validated resolved paths; all four spec reads capped at WORKFLOW_MAX_SPEC_BYTES+1. Fixes cancel journal fallback, .py byte cap, UnicodeDecodeError mapping; restores absolute-path contract.

* Potential fix for pull request finding 'CodeQL / Uncontrolled data used in path expression'

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* Potential fix for pull request finding 'CodeQL / Uncontrolled data used in path expression'

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* Potential fix for pull request finding 'CodeQL / Uncontrolled data used in path expression'

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* Potential fix for pull request finding 'CodeQL / Uncontrolled data used in path expression'

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* fix(workflow): use repo-standard realpath containment idiom

_safe_spec_path now mirrors path_validation.py (realpath + startswith(base+os.sep)) — the barrier this repo's CodeQL accepts; reverts the broken autofix re-anchoring (37 test failures) and removes the crashing WORKFLOW_NAME_RE.match str call. All 13b02de protections stand.

* fix(workflow): return plain realpath str from _safe_spec_path

Matches path_validation.py's return type exactly — the Path() wrap after the containment check was the remaining round-trip CodeQL's barrier model doesn't track (alerts #158/#160). Behavior-neutral type ripple; all sinks consume the validated str directly.

* fix(workflow): script-tier list crash + steps stuck RUNNING

Bug 1: cao workflow list crashed (TypeError) on a script spec — step_count
is None and was formatted with :<6; render it as '-'.

Bug 2: script per-step state was seeded RUNNING but never transitioned;
add record_step_completion (RUNNING->COMPLETED/FAILED, mirroring the BR-31
guard + YAML tier), wired into the run_step endpoint with best-effort
journal write-through so resume sees completed steps.

* fix(workflow): CI format, env_vars on session_name path, idle completion + interruptible cancel

Black-format two files; merge per-step env_vars into existing-session windows (#408); accept stable post-input IDLE as step completion and let cancel interrupt the in-flight wait so runs converge (#409).

* fix(workflow): address script run review findings

---------

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

Author: fanhongy

.github/workflows/publish-to-pypi.yml

@@ -115,6 +115,7 @@ jobs:
           cao --help
           cao-server --help
           cao-mcp-server --help
+          python -c "import cao_workflow; assert cao_workflow.run_step"
 
   publish-pypi:
     name: Publish to PyPI (maintainer-approved)

CHANGELOG.md

@@ -25,6 +25,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - add provider support (#272)
 
+- add script-tier workflows: a `.py` workflow spec is now runnable via `cao workflow run` (and the `workflow_run`/`workflow_cancel` MCP tools), with the same `resume`/`cancel`/`status` support as YAML workflows — tier is detected automatically from the file extension (#312)
+- add optional `skills` field to `AgentProfile` to scope the per-agent skill catalog via an fnmatch allowlist; runtime-prompt providers only, `load_skill` resolution unchanged (#351)
 - add herdr terminal backend with event-driven inbox delivery (#271)
 
 - bundle built-in memory plugins for Claude Code, Kiro, and Codex (#269)

docs/examples/conditional_example.py

@@ -0,0 +1,37 @@
+"""Conditional shape — shim-based (M1, FR-7.1).
+
+Branches on a simple author-controlled condition and runs a DIFFERENT step
+depending on which branch is taken — proving only the taken branch's
+``run_step`` call executes, never both.
+
+The branch condition is a plain in-script constant (NOT read from the
+environment or an external input) — U4's constructed spawn env carries only
+the fixed identity allowlist (``CAO_WORKFLOW_RUN_ID``/``GENERATION``/
+``CAO_API_BASE_URL``/``PATH``/``HOME``), so a script's own control flow must
+be deterministic from its own source, not from ambient environment state
+(the authoring guide's determinism obligation). Edit ``IS_URGENT`` below to
+exercise the other branch.
+
+Run standalone via ``cao workflow run --script docs/examples/conditional_example.py``.
+"""
+
+from __future__ import annotations
+
+from cao_workflow import emit_output, run_step
+
+IS_URGENT = True
+
+
+def main() -> None:
+    if IS_URGENT:
+        handle = run_step("kiro_cli", "reviewer", "escalate this incident", step_id="urgent-branch")
+        branch_taken = "urgent"
+    else:
+        handle = run_step("kiro_cli", "reviewer", "file this for later", step_id="routine-branch")
+        branch_taken = "routine"
+
+    emit_output({"branch_taken": branch_taken, "output": handle.output})
+
+
+if __name__ == "__main__":
+    main()

docs/examples/fanout_example.py

@@ -0,0 +1,37 @@
+"""Fan-out shape — shim-based (M1, FR-7.1).
+
+Runs several agent steps concurrently via ``concurrent.futures``. Each
+concurrent call passes an EXPLICIT ``step_id`` — the sequential counter
+fallback is UNSAFE across runs under concurrent scheduling (BR-13); an
+explicit label per shard is REQUIRED, not optional, whenever ``run_step`` is
+called from more than one thread.
+
+Run standalone via ``cao workflow run --script docs/examples/fanout_example.py``.
+"""
+
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+from cao_workflow import emit_output, run_step
+
+SHARDS = ["alpha", "beta", "gamma"]
+
+
+def _run_shard(shard: str):
+    handle = run_step("kiro_cli", "reviewer", f"review shard {shard}", step_id=f"shard-{shard}")
+    return shard, handle.output
+
+
+def main() -> None:
+    results = {}
+    with ThreadPoolExecutor(max_workers=len(SHARDS)) as pool:
+        futures = [pool.submit(_run_shard, shard) for shard in SHARDS]
+        for future in as_completed(futures):
+            shard, output = future.result()
+            results[shard] = output
+    emit_output({"shards": results})
+
+
+if __name__ == "__main__":
+    main()

docs/examples/loop_example.py

@@ -0,0 +1,26 @@
+"""Loop shape — shim-based (M1, FR-7.1).
+
+Runs the same agent step N times, once per iteration. Sequential calls with
+no explicit ``step_id`` are safe: the shim's lock-guarded counter assigns
+``call-1``, ``call-2``, ... in program order (ADR-10's sequential case).
+
+Run standalone via ``cao workflow run --script docs/examples/loop_example.py``.
+"""
+
+from __future__ import annotations
+
+from cao_workflow import emit_output, run_step
+
+ITERATIONS = 3
+
+
+def main() -> None:
+    outputs = []
+    for i in range(ITERATIONS):
+        handle = run_step("kiro_cli", "reviewer", f"summarize item {i}")
+        outputs.append(handle.output)
+    emit_output({"iterations": ITERATIONS, "outputs": outputs})
+
+
+if __name__ == "__main__":
+    main()

docs/examples/loop_raw_http_example.py

@@ -0,0 +1,52 @@
+"""Loop shape — raw HTTP, NO shim (Q2=A shim-optionality proof, FR-6.1).
+
+Demonstrates that the shim is a convenience, not a requirement: this script
+reads the SAME three identity env vars ``cao workflow run`` injects and posts
+to ``/terminals/run-step`` with stdlib ``urllib`` directly, with no
+``cao_workflow`` import at all. Compare against ``loop_example.py`` — same
+shape, same env contract, no shim.
+
+Run standalone via ``cao workflow run --script docs/examples/loop_raw_http_example.py``.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from urllib.request import Request, urlopen
+
+ITERATIONS = 3
+
+
+def main() -> None:
+    run_id = os.environ["CAO_WORKFLOW_RUN_ID"]
+    generation = os.environ["CAO_WORKFLOW_GENERATION"]
+    base_url = os.environ["CAO_API_BASE_URL"]
+
+    outputs = []
+    for i in range(ITERATIONS):
+        body = {
+            "provider": "kiro_cli",
+            "agent": "reviewer",
+            "prompt": f"summarize item {i}",
+            "env_vars": {
+                "CAO_WORKFLOW_RUN_ID": run_id,
+                "CAO_WORKFLOW_GENERATION": generation,
+                "CAO_WORKFLOW_STEP_ID": f"call-{i + 1}",
+            },
+        }
+        request = Request(
+            f"{base_url}/terminals/run-step",
+            data=json.dumps(body).encode("utf-8"),
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        with urlopen(request, timeout=630.0) as response:
+            data = json.loads(response.read().decode("utf-8"))
+        outputs.append(data["last_message"])
+
+    print(f"CAO_WORKFLOW_OUTPUT:{json.dumps({'iterations': ITERATIONS, 'outputs': outputs})}")
+
+
+if __name__ == "__main__":
+    main()

docs/workflow-scripts-authoring-guide.md

@@ -0,0 +1,168 @@
+# Authoring `cao workflow` scripts
+
+This guide is for anyone writing a Python script to run under
+`cao workflow run --script <path>` (the "script tier" — a full Python
+program driving one or more agent steps, as opposed to the declarative YAML
+tier). It covers the `cao_workflow` shim's contract, when to reach for a
+script instead of YAML, the determinism obligation resume relies on, and the
+resume boundary itself.
+
+## The contract: `run_step` and `emit_output`
+
+Your script's entire interface to the CAO server is the `cao_workflow`
+package — a thin, stdlib-only client library (`urllib` transport, zero
+`cli_agent_orchestrator` imports) that runs *inside your script's own
+process*, not inside the server.
+
+```python
+from cao_workflow import run_step, emit_output
+
+handle = run_step("kiro_cli", "reviewer", "review this diff")
+print(handle.output)   # handle.terminal_id, handle.status also available
+
+emit_output({"reviewed": True})
+```
+
+- **`run_step(provider, agent, prompt, *, step_id=None, timeout=None, **opts) -> StepHandle`**
+  runs one agent step through the same shared substrate the server's own
+  handoff path uses. It resolves your run's identity from the environment,
+  posts to `/terminals/run-step`, and returns a `StepHandle` with
+  `.step_id`, `.terminal_id`, `.output`, `.status`.
+- **`emit_output(value)`** prints the run-level `CAO_WORKFLOW_OUTPUT:{json}`
+  sentinel line the server scans once your script exits — a one-line
+  convenience so you never hand-format the prefix/JSON encoding yourself.
+  It is pure `print()`; no HTTP call, no state.
+- **Errors are typed and never retried:** `ShimIdentityError` (identity env
+  missing — nothing was attempted), `ShimTransportError` (network failure,
+  wraps the underlying `urllib` error), `ShimHTTPError` (non-2xx response,
+  carries `.status`/`.body`). All four (`ShimError` plus these three
+  subclasses) are importable from `cao_workflow` directly:
+  `from cao_workflow import run_step, ShimHTTPError`.
+- **`step_id` is required for concurrent fan-out.** If you call `run_step`
+  from more than one thread (e.g. via `concurrent.futures`), pass an
+  explicit, stable `step_id` per call. See "Fan-out and `step_id`" below —
+  this is not optional ergonomics, it is a correctness requirement.
+- **`reuse_terminal_id` is not supported through `run_step`.** The shim
+  always sends identity `env_vars`, and the server unconditionally rejects
+  `env_vars` + `reuse_terminal_id` together (422) — `run_step` fails fast
+  client-side with a `ShimError` instead of round-tripping an opaque 422. If
+  you genuinely need terminal reuse without the identity fence, call the
+  HTTP API directly.
+
+## YAML vs. script: which tier should this workflow use?
+
+Use the **YAML tier** (`cao workflow validate`/`list`/`get`/`run` against a
+declarative spec) when your workflow is a fixed sequence or simple
+branch/loop expressible in the YAML grammar — it is simpler to author,
+lint, and reason about, and it is the tier most of CAO's tooling assumes by
+default.
+
+Use a **script** (this guide) when your workflow's control flow needs
+something YAML can't express yet: nontrivial branching logic, real
+concurrent fan-out (`concurrent.futures`/`threading`), or per-iteration
+Python computation over agent output. A script is a full Python program —
+more power, more responsibility (see the determinism obligation next).
+
+If you're unsure, start with YAML. Reach for a script only when you hit a
+concrete limitation.
+
+## The determinism obligation (and repeated work on resume)
+
+`run_step` never retries, backs off, or reconnects. A client-side retry after
+an unknown-completion-state transport failure could issue the same agent work
+twice, so failures are returned to your script unchanged.
+
+This puts an obligation on **your script**: its control flow and the
+prompts it sends must be deterministic across runs of the *same* script
+source. If your script's behavior can vary run-to-run — a `random()` call
+that isn't seeded, a prompt that embeds `datetime.now()`, branching on
+wall-clock time — a **resume** can repeat different work from the original
+attempt. The current runtime does not replay completed calls or reject that
+divergence: it re-executes the frozen script from the top, including every
+`run_step` call. Determinism makes that repeated work predictable, but your
+workflow must tolerate it being issued again.
+
+## Fan-out and `step_id`
+
+`run_step`'s step key defaults to a lock-guarded sequential counter
+(`call-1`, `call-2`, ...) when you omit `step_id`. That counter is
+**race-free** under concurrent callers — the lock guarantees no two calls
+ever get the same key — but it is **not safe for fan-out**. Thread
+scheduling, not the counter's correctness, decides which call claims which
+`call-N`, so two runs of the same fan-out script can assign `call-1`/
+`call-2` to different logical calls depending on how the OS scheduled your
+threads that run. On resume, that reassignment makes journal history and
+diagnostics refer to different logical calls under the same `call-N` key.
+
+**The rule:** any time you call `run_step` from more than one thread
+(`concurrent.futures.ThreadPoolExecutor`, manual `threading.Thread`), pass
+an explicit, stable `step_id` per call:
+
+```python
+def _run_shard(shard):
+    return run_step("kiro_cli", "reviewer", f"review {shard}", step_id=f"shard-{shard}")
+
+with ThreadPoolExecutor(max_workers=3) as pool:
+    futures = [pool.submit(_run_shard, s) for s in ("alpha", "beta", "gamma")]
+```
+
+The linter (`cao workflow validate`) does not infer concurrency or check this
+rule. It reports syntax errors, disallowed or unverifiable dynamic imports,
+and imports associated with nondeterminism. Supplying stable `step_id` values
+for fan-out is the author's responsibility.
+
+## Resume re-executes the frozen script
+
+`run_step()` behaves **identically** whether the surrounding script is a
+fresh run or a resume drive. There is no `if resuming:` branch inside the
+shim. `cao workflow resume` re-executes the frozen source snapshot from the
+top with a new generation token, and each `run_step` call makes a new HTTP
+request that executes the agent step again. Completed calls are journaled,
+and a replay lookup primitive exists in the journal layer, but it is not
+connected to the run-step route and cannot currently suppress repeated work.
+
+Design scripts so repeated steps are acceptable, or add application-level
+idempotency around external side effects. `CAO_WORKFLOW_RESUME=1` is present
+in the resumed subprocess environment for code that must distinguish the
+drive, but the `cao_workflow` shim itself does not branch on it.
+
+## The `reuse_terminal_id` 422 trap
+
+If you pass `reuse_terminal_id` through `run_step(..., reuse_terminal_id=...)`,
+you'll get a `ShimError` immediately, before any network call:
+
+```
+ShimError: reuse_terminal_id is not supported by run_step() — the shim
+always sends env_vars (RUN_ID/GENERATION/STEP_ID), and the server rejects
+env_vars + reuse_terminal_id together (422). Omit reuse_terminal_id, or
+call the HTTP API directly if you need to reuse a terminal without the
+identity fence.
+```
+
+This isn't a shim bug or an arbitrary restriction: `run_step` **always**
+populates the identity `env_vars` fence (`CAO_WORKFLOW_RUN_ID`/
+`GENERATION`/`STEP_ID`), and the server's own request validator
+unconditionally rejects any request carrying both `env_vars` and
+`reuse_terminal_id` — the combination can never legitimately round-trip.
+Passing it through `**opts` would always produce an opaque 422; the shim
+fails fast instead so the mutual exclusivity is visible immediately. If you
+need terminal reuse without the identity fence, that's a case for calling
+`/terminals/run-step` directly over HTTP, outside the shim.
+
+## Examples
+
+Each example under [`docs/examples/`](examples/) demonstrates one of these
+patterns end-to-end, with a matching e2e test proving it runs:
+
+- [`loop_example.py`](examples/loop_example.py) — sequential loop, default
+  `step_id` counter.
+- [`conditional_example.py`](examples/conditional_example.py) — branching
+  control flow, explicit `step_id` per branch.
+- [`fanout_example.py`](examples/fanout_example.py) — concurrent fan-out via
+  `ThreadPoolExecutor`, explicit `step_id` per shard (the fan-out rule
+  above, applied).
+- [`loop_raw_http_example.py`](examples/loop_raw_http_example.py) — the
+  SAME loop shape with **no** `cao_workflow` import at all, using raw
+  `urllib` directly against the identity env vars `cao workflow run`
+  injects. Proves the shim is a convenience, not a requirement — a script
+  is free to skip it entirely and talk to `/terminals/run-step` on its own.

pyproject.toml

@@ -75,7 +75,7 @@ requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
-packages = ["src/cli_agent_orchestrator"]
+packages = ["src/cli_agent_orchestrator", "src/cao_workflow"]
 
 [tool.hatch.build]
 artifacts = [