herdr phase 1b: fleet_herdr dispatch module + node daemon ops

The second-runtime path uses the existing NodeHub control channel —
NOT SSH. Node daemons already accept typed ops over their outbound
websocket; adding three herdr_* ops keeps everything on the auth
model that already works fleet-wide (control-channel token, no new
SSH key management, no server-container-mounted keys).

Node daemon (clawmates-node):
  - New herdr_op handler in main.rs dispatching:
    * herdr_dispatch  — workspace create + pane split + rename + run
    * herdr_status    — pane get JSON (agent, agent_status, cwd)
    * herdr_read      — recent-unwrapped scrollback, N lines
  - Herdr binary resolved from ~/.local/bin, brew, /usr/local/bin.
    Missing binary returns clean error so cm-api can distinguish
    "node not set up for Herdr yet" from "Herdr op failed".

cm-api:
  - crates/cm-api/src/fleet_herdr.rs — dispatch / status /
    read_transcript / wait_for_completion helpers on top of
    hub.call_timeout(). wait_for_completion polls until agent_status
    hits 'done' or an idle-after-working state, matching the SKILL
    file's "either idle or done is completed" semantic.
  - routes::missions::herdr_dispatch — POST /api/missions/{id}/
    herdr-dispatch { cli, prompt }. Requires runtime_kind = 'local_herdr'
    and target_node_id set. Manual trigger so Phase 1b is exercisable
    end-to-end before Phase 1c wires the wizard + orchestrator.

Not yet wired: mission_orchestrator::on_launch still ignores
runtime_kind. Phase 1c adds the wizard picker AND the on_launch
branch that auto-dispatches on draft→running for local_herdr
missions. This commit only adds the primitives.

Verified: SQLX_OFFLINE=true cargo check --workspace green.
Phase 0 (Herdr install on fleet nodes) is the blocker to actually
exercising this end-to-end.

crates/bins/clawmates-node/src/main.rs

@@ -503,6 +503,20 @@ async fn handle_frame(
                 });
             }
         }
+        // Herdr dispatch ops. Server sends `herdr_dispatch` to open a
+        // sibling pane on the node's Herdr session and start the requested
+        // CLI (claude / codex / kimi / etc.) with a prompt. `herdr_status`
+        // polls that pane's agent_status; `herdr_read` scrapes its recent
+        // transcript. Node just shells out to the `herdr` binary — the
+        // Herdr background daemon is expected to already be running.
+        op @ ("herdr_dispatch" | "herdr_status" | "herdr_read") => {
+            if let Some(id) = v.get("id").and_then(Value::as_u64) {
+                let (ok, output) = herdr_op(op, &v).await;
+                let _ = out.send(
+                    json!({ "t": "result", "id": id, "ok": ok, "output": output }).to_string(),
+                );
+            }
+        }
         // Agent-sandbox container ops: drive the REAL DockerDriver so the
         // hardening (cap-drop ALL, seccomp, no-net, read-only, non-root) is
         // byte-identical to the gateway's local sandboxes.
@@ -839,6 +853,154 @@ fn handle_of(v: &Value) -> SandboxHandle {
 
 /// Run an agent-sandbox container op via the local DockerDriver (full hardening),
 /// returning the result payload as JSON or an error message.
+/// Herdr control ops. Requires `herdr` in PATH and a running background
+/// session (Phase 0 install). Returns raw JSON strings from the herdr
+/// CLI so the server can parse pane_id / agent_status without a
+/// second RPC hop.
+///
+/// Ops:
+///   herdr_dispatch { mission_id, cli, prompt, direction? } →
+///       runs `herdr pane split ... && herdr pane run ... "prompt"`
+///       output = the split's JSON response so the server can extract
+///       result.pane.pane_id
+///   herdr_status { pane_id } → `herdr pane get <pane_id>` JSON
+///   herdr_read   { pane_id, lines? } → recent-unwrapped scrollback
+async fn herdr_op(op: &str, v: &Value) -> (bool, String) {
+    let herdr = match std::env::var("HOME").ok().and_then(|h| {
+        [
+            format!("{h}/.local/bin/herdr"),
+            "/opt/homebrew/bin/herdr".to_string(),
+            "/usr/local/bin/herdr".to_string(),
+        ]
+        .into_iter()
+        .find(|p| std::path::Path::new(p).exists())
+    }) {
+        Some(p) => p,
+        None => return (false, "herdr binary not found on PATH".into()),
+    };
+    let herdr = std::sync::Arc::new(herdr);
+    let run = move |args: Vec<String>| {
+        let herdr = herdr.clone();
+        async move {
+            let fut = tokio::process::Command::new(herdr.as_str())
+                .args(&args)
+                .output();
+            match tokio::time::timeout(Duration::from_secs(30), fut).await {
+                Ok(Ok(o)) => {
+                    let mut s = String::from_utf8_lossy(&o.stdout).into_owned();
+                    if !o.status.success() {
+                        s.push_str(&String::from_utf8_lossy(&o.stderr));
+                    }
+                    (o.status.success(), s)
+                }
+                Ok(Err(e)) => (false, format!("spawn error: {e}")),
+                Err(_) => (false, "herdr op timed out".into()),
+            }
+        }
+    };
+    match op {
+        "herdr_dispatch" => {
+            let mission_id = v
+                .get("mission_id")
+                .and_then(Value::as_str)
+                .unwrap_or("unknown");
+            let cli = v.get("cli").and_then(Value::as_str).unwrap_or("claude");
+            let prompt = v.get("prompt").and_then(Value::as_str).unwrap_or("");
+            let direction = v
+                .get("direction")
+                .and_then(Value::as_str)
+                .unwrap_or("right");
+            // 1. Ensure a mission workspace exists (idempotent — label collision falls through).
+            let _ = run(vec![
+                "workspace".into(),
+                "create".into(),
+                "--label".into(),
+                format!("mission-{mission_id}"),
+            ])
+            .await;
+            // 2. Split off a fresh pane in that workspace and read its pane_id.
+            let (ok, split_out) = run(vec![
+                "pane".into(),
+                "split".into(),
+                "--direction".into(),
+                direction.into(),
+                "--no-focus".into(),
+            ])
+            .await;
+            if !ok {
+                return (false, format!("split failed: {split_out}"));
+            }
+            let pane_id = match serde_json::from_str::<Value>(&split_out) {
+                Ok(j) => j
+                    .pointer("/result/pane/pane_id")
+                    .and_then(Value::as_str)
+                    .unwrap_or_default()
+                    .to_string(),
+                Err(_) => String::new(),
+            };
+            if pane_id.is_empty() {
+                return (false, format!("no pane_id in split response: {split_out}"));
+            }
+            // 3. Rename for operator readability.
+            let _ = run(vec![
+                "pane".into(),
+                "rename".into(),
+                pane_id.clone(),
+                format!("mission-{mission_id}"),
+            ])
+            .await;
+            // 4. Launch the CLI with the prompt inline.
+            let launch = if prompt.is_empty() {
+                cli.to_string()
+            } else {
+                // Single-quoted so shell metacharacters in the prompt don't
+                // reinterpret. Herdr's pane.run sends this verbatim to the shell.
+                let escaped = prompt.replace('\'', "'\\''");
+                format!("{cli} '{escaped}'")
+            };
+            let (rok, rout) = run(vec![
+                "pane".into(),
+                "run".into(),
+                pane_id.clone(),
+                launch,
+            ])
+            .await;
+            let payload = serde_json::json!({
+                "pane_id": pane_id,
+                "split": split_out,
+                "run_output": rout,
+                "run_ok": rok,
+            });
+            (rok, payload.to_string())
+        }
+        "herdr_status" => {
+            let pane = v.get("pane_id").and_then(Value::as_str).unwrap_or_default();
+            if pane.is_empty() {
+                return (false, "pane_id required".into());
+            }
+            run(vec!["pane".into(), "get".into(), pane.to_string()]).await
+        }
+        "herdr_read" => {
+            let pane = v.get("pane_id").and_then(Value::as_str).unwrap_or_default();
+            let lines = v.get("lines").and_then(Value::as_u64).unwrap_or(200);
+            if pane.is_empty() {
+                return (false, "pane_id required".into());
+            }
+            run(vec![
+                "pane".into(),
+                "read".into(),
+                pane.to_string(),
+                "--source".into(),
+                "recent-unwrapped".into(),
+                "--lines".into(),
+                lines.to_string(),
+            ])
+            .await
+        }
+        _ => (false, format!("unknown herdr op {op}")),
+    }
+}
+
 async fn sb_op(op: &str, v: &Value) -> (bool, String) {
     let driver = match DockerDriver::connect() {
         Ok(d) => d,

crates/cm-api/src/fleet_herdr.rs

@@ -0,0 +1,165 @@
+//! Herdr second-runtime dispatch (Phase 1b).
+//!
+//! Server-side client for the `herdr_dispatch` / `herdr_status` /
+//! `herdr_read` ops the fleet-node daemon exposes. Missions with
+//! `runtime_kind = 'local_herdr'` route through this module instead
+//! of RuntimeProvisioner + ZeroClaw.
+//!
+//! Flow:
+//!   1. dispatch(mission, task) → node daemon spawns a Herdr pane +
+//!      launches the requested CLI. Returns the (workspace_id, tab_id,
+//!      pane_id) triple; caller persists it on topology_runs so a
+//!      resumed run can reattach rather than double-spawn.
+//!   2. poll_until_done(pane_id) → periodically issues herdr_status
+//!      until agent_status ∈ {done, idle} or a timeout. Between polls
+//!      the operator can watch the pane live on the node (Phase 2's
+//!      "Live pane" tab surfaces it in-browser).
+//!   3. read_transcript(pane_id) → final scrape after completion,
+//!      persisted to run_events for the Tasks tab.
+
+use cm_domain::NodeId;
+use serde_json::{json, Value};
+use std::sync::Arc;
+use std::time::Duration;
+use uuid::Uuid;
+
+use crate::fleet::NodeHub;
+
+/// What the caller needs to persist on the topology_run so a server
+/// restart can reattach to the same live pane.
+#[derive(Debug, Clone)]
+pub struct DispatchHandle {
+    pub pane_id: String,
+    pub raw_split_response: String,
+}
+
+/// Spawn a Herdr pane on `node` for `mission_id` and start `cli` with
+/// `prompt`. Returns the pane handle to persist.
+///
+/// `cli` is the executable name — "claude", "codex", "kimi", "opencode",
+/// "omp", "pi". Server callers should validate against the mission's
+/// team template (a rust_sdlc mission on tank probably wants Claude
+/// Code; a research mission on morpheus probably wants Kimi).
+pub async fn dispatch(
+    hub: Arc<NodeHub>,
+    node_id: NodeId,
+    mission_id: Uuid,
+    cli: &str,
+    prompt: &str,
+) -> Result<DispatchHandle, String> {
+    let args = json!({
+        "mission_id": mission_id.to_string(),
+        "cli": cli,
+        "prompt": prompt,
+    });
+    let out = hub
+        .call_timeout(node_id, "herdr_dispatch", args, 60)
+        .await?;
+    if !out.ok {
+        return Err(format!("node rejected dispatch: {}", truncate(&out.output, 400)));
+    }
+    let payload: Value = serde_json::from_str(&out.output)
+        .map_err(|e| format!("dispatch payload not json: {e}: {}", truncate(&out.output, 200)))?;
+    let pane_id = payload
+        .get("pane_id")
+        .and_then(Value::as_str)
+        .filter(|s| !s.is_empty())
+        .ok_or_else(|| "no pane_id in dispatch response".to_string())?
+        .to_string();
+    let split = payload
+        .get("split")
+        .and_then(Value::as_str)
+        .unwrap_or_default()
+        .to_string();
+    Ok(DispatchHandle {
+        pane_id,
+        raw_split_response: split,
+    })
+}
+
+/// Read the current agent state of a pane. Returns raw pane.get JSON
+/// so the caller can inspect any field (agent, agent_status, cwd,
+/// process metadata).
+pub async fn status(
+    hub: Arc<NodeHub>,
+    node_id: NodeId,
+    pane_id: &str,
+) -> Result<Value, String> {
+    let out = hub
+        .call_timeout(
+            node_id,
+            "herdr_status",
+            json!({ "pane_id": pane_id }),
+            20,
+        )
+        .await?;
+    if !out.ok {
+        return Err(format!("status failed: {}", truncate(&out.output, 300)));
+    }
+    serde_json::from_str(&out.output)
+        .map_err(|e| format!("status not json: {e}: {}", truncate(&out.output, 200)))
+}
+
+/// Pull the last `lines` of the pane's scrollback (unwrapped) — used
+/// to persist a completed run's transcript.
+pub async fn read_transcript(
+    hub: Arc<NodeHub>,
+    node_id: NodeId,
+    pane_id: &str,
+    lines: u32,
+) -> Result<String, String> {
+    let out = hub
+        .call_timeout(
+            node_id,
+            "herdr_read",
+            json!({ "pane_id": pane_id, "lines": lines }),
+            30,
+        )
+        .await?;
+    if !out.ok {
+        return Err(format!("read failed: {}", truncate(&out.output, 300)));
+    }
+    Ok(out.output)
+}
+
+/// Poll status every `poll_secs` until agent_status ∈ terminal set,
+/// or `timeout_secs` elapses. Returns the final status JSON.
+///
+/// Terminal set: 'done' | 'idle' after the pane has been seen at
+/// least once in a non-idle state (avoids returning immediately for
+/// a pane that hasn't yet started working).
+pub async fn wait_for_completion(
+    hub: Arc<NodeHub>,
+    node_id: NodeId,
+    pane_id: &str,
+    poll_secs: u64,
+    timeout_secs: u64,
+) -> Result<Value, String> {
+    let started = tokio::time::Instant::now();
+    let mut ever_working = false;
+    loop {
+        if started.elapsed() > Duration::from_secs(timeout_secs) {
+            return Err(format!("pane {pane_id} did not complete in {timeout_secs}s"));
+        }
+        let s = status(hub.clone(), node_id, pane_id).await?;
+        let state = s
+            .pointer("/result/agent_status")
+            .and_then(Value::as_str)
+            .unwrap_or("unknown");
+        match state {
+            "working" | "blocked" => ever_working = true,
+            "done" => return Ok(s),
+            "idle" if ever_working => return Ok(s),
+            _ => {}
+        }
+        tokio::time::sleep(Duration::from_secs(poll_secs)).await;
+    }
+}
+
+fn truncate(s: &str, max: usize) -> String {
+    if s.len() <= max {
+        s.to_string()
+    } else {
+        format!("{}…", &s[..max])
+    }
+}

crates/cm-api/src/lib.rs

@@ -7,6 +7,7 @@ pub mod cleanup_sweeper;
 mod error;
 mod extract;
 pub mod fleet;
+pub mod fleet_herdr;
 pub mod level_up;
 mod mcp_door;
 mod mcp_skills;
@@ -449,6 +450,10 @@ pub fn router(state: AppState) -> Router {
             "/api/missions/{id}/refine",
             post(routes::missions::refine),
         )
+        .route(
+            "/api/missions/{id}/herdr-dispatch",
+            post(routes::missions::herdr_dispatch),
+        )
         .route(
             "/api/missions/{id}/description",
             patch(routes::missions::set_description),

crates/cm-api/src/routes/missions.rs

@@ -361,6 +361,55 @@ pub async fn delete(
     Ok(Json(serde_json::json!({ "deleted": true })))
 }
 
+#[derive(Debug, Deserialize)]
+pub struct HerdrDispatchRequest {
+    pub cli: String,
+    pub prompt: String,
+}
+
+#[derive(Debug, Serialize)]
+pub struct HerdrDispatchResponse {
+    pub pane_id: String,
+    pub node_id: Uuid,
+}
+
+/// POST /api/missions/{id}/herdr-dispatch — manually spawn a Herdr
+/// pane on the mission's target_node running `cli` with `prompt`.
+/// Requires mission.runtime_kind = 'local_herdr' + target_node_id set.
+/// Wizard integration + auto-dispatch land in later phases; this
+/// exists so Phase 1b's fleet_herdr module can be exercised end-to-end
+/// against a real node while the rest of the arc builds out.
+pub async fn herdr_dispatch(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+    Path(id): Path<Uuid>,
+    Json(body): Json<HerdrDispatchRequest>,
+) -> Result<Json<HerdrDispatchResponse>, ApiError> {
+    let mission = cm_db::repo::missions::get(&state.pool, id, user.workspace_id.as_uuid())
+        .await?
+        .ok_or(ApiError::NotFound)?;
+    if mission.runtime_kind != "local_herdr" {
+        return Err(ApiError::BadRequest);
+    }
+    let node_id = mission.target_node_id.ok_or(ApiError::BadRequest)?;
+    let handle = crate::fleet_herdr::dispatch(
+        state.node_hub.clone(),
+        cm_domain::NodeId::from(node_id),
+        id,
+        body.cli.trim(),
+        body.prompt.trim(),
+    )
+    .await
+    .map_err(|e| {
+        eprintln!("herdr_dispatch mission {id}: {e}");
+        ApiError::Internal
+    })?;
+    Ok(Json(HerdrDispatchResponse {
+        pane_id: handle.pane_id,
+        node_id,
+    }))
+}
+
 pub async fn set_status(
     State(state): State<AppState>,
     Authed(user): Authed,