Author Npa Workflow

---
name: author-npa-workflow
description: Use when authoring, validating, or reviewing NPA workflow specs (apiVersion npa.workflow/v0.0.1) — declarative state machines that invoke workbench tools via SkyPilot.
---

# Author NPA Workflow

## When To Use

Load when creating or editing **NPA workflow YAML** under
`workflows/`, wiring tool stages, loops, or
transitions, or when helping agents/users convert SkyPilot bash pipelines into
specs.

Keep `workflows/main/` limited to `sim2real.yaml` and `paidf-cosmos3.yaml`.
Add all other catalog specs under `workflows/testing/`; keep catalog
documentation in `workflows/README.md`.

For **new creative pipelines**, also load `skills/workflows/generate-npa-workflow/SKILL.md`.

## Spec Contract

- **Guide:** `docs/workbench/npa-workflow-guide.md` (canonical examples + verify commands).
- **Catalog:** `docs/workbench/npa-workflow-tool-catalog.md` +
  `npa/src/npa/orchestration/npa_workflow/catalog.py`.
- **apiVersion:** `npa.workflow/v0.0.1` only (beta).
- **kind:** `Workflow`
- **States:** declarative nodes with `run` (shell/argv), `toolRef`, `sequence`, or `parallel`.
- **Tokens:** `{{config.key}}`, `{{run.id}}`, `{{run.prefix}}`, `{{state.NAME.uri}}` — no Jinja, no eval.
- **Predicates:** closed set: `promote_checkpoint`, `loop_back`.
- **Loops:** `loop.max: "{{config.attr}}"` or integer; `loop.until` for dynamic exit.
- **Parallel:** `parallel: [<leaf members>]` + optional `maxConcurrency`; the group's
  `next` state is the barrier. Members may not declare `next`/`transitions`.
  Use optional `parallelCount: "{{config.count}}"` when a public config count
  must exactly equal the explicit member list; override validation then fails
  before plan, rendering, or submission instead of silently under-fanning out.
- **Params:** `params: {k: v}` is a per-state config overlay — how sweep members share
  one `toolRef` and still differ.
- **Trigger:** `trigger: {uri, pollSeconds, maxPolls, minObjects}` makes the runtime
  driver wait for data before that state runs (the state must also do work).
  Numeric config expressions are resolved again after `--var` overrides;
  `pollSeconds` and `minObjects` must be positive, while `maxPolls: 0` retains
  the supported unbounded setting. Exercise the submitted configuration through
  the actual watcher when validating overrides: check observed object counts
  and polling behavior as well as successful spec parsing.
- **Decision states:** `writesDecision: true` when the state writes `config.decision_uri`.
- **needs:** ordering hints only (validated acyclic; not enforced at runtime).
- **I/O:** `inputs` / `outputs` with `uri` + optional `schema`.

## Validation Hardening (v0.0.1)

| Check | When |
| --- | --- |
| Unknown `toolRef` / predicate | `validate-spec` |
| Unbounded transition cycles | `validate-spec` (loops do **not** whitelist cycles) |
| Missing `{{config.*}}`, bad loop max | `validate-spec` via token resolution |
| Forward `{{state.*}}` refs | Allowed at validate; resolved during plan/execute |
| Execution depth | Guarded at `--execute` (no stack blowups) |
| `run.shell` | Resolves config tokens; spec authors are trusted (injection risk if config is untrusted) |

## Commands

### Validation and execution readiness

For planning-only work, preserve the requested operation in the workflow without
executing it. A restriction on provider calls during authoring does not require
replacing a planned inference step with a fixed answer. Use mocks only when the
user requests a mock or fixture, and label them as such.

Inspect the resolved plan against the task: do its tools or commands consume the
intended input, perform the requested operation, and produce the stated output?
Valid syntax alone cannot establish this. Report a mismatch as incomplete work,
not success, even when validation and planning commands pass.

Preserve failures in authoring and validation commands as well as workflow steps.
Run checks separately, or stop the command group at the first failure and return
that exit code. A later successful command must not hide an earlier failure.
If a search tool is unavailable, report its failure before using an available
alternative in a separate command.

Check input content with the consuming tool's local reader when available, not
just file existence or YAML validation. For a step that creates its own input,
check the generated format and prompt content without executing inference.
Keep an unchecked input unverified in the readiness record; do not claim task
fidelity from the workflow hash or planned command alone.

When saving a workflow, read [the readiness template](references/readiness-record.md)
and save `<workflow-stem>.readiness.json` beside it, within the authorized write
scope. Bind the record to the final workflow bytes. Record planning results and
each execution prerequisite separately; unverified prerequisites do not make a
correct planning-only task fail. Keep out-of-scope checks unverified.
For read-only reviews, report the same fields inline without creating files.
The final response can link the record and summarize the unresolved prerequisites.
Completeness does not establish execution readiness; verified claims need evidence.

An example bucket is a placeholder, not a writable destination. A local input
on the authoring machine is not automatically available to a remote worker;
identify its explicit staging, mount, or worker-readable URI. For a planning-only
task, list those unresolved prerequisites without creating storage, transferring
inputs, or submitting the workflow. Use
`skills/atomic/submit-workflow/SKILL.md` for the existing live-submit checks
when execution is requested.

### CLI commands

```bash
npa/.venv/bin/npa workbench workflow validate-spec <spec.yaml> --json
npa/.venv/bin/npa workbench workflow plan-spec <spec.yaml> --run-id demo --json
npa/.venv/bin/npa workbench workflow run-spec <spec.yaml> --plan-only --scheduler-plan --json
npa/.venv/bin/npa workbench workflow submit <spec.yaml> --run-id demo --plan-only
npa/.venv/bin/npa workbench workflow submit <spec.yaml> --run-id demo
```

`submit` accepts both `npa.workflow/v0.0.1` specs and legacy SkyPilot YAMLs.
For npa.workflow specs it plans → renders serial SkyPilot YAML → `sky jobs launch`.
Use `--plan-only` to inspect the rendered YAML without launching. Dynamic
branches still need `--assume-decision`.

Live infra (required before merge):

```bash
NPA_INTEGRATION_E2E=1 npa/.venv/bin/python -m pytest \
  npa/tests/e2e/test_npa_workflow_live_e2e.py \
  npa/tests/e2e/test_npa_workflow_live_infra.py -q
```

Live **submit** matrix (operator VM with SkyPilot + registry; burns GPU-hours):

```bash
# CPU-only first
NPA_E2E_NPA_WORKFLOW_SUBMIT_TIERS=cpu ./scripts/npa-workflow-submit-live-e2e.sh

# Full cpu+gpu+multi
./scripts/npa-workflow-submit-live-e2e.sh
```

Tmux full matrix (all npa.workflow YAMLs, real S3, credential leak checks):

```bash
./scripts/npa-workflow-real-infra-tmux.sh
```

## Authoring Rules

1. One workflow file = one variant; do not add sim2real-specific Python orchestrators.
2. Keep **terminal: true** on leaf completion states.
3. Use `--assume-decision` when planning specs with `transitions`.
4. `npa workbench workflow submit <npa.workflow.yaml>` plans the graph, renders
   a serial SkyPilot multi-doc YAML, and submits it (one-shot; unchanged).
   For a spec with a `parallel:` group, or a loop that must **early-exit on the
   real decision artifact**, add `--runtime`: the runtime orchestrator submits
   each wave, polls it to a terminal state, reads `config.decision_uri` from S3,
   and replans (parallel groups become SkyPilot JobGroups). Preview the wave
   shape offline with `plan-spec --waves`. See repo-root `DESIGN.md`.
5. Cross-stage data uses S3 URIs in `config` — tools are stateless.
6. Group config: runtime knobs first, then `*_uri` keys under `config.prefix`.
7. A `run.shell` state may import an npa module for logic that needs the package
   (npa is pip-installed in the task): `python3 -c "from npa.workflows.<mod>
   import <fn>; <fn>('{{config.a}}', '{{config.b}}')"` — prefer this over inlining
   heavy logic. Put testable logic in a real module (e.g.
   `npa/src/npa/workflows/data_factory_viz.py`) with a unit test.
8. A `toolRef` argv template must match the tool's **actual CLI option names**
   (e.g. `--input-uri`/`--output-uri`, not `--input-path`) and include required
   flags like `--run-id`; a mismatch validates/plans fine but crashes on real
   submit. Keep `catalog.py` and `docs/workbench/npa-workflow-tool-catalog.md` in
   sync.
9. **A blueprint never becomes a CLI command.** The spec is the unit of
   composition: `npa workbench workflow submit <spec.yaml>` runs any of them, and
   anything a blueprint needs on the way — provisioning, source staging,
   prerequisite reporting — belongs in the spec (`deployIfAbsent`) or in the
   generic submit flags (`--stage-src`, `--var`), never in a per-blueprint verb
   such as `npa paidf up`. One command per YAML does not scale: every new
   blueprint would add CLI surface, help text, docs and tests for behavior the
   generic path already has. Natural-language shortcuts belong in the chat
   agent's spec alias table (`npa/src/npa/cli/agent_workflow.py`), which maps a
   phrase to a spec name rather than adding a command.
10. **Live-infra is a priority** (`skills/atomic/testing-conventions/SKILL.md`): a
   new spec must be registered in `SUBMIT_LIVE_MATRIX`
   (`npa/src/npa/orchestration/npa_workflow/submit_matrix.py`); if it has a
   dynamic gate/loop, also add it to `DYNAMIC_SPECS` in
   `npa/tests/e2e/npa_workflow_live_helpers.py`. Use `plan_only=True` for
   stubs, placeholder/reference components, or a separately covered onboarding
   flow. An arbitrary execution gap is not a plan-only exemption: fix it or
   make the real path fail closed and exercise it. Don't stop at smoke.

## Reference Examples

| Spec | Purpose |
| --- | --- |
| `vlm-eval-single.yaml` | Single-tool minimal |
| `tokenfactory-rollout-judge.yaml` | Serial two-tool |
| `sim2real.yaml` | Canonical compositional 14-stage Sim2Real runtime; real component boundaries and nested durable loops |
| `bdd100k-pipeline.yaml` | AV failure-mode LanceDB → train → eval |
| `av-night-scene-hardening.yaml` | AV night-scene fan-out — two per-view detector train→eval branches |
| `cosmos-synth-fanout-curation.yaml` | Cosmos Transfer 2.5 synthetic fan-out → Voxel51 (FiftyOne) curation |
| `tokenfactory-cosmos-gate.yaml` | Creative reason → augment → VLM gate loop |
| `physical-ai-data-factory.yaml` | NVIDIA Physical AI Data Factory (no OSMO): annotate → Cosmos augment → evaluate/validate gate → re-label → FiftyOne curate → Rerun visualize; toolRefs + `run.shell` glue; dynamic gate |
| `token-factory-parallel-fanout.yaml` | Zero-GPU `parallel:` fan-out (JobGroup) + join barrier (`--runtime`) |
| `token-factory-gate-loop.yaml` | Zero-GPU runtime gate loop: real early-exit + `goto` branch (`--runtime`) |
| `isaac-lab-rl-sweep.yaml` | Parallel GPU sweep with `params:` overlays + ranking barrier (`--runtime`) |

## Verify

```bash
npa/.venv/bin/python -m pytest npa/tests/orchestration/npa_workflow/ \
  npa/tests/smoke/test_npa_workflow_smoke.py \
  npa/tests/smoke/test_all_workflow_yamls.py -q --tb=no
```

Tmux matrix: `./scripts/npa-workflow-creative-tmux.sh`