fix(viz): read the gateway's real tool_call keys, and correct the record

The frame is `{"type":"tool_call","id","name","args"}` — zeroclaw-gateway
/src/ws.rs. The tap read `tool` then `name`, and `arguments` then `input`.
`name` happened to be in the fallback chain; `args` was not in it at all,
so a container-tier tool call would have been recorded with its name and
NO path — a tool that reads as having touched nothing. `tool` and
`arguments` belong to `approval_request`, which is where they came from.

Also corrects what the histogram was read as saying. A mission turn on
gw-04 carried only chunk/done/session_start, and the first reading was
"there is no tool_call frame". Wrong: `grep -c tool_call` on the deployed
0.8.3 binary returns 46. Container-tier agents are provisioned tool-free
behind the MCP door (§15), so they call nothing — there is nothing to
observe on that tier, and nothing is broken.

That distinction is exactly what the histogram was shipped to make
possible: a tap matching no frame is otherwise indistinguishable from a
mission that used no tools.

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()),
+    //
-            None,
+    // Concurrency here is safe because `fcagent` is thread-per-connection: the
-            TURN_SECS,
+    // live log tail already relies on exactly that, on a second connection, for
-            &turn_env,
+    // the whole length of a turn. So this needs no fleet-node change.
-            run_id,
+    let turn_cmd = agent_command(&prompt, settings.as_deref());
-            Some(GUEST_LOG),
+    let turn = vm.exec_attributed(
-        )
+        &turn_cmd,
-        .await?;
+        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
+                    {
+                        // 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/topology_exec.rs

@@ -87,12 +87,19 @@ pub struct ToolTrace {
     pub calls: Vec<ToolCall>,
     /// Frame `type` values this drain did not recognise, counted.
     ///
-    /// Shipped in the same change as the tap on purpose: the frame name
+    /// Shipped in the same change as the tap on purpose: the frame name was
-    /// `tool_call` is taken from a comment in this file, not from a captured
+    /// taken from a comment in this file rather than from a captured frame. If
-    /// frame. If the runtime calls it something else, the tap records nothing
+    /// the runtime called it something else, the tap would record nothing and
-    /// and nothing anywhere errors — the World simply stays as sparse as it was
+    /// nothing anywhere would error — the World would simply stay as sparse as
-    /// before. This histogram is how one gw-04 run names the real frame instead
+    /// it was before.
-    /// of a bisect.
+    ///
+    /// MEASURED on gw-04 (v0.8.3, 2026-08-11): a mission turn's stream carried
+    /// `chunk`, `done` and `session_start` and no tool frames at all. That is
+    /// not a protocol mismatch — `tool_call` is in the deployed binary
+    /// (`zeroclaw-gateway/src/ws.rs` emits `{"type":"tool_call","id","name",
+    /// "args"}`) — it is §15: these agents are provisioned tool-free behind the
+    /// MCP door, so they call nothing. The histogram is what let us tell those
+    /// two apart, which was its whole purpose.
     pub unmatched: std::collections::BTreeMap<String, u32>,
 }
 
@@ -511,16 +518,27 @@ impl ZeroClawDriveExecutor {
                         // often enough to be believed and wrong often enough to
                         // put files on the map that nobody edited.
                         "tool_call" => {
+                            // `name` is what the gateway sends; `tool` is
+                            // what `approval_request` uses, kept as a fallback.
                             let tool = v
-                                .get("tool")
+                                .get("name")
-                                .or_else(|| v.get("name"))
+                                .or_else(|| v.get("tool"))
                                 .and_then(|t| t.as_str())
                                 .unwrap_or("")
                                 .trim()
                                 .to_string();
                             if !tool.is_empty() {
+                                // `args` FIRST: that is what the gateway
+                                // actually sends (`{"type":"tool_call","id",
+                                // "name","args"}` — zeroclaw-gateway/src/ws.rs).
+                                // The others were guesses, and a guess that
+                                // never matches costs the file path silently:
+                                // the tool call is still recorded, with no
+                                // target, and reads as a tool that touched
+                                // nothing.
                                 let args = v
-                                    .get("arguments")
+                                    .get("args")
+                                    .or_else(|| v.get("arguments"))
                                     .or_else(|| v.get("input"))
                                     .cloned()
                                     .unwrap_or(serde_json::Value::Null);
@@ -670,15 +688,17 @@ mod tests {
             let _ = socket.recv().await;
             for f in [
                 json!({"type": "session_start"}),
-                json!({"type": "tool_call", "tool": "Read",
+                // The REAL frame shape, copied from the gateway:
-                       "arguments": {"file_path": "/mission/repo/src/a.rs"}}),
+                // {"type":"tool_call","id","name","args"}.
+                json!({"type": "tool_call", "id": "t1", "name": "Read",
+                       "args": {"file_path": "/mission/repo/src/a.rs"}}),
                 // A tool whose arguments name no path at all.
-                json!({"type": "tool_call", "tool": "Bash",
+                json!({"type": "tool_call", "id": "t2", "name": "Bash",
-                       "arguments": {"command": "cargo test"}}),
+                       "args": {"command": "cargo test"}}),
                 // Prose that MENTIONS a path. It must not become a file touch.
-                json!({"type": "tool_call", "tool": "Grep",
+                json!({"type": "tool_call", "id": "t3", "name": "Grep",
                        "arguments_summary": "searching src/main.rs",
-                       "arguments": {"pattern": "fn main"}}),
+                       "args": {"pattern": "fn main"}}),
                 json!({"type": "a_frame_we_have_never_seen"}),
                 json!({"type": "a_frame_we_have_never_seen"}),
                 json!({"type": "done", "input_tokens": 1, "output_tokens": 1}),

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() {