//! A [`cm_orchestrator::TurnExecutor`] that runs each topology turn as a real //! ZeroClaw role-agent inside a container, driven over the gateway WebSocket //! "drive" recipe. //! //! The orchestrator owns the topology graph (it sequences edges, meters, and //! records the journal); this executor only runs *one* turn: it pairs with the //! gateway, opens `/ws/chat?agent=`, sends the role+task+context prompt, //! and streams the turn's events back into a [`TurnOutcome`]. //! //! **These agents are NOT tool-free.** That claim stood here for months and is //! false — see `docs/TOOL-CALL-ARCHITECTURE.md`. It was inferred from a frame //! stream that carried no tool events, and the emptiness has a different cause: //! `claude_cli` runs `claude -p --output-format json`, which returns a single //! final result object, and the provider hardcodes `tool_calls: Vec::new()`. //! The agent calls Claude Code's own tools; the transport discards them. //! `--output-format stream-json` emits `tool_use`/`tool_result` blocks — //! verified against the deployed Claude Code 2.1.228. //! //! The door-shaped provider that WOULD make this true (`--mcp-config` + //! `--disallowedTools` on the natives) is built and documented in //! `agent.config.example.toml`, and is not deployed: mission claws bind to //! `claude_cli.default`, which sets none of it. //! //! If the gateway emits an `approval_request` we still record it as a //! **blocked** `GatedAction` and end the turn — we never auto-approve. use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; use cm_domain::GatedCategory; use cm_orchestrator::{GatedAction, OrchestratorError, TurnExecutor, TurnOutcome, TurnRequest}; use futures::{SinkExt, StreamExt}; use tokio::sync::Mutex; use tokio_tungstenite::connect_async; use tokio_tungstenite::tungstenite::Message; /// Overall wall-clock budget for draining one turn's event stream. /// /// A turn is an agent LOOP, not one model call. Each call inside it is bounded /// separately by the daemon — `claude_cli`'s `timeout_secs`, 600s on gw-04 — /// so this has to cover however many calls the loop makes, not one of them. /// /// It was 700s, which is 100s more than a single call may take. MEASURED: a /// healthy research turn is ~157s, but a throttled one blew the budget with one /// slow call plus a second, and the executor killed it mid-flight after 11m43s /// with no error from the daemon — because nothing had failed yet. All the /// operator got was "turn timed out". /// /// An hour matches the phase's own budget. A genuinely stuck CALL is still /// caught at 600s by the daemon and surfaces as a real error; this only stops /// us killing turns that are working, slowly. const TURN_TIMEOUT: Duration = Duration::from_secs(3600); /// Drives ZeroClaw role-agents (in one container) to execute topology turns. /// Cap on the pinned-skill text injected into one mission turn. /// /// Skill bodies average ~3.5 KB and pinning is `idx < 2 || foundation`, so a /// role lands near 7-10 KB. The cap exists for the role that grows a long /// foundation set, and it is stated in the prompt when it fires. pub(crate) const MAX_PINNED_SKILL_BYTES: usize = 24_000; /// The line that introduces each skill in a prompt. /// /// NOT a markdown heading. The first version used `## `, and skill bodies /// are markdown that contain their own `##` headings — so anything reading the /// prompt back counted every section of every body as a separate skill. A live /// mission scored "Sizing heuristic" and "The output shape" as skills, which is /// what surfaced it. /// /// This marker cannot occur inside a body, so the prompt stays parseable by /// whatever reads it later. Skills are written by one function /// ([`render_pinned_skill`]) for the same reason: two renderers would drift and /// the reader would silently match only one. pub const SKILL_MARKER: &str = "--- SKILL: "; /// One skill, rendered for a prompt. pub fn render_pinned_skill(name: &str, body: &str) -> String { format!("\n{SKILL_MARKER}{name} ---\n{body}\n") } /// The skill names a rendered prompt delivered. pub fn skill_names_in(prompt: &str) -> Vec { prompt .lines() .filter_map(|l| l.trim().strip_prefix(SKILL_MARKER)) .map(|rest| rest.trim_end_matches(" ---").trim().to_string()) .filter(|n| !n.is_empty()) .collect() } pub struct ZeroClawDriveExecutor { /// Gateway base URL, e.g. `http://127.0.0.1:42617`. gateway_url: String, /// One-time pairing code (minted into a bearer token on first use). pairing_code: String, /// Topology `node.role` → ZeroClaw agent alias. role_aliases: HashMap, /// Alias used when a role isn't mapped. default_alias: String, /// Bearer token, paired lazily and reused across turns. token: Arc>>, 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>, } /// 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, /// Which workspace's live feed these frames belong to. Every subscriber is /// workspace-scoped, so a frame without this could not be routed. pub workspace_id: uuid::Uuid, pub mission_id: uuid::Uuid, pub phase_id: Option, pub run_id: Option, } /// 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, } /// What one turn's frames said about the work, beside its text. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct ToolTrace { pub calls: Vec, /// Frame `type` values this drain did not recognise, counted. /// /// 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"}`) — and it is NOT that the agents are tool-free, which is what /// this comment used to say. `claude_cli` asks for `--output-format json`, /// so the subprocess's tool calls never reach the gateway to be framed. /// The histogram still does its job: it distinguishes "no frames" from /// "frames we do not recognise", and the answer was the former. pub unmatched: std::collections::BTreeMap, } impl ZeroClawDriveExecutor { /// Build an executor explicitly (used in tests). pub fn new( gateway_url: String, pairing_code: String, role_aliases: HashMap, default_alias: String, ) -> Self { ZeroClawDriveExecutor { gateway_url: gateway_url.trim_end_matches('/').to_string(), pairing_code, role_aliases, default_alias, token: Arc::new(Mutex::new(None)), 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: /// - `ZEROCLAW_GATEWAY_URL` (required) e.g. `http://127.0.0.1:42617` /// - `ZEROCLAW_TOKEN` (preferred) a durable bearer token — pair once /// out-of-band, set this, and runs are repeatable. If set, no pairing. /// - `ZEROCLAW_PAIRING_CODE` (fallback) a one-time pairing code — consumed /// on first use, so only good for a single run. One of TOKEN/CODE required. /// - `ZEROCLAW_AGENT_MAP` (optional) `role=alias,role=alias` /// - `ZEROCLAW_DEFAULT_AGENT` (optional, default `scout`) pub fn from_env() -> Result { let gateway_url = std::env::var("ZEROCLAW_GATEWAY_URL").map_err(|_| "ZEROCLAW_GATEWAY_URL not set")?; Self::from_env_for_gateway(gateway_url) } /// Same as [`from_env`] but with a caller-supplied gateway URL. Used by /// the research pipeline to point the executor at the per-topic team /// container spawned in `research_container::spawn` instead of the /// workspace-wide gateway from `ZEROCLAW_GATEWAY_URL`. The auth token, /// role map, and default alias still come from the parent server's /// env — they're propagated into the team container by /// `research_container::inherited_env` so both endpoints use the same /// credentials. pub fn from_env_for_gateway(gateway_url: String) -> Result { let token = std::env::var("ZEROCLAW_TOKEN") .ok() .filter(|t| !t.is_empty()); let pairing_code = std::env::var("ZEROCLAW_PAIRING_CODE").unwrap_or_default(); if token.is_none() && pairing_code.is_empty() { return Err("set ZEROCLAW_TOKEN or ZEROCLAW_PAIRING_CODE".to_string()); } let default_alias = std::env::var("ZEROCLAW_DEFAULT_AGENT").unwrap_or_else(|_| "scout".to_string()); let role_aliases = std::env::var("ZEROCLAW_AGENT_MAP") .ok() .map(|s| parse_agent_map(&s)) .unwrap_or_default(); let mut exec = Self::new(gateway_url, pairing_code, role_aliases, default_alias); exec.token = Arc::new(Mutex::new(token)); Ok(exec) } /// Like [`from_env_for_gateway`] but with a caller-supplied pairing /// code — used by per-mission runtimes whose fresh daemons mint a /// new one-time code at startup. The env-derived ZEROCLAW_TOKEN /// is ignored (belongs to the shared runtime) so the lazy pair /// path runs and issues a bearer for this specific gateway. /// Reuse a token that was already paired and persisted. /// /// The pairing code is single-use, so a restarted server cannot pair again: /// it gets 403 and the mission is unrecoverable. Seeding the cache from /// `missions.runtime_token` is what makes a mission survive a restart. pub fn with_token(self, token: Option) -> Self { if let Some(t) = token.filter(|t| !t.trim().is_empty()) { // try_lock: this runs at construction, before any turn holds it. if let Ok(mut g) = self.token.try_lock() { *g = Some(t); } } self } pub fn from_env_for_gateway_with_code( gateway_url: String, pairing_code: String, ) -> Result { if pairing_code.is_empty() { return Err("empty pairing_code".to_string()); } let default_alias = std::env::var("ZEROCLAW_DEFAULT_AGENT").unwrap_or_else(|_| "scout".to_string()); let role_aliases = std::env::var("ZEROCLAW_AGENT_MAP") .ok() .map(|s| parse_agent_map(&s)) .unwrap_or_default(); Ok(Self::new( gateway_url, pairing_code, role_aliases, default_alias, )) } fn alias_for(&self, role: &str) -> String { self.role_aliases .get(role) .cloned() .unwrap_or_else(|| self.default_alias.clone()) } /// `POST {base}/pair` with the pairing code header → bearer token (cached). async fn ensure_paired(&self) -> Result { let mut guard = self.token.lock().await; if let Some(tok) = guard.as_ref() { return Ok(tok.clone()); } let res = self .http .post(format!("{}/pair", self.gateway_url)) .header("X-Pairing-Code", &self.pairing_code) .header("Content-Type", "application/json") .body("{}") .send() .await .map_err(|e| OrchestratorError::Executor(format!("pair request failed: {e}")))?; if !res.status().is_success() { return Err(OrchestratorError::Executor(format!( "pair failed: {}", res.status() ))); } let body: serde_json::Value = res .json() .await .map_err(|e| OrchestratorError::Executor(format!("pair response not json: {e}")))?; let token = body .get("token") .or_else(|| body.get("bearer")) .or_else(|| body.get("access_token")) .and_then(|v| v.as_str()) .ok_or_else(|| OrchestratorError::Executor("pair response had no token".into()))? .to_string(); *guard = Some(token.clone()); // Persist it. The code we just spent cannot be used again, so if this // token only ever lives in memory the next server process has no way // back in — that is the 403 that killed a 93k-token research phase. // Best-effort: failing to save must not fail a turn that just paired // successfully; the cost is that a restart before the next write // re-opens the original hole. if let Some(tap) = self.tap.as_ref() { if let Err(e) = sqlx::query("UPDATE missions SET runtime_token = $1 WHERE id = $2") .bind(&token) .bind(tap.mission_id) .execute(&tap.pool) .await { eprintln!( "topology_exec: could not persist runtime token for mission {}: {e}", tap.mission_id ); } } Ok(token) } /// The pinned skills for the claw behind `alias`, rendered for the prompt. /// /// Missions had NO path to a skill. The catalogue's only delivery channel /// is the `clawmates_skills` MCP server, and a mission agent cannot reach /// it for three independent reasons: `provision_claw` wrote a constant /// bundle list, the runtime config defines no such bundle, and mission /// claws run on `claude_cli`, which is text-only and cannot surface a tool /// call at all. Two doc comments in `cm-runtime` describe the mission path /// as already having this contract. It never did — so every skill authored /// for a mission role was unreachable prose, and no measurement of whether /// skills fire could have returned anything but zero. /// /// Bodies or an index, depending on the mission's arm — see /// [`crate::skill_delivery`]. Bodies were once the only honest option: /// there was no tool on the mission path that could fetch one, so an index /// would have advertised a capability that did not exist. The skills door /// changed that, and the arm is now recorded per mission so both can run. /// /// Pinned only (`pin_in_context`) in either arm, because everything else /// would go in unbounded and unread. pub async fn pinned_skills_text(&self, alias: &str) -> Option { let mode = self.skill_delivery_mode().await; self.pinned_skills_in_mode(alias, mode).await } /// The arm this mission was launched with. /// /// Read per turn rather than cached on the executor: the executor is /// constructed from the environment by `topology_worker`, which knows /// nothing about a mission, and the arm is decided at launch by the code /// that also learns whether the door installed. /// /// Anything unreadable — no tap, no row, an unrecognised value — resolves /// to `Inline`, which is the arm that needs nothing to be true. pub(crate) async fn skill_delivery_mode(&self) -> crate::skill_delivery::Mode { let Some(tap) = self.tap.as_ref() else { return crate::skill_delivery::Mode::Inline; }; sqlx::query_scalar::<_, Option>( "SELECT skill_delivery FROM missions WHERE id = $1", ) .bind(tap.mission_id) .fetch_optional(&tap.pool) .await .ok() .flatten() .flatten() .and_then(|s| crate::skill_delivery::parse(&s)) .unwrap_or(crate::skill_delivery::Mode::Inline) } pub(crate) async fn pinned_skills_in_mode( &self, alias: &str, mode: crate::skill_delivery::Mode, ) -> Option { let tap = self.tap.as_ref()?; let agent_id = crate::runtime_provision::claw_from_alias(alias)?; let link = cm_db::repo::agent_template_link::get(&tap.pool, agent_id) .await .ok() .flatten(); let (tpl_id, slot) = link .as_ref() .map(|l| (Some(l.template_id), Some(l.role_slot.as_str()))) .unwrap_or((None, None)); let bindings = cm_db::repo::skills_catalog::effective_for_agent(&tap.pool, agent_id, tpl_id, slot) .await .ok()?; let mut out = String::new(); let mut n = 0usize; for b in bindings.iter().filter(|b| b.pin_in_context) { // `always_inject` overrides the arm. Progressive disclosure asks // the agent to recognise that a procedure applies before fetching // it, and a CROSS-CUTTING procedure is the case that breaks: the // first A/B pair had `workspace-repo-commit-protocol` scored // Trigger=FAIL beside a passing boundary check, because a rule that // applies to everyone who writes reads as nobody's in particular. let text = match mode { crate::skill_delivery::Mode::Inline => b.skill.body.clone(), crate::skill_delivery::Mode::Index if b.skill.always_inject => { b.skill.body.clone() } // An entry is a few hundred bytes whatever the body weighs, so // the index arm cannot hit the cap that follows. That is the // point of it, and the reason the cap is checked against the // rendered text rather than against the body. crate::skill_delivery::Mode::Index => crate::skill_delivery::index_entry( &b.skill.description, b.skill.when_to_use.as_deref(), &crate::mcp_skills::skill_uri(b.skill.workspace_id, &b.skill.name), ), }; // Bounded, and truncation is STATED. A silently clipped procedure // is worse than an absent one: the agent follows the half it can // see and reports success against a rule it never read. if out.len() + text.len() > MAX_PINNED_SKILL_BYTES { out.push_str(&format!( "\n[skill \"{}\" omitted — the pinned set exceeded {} bytes]\n", b.skill.name, MAX_PINNED_SKILL_BYTES )); continue; } out.push_str(&render_pinned_skill(&b.skill.name, &text)); n += 1; } if n == 0 { return None; } Some(out) } /// Mirror of `ProviderExecutor`'s prompt, flattened to one `content` string /// (the gateway `message` envelope carries a single content field). fn build_prompt(req: &TurnRequest) -> String { let system = format!( "You are the \"{}\" agent in a multi-agent system. Do your part of the task \ concisely and return only your result.", req.role ); let mut user = format!("Task: {}", req.task); if !req.context.is_empty() { user.push_str("\n\nContext from upstream agents:\n"); for (i, c) in req.context.iter().enumerate() { user.push_str(&format!("[{i}] {c}\n")); } } format!("{system}\n\n{user}") } pub async fn drive(&self, alias: &str, prompt: &str) -> Result { 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 ws_base = if let Some(rest) = self.gateway_url.strip_prefix("https") { format!("wss{rest}") } else if let Some(rest) = self.gateway_url.strip_prefix("http") { format!("ws{rest}") } else { self.gateway_url.clone() }; let ws_url = format!( "{ws_base}/ws/chat?agent={}&name=clawmates&token={}", urlencoding::encode(alias), urlencoding::encode(&token), ); let (mut ws, _resp) = connect_async(&ws_url) .await .map_err(|e| OrchestratorError::Executor(format!("ws connect failed: {e}")))?; let envelope = serde_json::json!({ "type": "message", "content": prompt }).to_string(); ws.send(Message::Text(envelope.into())) .await .map_err(|e| OrchestratorError::Executor(format!("ws send failed: {e}")))?; let (outcome, trace) = match tokio::time::timeout( TURN_TIMEOUT, Self::drain( &mut ws, self.tap.as_ref().and_then(|t| { crate::live_bus::agent_id_from_alias(alias).map(|a| (t.workspace_id, a)) }), ), ) .await { Ok(res) => res?, Err(_) => { // "turn timed out" on its own is unactionable, and the one place // the reason lives — the per-mission runtime container — is torn // down after the phase, taking its log with it. Read the tail // while it still exists. // // MEASURED: a research phase timed out at exactly 700s having // produced zero steps and zero output, and the container was // already gone by the time anyone looked. All that survived was // the string. let container = self.container_name(); let tail = match &container { Some(c) => crate::container_exec::tail_logs(c, 40).await, None => "(could not derive the container name from the gateway url)".into(), }; return Err(OrchestratorError::Executor(format!( "turn timed out after {}s driving agent {alias} on {} — the agent \ never finished a turn. Last lines from {}:\n{tail}", TURN_TIMEOUT.as_secs(), self.gateway_url, container.as_deref().unwrap_or("its runtime container"), ))); } }; let _ = ws.close(None).await; 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 /// (`http://cm-runtime-mission-:42617`). Used only to fetch a log tail /// for an error message, so an unparseable URL is `None` rather than a /// failure. fn container_name(&self) -> Option { let rest = self .gateway_url .split("://") .nth(1) .unwrap_or(&self.gateway_url); let host = rest.split('/').next()?.split(':').next()?; (!host.is_empty()).then(|| host.to_string()) } /// Use a runtime agent as a governance judge: drive `alias` with the judge /// prompt and parse the verdict (`DENY` anywhere ⇒ deny, else allow). This /// lets a **subscription-only** model (e.g. Kimi via `kimi_cli`) be the judge /// with no platform API key — the registry/SDK path GLM and Kimi can't take. /// Fail-open (returns `(true, …)`) so a judge outage never halts agents. pub async fn judge(&self, alias: &str, system: &str, user: &str) -> (bool, String) { let prompt = format!("{system}\n\n{user}"); match self.drive(alias, &prompt).await { Ok(outcome) => { let text = outcome.output.trim().to_string(); let allow = !text.to_uppercase().contains("DENY"); (allow, text) } Err(e) => (true, format!("governor unreachable (fail-open): {e}")), } } /// Drive `alias` with a judging prompt and return its **raw** reply. /// /// [`Self::judge`] collapses the reply to a bool by substring-matching /// `DENY`, which only suits the governor's ALLOW/DENY contract and is /// fail-open. Callers that need a structured verdict — the phase /// completion evaluator wants `{"met":bool,"reason":string}` and must fail /// **closed** — need the text, and need the error rather than a /// synthesized permissive answer. pub async fn judge_raw(&self, alias: &str, system: &str, user: &str) -> Result { let prompt = format!("{system}\n\n{user}"); self.drive(alias, &prompt) .await .map(|outcome| outcome.output.trim().to_string()) .map_err(|e| e.to_string()) } /// Drive agent `alias` as a delegated sub-task and return its result. Reuses /// the same gateway drive as topology turns + the governor, so a delegated /// turn carries the same blocked-action / token instrumentation in its /// [`TurnOutcome`]. The target is tool-free behind the MCP door, so a /// delegated turn adds no new egress (§15 holds — the bridge only causes an /// in-workspace agent to run a turn; anything it does is independently gated /// at the door). pub async fn delegate( &self, alias: &str, task: &str, context: &[String], ) -> Result { let mut prompt = format!( "You are being delegated a sub-task by another agent on your team. \ Do it concisely and return only your result.\n\nTask: {task}" ); if !context.is_empty() { prompt.push_str("\n\nContext from the delegating agent:\n"); for (i, c) in context.iter().enumerate() { prompt.push_str(&format!("[{i}] {c}\n")); } } self.drive(alias, &prompt).await } /// Read frames until a terminal (`done`/`error`/`approval_request`) event. /// /// 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. /// `live` is the push target for this turn: `Some((workspace, agent))` when /// the turn belongs to a mission AND runs under a claw alias. `None` for the /// governor/door/evaluator, whose output belongs to no agent. async fn drain( ws: &mut S, live: Option<(uuid::Uuid, uuid::Uuid)>, ) -> Result<(TurnOutcome, ToolTrace), OrchestratorError> where S: StreamExt> + SinkExt + Unpin, { let mut output = String::new(); let mut tokens: u64 = 0; let mut gated: Vec = Vec::new(); let mut trace = ToolTrace::default(); while let Some(frame) = ws.next().await { let msg = frame.map_err(|e| OrchestratorError::Executor(format!("ws recv: {e}")))?; match msg { Message::Text(txt) => { let v: serde_json::Value = serde_json::from_str(txt.as_str()) .map_err(|e| OrchestratorError::Executor(format!("bad frame: {e}")))?; match v.get("type").and_then(|t| t.as_str()).unwrap_or("") { "chunk" => { if let Some(c) = v.get("content").and_then(|c| c.as_str()) { output.push_str(c); // Push, don't wait for the poll. This is the // whole point of the bus: the reasoning card // previously showed a step's text only after the // step ended and the row was written, so an // agent mid-thought looked idle for seconds. if let Some((ws_id, agent_id)) = live { if !c.trim().is_empty() { crate::live_bus::global().publish( ws_id, "agent.reasoning.delta", serde_json::json!({ "agentId": agent_id.to_string(), "text": c, "channel": "say", }), ); } } } } "done" => { let input = v.get("input_tokens").and_then(|n| n.as_u64()).unwrap_or(0); let out = v.get("output_tokens").and_then(|n| n.as_u64()).unwrap_or(0); tokens = input + out; break; } "approval_request" => { // §15: agents are tool-free behind the MCP door, so // this is unexpected. Record it as blocked, never // auto-approve, and end the turn. let tool = v.get("tool").and_then(|t| t.as_str()).unwrap_or("unknown"); let summary = v .get("arguments_summary") .and_then(|s| s.as_str()) .unwrap_or(""); gated.push(GatedAction { category: GatedCategory::OutboundMessage, summary: format!("{tool}: {summary}").trim().to_string(), approved: false, }); break; } "error" => { let m = v .get("message") .and_then(|m| m.as_str()) .unwrap_or("agent error"); return Err(OrchestratorError::Executor(m.to_string())); } "aborted" => { return Err(OrchestratorError::Executor("turn aborted".into())); } // 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" => { // `name` is what the gateway sends; `tool` is // what `approval_request` uses, kept as a fallback. let tool = v .get("name") .or_else(|| v.get("tool")) .and_then(|t| t.as_str()) .unwrap_or("") .trim() .to_string(); if !tool.is_empty() { // `args` FIRST: that is what the gateway // actually sends (`{"type":"tool_call","id", // "name","args"}` — zeroclaw-gateway/src/ws.rs). // The others were guesses, and a guess that // never matches costs the file path silently: // the tool call is still recorded, with no // target, and reads as a tool that touched // nothing. let args = v .get("args") .or_else(|| 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) => { let _ = ws.send(Message::Pong(p)).await; } Message::Close(_) => break, _ => {} } } Ok(( TurnOutcome { output: output.trim().to_string(), 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 { async fn run_turn(&self, req: TurnRequest) -> Result { // An explicit per-node agent (graph `node.attrs["agent"]`) wins, so one // request can pin a different model per role; otherwise use the map. let alias = req .agent .as_deref() .map(str::trim) .filter(|a| !a.is_empty()) .map(str::to_string) .unwrap_or_else(|| { // Falling back here means the graph node was never bound to a // claw, so the turn runs as the default agent with the DEFAULT // agent's workspace and tools — not the mission's. That silently // produced whole missions of unusable output, so say so loudly. let fallback = self.alias_for(&req.role); eprintln!( "topology_exec: node={} role={} has no bound agent — falling back to `{fallback}` \ (its workspace/tools, NOT the mission's)", req.node_id, req.role, ); fallback }); // One lookup, used for the section, its preamble and the record. // Deriving it three times would let a mission compose an index under // an inline heading if the row changed mid-run. let mode = self.skill_delivery_mode().await; let prompt = compose_turn_prompt( &Self::build_prompt(&req), self.pinned_skills_in_mode(&alias, mode).await.as_deref(), mode, ); // Record what this agent is ACTUALLY about to receive, before driving. // Re-deriving it later would re-run the skill lookup against a // catalogue that may have changed — and once agents author their own // skills, it certainly will have. if let Some(tap) = self.tap.as_ref() { let mut ev = crate::mission_events::MissionEvent::new( tap.mission_id, crate::mission_events::PROMPT_COMPOSED, ); ev.phase_id = tap.phase_id; ev.run_id = tap.run_id; ev.agent_id = crate::runtime_provision::claw_from_alias(&alias); ev.target = Some(req.role.clone()); ev.detail = serde_json::json!({ "text": prompt, "tier": "container", // The A/B arm, alongside the prompt it produced. `skill_use` // recovers this from the prompt text itself, so this field is // for reporting and for catching the two disagreeing. "skill_delivery": mode.as_str(), }); crate::mission_events::record(&tap.pool, ev).await; } self.drive(&alias, &prompt).await } } /// The base turn prompt with the agent's pinned skills appended, if it has any. /// /// Split out from `run_turn` so the wiring is testable: `pinned_skills_text` /// working and `run_turn` actually calling it are different claims, and the /// second is the one that was false for every skill in the catalogue. pub fn compose_turn_prompt( base: &str, skills: Option<&str>, mode: crate::skill_delivery::Mode, ) -> String { let Some(skills) = skills.map(str::trim).filter(|s| !s.is_empty()) else { // No heading when there is nothing under it. An empty "Your skills" // section tells the model it has skills and then shows it none, which // is worse than silence. return base.to_string(); }; // The preamble differs per arm and lives in `skill_delivery`, because it // is also what the scorer reads the arm back from. Two copies of this // sentence is two chances for the reader to stop recognising the writer. let preamble = crate::skill_delivery::preamble(mode); format!("{base}\n\n# Your skills\n\n{preamble}\n\n{skills}") } /// Parse `role=alias,role=alias` into a map (blank/malformed entries skipped). fn parse_agent_map(s: &str) -> HashMap { s.split(',') .filter_map(|pair| { let (role, alias) = pair.split_once('=')?; let role = role.trim(); let alias = alias.trim(); if role.is_empty() || alias.is_empty() { None } else { Some((role.to_string(), alias.to_string())) } }) .collect() } #[cfg(test)] mod tests { /// The container name comes out of the gateway URL, or nothing does. /// /// This is only used to fetch a log tail for a failure message, so a URL /// shape it does not recognise must degrade to "no log" rather than to a /// second error on top of the first. #[test] fn the_container_name_is_derived_or_absent_never_wrong() { let ex = |url: &str| { ZeroClawDriveExecutor::new( url.to_string(), String::new(), std::collections::HashMap::new(), "scout".into(), ) }; assert_eq!( ex("http://cm-runtime-mission-019fec2d596f:42617") .container_name() .as_deref(), Some("cm-runtime-mission-019fec2d596f") ); assert_eq!( ex("https://host.example:8443/base") .container_name() .as_deref(), Some("host.example") ); // No scheme is still a host. assert_eq!( ex("clawmates-runtime:42617").container_name().as_deref(), Some("clawmates-runtime") ); assert_eq!(ex("").container_name(), None); } use super::*; use axum::extract::ws::{Message as AxMsg, WebSocket, WebSocketUpgrade}; use axum::response::Response; use axum::routing::{get, post}; use axum::{Json, Router}; use serde_json::{json, Value}; async fn pair() -> Json { Json(json!({ "token": "test-token" })) } fn frames() -> Vec { vec![ json!({"type": "session_start", "session_id": "s1", "resumed": false}), json!({"type": "chunk", "content": "hel"}), json!({"type": "chunk", "content": "lo"}), json!({"type": "done", "input_tokens": 5, "output_tokens": 7}), ] } async fn ok_ws(ws: WebSocketUpgrade) -> Response { ws.on_upgrade(|mut socket: WebSocket| async move { let _ = socket.recv().await; // the client's message for f in frames() { let _ = socket.send(AxMsg::Text(f.to_string().into())).await; } }) } /// 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"}), // 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", "id": "t2", "name": "Bash", "args": {"command": "cargo test"}}), // Prose that MENTIONS a path. It must not become a file touch. json!({"type": "tool_call", "id": "t3", "name": "Grep", "arguments_summary": "searching src/main.rs", "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}), ] { let _ = socket.send(AxMsg::Text(f.to_string().into())).await; } }) } async fn approval_ws(ws: WebSocketUpgrade) -> Response { ws.on_upgrade(|mut socket: WebSocket| async move { let _ = socket.recv().await; for f in [ json!({"type": "chunk", "content": "working"}), json!({"type": "approval_request", "tool": "email.send", "arguments_summary": "to: a@b.c"}), ] { let _ = socket.send(AxMsg::Text(f.to_string().into())).await; } }) } async fn serve(router: Router) -> String { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, router).await.unwrap(); }); format!("http://{addr}") } fn req() -> TurnRequest { TurnRequest { node_id: "a".into(), role: "researcher".into(), agent: None, attrs: Default::default(), task: "say hi".into(), context: vec![], } } #[test] fn explicit_node_agent_overrides_role_map() { let mut map = HashMap::new(); map.insert("researcher".to_string(), "worker_glm".to_string()); let exec = ZeroClawDriveExecutor::new("http://x".into(), "c".into(), map, "scout".into()); // role map → worker_glm assert_eq!(exec.alias_for("researcher"), "worker_glm"); // but an explicit per-node agent should win in run_turn's selection let agent = Some("worker_kimi".to_string()); let chosen = agent .as_deref() .map(str::trim) .filter(|a| !a.is_empty()) .map(str::to_string) .unwrap_or_else(|| exec.alias_for("researcher")); assert_eq!(chosen, "worker_kimi"); } #[tokio::test] async fn drives_a_turn_and_accumulates_output_and_tokens() { let router = Router::new() .route("/pair", post(pair)) .route("/ws/chat", get(ok_ws)); let base = serve(router).await; let exec = ZeroClawDriveExecutor::new(base, "code".into(), HashMap::new(), "scout".into()); let out = exec.run_turn(req()).await.unwrap(); assert_eq!(out.output, "hello"); assert_eq!(out.tokens, 12); 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] async fn approval_request_is_recorded_as_blocked() { let router = Router::new() .route("/pair", post(pair)) .route("/ws/chat", get(approval_ws)); let base = serve(router).await; let exec = ZeroClawDriveExecutor::new(base, "code".into(), HashMap::new(), "scout".into()); let out = exec.run_turn(req()).await.unwrap(); assert_eq!(out.gated.len(), 1); assert!(!out.gated[0].approved); assert!(out.gated[0].summary.contains("email.send")); } #[test] fn agent_map_parses_pairs() { let m = parse_agent_map("researcher=scout, writer=quill ,bad=,=x,ok=y"); assert_eq!(m.get("researcher").unwrap(), "scout"); assert_eq!(m.get("writer").unwrap(), "quill"); assert_eq!(m.get("ok").unwrap(), "y"); assert_eq!(m.len(), 3); } }