diff --git a/Cargo.lock b/Cargo.lock index 2110002..7d91c96 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -998,7 +998,7 @@ dependencies = [ [[package]] name = "clawmates-node" -version = "0.4.0" +version = "0.5.0" dependencies = [ "base64 0.22.1", "bytes", diff --git a/crates/bins/clawmates-node/Cargo.toml b/crates/bins/clawmates-node/Cargo.toml index ad6189e..e629668 100644 --- a/crates/bins/clawmates-node/Cargo.toml +++ b/crates/bins/clawmates-node/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "clawmates-node" -version = "0.4.0" +version = "0.5.0" edition.workspace = true rust-version.workspace = true license.workspace = true diff --git a/crates/bins/clawmates-node/src/local_model.rs b/crates/bins/clawmates-node/src/local_model.rs index 95ab9a7..50c5d11 100644 --- a/crates/bins/clawmates-node/src/local_model.rs +++ b/crates/bins/clawmates-node/src/local_model.rs @@ -56,14 +56,42 @@ pub fn uses_local_model(backend: Option<&str>) -> bool { /// case. An error means it should have had one and could not — reported by the /// caller, never silently swallowed, because the symptom otherwise is an agent /// that hangs on its first turn. +/// Where a VM's model pipe leads: the node's own model, or the server's LLM +/// proxy for a backend whose credential the guest must not hold. +/// +/// The relay is how a microVM reaches a hosted model WITHOUT a provider key in +/// the guest. The guest's CLI points at its loopback (`127.0.0.1:11434`, which +/// fcagent already pipes here for every backend), holds a per-mission token, +/// and the server's proxy adds the real credential. This node copies bytes and +/// never sees a key. Only a tailnet address is accepted, so no message from the +/// server can point a node's pipe at the internet. +pub fn target_for(backend: Option<&str>, relay: Option<&str>) -> Result, String> { + if uses_local_model(backend) { + return Ok(Some(OLLAMA_ADDR.to_string())); + } + let Some(r) = relay.map(str::trim).filter(|r| !r.is_empty()) else { + return Ok(None); + }; + let addr: std::net::SocketAddr = r + .parse() + .map_err(|e| format!("model relay {r:?} is not an ip:port ({e})"))?; + match addr.ip() { + std::net::IpAddr::V4(ip) if ip.octets()[0] == 100 && (64..128).contains(&ip.octets()[1]) => { + Ok(Some(addr.to_string())) + } + _ => Err(format!("model relay {r} is not a tailnet (100.64.0.0/10) address — refused")), + } +} + pub fn start( uds: &Path, vm_id: &str, backend: Option<&str>, + relay: Option<&str>, ) -> Result)>, String> { - if !uses_local_model(backend) { + let Some(target) = target_for(backend, relay)? else { return Ok(None); - } + }; let path = PathBuf::from(format!("{}_{}", uds.display(), MODEL_PORT)); // Firecracker leaves these behind exactly as it does its own socket, and a // stale file makes bind fail with EADDRINUSE. @@ -72,7 +100,7 @@ pub fn start( UnixListener::bind(&path).map_err(|e| format!("bind {}: {e}", path.display()))?; eprintln!( - "microvm {vm_id}: local model socket on {} -> {OLLAMA_ADDR}", + "microvm {vm_id}: model socket on {} -> {target}", path.display() ); let vm = vm_id.to_string(); @@ -81,8 +109,9 @@ pub fn start( match listener.accept().await { Ok((s, _)) => { let vm = vm.clone(); + let target = target.clone(); tokio::spawn(async move { - if let Err(e) = pipe(s).await { + if let Err(e) = pipe(s, &target).await { // Loud, because the failure a mission sees is a turn // that never answers. A refused connection here means // the node's model server is down, and that is worth @@ -102,10 +131,10 @@ pub fn start( } /// Splice one guest connection onto a fresh connection to the node's model. -async fn pipe(mut guest: tokio::net::UnixStream) -> Result<(), String> { - let mut model = TcpStream::connect(OLLAMA_ADDR) +async fn pipe(mut guest: tokio::net::UnixStream, target: &str) -> Result<(), String> { + let mut model = TcpStream::connect(target) .await - .map_err(|e| format!("connect {OLLAMA_ADDR}: {e}"))?; + .map_err(|e| format!("connect {target}: {e}"))?; tokio::io::copy_bidirectional(&mut guest, &mut model) .await .map(|_| ()) @@ -116,6 +145,25 @@ async fn pipe(mut guest: tokio::net::UnixStream) -> Result<(), String> { mod tests { use super::*; + /// The local backend still pipes to Ollama, whatever relay is offered. + #[test] + fn the_local_backend_always_gets_the_nodes_own_model() { + assert_eq!(target_for(Some("local-ornith"), Some("100.102.112.85:8089")).unwrap().as_deref(), Some(OLLAMA_ADDR)); + } + + /// A hosted backend relays only when told to, and only to the tailnet. + #[test] + fn a_hosted_backend_relays_only_to_a_tailnet_address() { + assert_eq!(target_for(Some("claude"), None).unwrap(), None, "no relay offered: no pipe, as before"); + assert_eq!( + target_for(Some("glm"), Some("100.102.112.85:8089")).unwrap().as_deref(), + Some("100.102.112.85:8089") + ); + for bad in ["8.8.8.8:443", "127.0.0.1:8089", "10.0.0.5:8089", "100.128.0.1:8089", "api.z.ai:443", "100.102.112.85"] { + assert!(target_for(Some("claude"), Some(bad)).is_err(), "{bad} must be refused"); + } + } + /// Only the backends that are meant to have a local model get one. /// /// The negative half is the point: an unrecognised backend acquiring a route @@ -144,24 +192,31 @@ mod tests { } } - /// The guest cannot name a destination, so there is nothing to validate. + /// The guest still cannot name a destination. /// - /// This asserts the property that makes this module safe enough to skip the - /// allow-list entirely: the upstream address is a constant. If it ever - /// becomes a parameter, this file needs everything `egress` has. + /// This module used to dial one constant, and that was what let it skip the + /// allow-list. It now has two destinations — the node's own model, and the + /// server's LLM proxy as a relay — but the property the constant protected + /// holds: the pipe protocol carries no address, the destination is chosen + /// by `target_for` from the SERVER's `vm_create` message, and a relay is + /// accepted only on the tailnet (see `a_hosted_backend_relays_only_to_a_tailnet_address`). + /// If the guest ever gets to supply a target, this file needs everything + /// `egress` has. #[test] - fn the_upstream_address_is_a_constant_not_an_input() { + fn the_guest_never_chooses_where_the_pipe_goes() { let src = include_str!("local_model.rs"); // Needles are split so they do not match themselves in this file. assert_eq!( src.matches(concat!("TcpStream", "::connect(")).count(), 1, - "exactly one dial site, and it must use the constant" + "exactly one dial site" ); - assert!(src.contains(concat!("TcpStream", "::connect(OLLAMA_ADDR)"))); + assert!(src.contains(concat!("TcpStream", "::connect(target)"))); + // `target` reaches `pipe` only from `start`, which gets it only from `target_for`. + assert_eq!(src.matches(concat!("target_for", "(backend, relay)")).count(), 1); assert!( OLLAMA_ADDR.starts_with("127.0.0.1:"), - "the model server must be reached on loopback only" + "the node's model server must be reached on loopback only" ); } } diff --git a/crates/bins/clawmates-node/src/main.rs b/crates/bins/clawmates-node/src/main.rs index 857fd9a..dbcc7a7 100644 --- a/crates/bins/clawmates-node/src/main.rs +++ b/crates/bins/clawmates-node/src/main.rs @@ -328,6 +328,10 @@ fn capabilities_from(kvm: bool, firecracker: Option<&str>, backends: &[String]) // binary but no KVM is gw-04. Computed here rather than in the // scheduler so the rule sits next to the probe that feeds it. "microvm": kvm && firecracker.is_some(), + // The VM model pipe can relay to the server's LLM proxy, so a guest on a + // hosted backend needs no provider key (local_model::target_for). The + // server relays only to nodes that say so; older nodes keep the key. + "model_relay": true, }) } diff --git a/crates/bins/clawmates-node/src/microvm.rs b/crates/bins/clawmates-node/src/microvm.rs index 94304d6..c4ed148 100644 --- a/crates/bins/clawmates-node/src/microvm.rs +++ b/crates/bins/clawmates-node/src/microvm.rs @@ -328,6 +328,7 @@ pub async fn create( vcpus: u32, mem_mib: u32, backend: Option<&str>, + model_relay: Option<&str>, ) -> Result { check_id(vm_id)?; let golden = rootfs_for(backend)?; @@ -399,7 +400,7 @@ pub async fn create( // The node's own model, for a backend that has one. Bound before firecracker // starts for the same reason egress is: a guest that dials before the host // listens gets a refusal it will not retry. - let (model_uds, model_task) = match crate::local_model::start(&uds, vm_id, backend) { + let (model_uds, model_task) = match crate::local_model::start(&uds, vm_id, backend, model_relay) { Ok(Some((p, t))) => (Some(p), Some(t)), Ok(None) => (None, None), // This backend was supposed to have a local model and does not. Not @@ -678,6 +679,7 @@ pub async fn handle_op(op: &str, v: &Value, vms: &Vms) -> (bool, String) { u("vcpus", 2) as u32, u("mem_mib", 2048) as u32, v.get("backend").and_then(Value::as_str), + v.get("model_relay").and_then(Value::as_str), ) .await } @@ -747,7 +749,7 @@ pub async fn selftest() -> bool { // default — booting the wrong rootfs would report success for whatever came // out of it. Checked here so the guarantee is exercised on real hardware and // not only in a unit test with a temp dir. - match create(&vms, "selftest-absent", 2, 512, Some("definitely-not-built")).await { + match create(&vms, "selftest-absent", 2, 512, Some("definitely-not-built"), None).await { Err(e) if e.contains("rootfs-definitely-not-built.ext4") => { check(true, "an absent backend image fails by name", String::new()) } @@ -763,7 +765,7 @@ pub async fn selftest() -> bool { } let started = std::time::Instant::now(); - let created = match create(&vms, id, 2, 1024, backend.as_deref()).await { + let created = match create(&vms, id, 2, 1024, backend.as_deref(), None).await { Ok(v) => { check( true, diff --git a/crates/cm-api/src/llm_proxy.rs b/crates/cm-api/src/llm_proxy.rs index dbffef7..0712d68 100644 --- a/crates/cm-api/src/llm_proxy.rs +++ b/crates/cm-api/src/llm_proxy.rs @@ -98,6 +98,54 @@ pub fn base_url() -> Option { (!host.is_empty()).then(|| format!("http://{host}:{PORT}")) } +/// Where fleet nodes reach the proxy: the server's TAILNET address and the +/// proxy port, e.g. `100.102.112.85:8089` (published on that address only, +/// never on a public interface). Set = microVM missions may be relayed. +pub fn node_relay_addr() -> Option { + std::env::var("CLAWMATES_LLM_PROXY_NODE_ADDR") + .ok() + .map(|v| v.trim().to_string()) + .filter(|v| !v.is_empty()) +} + +/// The proxy route a microVM backend's CLI speaks to, or `None` for a backend +/// that reaches no hosted provider (`local-ornith`) or that nobody taught this +/// function about — the same fail-closed rule as the node's `provider_hosts`. +pub fn microvm_route(backend: Option<&str>) -> Option<&'static str> { + match backend { + None | Some("") | Some("default") | Some("claude") | Some("canary-claude") => Some("anthropic"), + Some("glm") => Some("glm"), + Some("kimi") => Some("kimi"), + _ => None, + } +} + +/// Should this microVM phase reach its model through the proxy? Only when the +/// proxy is on, a node address is configured, the backend has a route, and the +/// NODE says it can relay (daemon 0.5.0+ reports `model_relay`). An older node +/// keeps the old path — the key in the guest — rather than a guest whose model +/// calls go nowhere. +pub async fn microvm_relay(pool: &PgPool, node: Uuid, backend: Option<&str>) -> Option { + if !enabled() || microvm_route(backend).is_none() { + return None; + } + let addr = node_relay_addr()?; + let relays: bool = sqlx::query_scalar( + "SELECT coalesce((capabilities->>'model_relay')::boolean, false) FROM nodes WHERE id = $1", + ) + .bind(node) + .fetch_optional(pool) + .await + .ok() + .flatten() + .unwrap_or(false); + if !relays { + eprintln!("llm_proxy: node {node} cannot relay model calls (daemon < 0.5.0) — its guest gets the provider key"); + return None; + } + Some(addr) +} + /// One upstream: where it lives and the real credential it takes. #[derive(Debug, PartialEq, Eq)] pub struct Upstream { @@ -338,6 +386,33 @@ mod tests { assert!(DROP_REQUEST.contains(&"authorization") && DROP_REQUEST.contains(&"x-api-key")); } + /// A relayed guest holds the token under the name its CLI reads and points + /// at its own loopback model port — never a provider host, never a key. + #[test] + fn a_relayed_guest_gets_the_token_and_a_loopback_base_url() { + for (backend, cred, route) in [ + (Some("claude"), "CLAUDE_CODE_OAUTH_TOKEN", "anthropic"), + (None, "CLAUDE_CODE_OAUTH_TOKEN", "anthropic"), + (Some("glm"), "ANTHROPIC_AUTH_TOKEN", "glm"), + (Some("kimi"), "ANTHROPIC_AUTH_TOKEN", "kimi"), + ] { + let env = crate::mission_runtime::microvm_proxied_env(backend, "cmlp.m.s").unwrap(); + let get = |k: &str| env.iter().find(|(n, _)| n == k).map(|(_, v)| v.as_str()); + assert_eq!(get(cred), Some("cmlp.m.s"), "{backend:?}"); + assert_eq!(get("ANTHROPIC_BASE_URL"), Some(format!("http://127.0.0.1:11434/{route}").as_str())); + assert_eq!(env.len(), 2, "nothing else — in particular no other key: {env:?}"); + } + } + + /// A local model and an unknown backend have no route, so they are never + /// relayed (the local one keeps its own pipe to the node's model). + #[test] + fn local_and_unknown_backends_are_not_relayed() { + assert_eq!(microvm_route(Some("local-ornith")), None); + assert_eq!(microvm_route(Some("something-new")), None); + assert!(crate::mission_runtime::microvm_proxied_env(Some("local-ornith"), "t").is_err()); + } + #[test] fn only_known_providers_route() { assert!(upstream("evil.example").is_none()); diff --git a/crates/cm-api/src/microvm_client.rs b/crates/cm-api/src/microvm_client.rs index a61e87f..57036b7 100644 --- a/crates/cm-api/src/microvm_client.rs +++ b/crates/cm-api/src/microvm_client.rs @@ -98,14 +98,18 @@ impl<'a> MicroVm<'a> { vcpus: u32, mem_mib: u32, backend: Option<&str>, + model_relay: Option<&str>, ) -> Result { // 60s, not the hub default: a create that has to copy a rootfs and boot // is measured near 1s, but a node under load has no reason to be fast. - self.call( - "vm_create", - json!({ "vcpus": vcpus, "mem_mib": mem_mib, "backend": backend }), - 60, - ) + // + // `model_relay` is omitted, not sent as null, when absent: an older node + // ignores unknown fields either way, but the absence is the old path. + let mut req = json!({ "vcpus": vcpus, "mem_mib": mem_mib, "backend": backend }); + if let Some(r) = model_relay { + req["model_relay"] = json!(r); + } + self.call("vm_create", req, 60) .await } diff --git a/crates/cm-api/src/microvm_executor.rs b/crates/cm-api/src/microvm_executor.rs index f86d73f..7a3d978 100644 --- a/crates/cm-api/src/microvm_executor.rs +++ b/crates/cm-api/src/microvm_executor.rs @@ -493,10 +493,18 @@ pub struct ToolGateOutcome { pub async fn run_phase_in_vm(hub: &NodeHub, p: VmPhase<'_>) -> Result { // Resolved BEFORE the VM boots: a missing subscription token must fail the // phase, not boot a VM whose agent will sit there unauthenticated. - let env = crate::mission_runtime::microvm_provider_env(p.backend)?; + // + // With a relay, the guest holds the mission's proxy token instead of the + // provider key, and its model calls go loopback → vsock → node → server + // proxy, which adds the credential. See `llm_proxy`. + let env = match (&p.model_relay, crate::llm_proxy::token_for(p.mission_id)) { + (Some(_), Some(token)) => crate::mission_runtime::microvm_proxied_env(p.backend, &token)?, + _ => crate::mission_runtime::microvm_provider_env(p.backend)?, + }; + let relay = p.model_relay.as_deref().filter(|_| crate::llm_proxy::token_for(p.mission_id).is_some()); let vm = MicroVm::new(hub, p.node_id, vm_id_for(p.phase_id, p.iteration, p.step)); - let created = vm.create(VCPUS, MEM_MIB, p.backend).await?; + let created = vm.create(VCPUS, MEM_MIB, p.backend, relay).await?; // From here on every early return must still destroy the VM, so the work is // one call whose result is held while teardown runs unconditionally. @@ -542,6 +550,10 @@ pub struct VmPhase<'a> { pub task: &'a str, /// `missions.backend` — which rootfs image. `None` boots the node's default. pub backend: Option<&'a str>, + /// The server's LLM proxy as the node should reach it, when this phase's + /// model calls are relayed (`llm_proxy::microvm_relay`). `None` keeps the + /// provider key in the guest. + pub model_relay: Option, /// The host checkout, injected as a tar and collected back over the same /// path so `mission_delivery` needs no change. pub repo: &'a std::path::Path, diff --git a/crates/cm-api/src/microvm_turn_executor.rs b/crates/cm-api/src/microvm_turn_executor.rs index 155dae5..1aa3f30 100644 --- a/crates/cm-api/src/microvm_turn_executor.rs +++ b/crates/cm-api/src/microvm_turn_executor.rs @@ -191,6 +191,12 @@ impl TurnExecutor for MicroVmTurnExecutor { let outcome = self .vms .run(VmPhase { + model_relay: crate::llm_proxy::microvm_relay( + &self.pool, + fleet_node.as_uuid(), + backend.as_deref(), + ) + .await, task_policy: None, // Every node of a composed graph streams to the same outer run, // which is the one the operator is watching. diff --git a/crates/cm-api/src/mission_runtime.rs b/crates/cm-api/src/mission_runtime.rs index 0312a41..044f6a6 100644 --- a/crates/cm-api/src/mission_runtime.rs +++ b/crates/cm-api/src/mission_runtime.rs @@ -282,6 +282,24 @@ fn microvm_provider_env_from( Ok(env) } +/// The guest env for a microVM whose model calls are RELAYED to the server's +/// LLM proxy: the credential variable the backend's CLI reads carries the +/// mission's proxy token, and `ANTHROPIC_BASE_URL` points at the guest's own +/// loopback model port, which fcagent pipes to the node and the node relays. +/// The per-turn env overrides the base URL the image bakes in. +pub fn microvm_proxied_env( + backend: Option<&str>, + token: &str, +) -> Result, String> { + let want = microvm_credential_for(backend)?; + let route = crate::llm_proxy::microvm_route(backend) + .ok_or_else(|| format!("backend {backend:?} has no LLM proxy route"))?; + Ok(vec![ + (want.target.to_string(), token.to_string()), + ("ANTHROPIC_BASE_URL".to_string(), format!("http://127.0.0.1:11434/{route}")), + ]) +} + /// The testable half of [`forwarded_provider_env`]. The lookup is a parameter /// because a test cannot set process environment variables here — the workspace /// denies `unsafe`, and `set_var` is racy across test threads regardless. diff --git a/crates/cm-api/src/phase_runner.rs b/crates/cm-api/src/phase_runner.rs index d3426bd..cd41b48 100644 --- a/crates/cm-api/src/phase_runner.rs +++ b/crates/cm-api/src/phase_runner.rs @@ -1817,9 +1817,12 @@ async fn launch_microvm_phase( std::fs::create_dir_all(&repo) .map_err(|e| format!("create empty workspace {}: {e}", repo.display()))?; } + let model_relay = + crate::llm_proxy::microvm_relay(&pool2, node, backend.as_deref()).await; crate::microvm_executor::run_phase_in_vm( &hub, crate::microvm_executor::VmPhase { + model_relay, task_policy: Some(&task_policy), // Attribution for live output: this is the run a browser // subscribes to for this phase.