fix(mission_runtime): per-mission auto-pair via container log scrape (C3 auth)
ci / gates (push) Successful in 10s
ci / rust (push) Failing after 23s
ci / frontend (push) Successful in 38s
ci / e2e (push) Skipped
ci / publish (push) Skipped

The seed-mount approach didnt work: even with the shared runtimes
data dir bind-mounted, a fresh gateway instance mints a new pairing
key and requires re-pairing. The topology_worker connect returned
401 forever.

New approach — per-mission gateways self-pair:
- Provisioner tails container logs after start, extracts the
  X-Pairing-Code from the boot banner
- Persists it on missions.runtime_pairing_code (migration 0059)
- topology_worker constructs ZeroClawDriveExecutor with THAT code
  via from_env_for_gateway_with_code, which triggers the lazy
  /pair handshake on first turn and caches the returned bearer

Drops the shared-runtime data-dir mount — each per-mission gateway
now owns its own state, restoring the C3 isolation guarantee.
This commit is contained in:
Omar Sobh
2026-07-22 13:06:25 -07:00
parent 0210f5bf51
commit b569688e04
7 changed files with 173 additions and 44 deletions
+6 -3
View File
@@ -72,14 +72,15 @@ pub async fn on_launch(
// shared runtime endpoint in that case. // shared runtime endpoint in that case.
if let Some(prov) = crate::mission_runtime::MissionRuntimeProvisioner::from_env() { if let Some(prov) = crate::mission_runtime::MissionRuntimeProvisioner::from_env() {
match prov.ensure_container(mission_id).await { match prov.ensure_container(mission_id).await {
Ok(endpoint) => { Ok(ec) => {
let container_name = crate::mission_runtime::container_name(mission_id); let container_name = crate::mission_runtime::container_name(mission_id);
if let Err(e) = cm_db::repo::missions::set_runtime_binding( if let Err(e) = cm_db::repo::missions::set_runtime_binding(
pool, pool,
mission_id, mission_id,
workspace_id.as_uuid(), workspace_id.as_uuid(),
Some(&container_name), Some(&container_name),
Some(&endpoint), Some(&ec.endpoint),
ec.pairing_code.as_deref(),
) )
.await .await
{ {
@@ -88,7 +89,9 @@ pub async fn on_launch(
); );
} else { } else {
eprintln!( eprintln!(
"mission_orchestrator: runtime container {container_name} → {endpoint} for mission {mission_id}" "mission_orchestrator: runtime container {container_name} → {} (paired={}) for mission {mission_id}",
ec.endpoint,
ec.pairing_code.is_some()
); );
} }
} }
+99 -30
View File
@@ -28,8 +28,10 @@ use bollard::models::{
ContainerCreateBody, EndpointSettings, HostConfig, Mount, MountTypeEnum, NetworkConnectRequest, ContainerCreateBody, EndpointSettings, HostConfig, Mount, MountTypeEnum, NetworkConnectRequest,
}; };
use bollard::query_parameters::{ use bollard::query_parameters::{
CreateContainerOptions, InspectContainerOptions, RemoveContainerOptions, StartContainerOptions, CreateContainerOptions, InspectContainerOptions, LogsOptions, RemoveContainerOptions,
StartContainerOptions,
}; };
use futures::StreamExt;
use bollard::Docker; use bollard::Docker;
use std::collections::HashMap; use std::collections::HashMap;
use uuid::Uuid; use uuid::Uuid;
@@ -53,19 +55,6 @@ const EDGE_NETWORK: &str = "clawmates_edge";
/// by `clawmates-runtime.service`. /// by `clawmates-runtime.service`.
const MISSIONS_HOST_ROOT: &str = "/var/lib/clawmates-missions"; const MISSIONS_HOST_ROOT: &str = "/var/lib/clawmates-missions";
/// Host path holding the pre-paired ZeroClaw state (config.toml,
/// pairing key, brains, agent definitions) that the shared
/// `clawmates-runtime.service` uses. Per-mission runtime containers
/// mount this same dir so their gateway accepts the server's already-
/// provisioned ZEROCLAW_TOKEN instead of demanding a fresh /pair
/// handshake. Overridable via `CLAWMATES_RUNTIME_SEED_DIR` for
/// dev / test.
///
/// Known caveat: the sqlite sessions dir under here is currently
/// shared across all concurrent mission runtimes. In practice missions
/// don't collide often (topology_worker runs them one at a time per
/// mission) but this is an obvious next-iteration split point.
const DEFAULT_SEED_DIR: &str = "/root/clawmates-runtime/data";
/// Deterministic docker container name for a mission's runtime. /// Deterministic docker container name for a mission's runtime.
/// Uses the full UUID hex — UUIDv7 encodes time in the leading bytes, /// Uses the full UUID hex — UUIDv7 encodes time in the leading bytes,
@@ -88,6 +77,18 @@ pub struct MissionRuntimeProvisioner {
image: String, image: String,
} }
/// What `ensure_container` returns: everything the caller needs to
/// point a topology_worker at this mission's fresh gateway.
#[derive(Debug, Clone)]
pub struct EnsuredContainer {
pub endpoint: String,
/// One-time pairing code minted by the daemon at boot; may be
/// None on the reuse-existing path when we couldn't scrape it
/// back (log rotation). Callers keep the previously-persisted
/// value in that case.
pub pairing_code: Option<String>,
}
impl MissionRuntimeProvisioner { impl MissionRuntimeProvisioner {
/// Connect to the docker engine. Honors `DOCKER_HOST` (set in /// Connect to the docker engine. Honors `DOCKER_HOST` (set in
/// compose to the socket proxy) and falls back to the local /// compose to the socket proxy) and falls back to the local
@@ -108,7 +109,7 @@ impl MissionRuntimeProvisioner {
/// Idempotent: returns the endpoint URL, creating the container /// Idempotent: returns the endpoint URL, creating the container
/// on first call. If the container exists but is stopped, starts /// on first call. If the container exists but is stopped, starts
/// it. If it exists and is running, returns its endpoint. /// it. If it exists and is running, returns its endpoint.
pub async fn ensure_container(&self, mission_id: Uuid) -> Result<String, String> { pub async fn ensure_container(&self, mission_id: Uuid) -> Result<EnsuredContainer, String> {
let name = container_name(mission_id); let name = container_name(mission_id);
// Fast path: already running. // Fast path: already running.
if let Ok(inspect) = self if let Ok(inspect) = self
@@ -122,7 +123,14 @@ impl MissionRuntimeProvisioner {
.and_then(|s| s.running) .and_then(|s| s.running)
.unwrap_or(false); .unwrap_or(false);
if running { if running {
return Ok(endpoint_url(&name)); // Reuse; try to re-scrape the pairing code from logs,
// but it may have rotated out — caller falls back to
// the previously-persisted value in that case.
let pairing_code = self.scrape_pairing_code(&name).await;
return Ok(EnsuredContainer {
endpoint: endpoint_url(&name),
pairing_code,
});
} }
// Exists but not running — remove + recreate below rather // Exists but not running — remove + recreate below rather
// than trying to restart a dirty-state container. // than trying to restart a dirty-state container.
@@ -143,11 +151,13 @@ impl MissionRuntimeProvisioner {
// present or docker start fails with EACCES/ENOENT. // present or docker start fails with EACCES/ENOENT.
let mission_dir = format!("{MISSIONS_HOST_ROOT}/{mission_id}"); let mission_dir = format!("{MISSIONS_HOST_ROOT}/{mission_id}");
let _ = tokio::fs::create_dir_all(&mission_dir).await; let _ = tokio::fs::create_dir_all(&mission_dir).await;
let seed_dir = std::env::var("CLAWMATES_RUNTIME_SEED_DIR")
.unwrap_or_else(|_| DEFAULT_SEED_DIR.to_string());
let mounts = vec![ let mounts = vec![
// Mount just this mission's directory. Agents can navigate // Mount just this mission's directory. Agents can navigate
// its `/repo` subdir but never see other missions'. // its `/repo` subdir but never see other missions'.
// `/zeroclaw-data` is intentionally NOT bind-mounted — each
// per-mission gateway boots with a fresh state directory
// and mints its own pairing code, scraped in
// `wait_for_pairing_code` below.
Mount { Mount {
target: Some("/mission".to_string()), target: Some("/mission".to_string()),
source: Some(mission_dir.clone()), source: Some(mission_dir.clone()),
@@ -155,16 +165,6 @@ impl MissionRuntimeProvisioner {
read_only: Some(false), read_only: Some(false),
..Default::default() ..Default::default()
}, },
// Seed /zeroclaw-data from the shared runtime's paired
// data dir so the fresh gateway accepts our ZEROCLAW_TOKEN
// without a /pair handshake.
Mount {
target: Some("/zeroclaw-data".to_string()),
source: Some(seed_dir),
typ: Some(MountTypeEnum::BIND),
read_only: Some(false),
..Default::default()
},
]; ];
let host_config = HostConfig { let host_config = HostConfig {
@@ -238,9 +238,77 @@ impl MissionRuntimeProvisioner {
.await .await
.map_err(|e| format!("start mission runtime container: {e}"))?; .map_err(|e| format!("start mission runtime container: {e}"))?;
Ok(endpoint_url(&name)) // Give the daemon a moment to print its boot banner, then
// scrape the pairing code. Poll with a short deadline so a
// slow boot doesn't hang the launch — the topology_worker
// will retry pairing later if we came up empty.
let pairing_code = self.wait_for_pairing_code(&name).await;
Ok(EnsuredContainer {
endpoint: endpoint_url(&name),
pairing_code,
})
} }
/// Poll container logs for up to ~10s waiting for the ZeroClaw
/// pairing banner. Returns None on timeout — callers keep any
/// previously-persisted code, or the topology run fails and the
/// operator sees a clear error surface.
async fn wait_for_pairing_code(&self, name: &str) -> Option<String> {
let deadline = std::time::Duration::from_secs(10);
let start = std::time::Instant::now();
while start.elapsed() < deadline {
if let Some(code) = self.scrape_pairing_code(name).await {
return Some(code);
}
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
}
None
}
/// One-shot scrape of `docker logs <name>` for `X-Pairing-Code: NNNNNN`.
async fn scrape_pairing_code(&self, name: &str) -> Option<String> {
let mut stream = self.docker.logs(
name,
Some(LogsOptions {
stdout: true,
stderr: true,
tail: "200".to_string(),
..Default::default()
}),
);
let mut buf = String::new();
while let Some(chunk) = stream.next().await {
if let Ok(c) = chunk {
buf.push_str(&String::from_utf8_lossy(&c.into_bytes()));
if buf.len() > 64_000 {
break;
}
}
}
extract_pairing_code(&buf)
}
}
/// Parse a ZeroClaw daemon boot banner and pull out the pairing code
/// from a line like `Send: POST /pair with header X-Pairing-Code: 273200`.
fn extract_pairing_code(logs: &str) -> Option<String> {
for line in logs.lines() {
if let Some(idx) = line.find("X-Pairing-Code:") {
let rest = &line[idx + "X-Pairing-Code:".len()..];
let code: String = rest
.chars()
.skip_while(|c| c.is_whitespace())
.take_while(|c| c.is_ascii_digit())
.collect();
if !code.is_empty() {
return Some(code);
}
}
}
None
}
impl MissionRuntimeProvisioner {
/// Force-remove the mission's runtime container. Idempotent. /// Force-remove the mission's runtime container. Idempotent.
pub async fn teardown_container(&self, mission_id: Uuid) -> Result<(), String> { pub async fn teardown_container(&self, mission_id: Uuid) -> Result<(), String> {
let name = container_name(mission_id); let name = container_name(mission_id);
@@ -313,7 +381,8 @@ async fn sweep_once(pool: &sqlx::PgPool, grace: std::time::Duration) -> Result<(
eprintln!("mission_runtime::sweeper: teardown mission {id}: {e}"); eprintln!("mission_runtime::sweeper: teardown mission {id}: {e}");
} }
if let Err(e) = if let Err(e) =
cm_db::repo::missions::set_runtime_binding(pool, id, workspace_id, None, None).await cm_db::repo::missions::set_runtime_binding(pool, id, workspace_id, None, None, None)
.await
{ {
eprintln!("mission_runtime::sweeper: clear binding for {id}: {e}"); eprintln!("mission_runtime::sweeper: clear binding for {id}: {e}");
} }
+8 -3
View File
@@ -174,20 +174,25 @@ async fn launch_phase(
if m.runtime_endpoint.is_none() { if m.runtime_endpoint.is_none() {
if let Some(prov) = crate::mission_runtime::MissionRuntimeProvisioner::from_env() { if let Some(prov) = crate::mission_runtime::MissionRuntimeProvisioner::from_env() {
match prov.ensure_container(mission_id).await { match prov.ensure_container(mission_id).await {
Ok(endpoint) => { Ok(ec) => {
let name = crate::mission_runtime::container_name(mission_id); let name = crate::mission_runtime::container_name(mission_id);
if let Err(e) = cm_db::repo::missions::set_runtime_binding( if let Err(e) = cm_db::repo::missions::set_runtime_binding(
pool, pool,
mission_id, mission_id,
workspace_id, workspace_id,
Some(&name), Some(&name),
Some(&endpoint), Some(&ec.endpoint),
ec.pairing_code.as_deref(),
) )
.await .await
{ {
eprintln!("phase_runner: bind runtime container for {mission_id} failed: {e}"); eprintln!("phase_runner: bind runtime container for {mission_id} failed: {e}");
} else { } else {
eprintln!("phase_runner: runtime container {name} → {endpoint} for mission {mission_id}"); eprintln!(
"phase_runner: runtime container {name} → {} (paired={}) for mission {mission_id}",
ec.endpoint,
ec.pairing_code.is_some()
);
} }
} }
Err(e) => eprintln!("phase_runner: provision runtime container for {mission_id} failed (continuing): {e}"), Err(e) => eprintln!("phase_runner: provision runtime container for {mission_id} failed (continuing): {e}"),
+26
View File
@@ -108,6 +108,32 @@ impl ZeroClawDriveExecutor {
Ok(exec) 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.
pub fn from_env_for_gateway_with_code(
gateway_url: String,
pairing_code: String,
) -> Result<Self, String> {
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 { fn alias_for(&self, role: &str) -> String {
self.role_aliases self.role_aliases
.get(role) .get(role)
+11 -6
View File
@@ -152,8 +152,11 @@ 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_endpoint: Option<String> = sqlx::query_scalar::<_, Option<String>>( let mission_binding: Option<(Option<String>, Option<String>)> = sqlx::query_as::<
"SELECT m.runtime_endpoint _,
(Option<String>, Option<String>),
>(
"SELECT m.runtime_endpoint, m.runtime_pairing_code
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",
@@ -162,11 +165,13 @@ async fn run_job(
.fetch_optional(pool) .fetch_optional(pool)
.await .await
.ok() .ok()
.flatten()
.flatten(); .flatten();
let leaf_result = match mission_endpoint { let leaf_result = match mission_binding {
Some(url) => ZeroClawDriveExecutor::from_env_for_gateway(url), Some((Some(url), Some(code))) => {
None => ZeroClawDriveExecutor::from_env(), ZeroClawDriveExecutor::from_env_for_gateway_with_code(url, code)
}
Some((Some(url), None)) => ZeroClawDriveExecutor::from_env_for_gateway(url),
_ => ZeroClawDriveExecutor::from_env(),
}; };
let leaf = match leaf_result { let leaf = match leaf_result {
Ok(e) => e, Ok(e) => e,
+10 -2
View File
@@ -39,6 +39,10 @@ pub struct Mission {
pub runtime_container_name: Option<String>, pub runtime_container_name: Option<String>,
/// Gateway URL the topology_worker dials for this mission's runs. /// Gateway URL the topology_worker dials for this mission's runs.
pub runtime_endpoint: Option<String>, pub runtime_endpoint: Option<String>,
/// One-time pairing code captured from the fresh gateway's startup
/// log. topology_worker uses it to lazy-pair with this specific
/// mission runtime instead of the shared-runtime env token.
pub runtime_pairing_code: Option<String>,
#[serde(with = "time::serde::rfc3339")] #[serde(with = "time::serde::rfc3339")]
pub created_at: OffsetDateTime, pub created_at: OffsetDateTime,
#[serde(with = "time::serde::rfc3339")] #[serde(with = "time::serde::rfc3339")]
@@ -186,7 +190,7 @@ pub async fn get(pool: &PgPool, id: Uuid, workspace_id: Uuid) -> Result<Option<M
"SELECT id, workspace_id, title, template_kind, team_id, "SELECT id, workspace_id, title, template_kind, team_id,
team_template_id, repo_id, schedule, status, team_template_id, repo_id, schedule, status,
description, config, runtime_kind, target_node_id, description, config, runtime_kind, target_node_id,
runtime_container_name, runtime_endpoint, runtime_container_name, runtime_endpoint, runtime_pairing_code,
created_at, updated_at, completed_at created_at, updated_at, completed_at
FROM missions WHERE id = $1 AND workspace_id = $2", FROM missions WHERE id = $1 AND workspace_id = $2",
) )
@@ -227,7 +231,7 @@ pub async fn list_by_workspace(
"SELECT id, workspace_id, title, template_kind, team_id, "SELECT id, workspace_id, title, template_kind, team_id,
team_template_id, repo_id, schedule, status, team_template_id, repo_id, schedule, status,
description, config, runtime_kind, target_node_id, description, config, runtime_kind, target_node_id,
runtime_container_name, runtime_endpoint, runtime_container_name, runtime_endpoint, runtime_pairing_code,
created_at, updated_at, completed_at created_at, updated_at, completed_at
FROM missions WHERE workspace_id = $1 FROM missions WHERE workspace_id = $1
ORDER BY created_at DESC LIMIT $2", ORDER BY created_at DESC LIMIT $2",
@@ -254,6 +258,7 @@ pub async fn list_by_workspace(
target_node_id: r.get("target_node_id"), target_node_id: r.get("target_node_id"),
runtime_container_name: r.get("runtime_container_name"), runtime_container_name: r.get("runtime_container_name"),
runtime_endpoint: r.get("runtime_endpoint"), runtime_endpoint: r.get("runtime_endpoint"),
runtime_pairing_code: r.get("runtime_pairing_code"),
created_at: r.get("created_at"), created_at: r.get("created_at"),
updated_at: r.get("updated_at"), updated_at: r.get("updated_at"),
completed_at: r.get("completed_at"), completed_at: r.get("completed_at"),
@@ -315,11 +320,13 @@ pub async fn set_runtime_binding(
workspace_id: Uuid, workspace_id: Uuid,
container_name: Option<&str>, container_name: Option<&str>,
endpoint: Option<&str>, endpoint: Option<&str>,
pairing_code: Option<&str>,
) -> Result<(), DbError> { ) -> Result<(), DbError> {
sqlx::query( sqlx::query(
"UPDATE missions "UPDATE missions
SET runtime_container_name = $3, SET runtime_container_name = $3,
runtime_endpoint = $4, runtime_endpoint = $4,
runtime_pairing_code = $5,
updated_at = now() updated_at = now()
WHERE id = $1 AND workspace_id = $2", WHERE id = $1 AND workspace_id = $2",
) )
@@ -327,6 +334,7 @@ pub async fn set_runtime_binding(
.bind(workspace_id) .bind(workspace_id)
.bind(container_name) .bind(container_name)
.bind(endpoint) .bind(endpoint)
.bind(pairing_code)
.execute(pool) .execute(pool)
.await?; .await?;
Ok(()) Ok(())
@@ -0,0 +1,13 @@
-- Per-mission ZeroClaw pairing code.
--
-- Fresh mission runtime containers mint a new one-time pairing code
-- at startup. mission_runtime::ensure_container captures it from the
-- container's stdout and persists it here so the topology_worker's
-- ZeroClawDriveExecutor can lazy-pair with THIS gateway (instead of
-- reusing the shared runtime's env-derived ZEROCLAW_TOKEN, which the
-- fresh gateway doesn't accept).
--
-- Nullable: pre-C3-pair missions + non-mission topology runs fall
-- back to the shared-runtime env auth path.
ALTER TABLE missions ADD COLUMN runtime_pairing_code TEXT;