diff --git a/.airstack/modules/osmo.sh b/.airstack/modules/osmo.sh
index 053decbee..66ac29e02 100755
--- a/.airstack/modules/osmo.sh
+++ b/.airstack/modules/osmo.sh
@@ -120,19 +120,47 @@ function _osmo_prompt {
     fi
 }
 
-# osmo:setup — interactively register the three OSMO credentials AirStack
-# needs (airlab-docker-registry, airlab-docker-login, airlab-nucleus).
-# Idempotent — re-running rotates the credentials.
+# Helper: _osmo_prompt variant that falls back to $3 on empty input.
+function _osmo_prompt_default {
+    local var_name="$1"
+    local prompt_text="$2"
+    local default_val="$3"
+    local saved_stty=""
+
+    if [ -t 0 ]; then
+        saved_stty="$(stty -g 2>/dev/null || true)"
+        if [ -n "$saved_stty" ]; then
+            trap 'stty "$saved_stty" 2>/dev/null; trap - INT' INT
+            stty -icanon 2>/dev/null
+        fi
+    fi
+    read -r -p "${prompt_text} [${default_val}]: " "$var_name"
+    if [ -n "$saved_stty" ]; then
+        stty "$saved_stty" 2>/dev/null
+        trap - INT
+    fi
+
+    _osmo_trim "$var_name"
+    if [ -z "${!var_name}" ]; then
+        printf -v "$var_name" '%s' "$default_val"
+    fi
+}
+
+# osmo:setup — interactively register the OSMO credentials AirStack needs
+# (airlab-docker-registry, airlab-docker-login, airlab-nucleus, optional
+# airlab-storage). Idempotent — re-running rotates the credentials.
 function cmd_osmo_setup {
     _osmo_check_cli || return 1
 
     cat >&2 <<'EOF'
 
-This sets up the three per-user OSMO credentials AirStack-on-OSMO needs:
+This sets up the per-user OSMO credentials AirStack-on-OSMO needs:
 
   1. airlab-docker-registry  (REGISTRY) — for OSMO to pull the workspace image
   2. airlab-docker-login     (GENERIC)  — for the inner dockerd to pull AirStack images
   3. airlab-nucleus          (GENERIC)  — for Isaac Sim Nucleus access
+  4. airlab-storage          (GENERIC)  — OPTIONAL: push mission results to the
+                                          airlab-storage NAS + auto-teardown
 
 You'll be asked for:
 
@@ -140,6 +168,9 @@ You'll be asked for:
   - your AirLab Docker password (same as your Andrew password)
   - your Nucleus API token (https://airlab-nucleus.andrew.cmu.edu/omni/web3/
     → right-click cloud → API Tokens). NOT your Andrew password.
+  - (optional) your airlab-storage password — the SEPARATE Samba/SFTP/rsync
+    password set at https://airlab-storage.andrew.cmu.edu:4001/#/forgot-pwd,
+    NOT your Andrew/Docker password.
 
 Values go directly to OSMO; nothing is written to disk locally.
 
@@ -202,7 +233,33 @@ EOF
                   "omni_server=${omni_server}" \
         || { log_error "osmo credential set airlab-nucleus failed"; return 1; }
 
-    log_info "All three credentials registered. List them with: osmo credential list"
+    # Optional: airlab-storage NAS upload. Only the password is a secret; the
+    # destination path is set per-mission (nas_dest: in the mission spec).
+    local want_storage=""
+    read -r -p "Configure airlab-storage NAS upload (optional)? [y/N]: " want_storage
+    case "$want_storage" in
+        y|Y|yes|Yes)
+            local storage_user storage_pass storage_host
+            _osmo_prompt_default storage_user "airlab-storage username (Andrew ID)" "$andrew_id"
+            _osmo_prompt         storage_pass "airlab-storage password (hidden; the SEPARATE Samba/SFTP/rsync password)" true || return 1
+            _osmo_prompt_default storage_host "airlab-storage host" "airlab-storage.andrew.cmu.edu"
+
+            log_info "Refreshing airlab-storage (GENERIC)..."
+            osmo credential delete airlab-storage >/dev/null 2>&1 || true
+            osmo credential set airlab-storage \
+                --type GENERIC \
+                --payload "username=${storage_user}" \
+                          "password=${storage_pass}" \
+                          "host=${storage_host}" \
+                || { log_error "osmo credential set airlab-storage failed"; return 1; }
+            log_info "airlab-storage registered. Set 'nas_dest: /volume<X>/<share>/...' in a mission spec to upload results + auto-teardown."
+            ;;
+        *)
+            log_info "Skipping airlab-storage — NAS upload stays disabled (missions keep-alive as before)."
+            ;;
+    esac
+
+    log_info "Credentials registered. List them with: osmo credential list"
     log_info "Next: airstack osmo:up [--pool POOL]"
 }
 
@@ -686,6 +743,360 @@ function cmd_osmo_foxglove {
         --connect-timeout "$OSMO_PF_TIMEOUT"
 }
 
+# Parse a duration (90s / 30m / 8h / bare number = minutes) to seconds.
+function _osmo_parse_duration {
+    local v="$1"
+    if [[ "$v" =~ ^([0-9]+)([smh]?)$ ]]; then
+        local n="${BASH_REMATCH[1]}" u="${BASH_REMATCH[2]}"
+        case "$u" in
+            s) echo "$n" ;;
+            h) echo $(( n * 3600 )) ;;
+            *) echo $(( n * 60 )) ;;   # m or bare → minutes
+        esac
+    else
+        log_error "invalid duration '$v' — use e.g. 8h, 480m, 3600s, or a bare number (minutes)"
+        return 1
+    fi
+}
+
+function _osmo_hms {
+    printf '%dh%02dm' $(( $1 / 3600 )) $(( ($1 % 3600) / 60 ))
+}
+
+# Block until the mission finishes (the workspace log prints "mission_runner
+# exited") or cap_s elapses, then run osmo:fetch. Requires keep-alive so the
+# pod is still up to fetch from.
+function _osmo_wait_and_fetch {
+    local wf="$1" cap_s="$2"
+    local dest="${OSMO_AUTOFETCH_DEST:-./osmo-results}"
+    local poll="${OSMO_AUTOFETCH_POLL_S:-120}"
+    local start now elapsed snapshot status
+    start="$(date +%s)"
+    log_info "auto-fetch armed: polling every ${poll}s for completion (cap $(_osmo_hms "$cap_s")) → ${dest}"
+    log_info "  keep this terminal open (or background with nohup/tmux) until the fetch runs."
+    while :; do
+        now="$(date +%s)"; elapsed=$(( now - start ))
+        status="$(osmo workflow query "$wf" 2>/dev/null | awk -F': +' '/^Status/ {print $2; exit}' | tr -d ' \r\n' || true)"
+        case "$status" in
+            PENDING|RUNNING|"") ;;
+            *) log_warn "workflow ${wf} is ${status} — ending wait, attempting fetch."; break ;;
+        esac
+        snapshot="$(timeout 25 osmo workflow logs "$wf" -t workspace -n 600 2>/dev/null || true)"
+        if printf '%s\n' "$snapshot" | grep -q "mission_runner exited"; then
+            log_info "mission complete (after $(_osmo_hms "$elapsed")) — fetching."
+            break
+        fi
+        if [ "$elapsed" -ge "$cap_s" ]; then
+            log_warn "auto-fetch cap reached ($(_osmo_hms "$cap_s")) without completion — fetching what's there."
+            break
+        fi
+        sleep "$poll"
+    done
+    AIRSTACK_OSMO_WF="$wf" cmd_osmo_fetch "$dest"
+}
+
+# osmo:autofetch — attach the auto-fetch poller to an already-running mission
+# (one submitted without --auto-fetch, or whose terminal was closed). Polls the
+# workspace log for completion, then runs osmo:fetch.
+#
+# Usage: airstack osmo:autofetch [DUR]   (DUR = max-wait cap, default 8h; uses
+#        the saved workflow id, or AIRSTACK_OSMO_WF to target a specific one.)
+function cmd_osmo_autofetch {
+    _osmo_check_cli || return 1
+    local wf; wf="$(_osmo_wf_id)" || return 1
+    local cap_s
+    cap_s="$(_osmo_parse_duration "${1:-8h}")" || return 1
+    _osmo_wait_and_fetch "$wf" "$cap_s"
+}
+
+# osmo:mission — submit airstack-mission.yaml with a mission spec selected.
+#
+# Usage: airstack osmo:mission <mission.yaml> [--pool POOL] [--key PATH]
+#                              [--branch BRANCH] [--no-keep-alive]
+#                              [--auto-fetch DUR] [--nas-dest PATH]
+#                              [--no-nas-upload]
+#
+# <mission.yaml> is a repo-relative path (e.g. osmo/missions/example_takeoff_land.yaml).
+# The pod clones the branch and runs the mission spec from that clone, so the
+# mission file must be committed and pushed. --no-keep-alive makes the task
+# exit when the mission ends (frees the GPU, triggers the workflow's
+# `outputs:` upload) instead of sleeping for `airstack osmo:fetch`.
+#
+# --auto-fetch DUR blocks after submit, polling the workspace log for mission
+# completion and running osmo:fetch as soon as it's done (DUR — e.g. 8h, 480m —
+# is the max wait before fetching anyway). Requires keep-alive (the default).
+#
+# If the airlab-storage credential is set (airstack osmo:setup) and the mission
+# spec has `nas_dest:` (or --nas-dest PATH overrides it), the pod rsyncs results
+# to the NAS then tears itself down. --no-nas-upload suppresses that.
+function cmd_osmo_mission {
+    _osmo_check_cli || return 1
+
+    local mission=""
+    local pool="${OSMO_POOL:-}"
+    local pubkey_file=""
+    local branch=""
+    local branch_explicit=false
+    local keep_alive="true"
+    local auto_fetch_dur=""
+    local nas_dest=""
+    local no_nas_upload="false"
+    local extra_args=()
+
+    while [ $# -gt 0 ]; do
+        case "$1" in
+            --pool)          pool="$2"; shift 2 ;;
+            --key)           pubkey_file="$2"; shift 2 ;;
+            --branch)        branch="$2"; branch_explicit=true; shift 2 ;;
+            --no-keep-alive) keep_alive="false"; shift ;;
+            --auto-fetch)    auto_fetch_dur="$2"; shift 2 ;;
+            --nas-dest)      nas_dest="$2"; shift 2 ;;
+            --no-nas-upload) no_nas_upload="true"; shift ;;
+            -*)              extra_args+=("$1"); shift ;;
+            *)
+                if [ -z "$mission" ]; then mission="$1"; else extra_args+=("$1"); fi
+                shift ;;
+        esac
+    done
+
+    # Validate --auto-fetch up front (fail before submitting on a bad value).
+    local auto_fetch_cap_s=""
+    if [ -n "$auto_fetch_dur" ]; then
+        if [ "$keep_alive" != "true" ]; then
+            log_error "--auto-fetch requires the pod to stay alive; don't combine it with --no-keep-alive."
+            return 1
+        fi
+        auto_fetch_cap_s="$(_osmo_parse_duration "$auto_fetch_dur")" || return 1
+    fi
+
+    # NAS auto-upload tears the pod down on completion, so --auto-fetch (which
+    # needs the pod alive) is moot when it runs. Best-effort heads-up.
+    if [ -n "$auto_fetch_dur" ] && [ "$no_nas_upload" != "true" ] \
+       && osmo credential list 2>/dev/null | grep -q 'airlab-storage'; then
+        log_warn "airlab-storage is configured; a mission with nas_dest set uploads + tears down, making --auto-fetch moot."
+    fi
+
+    if [ -z "$mission" ]; then
+        log_error "Usage: airstack osmo:mission <mission.yaml> [--pool POOL] [--branch BRANCH] [--no-keep-alive]"
+        log_error "Available missions:"
+        ls "${PROJECT_ROOT}/osmo/missions/"*.yaml 2>/dev/null \
+            | sed "s|${PROJECT_ROOT}/|  |" >&2
+        return 1
+    fi
+    # Normalize to a repo-relative path — that's what the pod resolves
+    # against its clone of the branch.
+    mission="${mission#"${PROJECT_ROOT}"/}"
+    if [ ! -f "${PROJECT_ROOT}/${mission}" ]; then
+        log_error "Mission file not found locally: ${PROJECT_ROOT}/${mission}"
+        return 1
+    fi
+
+    if [ -z "$pubkey_file" ]; then
+        if ! pubkey_file="$(_osmo_pick_pubkey)"; then
+            log_error "No SSH public key found in ~/.ssh. Generate one with: ssh-keygen -t ed25519"
+            return 1
+        fi
+    fi
+
+    local workflow_yaml="${PROJECT_ROOT}/osmo/workflows/airstack-mission.yaml"
+    if [ ! -f "$workflow_yaml" ]; then
+        log_error "Workflow file not found: ${workflow_yaml}"
+        return 1
+    fi
+
+    # The pod runs the mission file from its clone of origin/<branch>, so an
+    # unpushed mission spec is the most common "why is it running the wrong
+    # thing" failure — same auto-pin + pushed check as osmo:up.
+    if [ "$branch_explicit" = false ] && [ -z "$branch" ]; then
+        branch="$(_osmo_local_branch)"
+        if [ -n "$branch" ]; then
+            log_info "Auto-detected local branch '${branch}'; pod will clone from origin/${branch} (override with --branch main)."
+        fi
+    fi
+    if [ -n "$branch" ]; then
+        _osmo_check_branch_pushed "$branch"
+    fi
+
+    local cmd=(osmo workflow submit "$workflow_yaml")
+    if [ -n "$pool" ]; then
+        cmd+=(--pool "$pool")
+    else
+        log_warn "No --pool provided and OSMO_POOL is unset; using your osmo profile's default pool."
+    fi
+    # Single --set-env: the flag is variadic and a second occurrence silently
+    # drops the first (see cmd_osmo_up).
+    local env_kvs=(
+        "SSH_PUB_KEY=$(cat "$pubkey_file")"
+        "OSMO_MISSION_FILE=${mission}"
+        "OSMO_MISSION_KEEP_ALIVE=${keep_alive}"
+    )
+    if [ -n "$branch" ]; then
+        env_kvs+=("AIRSTACK_BRANCH=${branch}")
+    fi
+    if [ -n "$nas_dest" ]; then
+        env_kvs+=("OSMO_MISSION_UPLOAD_DEST=${nas_dest}")
+    fi
+    if [ "$no_nas_upload" = "true" ]; then
+        env_kvs+=("OSMO_MISSION_NO_UPLOAD=true")
+    fi
+    cmd+=(--set-env "${env_kvs[@]}")
+    if [ ${#extra_args[@]} -gt 0 ]; then
+        cmd+=("${extra_args[@]}")
+    fi
+
+    log_info "Submitting mission '${mission}' (keep_alive=${keep_alive}): ${cmd[*]}"
+    local output
+    if ! output="$("${cmd[@]}" 2>&1)"; then
+        echo "$output" >&2
+        log_error "osmo workflow submit failed."
+        return 1
+    fi
+    echo "$output"
+
+    local wf_id
+    wf_id="$(echo "$output" | awk -F'- ' '/^Workflow ID/ {print $2; exit}' | tr -d ' \r\n')"
+    if [ -z "$wf_id" ]; then
+        log_warn "Could not parse workflow id from submit output. Set it manually:"
+        log_warn "  echo <wf-id> > ${OSMO_STATE_FILE}"
+        return 0
+    fi
+    _osmo_save_wf_id "$wf_id"
+
+    if [ -n "$auto_fetch_cap_s" ]; then
+        _osmo_wait_and_fetch "$wf_id" "$auto_fetch_cap_s"
+        return $?
+    fi
+
+    log_info "Next steps:"
+    log_info "  airstack osmo:logs        # follow mission progress"
+    log_info "  airstack osmo:fetch       # download bags + results (keep-alive mode)"
+    log_info "  airstack osmo:down        # cancel when done (results die with the pod!)"
+    if [ -n "$nas_dest" ] && [ "$no_nas_upload" != "true" ]; then
+        log_info "NAS upload armed → results go to ${nas_dest}/<name>/<stamp>/, then the pod tears itself down."
+    fi
+}
+
+# osmo:fetch — download mission results (mcap bags, logs, summaries) from the
+# pod to the laptop over the authenticated ssh port-forward.
+#
+# Usage: airstack osmo:fetch [dest-dir]
+#
+# Incremental and resumable (rsync): safe to run mid-mission to pull finished
+# iterations while the next one flies, and again later to top up. Falls back
+# to scp -r (non-incremental) if rsync isn't installed.
+#
+# Note: the osmo CLI also ships `osmo workflow rsync download`, which could
+# replace the port-forward + ssh below; we use the ssh path because the
+# sshd + port-forward channel is already validated infrastructure for this
+# workflow (osmo:ide) and works uniformly across osmo CLI versions.
+function cmd_osmo_fetch {
+    _osmo_check_cli || return 1
+    local wf; wf="$(_osmo_wf_id)" || return 1
+
+    local dest="${1:-./osmo-results}"
+    local remote_path="/root/AirStack/osmo/results/"   # trailing slash: rsync
+                                                       # follows the symlink to
+                                                       # /osmo/output/... on pods
+    local local_port="${OSMO_SSH_PORT%%:*}"
+    local ssh_opts=(-p "$local_port" -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o LogLevel=ERROR)
+
+    # Reuse an existing ssh port-forward (e.g. from osmo:ide) or spawn one
+    # for the duration of the fetch — same pattern as osmo:ide.
+    local pf_pid=""
+    if ! nc -z localhost "$local_port" 2>/dev/null; then
+        log_info "osmo workflow port-forward ${wf} workspace --port ${OSMO_SSH_PORT} (for the duration of the fetch)"
+        osmo workflow port-forward "$wf" workspace --port "$OSMO_SSH_PORT" --connect-timeout 600 \
+            > "${OSMO_STATE_DIR}/fetch-pf.log" 2>&1 &
+        pf_pid=$!
+        trap '[ -n "'"$pf_pid"'" ] && kill "'"$pf_pid"'" 2>/dev/null; trap - EXIT INT TERM' EXIT INT TERM
+        local waited=0
+        until nc -z localhost "$local_port" 2>/dev/null; do
+            sleep 1; waited=$((waited+1))
+            if [ "$waited" -ge 30 ]; then
+                log_error "Timed out waiting for port-forward on :${local_port}. Log: ${OSMO_STATE_DIR}/fetch-pf.log"
+                return 1
+            fi
+            if ! kill -0 "$pf_pid" 2>/dev/null; then
+                log_error "port-forward exited early. Tail:"
+                tail -10 "${OSMO_STATE_DIR}/fetch-pf.log" >&2
+                return 1
+            fi
+        done
+    fi
+
+    mkdir -p "$dest"
+    log_info "Fetching ${remote_path} → ${dest}"
+    local rc
+    if command -v rsync >/dev/null 2>&1; then
+        rsync -az --partial --info=progress2 -e "ssh ${ssh_opts[*]}" \
+            "root@localhost:${remote_path}" "$dest/"
+        rc=$?
+    else
+        log_warn "rsync not found; falling back to scp -r (non-incremental)."
+        scp "${ssh_opts[@]}" -r "root@localhost:${remote_path}." "$dest/"
+        rc=$?
+    fi
+
+    if [ -n "$pf_pid" ]; then
+        kill "$pf_pid" 2>/dev/null
+        trap - EXIT INT TERM
+    fi
+
+    if [ "$rc" -ne 0 ]; then
+        log_error "Fetch failed (exit ${rc}). Is the workflow still running, and has the mission produced results yet?"
+        return 1
+    fi
+    log_info "Done. Open any .mcap under ${dest} directly in Foxglove (Open local file)."
+}
+
+# osmo:stop — gracefully stop the running mission: SIGINT to mission_runner
+# on the pod → current step aborts, recorders finalize their mcaps, bags +
+# logs are collected, stack goes down. The pod itself stays alive
+# (keep-alive), so follow with `airstack osmo:fetch` to download results.
+function cmd_osmo_stop {
+    _osmo_check_cli || return 1
+    local wf; wf="$(_osmo_wf_id)" || return 1
+
+    local local_port="${OSMO_SSH_PORT%%:*}"
+    local ssh_opts=(-p "$local_port" -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o LogLevel=ERROR)
+
+    # Reuse an existing ssh port-forward or spawn one for this command —
+    # same pattern as osmo:fetch.
+    local pf_pid=""
+    if ! nc -z localhost "$local_port" 2>/dev/null; then
+        log_info "osmo workflow port-forward ${wf} workspace --port ${OSMO_SSH_PORT} (for the stop)"
+        osmo workflow port-forward "$wf" workspace --port "$OSMO_SSH_PORT" --connect-timeout 600 \
+            > "${OSMO_STATE_DIR}/stop-pf.log" 2>&1 &
+        pf_pid=$!
+        local waited=0
+        until nc -z localhost "$local_port" 2>/dev/null; do
+            sleep 1; waited=$((waited+1))
+            if [ "$waited" -ge 30 ]; then
+                log_error "Timed out waiting for port-forward on :${local_port}. Log: ${OSMO_STATE_DIR}/stop-pf.log"
+                [ -n "$pf_pid" ] && kill "$pf_pid" 2>/dev/null
+                return 1
+            fi
+        done
+    fi
+
+    log_info "Sending stop signal to mission_runner on the pod..."
+    ssh "${ssh_opts[@]}" root@localhost \
+        "pkill -INT -f 'osmo/workspace/mission_runner\.py' \
+         && echo 'stop signal delivered' \
+         || echo 'no running mission_runner found (mission already finished?)'"
+    local rc=$?
+
+    if [ -n "$pf_pid" ]; then
+        kill "$pf_pid" 2>/dev/null
+    fi
+    if [ "$rc" -ne 0 ]; then
+        log_error "ssh to pod failed (exit ${rc})."
+        return 1
+    fi
+    log_info "Mission stopping: current step aborts, recordings finalize, bags + logs are collected, stack goes down."
+    log_info "Watch progress: airstack osmo:logs    Download results: airstack osmo:fetch"
+}
+
 # osmo:down — cancel the active workflow. Reminds you to push first.
 function cmd_osmo_down {
     _osmo_check_cli || return 1
@@ -693,6 +1104,7 @@ function cmd_osmo_down {
 
     log_warn "About to cancel workflow '${wf}'."
     log_warn "Anything not pushed to git in /root/AirStack inside the pod will be LOST."
+    log_warn "Mission results (bags/logs) on the pod are lost too — run 'airstack osmo:fetch' first."
     log_warn "Hit Ctrl-C in the next 5 seconds to abort."
     sleep 5
     osmo workflow cancel "$wf"
@@ -703,14 +1115,22 @@ function cmd_osmo_down {
 function register_osmo_commands {
     COMMANDS["osmo:setup"]="cmd_osmo_setup"
     COMMANDS["osmo:up"]="cmd_osmo_up"
+    COMMANDS["osmo:mission"]="cmd_osmo_mission"
+    COMMANDS["osmo:fetch"]="cmd_osmo_fetch"
+    COMMANDS["osmo:autofetch"]="cmd_osmo_autofetch"
+    COMMANDS["osmo:stop"]="cmd_osmo_stop"
     COMMANDS["osmo:logs"]="cmd_osmo_logs"
     COMMANDS["osmo:ide"]="cmd_osmo_ide"
     COMMANDS["osmo:webrtc"]="cmd_osmo_webrtc"
     COMMANDS["osmo:foxglove"]="cmd_osmo_foxglove"
     COMMANDS["osmo:down"]="cmd_osmo_down"
 
-    COMMAND_HELP["osmo:setup"]="One-time per-user OSMO credential setup (airlab-docker-registry, airlab-docker-login, airlab-nucleus)"
+    COMMAND_HELP["osmo:setup"]="One-time per-user OSMO credential setup (airlab-docker-registry, airlab-docker-login, airlab-nucleus, optional airlab-storage for NAS upload)"
     COMMAND_HELP["osmo:up"]="Submit osmo/workflows/airstack-dev.yaml with your SSH pubkey injected (--pool POOL, --key PATH, --branch BRANCH)"
+    COMMAND_HELP["osmo:mission"]="Submit a batch mission (osmo/missions/*.yaml): repeated up→fly→record→down cycles (--pool POOL, --branch BRANCH, --no-keep-alive, --auto-fetch DUR, --nas-dest PATH, --no-nas-upload)"
+    COMMAND_HELP["osmo:fetch"]="Download mission results (mcap bags, logs, summaries) from the pod over ssh — incremental, safe to run mid-mission (osmo:fetch [dest-dir])"
+    COMMAND_HELP["osmo:autofetch"]="Attach the auto-fetch poller to the running mission (saved id or AIRSTACK_OSMO_WF): poll the workspace log until done, then osmo:fetch (osmo:autofetch [DUR], default 8h)"
+    COMMAND_HELP["osmo:stop"]="Gracefully stop the running mission: abort current step, finalize mcaps, collect bags+logs, stack down (pod stays up for osmo:fetch)"
     COMMAND_HELP["osmo:logs"]="Follow the workspace task logs (osmo workflow logs <id> -t workspace -n 500; OSMO_LOGS_TASK / OSMO_LOGS_TAIL override)"
     COMMAND_HELP["osmo:ide"]="Port-forward sshd (2200:22) and open VS Code/Cursor on Host airstack-osmo"
     COMMAND_HELP["osmo:webrtc"]="Port-forward Isaac Sim WebRTC ranges (TCP foreground + UDP background)"
diff --git a/.env b/.env
index 82cc01ccb..70138406b 100644
--- a/.env
+++ b/.env
@@ -12,7 +12,7 @@ PROJECT_NAME="airstack"
 # If you've run ./airstack.sh setup, then this will auto-generate from the git commit hash every time a change is made 
 # to a Dockerfile or docker-compose.yaml file. Otherwise this can also be set explicitly to make a release version.
 # auto-generated from git commit hash
-VERSION="0.19.0-alpha.3"
+VERSION="64594854"
 # Choose "dev" or "prebuilt". "dev" is for mounted code that must be built live. "prebuilt" is for built ros_ws baked into the image
 DOCKER_IMAGE_BUILD_MODE="dev"  
 # Where to push and pull images from. Can replace with your docker hub username if using docker hub.
diff --git a/.gitignore b/.gitignore
index 4868b5c74..1fb9d8592 100644
--- a/.gitignore
+++ b/.gitignore
@@ -94,6 +94,11 @@ simulation/ms-airsim/assets/scenes/*
 # Test results
 tests/results/
 
+# OSMO mission results: local runs of osmo/workspace/mission_runner.py write
+# to osmo/results/; `airstack osmo:fetch` downloads to ./osmo-results/.
+osmo/results/*
+osmo-results/*
+
 # Local-only — embedded sibling repo, not part of this branch
 common/rayfronts/
 
diff --git a/docs/tutorials/airstack_on_osmo.md b/docs/tutorials/airstack_on_osmo.md
index e9cfa8974..9f093e8c0 100644
--- a/docs/tutorials/airstack_on_osmo.md
+++ b/docs/tutorials/airstack_on_osmo.md
@@ -547,6 +547,45 @@ osmo workflow cancel $WF
 
 </details>
 
+## Batch missions (unattended runs)
+
+Everything above is the *interactive* workflow. The same pod can instead run
+**missions**: declarative YAML files (in
+[`osmo/missions/`](https://github.com/castacks/AirStack/blob/main/osmo/missions/))
+that script repeated cycles of bring-up → fly → record → tear-down with no
+human attached. Each iteration restarts the containers, records mcap bag
+files (Foxglove's native format — open the `.mcap` directly, no conversion),
+and snapshots container logs and per-step results.
+
+```bash
+# Submit a mission (auto-pins your current branch, like osmo:up):
+./airstack.sh osmo:mission osmo/missions/example_takeoff_land.yaml --pool airstack
+
+# Watch it fly:
+./airstack.sh osmo:logs
+
+# Pull bags + logs + summaries to your laptop — incremental, safe to run
+# mid-mission and again later to top up:
+./airstack.sh osmo:fetch ./results/
+
+# When you have everything (results die with the pod!):
+./airstack.sh osmo:down
+```
+
+A mission step can be any robot task action (`takeoff`, `land`, `navigate`,
+`semantic_search`, `exploration`, `coverage`, …), a timed wait, a topic pub,
+a service call, or an arbitrary `ros2`/shell command. Spec schema and step
+reference:
+[`osmo/missions/README.md`](https://github.com/castacks/AirStack/blob/main/osmo/missions/README.md).
+
+By default the pod **stays alive after the mission ends** so you can
+`osmo:fetch` whenever you're ready (mind the workflow's 24h `exec_timeout`).
+For fire-and-forget batches, submit with `--no-keep-alive`: the pod exits
+cleanly when the mission ends, freeing the GPU — and uploading the results
+directory to object storage automatically if the workflow's `outputs:`
+block is configured (lab-admin setup; see
+[`osmo/README.md`](https://github.com/castacks/AirStack/blob/main/osmo/README.md)).
+
 ## Troubleshooting
 
 | Symptom | Likely cause | Fix |
@@ -572,7 +611,8 @@ osmo workflow cancel $WF
 | Uncommitted edits in the IDE | Pod-local working tree | **No** |
 | `colcon build` outputs (`build/`, `install/`, `log/`) | `/root/AirStack/**/ros_ws/...` | **No** (gitignored Linux x86_64 binaries; rebuild trivially) |
 | Inner-dockerd image cache | Pod-local Docker layer cache | **No** |
-| Bag files, sim recordings, debug screenshots | `/root/AirStack/bags/`, etc. | **No** — pull selectively via `osmo workflow rsync download "$(cat ~/.airstack/osmo-state)" <pod-path>:<local-path>` *before* tearing down |
+| Mission results (mcap bags, logs, summaries) | `/root/AirStack/osmo/results/` | **No** — run `./airstack.sh osmo:fetch` *before* tearing down |
+| Other bag files, sim recordings, debug screenshots | `/root/AirStack/bags/`, etc. | **No** — pull selectively via `osmo workflow rsync download "$(cat ~/.airstack/osmo-state)" <pod-path>:<local-path>` *before* tearing down |
 
 The rule of thumb: **commit + push every time you'd save a file in a
 git-tracked sense.** The Source Control panel is the persistence boundary.
diff --git a/osmo/README.md b/osmo/README.md
index 91b41dbd5..410ed21c2 100644
--- a/osmo/README.md
+++ b/osmo/README.md
@@ -7,11 +7,17 @@ through [NVIDIA OSMO](https://github.com/NVIDIA/OSMO):
 osmo/
 ├── README.md                     # This file (admin / operator reference)
 ├── workflows/
-│   └── airstack-dev.yaml         # The OSMO workflow students submit
+│   ├── airstack-dev.yaml         # Interactive dev workflow (IDE over Remote-SSH)
+│   └── airstack-mission.yaml     # Batch mission workflow (unattended flights)
+├── missions/
+│   ├── README.md                 # Mission spec schema reference
+│   └── example_takeoff_land.yaml # Reference mission: takeoff → hover → land ×3
 └── workspace/
     ├── Dockerfile                # The airstack-osmo-workspace image
     ├── sshd_config               # Pubkey-only sshd config baked into the image
-    └── entrypoint.sh             # Pod startup: sshd, dockerd, clone, airstack up
+    ├── entrypoint.sh             # Pod startup: sshd, dockerd, clone, then
+    │                             #   dev mode (airstack up) or mission mode
+    └── mission_runner.py         # Batch executor (run from the clone, not the image)
 ```
 
 The student-facing walkthrough lives in
@@ -21,9 +27,9 @@ README is the **lab admin / operator** reference: pool requirements,
 workspace image build & push, validation stages, plus a credential summary
 for context.
 
-> **Scope:** developer workflow only. CI/CD on OSMO is **not** part of this
-> integration — the existing `system-tests.yml` + OpenStack orchestrator path
-> is unchanged.
+> **Scope:** developer workflow + batch missions. CI/CD on OSMO is **not**
+> part of this integration — the existing `system-tests.yml` + OpenStack
+> orchestrator path is unchanged.
 
 ## Architecture in one minute
 
@@ -47,6 +53,49 @@ app.foxglove.dev   ── ws ────►  port-forward 8766 ────►
                                             airstack.sh up brings these 3 up
 ```
 
+## Mission mode (batch runs)
+
+`airstack-mission.yaml` reuses the same workspace image and DinD pod, but
+instead of one interactive `airstack up`, the entrypoint hands off to
+[`workspace/mission_runner.py`](workspace/mission_runner.py), which executes
+a declarative mission spec from [`missions/`](missions/) — repeated cycles of:
+
+```
+airstack down → airstack up → wait for PX4 ready → record mcap bags
+→ run steps (takeoff / land / navigate / semantic search / any ros2 command)
+→ collect bags + container logs → airstack down
+```
+
+Submit, monitor, and download:
+
+```bash
+airstack osmo:mission osmo/missions/example_takeoff_land.yaml --pool <gpu-pool>
+airstack osmo:logs                 # follow mission progress
+airstack osmo:fetch ./results/     # rsync bags/logs/summaries to the laptop
+airstack osmo:down                 # cancel (fetch first — results die with the pod)
+```
+
+Key behaviors:
+
+- **The mission spec and runner come from the clone**, not the image — what
+  you push on your branch is what runs. The workspace image only needs a
+  rebuild when `Dockerfile`, `sshd_config`, or `entrypoint.sh` change.
+- **Bags are mcap** (`ros2 bag record -s mcap`) — open the `.mcap` files
+  directly in Foxglove, no conversion or local ROS install.
+- **Results location:** `/osmo/output/airstack-mission-results/<mission>/<stamp>/`
+  with a symlink at `/root/AirStack/osmo/results` (the path `osmo:fetch`
+  pulls). Artifacts are collected even for failed iterations.
+- **`OSMO_MISSION_KEEP_ALIVE`** (default `true`): the pod sleeps after the
+  mission so you can fetch over ssh. Set `false` (or submit with
+  `osmo:mission --no-keep-alive`) for fire-and-forget: the task exits
+  cleanly when the mission ends, freeing the GPU — and if the workflow's
+  `outputs:` block is configured with a destination bucket, OSMO uploads
+  `/osmo/output` automatically on that exit. A **canceled** workflow does
+  not upload outputs, so in keep-alive mode `osmo:fetch` is the retrieval
+  path.
+
+Mission spec schema and step types: [`missions/README.md`](missions/README.md).
+
 ## Pool requirements
 
 The OSMO pool the workflow runs on must satisfy:
diff --git a/osmo/missions/README.md b/osmo/missions/README.md
new file mode 100644
index 000000000..f33b977e7
--- /dev/null
+++ b/osmo/missions/README.md
@@ -0,0 +1,182 @@
+# AirStack mission specs
+
+A **mission** is a declarative YAML file executed by
+[`osmo/workspace/mission_runner.py`](../workspace/mission_runner.py). Each
+mission runs one or more full iterations of:
+
+```
+airstack down → airstack up → wait for PX4 ready → start mcap recording
+→ run steps → stop recording → collect bags + container logs → airstack down
+```
+
+Missions live in this directory so they're versioned with the code they
+exercise — the OSMO pod clones your branch, so whatever you push is what runs.
+
+Submit to OSMO with:
+
+```bash
+airstack osmo:mission osmo/missions/<mission>.yaml --pool <gpu-pool>
+```
+
+Or run locally on any machine that can `airstack up` (no OSMO involved):
+
+```bash
+python3 osmo/workspace/mission_runner.py osmo/missions/<mission>.yaml \
+  --airstack-root "$(pwd)"
+```
+
+`--dry-run` validates the spec and prints the merged config without touching
+Docker.
+
+## Results
+
+```
+osmo/results/<mission-name>/<UTC stamp>/
+├── summary.json              # per-iteration status + durations
+└── iter_001/
+    ├── bags/robot_1/*.mcap   # open directly in Foxglove (no conversion)
+    ├── logs/<container>.log  # docker logs snapshot per container
+    ├── logs/<container>/     # raw /tmp stdout tees (gossip, ddsrouter, relay) + rcl control-stack logs
+    ├── ready.json            # per-robot seconds-to-PX4-ready
+    ├── steps.json            # per-step command, output tail, pass/fail
+    └── iteration.json        # iteration summary
+```
+
+On an OSMO pod the actual storage is `/osmo/output/airstack-mission-results`
+(with `osmo/results` symlinked to it), so a workflow `outputs:` block uploads
+everything automatically when the task exits. Download from your laptop at
+any time while the pod is alive with `airstack osmo:fetch [dest]`.
+
+Artifacts are collected **even when an iteration fails** — a failed flight's
+bag is usually the most interesting one.
+
+## Schema
+
+Top-level keys (everything except `steps` is optional):
+
+| Key | Default | Meaning |
+|---|---|---|
+| `name` | filename stem | Results directory name |
+| `env` | `{}` | Env vars exported before each `airstack up` (`NUM_ROBOTS`, `COMPOSE_PROFILES`, `ISAAC_SIM_SCRIPT_NAME`, …) |
+| `iterations` | `1` | Full up→fly→down cycles. With `environments:` + `environment_order: grouped`, this is **per environment** (total = iterations × #environments) |
+| `iteration_attempts` | `1` | Max times to (re)run a single iteration until it passes. `>1`: a failed/errored iteration is redone (clean down→up→fly→down) up to this many attempts; a `passed`, manually `stopped`, or `abort_mission` outcome is never retried. Failed attempts' artifacts are preserved under `iter_NNN_failed_attempt_K`. With a fixed `SPAWN_SEED` the redo reproduces the same spawn layout |
+| `environment_order` | `round_robin` | With `environments:`: `round_robin` (iteration i → env[(i-1) % n], `iterations` is the total) \| `grouped` (each env runs `iterations` times in a row) |
+| `ready.timeout_s` | `600` | Max seconds to wait for PX4 readiness per iteration |
+| `ready.poll_interval_s` | `5` | Seconds between readiness polls |
+| `record.enabled` | `true` | Record an mcap per robot per iteration |
+| `record.scope` | `gcs` | `gcs` (one mcap on GCS domain 0) \| `robot` (one mcap per robot domain) \| `both` |
+| `record.topics` | tf + odom set | Topics to record; `{robot}` → `robot_N` |
+| `record.all` | `false` | Record **all** topics (`ros2 bag record -a`) — large |
+| `record.exclude` | — | With `all`: regex of topics to drop (`ros2 bag record -a --exclude-regex <regex>`) |
+| `record.required` | `false` | Abort the iteration if any recorder fails to start (don't fly unrecorded) |
+| `on_step_failure` | `abort_iteration` | `continue` \| `abort_iteration` \| `abort_mission` |
+| `up_timeout_s` | `3600` | `airstack up` timeout (first up on a fresh pod pulls images) |
+| `down_timeout_s` | `300` | `airstack down` timeout |
+| `robot_setup_bash` | robot ws `setup.bash` | Workspace sourced before `ros2` commands |
+| `nas_dest` | — | Base path on airlab-storage (e.g. `/volume3/<share>/airstack-missions`). When the `airlab-storage` OSMO credential is set (`airstack osmo:setup`), an OSMO pod rsyncs results to `<nas_dest>/<name>/<stamp>/` then tears itself down. Override per run with `osmo:mission --nas-dest PATH`; suppress with `--no-nas-upload`. Credentials never live in the spec |
+| `steps` | — | Ordered list of steps (below) |
+
+### Steps
+
+Every step type accepts the placeholders `{robot}` → `robot_N` and `{n}` → `N`
+in its strings, and runs once per robot in `robots:` (`all` by default, or a
+list like `[1, 3]`).
+
+**`action`** — send a goal to a robot task action server and wait for the
+result. The step passes when the action result reports `success: true`.
+
+```yaml
+- action:
+    task: takeoff                 # → /robot_N/tasks/<task>
+    goal: {target_altitude_m: 10.0, velocity_m_s: 1.0}
+    timeout_s: 120                # default 120 (per attempt)
+    attempts: 3                   # per-robot retries on failure (default 3)
+    retry_delay_s: 10             # wait between attempts (default 10)
+    feedback_timeout_s: 15        # via gcs: no relay_feedback within this
+                                  # window ⇒ goal presumed lost, retried
+    pass_on_feedback: false       # via gcs: pass as soon as feedback is seen
+                                  # (the task RAN), regardless of success/fail;
+                                  # only "never ran" (no feedback + no result
+                                  # after retries) fails the step
+    robots: all
+    # type: task_msgs/action/TakeoffTask   # derived from task name if omitted
+```
+
+Any step may also carry **`optional: true`** (a sibling key, not inside
+`action`/`run`/…): the step still runs and its result is recorded, but a
+failure neither trips `on_step_failure` nor counts toward an `iteration_attempts`
+redo. Use it for steps whose outcome doesn't gate the iteration — e.g. a `land`
+after the run is already done.
+
+**`pass_on_feedback` + `iteration_attempts` idiom** — to guarantee a task
+*runs* every iteration without caring whether it succeeds: set
+`pass_on_feedback: true` on that action and `iteration_attempts: >1` on the
+mission. If the task never gets feedback (goal lost / never started) after its
+`attempts` retries, the step fails and the whole iteration is redone; once
+feedback is seen the iteration is considered satisfied.
+
+Each robot's goal is logged per attempt and retried independently — a goal
+can be rejected transiently (relay has no GPS fix yet, PX4 position estimate
+not converged, action server still starting), so one robot failing its first
+attempt doesn't fail the step unless it exhausts all attempts.
+
+Available tasks (action type is derived as `task_msgs/action/<CamelCase>Task`):
+`takeoff`, `land`, `fixed_trajectory`, `navigate`, `exploration`, `coverage`,
+`semantic_search`, `chat`. Goal fields are defined in
+[`common/ros_packages/msgs/task_msgs/action/`](../../common/ros_packages/msgs/task_msgs/action/).
+Multi-robot action goals are sent **in parallel** across robots.
+
+**`wait`** — sleep for N seconds (e.g. hover, let a planner run):
+
+```yaml
+- wait: 30
+```
+
+**`run`** — arbitrary command; the escape hatch that makes any ROS 2 command
+work without runner changes. The step fails on non-zero exit unless
+`expect_success: false`. Set `attempts` to retry on failure (e.g. a flaky
+model download) before the step is marked failed.
+
+```yaml
+- run:
+    container: robot_1     # robot_N → exec in the robot container on robot
+                           #   N's DDS domain (ros2 is sourced for you);
+                           # pod → run on the pod itself (cwd = AirStack root);
+                           # any other value → literal container name, domain 0
+    cmd: ros2 topic echo --once /{robot}/odometry
+    timeout_s: 60
+    expect_success: true
+    attempts: 1            # retries on failure (default 1 = no retry)
+    retry_delay_s: 10      # wait between attempts (default 10)
+```
+
+**`topic_pub`** — `ros2 topic pub --once` per robot:
+
+```yaml
+- topic_pub:
+    topic: /{robot}/some_input
+    type: std_msgs/msg/Bool
+    msg: {data: true}
+```
+
+**`service_call`** — `ros2 service call` per robot:
+
+```yaml
+- service_call:
+    service: /{robot}/some_service
+    type: std_srvs/srv/Trigger
+    request: {}
+```
+
+## Notes
+
+- For `NUM_ROBOTS > 1` on Isaac Sim, set
+  `ISAAC_SIM_SCRIPT_NAME: example_multi_px4_pegasus_launch_script.py` in
+  `env:` — the default script spawns a single drone.
+- Missions run unattended: keep `ISAAC_SIM_HEADLESS: "true"`,
+  `ISAAC_SIM_USE_STANDALONE: "true"` and `PLAY_SIM_ON_START: "true"` unless
+  you're watching via the WebRTC livestream profile.
+- Recording camera/LiDAR topics (or `record.all: true`) is the bag-size
+  driver — budget pod `storage:` accordingly.
+- `/tf` and `/tf_static` are in the default topic set because without them a
+  Foxglove 3D panel can't pose anything during replay.
diff --git a/osmo/missions/example_takeoff_land.yaml b/osmo/missions/example_takeoff_land.yaml
new file mode 100644
index 000000000..e64bf7892
--- /dev/null
+++ b/osmo/missions/example_takeoff_land.yaml
@@ -0,0 +1,117 @@
+# Example mission: takeoff → hover 30s → land, 3 times, Isaac Sim headless.
+#
+# Run on OSMO:
+#   airstack osmo:mission osmo/missions/example_takeoff_land.yaml --pool <gpu-pool>
+#
+# Or locally against a machine that can run `airstack up` (no OSMO needed):
+#   python3 osmo/workspace/mission_runner.py osmo/missions/example_takeoff_land.yaml \
+#     --airstack-root "$(pwd)"
+#
+# Schema reference: osmo/missions/README.md
+
+name: example_takeoff_land
+
+# Optional: upload results to airlab-storage, then auto-teardown once the
+# upload succeeds (needs the airlab-storage OSMO credential — see
+# `airstack osmo:setup`). Override the path with osmo:mission --nas-dest,
+# or suppress with --no-nas-upload.
+# nas_dest: /volume<X>/<share>/airstack-missions
+
+# How to bring the stack up. `services` are passed straight through:
+#   ./airstack.sh up isaac-sim robot-desktop gcs
+# (docker compose auto-enables a named service's profile, so no
+# COMPOSE_PROFILES needed). On OSMO swap isaac-sim → isaac-sim-livestream to
+# watch the run over WebRTC. Alternatively use `profiles:` to export
+# COMPOSE_PROFILES and run a plain `airstack up`.
+stack:
+  services: [isaac-sim-livestream, robot-desktop, gcs]
+  # profiles: [desktop, isaac-sim]
+
+# Exported before every `airstack up`. Anything the compose stack reads from
+# the environment can go here (.env values, ISAAC_SIM_SCRIPT_NAME, etc.).
+env:
+  NUM_ROBOTS: 1
+  ISAAC_SIM_HEADLESS: "true"
+  ISAAC_SIM_USE_STANDALONE: "true"
+  # The default script only spawns one drone; for NUM_ROBOTS > 1 switch to
+  # example_multi_px4_pegasus_launch_script.py.
+  ISAAC_SIM_SCRIPT_NAME: example_one_px4_pegasus_launch_script.py
+  # Missions are unattended — the sim must start playing without a GUI click.
+  PLAY_SIM_ON_START: "true"
+
+# Where `action` steps are sent by default:
+#   gcs   — through the GCS action_relay, the same String-JSON → typed-goal
+#           path Foxglove and the GCS panels use (exercises the full
+#           GCS→robot chain, including the relay's airborne preconditions)
+#   robot — `ros2 action send_goal` directly on the robot's DDS domain
+# Override per step with `via:`.
+command_route: gcs
+
+# Full up → fly → collect → down cycles.
+iterations: 3
+
+# Gate before any steps run: per robot, MAVROS connected, then
+# local_position/odom publishing (PX4 EKF converged = ready to arm).
+ready:
+  timeout_s: 600
+
+# One mcap per iteration with ALL robots' data, recorded on the GCS
+# (domain 0) where each robot's domain_bridge forwards its state topics —
+# replaying the .mcap in Foxglove reproduces the live GCS view.
+# {robot} expands to robot_1..robot_N and the list is unioned.
+record:
+  enabled: true
+  scope: gcs               # or `robot`: one mcap per robot on its own
+                           # domain, for topics not bridged to the GCS
+  topics:
+    - /tf
+    - /tf_static
+    - /{robot}/odometry_conversion/odometry
+    - /{robot}/interface/mavros/global_position/global
+    - /{robot}/trajectory_controller/trajectory_vis
+    - /{robot}/global_plan
+    - /gcs/robot_markers
+    - /gcs/map_origin/location
+    - /gcs/map_origin/ground_msl
+    - /gcs/{robot}/location
+  # Or record everything visible on the recording domain (large):
+  # all: true
+
+# continue | abort_iteration (default) | abort_mission
+on_step_failure: abort_iteration
+
+# Steps run top to bottom. Every step that addresses a robot fans out over a
+# robot selection — `robots:` (default `all` = robots 1..NUM_ROBOTS, or a
+# list like [1, 3]) — and {robot} → robot_N / {n} → N expand per robot.
+# Per-step knobs (all optional, shown here with their defaults):
+#   action: robots=all  via=<command_route>  timeout_s=120 (per attempt)
+#           attempts=3 retry_delay_s=10 (per-robot retries on failure)
+#           via: gcs (through the GCS action_relay) | robot (direct send_goal)
+#   run:    robots=all  container=robot_{n} (per-robot) or robot_1 | gcs | pod |
+#           <name>;  fans out when cmd/container references {robot}/{n}
+#   wait:   <seconds>   (runs once)
+steps:
+  - action:
+      task: takeoff                      # → /<robot>/tasks/takeoff (TakeoffTask)
+      goal: {target_altitude_m: 10.0, velocity_m_s: 1.0}
+      #to specify robot(s)   
+      #robots: [1]   (1st robot)
+      #robots: [1, 3]   (1st and 3rd robot)
+      #robots: all   (all robots)
+
+  - wait: 30                             # hover
+
+  - run:                                 # arbitrary command, that runs per robot
+      cmd: ros2 topic echo --once /{robot}/odometry_conversion/odometry
+      timeout_s: 20
+      #to specify robot(s):   
+      #robots: [1]   (1st robot)
+      #robots: [1, 3]   (1st and 3rd robot)
+
+      #To specify the container:
+      #container: gcs   (GCS container)
+      #container: <container name> 
+
+  - action:
+      task: land                         # → /<robot>/tasks/land (LandTask)
+      goal: {velocity_m_s: 0.5}
diff --git a/osmo/workflows/airstack-mission.yaml b/osmo/workflows/airstack-mission.yaml
new file mode 100644
index 000000000..7429cae27
--- /dev/null
+++ b/osmo/workflows/airstack-mission.yaml
@@ -0,0 +1,138 @@
+# AirStack batch mission workflow on OSMO.
+#
+# Same workspace image and pod layout as airstack-dev.yaml, but instead of a
+# single `airstack up` for an interactive session, this runs a batch mission:
+# osmo/workspace/mission_runner.py executes the mission spec named by
+# OSMO_MISSION_FILE — repeated cycles of
+#
+#   airstack down → airstack up → wait for PX4 ready → record mcap bags
+#   → run steps (takeoff / land / navigate / any ros2 command) → collect
+#   bags + logs → airstack down
+#
+# NO IMAGE REBUILD NEEDED. The mission engine is NOT baked into the image:
+# the command below reuses the baked entrypoint for pod setup only
+# (OSMO_AIRSTACK_UP=false → sshd, inner dockerd, branch clone, creds, registry
+# login — no `airstack up`), then hands off to mission_launcher.sh FROM THE
+# CLONE. So `git push` your branch and resubmit; the launcher + runner + spec
+# all come from the branch. (The launcher pip-installs PyYAML at runtime if
+# the image predates it.)
+#
+# Mission specs live in osmo/missions/ (cloned with the branch, so what you
+# push is what runs). Schema: osmo/missions/README.md.
+#
+# To submit (or use the wrapper: airstack osmo:mission <mission.yaml>):
+#
+#   osmo workflow submit osmo/workflows/airstack-mission.yaml \
+#     --pool <gpu-pool> \
+#     --set-env "SSH_PUB_KEY=$(cat ~/.ssh/id_ed25519.pub)" \
+#                "OSMO_MISSION_FILE=osmo/missions/example_takeoff_land.yaml" \
+#                "AIRSTACK_BRANCH=<your-branch>"
+#
+# Results land in /osmo/output/airstack-mission-results (symlinked from
+# /root/AirStack/osmo/results). With the default OSMO_MISSION_KEEP_ALIVE=true
+# the pod stays alive after the mission so you can pull artifacts from your
+# laptop:
+#
+#   airstack osmo:fetch ./results/
+#
+# Set OSMO_MISSION_KEEP_ALIVE=false for fire-and-forget runs: the task exits
+# cleanly when the mission ends, which frees the GPU and triggers the
+# `outputs:` upload below (if a destination bucket is configured).
+#
+# NAS upload: if the airlab-storage credential is set (airstack osmo:setup) and
+# the mission spec has `nas_dest:`, the pod rsyncs results to airlab-storage
+# then tears itself down — no laptop fetch needed. See mission_launcher.sh.
+
+workflow:
+  name: airstack-mission
+  groups:
+  - name: airstack
+    tasks:
+    - name: workspace
+      lead: true
+      image: airlab-docker.andrew.cmu.edu/airstack/airstack-osmo-workspace:latest
+      # Required so the inner dockerd can run (same DinD setup as airstack-dev).
+      privileged: true
+      # Bootstrap: run the baked entrypoint for pod setup ONLY
+      # (OSMO_AIRSTACK_UP=false), backgrounded, then hand off to the launcher
+      # that ships in the freshly-cloned branch. This is what makes the
+      # mission engine rebuild-free — see the header comment.
+      command: ["bash", "-c"]
+      args:
+      - |
+        set -uo pipefail
+        log() { echo "[mission-bootstrap] $*"; }
+        log "starting pod setup via baked entrypoint (OSMO_AIRSTACK_UP=false)"
+        # Baked entrypoint for pod setup, backgrounded. `tee` streams its logs to
+        # `osmo workflow logs` and to a file the launcher greps for the terminal
+        # "sleeping forever" line. `tee` holds THIS task's stdout pipe open, so
+        # mission_launcher.sh kills BAKED_PID before it exits — otherwise
+        # osmo_exec never sees pipe EOF on our exit and the pod never frees the
+        # GPU (it hangs at RUNNING until exec_timeout).
+        OSMO_AIRSTACK_UP=false /usr/local/bin/entrypoint.sh 2>&1 \
+          | tee /tmp/baked-entrypoint.log &
+        export BAKED_PID=$!
+        LAUNCHER=/root/AirStack/osmo/workspace/mission_launcher.sh
+        log "waiting for branch clone to provide $LAUNCHER"
+        for i in $(seq 1 600); do [ -f "$LAUNCHER" ] && break; sleep 2; done
+        if [ ! -f "$LAUNCHER" ]; then
+          log "ERROR: $LAUNCHER not found after clone wait; sleeping for SSH debug"
+          exec sleep infinity
+        fi
+        log "handing off to mission_launcher.sh"
+        exec bash "$LAUNCHER"
+      environment:
+        # Mission selection — repo-relative path into the clone. Override at
+        # submit time to run a different mission.
+        OSMO_MISSION_FILE: "osmo/missions/example_takeoff_land.yaml"
+        # true  → pod sleeps after the mission; fetch results over ssh, then
+        #         `airstack osmo:down` (results die with the pod!).
+        # false → task exits when the mission ends: GPU freed, /osmo/output
+        #         uploaded to `outputs:` destinations.
+        OSMO_MISSION_KEEP_ALIVE: "true"
+        AIRSTACK_BRANCH: "krrishj18/osmo-mission-runner"        # branch entrypoint.sh clones
+        AIRSTACK_REPO_URL: "https://github.com/castacks/AirStack.git"
+        # SSH_PUB_KEY is supplied at submit time (needed for osmo:fetch):
+        #   --set-env "SSH_PUB_KEY=$(cat ~/.ssh/id_ed25519.pub)"
+      # Uncomment and point at a bucket the OSMO deployment can write to, to
+      # have /osmo/output uploaded automatically when the task exits (only
+      # meaningful with OSMO_MISSION_KEEP_ALIVE=false — a canceled workflow
+      # does not upload outputs):
+      #
+      # outputs:
+      #   - url: s3://<bucket>/airstack-missions/{{workflow_id}}/
+      credentials:
+        # Same per-user credentials as airstack-dev.yaml — see
+        # docs/tutorials/airstack_on_osmo.md "Step 0" / `airstack osmo:setup`.
+        # These map OSMO credential fields → pod env vars; no secret values live
+        # in this file.
+        airlab-nucleus:
+          OMNI_USER: omni_user
+          OMNI_PASS: omni_pass
+          OMNI_SERVER: omni_server
+        airlab-docker-login:
+          AIRLAB_REGISTRY_USER: username
+          AIRLAB_REGISTRY_PASS: password
+        # Optional: NAS result upload. Absent credential → nothing injected →
+        # upload disabled. Enabled per-mission via `nas_dest:` in the spec.
+        airlab-storage:
+          AIRLAB_STORAGE_USER: username
+          AIRLAB_STORAGE_PASS: password
+          AIRLAB_STORAGE_HOST: host
+
+  resources:
+    default:
+      cpu: 16
+      gpu: 1
+      memory: 64Gi
+      # Same sizing rationale as airstack-dev.yaml (inner image set alone
+      # exceeds 100Gi extracted), plus headroom for per-iteration mcap
+      # recordings — bump if a mission records camera/LiDAR over many
+      # iterations.
+      storage: 500Gi
+
+  timeout:
+    # With keep-alive the pod sleeps after the mission until this expires (then
+    # it's gone — fetch first). Sized to stay up ~30h post-mission for
+    # osmo:fetch (mission itself runs in the first ~1-2h).
+    exec_timeout: 32h
diff --git a/osmo/workspace/Dockerfile b/osmo/workspace/Dockerfile
index e80f3be59..d6a43f5d1 100644
--- a/osmo/workspace/Dockerfile
+++ b/osmo/workspace/Dockerfile
@@ -42,7 +42,9 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
     procps \
     python3 \
     python3-pip \
+    python3-yaml \
     rsync \
+    sshpass \
     sudo \
     tmux \
     tzdata \
diff --git a/osmo/workspace/entrypoint.sh b/osmo/workspace/entrypoint.sh
index bef9d9ba8..dd53a94ae 100755
--- a/osmo/workspace/entrypoint.sh
+++ b/osmo/workspace/entrypoint.sh
@@ -259,12 +259,17 @@ else
   export COMPOSE_PROFILES="${COMPOSE_PROFILES:-desktop,isaac-sim}"
 fi
 
+# OSMO_AIRSTACK_UP=false skips `airstack up` and goes straight to sleep — the
+# pod is fully set up (sshd, dockerd, clone, creds) for the caller to drive.
+# The airstack-mission.yaml workflow uses this hook: it runs this entrypoint
+# for setup only, then hands off to osmo/workspace/mission_launcher.sh (from
+# the clone, so the mission engine ships with the branch — no image rebuild).
 cd "$AIRSTACK_ROOT"
 if [ "${OSMO_AIRSTACK_UP:-true}" = "true" ]; then
   log "airstack up (COMPOSE_PROFILES=$COMPOSE_PROFILES, NUM_ROBOTS=$NUM_ROBOTS, livestream=$ISAAC_SIM_LIVESTREAM)"
   ./airstack.sh up || log "WARN: airstack up exited non-zero — pod stays alive for debugging via SSH"
 else
-  log "OSMO_AIRSTACK_UP=false — skipping airstack up; SSH in and run ./airstack.sh up manually"
+  log "OSMO_AIRSTACK_UP=false — skipping airstack up; caller drives the stack"
 fi
 
 # ─── 7. Sleep ──────────────────────────────────────────────────────────────
diff --git a/osmo/workspace/mission_launcher.sh b/osmo/workspace/mission_launcher.sh
new file mode 100755
index 000000000..b470d10c7
--- /dev/null
+++ b/osmo/workspace/mission_launcher.sh
@@ -0,0 +1,165 @@
+#!/usr/bin/env bash
+# mission_launcher.sh — no-rebuild mission entrypoint for AirStack-on-OSMO.
+#
+# The airstack-mission.yaml workflow runs the *baked* image entrypoint with
+# OSMO_AIRSTACK_UP=false (pod setup only: sshd, inner dockerd, branch clone,
+# Nucleus creds, registry login), then hands off to THIS script — which lives
+# in the clone, so the mission engine always comes from your branch and no
+# workspace-image rebuild is needed to iterate. (PyYAML, which the baked image
+# may predate, is installed at runtime below.)
+#
+# Responsibilities:
+#   1. Wait for the setup the baked entrypoint performs in the background
+#      (inner dockerd ready, clone present, registry login done).
+#   2. Ensure PyYAML is importable.
+#   3. Run mission_runner.py against $OSMO_MISSION_FILE.
+#   4. Keep-alive: stay foreground (sleep) so the pod survives for
+#      `airstack osmo:fetch`, or exit so OSMO uploads /osmo/output + frees GPU.
+
+set -uo pipefail
+
+AIRSTACK_ROOT="${AIRSTACK_ROOT:-/root/AirStack}"
+
+log()  { echo "[mission-launcher] $*"; }
+fail() { echo "[mission-launcher] ERROR: $*" >&2; }
+
+# Kill the backgrounded `tee` (BAKED_PID, from the workflow bootstrap) so it
+# releases this task's OSMO stdout pipe — else osmo_exec never sees EOF on our
+# exit and the pod never frees the GPU. Call right before an intentional exit.
+release_osmo_pipe() { [ -n "${BAKED_PID:-}" ] && kill "$BAKED_PID" 2>/dev/null; true; }
+
+# Fast DDS LARGE_DATA uses shared memory between same-host/same-version
+# participants; the segments live in /dev/shm, which every stack container
+# bind-mounts from the pod. The pod default is 64M — it fills immediately
+# and later participants then fail SHM registration. Regrow in place.
+mount -o remount,size=8G /dev/shm 2>/dev/null
+log "/dev/shm: $(df -h /dev/shm | awk 'NR==2{print $2" total, "$5" used"}')"
+
+# wait_for <description> <timeout_s> <command...> — poll until the command
+# succeeds or the timeout elapses. Returns non-zero on timeout.
+wait_for() {
+  local desc="$1" timeout="$2"; shift 2
+  local i=0
+  until "$@" >/dev/null 2>&1; do
+    i=$((i + 1))
+    if [ "$i" -ge "$timeout" ]; then
+      fail "timed out after ${timeout}s waiting for: ${desc}"
+      return 1
+    fi
+    [ $((i % 15)) -eq 0 ] && log "still waiting for ${desc} (${i}s)"
+    sleep 1
+  done
+  log "ready: ${desc}"
+}
+
+# ── 1. setup readiness (driven by the backgrounded baked entrypoint) ───────
+# If dockerd or the clone never appear, the pod is broken — stay alive (when
+# keep-alive) so it can be inspected over SSH rather than vanishing.
+if ! wait_for "inner dockerd" 180 docker info; then
+  [ "${OSMO_MISSION_KEEP_ALIVE:-true}" = "true" ] && exec sleep infinity
+  exit 1
+fi
+if ! wait_for "branch clone" 600 test -f "$AIRSTACK_ROOT/osmo/workspace/mission_runner.py"; then
+  [ "${OSMO_MISSION_KEEP_ALIVE:-true}" = "true" ] && exec sleep infinity
+  exit 1
+fi
+
+# The deployed :latest image's baked entrypoint ignores OSMO_AIRSTACK_UP and
+# runs its OWN `airstack up`. Running the mission's bring-up concurrently
+# collides (duplicate network, container-name conflicts, "network not found"
+# mid-teardown) and fails the first iteration. So wait for the baked
+# entrypoint to FINISH — the bootstrap tees its output to $BAKED_LOG, and its
+# terminal "sleeping forever" line is printed exactly once, after its
+# `airstack up` completes (success or fail). This also subsumes the clone +
+# registry-login steps it performs, and warms the inner image cache the
+# mission reuses. On a rebuilt image where the hook works, the entrypoint
+# skips `up` and prints that line almost immediately, so this is a fast no-op.
+#
+# We deliberately do NOT detect completion via `pgrep -f "sleep infinity"`:
+# the AirStack containers themselves run `sleep infinity`, so that matches the
+# instant the baked bring-up starts a container — firing the handoff in the
+# middle of its `airstack up`, which is exactly the race we're avoiding.
+BAKED_LOG="${BAKED_ENTRYPOINT_LOG:-/tmp/baked-entrypoint.log}"
+log "waiting for the baked entrypoint to finish its own bring-up (avoids a concurrent 'airstack up')"
+if ! wait_for "baked entrypoint complete" 2400 grep -q "sleeping forever" "$BAKED_LOG"; then
+  log "WARN: baked entrypoint didn't signal completion in 40m — proceeding; first iteration may race its bring-up"
+fi
+sleep 5  # let baked's compose settle before the mission's first ensure_down/up
+
+# ── 2. PyYAML (mission_runner imports yaml) ────────────────────────────────
+if ! python3 -c "import yaml" >/dev/null 2>&1; then
+  log "installing PyYAML (image predates python3-yaml)"
+  pip3 install --break-system-packages --quiet pyyaml \
+    || { apt-get update -qq && apt-get install -y -qq python3-yaml; } \
+    || fail "could not install PyYAML — mission_runner will fail to import yaml"
+fi
+
+# ── 3. run the mission ─────────────────────────────────────────────────────
+MISSION="${OSMO_MISSION_FILE:-}"
+if [ -z "$MISSION" ]; then
+  fail "OSMO_MISSION_FILE not set — nothing to run"
+else
+  MISSION_PATH="$AIRSTACK_ROOT/$MISSION"
+  [ -f "$MISSION_PATH" ] || MISSION_PATH="$MISSION"   # allow an absolute path
+  if [ -f "$MISSION_PATH" ]; then
+    log "running mission: $MISSION_PATH"
+    python3 "$AIRSTACK_ROOT/osmo/workspace/mission_runner.py" "$MISSION_PATH" \
+      --airstack-root "$AIRSTACK_ROOT"
+    log "mission_runner exited $?"
+  else
+    fail "mission file not found: $MISSION (looked under the clone and as an absolute path)"
+  fi
+fi
+
+# ── 3b. NAS upload + teardown (optional) ───────────────────────────────────
+# With the airlab-storage credential injected and a destination set (--nas-dest,
+# else the spec's nas_dest:), rsync results to the NAS. Success → pod exits (GPU
+# freed); failure → stay alive so results survive for osmo:fetch.
+NAS_YAML_DEST=""
+if [ -f "${MISSION_PATH:-}" ]; then
+  NAS_YAML_DEST="$(python3 -c 'import yaml,sys; d=yaml.safe_load(open(sys.argv[1])) or {}; print(d.get("nas_dest") or "")' "$MISSION_PATH" 2>/dev/null)"
+fi
+NAS_DEST="${OSMO_MISSION_UPLOAD_DEST:-$NAS_YAML_DEST}"
+
+if [ "${OSMO_MISSION_NO_UPLOAD:-false}" != "true" ] \
+   && [ -n "${AIRLAB_STORAGE_USER:-}" ] && [ -n "${AIRLAB_STORAGE_PASS:-}" ] \
+   && [ -n "$NAS_DEST" ]; then
+  if ! command -v sshpass >/dev/null 2>&1; then
+    log "installing sshpass (image predates it)"
+    { apt-get update -qq && apt-get install -y -qq sshpass; } \
+      || fail "could not install sshpass — skipping NAS upload"
+  fi
+  if command -v sshpass >/dev/null 2>&1; then
+    nas_host="${AIRLAB_STORAGE_HOST:-airlab-storage.andrew.cmu.edu}"
+    results_dir="${OSMO_RESULTS_ROOT:-$AIRSTACK_ROOT/osmo/results}"
+    ssh_cmd="ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o LogLevel=ERROR"
+    log "NAS upload: ${results_dir}/ → ${nas_host}:${NAS_DEST%/}/"
+    export SSHPASS="$AIRLAB_STORAGE_PASS"
+    # NAS share (Synology rsync 3.1.x) rejects --mkpath and can't chmod/chown the
+    # SMB-backed volume, so skip those attrs; nas_dest base must already exist.
+    sshpass -e rsync -rltz --partial --timeout=1800 \
+      --no-perms --no-owner --no-group --omit-dir-times -e "$ssh_cmd" \
+      "${results_dir}/" "${AIRLAB_STORAGE_USER}@${nas_host}:${NAS_DEST%/}/"
+    nas_rc=$?
+    unset SSHPASS
+    if [ "$nas_rc" -eq 0 ]; then
+      log "NAS upload OK → ${nas_host}:${NAS_DEST} — tearing pod down (GPU freed)"
+      release_osmo_pipe
+      exit 0
+    fi
+    fail "NAS upload FAILED (rc=$nas_rc) — keeping pod alive for 'airstack osmo:fetch' / SSH debug"
+    exec sleep infinity
+  fi
+elif [ "${OSMO_MISSION_NO_UPLOAD:-false}" != "true" ] \
+     && [ -n "${AIRLAB_STORAGE_USER:-}" ] && [ -z "$NAS_DEST" ]; then
+  log "airlab-storage set but no nas_dest / --nas-dest — skipping upload; pod stays alive."
+fi
+
+# ── 4. lifetime ────────────────────────────────────────────────────────────
+if [ "${OSMO_MISSION_KEEP_ALIVE:-true}" = "true" ]; then
+  log "OSMO_MISSION_KEEP_ALIVE=true — pod stays alive; fetch with 'airstack osmo:fetch'"
+  exec sleep infinity
+fi
+log "OSMO_MISSION_KEEP_ALIVE=false — exiting so OSMO uploads /osmo/output and frees the GPU"
+release_osmo_pipe
+exit 0
diff --git a/osmo/workspace/mission_runner.py b/osmo/workspace/mission_runner.py
new file mode 100644
index 000000000..110b8067d
--- /dev/null
+++ b/osmo/workspace/mission_runner.py
@@ -0,0 +1,1478 @@
+#!/usr/bin/env python3
+"""mission_runner.py — batch mission executor for AirStack-on-OSMO pods.
+
+Reads a declarative mission spec (YAML, see osmo/missions/README.md) and runs
+N full iterations of:
+
+    airstack down → airstack up → wait for PX4 ready → start mcap recording
+    → execute steps (ros2 action goals / topic pubs / service calls / raw
+    commands / waits) → stop recording → collect bags + container logs →
+    airstack down
+
+Artifacts land under one results root per mission run:
+
+    <results_root>/<mission-name>/<UTC stamp>/
+    ├── summary.json                 # per-iteration pass/fail + durations + method
+    └── iter_001__<env>__<method>/   # e.g. iter_001__warehouse__frontier
+                                     # (iter_NNN prefix kept; env+method appended)
+        ├── bags/robot_1/*.mcap      # Foxglove-ready (open the .mcap directly)
+        ├── logs/<container>.log     # docker logs snapshot, per container
+        ├── ready.json               # per-robot PX4 readiness timings
+        └── steps.json               # per-step command, output tail, status
+
+The results root prefers /osmo/output (OSMO uploads that directory to the
+workflow's `outputs:` destinations when the task exits) and falls back to
+<airstack>/osmo/results. Either way a symlink is left at
+<airstack>/osmo/results so `airstack osmo:fetch` always finds it.
+
+Designed to run on the OSMO workspace pod (invoked by entrypoint.sh when
+OSMO_MISSION_FILE is set), but has no OSMO dependency: it only needs docker,
+python3 + PyYAML, and a checkout with airstack.sh — so it can be tested on
+any dev machine that can run `airstack up`.
+
+Command patterns (ros2 exec into the robot container with a per-robot
+ROS_DOMAIN_ID, two-gate PX4 readiness, action-result parsing) mirror
+tests/conftest.py and tests/system/test_takeoff_hover_land.py.
+"""
+
+import argparse
+import json
+import os
+import re
+import shlex
+import signal
+import subprocess
+import sys
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timezone
+from pathlib import Path
+
+import yaml
+
+ROS_DISTRO_SETUP = "/opt/ros/jazzy/setup.bash"
+DEFAULT_ROBOT_SETUP_BASH = "/root/AirStack/robot/ros_ws/install/setup.bash"
+GCS_SETUP_BASH = "/root/AirStack/gcs/ros_ws/install/setup.bash"
+ROBOT_CONTAINER_PATTERN = "robot.*desktop"
+
+# Default topics for `record.scope: gcs` — one recorder in the GCS container
+# on domain 0, where each robot's domain_bridge (autonomy_bringup
+# onboard_all/config/domain_bridge.yaml) forwards its state topics. All
+# robots land in ONE mcap, and replaying it in Foxglove reproduces the live
+# GCS view (gcs_visualizer markers are already in the global 'map' frame).
+# {robot} expands to robot_1..robot_N and the result is unioned.
+DEFAULT_GCS_RECORD_TOPICS = [
+    "/tf",
+    "/tf_static",
+    "/{robot}/odometry_conversion/odometry",
+    "/{robot}/interface/mavros/global_position/global",
+    "/{robot}/trajectory_controller/trajectory_vis",
+    "/{robot}/global_plan",
+    "/gcs/robot_markers",
+    "/gcs/map_origin/location",
+    "/gcs/map_origin/ground_msl",
+    "/gcs/{robot}/location",
+]
+
+# Default topics for `record.scope: robot` — one recorder (and one mcap) per
+# robot on its own DDS domain. Use this scope for topics that are NOT
+# bridged to the GCS (raw sensors, high-rate local topics).
+DEFAULT_ROBOT_RECORD_TOPICS = [
+    "/tf",
+    "/tf_static",
+    "/{robot}/odometry_conversion/odometry",
+    "/{robot}/interface/mavros/global_position/global",
+    "/{robot}/global_plan",
+]
+
+# In-container staging dir for bag recordings (docker cp'd out before the
+# stack goes down).
+BAG_STAGING_DIR = "/tmp/osmo_bags"
+
+# Tasks the GCS action_relay bridges (gcs/ros_ws/src/action_relay). Goals for
+# these can be routed `via: gcs` — published as String JSON on
+# /<robot>/tasks/<task>/goal (GCS domain 0), exactly the path Foxglove uses.
+GCS_RELAY_TASKS = {"takeoff", "land", "navigate", "fixed_trajectory",
+                   "semantic_search", "exploration"}
+
+MISSION_DEFAULTS = {
+    "iterations": 1,
+    # How many times to (re)run a single iteration that doesn't pass before
+    # giving up and moving on. 1 = no redo (the historical behavior). >1 = if an
+    # iteration ends failed/errored (a step up to and including the gating step
+    # failed — e.g. takeoff never armed, or semantic_search never got feedback),
+    # tear the stack down and run that same iteration again, up to this many
+    # total attempts. A `stopped` (manual) or `abort_mission` outcome is never
+    # retried. Failed attempts' artifacts are preserved under
+    # iter_NNN_failed_attempt_K so the canonical iter_NNN holds the last attempt.
+    "iteration_attempts": 1,
+    # With an `environments:` list: round_robin → iteration i uses
+    # environments[(i-1) % n], so `iterations` is the TOTAL run count.
+    # grouped → each environment runs `iterations` times consecutively
+    # (retro×N, then fireacademy×N, …), so the total is iterations * n.
+    "environment_order": "round_robin",  # round_robin | grouped
+    "on_step_failure": "abort_iteration",  # continue | abort_iteration | abort_mission
+    "ready": {"timeout_s": 600, "poll_interval_s": 5},
+    # scope "gcs": one recorder on GCS domain 0 → one mcap with every robot's
+    # bridged topics (default). scope "robot": one recorder + one mcap per
+    # robot on its own domain (for unbridged/high-rate topics). scope "both":
+    # the GCS recorder AND the per-robot recorders together.
+    "record": {"enabled": True, "scope": "gcs"},
+    # How the stack is brought up. Either (or both):
+    #   services: [isaac-sim, robot-desktop, gcs]  → ./airstack.sh up <services>
+    #     (compose auto-enables a named service's profile)
+    #   profiles: [desktop, isaac-sim]             → exported as COMPOSE_PROFILES
+    # With neither, plain `airstack up` with COMPOSE_PROFILES=desktop,isaac-sim.
+    "stack": {"services": [], "profiles": []},
+    # Default route for `action` steps: "gcs" sends goals through the GCS
+    # action_relay (same path as Foxglove / the GCS panels — exercises the
+    # full GCS→robot chain incl. the relay's airborne preconditions);
+    # "robot" sends ros2 action goals directly on the robot's DDS domain.
+    # Override per step with `via:`.
+    "command_route": "gcs",
+    "up_timeout_s": 3600,    # first `up` on a fresh pod pulls the full image set
+    "down_timeout_s": 300,
+    "robot_setup_bash": DEFAULT_ROBOT_SETUP_BASH,
+}
+
+
+class MissionStop(Exception):
+    """Raised in the main thread on SIGINT/SIGTERM (e.g. `airstack
+    osmo:stop`): aborts the current step, finalizes recordings, collects
+    artifacts, and brings the stack down."""
+
+
+STOP_EVENT = threading.Event()
+
+
+def _stop_handler(signum, frame):
+    if STOP_EVENT.is_set():
+        return  # already stopping — don't interrupt artifact collection
+    STOP_EVENT.set()
+    raise MissionStop(f"stop requested (signal {signum})")
+
+
+def log(msg):
+    print(f"[mission {datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)
+
+
+def utc_stamp():
+    return datetime.now(timezone.utc).strftime("%Y-%m-%d_%H-%M-%S")
+
+
+# ── subprocess / docker plumbing ───────────────────────────────────────────
+
+def sh(cmd_list, timeout, env=None, cwd=None):
+    """Run a command, capturing output. Never raises on non-zero exit; a
+    TimeoutExpired is converted into a synthetic failed result so step
+    failures stay data, not exceptions."""
+    try:
+        return subprocess.run(cmd_list, capture_output=True, text=True,
+                              timeout=timeout, env=env, cwd=cwd)
+    except subprocess.TimeoutExpired as e:
+        return subprocess.CompletedProcess(
+            cmd_list, returncode=124,
+            stdout=(e.stdout or b"").decode() if isinstance(e.stdout, bytes) else (e.stdout or ""),
+            stderr=f"timed out after {timeout}s",
+        )
+
+
+def docker_exec(container, cmd, timeout):
+    return sh(["docker", "exec", container, "bash", "-c", cmd], timeout=timeout)
+
+
+def ros2_env_prefix(setup_bash, domain_id):
+    # Workspace setup is conditional: the GCS container has a different (or
+    # no) workspace at the robot path, and plain std_msgs pub/echo only needs
+    # the distro setup anyway.
+    return (f"source {ROS_DISTRO_SETUP} && "
+            f"{{ [ -f {setup_bash} ] && source {setup_bash}; }}; "
+            f"export ROS_DOMAIN_ID={domain_id}")
+
+
+def ros2_exec(container, ros2_cmd, domain_id, setup_bash, timeout):
+    return docker_exec(container, f"{ros2_env_prefix(setup_bash, domain_id)} && {ros2_cmd}",
+                       timeout=timeout)
+
+
+def list_containers(name_pattern=None, all_states=False):
+    cmd = ["docker", "ps", "--format", "{{.Names}}"]
+    if all_states:
+        cmd.insert(2, "-a")
+    if name_pattern:
+        cmd += ["--filter", f"name={name_pattern}"]
+    result = sh(cmd, timeout=15)
+    return [n for n in result.stdout.strip().splitlines() if n]
+
+
+def robot_containers():
+    """Running robot containers sorted by replica index (replica n ↔ robot_n)."""
+    def index(name):
+        tail = name.rsplit("-", 1)[-1]
+        return int(tail) if tail.isdigit() else 0
+    return sorted(list_containers(ROBOT_CONTAINER_PATTERN), key=index)
+
+
+def gcs_container():
+    names = list_containers("gcs")
+    return names[0] if names else None
+
+
+def stack_containers():
+    """Running containers belonging to the AirStack compose stack (robot, gcs,
+    and any sim variant — isaac-sim / isaac-sim-livestream / ms-airsim)."""
+    pats = ("airstack", "isaac-sim", "ms-airsim")
+    return [n for n in list_containers() if any(p in n for p in pats)]
+
+
+# ── mission spec ───────────────────────────────────────────────────────────
+
+def load_mission(path):
+    with open(path, encoding="utf-8") as f:
+        spec = yaml.safe_load(f)
+    if not isinstance(spec, dict):
+        raise ValueError(f"{path}: mission spec must be a YAML mapping")
+    if not spec.get("steps"):
+        raise ValueError(f"{path}: mission spec has no 'steps'")
+    merged = dict(MISSION_DEFAULTS)
+    for key, default in MISSION_DEFAULTS.items():
+        if isinstance(default, dict):
+            merged[key] = {**default, **(spec.get(key) or {})}
+    merged.update({k: v for k, v in spec.items() if not isinstance(MISSION_DEFAULTS.get(k), dict)})
+    merged.setdefault("name", Path(path).stem)
+    merged.setdefault("env", {})
+    if merged["on_step_failure"] not in ("continue", "abort_iteration", "abort_mission"):
+        raise ValueError(f"on_step_failure must be continue|abort_iteration|abort_mission, "
+                         f"got {merged['on_step_failure']!r}")
+    if merged["environment_order"] not in ("round_robin", "grouped"):
+        raise ValueError(f"environment_order must be round_robin|grouped, "
+                         f"got {merged['environment_order']!r}")
+    try:
+        merged["iteration_attempts"] = int(merged["iteration_attempts"])
+    except (TypeError, ValueError):
+        raise ValueError(f"iteration_attempts must be a positive integer, "
+                         f"got {merged['iteration_attempts']!r}")
+    if merged["iteration_attempts"] < 1:
+        raise ValueError(f"iteration_attempts must be >= 1, "
+                         f"got {merged['iteration_attempts']}")
+    if merged["command_route"] not in ("gcs", "robot"):
+        raise ValueError(f"command_route must be gcs|robot, got {merged['command_route']!r}")
+    if merged["record"].get("scope", "gcs") not in ("gcs", "robot", "both"):
+        raise ValueError(f"record.scope must be gcs|robot|both, "
+                         f"got {merged['record'].get('scope')!r}")
+    envs = merged.get("environments") or []
+    if not isinstance(envs, list):
+        raise ValueError("environments must be a list of environment entries")
+    for e in envs:
+        if not isinstance(e, dict) or "name" not in e:
+            raise ValueError(f"each environments entry must be a mapping with a "
+                             f"'name' key, got {e!r}")
+    merged["environments"] = envs
+    for step in merged["steps"]:
+        action = step.get("action") if isinstance(step, dict) else None
+        if action:
+            via = action.get("via", merged["command_route"])
+            if via not in ("gcs", "robot"):
+                raise ValueError(f"action via must be gcs|robot, got {via!r}")
+            if via == "gcs" and action.get("task") not in GCS_RELAY_TASKS:
+                raise ValueError(
+                    f"task '{action.get('task')}' is not bridged by the GCS action_relay "
+                    f"({', '.join(sorted(GCS_RELAY_TASKS))}); add `via: robot` to send it "
+                    f"directly on the robot's domain")
+            if action.get("pass_on_feedback") and via != "gcs":
+                raise ValueError(
+                    "pass_on_feedback requires via: gcs (feedback liveness is only "
+                    "tracked on the GCS action_relay path)")
+    return merged
+
+
+def task_action_type(task):
+    """tasks/semantic_search → task_msgs/action/SemanticSearchTask"""
+    camel = "".join(part.capitalize() for part in task.split("_"))
+    return f"task_msgs/action/{camel}Task"
+
+
+def expand(text, n):
+    """Substitute per-robot placeholders: {n} → robot index, {robot} → robot_<n>."""
+    return text.replace("{robot}", f"robot_{n}").replace("{n}", str(n))
+
+
+def has_placeholder(*strings):
+    """True if any string contains a per-robot placeholder ({robot} or {n})."""
+    return any(s and ("{robot}" in s or "{n}" in s) for s in strings)
+
+
+def step_robots(step_spec, num_robots):
+    robots = step_spec.get("robots", "all")
+    if robots == "all":
+        return list(range(1, num_robots + 1))
+    return [int(r) for r in robots]
+
+
+def uses_gcs_route(mission):
+    """True if any action step routes through the GCS action_relay."""
+    for step in mission["steps"]:
+        action = step.get("action") if isinstance(step, dict) else None
+        if action and action.get("via", mission["command_route"]) == "gcs":
+            return True
+    return False
+
+
+# ── per-iteration environment cycling ──────────────────────────────────────
+# An optional top-level `environments:` list lets one mission run several
+# scenes. environment_order=round_robin: iteration i uses environments[(i-1) %
+# len] (`iterations` = total). environment_order=grouped: each env runs
+# `iterations` times in a row (total = iterations * len). Within an entry,
+# UPPERCASE keys (ENV_URL, SPAWN_POLY, …) are exported as env vars before
+# `airstack up` (so they reach the isaac-sim container); any key can be
+# referenced from a step via a {{env.KEY}} placeholder (e.g. the search_area
+# polygon, which must track the scene).
+
+_ENV_PLACEHOLDER = re.compile(r"\{\{\s*env\.([A-Za-z0-9_]+)\s*\}\}")
+
+
+def env_exports(env_entry):
+    """UPPERCASE keys of an environment entry → env vars for `airstack up`.
+    Structured values are JSON-encoded so SPAWN_POLY can be written either as a
+    YAML list or as a JSON string in the mission spec."""
+    return {k: (v if isinstance(v, str) else json.dumps(v))
+            for k, v in env_entry.items() if k.isupper()}
+
+
+def substitute_env(obj, env_entry):
+    """Replace {{env.KEY}} placeholders in a (copy of a) step spec with values
+    from the current environment entry. A string that is *exactly* one
+    placeholder yields the raw value (so {{env.search_area}} can resolve to a
+    dict); a placeholder embedded in a larger string is substituted textually."""
+    if isinstance(obj, str):
+        whole = _ENV_PLACEHOLDER.fullmatch(obj.strip())
+        if whole:
+            return _lookup_env(env_entry, whole.group(1))
+
+        def repl(m):
+            val = _lookup_env(env_entry, m.group(1))
+            return val if isinstance(val, str) else json.dumps(val)
+        return _ENV_PLACEHOLDER.sub(repl, obj)
+    if isinstance(obj, dict):
+        return {k: substitute_env(v, env_entry) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [substitute_env(v, env_entry) for v in obj]
+    return obj
+
+
+def _lookup_env(env_entry, key):
+    if key not in env_entry:
+        raise ValueError(f"{{{{env.{key}}}}} referenced in a step but not found in "
+                         f"environment '{env_entry.get('name')}' "
+                         f"(available keys: {sorted(env_entry)})")
+    return env_entry[key]
+
+
+# ── iteration naming (env + method in the dir name) ─────────────────────────
+# `method:` is an optional free-form label (on an environment entry or at the
+# mission top level) for missions that compare algorithms/configurations across
+# iterations — it lands in the iteration dir name and summary.json so runs stay
+# distinguishable after fetch.
+
+def _slug(s):
+    """Filesystem-safe lowercase token."""
+    s = re.sub(r"[^A-Za-z0-9._-]+", "-", str(s).strip().lower())
+    return s.strip("-")
+
+
+def resolve_method(mission, env_entry):
+    """Normalized method label. Explicit `method:` on the environment entry
+    wins, then a mission-level `method:`; 'na' (omitted from names) otherwise."""
+    if env_entry and env_entry.get("method"):
+        return _slug(env_entry["method"])
+    if mission.get("method"):
+        return _slug(mission["method"])
+    return "na"
+
+
+def iter_dir_name(i, env_entry, method):
+    """iter_NNN[__<env>][__<method>]. The iter_NNN prefix is kept so `iter_*`
+    globs and numeric ordering still work; a trailing method suffix already in
+    the env name is dropped to avoid e.g. scene_a_frontier__frontier."""
+    parts = [f"iter_{i:03d}"]
+    env_slug = _slug(env_entry["name"]) if env_entry and env_entry.get("name") else ""
+    if env_slug:
+        for sep in ("_", "-"):
+            if method != "na" and env_slug.endswith(f"{sep}{method}"):
+                env_slug = env_slug[: -len(method) - 1]
+                break
+        if env_slug:
+            parts.append(env_slug)
+    if method and method != "na":
+        parts.append(method)
+    return "__".join(parts)
+
+
+# ── stack lifecycle ────────────────────────────────────────────────────────
+
+class Stack:
+    def __init__(self, airstack_root, mission):
+        self.root = airstack_root
+        self.mission = mission
+        self.env = os.environ.copy()
+        self.env.update({k: str(v) for k, v in mission["env"].items()})
+        self.env.setdefault("AUTOLAUNCH", "true")
+        self.env.setdefault("NUM_ROBOTS", "1")
+        # Bring-up selection (see MISSION_DEFAULTS["stack"]): an explicit
+        # service list is passed to `airstack up <services>`; explicit
+        # profiles become COMPOSE_PROFILES. With neither, default profiles.
+        self.services = [str(s) for s in mission["stack"].get("services") or []]
+        profiles = [str(p) for p in mission["stack"].get("profiles") or []]
+        if profiles:
+            self.env["COMPOSE_PROFILES"] = ",".join(profiles)
+        elif not self.services:
+            self.env.setdefault("COMPOSE_PROFILES", "desktop,isaac-sim")
+        self.num_robots = int(self.env["NUM_ROBOTS"])
+        self.setup_bash = mission["robot_setup_bash"]
+
+    def apply_env(self, overrides):
+        """Merge per-iteration env-var overrides (from an `environments:` entry)
+        into the bring-up environment so the next `airstack up` sees them."""
+        for k, v in overrides.items():
+            self.env[k] = v if isinstance(v, str) else str(v)
+
+    def _airstack(self, verb, timeout, extra_args=()):
+        log(f"airstack {verb} {' '.join(extra_args)} "
+            f"(NUM_ROBOTS={self.env['NUM_ROBOTS']}, "
+            f"COMPOSE_PROFILES={self.env.get('COMPOSE_PROFILES', '<from .env>')})")
+        return sh([str(Path(self.root) / "airstack.sh"), verb, *extra_args],
+                  timeout=timeout, env=self.env, cwd=self.root)
+
+    def up(self):
+        result = self._airstack("up", self.mission["up_timeout_s"],
+                                extra_args=self.services)
+        if result.returncode != 0:
+            raise RuntimeError(f"airstack up failed (exit {result.returncode}):\n"
+                               + "\n".join(result.stdout.splitlines()[-20:])
+                               + "\n" + "\n".join(result.stderr.splitlines()[-20:]))
+        deadline = time.time() + 120
+        while time.time() < deadline:
+            containers = robot_containers()
+            if len(containers) >= self.num_robots:
+                return containers
+            time.sleep(3)
+        raise RuntimeError(f"expected {self.num_robots} robot containers, "
+                           f"found {robot_containers()} after airstack up")
+
+    def down(self):
+        result = self._airstack("down", self.mission["down_timeout_s"])
+        if result.returncode != 0:
+            log(f"WARN: airstack down exited {result.returncode}")
+
+    def ensure_down(self):
+        """`airstack status`; if any stack containers exist, `airstack down`
+        and BLOCK until they're actually gone before returning.
+
+        This is the pre-`up` guard: a fresh `airstack up` must not race a
+        previous bring-up's containers/network (the baked entrypoint leaves a
+        stack up on a stale image; the prior iteration leaves one too).
+        Starting `up` while those still exist causes duplicate-network and
+        container-name conflicts."""
+        self._airstack("status", 60)
+        existing = stack_containers()
+        if not existing:
+            return
+        log(f"existing stack containers found {existing} — bringing them down first")
+        self.down()
+        deadline = time.time() + self.mission["down_timeout_s"]
+        while time.time() < deadline:
+            remaining = stack_containers()
+            if not remaining:
+                log("stack fully down; safe to bring up")
+                return
+            log(f"waiting for teardown: {remaining}")
+            time.sleep(2)
+        log(f"WARN: containers still present after down: {stack_containers()} "
+            f"— proceeding, `up` may conflict")
+
+    def wait_ready(self, container):
+        """Two sequential gates per robot (same as the system tests):
+        1. mavros/state reports connected=True (MAVROS ↔ PX4 heartbeat);
+        2. local_position/odom publishes (PX4 EKF converged — the actual
+           precondition for arming; `connected` alone fires ~25s too early).
+        Returns {robot_n: seconds_to_ready}; raises on timeout."""
+        cfg = self.mission["ready"]
+        started = time.time()
+        deadline = started + cfg["timeout_s"]
+        connected, ready_at = set(), {}
+        pending = list(range(1, self.num_robots + 1))
+
+        while pending and time.time() < deadline:
+            for n in list(pending):
+                if n not in connected:
+                    r = ros2_exec(container,
+                                  f"timeout 5 ros2 topic echo --once --csv "
+                                  f"--field connected /robot_{n}/interface/mavros/state",
+                                  domain_id=n, setup_bash=self.setup_bash, timeout=15)
+                    if any(line.strip() == "True" for line in r.stdout.splitlines()):
+                        connected.add(n)
+                    else:
+                        continue
+                r = ros2_exec(container,
+                              f"timeout 5 ros2 topic echo --once "
+                              f"/robot_{n}/interface/mavros/local_position/odom",
+                              domain_id=n, setup_bash=self.setup_bash, timeout=15)
+                if r.returncode == 0 and "pose:" in r.stdout:
+                    ready_at[n] = round(time.time() - started, 2)
+                    pending.remove(n)
+            if pending:
+                log(f"waiting for PX4: connected={sorted(connected)} pending={pending} "
+                    f"elapsed={time.time() - started:.0f}s")
+                time.sleep(cfg["poll_interval_s"])
+
+        if pending:
+            raise RuntimeError(f"robots {pending} not ready within {cfg['timeout_s']}s "
+                               f"(connected so far: {sorted(connected)})")
+        log(f"PX4 ready: {ready_at}")
+
+        # Gate 3 (only when actions route via the GCS): the per-robot
+        # action_relay nodes must be up on the GCS before goals are sent —
+        # the relay's goal subscription is volatile, so a goal published
+        # before the relay exists is silently lost.
+        if uses_gcs_route(self.mission):
+            while time.time() < deadline:
+                gcs = gcs_container()
+                if gcs:
+                    r = ros2_exec(gcs, "timeout 10 ros2 node list",
+                                  domain_id=0, setup_bash=GCS_SETUP_BASH, timeout=20)
+                    missing = [n for n in range(1, self.num_robots + 1)
+                               if f"action_relay_robot_{n}" not in r.stdout]
+                    if not missing:
+                        ready_at["gcs_relay"] = round(time.time() - started, 2)
+                        log(f"GCS action_relay ready ({ready_at['gcs_relay']}s)")
+                        break
+                    log(f"waiting for GCS action_relay: missing robots {missing}")
+                else:
+                    log("waiting for gcs container")
+                time.sleep(cfg["poll_interval_s"])
+            if "gcs_relay" not in ready_at:
+                raise RuntimeError(
+                    f"GCS action_relay not ready within {cfg['timeout_s']}s — "
+                    f"is the gcs service in stack.services/profiles and "
+                    f"AUTOLAUNCH=true? (or set command_route: robot)")
+        return ready_at
+
+
+# ── bag recording ──────────────────────────────────────────────────────────
+
+class Recorder:
+    """`ros2 bag record -s mcap` per the mission's record.scope.
+
+    scope "gcs" (default): a single recorder in the GCS container on domain
+    0, where every robot's domain_bridge forwards its state topics — one
+    mcap holds all robots, and Foxglove replay matches the live GCS view.
+
+    scope "robot": one recorder per robot inside robot container 1, each on
+    that robot's DDS domain — for topics that aren't bridged to the GCS.
+
+    scope "both": the GCS recorder and the per-robot recorders together —
+    one mcap with the fleet view from domain 0 plus one raw-data mcap per
+    robot domain.
+
+    Each recorder is started detached with its PID dropped to a file, and
+    stopped with SIGTERM so rosbag2 finalizes the mcap cleanly. (Not SIGINT:
+    jobs backgrounded from a non-interactive shell have SIGINT set to
+    SIG_IGN, so it would never be delivered; rosbag2 handles SIGTERM the
+    same way.) Bags are docker cp'd to the host before the stack goes down."""
+
+    def __init__(self, robot_container, mission, num_robots, setup_bash):
+        self.robot_container = robot_container
+        self.cfg = mission["record"]
+        self.scope = self.cfg.get("scope", "gcs")
+        self.num_robots = num_robots
+        self.setup_bash = setup_bash
+        # (container, domain_id, tag) per active recorder.
+        self.active = []
+
+    def _topic_selection(self, robots, scope):
+        """Build the `ros2 bag record` topic args; `robots` is the list of
+        robot indices whose {robot}/{n} placeholders to expand (unioned,
+        order-preserving, deduplicated). `scope` picks the default topic set
+        for the recorder being started ("gcs" or "robot")."""
+        if self.cfg.get("all"):
+            sel = "-a"
+            # record.exclude: regex of topics to drop from `-a`.
+            exclude = self.cfg.get("exclude")
+            if exclude:
+                sel += f" --exclude-regex {shlex.quote(exclude)}"
+            return sel
+        default = (DEFAULT_GCS_RECORD_TOPICS if scope == "gcs"
+                   else DEFAULT_ROBOT_RECORD_TOPICS)
+        topics = []
+        for t in self.cfg.get("topics", default):
+            if "{robot}" in t or "{n}" in t:
+                topics.extend(expand(t, n) for n in robots)
+            else:
+                topics.append(t)
+        seen = set()
+        unique = [t for t in topics if not (t in seen or seen.add(t))]
+        return " ".join(shlex.quote(t) for t in unique)
+
+    def _start_one(self, container, domain_id, tag, selection, setup_bash):
+        out_dir = f"{BAG_STAGING_DIR}/{tag}"
+        log_file = f"{BAG_STAGING_DIR}/record_{tag}.log"
+        pid_file = f"{BAG_STAGING_DIR}/record_{tag}.pid"
+        # Create the staging dir in the FOREGROUND, THEN background only the
+        # recorder. Folding `mkdir` into the same `... &` chain backgrounds the
+        # mkdir too, so the foreground pidfile write races ahead of the dir
+        # existing ("record_<tag>.pid: No such file or directory"). nohup +
+        # pidfile lets the recorder outlive this docker exec; topics that
+        # don't exist yet are fine (rosbag2 waits for them).
+        inner = (
+            f"mkdir -p {BAG_STAGING_DIR} && rm -rf {out_dir}\n"
+            f"{ros2_env_prefix(setup_bash, domain_id)}\n"
+            f"nohup ros2 bag record -s mcap -o {out_dir} {selection} "
+            f"> {log_file} 2>&1 &\n"
+            f"echo $! > {pid_file}\n"
+            # Confirm the recorder is actually alive — a bad topic name or a
+            # missing mcap plugin would make it exit immediately, and a silent
+            # recording failure is the worst outcome for a mission.
+            f"sleep 2\n"
+            f"if kill -0 \"$(cat {pid_file})\" 2>/dev/null; then echo RECORDER_ALIVE; "
+            f"else echo RECORDER_DEAD; tail -n 5 {log_file} 2>/dev/null; fi"
+        )
+        r = docker_exec(container, inner, timeout=30)
+        if "RECORDER_ALIVE" in r.stdout:
+            self.active.append((container, tag))
+            n_topics = ("all topics" if selection.startswith("-a")
+                        else f"{len(selection.split())} topics")
+            log(f"recording [{tag}] in {container} (domain {domain_id}) "
+                f"→ {out_dir} ({n_topics})")
+            return True
+        log(f"WARN: recorder [{tag}] failed to start / exited immediately: "
+            f"{tail(r.stdout + r.stderr, 6)}")
+        return False
+
+    def start(self):
+        if not self.cfg.get("enabled", True):
+            log("recording disabled by mission spec")
+            return
+        robots = list(range(1, self.num_robots + 1))
+        failed = []
+        if self.scope in ("gcs", "both"):
+            gcs = gcs_container()
+            if gcs:
+                if not self._start_one(gcs, 0, "gcs",
+                                       self._topic_selection(robots, "gcs"),
+                                       GCS_SETUP_BASH):
+                    failed.append("gcs")
+            else:
+                log("WARN: record.scope includes 'gcs' but no gcs container is "
+                    "running — GCS recording skipped (bring up the gcs service "
+                    "or use scope: robot)")
+                failed.append("gcs (no container)")
+                if self.scope == "gcs" and not self.cfg.get("required"):
+                    return
+        if self.scope in ("robot", "both"):
+            for n in robots:
+                if not self._start_one(self.robot_container, n, f"robot_{n}",
+                                       self._topic_selection([n], "robot"),
+                                       self.setup_bash):
+                    failed.append(f"robot_{n}")
+        # record.required: a mission whose deliverable is the bags shouldn't
+        # silently fly unrecorded.
+        if failed and self.cfg.get("required"):
+            raise RuntimeError(
+                f"record.required is set and recorder(s) failed to start: "
+                f"{failed} — aborting iteration")
+
+    def stop(self):
+        for container, tag in self.active:
+            docker_exec(container, f"""
+                pid=$(cat {BAG_STAGING_DIR}/record_{tag}.pid 2>/dev/null) || exit 0
+                kill -TERM "$pid" 2>/dev/null || exit 0
+                for i in $(seq 1 20); do
+                    kill -0 "$pid" 2>/dev/null || exit 0
+                    sleep 1
+                done
+                kill -9 "$pid" 2>/dev/null || true
+            """, timeout=40)
+        if self.active:
+            log(f"recorders stopped: {[tag for _, tag in self.active]}")
+
+    def collect(self, dest_dir):
+        if not self.active:
+            return
+        dest_dir.mkdir(parents=True, exist_ok=True)
+        for container in {c for c, _ in self.active}:
+            r = sh(["docker", "cp", f"{container}:{BAG_STAGING_DIR}/.", str(dest_dir)],
+                   timeout=1800)
+            if r.returncode != 0:
+                log(f"WARN: docker cp of bags from {container} failed: "
+                    f"{r.stderr.strip()[:200]}")
+        self.active = []
+        mcaps = list(dest_dir.rglob("*.mcap"))
+        log(f"collected {len(mcaps)} mcap file(s) → {dest_dir}")
+
+
+# ── step execution ─────────────────────────────────────────────────────────
+
+def action_ok(stdout):
+    """ros2 action send_goal --feedback prints the result as YAML;
+    AirStack task results carry `success: true` on completion."""
+    return "success: true" in stdout
+
+
+def tail(text, lines=15):
+    return "\n".join((text or "").strip().splitlines()[-lines:])
+
+
+def _run_one(stack, robot_container, target, cmd, timeout, expect_success):
+    """Run one resolved `run` command. `target` is already robot-expanded:
+      pod          → on the pod itself (cwd = AirStack root)
+      gcs          → in the gcs container on domain 0
+      robot_<n>    → in robot container 1 on robot N's DDS domain
+      <other>      → literal container name (domain 0)
+    """
+    if target == "pod":
+        r = sh(["bash", "-c", cmd], timeout=timeout, cwd=stack.root)
+    elif target == "gcs":
+        g = gcs_container()
+        if g:
+            r = ros2_exec(g, cmd, domain_id=0, setup_bash=GCS_SETUP_BASH,
+                          timeout=timeout + 15)
+        else:
+            r = subprocess.CompletedProcess(cmd, returncode=1, stdout="",
+                                            stderr="no gcs container running")
+    elif target.startswith("robot_") and target[6:].isdigit():
+        n = int(target[6:])
+        r = ros2_exec(robot_container, cmd, domain_id=n,
+                      setup_bash=stack.setup_bash, timeout=timeout + 15)
+    else:
+        r = docker_exec(target, cmd, timeout=timeout + 15)
+    ok = (r.returncode == 0) if expect_success else True
+    return {"target": target, "exit": r.returncode, "ok": ok,
+            "output_tail": tail(r.stdout + r.stderr)}
+
+
+def _run_one_retry(stack, robot_container, target, cmd, timeout, expect_success,
+                   attempts, retry_delay_s):
+    """_run_one with up to `attempts` tries (retry_delay_s between failures)."""
+    res = {"target": target, "exit": 1, "ok": False,
+           "output_tail": "not attempted — mission stop requested", "attempts": 0}
+    for attempt in range(1, max(1, attempts) + 1):
+        if STOP_EVENT.is_set():
+            break
+        res = _run_one(stack, robot_container, target, cmd, timeout, expect_success)
+        res["attempts"] = attempt
+        if res["ok"]:
+            break
+        if attempt < attempts and not STOP_EVENT.is_set():
+            log(f"run [{target}] failed (attempt {attempt}/{attempts}); "
+                f"retrying in {retry_delay_s}s")
+            time.sleep(retry_delay_s)
+    return res
+
+
+def run_step(stack, container, step_spec, step_index):
+    """Execute one step; returns a result dict with ok: bool."""
+    record = {"index": step_index, "spec": step_spec,
+              "started_at": datetime.now(timezone.utc).isoformat()}
+    t0 = time.time()
+
+    if "wait" in step_spec:
+        seconds = float(step_spec["wait"])
+        log(f"step {step_index}: wait {seconds}s")
+        time.sleep(seconds)
+        record.update(type="wait", ok=True)
+
+    elif "action" in step_spec:
+        spec = step_spec["action"]
+        task = spec["task"]
+        via = spec.get("via", stack.mission["command_route"])
+        goal = spec.get("goal", {})
+        # Normalize the goal to JSON: dicts directly; strings may be ros2-style
+        # YAML ("{target_altitude_m: 10}") — YAML-parse then re-dump. JSON is
+        # what the GCS relay requires, and it's a YAML subset so the direct
+        # send_goal path accepts it too.
+        goal_obj = yaml.safe_load(goal) if isinstance(goal, str) else goal
+        goal_json = json.dumps(goal_obj or {})
+        timeout = float(spec.get("timeout_s", 120))
+        # Per-robot retries for transient rejections (no GPS fix yet, PX4
+        # position not converged, goal lost on the relay's volatile sub).
+        max_attempts = int(spec.get("attempts", 3))
+        retry_delay_s = float(spec.get("retry_delay_s", 10))
+        # via: gcs only — no relay_feedback AND no result within this window
+        # after publishing ⇒ the goal is presumed lost; fail fast and retry.
+        feedback_timeout_s = int(spec.get("feedback_timeout_s", 15))
+        # cancel_on_timeout (via:gcs only): on `timeout_s`, publish a CancelGoal
+        # to the relay's .../cancel topic and wait up to cancel_grace_s for the
+        # cancelled result, so an on-cancel finalize (e.g. semantic_search's
+        # metrics) can run and return.
+        cancel_on_timeout = bool(spec.get("cancel_on_timeout", False))
+        cancel_grace_s = int(spec.get("cancel_grace_s", 120))
+        # pass_on_feedback (via:gcs only): treat the step as passed as soon as the
+        # action is confirmed to have actually started running — i.e. any
+        # relay_feedback was seen (or a result came back), regardless of whether
+        # the result is success or failure. The ONLY failure left is "goal lost /
+        # never ran" (no feedback and no result within the retries), which is what
+        # then fails the step (→ iteration redo when iteration_attempts>1). Use
+        # this when you only care that the task ran, not its outcome.
+        pass_on_feedback = bool(spec.get("pass_on_feedback", False))
+        robots = step_robots(spec, stack.num_robots)
+        log(f"step {step_index}: action {task} {goal_json} via {via} → robots {robots}")
+
+        if via == "gcs":
+            gcs = gcs_container()
+            if not gcs:
+                record.update(type="action", task=task, via=via, ok=False,
+                              error="no gcs container running — `via: gcs` needs the "
+                                    "gcs service up (or use `via: robot`)")
+                record["duration_s"] = round(time.time() - t0, 2)
+                log(f"step {step_index}: FAILED (no gcs container)")
+                return record
+
+            # The result echo must outlive `timeout` to catch the post-cancel
+            # result.
+            echo_timeout = int(timeout + cancel_grace_s) if cancel_on_timeout \
+                else int(timeout)
+
+            def send(n):
+                # Foxglove's path: String JSON on /<robot>/tasks/<task>/goal,
+                # result on .../relay_result. relay_result is latched, so a
+                # fresh subscriber gets the PREVIOUS goal's result — count
+                # messages before publishing and only accept a NEW one.
+                # relay_feedback is the liveness signal: nothing there (and
+                # no result) within feedback_timeout_s ⇒ goal lost, bail out
+                # so the retry loop re-sends.
+                base = f"/robot_{n}/tasks/{task}"
+                result_file = f"/tmp/relay_result_{task}_{n}.out"
+                fb_file = f"/tmp/relay_feedback_{task}_{n}.out"
+                msg_yaml = json.dumps({"data": expand(goal_json, n)})
+                cancel_yaml = json.dumps({"data": "osmo: timeout cancel"})
+                script = (
+                    f"rm -f {result_file} {fb_file}\n"
+                    f"touch {result_file} {fb_file}\n"
+                    f"( timeout {echo_timeout} ros2 topic echo --field data "
+                    f"{base}/relay_result >> {result_file} 2>&1 ) &\n"
+                    f"sub=$!\n"
+                    f"( timeout {echo_timeout} ros2 topic echo --field data "
+                    f"{base}/relay_feedback >> {fb_file} 2>&1 ) &\n"
+                    f"fb_sub=$!\n"
+                    f"sleep 3\n"
+                    f"pre=$(grep -c '^---' {result_file})\n"
+                    # -w 1: wait for the relay's (VOLATILE) goal sub to match
+                    # before publishing; --keep-alive 3: outlive the publish so
+                    # the RELIABLE handshake flushes (0.1s default drops goals).
+                    f"ros2 topic pub --once -w 1 --keep-alive 3 "
+                    f"{base}/goal std_msgs/msg/String "
+                    f"{shlex.quote(msg_yaml)} > /dev/null\n"
+                    f"sent=$(date +%s)\n"
+                    f"fb_seen=0\n"
+                    f"cancelled=0\n"
+                    f"deadline=$(( sent + {int(timeout)} ))\n"
+                    f"while :; do\n"
+                    f"  cur=$(grep -c '^---' {result_file})\n"
+                    f"  if [ \"$cur\" -gt \"$pre\" ]; then break; fi\n"
+                    f"  now=$(date +%s)\n"
+                    f"  if [ \"$fb_seen\" -eq 0 ] && "
+                    f"[ \"$(grep -c '^---' {fb_file})\" -gt 0 ]; then fb_seen=1; fi\n"
+                    f"  if [ \"$cancelled\" -eq 0 ] && [ \"$fb_seen\" -eq 0 ] && "
+                    f"[ $(( now - sent )) -ge {feedback_timeout_s} ]; then\n"
+                    f"    kill $sub $fb_sub 2>/dev/null\n"
+                    f"    echo 'no relay_feedback within {feedback_timeout_s}s "
+                    f"of goal publish — goal presumed lost'\n"
+                    f"    exit 0\n"
+                    f"  fi\n"
+                    f"  if [ $now -ge $deadline ]; then\n"
+                    # Deadline: cancel + keep waiting if cancel_on_timeout, else stop.
+                    f"    if [ {1 if cancel_on_timeout else 0} -eq 1 ] && "
+                    f"[ \"$cancelled\" -eq 0 ]; then\n"
+                    f"      echo 'osmo: {int(timeout)}s timeout — sending cancel'\n"
+                    f"      ros2 topic pub --once -w 1 --keep-alive 3 "
+                    f"{base}/cancel std_msgs/msg/String "
+                    f"{shlex.quote(cancel_yaml)} > /dev/null\n"
+                    f"      cancelled=1\n"
+                    f"      deadline=$(( now + {cancel_grace_s} ))\n"
+                    f"    else\n"
+                    f"      break\n"
+                    f"    fi\n"
+                    f"  fi\n"
+                    f"  kill -0 $sub 2>/dev/null || break\n"
+                    f"  sleep 1\n"
+                    f"done\n"
+                    f"kill $sub $fb_sub 2>/dev/null\n"
+                    # Surface whether the action ever produced feedback (it ran),
+                    # so the runner can pass on liveness when pass_on_feedback is
+                    # set. The early no-feedback bail above exits before here, so
+                    # this only prints when feedback was genuinely observed.
+                    f"if [ \"$fb_seen\" -eq 1 ]; then echo OSMO_FEEDBACK_SEEN; fi\n"
+                    f"cur=$(grep -c '^---' {result_file})\n"
+                    f"if [ \"$cur\" -gt \"$pre\" ]; then "
+                    f"grep -v '^---' {result_file} | tail -n 1; "
+                    f"else echo 'no relay_result within {echo_timeout}s'; fi"
+                )
+                r = ros2_exec(gcs, script, domain_id=0, setup_bash=GCS_SETUP_BASH,
+                              timeout=int(echo_timeout + 30))
+                # For cancel_on_timeout, a returned (success=false) cancel result
+                # is the intended outcome — treat it as ok so the step doesn't
+                # retry the search or trip on_step_failure.
+                got_result = '"success"' in r.stdout
+                feedback_seen = 'OSMO_FEEDBACK_SEEN' in r.stdout
+                ok = ('"success": true' in r.stdout
+                      or (cancel_on_timeout and got_result))
+                # pass_on_feedback: ran-at-all (feedback seen, or any result
+                # came back) is enough — outcome doesn't matter.
+                if pass_on_feedback:
+                    ok = ok or feedback_seen or got_result
+                return {"exit": r.returncode, "ok": ok,
+                        "feedback_seen": feedback_seen,
+                        "output_tail": tail(r.stdout + r.stderr)}
+        else:
+            action_type = spec.get("type", task_action_type(task))
+
+            def send(n):
+                cmd = (f"ros2 action send_goal --feedback /robot_{n}/tasks/{task} "
+                       f"{action_type} {shlex.quote(expand(goal_json, n))}")
+                r = ros2_exec(container, cmd, domain_id=n, setup_bash=stack.setup_bash,
+                              timeout=int(timeout + 15))
+                return {"exit": r.returncode, "ok": action_ok(r.stdout),
+                        "output_tail": tail(r.stdout + r.stderr)}
+
+        def send_with_retry(n):
+            res = {"ok": False, "exit": -1,
+                   "output_tail": "not attempted — mission stop requested"}
+            attempt = 0
+            for attempt in range(1, max_attempts + 1):
+                if STOP_EVENT.is_set():
+                    break
+                log(f"step {step_index}: sending {task} to robot_{n} "
+                    f"(attempt {attempt}/{max_attempts})")
+                res = send(n)
+                if res["ok"]:
+                    log(f"step {step_index}: {task} robot_{n} succeeded"
+                        + (f" on attempt {attempt}" if attempt > 1 else ""))
+                    break
+                log(f"step {step_index}: {task} robot_{n} attempt {attempt} "
+                    f"FAILED: {tail(res.get('output_tail', ''), 1)}")
+                if attempt < max_attempts and not STOP_EVENT.is_set():
+                    time.sleep(retry_delay_s)
+            res["attempts"] = attempt
+            return n, res
+
+        # No context manager: on MissionStop the `with` form would block in
+        # shutdown(wait=True) until the in-flight docker execs hit their
+        # timeouts. The workers die when `airstack down` kills the containers.
+        pool = ThreadPoolExecutor(max_workers=len(robots))
+        try:
+            results = dict(pool.map(send_with_retry, robots))
+        finally:
+            pool.shutdown(wait=False, cancel_futures=True)
+        record.update(type="action", task=task, via=via, per_robot=results,
+                      ok=all(v["ok"] for v in results.values()))
+
+    elif "run" in step_spec:
+        spec = step_spec["run"]
+        cmd = spec["cmd"]
+        timeout = float(spec.get("timeout_s", 60))
+        expect = spec.get("expect_success", True)
+        attempts = int(spec.get("attempts", 1))
+        retry_delay_s = float(spec.get("retry_delay_s", 10))
+        # `container` may reference {n}/{robot} to fan out over robots. If it's
+        # omitted, default to robot_{n} when the command is per-robot (has a
+        # placeholder) and robot_1 otherwise. A step fans out over `robots`
+        # (default all) iff its command or container references {n}/{robot}.
+        target = spec.get("container") or ("robot_{n}" if has_placeholder(cmd) else "robot_1")
+        if has_placeholder(cmd, target):
+            robots = step_robots(spec, stack.num_robots)
+            log(f"step {step_index}: run [{target}] {cmd} → robots {robots}")
+            results = {}
+            for n in robots:
+                results[n] = _run_one_retry(stack, container, expand(target, n),
+                                            expand(cmd, n), timeout, expect,
+                                            attempts, retry_delay_s)
+            record.update(type="run", per_robot=results,
+                          ok=all(v["ok"] for v in results.values()))
+        else:
+            log(f"step {step_index}: run [{target}] {cmd}")
+            res = _run_one_retry(stack, container, target, cmd, timeout, expect,
+                                 attempts, retry_delay_s)
+            record.update(type="run", **res)
+
+    elif "topic_pub" in step_spec:
+        spec = step_spec["topic_pub"]
+        msg = spec.get("msg", {})
+        msg_str = msg if isinstance(msg, str) else json.dumps(msg)
+        robots = step_robots(spec, stack.num_robots)
+        log(f"step {step_index}: topic_pub {spec['topic']} → robots {robots}")
+        results = {}
+        for n in robots:
+            cmd = (f"ros2 topic pub --once {expand(spec['topic'], n)} "
+                   f"{spec['type']} {shlex.quote(expand(msg_str, n))}")
+            r = ros2_exec(container, cmd, domain_id=n, setup_bash=stack.setup_bash,
+                          timeout=float(spec.get("timeout_s", 30)) + 15)
+            results[n] = {"exit": r.returncode, "ok": r.returncode == 0,
+                          "output_tail": tail(r.stdout + r.stderr)}
+        record.update(type="topic_pub", per_robot=results,
+                      ok=all(v["ok"] for v in results.values()))
+
+    elif "service_call" in step_spec:
+        spec = step_spec["service_call"]
+        req = spec.get("request", {})
+        req_str = req if isinstance(req, str) else json.dumps(req)
+        robots = step_robots(spec, stack.num_robots)
+        log(f"step {step_index}: service_call {spec['service']} → robots {robots}")
+        results = {}
+        for n in robots:
+            cmd = (f"ros2 service call {expand(spec['service'], n)} "
+                   f"{spec['type']} {shlex.quote(expand(req_str, n))}")
+            r = ros2_exec(container, cmd, domain_id=n, setup_bash=stack.setup_bash,
+                          timeout=float(spec.get("timeout_s", 30)) + 15)
+            results[n] = {"exit": r.returncode, "ok": r.returncode == 0,
+                          "output_tail": tail(r.stdout + r.stderr)}
+        record.update(type="service_call", per_robot=results,
+                      ok=all(v["ok"] for v in results.values()))
+
+    else:
+        record.update(type="unknown", ok=False,
+                      error=f"unrecognized step keys: {sorted(step_spec)}")
+
+    record["duration_s"] = round(time.time() - t0, 2)
+    log(f"step {step_index}: {'OK' if record['ok'] else 'FAILED'} "
+        f"({record['duration_s']}s)")
+    return record
+
+
+# ── artifacts ──────────────────────────────────────────────────────────────
+
+# Navigation / control-stack node-name globs. These log via rcl to ~/.ros/log
+# (not /tmp), so they're collected separately — they're what reveals why a
+# planner's plan isn't becoming motion (e.g. a robot stuck mid-mission). Task
+# nodes are here too so their rcl log is captured as a SECOND path independent
+# of the /tmp tee (TEE_LOG_GLOBS) — if one collection path misses, the other
+# still shows why the drone wasn't commanded.
+CONTROL_STACK_GLOBS = ("droan_gl*", "droan_local_planner*", "trajectory_controller*",
+                       "fixed_trajectory*", "takeoff_landing*", "mavros*",
+                       "semantic_search_task*")
+
+
+# /tmp tee logs written by task-spawned subprocesses and the GCS
+# relay/gossip/ddsrouter — the stdout that rcl logging doesn't capture.
+TEE_LOG_GLOBS = ("/tmp/gossip_*.log", "/tmp/ddsrouter_*.log", "/tmp/relay_*.log")
+
+
+def _copy_tmp_tee_logs(dest_dir):
+    """Copy the raw /tmp tee logs from each container into <dest_dir>/<container>/.
+    Idempotent (docker cp overwrites), so it is safe to call repeatedly — the
+    snapshot loop calls it periodically so a hard pod cancel mid-iteration still
+    leaves the most recent stdout tees on disk."""
+    targets = list(robot_containers())
+    gcs = gcs_container()
+    if gcs:
+        targets.append(gcs)
+    total = 0
+    for name in targets:
+        r = docker_exec(name, "ls " + " ".join(TEE_LOG_GLOBS) + " 2>/dev/null",
+                        timeout=15)
+        files = [f for f in r.stdout.split() if f]
+        if not files:
+            continue
+        out = dest_dir / name
+        out.mkdir(parents=True, exist_ok=True)
+        for f in files:
+            sh(["docker", "cp", f"{name}:{f}", str(out)], timeout=120)
+        total += len(files)
+    return total
+
+
+def snapshot_task_logs(dest_dir):
+    """Copy the raw gossip/ddsrouter/relay tees (written to /tmp inside each
+    container) into <dest_dir>/<container>/, plus the navigation /
+    control-stack node logs that rcl writes under ~/.ros/log rather than /tmp.
+    Includes the GCS container so the GCS-side gossip ddsrouter log is captured."""
+    targets = list(robot_containers())
+    gcs = gcs_container()
+    if gcs:
+        targets.append(gcs)
+    _copy_tmp_tee_logs(dest_dir)
+    name_pred = " -o ".join(f"-name '{g}'" for g in CONTROL_STACK_GLOBS)
+    tar = "/tmp/control_stack_logs.tar.gz"
+    for name in targets:
+        r = docker_exec(name, "ls " + " ".join(TEE_LOG_GLOBS) + " 2>/dev/null",
+                        timeout=15)
+        files = [f for f in r.stdout.split() if f]
+        out = dest_dir / name
+        made = docker_exec(
+            name,
+            f"f=$(find /root/.ros/log -type f \\( {name_pred} \\) 2>/dev/null); "
+            f"[ -n \"$f\" ] && tar czf {tar} $f 2>/dev/null && echo COLLECTED",
+            timeout=30)
+        if "COLLECTED" in (made.stdout or ""):
+            out.mkdir(parents=True, exist_ok=True)
+            sh(["docker", "cp", f"{name}:{tar}",
+                str(out / "control_stack_logs.tar.gz")], timeout=120)
+            files = files + ["control_stack_logs.tar.gz"]
+        if files:
+            log(f"collected {len(files)} node log(s) from {name}")
+
+
+def snapshot_container_logs(dest_dir):
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    pats = ("airstack", "isaac-sim", "ms-airsim")
+    names = [n for n in list_containers(all_states=True)
+             if any(p in n for p in pats)]
+    for name in names:
+        r = sh(["docker", "logs", name], timeout=120)
+        (dest_dir / f"{name}.log").write_text(
+            (r.stdout or "") + (("\n--- stderr ---\n" + r.stderr) if r.stderr else ""),
+            encoding="utf-8")
+
+
+def write_json(path, data):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(data, indent=2, default=str) + "\n", encoding="utf-8")
+
+
+def resolve_results_root(airstack_root):
+    """Prefer /osmo/output (auto-uploaded by OSMO `outputs:` when the task
+    exits); keep <airstack>/osmo/results working in both cases so
+    `airstack osmo:fetch` has one stable path to rsync."""
+    fallback = Path(airstack_root) / "osmo" / "results"
+    override = os.environ.get("OSMO_RESULTS_ROOT")
+    if override:
+        root = Path(override)
+        root.mkdir(parents=True, exist_ok=True)
+        return root
+    osmo_output = Path(os.environ.get("OSMO_OUTPUT_DIR", "/osmo/output"))
+    if osmo_output.is_dir() and os.access(osmo_output, os.W_OK):
+        root = osmo_output / "airstack-mission-results"
+        root.mkdir(parents=True, exist_ok=True)
+        if not fallback.exists():
+            fallback.parent.mkdir(parents=True, exist_ok=True)
+            fallback.symlink_to(root)
+        return root
+    fallback.mkdir(parents=True, exist_ok=True)
+    return fallback
+
+
+# ── main loop ──────────────────────────────────────────────────────────────
+
+# Per-container health snapshot. Sections are emitted on tagged lines so the
+# multi-line ps output stays parseable. cgroup paths are v2; absent on v1 (the
+# fields just come back empty).
+_SNAP_CMD = r"""
+echo "T|$(date '+%F %T')"
+echo "DDS|procs=$(pgrep -c ddsrouter 2>/dev/null) estab=$(ss -Htn 2>/dev/null | grep -c ESTAB) ddstcp=$(ss -Htnp 2>/dev/null | grep -c ddsrouter) cpu=$(ps -C ddsrouter -o %cpu= 2>/dev/null | tr '\n' ',')"
+echo "MEM|cur=$(cat /sys/fs/cgroup/memory.current 2>/dev/null) max=$(cat /sys/fs/cgroup/memory.max 2>/dev/null) oom_kill=$(awk '/^oom_kill /{print $2}' /sys/fs/cgroup/memory.events 2>/dev/null)"
+echo "PS|"
+ps -eo rss=,args= --sort=-rss 2>/dev/null | head -12
+"""
+
+# Processes whose early death we want to catch. Matched as substrings of the
+# full command line (python nodes all run as `python3 ...`). Extend per-mission
+# with any long-running node whose silent crash would otherwise only show up as
+# a stuck drone.
+_WATCH_PROCS = ("ddsrouter", "gossip")
+
+
+def _parse_snapshot(out):
+    ts = dds = mem = ""
+    oom = None
+    ps_rows = []
+    in_ps = False
+    for ln in out.splitlines():
+        if ln.startswith("T|"):
+            ts = ln[2:].strip()
+        elif ln.startswith("DDS|"):
+            dds = ln[4:].strip()
+        elif ln.startswith("MEM|"):
+            mem = ln[4:].strip()
+            m = re.search(r"oom_kill=(\d+)", mem)
+            oom = int(m.group(1)) if m else None
+        elif ln.startswith("PS|"):
+            in_ps = True
+        elif in_ps:
+            p = ln.split(None, 1)
+            if len(p) == 2 and p[0].isdigit():
+                ps_rows.append((int(p[0]), p[1].strip()))
+    return ts, dds, mem, oom, ps_rows
+
+
+def _format_resources(mem, ps_rows):
+    def mb(v):
+        return f"{int(v) // 1048576}MB" if v.isdigit() else (v or "?")
+    m = re.search(r"cur=(\S+) max=(\S+)", mem)
+    memstr = f"mem={mb(m.group(1))}/{mb(m.group(2))}" if m else mem
+    oom = re.search(r"oom_kill=\S+", mem)
+    key = {k: 0 for k in _WATCH_PROCS}
+    for rss, args in ps_rows:
+        for k in _WATCH_PROCS:
+            if k in args:
+                key[k] += rss
+    keystr = " ".join(f"{k}={v // 1024}MB" for k, v in key.items())
+    top = []
+    for rss, args in ps_rows[:5]:
+        label = next((k for k in _WATCH_PROCS if k in args),
+                     args.split()[0].rsplit("/", 1)[-1][:16])
+        top.append(f"{label}:{rss // 1024}MB")
+    return f"{memstr} {oom.group(0) if oom else ''} | {keystr} | top: {' '.join(top)}", key
+
+
+def _capture_oom_dmesg(iter_dir, container, ts, why):
+    """Append the kernel OOM-killer trail (host ring buffer, readable from the
+    privileged container) when a watched process dies or oom_kill increments."""
+    r = docker_exec(
+        container,
+        "dmesg -T 2>/dev/null | grep -iE 'killed process|out of memory|oom-kill|"
+        "invoked oom|memory cgroup' | tail -40", timeout=15)
+    txt = (r.stdout or "").strip()
+    block = (f"### {container} @ {ts} — {why}\n"
+             f"{txt or '(dmesg unavailable in container — run host: dmesg -T | grep -i oom)'}\n")
+    with open(iter_dir / "oom_dmesg.log", "a", encoding="utf-8") as f:
+        f.write(block)
+
+
+def transport_snapshot_loop(iter_dir, stop_event, interval_s=5):
+    """Periodic per-container health snapshot for debugging early process death:
+      - transport.log : ddsrouter procs + domain-99 TCP (gossip dropout)
+      - resources.log : cgroup memory, oom_kill counter, watched + heaviest procs
+      - oom_dmesg.log : kernel OOM trail, captured when a watched process
+        vanishes or the cgroup oom_kill counter increments
+    """
+    iter_dir.mkdir(parents=True, exist_ok=True)
+    tlog, rlog = iter_dir / "transport.log", iter_dir / "resources.log"
+    prev_oom, prev_key = {}, {}
+    # Periodically copy the /tmp tee logs into iter_dir/logs so a hard pod cancel
+    # (SIGKILL — run_iteration's finally never runs) still leaves recent
+    # stdout tees. ~every 30s regardless of the health interval.
+    tee_every = max(1, round(30 / interval_s))
+    tick = 0
+    while not stop_event.is_set():
+        targets = list(robot_containers())
+        gcs = gcs_container()
+        if gcs:
+            targets.append(gcs)
+        tlines, rlines = [], []
+        for name in targets:
+            r = docker_exec(name, _SNAP_CMD, timeout=15)
+            ts, dds, mem, oom, ps_rows = _parse_snapshot(r.stdout or "")
+            if not ts:
+                continue
+            tlines.append(f"{name} [{ts}] {dds}")
+            res, key = _format_resources(mem, ps_rows)
+            rlines.append(f"{name} [{ts}] {res}")
+            if oom is not None:
+                if prev_oom.get(name, oom) < oom:
+                    _capture_oom_dmesg(iter_dir, name, ts, f"cgroup oom_kill -> {oom}")
+                prev_oom[name] = oom
+            if ps_rows:   # only trust presence when we got a real ps sample
+                for k in _WATCH_PROCS:
+                    if prev_key.get((name, k), 0) > 0 and key[k] == 0:
+                        _capture_oom_dmesg(iter_dir, name, ts, f"{k} vanished")
+                    prev_key[(name, k)] = key[k]
+        if tlines:
+            with open(tlog, "a", encoding="utf-8") as f:
+                f.write("\n".join(tlines) + "\n")
+        if rlines:
+            with open(rlog, "a", encoding="utf-8") as f:
+                f.write("\n".join(rlines) + "\n")
+        if tick % tee_every == 0:
+            try:
+                _copy_tmp_tee_logs(iter_dir / "logs")
+            except Exception as e:
+                log(f"WARN: periodic tee-log copy failed: {e}")
+        tick += 1
+        stop_event.wait(interval_s)
+
+
+def run_iteration(stack, mission, iter_dir):
+    """One full up → ready → record → steps → collect → down cycle.
+    Returns the iteration summary dict; never raises (failures are data)."""
+    summary = {"status": "passed", "steps_ok": 0, "steps_failed": 0}
+    recorder = None
+    container = None
+    snap_stop = None
+    t0 = time.time()
+    try:
+        # `airstack status`; if a stack is already up (baked entrypoint on a
+        # stale image, or the previous iteration), down it and wait for the
+        # containers to fully disappear before bringing up — otherwise the
+        # `up` races leftover containers/network.
+        stack.ensure_down()
+        containers = stack.up()
+        container = containers[0]
+        summary["up_duration_s"] = round(time.time() - t0, 2)
+
+        snap_stop = threading.Event()
+        threading.Thread(target=transport_snapshot_loop,
+                         args=(iter_dir, snap_stop), daemon=True).start()
+
+        ready_at = stack.wait_ready(container)
+        write_json(iter_dir / "ready.json", ready_at)
+
+        recorder = Recorder(container, mission, stack.num_robots, stack.setup_bash)
+        recorder.start()
+
+        steps = []
+        for i, step_spec in enumerate(mission["steps"], start=1):
+            record = run_step(stack, container, step_spec, i)
+            steps.append(record)
+            if record["ok"]:
+                summary["steps_ok"] += 1
+                continue
+            summary["steps_failed"] += 1
+            # An `optional: true` step records its failure but never fails the
+            # iteration (so it doesn't trip on_step_failure or an iteration redo)
+            # — e.g. a post-search land after the gating step already ran.
+            if isinstance(step_spec, dict) and step_spec.get("optional"):
+                log(f"step {i}: failed but marked optional — continuing")
+                summary.setdefault("optional_failures", []).append(i)
+                continue
+            summary["status"] = "failed"
+            policy = mission["on_step_failure"]
+            if policy == "continue":
+                continue
+            if policy == "abort_mission":
+                summary["abort_mission"] = True
+            break
+        write_json(iter_dir / "steps.json", steps)
+
+    except MissionStop as e:
+        summary["status"] = "stopped"
+        summary["abort_mission"] = True
+        log(f"STOP: {e} — finalizing recordings and collecting artifacts")
+
+    except Exception as e:
+        summary["status"] = "error"
+        summary["error"] = str(e)
+        log(f"ERROR: iteration aborted: {e}")
+
+    finally:
+        if snap_stop is not None:
+            snap_stop.set()
+        # Artifact collection happens even on failure — a failed flight's
+        # bag is usually the most interesting one.
+        if recorder is not None:
+            recorder.stop()
+            recorder.collect(iter_dir / "bags")
+        if container is not None or list_containers("airstack", all_states=True):
+            snapshot_container_logs(iter_dir / "logs")
+            snapshot_task_logs(iter_dir / "logs")
+        stack.down()
+        summary["duration_s"] = round(time.time() - t0, 2)
+        write_json(iter_dir / "iteration.json", summary)
+    return summary
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("mission_file", help="Path to mission spec YAML")
+    parser.add_argument("--airstack-root",
+                        default=os.environ.get("AIRSTACK_ROOT", "/root/AirStack"))
+    parser.add_argument("--dry-run", action="store_true",
+                        help="Validate the spec and print the plan without touching docker")
+    args = parser.parse_args()
+
+    mission = load_mission(args.mission_file)
+    stack = Stack(args.airstack_root, mission)
+
+    # Graceful stop: SIGINT/SIGTERM (Ctrl-C locally, `airstack osmo:stop` on
+    # a pod) aborts the current step but still finalizes the mcaps, collects
+    # bags + logs, and brings the stack down.
+    signal.signal(signal.SIGINT, _stop_handler)
+    signal.signal(signal.SIGTERM, _stop_handler)
+
+    log(f"mission '{mission['name']}': {mission['iterations']} iteration(s), "
+        f"{len(mission['steps'])} step(s), {stack.num_robots} robot(s)"
+        + (f", up to {mission['iteration_attempts']} attempt(s)/iteration"
+           if mission["iteration_attempts"] > 1 else ""))
+    if args.dry_run:
+        print(yaml.safe_dump(mission, sort_keys=False))
+        return 0
+
+    results_root = resolve_results_root(args.airstack_root)
+    run_dir = results_root / mission["name"] / utc_stamp()
+    run_dir.mkdir(parents=True)
+    log(f"results → {run_dir}")
+
+    environments = mission.get("environments") or []
+    n_iter = mission["iterations"]
+    # Build the per-iteration environment schedule. round_robin: cycle envs,
+    # `iterations` is the total. grouped: run each env `iterations` times in a
+    # row, so the total is iterations * len(environments).
+    if not environments:
+        schedule = [None] * n_iter
+    elif mission["environment_order"] == "grouped":
+        schedule = [env for env in environments for _ in range(n_iter)]
+    else:
+        schedule = [environments[i % len(environments)] for i in range(n_iter)]
+    total = len(schedule)
+    if environments:
+        log(f"{mission['environment_order']} over {len(environments)} "
+            f"environment(s) → {total} iteration(s): "
+            f"{[e['name'] for e in schedule]}")
+
+    iterations = []
+    max_attempts = mission["iteration_attempts"]
+    for i, env_entry in enumerate(schedule, start=1):
+        log(f"━━━ iteration {i}/{total} ━━━")
+        method = resolve_method(mission, env_entry)
+        iter_dir = run_dir / iter_dir_name(i, env_entry, method)
+        # Apply this iteration's environment (scene URL + spawn area as env vars
+        # for `airstack up`; search_area et al. templated into steps).
+        if env_entry is not None:
+            log(f"environment: {env_entry['name']} "
+                f"(ENV_URL={env_entry.get('ENV_URL')})")
+            stack.apply_env(env_exports(env_entry))
+            iter_mission = {**mission, "steps": substitute_env(mission["steps"], env_entry)}
+        else:
+            iter_mission = mission
+
+        # Iteration redo: re-run the SAME iteration up to iteration_attempts
+        # times until it passes. A passed run, a manual stop, or an explicit
+        # abort_mission is never retried; a failed/errored run is (the stack is
+        # already down by run_iteration's finally, so each attempt is a clean
+        # down→up→fly→down cycle — and with a fixed SPAWN_SEED the spawn layout
+        # is reproduced exactly).
+        attempt = 0
+        while True:
+            attempt += 1
+            if attempt > 1:
+                log(f"iteration {i}: attempt {attempt}/{max_attempts}")
+            summary = run_iteration(stack, iter_mission, iter_dir)
+            done = (summary["status"] in ("passed", "stopped")
+                    or summary.get("abort_mission")
+                    or attempt >= max_attempts)
+            if done:
+                break
+            # Failed with retries left: stash this attempt's artifacts so the
+            # canonical iter_NNN ends up holding the final attempt.
+            failed_dir = run_dir / f"{iter_dir.name}_failed_attempt_{attempt}"
+            try:
+                if iter_dir.exists():
+                    iter_dir.rename(failed_dir)
+                log(f"iteration {i} {summary['status']} on attempt {attempt}"
+                    f"/{max_attempts}; artifacts → {failed_dir.name}; redoing")
+            except OSError as e:
+                log(f"WARN: could not stash failed attempt dir ({e}); "
+                    f"redoing into {iter_dir.name}")
+
+        summary["iteration"] = i
+        summary["attempts"] = attempt
+        summary["method"] = method
+        if env_entry is not None:
+            summary["environment"] = env_entry["name"]
+        iterations.append(summary)
+        write_json(run_dir / "summary.json",
+                   {"mission": mission["name"], "mission_file": args.mission_file,
+                    "iterations": iterations})
+        if summary.get("abort_mission"):
+            log("stopping remaining iterations"
+                + (" (manual stop)" if summary["status"] == "stopped" else
+                   " (on_step_failure=abort_mission)"))
+            break
+
+    passed = sum(1 for s in iterations if s["status"] == "passed")
+    log(f"mission complete: {passed}/{len(iterations)} iteration(s) passed; "
+        f"results in {run_dir}")
+    return 0 if passed == len(iterations) else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/simulation/isaac-sim/extensions/PegasusSimulator b/simulation/isaac-sim/extensions/PegasusSimulator
index fe8b5a101..c94a4fb1f 160000
--- a/simulation/isaac-sim/extensions/PegasusSimulator
+++ b/simulation/isaac-sim/extensions/PegasusSimulator
@@ -1 +1 @@
-Subproject commit fe8b5a101857f2cda290b9b677b3a95c4cca6b09
+Subproject commit c94a4fb1f1a2abff03e28f6d42ee47ca705d10cb