feat: dispatch each task via spawn task run (spawn#386 adapter migration)

[scttfrdmn]

Summary

Ports spawn-airflow to the spawn task-execution protocol (spawn#386), following cwl-spawn and miniwdl-spawn. The operator stops orchestrating the launch itself and instead builds a TaskSpec and shells out to spawn task run.

Before → after

• Before: execute built a user-data script, called spawn launch … --user-data-file … --on-complete terminate, and polled a durable .exitcode object in S3 (sync loop or SpawnExitCodeTrigger).
• After: build a TaskSpec → spawn task run (detached) → poll spawn task status --check-complete (sync) or defer to SpawnTaskStatusTrigger → read the CompletionRecord exit code.

spawn now owns: instance sizing (truffle), S3 staging, the durable completion record, and a scoped least-privilege IAM profile (was --iam-policy s3:FullAccess).

Trigger rename (breaking for deferrable users)

SpawnExitCodeTrigger → SpawnTaskStatusTrigger: it polls spawn task status instead of the raw .exitcode S3 object. Serialized kwargs change from {s3_prefix, region, poll_interval} to {task_id, region, poll_interval}.

Notes

• Job dir /mnt/spawn_airflow_job → /var/tmp/spawn_airflow_job (user-writable — spawn runs the command as the unprivileged login user, which can't mkdir under root-owned /mnt; same class of fix as cwl-spawn#2 / the miniwdl port).
• No input manifest — an Airflow task carries no pre-staged input tree; the command fetches its own inputs (mirroring EcsRunTaskOperator). The TaskSpec has only an output manifest syncing the job dir back to workdir_s3.
• instance_type now steers the family (e.g. c7i.4xlarge → c7i), not an exact pin (spawn#413 follow-up). on_kill terminates by task_id (captured mid-flight).

Removed

launch.py, completion.py, bootstrap.py, sizing.py + test_helpers.py (spawn owns these). New pure taskspec.py.

Testing

• pytest 22 pass (new test_taskspec.py; reworked test_operator.py asserts detached dispatch, status-poll, deferral to SpawnTaskStatusTrigger, terminate-by-task-id; test_drift_guard.py unchanged). ruff + mypy clean.
• Gated real-AWS smoke is a follow-up (needs a TTL + approval per cost-safety): run examples/example_dag.py's hello task end-to-end and leak-check. Not shipping the release until that passes.

Refs spore-host/spawn#386