Add a GA4GH WES execution engine to `planemo run` (`--engine wes`)

> **Posted by Claude (AI assistant) on behalf of @jmchilton.** Drafted by an AI assistant from a source-level review of `planemo` and the Galaxy WES API; not personally authored by @jmchilton. Treat as a proposal/RFC to react to.

## Summary

Add a new Planemo execution engine, selected with `planemo run --engine wes`, that runs a workflow by POSTing it to a [GA4GH Workflow Execution Service (WES)](https://ga4gh.github.io/workflow-execution-service-schemas/) endpoint (`/ga4gh/wes/v1/runs`), polling the run, and returning outputs — mirroring how `external_galaxy` runs against a remote Galaxy, but over the WES wire protocol instead of bioblend.

The immediate target is Galaxy's own WES API ([galaxyproject/galaxy #21335](https://github.com/galaxyproject/galaxy/pull/21335)), but the engine is intended to stay WES-generic where practical.

Two realities (verified live against Galaxy's WES) shape the design:

1. **WES has no data-staging endpoint.** Data inputs must already exist in the backend (for Galaxy, as HDAs referenced `{"src":"hda","id":...}` in `workflow_params`). Planemo's job-file model assumes local `class: File` inputs it uploads. So the engine needs a staging strategy — and Planemo already has the abstraction for it (`galaxy.tool_util.client.staging` / `galactic_job_json`, used by the Galaxy engines via `stage_in`); the WES engine should reuse it, not reinvent it.
2. **Run success comes from the WES run `state`, not per-job exit codes.** For Galaxy, a finished invocation reports WES `COMPLETE`; a failed *job* does **not** fail the *run* (a failed job can be a normal, valid path, e.g. filter-failed) — only an invocation that fails to schedule maps to `EXECUTOR_ERROR`. (Galaxy's `completed → COMPLETE` state mapping was just fixed; previously a finished run reported `UNKNOWN`.)

## Background: how Planemo engines work

- `planemo/commands/cmd_run.py` auto-picks an engine (CWL→`cwltool`, Galaxy+`--galaxy_url`→`external_galaxy`, else `galaxy`), then `engine.run([runnable],[job_path])[0]` and checks `was_successful`. **Note:** `--download_outputs` is registered but the run body does not actually download for any engine — there is no existing download hook to lean on.
- `planemo/engine/interface.py` — `Engine`/`BaseEngine`; a concrete engine sets `handled_runnable_types` and implements `_run(self, runnables, job_path, output_collectors=None)` + `cleanup()`. Public `run(...)` fans out.
- `planemo/engine/factory.py` — `build_engine` is an `if/elif` over the engine string (`galaxy`, `docker_galaxy`, `external_galaxy`, `cwltool`, `toil`).
- `planemo/engine/galaxy.py` → `planemo/galaxy/activity.py::execute` calls `stage_in` → `galaxy.tool_util.client.staging` → `galactic_job_json` — **the reusable job-file parser + stager**: it loads the job file, separates `class: File` inputs from scalar params, uploads files, and returns a `job_dict` rewritten to `{src,id}` plus a `history_id`.
- `planemo/runnable.py` — `RunResponse` ABC; `SuccessfulRunResponse` (**`was_successful` hardcoded `True`**); `ErrorRunResponse`. `CwlToolRunResponse` is the minimal success template; `run_cwltool` returning `ErrorRunResponse` on failure is the failure template.
- `planemo/options.py` — `run_engine_option()` (the `click.Choice`), the galaxy url/key options, the `engine_options()` decorator (already wires `--galaxy_url`, keys, `history_name`/`history_id`), `no_wait_option()`.
- `responses>=0.23.0` is already in `dev-requirements.txt`.

## Phase 0 (Galaxy side, done): `completed → COMPLETE` state mapping

Galaxy's `GALAXY_TO_WES_STATE` lacked the terminal `completed` state, so successful runs reported WES `UNKNOWN`. Fixed (`completed→COMPLETE`, `requires_materialization→INITIALIZING`) with a red→green API test. ⇒ WES `state` is now a reliable completion signal for clients.

## Phase 1 (MVP): submit + monitor over WES, no File staging

Scope: run a local Galaxy workflow whose inputs are parameters only (or already `{src,id}` references). Prove the wire protocol.

**Options (`planemo/options.py`):** add `"wes"` to the engine `click.Choice`; add `--wes_url`, `--wes_key` (`x-api-key` for Galaxy; `--wes_auth_scheme api_key|bearer` for generic servers), and `--wes_engine_parameters` (passthrough to `workflow_engine_parameters` for the long tail like `preferred_object_store_id`/`use_cached_job`; reuse the existing `history_id`/`history_name` options and fold them in). All copy the galaxy-option `use_global_config`/`extra_global_config_vars` pattern and must be added to `engine_options()`.

**Engine (`planemo/engine/wes.py`):** `WesEngine(BaseEngine)` with `handled_runnable_types = [RunnableType.galaxy_workflow]`. `_run` maps over `(runnable, job_path)`. Per run:
1. Read workflow body from `runnable.path`; choose `workflow_type` by sniffing content (`class: GalaxyWorkflow`→`gx_workflow_format2`; top-level `steps`/`workflow`→`gx_workflow_ga`), not just extension (Galaxy re-validates and 400s on mismatch).
2. Parse job inputs reusing `galactic_job_json`'s File-vs-param classification. MVP: pass scalars through; if any input is `class: File`/`Directory` and no staging is configured, **fail fast** with an `ErrorRunResponse` ("pre-stage and reference by id, or use `--engine external_galaxy`").
3. `POST multipart/form-data` to `/ga4gh/wes/v1/runs` (`workflow_type`, `workflow_type_version`, `workflow_attachment`, `workflow_params` JSON, `workflow_engine_parameters` JSON). On non-2xx, parse the JSON `err_msg` into an `ErrorRunResponse` — don't leak a raw `requests` exception.
4. Poll `/runs/{run_id}/status` with bounded backoff + overall timeout, honoring `--no_wait`. Terminal: `COMPLETE` (success); `EXECUTOR_ERROR`/`SYSTEM_ERROR`/`CANCELED` (failure). `UNKNOWN`/`QUEUED`/`INITIALIZING`/`RUNNING`/`CANCELING` are non-terminal. Success is derived from run `state` only, never from per-task `exit_code`.
5. `GET /runs/{run_id}` → `WesRunResponse` (success) or `ErrorRunResponse` (failure).

**RunResponses:** mirror `run_cwltool` — `WesRunResponse(SuccessfulRunResponse)` for success only (don't try to flip `was_successful`; it's hardcoded `True`), and `ErrorRunResponse` for every failure path (submit 4xx, terminal error/cancel, poll timeout). `WesRunResponse.outputs_dict` = the run-log `outputs` map; `log` lazily fetches task stdout/stderr (which resolve to Galaxy `/api/jobs/{id}/stdout|stderr` and need the same auth).

**Factory:** add `elif engine_type_str == "wes": engine_type = WesEngine`.

**Tests (red→green):**
- Unit (`tests/test_wes_engine.py`): with `responses` (or a `FakeWesServer` à la `tests/fake_trs.py`), assert the multipart submit payload + auth header, success polling → `WesRunResponse.outputs_dict`, and that a terminal `EXECUTOR_ERROR` and a submit-4xx both yield `ErrorRunResponse`.
- `can_run` (`tests/test_engines.py`): `wes` handles `galaxy_workflow`, not tools/CWL (MVP).
- Integration (`tests/test_cmds_wes.py`, mirrors `tests/test_cmds_with_workflow_id.py`): start a `--daemon` Galaxy serving the workflow's tool via `--extra_tools`, mint an API key, pre-stage one HDA, then `planemo run --engine wes --wes_url ... --wes_key ... wf.gxwf.yml job.yml`. Caveats: a workflow `data` input is **not** "params-only" — it needs a pre-staged HDA referenced by id; the fixture tool id must match the served tool (e.g. `tests/data/wf1.gxwf.yml` uses `tool_id: cat`).

## Phase 2: input staging + output download

- **Galaxy-specific path (recommended; the only thing that runs File-input workflows against Galaxy WES today):** when `--galaxy_url`/`--galaxy_user_key` are supplied, **reuse `stage_in`** (via a bioblend `GalaxyInstance`) to upload `class: File` inputs and obtain the `{src,id}` `job_dict` + `history_id`, then submit via WES. Do **not** write a parallel rewriter. For `--download_outputs` (no existing hook), implement explicitly — Galaxy API fetch handles HDA + HDCA; DRS handles HDA only (HDCA outputs carry no `drs_uri`).
- **Generic path (future):** if `service-info.supported_filesystem_protocols` allows, upload File inputs to an accessible location (`s3`/`gs`/http) and reference by URL; download via DRS only.

## Out of scope (initially)

CWL/Nextflow/WDL via Galaxy WES (only `gx_workflow_ga`/`gx_workflow_format2`); `planemo test --engine wes`; tool (non-workflow) execution; batch invocations (Galaxy WES rejects them); `--download_outputs` in the MVP.

## Open questions

1. **Galaxy-specific or generic-GA4GH engine?** Forks the staging/output design. Recommend Galaxy-specific MVP.
2. **Auth surface** — separate `--wes_url`/`--wes_key` (recommended) plus optional `--galaxy_*` for staging/download, vs reuse only `--galaxy_*`?
3. **Auto-select the engine when `--wes_url` is set** (like `--galaxy_url`→`external_galaxy`)?
4. **`--download_outputs`** — Galaxy API (complete; handles HDCA) vs DRS (standards-pure; HDA only)? Must be implemented from scratch.
5. **`workflow_type_version`** — hardcode `"1.0.0"` or expose? Galaxy treats it as free-form.

Start with planemo/engine/interface.py, planemo/engine/factory.py, planemo/options.py, and planemo/engine/galaxy.py to understand engine registration, options, and staging boundaries. Review the proposed tests in tests/test_wes_engine.py, tests/test_engines.py, and tests/test_cmds_wes.py before resolving the open design questions. Done means the agreed MVP submits and polls WES workflows with the specified success and failure responses, with passing unit and integration tests.

Written by the indexing model from the issue text.