feat(viz): microVM tool motion is live, and needs no fleet-node change

The plan deferred this as "the only fleet-node binary change". It is not
one. `fcagent` is thread-per-connection — its own comment says so, and
the live log tail has relied on exactly that for the whole length of a
turn, on a second connection. So the host can drain the tap WHILE the
turn's exec is in flight, from the server alone.

The turn and a 20s drain loop now run concurrently. A coding phase shows
its files being touched as it works rather than an hour later, all at
once, and the drain is bounded by a cursor so a repeated poll returns
only what is new.

The cursor counts LINES, not parsed events, and that distinction is the
bug this commit would otherwise have shipped. The hook appends the event
and then a newline of its own, so a two-event tap is four lines; advancing
by event count leaves the cursor two lines short, `tail -n +N` hands back
events already recorded, and the live drain re-records everything it has
already written — worse the longer the turn runs, and silent throughout.
Caught while writing the test, not by it.

`tap_sink` and `VmOutcome::tools` are mutually exclusive by contract: with
a sink, the sink owns recording including the final batch and `tools`
comes back empty. Handing the same calls back on both would double every
file orb's weight with no way for the caller to tell which it was
looking at.

The sink is an unbounded channel to a recorder task, so the VM executor
stays free of the database: it observes, phase_runner records. The task
ends when the sender drops with the phase.

Verified before this change: the microVM tap is real. The `microvm`
scenario passed 6/6 and left ten `tool.call` rows and a `file.touch` on
MICROVM.md, repo-relative, from Claude Code's own PostToolUse hook.

Co-Authored-By: Claude Opus 5 <[email protected]>

crates/cm-api/src/microvm_executor.rs

@@ -350,6 +350,13 @@ fn agent_command(prompt: &str, settings: Option<&str>) -> String {
     )
 }
 
+/// How often the tap is drained while a turn runs.
+///
+/// A turn can last an hour; the World is meant to show what is happening now.
+/// Every 20s is one extra guest connection per 20s — negligible beside the log
+/// tail, which reconnects every 300ms for the same length of time.
+const LIVE_TAP_SECS: u64 = 20;
+
 /// 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.
@@ -419,6 +426,7 @@ pub async fn run_phase_in_vm(hub: &NodeHub, p: VmPhase<'_>) -> Result<VmOutcome,
         p.team_engine,
         p.gate,
         p.run_id,
+        p.tap_sink.as_ref(),
     )
     .await;
 
@@ -474,6 +482,19 @@ pub struct VmPhase<'a> {
     /// qualification. Part of the vm id, so the nodes of one phase-iteration
     /// cannot collide on a fleet node.
     pub step: Option<u32>,
+    /// Where tool observations go WHILE the turn runs.
+    ///
+    /// The plan for this assumed a fleet-node binary change. It does not need
+    /// one: `fcagent` is thread-per-connection (`main.rs`: "One THREAD per
+    /// connection, not one at a time"), which is the same property the live log
+    /// tail already relies on, so the host can drain the tap on a second
+    /// connection while the turn's exec is still in flight.
+    ///
+    /// `Some` ⇒ the sink owns recording and [`VmOutcome::tools`] comes back
+    /// EMPTY. `None` ⇒ one drain at the end into `tools`. Exactly one of the
+    /// two, never both: they read the same file, and a phase whose tools were
+    /// recorded twice would draw every file orb at double weight.
+    pub tap_sink: Option<tokio::sync::mpsc::UnboundedSender<Vec<crate::vm_tool_tap::Observed>>>,
 }
 
 /// Runs one phase (or one graph node of one) in a VM.
@@ -520,6 +541,10 @@ async fn run_inside(
     // 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>,
+    // Where tool observations go while the turn is still running. See
+    // `VmPhase::tap_sink` — `Some` means the sink owns recording and the
+    // returned `tools` is empty.
+    tap_sink: Option<&tokio::sync::mpsc::UnboundedSender<Vec<crate::vm_tool_tap::Observed>>>,
 ) -> 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
@@ -688,30 +713,112 @@ async fn run_inside(
         }
     };
 
-    let out = vm
+    // The turn, and — concurrently — the tap drain that makes its work visible
-        .exec_attributed(
+    // WHILE it happens rather than an hour later.
-            &agent_command(&prompt, settings.as_deref()),
+    //
+    // Concurrency here is safe because `fcagent` is thread-per-connection: the
+    // live log tail already relies on exactly that, on a second connection, for
+    // the whole length of a turn. So this needs no fleet-node change.
+    let turn_cmd = agent_command(&prompt, settings.as_deref());
+    let turn = vm.exec_attributed(
+        &turn_cmd,
         None,
         TURN_SECS,
         &turn_env,
         run_id,
         Some(GUEST_LOG),
+    );
+    // Lines already drained and handed to the sink. Shared with the final
+    // drain below so the two cannot overlap — the same line recorded twice is
+    // a file orb at double weight, and nothing would report it.
+    let drained_lines = std::sync::atomic::AtomicUsize::new(0);
+    let out = match (&tap_dir, &tap_sink) {
+        (Some(_), Some(sink)) => {
+            let done = std::sync::atomic::AtomicBool::new(false);
+            let live = async {
+                // Checked AFTER a pass, never before: exiting on the flag alone
+                // would drop whatever the agent did in the last interval.
+                loop {
+                    tokio::time::sleep(std::time::Duration::from_secs(LIVE_TAP_SECS)).await;
+                    let finished = done.load(std::sync::atomic::Ordering::Relaxed);
+                    if let Ok(o) = vm
+                        .exec(
+                            &crate::vm_tool_tap::drain_from(
+                                drained_lines.load(std::sync::atomic::Ordering::Relaxed),
+                            ),
+                            None,
+                            30,
+                            &[],
                         )
-        .await?;
+                        .await
+                    {
+                        // LINES, not parsed events — see `consumed_lines`.
+                        let lines = crate::vm_tool_tap::consumed_lines(&o.stdout);
+                        if lines > 0 {
+                            drained_lines.fetch_add(lines, std::sync::atomic::Ordering::Relaxed);
+                            let batch = crate::vm_tool_tap::parse(&o.stdout);
+                            if !batch.is_empty() && sink.send(batch).is_err() {
+                                // Nobody is recording any more. Stop polling the
+                                // guest rather than burning a connection every
+                                // interval for output with no destination.
+                                return;
+                            }
+                        }
+                    }
+                    if finished {
+                        return;
+                    }
+                }
+            };
+            let turn = async {
+                let r = turn.await;
+                done.store(true, std::sync::atomic::Ordering::Relaxed);
+                r
+            };
+            let (r, ()) = tokio::join!(turn, live);
+            r?
+        }
+        _ => turn.await?,
+    };
 
     // The tool tap, drained BEFORE collect: it lives in /root, outside the
     // collected tree, and the VM is destroyed moments later. This is the only
     // chance to read it.
     let tools = match tap_dir {
         None => Vec::new(),
-        Some(_) => match vm.exec(crate::vm_tool_tap::DRAIN_PROBE, None, 60, &[]).await {
+        // Always from the cursor, so a sink-driven turn reads only what its
-            Ok(p) => crate::vm_tool_tap::parse(&p.stdout),
+        // last live pass did not, and a sink-less one reads the whole file
+        // (its cursor never moved).
+        Some(_) => match vm
+            .exec(
+                &crate::vm_tool_tap::drain_from(
+                    drained_lines.load(std::sync::atomic::Ordering::Relaxed),
+                ),
+                None,
+                60,
+                &[],
+            )
+            .await
+        {
+            Ok(o) => crate::vm_tool_tap::parse(&o.stdout),
             Err(e) => {
                 eprintln!("microvm_executor: tap drain failed on {}: {e}", vm.vm_id());
                 Vec::new()
             }
         },
     };
+    // With a sink, the sink owns recording — including this last batch. Handing
+    // the same calls back on `VmOutcome::tools` as well would record them
+    // twice, and the caller has no way to tell which it is looking at.
+    let tools = match &tap_sink {
+        Some(sink) => {
+            if !tools.is_empty() {
+                let _ = sink.send(tools);
+            }
+            Vec::new()
+        }
+        None => tools,
+    };
     if tap_dir.is_some() && tools.is_empty() {
         // A tap that installed and drained nothing is the silent case: the
         // phase looks the same as it did before the tap existed. Say so, or the

crates/cm-api/src/microvm_turn_executor.rs

@@ -208,6 +208,15 @@ impl<V: PhaseVm> TurnExecutor for MicroVmTurnExecutor<V> {
                 // one source for what "done" means, whichever executor asks.
                 gate: self.gate.as_ref(),
                 step: Some(step),
+                // Same live drain as the solo path. A composed graph node can
+                // run for an hour too, and its files are the only account of
+                // what it did until the next node collects.
+                tap_sink: Some(crate::phase_runner::vm_tool_recorder(
+                    &self.pool,
+                    self.mission_id,
+                    self.phase_id,
+                    self.run_id,
+                )),
             })
             .await
             .map_err(|e| {

crates/cm-api/src/phase_runner.rs

@@ -1168,6 +1168,10 @@ async fn launch_microvm_phase(
                     // The solo path is one VM for the whole phase; only a
                     // composed run needs the id qualified per graph node.
                     step: None,
+                    // Live tool motion: the tap is drained WHILE the turn runs,
+                    // so the World shows a coding phase touching files as it
+                    // happens rather than an hour later, all at once.
+                    tap_sink: Some(vm_tool_recorder(&pool2, mission_id, phase_id, run_id)),
                 },
             )
             .await
@@ -1198,6 +1202,11 @@ async fn launch_microvm_phase(
         // consumed. Recorded whatever the phase's verdict: a phase that failed
         // still did work, and the map of what it touched is exactly what makes
         // the failure legible.
+        // With a sink attached, `tools` comes back EMPTY by contract — the
+        // recorder task has already written every batch, including the final
+        // one. Kept as a no-op rather than deleted so a future sink-less path
+        // still records; recording the same calls twice is what the empty
+        // contract exists to prevent.
         if let Ok(o) = &outcome {
             record_vm_tools(&pool2, mission_id, phase_id, run_id, &o.tools).await;
         }
@@ -1495,6 +1504,29 @@ fn phase_task_text(
 /// `agent_id` is deliberately absent: a microVM phase has no platform agent, so
 /// there is no pawn to attribute the touch to. Inventing one would put a named
 /// crew member's face on work a VM did alone.
+/// A sink that records tool observations into `mission_events` as they arrive,
+/// plus the task draining it.
+///
+/// The channel exists so the VM executor stays free of the database: it
+/// observes, this records. The task ends when the sender is dropped, which
+/// happens when the phase's `VmPhase` goes out of scope — so there is no
+/// lifetime to manage and no way to leak one per phase.
+pub(crate) fn vm_tool_recorder(
+    pool: &PgPool,
+    mission_id: Uuid,
+    phase_id: Uuid,
+    run_id: Uuid,
+) -> tokio::sync::mpsc::UnboundedSender<Vec<crate::vm_tool_tap::Observed>> {
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Vec<crate::vm_tool_tap::Observed>>();
+    let pool = pool.clone();
+    tokio::spawn(async move {
+        while let Some(batch) = rx.recv().await {
+            record_vm_tools(&pool, mission_id, phase_id, run_id, &batch).await;
+        }
+    });
+    tx
+}
+
 pub(crate) async fn record_vm_tools(
     pool: &PgPool,
     mission_id: Uuid,

crates/cm-api/src/vm_tool_tap.rs

@@ -39,6 +39,32 @@ pub const SETTINGS_PATH: &str = "/root/guest-settings.json";
 /// those two is the histogram in the log, not an error here.
 pub const DRAIN_PROBE: &str = "cat /root/tap/tools.jsonl 2>/dev/null || true";
 
+/// Read the tap from line `from` onward, so a repeated drain returns only what
+/// is new.
+///
+/// A cursor rather than a re-read: the live drain runs every few seconds
+/// against a file the agent is still appending to, and re-sending the whole
+/// file each pass would record every tool call once per poll — a phase would
+/// finish with its early files weighted by how long it ran.
+///
+/// `tail -n +N` is 1-based on the FIRST line to print, so `from` is a line
+/// count already consumed and the probe asks for `from + 1`.
+pub fn drain_from(from: usize) -> String {
+    format!("tail -n +{} {TAP_FILE} 2>/dev/null || true", from + 1)
+}
+
+/// How far a drain advanced the cursor — the number of LINES it consumed.
+///
+/// Counts every line, including blank ones, and that is the whole point. The
+/// hook appends the event and then a newline of its own, so the tap is
+/// `{json}\n\n{json}\n\n…` and `parse` skips the blanks. Advancing the cursor
+/// by the number of PARSED events instead would leave it short by one line per
+/// event, and `tail -n +N` would hand back events already recorded — every one
+/// of them written again on the next poll, with nothing anywhere reporting it.
+pub fn consumed_lines(raw: &str) -> usize {
+    raw.lines().count()
+}
+
 /// One observed tool call.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Observed {
@@ -265,6 +291,37 @@ mod tests {
         );
     }
 
+    /// The cursor must not re-read what it already returned.
+    ///
+    /// Off by one here is not a crash, it is a double-count: `tail -n +1` and
+    /// `tail -n +2` both return output, and the wrong one quietly records every
+    /// early tool call once per poll.
+    #[test]
+    fn the_drain_cursor_asks_for_what_it_has_not_seen() {
+        assert!(drain_from(0).contains("tail -n +1 "));
+        assert!(drain_from(3).contains("tail -n +4 "));
+        assert!(drain_from(0).contains(TAP_FILE));
+    }
+
+    /// The cursor counts LINES, not events.
+    ///
+    /// The hook writes the event and then a newline of its own, so a two-event
+    /// tap is four lines. Advancing by parsed-event count would leave the
+    /// cursor two lines short, `tail` would return both events again, and the
+    /// live drain would re-record everything it had already recorded — growing
+    /// worse the longer the turn ran, and silent throughout.
+    #[test]
+    fn the_cursor_counts_lines_not_events() {
+        let raw = concat!(
+            r#"{"tool_name":"Edit","tool_input":{"file_path":"a.rs"}}"#,
+            "\n\n",
+            r#"{"tool_name":"Bash","tool_input":{"command":"ls"}}"#,
+            "\n\n",
+        );
+        assert_eq!(parse(raw).len(), 2, "two events");
+        assert_eq!(consumed_lines(raw), 4, "…written across four lines");
+    }
+
     /// An event that is not a tool call is not a tool named "".
     #[test]
     fn a_non_tool_event_is_skipped() {