Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d810fc0a86 | ||
|
|
31158467f4 |
@@ -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
|
||||
.exec_attributed(
|
||||
&agent_command(&prompt, settings.as_deref()),
|
||||
// The turn, and — concurrently — the tap drain that makes its work visible
|
||||
// WHILE it happens rather than an hour later.
|
||||
//
|
||||
// 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 {
|
||||
Ok(p) => crate::vm_tool_tap::parse(&p.stdout),
|
||||
// Always from the cursor, so a sink-driven turn reads only what its
|
||||
// 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
|
||||
|
||||
@@ -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| {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
/// `tool_call` is taken from a comment in this file, not from a captured
|
||||
/// frame. If the runtime calls it something else, the tap records nothing
|
||||
/// and nothing anywhere errors — the World simply stays as sparse as it was
|
||||
/// before. This histogram is how one gw-04 run names the real frame instead
|
||||
/// of a bisect.
|
||||
/// Shipped in the same change as the tap on purpose: the frame name was
|
||||
/// taken from a comment in this file rather than from a captured frame. If
|
||||
/// the runtime called it something else, the tap would record nothing and
|
||||
/// nothing anywhere would error — the World would simply stay as sparse as
|
||||
/// it was before.
|
||||
///
|
||||
/// 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")
|
||||
.or_else(|| v.get("name"))
|
||||
.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",
|
||||
"arguments": {"file_path": "/mission/repo/src/a.rs"}}),
|
||||
// The REAL frame shape, copied from the gateway:
|
||||
// {"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",
|
||||
"arguments": {"command": "cargo test"}}),
|
||||
json!({"type": "tool_call", "id": "t2", "name": "Bash",
|
||||
"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}),
|
||||
|
||||
@@ -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() {
|
||||
|
||||
Reference in New Issue
Block a user