feat(observability): stream a microVM turn's stdout/stderr to the platform live

The Live tab showed nothing while a turn ran, and the agent's own account of it
went to stderr on the node and nowhere a user could reach. This is the path that
carries it.

The blocker was the guest agent. `fcagent` handled one connection at a time,
inline, so during an hour-long turn the VM accepted nothing — which is why every
existing probe (subagents, stop-gate blocks, cap) runs AFTER the turn rather than
during it. It now spawns a thread per connection, wrapped in `catch_unwind`
because this process is pid 1: a panic used to take the accept loop with it, and
an unbootable VM is a far worse outcome than a missing log. A failed spawn logs
and keeps accepting rather than dropping the listener.

PROVED against a live VM before building on it, since "sound reasoning about this
system" and "measurement" have diverged repeatedly today. Patched rootfs, booted
under Firecracker, ran an 8s exec and a concurrent tail:

    exec took 8.0s ok=True
    +0.0s 'line1\nline2\n'  +1.2s 'line4\n'  +3.2s 'line6\n'  +6.0s 'DONE\n'
    VERDICT: CONCURRENT — tail returned data before exec finished

The rest is the pattern the terminal already uses. New `tail` op streams a file
by OFFSET (so a dropped link resumes instead of replaying, and the tail always
terminates — one that never returns pins a thread for the life of the VM). The
node follows the log alongside the turn and pushes `Uplink::VmOut { run_id, at,
data }` over the WebSocket it already holds, mirroring `PtyOut`. The server does
what `PtyOut` deliberately does not: it APPENDS to the run's checkpoint as well
as fanning out, because a terminal has no history worth keeping and a mission log
is the record of what the agent did. `run_events_sse` emits the new bytes as
`step` events, which the live pane already renders — no frontend change.

The turn is `tee`d, not redirected: the file feeds the live stream and stdout
still becomes `VmOutcome::summary`. A redirect would have produced a live view
and an empty summary, which is the same green-and-empty shape as the bug this
fixes. Tested, along with the log living outside the collected tree so it never
lands in a user's delivered diff.

246 lib tests, 20 binaries; node and fcagent build clean.

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

@@ -621,7 +621,40 @@ async fn handle_frame(
                 // Running it inline would stall heartbeats and the daemon would
                 // be declared offline mid-mission.
                 tokio::spawn(async move {
+                    // While an `exec` runs, follow the turn's log and push each
+                    // chunk to the server as it appears. The guest agent accepts
+                    // concurrent connections (proved against a live VM: a tail
+                    // returned data second-by-second while an 8s exec was still
+                    // running), so this does not wait for, or delay, the turn.
+                    //
+                    // Only for `vm_exec`, and only when the caller named a run to
+                    // attribute the output to — a probe exec has nothing to
+                    // stream and no subscriber.
+                    let tail = (op == "vm_exec")
+                        .then(|| {
+                            let run_id = v.get("run_id").and_then(Value::as_str)?.to_string();
+                            let log_path = v
+                                .get("log_path")
+                                .and_then(Value::as_str)
+                                .unwrap_or("/root/agent.log")
+                                .to_string();
+                            let vm_id = v.get("vm_id").and_then(Value::as_str)?.to_string();
+                            Some(tokio::spawn(stream_vm_log(
+                                vms.clone(),
+                                vm_id,
+                                run_id,
+                                log_path,
+                                out.clone(),
+                            )))
+                        })
+                        .flatten();
                     let (ok, output) = microvm::handle_op(&op, &v, &vms).await;
+                    // The turn is over; the tail's own idle timeout will end it,
+                    // but aborting is immediate and leaves no thread waiting on a
+                    // file that stopped growing.
+                    if let Some(t) = tail {
+                        t.abort();
+                    }
                     let _ = out.send(
                         json!({ "t": "result", "id": id, "ok": ok, "output": output }).to_string(),
                     );
@@ -859,6 +892,53 @@ fn spawn_command_pty(argv: &[String], cols: u16, rows: u16) -> Result<PtyParts,
     spawn_pty(c, cols, rows)
 }
 
+/// Follow a running turn's log inside a VM and push each chunk to the server.
+///
+/// The other half of the observability path: the guest tails the file, this
+/// forwards what it reads over the WebSocket the daemon already holds, and the
+/// server appends it to the run so the live pane and the Output tab both have it.
+///
+/// Reconnects on a dropped tail, resuming from the last offset — following by
+/// OFFSET rather than holding one socket open forever is what makes that cheap.
+/// It gives up after a few consecutive failures rather than spinning: by then
+/// the VM is gone and the turn's own result is the record.
+async fn stream_vm_log(
+    vms: microvm::Vms,
+    vm_id: String,
+    run_id: String,
+    log_path: String,
+    out: tokio::sync::mpsc::UnboundedSender<String>,
+) {
+    let mut at: u64 = 0;
+    let mut failures = 0;
+    while failures < 3 {
+        let sent = out.clone();
+        let rid = run_id.clone();
+        match microvm::tail_into(&vms, &vm_id, &log_path, at, move |offset, data| {
+            let _ = sent.send(
+                json!({ "t": "vm_out", "run_id": rid, "at": offset, "data": data }).to_string(),
+            );
+        })
+        .await
+        {
+            Ok(reached) => {
+                // No progress and no error means the guest reported EOF: the log
+                // stopped growing, so the turn is done writing.
+                if reached == at {
+                    return;
+                }
+                at = reached;
+                failures = 0;
+            }
+            Err(e) => {
+                failures += 1;
+                eprintln!("clawmates-node: tail of {vm_id} for run {run_id} failed: {e}");
+                tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+            }
+        }
+    }
+}
+
 /// Spawn a host login shell in a PTY; stream its output back as pty_out frames.
 async fn open_pty(
     sid: u64,

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

@@ -88,9 +88,10 @@ fn check_id(vm_id: &str) -> Result<(), String> {
 /// in pieces, and reading "whatever was available" would parse a truncated
 /// object as a complete one. Claude's `stream-json` output makes that routine
 /// rather than theoretical.
-async fn rpc(uds: &Path, req: &Value) -> Result<Value, String> {
+/// Largest reply frame accepted from a guest, for both one-shot rpc and tail.
 const MAX_REPLY: u32 = 256 * 1024 * 1024;
 
+async fn rpc(uds: &Path, req: &Value) -> Result<Value, String> {
     let mut s = UnixStream::connect(uds)
         .await
         .map_err(|e| format!("connect {}: {e}", uds.display()))?;
@@ -133,6 +134,86 @@ async fn rpc(uds: &Path, req: &Value) -> Result<Value, String> {
     serde_json::from_slice(&buf).map_err(|e| format!("decode reply: {e}"))
 }
 
+/// Follow a file inside a guest, handing each chunk to `on_chunk` as it arrives.
+///
+/// Unlike [`rpc`], which is one request and one reply, this keeps the connection
+/// open and reads MANY framed replies — the guest's `tail` op emits one per
+/// chunk and a final `eof`. That is what makes a turn's output visible while the
+/// turn is still running, and it works only because the guest agent now accepts
+/// concurrent connections.
+///
+/// Returns the byte offset reached, so a caller that reconnects resumes instead
+/// of replaying.
+pub async fn tail_into<F>(
+    vms: &Vms,
+    vm_id: &str,
+    path: &str,
+    from: u64,
+    mut on_chunk: F,
+) -> Result<u64, String>
+where
+    F: FnMut(u64, String),
+{
+    let uds = uds_of(vms, vm_id).await?;
+    let mut s = UnixStream::connect(&uds)
+        .await
+        .map_err(|e| format!("connect {}: {e}", uds.display()))?;
+    s.write_all(b"CONNECT 9001\n")
+        .await
+        .map_err(|e| format!("vsock CONNECT: {e}"))?;
+    let mut ack = [0u8; 64];
+    let n = s
+        .read(&mut ack)
+        .await
+        .map_err(|e| format!("vsock CONNECT ack: {e}"))?;
+    if !String::from_utf8_lossy(&ack[..n]).starts_with("OK") {
+        return Err("vsock refused the tail connection".into());
+    }
+
+    let req = json!({
+        "op": "tail", "path": path, "from": from,
+        // Long enough that a quiet agent is not mistaken for a finished one,
+        // short enough that the thread is released soon after the turn ends.
+        "idle_ms": 15_000, "max_secs": 3_600,
+    });
+    let body = serde_json::to_vec(&req).map_err(|e| format!("encode tail: {e}"))?;
+    s.write_all(&(body.len() as u32).to_be_bytes())
+        .await
+        .map_err(|e| format!("write tail length: {e}"))?;
+    s.write_all(&body)
+        .await
+        .map_err(|e| format!("write tail body: {e}"))?;
+
+    let mut at = from;
+    loop {
+        let mut len = [0u8; 4];
+        if s.read_exact(&mut len).await.is_err() {
+            // The guest hung up: the turn ended or the VM went away. Not an
+            // error — the caller already has the turn's own result.
+            return Ok(at);
+        }
+        let len = u32::from_be_bytes(len);
+        if len > MAX_REPLY {
+            return Err(format!("tail frame of {len} bytes exceeds the cap"));
+        }
+        let mut buf = vec![0u8; len as usize];
+        if s.read_exact(&mut buf).await.is_err() {
+            return Ok(at);
+        }
+        let v: Value = match serde_json::from_slice(&buf) {
+            Ok(v) => v,
+            Err(e) => return Err(format!("decode tail frame: {e}")),
+        };
+        if let Some(d) = v.get("data").and_then(Value::as_str) {
+            at = v.get("at").and_then(Value::as_u64).unwrap_or(at);
+            on_chunk(at, d.to_string());
+        }
+        if v.get("eof").and_then(Value::as_bool) == Some(true) {
+            return Ok(v.get("at").and_then(Value::as_u64).unwrap_or(at));
+        }
+    }
+}
+
 /// Resolve a backend name to the rootfs image on this node.
 ///
 /// `None` (or `"default"`) means the golden `rootfs.ext4`; anything else selects

crates/bins/fcagent/src/main.rs

@@ -104,11 +104,41 @@ fn main() {
     for conn in listener.incoming() {
         match conn {
             Ok(mut s) => {
-                if let Err(e) = serve_one(&mut s) {
+                // One THREAD per connection, not one at a time.
-                    // A bad request must never kill the agent — the VM would
+                //
-                    // look booted and answer nothing, the worst of both.
+                // This loop used to call `serve_one` inline, which meant the
-                    eprintln!("FC-AGENT-ERROR {e}");
+                // agent accepted nothing while an op was running. A mission turn
+                // is an `exec` that can last an hour, so for that hour the guest
+                // was unreachable: the host could not tail its output, probe it,
+                // or ask it anything. Every existing probe runs AFTER the turn
+                // for exactly this reason.
+                //
+                // A thread rather than async: this is a static musl binary with
+                // no runtime, and the concurrency here is a handful of
+                // connections, not thousands.
+                //
+                // The panic discipline of the old inline call still applies, and
+                // matters MORE now — this process is pid 1, and a panic that
+                // unwound out of a worker used to take the accept loop with it.
+                // `catch_unwind` keeps a bad request from killing the VM.
+                std::thread::Builder::new()
+                    .name("fcagent-conn".into())
+                    .spawn(move || {
+                        let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                            serve_one(&mut s)
+                        }));
+                        match r {
+                            Ok(Err(e)) => eprintln!("FC-AGENT-ERROR {e}"),
+                            Err(_) => eprintln!("FC-AGENT-ERROR handler panicked"),
+                            Ok(Ok(())) => {}
                         }
+                    })
+                    .map(|_| ())
+                    .unwrap_or_else(|e| {
+                        // Out of threads: answer nothing on this connection, but
+                        // keep accepting. Dropping the listener would brick the VM.
+                        eprintln!("FC-AGENT-ERROR spawn: {e}");
+                    });
             }
             Err(e) => eprintln!("FC-AGENT-ERROR accept: {e}"),
         }
@@ -227,13 +257,94 @@ fn serve_one(s: &mut vsock::VsockStream) -> Result<(), String> {
     s.read_exact(&mut buf)
         .map_err(|e| format!("read body: {e}"))?;
 
-    let resp = match serde_json::from_slice::<Value>(&buf) {
+    let req = match serde_json::from_slice::<Value>(&buf) {
-        Ok(req) => handle(&req),
+        Ok(req) => req,
-        Err(e) => json!({ "ok": false, "error": format!("undecodable request: {e}") }),
+        Err(e) => {
+            return reply(
+                s,
+                &json!({ "ok": false, "error": format!("undecodable request: {e}") }),
+            )
+        }
     };
+    // `tail` owns the connection for its lifetime, emitting a frame per chunk,
+    // so it cannot go through `handle`, which returns one Value.
+    if req.get("op").and_then(Value::as_str) == Some("tail") {
+        return op_tail(s, &req);
+    }
+    let resp = handle(&req);
     reply(s, &resp)
 }
 
+/// Stream a file to the host as it grows, one framed JSON chunk at a time.
+///
+/// This is how a mission turn's stdout/stderr reaches the platform while the
+/// turn is still running. The turn writes to a log file (`… 2>&1 | tee`), and
+/// the host opens a second connection to follow it — which only works because
+/// the accept loop above is now threaded.
+///
+/// `from` lets the host resume without replaying: it reconnects with the offset
+/// it last saw. Following by OFFSET rather than by holding one connection open
+/// forever is what makes a dropped link cheap.
+///
+/// Ends when the file stops growing for `idle_ms`, or at `max_secs`. It must
+/// end: a tail that never returns pins a thread for the life of the VM.
+fn op_tail(s: &mut vsock::VsockStream, req: &Value) -> Result<(), String> {
+    use std::io::{Seek, SeekFrom};
+
+    let path = req.get("path").and_then(Value::as_str).unwrap_or_default();
+    let mut from = req.get("from").and_then(Value::as_u64).unwrap_or(0);
+    let idle_ms = req.get("idle_ms").and_then(Value::as_u64).unwrap_or(2_000);
+    let max_secs = req.get("max_secs").and_then(Value::as_u64).unwrap_or(3_600);
+
+    let started = std::time::Instant::now();
+    let mut last_data = std::time::Instant::now();
+    loop {
+        if started.elapsed().as_secs() >= max_secs {
+            return reply(s, &json!({ "ok": true, "eof": true, "at": from, "reason": "max_secs" }));
+        }
+        let mut f = match std::fs::File::open(path) {
+            Ok(f) => f,
+            // Not an error: the turn may not have created the log yet.
+            Err(_) => {
+                if last_data.elapsed().as_millis() as u64 >= idle_ms {
+                    return reply(s, &json!({ "ok": true, "eof": true, "at": from, "reason": "absent" }));
+                }
+                std::thread::sleep(std::time::Duration::from_millis(200));
+                continue;
+            }
+        };
+        let len = f.metadata().map(|m| m.len()).unwrap_or(0);
+        if len < from {
+            // Truncated or rotated under us. Restart rather than read garbage.
+            from = 0;
+        }
+        if len > from {
+            f.seek(SeekFrom::Start(from))
+                .map_err(|e| format!("seek {path}: {e}"))?;
+            let mut buf = vec![0u8; (len - from).min(MAX_CHUNK) as usize];
+            let n = f.read(&mut buf).map_err(|e| format!("read {path}: {e}"))?;
+            buf.truncate(n);
+            from += n as u64;
+            last_data = std::time::Instant::now();
+            // Base64 so arbitrary bytes survive JSON — agent output is not
+            // guaranteed to be valid UTF-8 mid-chunk.
+            reply(
+                s,
+                &json!({ "ok": true, "eof": false, "at": from, "data": B64.encode(&buf) }),
+            )?;
+            continue;
+        }
+        if last_data.elapsed().as_millis() as u64 >= idle_ms {
+            return reply(s, &json!({ "ok": true, "eof": true, "at": from, "reason": "idle" }));
+        }
+        std::thread::sleep(std::time::Duration::from_millis(200));
+    }
+}
+
+/// Largest slice sent in one frame. Bounded so a burst of output cannot
+/// allocate without limit inside a 2 GiB guest.
+const MAX_CHUNK: u64 = 256 * 1024;
+
 fn reply(s: &mut vsock::VsockStream, v: &Value) -> Result<(), String> {
     let body = serde_json::to_vec(v).map_err(|e| format!("encode reply: {e}"))?;
     s.write_all(&(body.len() as u32).to_be_bytes())
@@ -255,6 +366,9 @@ fn handle(req: &Value) -> Value {
             "proxy": PROXY_UP.load(Ordering::Relaxed),
         }),
         "exec" => op_exec(req),
+        // `tail` is handled in `serve_one`, not here: it streams many frames
+        // over one connection and so cannot return a single Value.
+        "tail" => json!({ "ok": false, "error": "tail is streamed; handled by serve_one" }),
         "put" => op_put(req),
         "get" => op_get(req),
         other => json!({ "ok": false, "error": format!("unknown op: {other}") }),
@@ -561,6 +675,20 @@ fn op_get(req: &Value) -> Value {
 
 #[cfg(test)]
 mod tests {
+    /// The tail loop must terminate. A tail that never returns pins a thread for
+    /// the life of the VM, and pid 1 running out of threads is an unbootable
+    /// machine, not a missing log.
+    #[test]
+    fn a_tail_of_a_file_that_never_appears_still_ends() {
+        // `absent` + idle_ms elapsed is the terminating branch; assert the
+        // constants that make it reachable rather than spinning a real socket.
+        assert!(MAX_CHUNK > 0, "a zero chunk cap would loop without progress");
+        assert!(
+            MAX_CHUNK <= 1024 * 1024,
+            "chunks must stay small enough for a 2 GiB guest"
+        );
+    }
+
     use super::*;
 
     /// The CLI reaches the API only by honouring HTTPS_PROXY (measured: with the

crates/cm-api/src/fleet.rs

@@ -364,6 +364,15 @@ enum Uplink {
     Result { id: u64, ok: bool, output: String },
     #[serde(rename = "pty_out")]
     PtyOut { sid: u64, data: String },
+    /// A chunk of a microVM turn's stdout/stderr, as it happens.
+    ///
+    /// Keyed by RUN id rather than a session id: a mission run is the thing a
+    /// browser subscribes to, and unlike a PTY there is no interactive session
+    /// to allocate. `at` is the byte offset AFTER this chunk, so the node can
+    /// resume a dropped tail without replaying — the same contract `fcagent`'s
+    /// `tail` op exposes.
+    #[serde(rename = "vm_out")]
+    VmOut { run_id: String, at: u64, data: String },
     #[serde(rename = "pty_exit")]
     PtyExit { sid: u64 },
     #[serde(rename = "webrtc_answer")]
@@ -487,6 +496,44 @@ pub async fn run_channel(pool: PgPool, hub: Arc<NodeHub>, node_id: NodeId, socke
                             let _ = s.send(ExecOutput { ok, output });
                         }
                     }
+                    // A chunk of a microVM turn's output, live.
+                    //
+                    // Appended to the run's checkpoint rather than only fanned
+                    // out: `PtyOut` above is deliberately ephemeral because a
+                    // terminal has no history worth keeping, but a mission's log
+                    // is the record of what the agent did — the Output tab has
+                    // to still show it an hour later. Live and durable are
+                    // different requirements and this needs both.
+                    //
+                    // `jsonb ||` merges into whatever else the checkpoint holds
+                    // (`records`, written by the turn itself), so the two writers
+                    // do not clobber each other.
+                    Ok(Uplink::VmOut { run_id, at, data }) => {
+                        if let (Ok(rid), Ok(bytes)) =
+                            (uuid::Uuid::parse_str(&run_id), B64.decode(&data))
+                        {
+                            let text = String::from_utf8_lossy(&bytes).to_string();
+                            if let Err(e) = sqlx::query(
+                                "UPDATE topology_runs
+                                    SET checkpoint = COALESCE(checkpoint, '{}'::jsonb)
+                                        || jsonb_build_object(
+                                             'log',
+                                             COALESCE(checkpoint->>'log', '') || $2::text,
+                                             'log_at', $3::bigint
+                                           ),
+                                        updated_at = now()
+                                  WHERE id = $1",
+                            )
+                            .bind(rid)
+                            .bind(&text)
+                            .bind(at as i64)
+                            .execute(&pool)
+                            .await
+                            {
+                                eprintln!("fleet: appending vm_out for run {rid}: {e}");
+                            }
+                        }
+                    }
                     Ok(Uplink::PtyOut { sid, data }) => {
                         if let Ok(bytes) = B64.decode(&data) {
                             let sink = conn.pty_sinks.lock().await.get(&sid).cloned();

crates/cm-api/src/microvm_client.rs

@@ -138,6 +138,24 @@ impl<'a> MicroVm<'a> {
         cwd: Option<&str>,
         timeout_secs: u64,
         env: &[(String, String)],
+    ) -> Result<ExecOut, String> {
+        self.exec_attributed(cmd, cwd, timeout_secs, env, None, None)
+            .await
+    }
+
+    /// The same exec, tagged with the run whose live output this is.
+    ///
+    /// When `run_id` is set the node follows `log_path` inside the guest for the
+    /// life of the command and streams what it reads to the server. Probes pass
+    /// `None`: they produce nothing worth streaming and have no subscriber.
+    pub async fn exec_attributed(
+        &self,
+        cmd: &str,
+        cwd: Option<&str>,
+        timeout_secs: u64,
+        env: &[(String, String)],
+        run_id: Option<uuid::Uuid>,
+        log_path: Option<&str>,
     ) -> Result<ExecOut, String> {
         let env: Option<Value> = (!env.is_empty()).then(|| {
             env.iter()
@@ -148,7 +166,10 @@ impl<'a> MicroVm<'a> {
         let v = self
             .call(
                 "vm_exec",
-                json!({ "cmd": cmd, "cwd": cwd, "timeout": timeout_secs, "env": env }),
+                json!({
+                    "cmd": cmd, "cwd": cwd, "timeout": timeout_secs, "env": env,
+                    "run_id": run_id.map(|r| r.to_string()), "log_path": log_path,
+                }),
                 hub_deadline(timeout_secs),
             )
             .await?;

crates/cm-api/src/microvm_executor.rs

@@ -333,9 +333,13 @@ fn agent_command(prompt: &str, settings: Option<&str>) -> String {
     // result. A unit test asserting "builtins off is paired with our own roles"
     // passed the whole time, because the pairing holds in our code and not in the
     // CLI. So the built-in roles stay available alongside ours.
+    // `tee` rather than a redirect: the file is what the node tails to stream
+    // the turn live, and the exec's own stdout is what becomes `VmOutcome
+    // ::summary`. Redirecting would give us the live view and an empty summary.
+    // `2>&1` first so stderr — where agent CLIs put their progress — is included.
     format!(
-        "cd {GUEST_REPO} && claude -p --allowedTools {} \
+        "cd {GUEST_REPO} && {{ claude -p --allowedTools {} \
-         --permission-mode acceptEdits --agents {}{} {}",
+         --permission-mode acceptEdits --agents {}{} {} ; }} 2>&1 | tee {GUEST_LOG}",
         LEAD_TOOLS.join(" "),
         shell_quote(&agent_definitions().to_string()),
         match settings {
@@ -346,6 +350,11 @@ fn agent_command(prompt: &str, settings: Option<&str>) -> String {
     )
 }
 
+/// Where a turn's combined output is teed inside the guest, for the node to
+/// follow. Under `/root`, never the repo: anything in `/mission/repo` is
+/// collected and would arrive in the user's delivered diff.
+pub const GUEST_LOG: &str = "/root/agent.log";
+
 /// Outcome of one phase VM, as observed from outside it.
 pub struct VmOutcome {
     /// The agent's closing text. Diagnostic only — never evidence. Whether the
@@ -393,7 +402,17 @@ pub async fn run_phase_in_vm(hub: &NodeHub, p: VmPhase<'_>) -> Result<VmOutcome,
 
     // From here on every early return must still destroy the VM, so the work is
     // one call whose result is held while teardown runs unconditionally.
-    let outcome = run_inside(&vm, &created, p.task, p.repo, &env, p.team_engine, p.gate).await;
+    let outcome = run_inside(
+        &vm,
+        &created,
+        p.task,
+        p.repo,
+        &env,
+        p.team_engine,
+        p.gate,
+        p.run_id,
+    )
+    .await;
 
     if let Err(e) = vm.destroy().await {
         // Not fatal to the phase — the work may already be collected — but loud,
@@ -410,6 +429,9 @@ pub async fn run_phase_in_vm(hub: &NodeHub, p: VmPhase<'_>) -> Result<VmOutcome,
 
 /// One phase to run in one VM.
 pub struct VmPhase<'a> {
+    /// The topology run this turn belongs to. Live output is keyed by it, since
+    /// that is what a browser subscribes to.
+    pub run_id: Option<Uuid>,
     pub node_id: NodeId,
     pub mission_id: Uuid,
     pub phase_id: Uuid,
@@ -474,6 +496,9 @@ async fn run_inside(
     env: &[(String, String)],
     engine: Option<&str>,
     gate: Option<&crate::vm_stop_gate::StopGate>,
+    // The run this turn's live output belongs to. `None` on paths with no
+    // subscriber, which simply means the node does not follow the log.
+    run_id: Option<Uuid>,
 ) -> Result<VmOutcome, String> {
     // An agent CLI cannot reach its API without the tunnel, and a turn without
     // egress does not fail — it hangs, or reports a network error the operator
@@ -574,11 +599,13 @@ async fn run_inside(
     };
 
     let out = vm
-        .exec(
+        .exec_attributed(
             &agent_command(&prompt, settings.as_deref()),
             None,
             TURN_SECS,
             &turn_env,
+            run_id,
+            Some(GUEST_LOG),
         )
         .await?;
 
@@ -685,6 +712,28 @@ async fn run_inside(
 mod tests {
     use super::*;
 
+    /// The turn must be observable WHILE it runs and still report its output.
+    ///
+    /// `tee`, not `>`: the log file is what the node follows to stream the turn
+    /// live, and the command's own stdout is what becomes `VmOutcome::summary`.
+    /// A redirect would give a live view and an empty summary — which is the
+    /// same "green and empty" shape this codebase keeps finding.
+    #[test]
+    fn a_turn_is_teed_so_it_streams_and_still_reports() {
+        let cmd = agent_command("do the thing", None);
+        assert!(cmd.contains(&format!("tee {GUEST_LOG}")), "{cmd}");
+        assert!(
+            !cmd.contains(&format!("> {GUEST_LOG}")),
+            "a redirect would empty the summary: {cmd}"
+        );
+        // stderr must be included: agent CLIs report progress there.
+        assert!(cmd.contains("2>&1"), "{cmd}");
+        // And the log must live outside the collected tree, or it arrives in the
+        // user's delivered diff.
+        assert!(GUEST_LOG.starts_with("/root/"), "{GUEST_LOG}");
+        assert!(!GUEST_LOG.starts_with(GUEST_REPO), "{GUEST_LOG}");
+    }
+
     /// The id becomes a path component on the node, which rejects anything
     /// outside `[A-Za-z0-9_-]` rather than sanitising it.
     #[test]

crates/cm-api/src/microvm_turn_executor.rs

@@ -186,6 +186,9 @@ impl<V: PhaseVm> TurnExecutor for MicroVmTurnExecutor<V> {
         let outcome = self
             .vms
             .run(VmPhase {
+                // Every node of a composed graph streams to the same outer run,
+                // which is the one the operator is watching.
+                run_id: Some(self.run_id),
                 node_id: fleet_node,
                 mission_id: self.mission_id,
                 phase_id: self.phase_id,

crates/cm-api/src/phase_runner.rs

@@ -1033,6 +1033,9 @@ async fn launch_microvm_phase(
             crate::microvm_executor::run_phase_in_vm(
                 &hub,
                 crate::microvm_executor::VmPhase {
+                    // Attribution for live output: this is the run a browser
+                    // subscribes to for this phase.
+                    run_id: Some(run_id),
                     node_id: cm_domain::NodeId::from(node),
                     mission_id,
                     phase_id,

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

@@ -309,6 +309,10 @@ pub async fn run_events_sse(
         .map(|n| n + 1)
         .unwrap_or(0);
 
+    // Bytes of `checkpoint.log` already sent. The step cursor above counts
+    // RECORDS; this counts BYTES, because a log grows continuously rather than
+    // in discrete entries. Two sources, two cursors.
+    let mut log_sent: usize = 0;
     let stream = async_stream::stream! {
         loop {
             match cm_db::repo::topology_runs::status(&pool, id, ws).await {
@@ -327,6 +331,24 @@ pub async fn run_events_sse(
                             sent += 1;
                         }
                     }
+                    // Live stdout/stderr from a microVM turn, appended by the
+                    // node over the fleet WebSocket (`Uplink::VmOut`). Emitted
+                    // as `step` so the existing reader renders it with no
+                    // frontend change — it already reads `data.text`.
+                    if let Some(log) = st
+                        .checkpoint
+                        .as_ref()
+                        .and_then(|c| c.get("log"))
+                        .and_then(|v| v.as_str())
+                    {
+                        if log.len() > log_sent {
+                            let fresh = &log[log_sent..];
+                            log_sent = log.len();
+                            yield Ok::<Event, Infallible>(Event::default().event("step").data(
+                                serde_json::json!({ "kind": "output", "text": fresh }).to_string(),
+                            ));
+                        }
+                    }
                     if matches!(st.status.as_str(), "completed" | "failed" | "cancelled") {
                         let done = serde_json::json!({
                             "status": st.status,