Omar Sobh
a2bc06d313
- mission_orchestrator: replace hardcoded cli=\"claude\" with real precedence chain: mission.config.cli → template.config.default_cli → \"claude\". Missions can now A/B by CLI without a schema change. - fleet.rs: scope clippy::too_many_arguments allow on NodeHub::open_pty — 8 params (session address x4 + target address x3) is the actual dimensionality; a struct would be ceremony. - security_scan.rs: delete future_artifact_root — pure hint-stub, no callers, nothing depends on it. - routes/world.rs: delete the empty normalize() seam + its 30-line comment block. normalize_run_event higher up does the real work; the empty stub was pre-cleanup scaffolding. Post-sweep validation across the whole workspace: * cargo check --workspace → clean * cargo test --workspace --no-run → all bins build * cargo test -p cm-api --test mission_orchestrator → 3/3 pass * cargo clippy -p cm-api -p clawmates-node --tests → 0 warnings * tsc --noEmit → 0 errors Wiring audit: every recent route handler is registered in lib.rs. Every recent frontend component has at least one importer. Remaining #[allow(dead_code)] entries are deprovision_claw + unpublish_claw on RuntimeProvisioner — real rollback paths waiting on the team-delete route. Documented future hooks, not stubs.
322 lines
12 KiB
Rust
322 lines
12 KiB
Rust
//! Mission-launch orchestrator. Slice 4.
|
||
//!
|
||
//! When a mission's status flips from `draft` to `running`, this
|
||
//! module fires:
|
||
//! 1. Materialize a team from `team_template_id` (if set + no
|
||
//! `team_id` yet) — inserts agents, ingests brain seeds,
|
||
//! records agent_template_link lineage, wires team_members.
|
||
//! 2. Bind the resulting `team_id` back onto the mission.
|
||
//!
|
||
//! Phase execution (running research/coding/benchmark/security_scan
|
||
//! against the world) is layered on top by Slices 5–8.
|
||
//!
|
||
//! Design notes:
|
||
//! - Runtime provisioning is opt-in via `RuntimeProvisioner::from_env`.
|
||
//! Missing runtime = "insert DB rows only, no live claw" — the
|
||
//! mission still boots; live claws land the moment the runtime
|
||
//! env is configured + the mission re-launches.
|
||
//! - Every step is best-effort logged so a partial materialization
|
||
//! can be picked up by a subsequent launch rather than blocking.
|
||
|
||
use cm_db::repo::team_templates::TeamTemplateDetail;
|
||
use cm_domain::{AccessPolicy, Agent, AgentStatus, WorkspaceId};
|
||
use sqlx::PgPool;
|
||
use uuid::Uuid;
|
||
|
||
use crate::runtime_provision::RuntimeProvisioner;
|
||
|
||
/// Called from `routes::missions::set_status` when the transition is
|
||
/// draft→running. Materializes the team + returns the new team_id
|
||
/// (or an existing one — no-op when the mission already has a team).
|
||
///
|
||
/// Non-fatal on failure: logs + returns Ok(None) so the mission still
|
||
/// becomes `running` and the user can inspect the error via server
|
||
/// logs + re-attempt. Later slices surface this on the canvas.
|
||
pub async fn on_launch(
|
||
pool: &PgPool,
|
||
workspace_id: WorkspaceId,
|
||
user_id: cm_domain::UserId,
|
||
mission_id: Uuid,
|
||
node_hub: Option<std::sync::Arc<crate::fleet::NodeHub>>,
|
||
) -> Result<Option<Uuid>, String> {
|
||
let Some(mission) = cm_db::repo::missions::get(pool, mission_id, workspace_id.as_uuid())
|
||
.await
|
||
.map_err(|e| format!("load mission: {e}"))?
|
||
else {
|
||
return Err("mission not found".into());
|
||
};
|
||
|
||
// Skip if already bound.
|
||
if mission.team_id.is_some() {
|
||
return Ok(mission.team_id);
|
||
}
|
||
let Some(template_id) = mission.team_template_id else {
|
||
// No template + no team = phase execution will auto-provision
|
||
// via the LLM path (Slice 2's fallback), or run against the
|
||
// shared runtime. Nothing to do here.
|
||
return Ok(None);
|
||
};
|
||
|
||
let template = cm_db::repo::team_templates::get(pool, template_id)
|
||
.await
|
||
.map_err(|e| format!("load template: {e}"))?
|
||
.ok_or_else(|| format!("template {template_id} not found"))?;
|
||
|
||
let provisioner = RuntimeProvisioner::from_env();
|
||
|
||
let team_id = mint_team_from_template(
|
||
pool,
|
||
workspace_id,
|
||
user_id,
|
||
provisioner.as_ref(),
|
||
&template,
|
||
&mission.title,
|
||
"claude-sonnet-5",
|
||
)
|
||
.await?;
|
||
|
||
// Bind the team onto the mission.
|
||
sqlx::query("UPDATE missions SET team_id = $1, updated_at = now() WHERE id = $2")
|
||
.bind(team_id)
|
||
.bind(mission_id)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| format!("bind team on mission: {e}"))?;
|
||
|
||
// Ensure the mission's repo is checked out at
|
||
// $CLAWMATES_MISSIONS_ROOT/{mission_id}/repo — where
|
||
// security_scan + benchmark_runner exec against. Non-fatal:
|
||
// missions without a repo (research_only, custom) skip cleanly,
|
||
// and clone failures log without blocking launch (the operator
|
||
// sees the error on the canvas via the run's failed status when
|
||
// a repo-dependent phase tries to fire).
|
||
match crate::mission_workspace::ensure_checkout(pool, workspace_id, mission_id).await {
|
||
Ok(Some(path)) => eprintln!(
|
||
"mission_orchestrator: repo checked out at {}",
|
||
path.display()
|
||
),
|
||
Ok(None) => {}
|
||
Err(e) => eprintln!(
|
||
"mission_orchestrator: repo checkout for {mission_id} failed (continuing): {e}"
|
||
),
|
||
}
|
||
|
||
// Herdr second-runtime: if runtime_kind='local_herdr', spawn a
|
||
// pane on target_node running the first available local CLI.
|
||
// Non-fatal on failure — the operator sees the error in server
|
||
// logs and can manually retry via POST /herdr-dispatch.
|
||
if mission.runtime_kind == "local_herdr" {
|
||
if let (Some(hub), Some(node_id)) = (node_hub, mission.target_node_id) {
|
||
let prompt = mission.description.clone().unwrap_or_default();
|
||
// CLI selection precedence:
|
||
// mission.config.cli → template.config.default_cli → "claude"
|
||
// Templates encode which agent CLI fits their stack; missions can
|
||
// override per-run for A/B (kimi on morpheus vs claude on tank).
|
||
let cli = mission
|
||
.config
|
||
.get("cli")
|
||
.and_then(|v| v.as_str())
|
||
.map(str::to_string)
|
||
.or_else(|| {
|
||
template
|
||
.template
|
||
.config
|
||
.get("default_cli")
|
||
.and_then(|v| v.as_str())
|
||
.map(str::to_string)
|
||
})
|
||
.unwrap_or_else(|| "claude".to_string());
|
||
match crate::fleet_herdr::dispatch(
|
||
hub,
|
||
cm_domain::NodeId::from(node_id),
|
||
mission_id,
|
||
&cli,
|
||
&prompt,
|
||
)
|
||
.await
|
||
{
|
||
Ok(handle) => eprintln!(
|
||
"mission_orchestrator: herdr pane {} spawned on node {}",
|
||
handle.pane_id, node_id
|
||
),
|
||
Err(e) => eprintln!(
|
||
"mission_orchestrator: herdr dispatch for {mission_id} failed (continuing): {e}"
|
||
),
|
||
}
|
||
} else {
|
||
eprintln!(
|
||
"mission_orchestrator: mission {mission_id} is local_herdr but node_hub or target_node missing"
|
||
);
|
||
}
|
||
}
|
||
|
||
Ok(Some(team_id))
|
||
}
|
||
|
||
async fn mint_team_from_template(
|
||
pool: &PgPool,
|
||
workspace_id: WorkspaceId,
|
||
user_id: cm_domain::UserId,
|
||
provisioner: Option<&RuntimeProvisioner>,
|
||
template: &TeamTemplateDetail,
|
||
team_name: &str,
|
||
default_model: &str,
|
||
) -> Result<Uuid, String> {
|
||
// Build the topology graph from role slots so the team's `graph`
|
||
// NOT NULL column is satisfied + downstream topology executors
|
||
// have a valid shape to iterate over.
|
||
let roles: Vec<&str> = template.roles.iter().map(|r| r.slot.as_str()).collect();
|
||
let topology_kind = parse_topology_kind(&template.template.default_topology);
|
||
let graph = cm_topology::build(topology_kind, &roles)
|
||
.map_err(|e| format!("build topology graph: {e}"))?;
|
||
let graph_json =
|
||
serde_json::to_value(&graph).map_err(|e| format!("serialize topology graph: {e}"))?;
|
||
|
||
let team_id = Uuid::now_v7();
|
||
cm_db::repo::teams::insert_team_with_lifecycle(
|
||
pool,
|
||
team_id,
|
||
workspace_id,
|
||
team_name,
|
||
&template.template.default_topology,
|
||
&graph_json,
|
||
"permanent",
|
||
)
|
||
.await
|
||
.map_err(|e| format!("insert team: {e}"))?;
|
||
|
||
// Stamp template lineage on the team row (Slice 3 columns).
|
||
sqlx::query(
|
||
"UPDATE teams SET template_id = $1, template_version = $2,
|
||
risk_profile = $3, mcp_bundles = $4
|
||
WHERE id = $5",
|
||
)
|
||
.bind(template.template.id)
|
||
.bind(template.template.version)
|
||
.bind(&template.template.risk_profile)
|
||
.bind(serde_json::json!(template.template.mcp_bundles))
|
||
.bind(team_id)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| format!("stamp template lineage: {e}"))?;
|
||
|
||
// For each role: create agent, provision runtime, ingest brain
|
||
// seed, record link, bind to topology node.
|
||
for (idx, role) in template.roles.iter().enumerate() {
|
||
// Node id in the graph follows cm_topology's `n0..` convention
|
||
// — read the actual node id from the graph so team_members
|
||
// stays consistent with the topology.
|
||
let Some(node) = graph.nodes.get(idx) else {
|
||
return Err(format!(
|
||
"topology graph produced {} nodes but template has {} roles",
|
||
graph.nodes.len(),
|
||
template.roles.len(),
|
||
));
|
||
};
|
||
let agent = Agent {
|
||
id: cm_domain::AgentId::new(),
|
||
workspace_id,
|
||
name: format!("{} · {}", team_name, role.slot),
|
||
job_title: role.slot.clone(),
|
||
system_prompt: role.system_prompt.clone(),
|
||
avatar: String::new(),
|
||
accent: default_accent_for(&role.slot).to_string(),
|
||
wallpaper: String::new(),
|
||
managed_by: user_id,
|
||
status: AgentStatus::Online,
|
||
};
|
||
cm_db::repo::agents::insert(pool, &agent, &AccessPolicy::default())
|
||
.await
|
||
.map_err(|e| format!("insert agent {}: {e}", role.slot))?;
|
||
let claw_id = agent.id.as_uuid();
|
||
|
||
cm_db::repo::agents::set_model_binding(pool, agent.id, default_model)
|
||
.await
|
||
.map_err(|e| format!("set_model_binding {claw_id}: {e}"))?;
|
||
|
||
// Runtime provisioning is opt-in — no-op if unconfigured.
|
||
if let Some(p) = provisioner {
|
||
if let Err(e) = p.provision_claw(claw_id, default_model).await {
|
||
eprintln!(
|
||
"mission_orchestrator: provision claw {claw_id} failed (continuing): {e}"
|
||
);
|
||
}
|
||
}
|
||
|
||
// Ingest brain seed (Slice 3.5d). Non-fatal on failure —
|
||
// agent still works from system_prompt alone.
|
||
if let Some(seed) = role.brain_seed.as_deref().filter(|s| !s.trim().is_empty()) {
|
||
if let Err(e) =
|
||
crate::brain_seed::ingest(claw_id, seed.to_string(), role.system_prompt.clone())
|
||
.await
|
||
{
|
||
eprintln!(
|
||
"mission_orchestrator: brain_seed ingest for {claw_id} failed (continuing): {e}"
|
||
);
|
||
}
|
||
}
|
||
|
||
// Record lineage (Slice 3.5d) so the MCP skills server can
|
||
// merge template default skills with per-agent overrides.
|
||
cm_db::repo::agent_template_link::upsert(
|
||
pool,
|
||
claw_id,
|
||
template.template.id,
|
||
template.template.version,
|
||
&role.slot,
|
||
)
|
||
.await
|
||
.map_err(|e| format!("agent_template_link upsert {claw_id}: {e}"))?;
|
||
cm_db::repo::agent_template_link::mark_seeded(pool, claw_id)
|
||
.await
|
||
.map_err(|e| format!("mark_seeded {claw_id}: {e}"))?;
|
||
|
||
// Wire team_members using the topology node id.
|
||
cm_db::repo::teams::add_member(pool, team_id, &node.id, claw_id, &role.slot)
|
||
.await
|
||
.map_err(|e| format!("team_members add {claw_id}: {e}"))?;
|
||
|
||
// Audit for parity with individual claw creation path.
|
||
let _ = cm_db::repo::audit::append(
|
||
pool,
|
||
workspace_id,
|
||
cm_db::repo::audit::Actor::User(user_id),
|
||
"agent.created",
|
||
"agent",
|
||
&agent.id.to_string(),
|
||
serde_json::json!({
|
||
"name": agent.name,
|
||
"job_title": agent.job_title,
|
||
"source": "mission_orchestrator",
|
||
"template_id": template.template.id.to_string(),
|
||
"template_version": template.template.version,
|
||
"role_slot": role.slot,
|
||
}),
|
||
)
|
||
.await;
|
||
}
|
||
|
||
Ok(team_id)
|
||
}
|
||
|
||
fn parse_topology_kind(s: &str) -> cm_topology::TopologyKind {
|
||
use cm_topology::TopologyKind;
|
||
match s {
|
||
"pipeline" => TopologyKind::Pipeline,
|
||
"hierarchical" => TopologyKind::Hierarchical,
|
||
"star_moe" => TopologyKind::StarMoe,
|
||
_ => TopologyKind::HubSpoke,
|
||
}
|
||
}
|
||
|
||
fn default_accent_for(slot: &str) -> &'static str {
|
||
match slot {
|
||
"planner" | "arch_analyst" | "designer" | "scene_designer" | "api_designer" => "#7cd6e0",
|
||
"coder" | "kernel_author" | "shader_author" => "#5fd08a",
|
||
"tester" | "bench_engineer" | "perf_engineer" => "#ffb44a",
|
||
"reviewer" | "db_engineer" => "#c98af0",
|
||
"committer" | "spec_integrator" => "#ff8a7a",
|
||
_ => "#8a8a92",
|
||
}
|
||
}
|