Topology P-zc 1A: ZeroClawDriveExecutor — real role-agents over /ws/chat
cm-orchestrator owns the topology graph; each turn now drives a real ZeroClaw
role-agent in a container via the proven gateway drive recipe, instead of a
tool-free cm-llm call.
- crates/cm-api/src/topology_exec.rs: ZeroClawDriveExecutor impl TurnExecutor —
pair (POST /pair + X-Pairing-Code, token cached) -> ws /ws/chat?agent=<alias>
-> send {type:message,content} -> drain chunk/done/approval_request/error.
approval_request is recorded as a BLOCKED GatedAction, never auto-approved (§15).
Role->alias via ZEROCLAW_AGENT_MAP, fallback ZEROCLAW_DEFAULT_AGENT (scout).
- POST /api/topologies/run {task,graph} -> execute() -> RunRecord, persisted
best-effort to the existing topology_runs table (no migration). compare stays
tool-free. from_env() is read in-handler so cm-api still boots unset.
- deploy/clawmates-runtime: example config now declares a tool-free multi-agent
role-cast; README documents the ZEROCLAW_* knobs + run endpoint.
tokio-tungstenite 0.26 (already in lock) + dev axum `ws` for the hermetic test.
3 lib tests green, clippy clean, SQLX_OFFLINE build clean.
Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
7baf2082d0
commit
6d77a0acc1
Generated
+40
-2
@@ -140,6 +140,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||||||
checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
|
checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"axum-core",
|
"axum-core",
|
||||||
|
"base64",
|
||||||
"bytes",
|
"bytes",
|
||||||
"form_urlencoded",
|
"form_urlencoded",
|
||||||
"futures-util",
|
"futures-util",
|
||||||
@@ -158,8 +159,10 @@ dependencies = [
|
|||||||
"serde_json",
|
"serde_json",
|
||||||
"serde_path_to_error",
|
"serde_path_to_error",
|
||||||
"serde_urlencoded",
|
"serde_urlencoded",
|
||||||
|
"sha1",
|
||||||
"sync_wrapper",
|
"sync_wrapper",
|
||||||
"tokio",
|
"tokio",
|
||||||
|
"tokio-tungstenite 0.29.0",
|
||||||
"tower",
|
"tower",
|
||||||
"tower-layer",
|
"tower-layer",
|
||||||
"tower-service",
|
"tower-service",
|
||||||
@@ -513,6 +516,7 @@ dependencies = [
|
|||||||
"thiserror",
|
"thiserror",
|
||||||
"time",
|
"time",
|
||||||
"tokio",
|
"tokio",
|
||||||
|
"tokio-tungstenite 0.26.2",
|
||||||
"tower-http",
|
"tower-http",
|
||||||
"urlencoding",
|
"urlencoding",
|
||||||
"uuid",
|
"uuid",
|
||||||
@@ -1953,7 +1957,7 @@ dependencies = [
|
|||||||
"serde_yaml",
|
"serde_yaml",
|
||||||
"thiserror",
|
"thiserror",
|
||||||
"tokio",
|
"tokio",
|
||||||
"tokio-tungstenite",
|
"tokio-tungstenite 0.26.2",
|
||||||
"tokio-util",
|
"tokio-util",
|
||||||
"tower",
|
"tower",
|
||||||
"tower-http",
|
"tower-http",
|
||||||
@@ -4005,8 +4009,24 @@ checksum = "7a9daff607c6d2bf6c16fd681ccb7eecc83e4e2cdc1ca067ffaadfca5de7f084"
|
|||||||
dependencies = [
|
dependencies = [
|
||||||
"futures-util",
|
"futures-util",
|
||||||
"log",
|
"log",
|
||||||
|
"rustls",
|
||||||
|
"rustls-pki-types",
|
||||||
"tokio",
|
"tokio",
|
||||||
"tungstenite",
|
"tokio-rustls",
|
||||||
|
"tungstenite 0.26.2",
|
||||||
|
"webpki-roots 0.26.11",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "tokio-tungstenite"
|
||||||
|
version = "0.29.0"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c"
|
||||||
|
dependencies = [
|
||||||
|
"futures-util",
|
||||||
|
"log",
|
||||||
|
"tokio",
|
||||||
|
"tungstenite 0.29.0",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
@@ -4251,11 +4271,29 @@ dependencies = [
|
|||||||
"httparse",
|
"httparse",
|
||||||
"log",
|
"log",
|
||||||
"rand 0.9.4",
|
"rand 0.9.4",
|
||||||
|
"rustls",
|
||||||
|
"rustls-pki-types",
|
||||||
"sha1",
|
"sha1",
|
||||||
"thiserror",
|
"thiserror",
|
||||||
"utf-8",
|
"utf-8",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "tungstenite"
|
||||||
|
version = "0.29.0"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8"
|
||||||
|
dependencies = [
|
||||||
|
"bytes",
|
||||||
|
"data-encoding",
|
||||||
|
"http",
|
||||||
|
"httparse",
|
||||||
|
"log",
|
||||||
|
"rand 0.9.4",
|
||||||
|
"sha1",
|
||||||
|
"thiserror",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "typenum"
|
name = "typenum"
|
||||||
version = "1.20.1"
|
version = "1.20.1"
|
||||||
|
|||||||
@@ -29,6 +29,7 @@ cm-scheduler = { path = "../cm-scheduler" }
|
|||||||
cm-secrets = { path = "../cm-secrets" }
|
cm-secrets = { path = "../cm-secrets" }
|
||||||
cm-topology = { path = "../cm-topology" }
|
cm-topology = { path = "../cm-topology" }
|
||||||
thiserror = { workspace = true }
|
thiserror = { workspace = true }
|
||||||
|
tokio-tungstenite = { version = "0.26", features = ["rustls-tls-webpki-roots"] }
|
||||||
tower-http = { version = "0.6", features = ["trace"] }
|
tower-http = { version = "0.6", features = ["trace"] }
|
||||||
time = { workspace = true }
|
time = { workspace = true }
|
||||||
tokio = { workspace = true }
|
tokio = { workspace = true }
|
||||||
@@ -36,6 +37,7 @@ urlencoding = "2"
|
|||||||
uuid = { workspace = true }
|
uuid = { workspace = true }
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
|
axum = { version = "0.8", features = ["ws"] }
|
||||||
jsonwebtoken = "9"
|
jsonwebtoken = "9"
|
||||||
eventsource-stream = "0.2"
|
eventsource-stream = "0.2"
|
||||||
reqwest = { version = "0.12", default-features = false, features = [
|
reqwest = { version = "0.12", default-features = false, features = [
|
||||||
|
|||||||
@@ -3,6 +3,7 @@
|
|||||||
mod error;
|
mod error;
|
||||||
mod extract;
|
mod extract;
|
||||||
mod routes;
|
mod routes;
|
||||||
|
mod topology_exec;
|
||||||
|
|
||||||
use axum::routing::{delete, get, patch, post};
|
use axum::routing::{delete, get, patch, post};
|
||||||
use axum::Router;
|
use axum::Router;
|
||||||
@@ -139,6 +140,7 @@ pub fn router(state: AppState) -> Router {
|
|||||||
.route("/api/topologies/classify", post(routes::topology::classify_graph))
|
.route("/api/topologies/classify", post(routes::topology::classify_graph))
|
||||||
.route("/api/topologies/build", post(routes::topology::build_graph))
|
.route("/api/topologies/build", post(routes::topology::build_graph))
|
||||||
.route("/api/topologies/compare", post(routes::topology::compare_topologies))
|
.route("/api/topologies/compare", post(routes::topology::compare_topologies))
|
||||||
|
.route("/api/topologies/run", post(routes::topology::run_topology))
|
||||||
.route("/api/topology-runs", get(routes::topology::list_runs))
|
.route("/api/topology-runs", get(routes::topology::list_runs))
|
||||||
.route("/api/topology-runs/{id}", get(routes::topology::get_run))
|
.route("/api/topology-runs/{id}", get(routes::topology::get_run))
|
||||||
.layer(tower_http::trace::TraceLayer::new_for_http())
|
.layer(tower_http::trace::TraceLayer::new_for_http())
|
||||||
|
|||||||
@@ -5,7 +5,7 @@
|
|||||||
|
|
||||||
use axum::extract::{Path, State};
|
use axum::extract::{Path, State};
|
||||||
use axum::Json;
|
use axum::Json;
|
||||||
use cm_orchestrator::{compare, Comparison, JudgeScorer, ProviderExecutor};
|
use cm_orchestrator::{compare, execute, Comparison, JudgeScorer, ProviderExecutor, RunRecord};
|
||||||
use cm_topology::{build, classify, heuristics, Classification, TopologyGraph, TopologyKind};
|
use cm_topology::{build, classify, heuristics, Classification, TopologyGraph, TopologyKind};
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use time::format_description::well_known::Rfc3339;
|
use time::format_description::well_known::Rfc3339;
|
||||||
@@ -113,6 +113,42 @@ pub async fn compare_topologies(
|
|||||||
Ok(Json(cmp))
|
Ok(Json(cmp))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Request body for executing a single topology on a real agent container.
|
||||||
|
#[derive(Deserialize)]
|
||||||
|
pub struct RunRequest {
|
||||||
|
pub task: String,
|
||||||
|
pub graph: TopologyGraph,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `POST /api/topologies/run` — execute one topology by driving real ZeroClaw
|
||||||
|
/// role-agents (in a container) for each turn, returning the run journal. The
|
||||||
|
/// orchestrator owns the graph; agents are tool-free behind the Clawmates MCP
|
||||||
|
/// door, so §15 holds by construction. Result is persisted best-effort to the
|
||||||
|
/// existing `topology_runs` table.
|
||||||
|
pub async fn run_topology(
|
||||||
|
State(state): State<AppState>,
|
||||||
|
Authed(user): Authed,
|
||||||
|
Json(req): Json<RunRequest>,
|
||||||
|
) -> Result<Json<RunRecord>, ApiError> {
|
||||||
|
let executor =
|
||||||
|
crate::topology_exec::ZeroClawDriveExecutor::from_env().map_err(|_| ApiError::Internal)?;
|
||||||
|
let record = execute(&req.graph, &req.task, &executor)
|
||||||
|
.await
|
||||||
|
.map_err(|_| ApiError::Internal)?;
|
||||||
|
|
||||||
|
if let Ok(value) = serde_json::to_value(&record) {
|
||||||
|
let _ = cm_db::repo::topology_runs::insert(
|
||||||
|
&state.pool,
|
||||||
|
Uuid::now_v7(),
|
||||||
|
user.workspace_id,
|
||||||
|
&req.task,
|
||||||
|
&value,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
Ok(Json(record))
|
||||||
|
}
|
||||||
|
|
||||||
/// A saved comparison run, summarized.
|
/// A saved comparison run, summarized.
|
||||||
#[derive(Serialize)]
|
#[derive(Serialize)]
|
||||||
pub struct RunSummary {
|
pub struct RunSummary {
|
||||||
|
|||||||
@@ -0,0 +1,375 @@
|
|||||||
|
//! 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=<alias>`, sends the role+task+context prompt,
|
||||||
|
//! and streams the turn's events back into a [`TurnOutcome`].
|
||||||
|
//!
|
||||||
|
//! **§15 by construction:** the agents are provisioned tool-free (every
|
||||||
|
//! sensitive capability is a gated Clawmates MCP tool — the "door"), so a turn
|
||||||
|
//! takes no sandbox-leaving action here. If the gateway nonetheless emits an
|
||||||
|
//! `approval_request`, we 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.
|
||||||
|
const TURN_TIMEOUT: Duration = Duration::from_secs(90);
|
||||||
|
|
||||||
|
/// Drives ZeroClaw role-agents (in one container) to execute topology turns.
|
||||||
|
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<String, String>,
|
||||||
|
/// Alias used when a role isn't mapped.
|
||||||
|
default_alias: String,
|
||||||
|
/// Bearer token, paired lazily and reused across turns.
|
||||||
|
token: Arc<Mutex<Option<String>>>,
|
||||||
|
http: reqwest::Client,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ZeroClawDriveExecutor {
|
||||||
|
/// Build an executor explicitly (used in tests).
|
||||||
|
pub fn new(
|
||||||
|
gateway_url: String,
|
||||||
|
pairing_code: String,
|
||||||
|
role_aliases: HashMap<String, String>,
|
||||||
|
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(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Build from the environment:
|
||||||
|
/// - `ZEROCLAW_GATEWAY_URL` (required) e.g. `http://127.0.0.1:42617`
|
||||||
|
/// - `ZEROCLAW_PAIRING_CODE` (required)
|
||||||
|
/// - `ZEROCLAW_AGENT_MAP` (optional) `role=alias,role=alias`
|
||||||
|
/// - `ZEROCLAW_DEFAULT_AGENT` (optional, default `scout`)
|
||||||
|
pub fn from_env() -> Result<Self, String> {
|
||||||
|
let gateway_url =
|
||||||
|
std::env::var("ZEROCLAW_GATEWAY_URL").map_err(|_| "ZEROCLAW_GATEWAY_URL not set")?;
|
||||||
|
let pairing_code =
|
||||||
|
std::env::var("ZEROCLAW_PAIRING_CODE").map_err(|_| "ZEROCLAW_PAIRING_CODE not set")?;
|
||||||
|
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<String, OrchestratorError> {
|
||||||
|
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());
|
||||||
|
Ok(token)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 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}")
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn drive(&self, alias: &str, prompt: &str) -> Result<TurnOutcome, 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 = tokio::time::timeout(TURN_TIMEOUT, Self::drain(&mut ws))
|
||||||
|
.await
|
||||||
|
.map_err(|_| OrchestratorError::Executor("turn timed out".into()))??;
|
||||||
|
let _ = ws.close(None).await;
|
||||||
|
Ok(outcome)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Read frames until a terminal (`done`/`error`/`approval_request`) event.
|
||||||
|
async fn drain<S>(ws: &mut S) -> Result<TurnOutcome, OrchestratorError>
|
||||||
|
where
|
||||||
|
S: StreamExt<Item = Result<Message, tokio_tungstenite::tungstenite::Error>>
|
||||||
|
+ SinkExt<Message>
|
||||||
|
+ Unpin,
|
||||||
|
{
|
||||||
|
let mut output = String::new();
|
||||||
|
let mut tokens: u64 = 0;
|
||||||
|
let mut gated: Vec<GatedAction> = Vec::new();
|
||||||
|
|
||||||
|
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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
"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()));
|
||||||
|
}
|
||||||
|
// session_start, thinking, tool_call, tool_result, …
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Message::Ping(p) => {
|
||||||
|
let _ = ws.send(Message::Pong(p)).await;
|
||||||
|
}
|
||||||
|
Message::Close(_) => break,
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(TurnOutcome {
|
||||||
|
output: output.trim().to_string(),
|
||||||
|
tokens,
|
||||||
|
gated,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl TurnExecutor for ZeroClawDriveExecutor {
|
||||||
|
async fn run_turn(&self, req: TurnRequest) -> Result<TurnOutcome, OrchestratorError> {
|
||||||
|
let alias = self.alias_for(&req.role);
|
||||||
|
let prompt = Self::build_prompt(&req);
|
||||||
|
self.drive(&alias, &prompt).await
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Parse `role=alias,role=alias` into a map (blank/malformed entries skipped).
|
||||||
|
fn parse_agent_map(s: &str) -> HashMap<String, String> {
|
||||||
|
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 {
|
||||||
|
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<Value> {
|
||||||
|
Json(json!({ "token": "test-token" }))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn frames() -> Vec<Value> {
|
||||||
|
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;
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
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: [email protected]"}),
|
||||||
|
] {
|
||||||
|
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(),
|
||||||
|
task: "say hi".into(),
|
||||||
|
context: vec![],
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[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());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[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);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -74,6 +74,22 @@ Drive client: `tools/runtime-spike/drive.mjs` (Node ≥22). Example:
|
|||||||
PAIR_CODE=<code> node tools/runtime-spike/drive.mjs http://<host>:42617 scout "hi"
|
PAIR_CODE=<code> node tools/runtime-spike/drive.mjs http://<host>:42617 scout "hi"
|
||||||
```
|
```
|
||||||
|
|
||||||
|
## Driving topologies from Clawmates (`POST /api/topologies/run`)
|
||||||
|
|
||||||
|
`cm-orchestrator` owns the topology graph; for each turn it drives one ZeroClaw role-agent in this
|
||||||
|
container over the recipe above, via `ZeroClawDriveExecutor` (`crates/cm-api/src/topology_exec.rs`).
|
||||||
|
The server reads these env knobs (unset = the endpoint 500s; the rest of cm-api is unaffected):
|
||||||
|
|
||||||
|
- `ZEROCLAW_GATEWAY_URL` — e.g. `http://127.0.0.1:42617`
|
||||||
|
- `ZEROCLAW_PAIRING_CODE` — the one-time code from the daemon startup log
|
||||||
|
- `ZEROCLAW_AGENT_MAP` — optional `role=alias,role=alias` (e.g. `analyst=researcher`)
|
||||||
|
- `ZEROCLAW_DEFAULT_AGENT` — fallback alias for unmapped roles (default `scout`)
|
||||||
|
|
||||||
|
`agent.config.example.toml` declares the multi-agent role-cast (coordinator/researcher/writer/worker
|
||||||
|
+ `scout`), all tool-free. Manual E2E: bring up the stack, grab the pair code, sanity-drive one alias
|
||||||
|
with `drive.mjs`, then `POST /api/topologies/run {task, graph}` (build a graph via
|
||||||
|
`/api/topologies/build`) and confirm the `RunRecord` has real per-step outputs + per-turn tokens.
|
||||||
|
|
||||||
## Next
|
## Next
|
||||||
|
|
||||||
- [ ] **Demonstrate a §15 gate**: give `scout` a tool under the Supervised profile and confirm the
|
- [ ] **Demonstrate a §15 gate**: give `scout` a tool under the Supervised profile and confirm the
|
||||||
|
|||||||
@@ -1,19 +1,59 @@
|
|||||||
# Phase-1 spike config for a tenant runtime: one provider model + one agent.
|
# Phase-1 runtime config for a tenant: one provider model + a multi-agent
|
||||||
|
# role-cast. ONE container hosts the whole cast; Clawmates' cm-orchestrator owns
|
||||||
|
# the topology graph and drives a specific role by alias over /ws/chat?agent=...
|
||||||
|
#
|
||||||
# Secrets are NOT committed — the API key is injected at runtime via env
|
# Secrets are NOT committed — the API key is injected at runtime via env
|
||||||
# ZEROCLAW_providers__models__groq__default__api_key=<key>
|
# ZEROCLAW_providers__models__groq__default__api_key=<key>
|
||||||
# (double-underscore = config nesting). Swap groq→anthropic for prod.
|
# (double-underscore = config nesting). Swap groq→anthropic for prod.
|
||||||
|
|
||||||
# Mark onboarding complete (headless equivalent of the browser Quickstart) so
|
# Mark onboarding complete (headless equivalent of the browser Quickstart) so
|
||||||
# the agent will answer; otherwise /ws/chat returns NEEDS_ONBOARDING.
|
# the agents will answer; otherwise /ws/chat returns NEEDS_ONBOARDING.
|
||||||
[onboard_state]
|
[onboard_state]
|
||||||
quickstart_completed = true
|
quickstart_completed = true
|
||||||
|
|
||||||
[providers.models.groq.default]
|
[providers.models.groq.default]
|
||||||
model = "llama-3.3-70b-versatile"
|
model = "llama-3.3-70b-versatile"
|
||||||
|
|
||||||
# One named agent == one Clawmates "claw". For real §15 we lock tools.allow and
|
# §15 door (Step 1A): a tool-free profile — agents can only reason, no native
|
||||||
# add an MCP client → the Clawmates gated door; for the spike we keep it minimal
|
# outbound tools. Step 1B adds an MCP client bundle (the Clawmates gated door),
|
||||||
# and use the Supervised profile so sensitive tools raise an approval_request.
|
# whose tools are injected AFTER this allowlist filter, so a tool-free agent can
|
||||||
|
# still call ONLY the gated Clawmates tools.
|
||||||
|
[risk_profiles.toolfree]
|
||||||
|
level = "supervised"
|
||||||
|
allowed_tools = []
|
||||||
|
excluded_tools = ["shell", "file_read", "file_write", "http_request", "browser", "composio"]
|
||||||
|
|
||||||
|
# The role-cast. node.role → agent alias is configured Clawmates-side via
|
||||||
|
# ZEROCLAW_AGENT_MAP (e.g. "analyst=researcher"); `scout` is the default
|
||||||
|
# fallback (ZEROCLAW_DEFAULT_AGENT) for any unmapped role.
|
||||||
|
[agents.coordinator]
|
||||||
|
model_provider = "groq.default"
|
||||||
|
risk_profile = "toolfree"
|
||||||
|
|
||||||
|
[agents.researcher]
|
||||||
|
model_provider = "groq.default"
|
||||||
|
risk_profile = "toolfree"
|
||||||
|
|
||||||
|
[agents.writer]
|
||||||
|
model_provider = "groq.default"
|
||||||
|
risk_profile = "toolfree"
|
||||||
|
|
||||||
|
[agents.worker]
|
||||||
|
model_provider = "groq.default"
|
||||||
|
risk_profile = "toolfree"
|
||||||
|
|
||||||
|
# Default fallback alias for roles not present above.
|
||||||
[agents.scout]
|
[agents.scout]
|
||||||
model_provider = "groq.default"
|
model_provider = "groq.default"
|
||||||
risk_profile = "supervised"
|
risk_profile = "toolfree"
|
||||||
|
|
||||||
|
# Step 1B — the Clawmates §15 MCP door (uncomment + point at the MCP server):
|
||||||
|
# [mcp]
|
||||||
|
# enabled = true
|
||||||
|
# deferred_loading = true
|
||||||
|
# [mcp.servers.clawmates]
|
||||||
|
# transport = "http"
|
||||||
|
# url = "http://mcp:3000/mcp"
|
||||||
|
# [mcp_bundles.clawmates_door]
|
||||||
|
# servers = ["clawmates"]
|
||||||
|
# then add `mcp_bundles = ["clawmates_door"]` to each [agents.*] block above.
|
||||||
|
|||||||
Reference in New Issue
Block a user