fix(node): aborting the tail at turn end raced the flush that matters most

Composed runs streamed NOTHING while solo runs streamed fine — same code path,
`HubVms::run` is a straight passthrough, and the node logged a tail starting for
all five graph nodes with the correct outer run id. The difference was timing.

`claude -p ... | tee` makes stdout a PIPE, so the CLI block-buffers and flushes
at EXIT. The most valuable output — the agent's summary of what it did — arrives
in the instant the turn ends. The node aborted the tail the moment `handle_op`
returned, so that flush was a race: a solo turn (minutes long, output already
flushed by size) won it and streamed 337 bytes; each node of a composed run
(~20s) lost it and streamed zero.

The tail now DRAINS. A flag is set when the turn returns, and the loop exits only
after a pass that read nothing new — checked AFTER a read, never before one,
because exiting on the flag alone would drop exactly the bytes this exists to
capture. Bounded by a 20s timeout with the abort kept as a backstop rather than
the mechanism, so a VM that stopped answering cannot hold the task open.

Worth naming: 5 tails started, 5 logged cleanly, 0 bytes arrived. Every
individual step reported success and the feature did nothing — the same shape as
the empty Live tab this whole thread began with, one layer down.

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

@@ -630,6 +630,9 @@ async fn handle_frame(
                     // 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.
+                    // Set when the turn returns, so the tail can DRAIN before it
+                    // stops rather than being cut off mid-flush.
+                    let turn_done = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
                     let tail = (op == "vm_exec")
                         .then(|| {
                             let run_id = v.get("run_id").and_then(Value::as_str)?.to_string();
@@ -645,15 +648,29 @@ async fn handle_frame(
                                 run_id,
                                 log_path,
                                 out.clone(),
+                                turn_done.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.
+                    // Let the tail DRAIN, then stop. Aborting here was wrong:
+                    // `claude -p | tee` makes stdout a pipe, so the CLI block-
+                    // buffers and flushes at EXIT — the most valuable output
+                    // arrives in the instant the turn ends. Aborting raced that
+                    // flush and lost it. Measured: a solo turn (minutes long) won
+                    // the race and streamed 337 bytes; every node of a composed
+                    // run (~20s each) lost it and streamed nothing at all.
+                    //
+                    // Bounded, because a VM that stopped answering must not hold
+                    // this task open — the abort remains, as a backstop rather
+                    // than the mechanism.
                     if let Some(t) = tail {
-                        t.abort();
+                        turn_done.store(true, std::sync::atomic::Ordering::Relaxed);
+                        let drained =
+                            tokio::time::timeout(std::time::Duration::from_secs(20), t).await;
+                        if drained.is_err() {
+                            eprintln!("clawmates-node: tail drain timed out for {op}");
+                        }
                     }
                     let _ = out.send(
                         json!({ "t": "result", "id": id, "ok": ok, "output": output }).to_string(),
@@ -908,6 +925,7 @@ async fn stream_vm_log(
     run_id: String,
     log_path: String,
     out: tokio::sync::mpsc::UnboundedSender<String>,
+    turn_done: std::sync::Arc<std::sync::atomic::AtomicBool>,
 ) {
     // Said out loud at the start, because the failure this replaced was
     // invisible: the tail gave up during VM boot and logged nothing, so an empty
@@ -916,6 +934,7 @@ async fn stream_vm_log(
     let mut at: u64 = 0;
     let mut failures = 0;
     while failures < 3 {
+        let at_before = at;
         let sent = out.clone();
         let rid = run_id.clone();
         match microvm::tail_into(&vms, &vm_id, &log_path, at, move |offset, data| {
@@ -938,7 +957,14 @@ async fn stream_vm_log(
                 // not a guess about idleness.
                 at = reached;
                 failures = 0;
-                tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+                // The turn has returned AND this pass read nothing new: the
+                // final flush is already in hand, so stop. Checked after a read,
+                // never before one — exiting on the flag alone would drop
+                // exactly the bytes this exists to capture.
+                if turn_done.load(std::sync::atomic::Ordering::Relaxed) && reached == at_before {
+                    return;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(300)).await;
             }
             Err(e) => {
                 failures += 1;