osobh/clawmates
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(viz): what the agents actually did, as structured events

Browse Source 
The World could draw a mission's shape but nothing about the work. The
detail existed only as prose in checkpoint.log and model output, where a
tool name is indistinguishable from an agent *talking about* a tool — so
it was never parsed, deliberately. `mission_events` is the structured
channel that replaces it.

Three taps, one table:

- Container tier: the `_ => {}` at the end of topology_exec's typed frame
  stream now matches `tool_call` and reads the tool's JSON ARGUMENTS for a
  path. Never the prose summary — a path scraped from a sentence would put
  files on the map that no agent opened, and the test proves a Grep whose
  summary says "src/main.rs" produces no file touch. The frame name itself
  is unverified, so the same commit ships an unmatched-frame-type
  histogram: a tap that matches nothing looks exactly like a mission that
  used no tools, and this is how one gw-04 run names the real frame.

- microVM tier: a `PostToolUse` hook, the seam vm_stop_gate already proved
  fires under `claude -p`. It copies stdin to /root/tap and exits 0
  unconditionally — a non-zero PostToolUse hook talks back to the model,
  which would turn the observer into a participant. Drained before collect,
  since the VM is destroyed moments later.

- Phase transitions: five identical copies of the pending→running UPDATE
  became one `mark_phase_running`, and `close_finished_phases` grew
  RETURNING. Its CASE decides each phase's status inside SQL from rows the
  statement does not change, so it cannot be re-derived afterwards without
  writing that CASE twice — without RETURNING it emits zero phase.completed
  and reports success.

The settings.json hazard the plan called out: the stop gate wrote the
WHOLE document, so a second hook writer would have silently erased it and
a coding phase would then complete having written nothing — the exact
failure the gate exists to catch. There is now one composer,
`vm_tool_tap::guest_settings`, one writer, and a source-walk test that
fails if anything else writes a settings document.

`mission_events.run_id` carries no FK on purpose: phase_runner DELETEs
topology_runs on retry, and a cascade would erase a phase's whole history
the moment it retried — silently, since a cascade is not an error.

world.rs streams it with a cursor that separates backfill from motion.
Everything already in the table when a subscriber arrives is drawn as
settled history; only what lands afterwards animates. Otherwise opening a
finished mission replays an hour of tool calls as a burst storm.

Bounded twice: 400 events per phase (enforced inside the INSERT, since
two concurrent taps would each read a count below the cap) and a 7-day
retention sweep in mission_gc.

Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Omar Sobh
2026-08-11 09:16:50 -07:00
co-authored by Claude Opus 5
parent c2fa8067e1
commit 9e61e3ba35
13 changed files with 1257 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/cm-api/src/lib.rs
+2
View File
@@ -23,12 +23,14 @@ pub mod corpus;
pub mod harvest; pub mod harvest;
pub mod library; pub mod library;
pub mod mission_delivery; pub mod mission_delivery;
pub mod mission_events;
pub mod microvm_client; pub mod microvm_client;
pub mod microvm_executor; pub mod microvm_executor;
pub mod microvm_turn_executor; pub mod microvm_turn_executor;
pub mod subscription; pub mod subscription;
pub mod vm_placement; pub mod vm_placement;
pub mod vm_stop_gate; pub mod vm_stop_gate;
pub mod vm_tool_tap;
pub mod mission_fs; pub mod mission_fs;
pub mod mission_outputs; pub mod mission_outputs;
pub mod papers; pub mod papers;
crates/cm-api/src/microvm_executor.rs
+99 -11
View File
@@ -386,6 +386,13 @@ pub struct VmOutcome {
/// `Some(true)` is a FAILED turn. The gate is the only thing that ever runs /// `Some(true)` is a FAILED turn. The gate is the only thing that ever runs
/// a `done_when_check`, so if it gave up, nothing downstream will notice. /// a `done_when_check`, so if it gave up, nothing downstream will notice.
pub released_at_cap: Option<bool>, pub released_at_cap: Option<bool>,
/// What the agent's tools touched, drained from the guest tap.
///
/// Empty when no tap was installed, when the hook never fired, or when the
/// agent genuinely called nothing. Those are three different facts and this
/// field cannot tell them apart — the unmatched-frame log and the install
/// error are what separate them.
pub tools: Vec<crate::vm_tool_tap::Observed>,
} }
/// Boot a VM, run the phase in it, collect the result, and destroy it. /// Boot a VM, run the phase in it, collect the result, and destroy it.
@@ -579,15 +586,17 @@ async fn run_inside(
// to "no gate" and logged: the gate makes a phase converge in ONE VM instead // to "no gate" and logged: the gate makes a phase converge in ONE VM instead
// of a second one, so losing it costs money and time — while failing the // of a second one, so losing it costs money and time — while failing the
// turn over it would cost the work. // turn over it would cost the work.
let settings = match gate { // Asked ONCE, of the binary rather than of a version number, and shared by
// both hooks: `--settings` is how either of them gets installed at all.
let settings_supported = vm
.exec(SETTINGS_PROBE, None, 60, &[])
.await
.map(|p| p.stdout.contains("SETTINGS-OK"))
.unwrap_or(false);
let gate_dir = match gate {
None => None, None => None,
Some(g) => { Some(g) => {
let supported = vm if !settings_supported {
.exec(SETTINGS_PROBE, None, 60, &[])
.await
.map(|p| p.stdout.contains("SETTINGS-OK"))
.unwrap_or(false);
if !supported {
eprintln!( eprintln!(
"microvm_executor: {} has no `claude --settings`, so the stop gate \ "microvm_executor: {} has no `claude --settings`, so the stop gate \
cannot be installed — this phase runs ungated and its completion \ cannot be installed — this phase runs ungated and its completion \
@@ -598,10 +607,7 @@ async fn run_inside(
} else { } else {
let install = g.install_command(GUEST_REPO, crate::vm_stop_gate::GATE_DIR); let install = g.install_command(GUEST_REPO, crate::vm_stop_gate::GATE_DIR);
match vm.exec(&install, None, 60, &[]).await { match vm.exec(&install, None, 60, &[]).await {
Ok(o) if o.rc == 0 => Some(format!( Ok(o) if o.rc == 0 => Some(crate::vm_stop_gate::GATE_DIR),
"{}/settings.json",
crate::vm_stop_gate::GATE_DIR
)),
Ok(o) => { Ok(o) => {
eprintln!( eprintln!(
"microvm_executor: could not install the stop gate on {} \ "microvm_executor: could not install the stop gate on {} \
@@ -625,6 +631,63 @@ async fn run_inside(
} }
}; };
// The tool tap, on the same `--settings` seam as the gate. Best-effort in
// exactly the same way: a phase that runs without telemetry is a phase that
// still delivers, and failing the turn to protect a picture would be the
// wrong trade.
let tap_dir = match settings_supported {
false => None,
true => match vm
.exec(&crate::vm_tool_tap::install_command(crate::vm_tool_tap::TAP_DIR), None, 60, &[])
.await
{
Ok(o) if o.rc == 0 => Some(crate::vm_tool_tap::TAP_DIR),
Ok(o) => {
eprintln!(
"microvm_executor: could not install the tool tap on {} (rc={}): {} \
— this phase's actions will not appear in the World",
vm.vm_id(),
o.rc,
o.stderr
);
None
}
Err(e) => {
eprintln!(
"microvm_executor: could not install the tool tap on {}: {e} \
— this phase's actions will not appear in the World",
vm.vm_id()
);
None
}
},
};
// ONE settings document, written once, carrying whichever hooks installed.
// Two writers here is the silent clobber `guest_settings` exists to stop:
// whichever ran second would erase the other's hook with no error at all.
let settings = match (gate_dir, tap_dir) {
(None, None) => None,
(g, t) => {
let doc = crate::vm_tool_tap::guest_settings(g, t);
let cmd = crate::vm_tool_tap::settings_command(
crate::vm_tool_tap::SETTINGS_PATH,
&doc,
);
match vm.exec(&cmd, None, 60, &[]).await {
Ok(o) if o.rc == 0 => Some(crate::vm_tool_tap::SETTINGS_PATH.to_string()),
other => {
eprintln!(
"microvm_executor: could not write the guest settings on {} ({other:?}) \
— running with NO hooks: neither the stop gate nor the tool tap",
vm.vm_id()
);
None
}
}
}
};
let out = vm let out = vm
.exec_attributed( .exec_attributed(
&agent_command(&prompt, settings.as_deref()), &agent_command(&prompt, settings.as_deref()),
@@ -636,6 +699,30 @@ async fn run_inside(
) )
.await?; .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),
Err(e) => {
eprintln!("microvm_executor: tap drain failed on {}: {e}", vm.vm_id());
Vec::new()
}
},
};
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
// next person debugging an empty World has no thread to pull.
eprintln!(
"microvm_executor: {} installed the tool tap and drained ZERO tool calls — \
either the agent called no tools or `PostToolUse` did not fire in this image",
vm.vm_id()
);
}
// Ask the guest how many subagents ran, before collecting: the transcripts // Ask the guest how many subagents ran, before collecting: the transcripts
// live in /root, outside the collected tree, so this is the only chance. // live in /root, outside the collected tree, so this is the only chance.
// Failure to probe is `None`, never 0 — "we could not look" and "it delegated // Failure to probe is `None`, never 0 — "we could not look" and "it delegated
@@ -732,6 +819,7 @@ async fn run_inside(
teammates, teammates,
stop_blocks, stop_blocks,
released_at_cap, released_at_cap,
tools,
}) })
} }
crates/cm-api/src/microvm_turn_executor.rs
+17
View File
@@ -214,6 +214,19 @@ impl<V: PhaseVm> TurnExecutor for MicroVmTurnExecutor<V> {
OrchestratorError::Executor(format!("node {} in a microVM: {e}", req.node_id)) OrchestratorError::Executor(format!("node {} in a microVM: {e}", req.node_id))
})?; })?;
// Recorded BEFORE the failure branches below. A node that could not be
// collected, or whose gate capped, still touched files — and on this
// path those touches are the only account of what it did, since the
// work never reached a diff.
crate::phase_runner::record_vm_tools(
&self.pool,
self.mission_id,
self.phase_id,
self.run_id,
&outcome.tools,
)
.await;
// A node whose work never came back must fail the run rather than hand // A node whose work never came back must fail the run rather than hand
// the next node a tree missing the previous one's edits. On this path an // the next node a tree missing the previous one's edits. On this path an
// uncollected turn is worse than on the solo one: the loss is silent, // uncollected turn is worse than on the solo one: the loss is silent,
@@ -399,6 +412,7 @@ mod tests {
teammates: None, teammates: None,
stop_blocks: None, stop_blocks: None,
released_at_cap: None, released_at_cap: None,
tools: Vec::new(),
}) })
} }
} }
@@ -593,6 +607,7 @@ mod tests {
teammates: None, teammates: None,
stop_blocks: None, stop_blocks: None,
released_at_cap: None, released_at_cap: None,
tools: Vec::new(),
}) })
} }
} }
@@ -624,6 +639,7 @@ mod tests {
teammates: None, teammates: None,
stop_blocks: Some(crate::vm_stop_gate::MAX_BLOCKS), stop_blocks: Some(crate::vm_stop_gate::MAX_BLOCKS),
released_at_cap: Some(true), released_at_cap: Some(true),
tools: Vec::new(),
}) })
} }
} }
@@ -650,6 +666,7 @@ mod tests {
teammates: None, teammates: None,
stop_blocks: Some(crate::vm_stop_gate::MAX_BLOCKS), stop_blocks: Some(crate::vm_stop_gate::MAX_BLOCKS),
released_at_cap: Some(false), released_at_cap: Some(false),
tools: Vec::new(),
}) })
} }
} }
crates/cm-api/src/mission_events.rs
+229
View File
@@ -0,0 +1,229 @@
//! Structured mission activity — the channel that replaced parsing prose.
//!
//! The operator decision behind this module: action detail comes from
//! **structured events at the source**, never from `checkpoint.log` or model
//! output. A tool name in a log line is indistinguishable from an agent
//! *discussing* a tool, and a visualization built on that distinction reads as
//! confident fact while being partly fiction.
//!
//! Everything here is best-effort. A mission must not fail because its
//! telemetry could not be written — so every write logs and swallows. That is a
//! deliberate exception to this codebase's usual rule, and it is bounded: the
//! only thing lost is detail in a picture.
use serde_json::Value;
use sqlx::PgPool;
use uuid::Uuid;
/// A phase entered `running`.
pub const PHASE_STARTED: &str = "phase.started";
/// A phase reached a terminal state. `detail.status` says which.
pub const PHASE_COMPLETED: &str = "phase.completed";
/// An agent called a tool. `target` is the tool name.
pub const TOOL_CALL: &str = "tool.call";
/// A tool touched a path. `target` is the path, repo-relative where known.
pub const FILE_TOUCH: &str = "file.touch";
/// Most events one phase may record.
///
/// A capped stream that says so beats an uncapped one that quietly becomes the
/// largest table in the database: a coding phase can call thousands of tools,
/// and every one of them would be replayed to every World subscriber. Past the
/// cap the picture is already complete — nobody reads the four-thousandth file
/// orb.
pub const PER_PHASE_CAP: i64 = 400;
/// One recorded event.
#[derive(Debug, Clone, Default)]
pub struct MissionEvent {
pub mission_id: Uuid,
pub phase_id: Option<Uuid>,
pub run_id: Option<Uuid>,
pub agent_id: Option<Uuid>,
pub kind: String,
pub target: Option<String>,
pub detail: Value,
}
impl MissionEvent {
pub fn new(mission_id: Uuid, kind: &str) -> Self {
MissionEvent {
mission_id,
kind: kind.to_string(),
detail: Value::Null,
..Default::default()
}
}
pub fn phase(mut self, id: Uuid) -> Self {
self.phase_id = Some(id);
self
}
pub fn run(mut self, id: Uuid) -> Self {
self.run_id = Some(id);
self
}
pub fn agent(mut self, id: Option<Uuid>) -> Self {
self.agent_id = id;
self
}
pub fn target(mut self, t: impl Into<String>) -> Self {
self.target = Some(t.into());
self
}
pub fn detail(mut self, d: Value) -> Self {
self.detail = d;
self
}
}
/// Record one event, best-effort.
///
/// The per-phase cap is enforced in the INSERT itself rather than by a read
/// followed by a write: two tool taps writing concurrently would both read a
/// count below the cap and both insert, and the cap would drift by however many
/// writers there are. `INSERT … SELECT … WHERE (subquery) < cap` makes the
/// decision inside the statement.
pub async fn record(pool: &PgPool, e: MissionEvent) {
let detail = if e.detail.is_null() {
Value::Object(Default::default())
} else {
e.detail
};
let res = sqlx::query(
"INSERT INTO mission_events
(mission_id, phase_id, run_id, agent_id, kind, target, detail)
SELECT $1, $2, $3, $4, $5, $6, $7
WHERE $2::uuid IS NULL
OR (SELECT count(*) FROM mission_events WHERE phase_id = $2) < $8",
)
.bind(e.mission_id)
.bind(e.phase_id)
.bind(e.run_id)
.bind(e.agent_id)
.bind(&e.kind)
.bind(&e.target)
.bind(&detail)
.bind(PER_PHASE_CAP)
.execute(pool)
.await;
if let Err(err) = res {
eprintln!("mission_events: record {} failed: {err}", e.kind);
}
}
/// Record several events under one round trip's worth of intent.
pub async fn record_all(pool: &PgPool, events: Vec<MissionEvent>) {
for e in events {
record(pool, e).await;
}
}
/// The path a tool's **arguments** name, if any.
///
/// Reads the arguments as JSON — never the tool's prose summary. The summary is
/// a sentence written for a human; a path pulled out of it by regex would be
/// right often enough to be trusted and wrong often enough to matter.
///
/// The key names are the ones Claude Code and the ZeroClaw tools actually use.
/// An unrecognised shape returns `None`, which renders as a tool call with no
/// file — accurate, rather than a guess at which argument was a path.
pub fn tool_path(args: &Value) -> Option<String> {
const KEYS: [&str; 6] = [
"file_path",
"filePath",
"path",
"notebook_path",
"file",
"target_file",
];
let obj = args.as_object()?;
for k in KEYS {
if let Some(s) = obj.get(k).and_then(Value::as_str) {
let s = s.trim();
if !s.is_empty() {
return Some(s.to_string());
}
}
}
None
}
/// Strip the guest/host workspace prefix so a path is repo-relative.
///
/// Tool arguments are absolute inside the sandbox (`/mission/repo/src/a.rs`).
/// Left alone, every mission's file tree would nest under a `mission` → `repo`
/// pair of directory orbs that exist in no repository and mean nothing to the
/// person reading the map.
pub fn repo_relative(path: &str, roots: &[&str]) -> String {
let p = path.trim();
for root in roots {
let root = root.trim_end_matches('/');
if let Some(rest) = p.strip_prefix(root) {
let rest = rest.trim_start_matches('/');
if !rest.is_empty() {
return rest.to_string();
}
}
}
p.trim_start_matches("./").to_string()
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
/// Paths come from arguments, and only from argument keys we know.
///
/// The alternative — scanning the values for anything that looks like a
/// path — is what makes a viz confidently wrong: a `pattern` of `*.rs` or a
/// `command` of `ls src/` would both become "the agent edited a file".
#[test]
fn a_path_comes_from_a_known_argument_or_not_at_all() {
assert_eq!(
tool_path(&json!({"file_path": "/mission/repo/src/a.rs"})).as_deref(),
Some("/mission/repo/src/a.rs")
);
assert_eq!(tool_path(&json!({"path": "docs/x.md"})).as_deref(), Some("docs/x.md"));
// A shell command mentions paths and touches none we can name.
assert_eq!(tool_path(&json!({"command": "ls src/"})), None);
// A glob is a query, not a file.
assert_eq!(tool_path(&json!({"pattern": "**/*.rs"})), None);
// Blank is absence, not a file called "".
assert_eq!(tool_path(&json!({"file_path": " "})), None);
assert_eq!(tool_path(&json!("not an object")), None);
}
/// The sandbox prefix must not become two directory orbs in every mission.
#[test]
fn paths_are_made_repo_relative() {
let roots = ["/mission/repo", "/workspace"];
assert_eq!(repo_relative("/mission/repo/src/a.rs", &roots), "src/a.rs");
assert_eq!(repo_relative("/workspace/README.md", &roots), "README.md");
assert_eq!(repo_relative("./src/a.rs", &roots), "src/a.rs");
// Outside every root, it is left alone rather than mangled.
assert_eq!(repo_relative("/etc/hosts", &roots), "/etc/hosts");
// The root ITSELF is not a file, so it must not collapse to "".
assert_eq!(repo_relative("/mission/repo", &roots), "/mission/repo");
}
/// The cap must be decided inside the INSERT.
///
/// A count-then-insert is the classic version of this and it is wrong here:
/// the container tap and the microVM drain both write for the same phase,
/// and each would see a count below the cap and insert. Nothing errors —
/// the table simply grows past the bound that exists to hold it.
#[test]
fn the_cap_is_enforced_in_one_statement() {
let src = include_str!("mission_events.rs");
let body = src
.split("pub async fn record(")
.nth(1)
.and_then(|s| s.split("pub async fn").next())
.expect("record body");
assert!(
body.contains("INSERT INTO mission_events") && body.contains("SELECT count(*)"),
"the cap must be a subquery in the INSERT, not a separate read"
);
}
}
crates/cm-api/src/mission_gc.rs
+37 -1
View File
@@ -84,6 +84,8 @@ pub struct Reclaimed {
/// Directories we tried and failed to remove. Reported rather than swallowed /// Directories we tried and failed to remove. Reported rather than swallowed
/// — a GC that cannot collect is the thing being fixed. /// — a GC that cannot collect is the thing being fixed.
pub failed: u64, pub failed: u64,
/// Rows swept from `mission_events`.
pub events: u64,
} }
impl Reclaimed { impl Reclaimed {
@@ -96,10 +98,12 @@ impl std::fmt::Display for Reclaimed {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!( write!(
f, f,
"reclaimed {} orphan mission dir(s), {} scratch dir(s), {} output(s), {:.1} MiB{}", "reclaimed {} orphan mission dir(s), {} scratch dir(s), {} output(s), \
{} mission event(s), {:.1} MiB{}",
self.orphan_dirs, self.orphan_dirs,
self.scratch_dirs, self.scratch_dirs,
self.outputs, self.outputs,
self.events,
self.bytes as f64 / (1024.0 * 1024.0), self.bytes as f64 / (1024.0 * 1024.0),
if self.failed > 0 { if self.failed > 0 {
format!(" — {} COULD NOT BE REMOVED", self.failed) format!(" — {} COULD NOT BE REMOVED", self.failed)
@@ -116,9 +120,41 @@ async fn sweep_once(pool: &PgPool) -> Result<Reclaimed, String> {
reap_orphan_missions(pool, &root, &mut out).await?; reap_orphan_missions(pool, &root, &mut out).await?;
reap_scratch(&root, &mut out).await; reap_scratch(&root, &mut out).await;
reap_outputs(pool, &root, &mut out).await; reap_outputs(pool, &root, &mut out).await;
reap_mission_events(pool, &mut out).await;
Ok(out) Ok(out)
} }
/// How long a mission's structured activity is kept.
///
/// The World shows the last 24 hours of finished missions, so a week is
/// generous and still bounds a table that a single busy coding phase can add
/// hundreds of rows to. The per-phase cap bounds ONE phase; this bounds time.
const EVENT_RETENTION_DAYS: i32 = 7;
/// Sweep expired `mission_events`.
///
/// Bounded per pass rather than deleting the whole backlog in one statement: a
/// deployment that has been accumulating for months would otherwise take a long
/// lock on its first sweep after this ships. The sweep runs on a timer, so a
/// large backlog simply drains over several passes.
async fn reap_mission_events(pool: &PgPool, out: &mut Reclaimed) {
let res = sqlx::query(
"DELETE FROM mission_events
WHERE id IN (
SELECT id FROM mission_events
WHERE created_at < now() - make_interval(days => $1)
LIMIT 10000
)",
)
.bind(EVENT_RETENTION_DAYS)
.execute(pool)
.await;
match res {
Ok(r) => out.events += r.rows_affected(),
Err(e) => eprintln!("mission_gc: sweeping mission_events failed: {e}"),
}
}
/// Directories under the missions root with no mission row. /// Directories under the missions root with no mission row.
async fn reap_orphan_missions( async fn reap_orphan_missions(
pool: &PgPool, pool: &PgPool,
crates/cm-api/src/phase_runner.rs
+139 -49
View File
@@ -895,15 +895,7 @@ async fn launch_phase(
} }
// Flip phase to running. // Flip phase to running.
sqlx::query( mark_phase_running(pool, mission_id, phase_id).await?;
"UPDATE mission_phases
SET status = 'running', started_at = now()
WHERE id = $1 AND status = 'pending'",
)
.bind(phase_id)
.execute(pool)
.await
.map_err(|e| format!("mark phase {phase_id} running: {e}"))?;
eprintln!( eprintln!(
"phase_runner: mission {mission_id} phase {phase_id} ({kind}) launched with {} team(s)", "phase_runner: mission {mission_id} phase {phase_id} ({kind}) launched with {} team(s)",
team_rows.len() team_rows.len()
@@ -981,13 +973,7 @@ async fn launch_composed_microvm_phase(
.execute(pool) .execute(pool)
.await; .await;
// The phase must still leave `pending`, or the sweep re-launches it. // The phase must still leave `pending`, or the sweep re-launches it.
let _ = sqlx::query( let _ = mark_phase_running(pool, mission_id, phase_id).await;
"UPDATE mission_phases SET status = 'running', started_at = now()
WHERE id = $1 AND status = 'pending'",
)
.bind(phase_id)
.execute(pool)
.await;
Ok(()) Ok(())
}; };
@@ -1055,15 +1041,7 @@ async fn launch_composed_microvm_phase(
.await .await
.map_err(|e| format!("enqueue composed run for phase {phase_id}: {e}"))?; .map_err(|e| format!("enqueue composed run for phase {phase_id}: {e}"))?;
sqlx::query( mark_phase_running(pool, mission_id, phase_id).await?;
"UPDATE mission_phases
SET status = 'running', started_at = now()
WHERE id = $1 AND status = 'pending'",
)
.bind(phase_id)
.execute(pool)
.await
.map_err(|e| format!("mark phase {phase_id} running: {e}"))?;
eprintln!( eprintln!(
"phase_runner: mission {mission_id} phase {phase_id} queued as a COMPOSED run \ "phase_runner: mission {mission_id} phase {phase_id} queued as a COMPOSED run \
@@ -1130,15 +1108,7 @@ async fn launch_microvm_phase(
.await .await
.map_err(|e| format!("enqueue microvm run for phase {phase_id}: {e}"))?; .map_err(|e| format!("enqueue microvm run for phase {phase_id}: {e}"))?;
sqlx::query( mark_phase_running(pool, mission_id, phase_id).await?;
"UPDATE mission_phases
SET status = 'running', started_at = now()
WHERE id = $1 AND status = 'pending'",
)
.bind(phase_id)
.execute(pool)
.await
.map_err(|e| format!("mark phase {phase_id} running: {e}"))?;
let repo = crate::mission_workspace::checkout_path(mission_id); let repo = crate::mission_workspace::checkout_path(mission_id);
let task = task.to_string(); let task = task.to_string();
@@ -1224,6 +1194,13 @@ async fn launch_microvm_phase(
Ok(o) => o.stop_blocks.map(|n| n.to_string()).unwrap_or_else(|| "-".into()), Ok(o) => o.stop_blocks.map(|n| n.to_string()).unwrap_or_else(|| "-".into()),
Err(_) => "-".into(), Err(_) => "-".into(),
}; };
// What the agent touched inside the VM, recorded before the outcome is
// 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.
if let Ok(o) = &outcome {
record_vm_tools(&pool2, mission_id, phase_id, run_id, &o.tools).await;
}
let (status, note) = match outcome { let (status, note) = match outcome {
// The gate gave up. It is the ONLY thing that runs a // The gate gave up. It is the ONLY thing that runs a
// `done_when_check`, so a release at the cap means the phase's own // `done_when_check`, so a release at the cap means the phase's own
@@ -1344,15 +1321,7 @@ async fn launch_direct_session(
.await .await
.map_err(|e| format!("enqueue session run for phase {phase_id}: {e}"))?; .map_err(|e| format!("enqueue session run for phase {phase_id}: {e}"))?;
sqlx::query( mark_phase_running(pool, mission_id, phase_id).await?;
"UPDATE mission_phases
SET status = 'running', started_at = now()
WHERE id = $1 AND status = 'pending'",
)
.bind(phase_id)
.execute(pool)
.await
.map_err(|e| format!("mark phase {phase_id} running: {e}"))?;
let container = crate::mission_runtime::container_name(mission_id); let container = crate::mission_runtime::container_name(mission_id);
let task = task.to_string(); let task = task.to_string();
@@ -1521,6 +1490,83 @@ fn phase_task_text(
) )
} }
/// Record a microVM turn's drained tool tap.
///
/// `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.
pub(crate) async fn record_vm_tools(
pool: &PgPool,
mission_id: Uuid,
phase_id: Uuid,
run_id: Uuid,
tools: &[crate::vm_tool_tap::Observed],
) {
if tools.is_empty() {
return;
}
let mut events = Vec::new();
for t in tools {
events.push(crate::mission_events::MissionEvent {
mission_id,
phase_id: Some(phase_id),
run_id: Some(run_id),
agent_id: None,
kind: crate::mission_events::TOOL_CALL.to_string(),
target: Some(t.tool.clone()),
detail: serde_json::Value::Null,
});
if let Some(path) = &t.path {
events.push(crate::mission_events::MissionEvent {
mission_id,
phase_id: Some(phase_id),
run_id: Some(run_id),
agent_id: None,
kind: crate::mission_events::FILE_TOUCH.to_string(),
target: Some(crate::mission_events::repo_relative(
path,
&["/mission/repo", "/workspace"],
)),
detail: serde_json::json!({ "tool": t.tool }),
});
}
}
crate::mission_events::record_all(pool, events).await;
}
/// Flip a phase from `pending` to `running`, and say so on the wire.
///
/// The `WHERE … AND status = 'pending'` guard means this UPDATE is a claim, not
/// an assignment: a phase already claimed by another launcher matches nothing.
/// `RETURNING` is what turns that into information — without it the statement
/// reports the same `Ok(())` whether it started a phase or lost the race, and
/// the World would announce a start that never happened.
///
/// Five copies of this UPDATE existed, one per launch path. They were identical
/// and independently maintained, which is how a sixth path would have been
/// written with no event at all.
async fn mark_phase_running(pool: &PgPool, mission_id: Uuid, phase_id: Uuid) -> Result<(), String> {
let claimed = sqlx::query(
"UPDATE mission_phases
SET status = 'running', started_at = now()
WHERE id = $1 AND status = 'pending'
RETURNING id",
)
.bind(phase_id)
.fetch_optional(pool)
.await
.map_err(|e| format!("mark phase {phase_id} running: {e}"))?;
if claimed.is_some() {
crate::mission_events::record(
pool,
crate::mission_events::MissionEvent::new(mission_id, crate::mission_events::PHASE_STARTED)
.phase(phase_id),
)
.await;
}
Ok(())
}
/// Close phases whose topology_runs are all terminal. /// Close phases whose topology_runs are all terminal.
/// ///
/// A phase that declares a `done_when` condition lands in `evaluating` instead /// A phase that declares a `done_when` condition lands in `evaluating` instead
@@ -1529,7 +1575,7 @@ fn phase_task_text(
/// — there is nothing to evaluate — and a phase with no condition completes /// — there is nothing to evaluate — and a phase with no condition completes
/// exactly as it always did, so untouched missions are unaffected. /// exactly as it always did, so untouched missions are unaffected.
async fn close_finished_phases(pool: &PgPool) -> Result<(), String> { async fn close_finished_phases(pool: &PgPool) -> Result<(), String> {
sqlx::query( let closed = sqlx::query(
"UPDATE mission_phases mp "UPDATE mission_phases mp
SET status = SET status =
CASE CASE
@@ -1563,11 +1609,38 @@ async fn close_finished_phases(pool: &PgPool) -> Result<(), String> {
WHERE r.mission_phase_id = mp.id WHERE r.mission_phase_id = mp.id
AND r.iteration = mp.iteration AND r.iteration = mp.iteration
AND r.status NOT IN ('completed', 'failed', 'cancelled') AND r.status NOT IN ('completed', 'failed', 'cancelled')
)", )
RETURNING mp.id, mp.mission_id, mp.status",
) )
.execute(pool) .fetch_all(pool)
.await .await
.map_err(|e| format!("close finished phases: {e}"))?; .map_err(|e| format!("close finished phases: {e}"))?;
// `RETURNING` rather than a follow-up SELECT, and this is the only way to
// get these rows: the `CASE` decides each phase's status INSIDE the
// statement, from `topology_runs` rows whose state the statement itself
// does not change — so re-deriving it afterwards would be a second
// implementation of that CASE, free to disagree with the first. Without it
// this function emits zero `phase.completed` events and reports success.
for row in closed {
let phase_id: Uuid = row.get("id");
let mission_id: Uuid = row.get("mission_id");
let status: String = row.get("status");
// `evaluating` is not terminal — the judge has not spoken yet — so it
// is not a completion. `evaluate_finished_phases` closes those.
if status == "evaluating" {
continue;
}
crate::mission_events::record(
pool,
crate::mission_events::MissionEvent::new(
mission_id,
crate::mission_events::PHASE_COMPLETED,
)
.phase(phase_id)
.detail(serde_json::json!({ "status": status })),
)
.await;
}
Ok(()) Ok(())
} }
@@ -1626,15 +1699,32 @@ async fn evaluate_finished_phases(
// correctly refused a phase whose suite had a failing test, and the // correctly refused a phase whose suite had a failing test, and the
// mission still closed `completed`. // mission still closed `completed`.
let status = if verdict.met { "completed" } else { "failed" }; let status = if verdict.met { "completed" } else { "failed" };
sqlx::query( let closed = sqlx::query(
"UPDATE mission_phases SET status = $2, completed_at = now() "UPDATE mission_phases SET status = $2, completed_at = now()
WHERE id = $1 AND status = 'evaluating'", WHERE id = $1 AND status = 'evaluating'
RETURNING id",
) )
.bind(phase_id) .bind(phase_id)
.bind(status) .bind(status)
.execute(pool) .fetch_optional(pool)
.await .await
.map_err(|e| format!("close phase {phase_id}: {e}"))?; .map_err(|e| format!("close phase {phase_id}: {e}"))?;
if closed.is_some() {
crate::mission_events::record(
pool,
crate::mission_events::MissionEvent::new(
mission_id,
crate::mission_events::PHASE_COMPLETED,
)
.phase(phase_id)
.detail(serde_json::json!({
"status": status,
"judge": verdict.model,
"reason": verdict.reason,
})),
)
.await;
}
eprintln!( eprintln!(
"phase_runner: phase {phase_id} ({kind}) {status} after {} pass(es) — met={} \ "phase_runner: phase {phase_id} ({kind}) {status} after {} pass(es) — met={} \
(judge={}, independent={}) — {}", (judge={}, independent={}) — {}",
crates/cm-api/src/routes/world.rs
+125
View File
@@ -228,6 +228,61 @@ async fn world_files(pool: &PgPool, ws: WorkspaceId, only: Option<Uuid>) -> Vec<
.collect() .collect()
} }
/// One row of `mission_events` — what an agent actually did.
struct ActRow {
id: i64,
mission_id: String,
phase_id: Option<String>,
agent_id: Option<String>,
kind: String,
target: String,
}
/// Structured mission activity since `after`, oldest first.
///
/// `after < 0` means "the first pass has not run yet": everything is returned
/// so the caller can seed its cursor and draw the backlog as settled history.
async fn world_acts(pool: &PgPool, ws: WorkspaceId, only: Option<Uuid>, after: i64) -> Vec<ActRow> {
let rows = sqlx::query(
"SELECT e.id,
e.mission_id::text AS mission_id,
e.phase_id::text AS phase_id,
e.agent_id::text AS agent_id,
e.kind,
e.target
FROM mission_events e
JOIN missions m ON m.id = e.mission_id
WHERE m.workspace_id = $1
AND e.kind IN ('tool.call', 'file.touch')
AND e.target IS NOT NULL
AND e.id > $2
AND ( m.status = 'running'
OR ( m.status IN ('completed', 'failed')
AND m.completed_at > now() - interval '24 hours' ) )
AND ($3::uuid IS NULL OR m.id = $3)
ORDER BY e.id
LIMIT 500",
)
.bind(ws.as_uuid())
.bind(after)
.bind(only)
.fetch_all(pool)
.await
.unwrap_or_default();
rows.into_iter()
.filter_map(|r| {
Some(ActRow {
id: r.get::<i64, _>("id"),
mission_id: r.get::<String, _>("mission_id"),
phase_id: r.get::<Option<String>, _>("phase_id"),
agent_id: r.get::<Option<String>, _>("agent_id"),
kind: r.get::<String, _>("kind"),
target: r.get::<Option<String>, _>("target")?,
})
})
.collect()
}
/// Agents that execute a phase of this kind, via the purposes the phase runner /// Agents that execute a phase of this kind, via the purposes the phase runner
/// itself uses. Returns empty for a teamless (microVM) mission. /// itself uses. Returns empty for a teamless (microVM) mission.
async fn phase_agents( async fn phase_agents(
@@ -550,6 +605,13 @@ pub async fn world_live(
// (phase, path) pairs already announced — a delivered file is a fact // (phase, path) pairs already announced — a delivered file is a fact
// that happened once, not a recurring event. // that happened once, not a recurring event.
let mut last_file: HashSet<String> = HashSet::new(); let mut last_file: HashSet<String> = HashSet::new();
// `mission_events` cursor. -1 until the first pass seeds it, which is
// what separates BACKFILL from MOTION: everything already in the table
// when a subscriber arrives is history and is drawn as a settled map,
// and only what lands afterwards is animated. Without the distinction,
// opening a finished mission would replay an hour of tool calls as a
// burst storm and read as a mission that just did all of it at once.
let mut event_cursor: i64 = -1;
// Audit-log cursor for edge-initiated inter-agent events (delegation, // Audit-log cursor for edge-initiated inter-agent events (delegation,
// A2A) that bypass the run loop. -1 until seeded on the first pass. // A2A) that bypass the run loop. -1 until seeded on the first pass.
let mut audit_cursor: i64 = -1; let mut audit_cursor: i64 = -1;
@@ -730,6 +792,69 @@ pub async fn world_live(
); );
} }
// Structured activity — the motion channel. Tool calls and file
// touches recorded at the source by the container tap and the
// microVM `PostToolUse` hook. Never parsed from prose: a tool name
// in a log is indistinguishable from an agent TALKING about a tool.
{
let backfill = event_cursor < 0;
for a in world_acts(&pool, ws, only_mission, event_cursor.max(0)).await {
event_cursor = event_cursor.max(a.id);
match a.kind.as_str() {
"file.touch" => {
let key = format!("{}|{}", a.phase_id.as_deref().unwrap_or(""), a.target);
if !last_file.insert(key) {
continue;
}
yield sse(
"mission.file",
json!({
"missionId": a.mission_id,
"phaseId": a.phase_id,
"agentId": a.agent_id,
"path": a.target,
// Backlog is history: it draws the file and
// stops there. Only what lands while someone
// is watching is motion.
"source": if backfill { "diff" } else { "tool" },
}),
);
}
// A tool call with an agent is a pawn leaving its
// station and coming back; without one (a microVM phase
// has no platform agent) it is only a node lighting up,
// which is the truth rather than an invented traveller.
_ if !backfill => {
let node_id = format!("tool:{}", a.target);
match &a.agent_id {
Some(agent) => yield sse(
"world.touch",
json!({
"agentId": agent,
"nodeId": node_id,
"kind": "event",
"weight": 0.45,
}),
),
None => yield sse(
"node.activity",
json!({
"nodeId": node_id,
"label": a.target,
"kind": "event",
"heat": 0.5,
}),
),
}
}
// A backlogged tool call draws nothing: the orb would be
// a tool nobody is using, permanently lit on a map of
// work that already finished.
_ => {}
}
}
}
// Real convergence: each running agent beams toward its active-run node. // Real convergence: each running agent beams toward its active-run node.
for (run_id, agent_id) in &runs { for (run_id, agent_id) in &runs {
let node_id = format!("run:{}", &run_id[..run_id.len().min(8)]); let node_id = format!("run:{}", &run_id[..run_id.len().min(8)]);
crates/cm-api/src/runtime_provision.rs
+27
View File
@@ -18,6 +18,18 @@ pub fn claw_alias(claw_id: Uuid) -> String {
format!("claw_{}", claw_id.simple()) format!("claw_{}", claw_id.simple())
} }
/// The claw behind a runtime alias, or `None` if it is not one of ours.
///
/// The inverse of [`claw_alias`], and it lives beside it so the two cannot
/// drift — a changed prefix breaks the round-trip test rather than quietly
/// returning `None` for every agent and dropping their attribution.
///
/// `None` is the honest answer for `scout` and the other configured aliases
/// that are not claws: they have no row in `agents` to point at.
pub fn claw_from_alias(alias: &str) -> Option<Uuid> {
Uuid::parse_str(alias.trim().strip_prefix("claw_")?).ok()
}
/// Map a claw's chosen model to a configured provider alias. /// Map a claw's chosen model to a configured provider alias.
/// ///
/// Claude models resolve to `claude_cli.default`, which spawns the real /// Claude models resolve to `claude_cli.default`, which spawns the real
@@ -435,4 +447,19 @@ mod tests {
let id = Uuid::nil(); let id = Uuid::nil();
assert_eq!(claw_alias(id), "claw_00000000000000000000000000000000"); assert_eq!(claw_alias(id), "claw_00000000000000000000000000000000");
} }
/// The alias must round-trip, and must NOT invent a claw for one of the
/// configured non-claw aliases.
///
/// The failure this guards is silent both ways: a broken round-trip drops
/// every tool call's agent attribution (files appear, nobody moves), and a
/// too-eager parse would attribute work to a claw id that matches no row.
#[test]
fn an_alias_round_trips_to_its_claw_and_nothing_else_does() {
let id = Uuid::from_u128(0x0198_2f11_7ac0_7d51_9c3e_44a1_09b2_5e77);
assert_eq!(claw_from_alias(&claw_alias(id)), Some(id));
assert_eq!(claw_from_alias("scout"), None);
assert_eq!(claw_from_alias("claude_cli.default"), None);
assert_eq!(claw_from_alias("claw_not-a-uuid"), None);
}
} }
crates/cm-api/src/topology_exec.rs
+227 -10
View File
@@ -54,6 +54,46 @@ pub struct ZeroClawDriveExecutor {
/// Bearer token, paired lazily and reused across turns. /// Bearer token, paired lazily and reused across turns.
token: Arc<Mutex<Option<String>>>, token: Arc<Mutex<Option<String>>>,
http: reqwest::Client, http: reqwest::Client,
/// Where this executor's turns record what they did. `None` on every path
/// that is not a mission phase (the governor, the door, the evaluator) —
/// those turns belong to no phase and have nothing to attribute to.
tap: Option<Arc<MissionTap>>,
}
/// Where a turn's tool activity is written, and what it belongs to.
///
/// Carried on the executor rather than passed per turn because `TurnRequest`
/// is the shared orchestrator contract: threading a mission id through it would
/// put mission concepts into every tier that has no missions.
pub struct MissionTap {
pub pool: sqlx::PgPool,
pub mission_id: uuid::Uuid,
pub phase_id: Option<uuid::Uuid>,
pub run_id: Option<uuid::Uuid>,
}
/// One tool call, as the frame stream reported it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ToolCall {
pub tool: String,
/// The path the tool's **arguments** named, if any. Never extracted from a
/// prose summary — see [`crate::mission_events::tool_path`].
pub path: Option<String>,
}
/// What one turn's frames said about the work, beside its text.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
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.
pub unmatched: std::collections::BTreeMap<String, u32>,
} }
impl ZeroClawDriveExecutor { impl ZeroClawDriveExecutor {
@@ -71,9 +111,17 @@ impl ZeroClawDriveExecutor {
default_alias, default_alias,
token: Arc::new(Mutex::new(None)), token: Arc::new(Mutex::new(None)),
http: reqwest::Client::new(), http: reqwest::Client::new(),
tap: None,
} }
} }
/// Attach the mission this executor's turns belong to, so their tool calls
/// are recorded. Without it the executor behaves exactly as it did.
pub fn with_tap(mut self, tap: MissionTap) -> Self {
self.tap = Some(Arc::new(tap));
self
}
/// Build from the environment: /// Build from the environment:
/// - `ZEROCLAW_GATEWAY_URL` (required) e.g. `http://127.0.0.1:42617` /// - `ZEROCLAW_GATEWAY_URL` (required) e.g. `http://127.0.0.1:42617`
/// - `ZEROCLAW_TOKEN` (preferred) a durable bearer token — pair once /// - `ZEROCLAW_TOKEN` (preferred) a durable bearer token — pair once
@@ -203,6 +251,19 @@ impl ZeroClawDriveExecutor {
} }
pub async fn drive(&self, alias: &str, prompt: &str) -> Result<TurnOutcome, OrchestratorError> { pub async fn drive(&self, alias: &str, prompt: &str) -> Result<TurnOutcome, OrchestratorError> {
self.drive_traced(alias, prompt).await.map(|(o, _)| o)
}
/// [`Self::drive`], also returning what the turn's frames said it did.
///
/// Exists so the tool tap is testable at all: `drive` discards the trace
/// after recording it, and a tap whose extraction is never asserted is
/// exactly the kind of code that silently records nothing.
pub(crate) async fn drive_traced(
&self,
alias: &str,
prompt: &str,
) -> Result<(TurnOutcome, ToolTrace), OrchestratorError> {
let token = self.ensure_paired().await?; let token = self.ensure_paired().await?;
let ws_base = if let Some(rest) = self.gateway_url.strip_prefix("https") { let ws_base = if let Some(rest) = self.gateway_url.strip_prefix("https") {
format!("wss{rest}") format!("wss{rest}")
@@ -226,7 +287,8 @@ impl ZeroClawDriveExecutor {
.await .await
.map_err(|e| OrchestratorError::Executor(format!("ws send failed: {e}")))?; .map_err(|e| OrchestratorError::Executor(format!("ws send failed: {e}")))?;
let outcome = match tokio::time::timeout(TURN_TIMEOUT, Self::drain(&mut ws)).await { let (outcome, trace) = match tokio::time::timeout(TURN_TIMEOUT, Self::drain(&mut ws)).await
{
Ok(res) => res?, Ok(res) => res?,
Err(_) => { Err(_) => {
// "turn timed out" on its own is unactionable, and the one place // "turn timed out" on its own is unactionable, and the one place
@@ -253,7 +315,60 @@ impl ZeroClawDriveExecutor {
} }
}; };
let _ = ws.close(None).await; let _ = ws.close(None).await;
Ok(outcome) self.record_trace(alias, &trace).await;
Ok((outcome, trace))
}
/// Persist what this turn's frames said the agent did.
///
/// Best-effort and after the fact: a telemetry write must not be able to
/// fail a turn that already succeeded.
async fn record_trace(&self, alias: &str, trace: &ToolTrace) {
if !trace.unmatched.is_empty() {
// Logged whether or not a tap is attached — the point is to learn
// the real frame names, and the paths with no tap see the same
// stream.
eprintln!(
"topology_exec: unmatched frame types this turn ({alias}): {:?}",
trace.unmatched
);
}
let Some(tap) = self.tap.as_ref() else { return };
if trace.calls.is_empty() {
return;
}
let agent_id = crate::runtime_provision::claw_from_alias(alias);
let event = |kind: &str, target: String, detail: serde_json::Value| {
crate::mission_events::MissionEvent {
mission_id: tap.mission_id,
phase_id: tap.phase_id,
run_id: tap.run_id,
agent_id,
kind: kind.to_string(),
target: Some(target),
detail,
}
};
let mut events = Vec::new();
for call in &trace.calls {
events.push(event(
crate::mission_events::TOOL_CALL,
call.tool.clone(),
serde_json::Value::Null,
));
// A file touch is a SECOND event, not a replacement: the tool call
// happened whether or not we could name a path in its arguments,
// and collapsing the two would make every unparseable tool call
// disappear from the record entirely.
if let Some(path) = &call.path {
events.push(event(
crate::mission_events::FILE_TOUCH,
crate::mission_events::repo_relative(path, GUEST_ROOTS),
serde_json::json!({ "tool": call.tool }),
));
}
}
crate::mission_events::record_all(&tap.pool, events).await;
} }
/// The runtime container behind this executor, derived from its gateway URL /// The runtime container behind this executor, derived from its gateway URL
@@ -330,7 +445,12 @@ impl ZeroClawDriveExecutor {
} }
/// Read frames until a terminal (`done`/`error`/`approval_request`) event. /// Read frames until a terminal (`done`/`error`/`approval_request`) event.
async fn drain<S>(ws: &mut S) -> Result<TurnOutcome, OrchestratorError> ///
/// Returns the turn's outcome AND what its frames said the agent did. The
/// trace is separate from [`TurnOutcome`] deliberately: that type is the
/// shared orchestrator contract used by every tier, and tool telemetry is a
/// mission concern.
async fn drain<S>(ws: &mut S) -> Result<(TurnOutcome, ToolTrace), OrchestratorError>
where where
S: StreamExt<Item = Result<Message, tokio_tungstenite::tungstenite::Error>> S: StreamExt<Item = Result<Message, tokio_tungstenite::tungstenite::Error>>
+ SinkExt<Message> + SinkExt<Message>
@@ -339,6 +459,7 @@ impl ZeroClawDriveExecutor {
let mut output = String::new(); let mut output = String::new();
let mut tokens: u64 = 0; let mut tokens: u64 = 0;
let mut gated: Vec<GatedAction> = Vec::new(); let mut gated: Vec<GatedAction> = Vec::new();
let mut trace = ToolTrace::default();
while let Some(frame) = ws.next().await { while let Some(frame) = ws.next().await {
let msg = frame.map_err(|e| OrchestratorError::Executor(format!("ws recv: {e}")))?; let msg = frame.map_err(|e| OrchestratorError::Executor(format!("ws recv: {e}")))?;
@@ -384,8 +505,39 @@ impl ZeroClawDriveExecutor {
"aborted" => { "aborted" => {
return Err(OrchestratorError::Executor("turn aborted".into())); return Err(OrchestratorError::Executor("turn aborted".into()));
} }
// session_start, thinking, tool_call, tool_result, … // The action channel. `arguments` is read as JSON and
_ => {} // nothing else is: the frame also carries a prose
// summary, and a path pulled out of THAT would be right
// often enough to be believed and wrong often enough to
// put files on the map that nobody edited.
"tool_call" => {
let tool = v
.get("tool")
.or_else(|| v.get("name"))
.and_then(|t| t.as_str())
.unwrap_or("")
.trim()
.to_string();
if !tool.is_empty() {
let args = v
.get("arguments")
.or_else(|| v.get("input"))
.cloned()
.unwrap_or(serde_json::Value::Null);
trace.calls.push(ToolCall {
path: crate::mission_events::tool_path(&args),
tool,
});
}
}
// session_start, thinking, tool_result, …
other => {
// Counted, not ignored. See `ToolTrace::unmatched`:
// the frame name above is unverified, and a tap
// that matches nothing looks exactly like a mission
// that used no tools.
*trace.unmatched.entry(other.to_string()).or_insert(0) += 1;
}
} }
} }
Message::Ping(p) => { Message::Ping(p) => {
@@ -396,14 +548,21 @@ impl ZeroClawDriveExecutor {
} }
} }
Ok(TurnOutcome { Ok((
output: output.trim().to_string(), TurnOutcome {
tokens, output: output.trim().to_string(),
gated, tokens,
}) gated,
},
trace,
))
} }
} }
/// Guest workspace roots, stripped so a tool's absolute path becomes the
/// repo-relative one a person recognises.
const GUEST_ROOTS: &[&str] = &["/mission/repo", "/workspace", "/repo"];
impl TurnExecutor for ZeroClawDriveExecutor { impl TurnExecutor for ZeroClawDriveExecutor {
async fn run_turn(&self, req: TurnRequest) -> Result<TurnOutcome, OrchestratorError> { async fn run_turn(&self, req: TurnRequest) -> Result<TurnOutcome, OrchestratorError> {
// An explicit per-node agent (graph `node.attrs["agent"]`) wins, so one // An explicit per-node agent (graph `node.attrs["agent"]`) wins, so one
@@ -505,6 +664,30 @@ mod tests {
}) })
} }
/// A stream carrying tool calls and one frame type we do not know.
async fn tool_ws(ws: WebSocketUpgrade) -> Response {
ws.on_upgrade(|mut socket: WebSocket| async move {
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"}}),
// A tool whose arguments name no path at all.
json!({"type": "tool_call", "tool": "Bash",
"arguments": {"command": "cargo test"}}),
// Prose that MENTIONS a path. It must not become a file touch.
json!({"type": "tool_call", "tool": "Grep",
"arguments_summary": "searching src/main.rs",
"arguments": {"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}),
] {
let _ = socket.send(AxMsg::Text(f.to_string().into())).await;
}
})
}
async fn approval_ws(ws: WebSocketUpgrade) -> Response { async fn approval_ws(ws: WebSocketUpgrade) -> Response {
ws.on_upgrade(|mut socket: WebSocket| async move { ws.on_upgrade(|mut socket: WebSocket| async move {
let _ = socket.recv().await; let _ = socket.recv().await;
@@ -569,6 +752,40 @@ mod tests {
assert!(out.gated.is_empty()); assert!(out.gated.is_empty());
} }
/// Tool detail comes from arguments, and unknown frames are counted.
///
/// The two halves are one test because they are one risk. The frame type
/// `tool_call` is taken from a comment in this file, not from a captured
/// frame — so if it is wrong, the tap records nothing, the World stays as
/// sparse as it was, and NOTHING errors. The histogram is what turns that
/// into a log line naming the real frame.
#[tokio::test]
async fn tool_frames_give_up_their_arguments_and_unknown_frames_are_counted() {
let router = Router::new()
.route("/pair", post(pair))
.route("/ws/chat", get(tool_ws));
let base = serve(router).await;
let exec = ZeroClawDriveExecutor::new(base, "code".into(), HashMap::new(), "scout".into());
let (_out, trace) = exec.drive_traced("scout", "go").await.unwrap();
assert_eq!(
trace.calls,
vec![
ToolCall { tool: "Read".into(), path: Some("/mission/repo/src/a.rs".into()) },
ToolCall { tool: "Bash".into(), path: None },
// `arguments_summary` said "src/main.rs". It is prose, so it is
// not a file touch — a path scraped from a sentence would put
// files on the map that no agent opened.
ToolCall { tool: "Grep".into(), path: None },
]
);
assert_eq!(trace.unmatched.get("a_frame_we_have_never_seen"), Some(&2));
assert_eq!(trace.unmatched.get("session_start"), Some(&1));
// `done` terminates the drain and is not an unmatched frame.
assert!(!trace.unmatched.contains_key("done"), "{:?}", trace.unmatched);
}
#[tokio::test] #[tokio::test]
async fn approval_request_is_recorded_as_blocked() { async fn approval_request_is_recorded_as_blocked() {
let router = Router::new() let router = Router::new()
crates/cm-api/src/topology_worker.rs
+28 -11
View File
@@ -185,23 +185,33 @@ async fn run_job(
// the missions row; else fall back to the shared env-derived // the missions row; else fall back to the shared env-derived
// gateway (pre-C3 missions + non-mission runs). This is what // gateway (pre-C3 missions + non-mission runs). This is what
// isolates agents' workspace filesystem to that mission's repo. // isolates agents' workspace filesystem to that mission's repo.
let mission_binding: Option<(Option<String>, Option<String>)> = type MissionBinding = (Option<String>, Option<String>, Uuid, Option<Uuid>);
sqlx::query_as::<_, (Option<String>, Option<String>)>( let mission_binding: Option<MissionBinding> = sqlx::query_as::<_, MissionBinding>(
"SELECT m.runtime_endpoint, m.runtime_pairing_code "SELECT m.runtime_endpoint, m.runtime_pairing_code, m.id, r.mission_phase_id
FROM topology_runs r FROM topology_runs r
JOIN missions m ON m.id = r.mission_id JOIN missions m ON m.id = r.mission_id
WHERE r.id = $1", WHERE r.id = $1",
) )
.bind(id) .bind(id)
.fetch_optional(pool) .fetch_optional(pool)
.await .await
.ok() .ok()
.flatten(); .flatten();
// What this run's turns will be attributed to. `None` when the run belongs
// to no mission — a bare topology run has no phase to hang tool calls on.
let tap = mission_binding
.as_ref()
.map(|(_, _, mission_id, phase_id)| crate::topology_exec::MissionTap {
pool: pool.clone(),
mission_id: *mission_id,
phase_id: *phase_id,
run_id: Some(id),
});
let leaf_result = match mission_binding { let leaf_result = match mission_binding {
Some((Some(url), Some(code))) => { Some((Some(url), Some(code), _, _)) => {
ZeroClawDriveExecutor::from_env_for_gateway_with_code(url, code) ZeroClawDriveExecutor::from_env_for_gateway_with_code(url, code)
} }
Some((Some(url), None)) => ZeroClawDriveExecutor::from_env_for_gateway(url), Some((Some(url), None, _, _)) => ZeroClawDriveExecutor::from_env_for_gateway(url),
_ => ZeroClawDriveExecutor::from_env(), _ => ZeroClawDriveExecutor::from_env(),
}; };
let leaf = match leaf_result { let leaf = match leaf_result {
@@ -211,6 +221,13 @@ async fn run_job(
return; return;
} }
}; };
// The tap rides on the leaf executor, so the recursive tiers get it too:
// they drive the same leaf all the way down, and a company-tier mission's
// tool calls belong to its phase exactly as a team-tier one's do.
let leaf = match tap {
Some(t) => leaf.with_tap(t),
None => leaf,
};
// Select the executor by deploy tier: `team` drives claws directly; the // Select the executor by deploy tier: `team` drives claws directly; the
// upper tiers drive the recursive sub-topology executor (which runs each // upper tiers drive the recursive sub-topology executor (which runs each
crates/cm-api/src/vm_stop_gate.rs
+10 -19
View File
@@ -36,7 +36,6 @@
//! `Stop`, `SubagentStop`, `PreToolUse`, `PostToolUse`, `UserPromptSubmit` and //! `Stop`, `SubagentStop`, `PreToolUse`, `PostToolUse`, `UserPromptSubmit` and
//! `SessionStart` do. //! `SessionStart` do.
use serde_json::json;
/// How many times the gate may refuse a stop before it gives up and lets the /// How many times the gate may refuse a stop before it gives up and lets the
/// agent finish. Three is enough for "you wrote nothing" → "you wrote something" /// agent finish. Three is enough for "you wrote nothing" → "you wrote something"
@@ -183,21 +182,12 @@ impl StopGate {
s s
} }
/// The settings file that installs the script as a `Stop` hook. /// One shell command that writes the gate SCRIPT into the guest.
pub fn settings(&self, dir: &str) -> serde_json::Value { ///
json!({ /// It deliberately does NOT write `settings.json`. It used to, and it wrote
"hooks": { /// the whole document — so the moment a second feature needed a hook, the
"Stop": [{ /// later writer would silently erase this one. The composed document is
"hooks": [{ /// built in exactly one place: [`crate::vm_tool_tap::guest_settings`].
"type": "command",
"command": format!("{dir}/stop-gate.sh"),
}]
}]
}
})
}
/// One shell command that writes the gate into the guest.
/// ///
/// Written by `printf` through an exec rather than injected as part of the /// Written by `printf` through an exec rather than injected as part of the
/// tar: the tar lands in `/mission/repo`, which is exactly where this must /// tar: the tar lands in `/mission/repo`, which is exactly where this must
@@ -206,10 +196,9 @@ impl StopGate {
format!( format!(
"mkdir -p {d} && rm -f {d}/blocks {d}/log {d}/capped \ "mkdir -p {d} && rm -f {d}/blocks {d}/log {d}/capped \
&& printf '%s' {script} > {d}/stop-gate.sh \ && printf '%s' {script} > {d}/stop-gate.sh \
&& chmod +x {d}/stop-gate.sh && printf '%s' {settings} > {d}/settings.json", && chmod +x {d}/stop-gate.sh",
d = dir, d = dir,
script = q(&self.script(repo, dir)), script = q(&self.script(repo, dir)),
settings = q(&self.settings(dir).to_string()),
) )
} }
} }
@@ -223,6 +212,7 @@ fn q(s: &str) -> String {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use serde_json::json;
use std::path::Path; use std::path::Path;
use std::process::Command; use std::process::Command;
@@ -474,7 +464,8 @@ mod tests {
assert!(!install.contains(write), "{install}"); assert!(!install.contains(write), "{install}");
} }
assert_eq!( assert_eq!(
gate.settings(GATE_DIR)["hooks"]["Stop"][0]["hooks"][0]["command"], crate::vm_tool_tap::guest_settings(Some(GATE_DIR), None)["hooks"]["Stop"][0]["hooks"]
[0]["command"],
json!("/root/gate/stop-gate.sh") json!("/root/gate/stop-gate.sh")
); );
} }
crates/cm-api/src/vm_tool_tap.rs
+274
View File
@@ -0,0 +1,274 @@
//! What the agent did inside a microVM, taken from Claude Code's own hooks.
//!
//! The microVM tier had no action channel at all: a phase ran, a diff came
//! back, and everything between was invisible. The seam is the same one
//! [`crate::vm_stop_gate`] proved works in this image — `PostToolUse` fires
//! under `claude -p`, measured, not read off documentation.
//!
//! # The observer must not become a participant
//!
//! The hook `exit 0`s unconditionally. A `PostToolUse` hook that exits non-zero
//! feeds its stderr back to the model, so a tap with a bug would start
//! *instructing* the agent it exists to watch — and the resulting transcript
//! would look like a model that lost the plot rather than a broken hook.
//!
//! # Never inside the repository
//!
//! Everything lives under `/root`. `/mission/repo` is collected and diffed, so
//! a tap file written there would arrive in the user's delivered patch as
//! though an agent had authored it — the same rule, and the same reason, as the
//! stop gate's [`crate::vm_stop_gate::GATE_DIR`].
use serde_json::{json, Value};
/// Where the tap writes in the guest. Under `/root`, never the repo.
pub const TAP_DIR: &str = "/root/tap";
/// The file the hook appends to, one JSON object per line.
pub const TAP_FILE: &str = "/root/tap/tools.jsonl";
/// The single settings document the guest agent runs with.
///
/// One path, because there is only ever one writer — see [`guest_settings`].
pub const SETTINGS_PATH: &str = "/root/guest-settings.json";
/// Read the tap out of the guest, before collection destroys the VM.
///
/// `|| true` so a phase whose agent called no tools — or where the hook never
/// fired — reads as empty rather than as a failed probe. The difference between
/// 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";
/// One observed tool call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Observed {
pub tool: String,
/// The path the tool's **input** named, if any. From JSON, never prose.
pub path: Option<String>,
}
/// The hook script. Copies stdin verbatim to the tap file and gets out of the
/// way.
///
/// The parsing happens host-side, on purpose: a `jq` or `sed` pipeline in the
/// guest would need the tool's JSON schema baked into a shell script, inside an
/// image we do not rebuild for a parser change, with no way to tell a parse
/// failure from a quiet turn.
pub fn hook_script(dir: &str) -> String {
format!(
"#!/bin/sh\n\
# The tool tap. See cm-api/src/vm_tool_tap.rs.\n\
mkdir -p {dir} 2>/dev/null\n\
# `cat` of stdin, appended whole. One JSON object per line, because\n\
# Claude Code hands the hook one event per invocation.\n\
cat >> {dir}/tools.jsonl 2>/dev/null\n\
printf '\\n' >> {dir}/tools.jsonl 2>/dev/null\n\
# ALWAYS zero. A non-zero PostToolUse hook talks back to the model.\n\
exit 0\n"
)
}
/// The settings document for the guest, carrying **every** hook at once.
///
/// This function exists because the alternative — each feature writing its own
/// `settings.json` — is a silent clobber. The stop gate wrote the whole
/// document; a tap that did the same would erase the gate, and a coding phase
/// would then complete having written nothing, which is the exact failure the
/// gate exists to catch. One writer, one document, one test that both hooks
/// survive it.
///
/// `None` for either half means that hook is simply absent.
pub fn guest_settings(gate_dir: Option<&str>, tap_dir: Option<&str>) -> Value {
let mut hooks = serde_json::Map::new();
if let Some(dir) = gate_dir {
hooks.insert(
"Stop".into(),
json!([{ "hooks": [{ "type": "command", "command": format!("{dir}/stop-gate.sh") }] }]),
);
}
if let Some(dir) = tap_dir {
hooks.insert(
"PostToolUse".into(),
json!([{ "hooks": [{ "type": "command", "command": format!("{dir}/tap.sh") }] }]),
);
}
json!({ "hooks": Value::Object(hooks) })
}
/// One shell command that installs the tap.
///
/// Written by `printf` through an exec rather than injected with the workspace
/// tar: the tar lands in `/mission/repo`, which is exactly where this must not.
pub fn install_command(dir: &str) -> String {
format!(
"mkdir -p {dir} && rm -f {dir}/tools.jsonl \
&& printf '%s' {script} > {dir}/tap.sh && chmod +x {dir}/tap.sh",
dir = dir,
script = q(&hook_script(dir)),
)
}
/// Write the composed settings document.
pub fn settings_command(path: &str, settings: &Value) -> String {
format!("printf '%s' {} > {path}", q(&settings.to_string()))
}
/// Parse a drained tap.
///
/// Tolerant by construction: the file is appended to by a shell hook in a VM
/// that may be killed mid-write, so a truncated last line is expected and is
/// skipped rather than failing the whole drain. Losing the last tool call of a
/// phase costs one orb; losing all of them because of it would cost the tier.
pub fn parse(raw: &str) -> Vec<Observed> {
raw.lines()
.map(str::trim)
.filter(|l| !l.is_empty())
.filter_map(|line| {
let v: Value = serde_json::from_str(line).ok()?;
// Only tool events. The same hook file would carry others if the
// settings ever install one, and a `hook_event_name` we do not
// recognise must not be read as a tool named "".
let tool = v
.get("tool_name")
.or_else(|| v.get("toolName"))
.and_then(Value::as_str)?
.trim()
.to_string();
if tool.is_empty() {
return None;
}
let input = v
.get("tool_input")
.or_else(|| v.get("toolInput"))
.cloned()
.unwrap_or(Value::Null);
Some(Observed {
path: crate::mission_events::tool_path(&input),
tool,
})
})
.collect()
}
/// Single-quote for `sh`. Local copy, same rule as the stop gate's — these two
/// modules deliberately share no code, so neither can break the other.
fn q(s: &str) -> String {
format!("'{}'", s.replace('\'', r"'\''"))
}
#[cfg(test)]
mod tests {
use super::*;
/// The gate and the tap must BOTH survive one settings document.
///
/// This is the whole reason `guest_settings` exists. Two writers each
/// producing a whole `settings.json` is not a merge conflict — the second
/// simply wins, no error, and the loser's hook never runs. When the loser is
/// the stop gate, a coding phase completes having written nothing: the exact
/// failure the gate was built to catch.
#[test]
fn both_hooks_survive_one_settings_document() {
let s = guest_settings(Some("/root/gate"), Some(TAP_DIR));
let hooks = s.get("hooks").expect("hooks");
assert_eq!(
hooks["Stop"][0]["hooks"][0]["command"],
json!("/root/gate/stop-gate.sh"),
"the stop gate must survive the tap being installed"
);
assert_eq!(
hooks["PostToolUse"][0]["hooks"][0]["command"],
json!("/root/tap/tap.sh")
);
}
/// Either half absent leaves the other exactly as it was.
#[test]
fn one_hook_alone_is_a_valid_document() {
let gate_only = guest_settings(Some("/root/gate"), None);
assert!(gate_only["hooks"].get("Stop").is_some());
assert!(gate_only["hooks"].get("PostToolUse").is_none());
let tap_only = guest_settings(None, Some(TAP_DIR));
assert!(tap_only["hooks"].get("Stop").is_none());
assert!(tap_only["hooks"].get("PostToolUse").is_some());
}
/// The observer must never talk back to the model.
#[test]
fn the_hook_always_exits_zero() {
let s = hook_script(TAP_DIR);
assert!(s.contains("exit 0"));
// No conditional exits at all: a `PostToolUse` hook that exits non-zero
// feeds stderr back to the agent, so the tap would become an instruction.
assert!(!s.contains("exit 1") && !s.contains("exit 2"), "{s}");
}
/// Nothing the tap writes may land in the delivered tree.
#[test]
fn the_tap_never_writes_into_the_repository() {
assert!(TAP_DIR.starts_with("/root/"));
assert!(TAP_FILE.starts_with("/root/"));
assert!(SETTINGS_PATH.starts_with("/root/"));
let cmd = install_command(TAP_DIR);
assert!(!cmd.contains("/mission/repo"), "{cmd}");
assert!(!hook_script(TAP_DIR).contains("/mission/repo"));
}
/// A real `PostToolUse` payload gives up its tool and its path — and a
/// truncated final line does not take the rest of the phase with it.
#[test]
fn a_drained_tap_parses_and_tolerates_a_torn_last_line() {
let raw = concat!(
r#"{"hook_event_name":"PostToolUse","tool_name":"Edit","#,
r#""tool_input":{"file_path":"/mission/repo/src/a.rs"}}"#,
"\n",
r#"{"hook_event_name":"PostToolUse","tool_name":"Bash","tool_input":{"command":"ls"}}"#,
"\n",
"\n",
// The VM was destroyed mid-write.
r#"{"hook_event_name":"PostToolUse","tool_name":"Wri"#,
);
assert_eq!(
parse(raw),
vec![
Observed { tool: "Edit".into(), path: Some("/mission/repo/src/a.rs".into()) },
Observed { tool: "Bash".into(), path: None },
]
);
}
/// Exactly one place in the tree writes the guest settings document.
///
/// The unit test above proves `guest_settings` composes correctly; it says
/// nothing about whether anyone bypasses it. A second `> …settings.json`
/// anywhere is the silent clobber itself, and it would pass every other
/// test in this file.
#[test]
fn nothing_else_writes_the_guest_settings() {
for (name, src) in [
("vm_stop_gate.rs", include_str!("vm_stop_gate.rs")),
("microvm_executor.rs", include_str!("microvm_executor.rs")),
] {
assert!(
!src.contains("> {d}/settings.json") && !src.contains("settings.json\","),
"{name} writes a settings document of its own; compose it through \
vm_tool_tap::guest_settings instead"
);
}
// And the one legitimate writer is this module's own helper.
let exec = include_str!("microvm_executor.rs");
assert_eq!(
exec.matches("vm_tool_tap::settings_command").count(),
1,
"the settings document must be written exactly once per turn"
);
}
/// An event that is not a tool call is not a tool named "".
#[test]
fn a_non_tool_event_is_skipped() {
assert!(parse(r#"{"hook_event_name":"SessionStart","session_id":"x"}"#).is_empty());
assert!(parse(r#"{"tool_name":" ","tool_input":{}}"#).is_empty());
}
}
migrations/0075_mission_events.sql
+43
View File
@@ -0,0 +1,43 @@
-- Structured mission activity: what an agent actually DID, as data.
--
-- Until now the World could show a mission's shape (phases, status, who is on
-- which station) but nothing about the work itself. The detail existed only as
-- prose in `checkpoint.log` and agent output, where a tool name is
-- indistinguishable from an agent *talking about* a tool — so it was never
-- parsed, deliberately. This table is the structured channel that replaces it.
--
-- `run_events` could not be reused: 0003 declares `run_id REFERENCES
-- agent_runs(id)`, and mission phases insert into `topology_runs`. Those are
-- independent id spaces, so every write would have been an FK violation.
CREATE TABLE IF NOT EXISTS mission_events (
id BIGSERIAL PRIMARY KEY,
mission_id UUID NOT NULL REFERENCES missions(id) ON DELETE CASCADE,
-- No FK. `mission_phases` rows survive, but a phase that loops is re-run
-- and this column is only ever read as a grouping key.
phase_id UUID,
-- Deliberately NO FK, and this is the load-bearing decision in the file:
-- `phase_runner` DELETEs the `topology_runs` row on every retry. With a
-- cascading FK, a phase's whole history would vanish the moment it retried
-- — silently, since a cascade is not an error. The id is kept as a plain
-- correlation value.
run_id UUID,
-- The platform agent, when there is one. NULL for a microVM phase, which
-- has no platform agent at all — that is the truth, not missing data.
agent_id UUID,
-- phase.started | phase.completed | tool.call | file.touch
-- | finding.raised | benchmark.delta
kind TEXT NOT NULL,
-- The tool name, or the file path — whatever the event is *about*.
target TEXT,
detail JSONB NOT NULL DEFAULT '{}'::jsonb,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- The World reads by mission, newest last, and by phase for the per-phase cap.
CREATE INDEX IF NOT EXISTS mission_events_mission_idx
ON mission_events (mission_id, id);
CREATE INDEX IF NOT EXISTS mission_events_phase_idx
ON mission_events (phase_id, id);
-- The retention sweep in `mission_gc` scans by age.
CREATE INDEX IF NOT EXISTS mission_events_created_idx
ON mission_events (created_at);