use crate::blueprint::{
resolve_bound_agents, AgentDef, AgentKind, AgentProfile, Blueprint, BlueprintMetadata,
BoundAgent, BoundAgentResolveError, Runner,
};
use crate::core::ctx::Ctx;
use crate::core::engine::Engine;
use crate::core::projection_placement::{ProjectionPlacement, ProjectionPlacementError};
use crate::core::step_naming::{StepNaming, StepNamingError};
use crate::operator::{Operator, OperatorSpawner, WorkerBinding};
use crate::types::{CapToken, StepId};
use crate::worker::adapter::{InProcSpawner, SpawnError, SpawnerAdapter, WorkerFn};
use crate::worker::process_spawner::{ProcessSpawner, StreamMode};
use crate::worker::Worker;
use async_trait::async_trait;
use mlua_flow_ir::{Expr, Node as FlowNode, Path};
use mlua_swarm_schema::{VerdictChannel, VerdictContract};
use serde_json::Value;
use std::collections::HashMap;
use std::sync::Arc;
use thiserror::Error;
#[derive(Debug, Error)]
pub enum CompileError {
#[error("bound agent resolution: {0}")]
BoundAgent(#[from] BoundAgentResolveError),
#[error("unknown agent kind in SpawnerRegistry: {0:?}")]
UnknownKind(AgentKind),
#[error("agent '{name}' spec invalid: {msg}")]
InvalidSpec {
name: String,
msg: String,
},
#[error("flow references agent '{0}' but no AgentDef matches")]
UnresolvedRef(String),
#[error("duplicate AgentDef name: {0}")]
DuplicateAgent(String),
#[error("agent '{agent}' operator_ref '{op_ref}' does not match any OperatorDef.name in Blueprint.operators (defined: {defined:?})")]
UnresolvedOperatorRef {
agent: String,
op_ref: String,
defined: Vec<String>,
},
#[error("{where_} names an undefined MetaDef: '{meta_ref}' (defined: {defined:?})")]
UnresolvedMetaRef {
where_: String,
meta_ref: String,
defined: Vec<String>,
},
#[error("StepNaming collision: {0}")]
StepNamingCollision(#[from] StepNamingError),
#[error("invalid projection_placement: {0}")]
InvalidProjectionPlacement(#[from] ProjectionPlacementError),
#[error("audits[].agent '{agent}' does not match any AgentDef.name in Blueprint.agents (defined: {defined:?})")]
UnresolvedAuditAgent {
agent: String,
defined: Vec<String>,
},
#[error(
"agent '{agent}' declares verdict channel '{expected_channel}' but {where_} \
addresses it as '{actual_shape}' output — see the \"Returning verdicts to drive \
BP flow\" guide's Pattern A (channel: \"body\") / Pattern B (channel: \"part\")"
)]
VerdictChannelMismatch {
where_: String,
agent: String,
expected_channel: String,
actual_shape: String,
},
#[error(
"agent '{agent}' verdict Lit '{value}' at {where_} is not a member of the declared \
values {values:?}"
)]
VerdictValueNotInContract {
where_: String,
agent: String,
value: String,
values: Vec<String>,
},
#[error(
"agent '{agent}' declares verdict value '{value}' but no downstream Branch/Loop \
cond references it (declared: {declared_values:?}, at step '{step_ref}') — either \
handle the value downstream or drop it from `verdict.values`"
)]
VerdictValueUnhandled {
agent: String,
value: String,
declared_values: Vec<String>,
step_ref: String,
},
}
pub const WORKER_BINDING_REQUIRED_MSG_PREFIX: &str =
"profile.worker_binding is required for this operator backend";
impl From<&CompileError> for mlua_swarm_diag::Diagnostic {
fn from(err: &CompileError) -> Self {
use mlua_swarm_diag::{
Applicability, DiagElement, DiagLevel, DiagSpan, DiagStage, Diagnostic, DocsRef,
Suggestion,
};
let base = |kind: &'static str| {
Diagnostic::new(
kind,
DiagStage::CompileLint,
DiagLevel::Error,
err.to_string(),
)
};
let agent_span = |name: &str| DiagSpan {
element: DiagElement::Agent {
name: name.to_string(),
},
json_path: Some(format!("$.agents[?(@.name=='{name}')]")),
};
match err {
CompileError::BoundAgent(_) => base("bound-agent-resolution"),
CompileError::UnknownKind(_) => base("unknown-agent-kind").with_help(
"register a SpawnerFactory for this kind, or disable strategy.strict_kind",
),
CompileError::InvalidSpec { name, msg }
if msg.starts_with(WORKER_BINDING_REQUIRED_MSG_PREFIX) =>
{
Diagnostic::new(
"worker-binding-missing",
DiagStage::CompileLint,
DiagLevel::Error,
format!(
"operator agent '{name}' has no explicit Runner or legacy \
`profile.worker_binding`"
),
)
.with_note(msg.clone())
.with_suggestion(Suggestion {
msg: "add an explicit Runner (or legacy profile.worker_binding)".into(),
patch: "runner = { backend = \"ws_operator\", variant = \"claude\", \
tools = {} }"
.into(),
applicability: Applicability::HasPlaceholders,
})
.with_docs_ref(DocsRef {
uri: "mse://guides/bp-dsl-templates",
anchor: None,
})
.with_span(agent_span(name))
}
CompileError::InvalidSpec { name, .. } => {
base("invalid-agent-spec").with_span(agent_span(name))
}
CompileError::UnresolvedRef(ref_) => base("unresolved-agent-ref").with_span(DiagSpan {
element: DiagElement::Step { ref_: ref_.clone() },
json_path: None,
}),
CompileError::DuplicateAgent(name) => {
base("duplicate-agent-name").with_span(agent_span(name))
}
CompileError::UnresolvedOperatorRef { agent, defined, .. } => {
base("unresolved-operator-ref")
.with_note(format!("declared OperatorDef names: {defined:?}"))
.with_span(agent_span(agent))
}
CompileError::UnresolvedMetaRef { defined, .. } => base("unresolved-meta-ref")
.with_note(format!("declared MetaDef names: {defined:?}")),
CompileError::StepNamingCollision(_) => base("step-naming-collision"),
CompileError::InvalidProjectionPlacement(_) => base("invalid-projection-placement")
.with_span(DiagSpan {
element: DiagElement::BlueprintRoot,
json_path: Some("$.projection_placement".into()),
}),
CompileError::UnresolvedAuditAgent { defined, .. } => base("unresolved-audit-agent")
.with_note(format!("declared AgentDef names: {defined:?}"))
.with_span(DiagSpan {
element: DiagElement::BlueprintRoot,
json_path: Some("$.audits".into()),
}),
CompileError::VerdictChannelMismatch { agent, .. } => base("verdict-channel-mismatch")
.with_help(
"see the \"Returning verdicts to drive BP flow\" guide's Pattern A \
(channel: \"body\") / Pattern B (channel: \"part\")",
)
.with_docs_ref(DocsRef {
uri: "mse://guides/blueprint-authoring",
anchor: None,
})
.with_span(agent_span(agent)),
CompileError::VerdictValueNotInContract { agent, .. } => {
base("verdict-value-not-in-contract")
.with_suggestion(Suggestion {
msg: "align the cond literal with the agent's declared verdict \
contract"
.into(),
patch: "either add the cond's literal to `agents[N].verdict.values`, \
or change the cond to a value that is already declared"
.into(),
applicability: Applicability::MaybeIncorrect,
})
.with_docs_ref(DocsRef {
uri: "mse://guides/blueprint-authoring",
anchor: None,
})
.with_span(agent_span(agent))
}
CompileError::VerdictValueUnhandled {
agent,
declared_values,
..
} => base("verdict-value-unhandled")
.with_note(format!("declared verdict.values: {declared_values:?}"))
.with_help(
"either handle the value in a downstream Branch/Loop cond, or drop it \
from verdict.values",
)
.with_span(agent_span(agent)),
}
}
}
pub trait SpawnerFactory: Send + Sync {
fn build(
&self,
agent_def: &AgentDef,
hint: Option<&Value>,
) -> Result<Arc<dyn SpawnerAdapter>, CompileError>;
}
pub trait SpawnerFactoryKind: SpawnerFactory {
const KIND: AgentKind;
type Worker: crate::worker::Worker;
}
#[derive(Clone)]
pub struct SpawnerRegistry {
factories: HashMap<AgentKind, Arc<dyn SpawnerFactory>>,
}
impl SpawnerRegistry {
pub fn new() -> Self {
Self {
factories: HashMap::new(),
}
}
pub fn register<F: SpawnerFactoryKind + 'static>(&mut self, factory: Arc<F>) -> &mut Self {
let f: Arc<dyn SpawnerFactory> = factory;
self.factories.insert(F::KIND, f);
self
}
}
impl Default for SpawnerRegistry {
fn default() -> Self {
Self::new()
}
}
pub struct Compiler {
registry: SpawnerRegistry,
default_spawner: Option<Arc<dyn SpawnerAdapter>>,
}
pub struct CompiledBlueprint {
pub router: Arc<CompiledAgentTable>,
pub flow: FlowNode,
pub metadata: BlueprintMetadata,
pub step_naming: Arc<StepNaming>,
pub projection_placement: Arc<ProjectionPlacement>,
}
fn project_bound_agent_for_legacy_factories(bound: &BoundAgent) -> AgentDef {
let mut agent = bound.agent.clone();
match &bound.runner {
Some(Runner::WsOperator { variant, tools })
| Some(Runner::WsClaudeCode { variant, tools }) => {
let profile = agent.profile.get_or_insert_with(AgentProfile::default);
profile.worker_binding = Some(variant.clone());
profile.tools = tools.clone();
}
Some(Runner::AgentBlockInProcess { tools }) => {
let profile = agent.profile.get_or_insert_with(AgentProfile::default);
profile.worker_binding = None;
profile.tools = tools.clone();
}
Some(Runner::Subprocess { .. }) => {}
None => {}
}
let meta = agent.meta.get_or_insert_with(Default::default);
meta.context_policy = bound.context_policy.clone();
agent
}
pub(crate) fn materialize_bound_blueprint(
bp: &Blueprint,
bound_agents: &[BoundAgent],
) -> Blueprint {
let mut effective = bp.clone();
effective.agents = bound_agents
.iter()
.map(project_bound_agent_for_legacy_factories)
.collect();
effective.default_context_policy = None;
effective
}
impl Compiler {
pub fn new(registry: SpawnerRegistry) -> Self {
Self {
registry,
default_spawner: None,
}
}
pub fn with_default(mut self, sp: Arc<dyn SpawnerAdapter>) -> Self {
self.default_spawner = Some(sp);
self
}
pub fn compile(&self, bp: &Blueprint) -> Result<CompiledBlueprint, CompileError> {
let bound_agents = resolve_bound_agents(bp)?;
self.compile_bound(bp, &bound_agents)
}
pub fn compile_bound(
&self,
bp: &Blueprint,
bound_agents: &[BoundAgent],
) -> Result<CompiledBlueprint, CompileError> {
let effective = materialize_bound_blueprint(bp, bound_agents);
self.compile_resolved(&effective)
}
fn compile_resolved(&self, bp: &Blueprint) -> Result<CompiledBlueprint, CompileError> {
let mut routes: HashMap<String, Arc<dyn SpawnerAdapter>> = HashMap::new();
let mut seen: HashMap<String, ()> = HashMap::new();
let mut verdict_contracts: HashMap<String, VerdictContract> = HashMap::new();
let defined: Vec<String> = bp.operators.iter().map(|o| o.name.clone()).collect();
for ad in &bp.agents {
if !matches!(ad.kind, AgentKind::Operator) {
continue;
}
let op_ref = ad.spec.get("operator_ref").and_then(|v| v.as_str());
if let Some(op_ref) = op_ref {
if !defined.iter().any(|n| n == op_ref) {
return Err(CompileError::UnresolvedOperatorRef {
agent: ad.name.clone(),
op_ref: op_ref.to_string(),
defined: defined.clone(),
});
}
}
}
let metas_defined: Vec<String> = bp.metas.iter().map(|m| m.name.clone()).collect();
for ad in &bp.agents {
let meta_ref = ad.meta.as_ref().and_then(|m| m.meta_ref.as_ref());
if let Some(meta_ref) = meta_ref {
if !metas_defined.iter().any(|n| n == meta_ref) {
return Err(CompileError::UnresolvedMetaRef {
where_: format!("AgentMeta.meta_ref of agent '{}'", ad.name),
meta_ref: meta_ref.clone(),
defined: metas_defined.clone(),
});
}
}
}
let mut static_step_meta_refs: Vec<(String, String)> = Vec::new();
collect_step_meta_refs(&bp.flow, &mut static_step_meta_refs);
for (where_, meta_ref) in static_step_meta_refs {
if !metas_defined.iter().any(|n| n == &meta_ref) {
return Err(CompileError::UnresolvedMetaRef {
where_,
meta_ref,
defined: metas_defined.clone(),
});
}
}
let agents_defined: Vec<String> = bp.agents.iter().map(|a| a.name.clone()).collect();
for audit in &bp.audits {
if !agents_defined.iter().any(|n| n == &audit.agent) {
return Err(CompileError::UnresolvedAuditAgent {
agent: audit.agent.clone(),
defined: agents_defined.clone(),
});
}
}
for ad in &bp.agents {
if seen.contains_key(&ad.name) {
return Err(CompileError::DuplicateAgent(ad.name.clone()));
}
seen.insert(ad.name.clone(), ());
if let Some(contract) = &ad.verdict {
verdict_contracts.insert(ad.name.clone(), contract.clone());
}
let factory = match self.registry.factories.get(&ad.kind) {
Some(f) => f.clone(),
None => {
if bp.strategy.strict_kind {
return Err(CompileError::UnknownKind(ad.kind.clone()));
} else {
tracing::warn!(
agent = %ad.name,
kind = ?ad.kind,
"no spawner factory registered for agent kind; \
dropping agent from routing table (strict_kind=false)"
);
continue;
}
}
};
let hint = bp.hints.per_agent.get(&ad.name);
let subprocess_hint = if ad.kind == AgentKind::Subprocess {
resolve_subprocess_template_hint(bp, ad)?
} else {
None
};
let spawner = factory.build(ad, subprocess_hint.as_ref().or(hint))?;
routes.insert(ad.name.clone(), spawner);
}
let strict_verdict_handling = bp.metadata.strict_verdict_handling.unwrap_or(false);
verify_verdict_conds(&bp.flow, &verdict_contracts, strict_verdict_handling)?;
if bp.strategy.strict_refs {
verify_refs(&bp.flow, &routes, self.default_spawner.is_some())?;
}
let (step_naming, step_naming_warnings) = StepNaming::from_blueprint(bp)?;
for warning in &step_naming_warnings {
tracing::warn!(
name = %warning.name,
first_step_ref = %warning.first_step_ref,
second_step_ref = %warning.second_step_ref,
"StepNaming: undeclared steps' canonical/alias names collide; \
the step whose own ref matches the name keeps it (data-plane priority)"
);
}
let projection_placement =
ProjectionPlacement::from_spec(bp.projection_placement.as_ref())?;
let router = Arc::new(CompiledAgentTable {
routes,
default: self.default_spawner.clone(),
verdict_contracts,
});
Ok(CompiledBlueprint {
router,
flow: bp.flow.clone(),
metadata: bp.metadata.clone(),
step_naming: Arc::new(step_naming),
projection_placement: Arc::new(projection_placement),
})
}
}
fn verify_refs(
node: &FlowNode,
routes: &HashMap<String, Arc<dyn SpawnerAdapter>>,
has_default: bool,
) -> Result<(), CompileError> {
let mut refs: Vec<String> = Vec::new();
collect_refs(node, &mut refs);
for r in refs {
if !routes.contains_key(&r) && !has_default {
return Err(CompileError::UnresolvedRef(r));
}
}
Ok(())
}
fn collect_refs(node: &FlowNode, out: &mut Vec<String>) {
match node {
FlowNode::Step { ref_, .. } => out.push(ref_.clone()),
FlowNode::Seq { children } => {
for c in children {
collect_refs(c, out);
}
}
FlowNode::Branch { then_, else_, .. } => {
collect_refs(then_, out);
collect_refs(else_, out);
}
FlowNode::Fanout { body, .. } => collect_refs(body, out),
FlowNode::Loop { body, .. } => collect_refs(body, out),
FlowNode::Try { body, catch, .. } => {
collect_refs(body, out);
collect_refs(catch, out);
}
FlowNode::Assign { .. } => {} }
}
fn collect_step_meta_refs(node: &FlowNode, out: &mut Vec<(String, String)>) {
match node {
FlowNode::Step { ref_, in_, .. } => {
if let Expr::Lit { value } = in_ {
if let Some(meta_ref) = static_step_meta_ref(value) {
out.push((format!("Step '{ref_}' $step_meta.ref"), meta_ref));
}
}
}
FlowNode::Seq { children } => {
for c in children {
collect_step_meta_refs(c, out);
}
}
FlowNode::Branch { then_, else_, .. } => {
collect_step_meta_refs(then_, out);
collect_step_meta_refs(else_, out);
}
FlowNode::Fanout { body, .. } => collect_step_meta_refs(body, out),
FlowNode::Loop { body, .. } => collect_step_meta_refs(body, out),
FlowNode::Try { body, catch, .. } => {
collect_step_meta_refs(body, out);
collect_step_meta_refs(catch, out);
}
FlowNode::Assign { .. } => {} }
}
fn static_step_meta_ref(value: &Value) -> Option<String> {
value
.as_object()?
.get("$step_meta")?
.as_object()?
.get("ref")?
.as_str()
.map(str::to_string)
}
fn verify_verdict_conds(
flow: &FlowNode,
verdict_contracts: &HashMap<String, VerdictContract>,
strict_verdict_handling: bool,
) -> Result<(), CompileError> {
let mut step_outputs: HashMap<String, String> = HashMap::new();
let mut step_agents: HashMap<String, String> = HashMap::new();
collect_step_outputs_and_agents(flow, &mut step_outputs, &mut step_agents);
let mut errors: Vec<CompileError> = Vec::new();
let mut referenced_values: HashMap<String, std::collections::HashSet<String>> = HashMap::new();
collect_verdict_conds(
flow,
&step_outputs,
verdict_contracts,
&mut referenced_values,
&mut errors,
);
check_unhandled_verdict_values(
verdict_contracts,
&referenced_values,
&step_agents,
strict_verdict_handling,
&mut errors,
);
match errors.into_iter().next() {
Some(e) => Err(e),
None => Ok(()),
}
}
fn collect_step_outputs_and_agents(
node: &FlowNode,
out: &mut HashMap<String, String>,
step_agents: &mut HashMap<String, String>,
) {
match node {
FlowNode::Step {
ref_,
out: out_expr,
..
} => {
if let Expr::Path { at } = out_expr {
out.insert(at.to_string(), ref_.clone());
}
step_agents
.entry(ref_.clone())
.or_insert_with(|| ref_.clone());
}
FlowNode::Seq { children } => {
for c in children {
collect_step_outputs_and_agents(c, out, step_agents);
}
}
FlowNode::Branch { then_, else_, .. } => {
collect_step_outputs_and_agents(then_, out, step_agents);
collect_step_outputs_and_agents(else_, out, step_agents);
}
FlowNode::Fanout { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
FlowNode::Loop { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
FlowNode::Try { body, catch, .. } => {
collect_step_outputs_and_agents(body, out, step_agents);
collect_step_outputs_and_agents(catch, out, step_agents);
}
FlowNode::Assign { .. } => {} }
}
fn collect_verdict_conds(
node: &FlowNode,
step_outputs: &HashMap<String, String>,
verdict_contracts: &HashMap<String, VerdictContract>,
referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
errors: &mut Vec<CompileError>,
) {
match node {
FlowNode::Branch { cond, then_, else_ } => {
lint_cond_expr(
cond,
"Branch cond",
step_outputs,
verdict_contracts,
referenced_values,
errors,
);
collect_verdict_conds(
then_,
step_outputs,
verdict_contracts,
referenced_values,
errors,
);
collect_verdict_conds(
else_,
step_outputs,
verdict_contracts,
referenced_values,
errors,
);
}
FlowNode::Loop { cond, body, .. } => {
lint_cond_expr(
cond,
"Loop cond",
step_outputs,
verdict_contracts,
referenced_values,
errors,
);
collect_verdict_conds(
body,
step_outputs,
verdict_contracts,
referenced_values,
errors,
);
}
FlowNode::Seq { children } => {
for c in children {
collect_verdict_conds(
c,
step_outputs,
verdict_contracts,
referenced_values,
errors,
);
}
}
FlowNode::Fanout { body, .. } => collect_verdict_conds(
body,
step_outputs,
verdict_contracts,
referenced_values,
errors,
),
FlowNode::Try { body, catch, .. } => {
collect_verdict_conds(
body,
step_outputs,
verdict_contracts,
referenced_values,
errors,
);
collect_verdict_conds(
catch,
step_outputs,
verdict_contracts,
referenced_values,
errors,
);
}
FlowNode::Step { .. } | FlowNode::Assign { .. } => {}
}
}
fn lint_cond_expr(
expr: &Expr,
where_: &str,
step_outputs: &HashMap<String, String>,
verdict_contracts: &HashMap<String, VerdictContract>,
referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
errors: &mut Vec<CompileError>,
) {
match expr {
Expr::Eq { lhs, rhs } | Expr::Ne { lhs, rhs } => {
if let Some((path, lit)) = path_lit_operands(lhs, rhs) {
resolve_and_check(
path,
&[lit],
where_,
step_outputs,
verdict_contracts,
referenced_values,
errors,
);
}
}
Expr::In { needle, haystack } => {
if let (
Expr::Path { at },
Expr::Lit {
value: Value::Array(items),
},
) = (needle.as_ref(), haystack.as_ref())
{
let lits: Vec<&Value> = items.iter().collect();
resolve_and_check(
at,
&lits,
where_,
step_outputs,
verdict_contracts,
referenced_values,
errors,
);
}
}
Expr::And { args } | Expr::Or { args } => {
for a in args {
lint_cond_expr(
a,
where_,
step_outputs,
verdict_contracts,
referenced_values,
errors,
);
}
}
Expr::Not { arg } => lint_cond_expr(
arg,
where_,
step_outputs,
verdict_contracts,
referenced_values,
errors,
),
_ => {}
}
}
fn path_lit_operands<'a>(lhs: &'a Expr, rhs: &'a Expr) -> Option<(&'a Path, &'a Value)> {
match (lhs, rhs) {
(Expr::Path { at }, Expr::Lit { value }) => Some((at, value)),
(Expr::Lit { value }, Expr::Path { at }) => Some((at, value)),
_ => None,
}
}
fn resolve_and_check(
path: &Path,
lits: &[&Value],
where_: &str,
step_outputs: &HashMap<String, String>,
verdict_contracts: &HashMap<String, VerdictContract>,
referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
errors: &mut Vec<CompileError>,
) {
let path_str = path.to_string();
let (agent, actual_shape) = if let Some(agent) = step_outputs.get(&path_str) {
(agent, "body")
} else if let Some(stripped) = path_str.strip_suffix(".parts.verdict") {
match step_outputs.get(stripped) {
Some(agent) => (agent, "part"),
None => return,
}
} else {
return;
};
let Some(contract) = verdict_contracts.get(agent) else {
tracing::warn!(
agent = %agent,
where_ = %where_,
"cond references agent output but no verdict contract declared"
);
return;
};
let expected_channel = match contract.channel {
VerdictChannel::Body => "body",
VerdictChannel::Part => "part",
};
if expected_channel != actual_shape {
errors.push(CompileError::VerdictChannelMismatch {
where_: where_.to_string(),
agent: agent.clone(),
expected_channel: expected_channel.to_string(),
actual_shape: actual_shape.to_string(),
});
return;
}
for lit in lits {
let value_str = lit
.as_str()
.map(str::to_string)
.unwrap_or_else(|| lit.to_string());
if !contract.values.iter().any(|v| v == &value_str) {
errors.push(CompileError::VerdictValueNotInContract {
where_: where_.to_string(),
agent: agent.clone(),
value: value_str.clone(),
values: contract.values.clone(),
});
}
referenced_values
.entry(agent.clone())
.or_default()
.insert(value_str);
}
}
fn check_unhandled_verdict_values(
verdict_contracts: &HashMap<String, VerdictContract>,
referenced_values: &HashMap<String, std::collections::HashSet<String>>,
step_agents: &HashMap<String, String>,
strict_verdict_handling: bool,
errors: &mut Vec<CompileError>,
) {
let mut agents: Vec<&String> = verdict_contracts.keys().collect();
agents.sort();
for agent in agents {
let contract = &verdict_contracts[agent];
let referenced = referenced_values.get(agent);
let step_ref = step_agents
.get(agent)
.cloned()
.unwrap_or_else(|| agent.clone());
for value in &contract.values {
let handled = referenced.map(|set| set.contains(value)).unwrap_or(false);
if handled {
continue;
}
if strict_verdict_handling {
errors.push(CompileError::VerdictValueUnhandled {
agent: agent.clone(),
value: value.clone(),
declared_values: contract.values.clone(),
step_ref: step_ref.clone(),
});
} else {
tracing::warn!(
agent = %agent,
value = %value,
step_ref = %step_ref,
"declared verdict value has no downstream cond handler; \
opt in to `metadata.strict_verdict_handling` to reject at compile"
);
}
}
}
}
pub struct CompiledAgentTable {
pub(crate) routes: HashMap<String, Arc<dyn SpawnerAdapter>>,
pub(crate) default: Option<Arc<dyn SpawnerAdapter>>,
pub(crate) verdict_contracts: HashMap<String, VerdictContract>,
}
impl CompiledAgentTable {
pub fn has_route(&self, agent: &str) -> bool {
self.routes.contains_key(agent)
}
pub fn routed_agents(&self) -> Vec<String> {
self.routes.keys().cloned().collect()
}
pub fn verdict_contract_for(&self, agent: &str) -> Option<&VerdictContract> {
self.verdict_contracts.get(agent)
}
}
#[async_trait]
impl SpawnerAdapter for CompiledAgentTable {
async fn spawn(
&self,
engine: &Engine,
ctx: &Ctx,
task_id: StepId,
attempt: u32,
token: CapToken,
) -> Result<Box<dyn Worker>, SpawnError> {
let sp = self
.routes
.get(&ctx.agent)
.cloned()
.or_else(|| self.default.clone())
.ok_or_else(|| SpawnError::NotRegistered(ctx.agent.clone()))?;
sp.spawn(engine, ctx, task_id, attempt, token).await
}
}
pub struct SubprocessProcessSpawnerFactory;
impl SpawnerFactoryKind for SubprocessProcessSpawnerFactory {
const KIND: AgentKind = AgentKind::Subprocess;
type Worker = crate::worker::process_spawner::ProcessWorker;
}
pub const SUBPROCESS_TEMPLATE_HINT_KEY: &str = "subprocess_template";
pub const SUBPROCESS_OVERRIDES_HINT_KEY: &str = "subprocess_overrides";
fn validate_embed_placeholders(s: &str, where_: &str) -> Result<(), String> {
let mut rest = s;
while let Some(start) = rest.find('{') {
let after = &rest[start + 1..];
let Some(end) = after.find('}') else {
break;
};
let token = &after[..end];
let is_candidate =
!token.is_empty() && token.chars().all(|c| c.is_ascii_lowercase() || c == '_');
if is_candidate {
if !crate::worker::process_spawner::EMBED_PLACEHOLDERS.contains(&token) {
return Err(format!(
"unknown placeholder '{{{token}}}' in {where_}; closed set is \
{{system, system_file, prompt, model, tools_csv, work_dir, task_id, attempt}}"
));
}
rest = &after[end + 1..];
} else {
rest = after;
}
}
Ok(())
}
fn resolve_subprocess_template_hint(
bp: &Blueprint,
ad: &AgentDef,
) -> Result<Option<Value>, CompileError> {
let invalid = |msg: String| CompileError::InvalidSpec {
name: ad.name.clone(),
msg,
};
let runner = mlua_swarm_schema::resolve_runner(bp, ad).map_err(|e| invalid(e.to_string()))?;
let Some(Runner::Subprocess {
template,
overrides,
}) = runner
else {
return Ok(None);
};
let def = bp
.subprocesses
.iter()
.find(|d| d.name == template)
.ok_or_else(|| {
let mut names: Vec<&str> = bp.subprocesses.iter().map(|d| d.name.as_str()).collect();
names.sort_unstable();
invalid(format!(
"Runner::Subprocess template '{template}' not found in \
Blueprint.subprocesses (defined: [{}])",
names.join(", ")
))
})?;
Ok(Some(serde_json::json!({
SUBPROCESS_TEMPLATE_HINT_KEY: def,
SUBPROCESS_OVERRIDES_HINT_KEY: overrides,
})))
}
impl SubprocessProcessSpawnerFactory {
fn build_embed(
agent_def: &AgentDef,
template: &Value,
overrides: Option<&Value>,
) -> Result<ProcessSpawner, CompileError> {
use crate::worker::process_spawner::EmbedTemplate;
use mlua_swarm_schema::{SubprocessDef, SubprocessOverrides};
let agent_name = &agent_def.name;
let invalid = |msg: String| CompileError::InvalidSpec {
name: agent_name.to_string(),
msg,
};
let def: SubprocessDef = serde_json::from_value(template.clone())
.map_err(|e| invalid(format!("subprocess_template hint: {e}")))?;
let overrides: SubprocessOverrides = match overrides {
Some(v) => serde_json::from_value(v.clone())
.map_err(|e| invalid(format!("subprocess_overrides hint: {e}")))?,
None => SubprocessOverrides::default(),
};
if def.argv.is_empty() {
return Err(invalid(format!(
"SubprocessDef '{}': argv must not be empty",
def.name
)));
}
for (i, a) in def.argv.iter().enumerate() {
validate_embed_placeholders(a, &format!("argv[{i}]")).map_err(&invalid)?;
}
if let Some(stdin) = &def.stdin {
validate_embed_placeholders(stdin, "stdin").map_err(&invalid)?;
}
for (k, v) in &def.env {
validate_embed_placeholders(v, &format!("env['{k}']")).map_err(&invalid)?;
}
if let Some(cwd) = &def.cwd {
validate_embed_placeholders(cwd, "cwd").map_err(&invalid)?;
}
let stream_mode = match def.stream_mode.as_deref() {
Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
Some("sse_events") => Some(StreamMode::SseEvents),
Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
None => None,
};
if let Some(output) = &def.output {
if stream_mode.is_some() {
return Err(invalid(format!(
"SubprocessDef '{}': output normalization is a plain-mode \
declaration; remove either `output` or `stream_mode`",
def.name
)));
}
if let Some(format) = output.format.as_deref() {
if format != "json" {
return Err(invalid(format!(
"SubprocessDef '{}': unknown output.format '{format}' \
(supported: \"json\")",
def.name
)));
}
}
if let Some(ptr) = output.result_ptr.as_deref() {
if !ptr.starts_with('/') {
return Err(invalid(format!(
"SubprocessDef '{}': output.result_ptr '{ptr}' is not a \
JSON Pointer (RFC 6901 — must start with '/')",
def.name
)));
}
}
if let Some(ok_from) = output.ok_from.as_deref() {
if ok_from != "exit_code" && !ok_from.starts_with('/') {
return Err(invalid(format!(
"SubprocessDef '{}': output.ok_from '{ok_from}' must be \
\"exit_code\" or a JSON Pointer (starting with '/')",
def.name
)));
}
}
}
let profile = agent_def.profile.as_ref();
let system_prompt = profile
.map(|p| p.system_prompt.clone())
.filter(|s| !s.is_empty());
let model = overrides
.model
.clone()
.or_else(|| profile.and_then(|p| p.model.clone()));
let tools: Vec<String> = if overrides.tools.is_empty() {
profile.map(|p| p.tools.clone()).unwrap_or_default()
} else {
overrides.tools.clone()
};
let cwd = overrides.cwd.clone().or_else(|| def.cwd.clone());
if let Some(c) = &cwd {
validate_embed_placeholders(c, "overrides.cwd").map_err(&invalid)?;
}
let program = def.argv[0].clone();
let sp = ProcessSpawner {
program,
args: Vec::new(),
use_stdin: def.stdin.is_some(),
stream_mode,
embed: Some(EmbedTemplate {
argv: def.argv,
stdin: def.stdin,
env: def.env,
cwd,
output: def.output,
system_prompt,
model,
tools_csv: tools.join(","),
}),
};
Ok(sp)
}
}
impl SpawnerFactory for SubprocessProcessSpawnerFactory {
fn build(
&self,
agent_def: &AgentDef,
hint: Option<&Value>,
) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
if let Some(template) = hint.and_then(|h| h.get(SUBPROCESS_TEMPLATE_HINT_KEY)) {
let overrides = hint.and_then(|h| h.get(SUBPROCESS_OVERRIDES_HINT_KEY));
return Self::build_embed(agent_def, template, overrides).map(|sp| {
let arc: Arc<dyn SpawnerAdapter> = Arc::new(sp);
arc
});
}
let agent_name = &agent_def.name;
let spec = &agent_def.spec;
let invalid = |msg: String| CompileError::InvalidSpec {
name: agent_name.to_string(),
msg,
};
let program = spec
.get("program")
.and_then(|v| v.as_str())
.ok_or_else(|| invalid("shell spec: 'program' (string) required".into()))?
.to_string();
let args: Vec<String> = spec
.get("args")
.and_then(|v| v.as_array())
.map(|a| {
a.iter()
.filter_map(|x| x.as_str().map(|s| s.to_string()))
.collect()
})
.unwrap_or_default();
let use_stdin = spec
.get("use_stdin")
.and_then(|v| v.as_bool())
.unwrap_or(true);
let stream_mode = match spec.get("stream_mode").and_then(|v| v.as_str()) {
Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
Some("sse_events") => Some(StreamMode::SseEvents),
Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
None => None,
};
let mut sp = ProcessSpawner {
program,
args,
use_stdin,
stream_mode,
embed: None,
};
if let Some(mode) = sp.stream_mode.clone() {
sp = sp.stream_mode(mode);
}
Ok(Arc::new(sp))
}
}
pub struct LuaInProcessSpawnerFactory {
registry: HashMap<String, WorkerFn>,
bridges: HashMap<String, HostBridge>,
}
#[derive(Clone)]
pub struct HostBridge(
Arc<dyn Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync>,
);
impl HostBridge {
pub fn new<F>(f: F) -> Self
where
F: Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync + 'static,
{
Self(Arc::new(f))
}
pub fn call(&self, arg: serde_json::Value) -> Result<serde_json::Value, String> {
(self.0)(arg)
}
}
#[derive(Clone)]
pub struct LuaScriptSource {
pub source: String,
pub label: String,
}
impl LuaScriptSource {
pub fn new(source: impl Into<String>, label: impl Into<String>) -> Self {
Self {
source: source.into(),
label: label.into(),
}
}
}
impl LuaInProcessSpawnerFactory {
pub fn new() -> Self {
Self {
registry: HashMap::new(),
bridges: HashMap::new(),
}
}
pub fn with_bridge(mut self, name: impl Into<String>, bridge: HostBridge) -> Self {
self.bridges.insert(name.into(), bridge);
self
}
pub fn register_lua(mut self, fn_id: impl Into<String>, source: LuaScriptSource) -> Self {
let source = Arc::new(source);
let bridges = Arc::new(self.bridges.clone());
let wrapped: WorkerFn = Arc::new(move |inv| {
let source = source.clone();
let bridges = bridges.clone();
Box::pin(run_lua_worker(source, bridges, inv))
});
self.registry.insert(fn_id.into(), wrapped);
self
}
}
async fn run_lua_worker(
source: Arc<LuaScriptSource>,
bridges: Arc<HashMap<String, HostBridge>>,
inv: crate::worker::adapter::WorkerInvocation,
) -> Result<crate::worker::adapter::WorkerResult, crate::worker::adapter::WorkerError> {
use crate::worker::adapter::WorkerError;
use mlua::LuaSerdeExt;
let label = source.label.clone();
let outcome =
tokio::task::spawn_blocking(move || -> Result<(serde_json::Value, bool), String> {
let lua = mlua::Lua::new();
let g = lua.globals();
g.set("_PROMPT", inv.prompt.clone())
.map_err(|e| format!("set _PROMPT: {e}"))?;
g.set("_AGENT", inv.agent.clone())
.map_err(|e| format!("set _AGENT: {e}"))?;
g.set("_TASK_ID", inv.task_id.to_string())
.map_err(|e| format!("set _TASK_ID: {e}"))?;
g.set("_ATTEMPT", inv.attempt as i64)
.map_err(|e| format!("set _ATTEMPT: {e}"))?;
if let Ok(json_val) = serde_json::from_str::<serde_json::Value>(&inv.prompt) {
let lua_val = lua
.to_value(&json_val)
.map_err(|e| format!("_CTX to_value: {e}"))?;
g.set("_CTX", lua_val)
.map_err(|e| format!("set _CTX: {e}"))?;
}
if !bridges.is_empty() {
let host = lua
.create_table()
.map_err(|e| format!("create host table: {e}"))?;
for (name, bridge) in bridges.iter() {
let bridge = bridge.clone();
let bname = name.clone();
let f = lua
.create_function(move |lua, arg: mlua::Value| {
let json_arg: serde_json::Value = lua.from_value(arg).map_err(|e| {
mlua::Error::external(format!("bridge {bname} arg → json: {e}"))
})?;
let result_json =
bridge.call(json_arg).map_err(mlua::Error::external)?;
lua.to_value(&result_json).map_err(|e| {
mlua::Error::external(format!("bridge {bname} ret → lua: {e}"))
})
})
.map_err(|e| format!("create_function {name}: {e}"))?;
host.set(name.as_str(), f)
.map_err(|e| format!("host.{name} set: {e}"))?;
}
g.set("host", host).map_err(|e| format!("set host: {e}"))?;
}
let result: mlua::Value = lua
.load(&source.source)
.set_name(&source.label)
.eval()
.map_err(|e| format!("lua eval [{}]: {e}", source.label))?;
let json_result: serde_json::Value = lua
.from_value(result)
.map_err(|e| format!("lua → json [{}]: {e}", source.label))?;
let (value, ok) = match &json_result {
serde_json::Value::Object(map)
if map.contains_key("value") || map.contains_key("ok") =>
{
let ok = map.get("ok").and_then(|v| v.as_bool()).unwrap_or(true);
let value = map.get("value").cloned().unwrap_or(json_result.clone());
(value, ok)
}
_ => (json_result, true),
};
Ok((value, ok))
})
.await
.map_err(|e| WorkerError::Failed(format!("spawn_blocking join [{label}]: {e}")))?
.map_err(WorkerError::Failed)?;
Ok(crate::worker::adapter::WorkerResult {
value: outcome.0,
ok: outcome.1,
stats: None,
}
.ensure_worker_kind("lua"))
}
impl Default for LuaInProcessSpawnerFactory {
fn default() -> Self {
Self::new()
}
}
impl SpawnerFactoryKind for LuaInProcessSpawnerFactory {
const KIND: AgentKind = AgentKind::Lua;
type Worker = LuaWorker;
}
impl SpawnerFactory for LuaInProcessSpawnerFactory {
fn build(
&self,
agent_def: &AgentDef,
_hint: Option<&Value>,
) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
if let Some(source) = agent_def.spec.get("source").and_then(|v| v.as_str()) {
let label = agent_def
.spec
.get("label")
.and_then(|v| v.as_str())
.map(str::to_string)
.unwrap_or_else(|| format!("{}.lua", agent_def.name));
let script = Arc::new(LuaScriptSource::new(source.to_string(), label));
let bridges = Arc::new(self.bridges.clone());
let wrapped: WorkerFn = Arc::new(move |inv| {
let source = script.clone();
let bridges = bridges.clone();
Box::pin(run_lua_worker(source, bridges, inv))
});
let mut sp: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
sp.registry.insert(agent_def.name.to_string(), wrapped);
return Ok(Arc::new(sp));
}
build_inproc_from_registry::<LuaWorker>(&self.registry, agent_def, "lua")
}
}
pub struct RustFnInProcessSpawnerFactory {
registry: HashMap<String, WorkerFn>,
}
impl RustFnInProcessSpawnerFactory {
pub fn new() -> Self {
Self {
registry: HashMap::new(),
}
}
pub fn register_fn<F, Fut>(mut self, fn_id: impl Into<String>, f: F) -> Self
where
F: Fn(crate::worker::adapter::WorkerInvocation) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<
Output = Result<
crate::worker::adapter::WorkerResult,
crate::worker::adapter::WorkerError,
>,
> + Send
+ 'static,
{
let f = Arc::new(f);
let wrapped: WorkerFn = Arc::new(move |inv| {
let f = f.clone();
Box::pin(f(inv))
});
self.registry.insert(fn_id.into(), wrapped);
self
}
}
impl Default for RustFnInProcessSpawnerFactory {
fn default() -> Self {
Self::new()
}
}
impl SpawnerFactoryKind for RustFnInProcessSpawnerFactory {
const KIND: AgentKind = AgentKind::RustFn;
type Worker = RustFnWorker;
}
impl SpawnerFactory for RustFnInProcessSpawnerFactory {
fn build(
&self,
agent_def: &AgentDef,
_hint: Option<&Value>,
) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
build_inproc_from_registry::<RustFnWorker>(&self.registry, agent_def, "rust_fn")
}
}
fn build_inproc_from_registry<W>(
registry: &HashMap<String, WorkerFn>,
agent_def: &AgentDef,
kind_label: &str,
) -> Result<Arc<dyn SpawnerAdapter>, CompileError>
where
W: crate::worker::Worker + From<crate::worker::WorkerJoinHandler> + Send + Sync + 'static,
{
let agent_name = &agent_def.name;
let spec = &agent_def.spec;
let invalid = |msg: String| CompileError::InvalidSpec {
name: agent_name.to_string(),
msg,
};
let fn_id = spec
.get("fn_id")
.and_then(|v| v.as_str())
.ok_or_else(|| invalid(format!("{kind_label} spec: 'fn_id' (string) required")))?;
let f = registry
.get(fn_id)
.cloned()
.ok_or_else(|| invalid(format!("fn_id '{fn_id}' not registered in factory")))?;
let mut sp: InProcSpawner<W> = InProcSpawner::<W>::typed();
sp.registry.insert(agent_name.to_string(), f);
Ok(Arc::new(sp))
}
pub struct LuaWorker {
pub handler: crate::worker::WorkerJoinHandler,
}
impl From<crate::worker::WorkerJoinHandler> for LuaWorker {
fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
Self { handler }
}
}
#[async_trait::async_trait]
impl crate::worker::Worker for LuaWorker {
fn id(&self) -> &crate::types::WorkerId {
&self.handler.worker_id
}
fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
self.handler.cancel.clone()
}
async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
self.handler.await_completion().await
}
}
pub struct RustFnWorker {
pub handler: crate::worker::WorkerJoinHandler,
}
impl From<crate::worker::WorkerJoinHandler> for RustFnWorker {
fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
Self { handler }
}
}
#[async_trait::async_trait]
impl crate::worker::Worker for RustFnWorker {
fn id(&self) -> &crate::types::WorkerId {
&self.handler.worker_id
}
fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
self.handler.cancel.clone()
}
async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
self.handler.await_completion().await
}
}
pub struct OperatorSpawnerFactory {
operators: Arc<std::sync::RwLock<HashMap<String, Arc<dyn Operator>>>>,
}
impl OperatorSpawnerFactory {
pub fn new() -> Self {
Self {
operators: Arc::new(std::sync::RwLock::new(HashMap::new())),
}
}
pub fn register_operator(&self, id: impl Into<String>, op: Arc<dyn Operator>) -> &Self {
self.operators
.write()
.expect("OperatorSpawnerFactory.operators RwLock poisoned")
.insert(id.into(), op);
self
}
pub fn unregister_operator(&self, id: &str) -> &Self {
self.operators
.write()
.expect("OperatorSpawnerFactory.operators RwLock poisoned")
.remove(id);
self
}
}
impl Default for OperatorSpawnerFactory {
fn default() -> Self {
Self::new()
}
}
impl SpawnerFactoryKind for OperatorSpawnerFactory {
const KIND: AgentKind = AgentKind::Operator;
type Worker = crate::operator::OperatorWorker;
}
impl SpawnerFactory for OperatorSpawnerFactory {
fn build(
&self,
agent_def: &AgentDef,
_hint: Option<&Value>,
) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
let agent_name = &agent_def.name;
let spec = &agent_def.spec;
let system_prompt = agent_def.profile.as_ref().map(|p| p.system_prompt.clone());
let invalid = |msg: String| CompileError::InvalidSpec {
name: agent_name.to_string(),
msg,
};
let op_ref = spec
.get("operator_ref")
.and_then(|v| v.as_str())
.ok_or_else(|| invalid("operator spec: 'operator_ref' (string) required".into()))?;
let operators = self
.operators
.read()
.expect("OperatorSpawnerFactory.operators RwLock poisoned");
let op = operators.get(op_ref).cloned().ok_or_else(|| {
let mut names: Vec<String> = operators.keys().cloned().collect();
names.sort();
let names_list = if names.is_empty() {
"<none>".to_string()
} else {
names.join(", ")
};
invalid(format!(
"operator_ref '{op_ref}' not registered in factory. \
Registered sids: [{names_list}]. \
Hint: call mse_operator_join(roles=[...]) to mint the sid first."
))
})?;
drop(operators);
let worker_binding = agent_def
.profile
.as_ref()
.and_then(|p| p.worker_binding.as_ref())
.map(|variant| WorkerBinding {
variant: variant.clone(),
tools: agent_def
.profile
.as_ref()
.map(|p| p.tools.clone())
.unwrap_or_default(),
request_digest: None,
requested_model: None,
});
if op.requires_worker_binding() && worker_binding.is_none() {
return Err(invalid(format!(
"{WORKER_BINDING_REQUIRED_MSG_PREFIX}. \
Fix by either: \
(a) if authoring the Blueprint JSON directly, add \
`agents[N].profile.worker_binding: \"<subagent-type>\"` \
to the JSON literal; or \
(b) if using an $agent_md file ref, add \
`worker_binding: <subagent-type>` to the agent .md frontmatter."
)));
}
Ok(Arc::new(OperatorSpawner::new(
op,
system_prompt,
worker_binding,
)))
}
}
#[cfg(test)]
mod operator_spawner_factory_worker_binding_tests {
use super::*;
use crate::blueprint::AgentProfile;
use crate::core::ctx::Ctx;
use crate::types::CapToken;
use crate::worker::adapter::{WorkerError, WorkerResult};
struct StubOperator {
requires_binding: bool,
}
#[async_trait]
impl Operator for StubOperator {
async fn execute(
&self,
_ctx: &Ctx,
_system: Option<String>,
_prompt: Value,
_worker: Option<WorkerBinding>,
_worker_token: CapToken,
) -> Result<WorkerResult, WorkerError> {
Ok(WorkerResult {
value: Value::Null,
ok: true,
stats: None,
})
}
fn requires_worker_binding(&self) -> bool {
self.requires_binding
}
}
fn agent_def_with(profile: Option<AgentProfile>) -> AgentDef {
AgentDef {
name: "test-agent".to_string(),
kind: AgentKind::Operator,
spec: serde_json::json!({ "operator_ref": "op1" }),
profile,
meta: None,
runner: None,
runner_ref: None,
verdict: None,
}
}
#[test]
fn build_fails_loud_when_binding_required_but_absent() {
let factory = OperatorSpawnerFactory::new();
factory.register_operator(
"op1",
Arc::new(StubOperator {
requires_binding: true,
}) as Arc<dyn Operator>,
);
let def = agent_def_with(Some(AgentProfile::default()));
match factory.build(&def, None) {
Err(CompileError::InvalidSpec { name, msg }) => {
assert_eq!(name, "test-agent");
assert!(
msg.contains("worker_binding is required"),
"unexpected message: {msg}"
);
assert!(
msg.contains("agents[N].profile.worker_binding"),
"message missing JSON-direct hint (issue #9): {msg}"
);
assert!(
msg.contains("agent .md frontmatter"),
"message missing $agent_md hint: {msg}"
);
}
Err(other) => panic!("expected InvalidSpec, got: {other:?}"),
Ok(_) => panic!("expected compile-time failure, got Ok"),
}
}
#[test]
fn factory_error_message_carries_the_shared_prefix_and_specializes_the_diagnostic() {
let factory = OperatorSpawnerFactory::new();
factory.register_operator(
"op1",
Arc::new(StubOperator {
requires_binding: true,
}) as Arc<dyn Operator>,
);
let def = agent_def_with(Some(AgentProfile::default()));
let err = match factory.build(&def, None) {
Err(err) => err,
Ok(_) => panic!("expected compile-time failure, got Ok"),
};
match &err {
CompileError::InvalidSpec { msg, .. } => {
assert!(
msg.starts_with(WORKER_BINDING_REQUIRED_MSG_PREFIX),
"factory message must start with the shared prefix, got: {msg}"
);
}
other => panic!("expected InvalidSpec, got: {other:?}"),
}
let d = mlua_swarm_diag::Diagnostic::from(&err);
assert_eq!(d.kind, "worker-binding-missing");
}
#[test]
fn build_succeeds_when_binding_required_and_present() {
let factory = OperatorSpawnerFactory::new();
factory.register_operator(
"op1",
Arc::new(StubOperator {
requires_binding: true,
}) as Arc<dyn Operator>,
);
let profile = AgentProfile {
worker_binding: Some("mse-worker-coder".to_string()),
tools: vec!["Read".to_string(), "Edit".to_string()],
..Default::default()
};
let def = agent_def_with(Some(profile));
assert!(
factory.build(&def, None).is_ok(),
"expected Ok when worker_binding is declared"
);
}
#[test]
fn build_succeeds_when_binding_not_required_and_absent() {
let factory = OperatorSpawnerFactory::new();
factory.register_operator(
"op1",
Arc::new(StubOperator {
requires_binding: false,
}) as Arc<dyn Operator>,
);
let def = agent_def_with(Some(AgentProfile::default()));
assert!(
factory.build(&def, None).is_ok(),
"backends that don't require a binding must not be gated by its absence"
);
}
}
#[cfg(test)]
mod lua_inline_source_tests {
use super::*;
use crate::types::{CapToken, Role, StepId};
fn agent(name: &str, spec: Value) -> AgentDef {
AgentDef {
name: name.to_string(),
kind: AgentKind::Lua,
spec,
profile: None,
meta: None,
runner: None,
runner_ref: None,
verdict: None,
}
}
fn test_invocation(prompt: &str) -> crate::worker::adapter::WorkerInvocation {
crate::worker::adapter::WorkerInvocation {
token: CapToken {
agent_id: "a".into(),
role: Role::Worker,
scopes: vec!["*".into()],
issued_at: 0,
expire_at: u64::MAX / 2,
max_uses: None,
nonce: "test-nonce".into(),
sig_hex: "".into(),
},
task_id: StepId::parse("ST-test").expect("StepId parse"),
attempt: 1,
agent: "g".into(),
prompt: prompt.into(),
sink: None,
cancel_token: None,
}
}
#[test]
fn build_accepts_inline_source_without_pre_registration() {
let factory = LuaInProcessSpawnerFactory::new();
let def = agent(
"g",
serde_json::json!({ "source": "return { value = 42, ok = true }" }),
);
assert!(
factory.build(&def, None).is_ok(),
"inline spec.source must build without a pre-registered fn_id"
);
}
#[test]
fn build_rejects_when_neither_source_nor_fn_id_is_present() {
let factory = LuaInProcessSpawnerFactory::new();
let def = agent("g", serde_json::json!({}));
match factory.build(&def, None) {
Err(CompileError::InvalidSpec { msg, .. }) => {
assert!(
msg.contains("fn_id"),
"empty spec must still surface the fn_id-required message: {msg}"
);
}
Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
Ok(_) => panic!("expected InvalidSpec, got Ok(SpawnerAdapter)"),
}
}
#[tokio::test]
async fn inline_source_evaluates_and_marshals_result() {
let source =
LuaScriptSource::new("return { value = _PROMPT .. '!', ok = true }", "smoke.lua");
let out = run_lua_worker(
std::sync::Arc::new(source),
std::sync::Arc::new(HashMap::new()),
test_invocation("hello"),
)
.await
.expect("lua worker ok");
assert_eq!(out.value, serde_json::json!("hello!"));
assert!(out.ok);
}
#[tokio::test]
async fn inline_source_can_signal_agent_level_failure() {
let source = LuaScriptSource::new("return { value = 'nope', ok = false }", "gate.lua");
let out = run_lua_worker(
std::sync::Arc::new(source),
std::sync::Arc::new(HashMap::new()),
test_invocation("input"),
)
.await
.expect("lua worker ok");
assert_eq!(out.value, serde_json::json!("nope"));
assert!(!out.ok);
}
}
#[cfg(test)]
mod meta_ref_validation_tests {
use super::*;
use crate::blueprint::{AgentMeta, MetaDef};
use crate::worker::adapter::WorkerResult;
fn registry_with_echo() -> SpawnerRegistry {
let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
Ok(WorkerResult {
value: Value::String(inv.prompt),
ok: true,
stats: None,
})
});
let mut reg = SpawnerRegistry::new();
reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
reg
}
fn rustfn_agent(name: &str) -> AgentDef {
AgentDef {
name: name.to_string(),
kind: AgentKind::RustFn,
spec: serde_json::json!({ "fn_id": "echo" }),
profile: None,
meta: None,
runner: None,
runner_ref: None,
verdict: None,
}
}
fn simple_flow(agent_ref: &str, in_: Expr) -> FlowNode {
FlowNode::Step {
ref_: agent_ref.to_string(),
in_,
out: Expr::Path {
at: "$.output".parse().expect("literal test path: $.output"),
},
}
}
fn minimal_bp(agents: Vec<AgentDef>, metas: Vec<MetaDef>, flow: FlowNode) -> Blueprint {
Blueprint {
schema_version: crate::blueprint::current_schema_version(),
id: "meta-ref-ut".into(),
flow,
agents,
operators: vec![],
metas,
hints: Default::default(),
strategy: Default::default(),
metadata: BlueprintMetadata::default(),
spawner_hints: Default::default(),
default_agent_kind: AgentKind::Operator,
default_operator_kind: None,
default_init_ctx: None,
default_agent_ctx: None,
default_context_policy: None,
projection_placement: None,
audits: vec![],
degradation_policy: None,
runners: vec![],
default_runner: None,
subprocesses: vec![],
check_policy: None,
blueprint_ref_includes: Vec::new(),
}
}
#[test]
fn valid_meta_ref_compiles() {
let mut agent = rustfn_agent("worker");
agent.meta = Some(AgentMeta {
meta_ref: Some("shared".to_string()),
..Default::default()
});
let bp = minimal_bp(
vec![agent],
vec![MetaDef {
name: "shared".into(),
ctx: serde_json::json!({ "k": "v" }),
}],
simple_flow(
"worker",
Expr::Path {
at: "$.input".parse().expect("literal test path: $.input"),
},
),
);
let compiler = Compiler::new(registry_with_echo());
assert!(
compiler.compile(&bp).is_ok(),
"a resolvable AgentMeta.meta_ref must compile"
);
}
#[test]
fn unknown_agent_meta_ref_is_unresolved_meta_ref() {
let mut agent = rustfn_agent("worker");
agent.meta = Some(AgentMeta {
meta_ref: Some("missing".to_string()),
..Default::default()
});
let bp = minimal_bp(
vec![agent],
vec![],
simple_flow(
"worker",
Expr::Path {
at: "$.input".parse().expect("literal test path: $.input"),
},
),
);
let compiler = Compiler::new(registry_with_echo());
match compiler.compile(&bp) {
Err(CompileError::UnresolvedMetaRef {
where_,
meta_ref,
defined,
}) => {
assert!(
where_.contains("worker"),
"where_ must name the agent: {where_}"
);
assert_eq!(meta_ref, "missing");
assert!(defined.is_empty());
}
Err(other) => {
panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
}
Ok(_) => panic!("expected compile-time failure, got Ok"),
}
}
#[test]
fn unknown_static_step_meta_ref_in_lit_is_unresolved_meta_ref() {
let agent = rustfn_agent("worker");
let in_ = Expr::Lit {
value: serde_json::json!({ "$step_meta": { "ref": "missing" }, "$in": "go" }),
};
let bp = minimal_bp(vec![agent], vec![], simple_flow("worker", in_));
let compiler = Compiler::new(registry_with_echo());
match compiler.compile(&bp) {
Err(CompileError::UnresolvedMetaRef {
where_, meta_ref, ..
}) => {
assert!(
where_.contains("worker"),
"where_ must name the offending step: {where_}"
);
assert_eq!(meta_ref, "missing");
}
Err(other) => {
panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
}
Ok(_) => panic!("expected compile-time failure, got Ok"),
}
}
#[test]
fn path_op_input_with_no_static_envelope_compiles_fine() {
let agent = rustfn_agent("worker");
let bp = minimal_bp(
vec![agent],
vec![],
simple_flow(
"worker",
Expr::Path {
at: "$.input".parse().expect("literal test path: $.input"),
},
),
);
let compiler = Compiler::new(registry_with_echo());
assert!(
compiler.compile(&bp).is_ok(),
"a non-Lit Step.in must not trigger the best-effort static $step_meta check"
);
}
}
#[cfg(test)]
mod audit_agent_validation_tests {
use super::*;
use crate::worker::adapter::WorkerResult;
use mlua_swarm_schema::{AuditDef, AuditMode};
fn registry_with_echo() -> SpawnerRegistry {
let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
Ok(WorkerResult {
value: Value::String(inv.prompt),
ok: true,
stats: None,
})
});
let mut reg = SpawnerRegistry::new();
reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
reg
}
fn rustfn_agent(name: &str) -> AgentDef {
AgentDef {
name: name.to_string(),
kind: AgentKind::RustFn,
spec: serde_json::json!({ "fn_id": "echo" }),
profile: None,
meta: None,
runner: None,
runner_ref: None,
verdict: None,
}
}
fn minimal_bp(agents: Vec<AgentDef>, audits: Vec<AuditDef>) -> Blueprint {
Blueprint {
schema_version: crate::blueprint::current_schema_version(),
id: "audit-ref-ut".into(),
flow: FlowNode::Step {
ref_: "worker".to_string(),
in_: Expr::Path {
at: "$.input".parse().expect("literal test path: $.input"),
},
out: Expr::Path {
at: "$.output".parse().expect("literal test path: $.output"),
},
},
agents,
operators: vec![],
metas: vec![],
hints: Default::default(),
strategy: Default::default(),
metadata: BlueprintMetadata::default(),
spawner_hints: Default::default(),
default_agent_kind: AgentKind::Operator,
default_operator_kind: None,
default_init_ctx: None,
default_agent_ctx: None,
default_context_policy: None,
projection_placement: None,
audits,
degradation_policy: None,
runners: vec![],
default_runner: None,
subprocesses: vec![],
check_policy: None,
blueprint_ref_includes: Vec::new(),
}
}
#[test]
fn unresolved_audit_agent_is_a_loud_compile_error() {
let bp = minimal_bp(
vec![rustfn_agent("worker")],
vec![AuditDef {
agent: "missing-auditor".to_string(),
steps: None,
mode: AuditMode::default(),
}],
);
let compiler = Compiler::new(registry_with_echo());
match compiler.compile(&bp) {
Err(CompileError::UnresolvedAuditAgent { agent, defined }) => {
assert_eq!(agent, "missing-auditor");
assert_eq!(defined, vec!["worker".to_string()]);
}
Err(other) => {
panic!("expected UnresolvedAuditAgent, got a different CompileError: {other}")
}
Ok(_) => panic!("expected compile-time failure, got Ok"),
}
}
#[test]
fn resolved_audit_agent_compiles_fine() {
let bp = minimal_bp(
vec![rustfn_agent("worker"), rustfn_agent("auditor")],
vec![AuditDef {
agent: "auditor".to_string(),
steps: None,
mode: AuditMode::default(),
}],
);
let compiler = Compiler::new(registry_with_echo());
assert!(
compiler.compile(&bp).is_ok(),
"an audits[].agent that names a declared AgentDef must compile"
);
}
}
#[cfg(test)]
mod projection_placement_compile_tests {
use super::*;
use crate::core::projection_placement::{ProjectionPlacement, RootPreference};
use crate::worker::adapter::WorkerResult;
use mlua_swarm_schema::ProjectionPlacementSpec;
fn registry_with_echo() -> SpawnerRegistry {
let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
Ok(WorkerResult {
value: Value::String(inv.prompt),
ok: true,
stats: None,
})
});
let mut reg = SpawnerRegistry::new();
reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
reg
}
fn minimal_bp(projection_placement: Option<ProjectionPlacementSpec>) -> Blueprint {
Blueprint {
schema_version: crate::blueprint::current_schema_version(),
id: "projection-placement-ut".into(),
flow: FlowNode::Step {
ref_: "worker".to_string(),
in_: Expr::Path {
at: "$.input".parse().expect("literal test path: $.input"),
},
out: Expr::Path {
at: "$.output".parse().expect("literal test path: $.output"),
},
},
agents: vec![AgentDef {
name: "worker".to_string(),
kind: AgentKind::RustFn,
spec: serde_json::json!({ "fn_id": "echo" }),
profile: None,
meta: None,
runner: None,
runner_ref: None,
verdict: None,
}],
operators: vec![],
metas: vec![],
hints: Default::default(),
strategy: Default::default(),
metadata: BlueprintMetadata::default(),
spawner_hints: Default::default(),
default_agent_kind: AgentKind::Operator,
default_operator_kind: None,
default_init_ctx: None,
default_agent_ctx: None,
default_context_policy: None,
projection_placement,
audits: vec![],
degradation_policy: None,
runners: vec![],
default_runner: None,
subprocesses: vec![],
check_policy: None,
blueprint_ref_includes: Vec::new(),
}
}
#[test]
fn undeclared_projection_placement_compiles_to_byte_compat_default() {
let bp = minimal_bp(None);
let compiled = Compiler::new(registry_with_echo())
.compile(&bp)
.expect("undeclared projection_placement compiles");
assert_eq!(
*compiled.projection_placement,
ProjectionPlacement::default()
);
}
#[test]
fn declared_valid_projection_placement_compiles_to_matching_resolver() {
let bp = minimal_bp(Some(ProjectionPlacementSpec {
root: Some("project_root".to_string()),
dir_template: Some("custom/{task_id}/out".to_string()),
}));
let compiled = Compiler::new(registry_with_echo())
.compile(&bp)
.expect("valid projection_placement compiles");
assert_eq!(
compiled.projection_placement.root_preference,
RootPreference::ProjectRoot
);
assert_eq!(
compiled.projection_placement.dir_template,
"custom/{task_id}/out"
);
}
#[test]
fn declared_invalid_dir_template_rejects_compile() {
let bp = minimal_bp(Some(ProjectionPlacementSpec {
root: None,
dir_template: Some("workspace/tasks/ctx".to_string()), }));
match Compiler::new(registry_with_echo()).compile(&bp) {
Err(CompileError::InvalidProjectionPlacement(_)) => {}
Err(other) => {
panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
}
Ok(_) => {
panic!("expected compile-time rejection for a missing {{task_id}} placeholder")
}
}
}
#[test]
fn declared_invalid_root_literal_rejects_compile() {
let bp = minimal_bp(Some(ProjectionPlacementSpec {
root: Some("nope".to_string()),
dir_template: None,
}));
match Compiler::new(registry_with_echo()).compile(&bp) {
Err(CompileError::InvalidProjectionPlacement(_)) => {}
Err(other) => {
panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
}
Ok(_) => panic!("expected compile-time rejection for an invalid root literal"),
}
}
}
#[cfg(test)]
mod verdict_contract_lint_tests {
use super::*;
use crate::worker::adapter::WorkerResult;
fn registry_with_echo() -> SpawnerRegistry {
let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
Ok(WorkerResult {
value: Value::String(inv.prompt),
ok: true,
stats: None,
})
});
let mut reg = SpawnerRegistry::new();
reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
reg
}
fn gate_agent(verdict: Option<VerdictContract>) -> AgentDef {
AgentDef {
name: "gate".to_string(),
kind: AgentKind::RustFn,
spec: serde_json::json!({ "fn_id": "echo" }),
profile: None,
meta: None,
runner: None,
runner_ref: None,
verdict,
}
}
fn minimal_bp(agent: AgentDef, flow: FlowNode) -> Blueprint {
Blueprint {
schema_version: crate::blueprint::current_schema_version(),
id: "verdict-contract-ut".into(),
flow,
agents: vec![agent],
operators: vec![],
metas: vec![],
hints: Default::default(),
strategy: Default::default(),
metadata: BlueprintMetadata::default(),
spawner_hints: Default::default(),
default_agent_kind: AgentKind::Operator,
default_operator_kind: None,
default_init_ctx: None,
default_agent_ctx: None,
default_context_policy: None,
projection_placement: None,
audits: vec![],
degradation_policy: None,
runners: vec![],
default_runner: None,
subprocesses: vec![],
check_policy: None,
blueprint_ref_includes: Vec::new(),
}
}
fn step(ref_: &str, out_path: &str) -> FlowNode {
FlowNode::Step {
ref_: ref_.to_string(),
in_: Expr::Lit { value: Value::Null },
out: Expr::Path {
at: out_path.parse().expect("literal test path"),
},
}
}
fn noop() -> FlowNode {
FlowNode::Seq { children: vec![] }
}
fn eq_cond(path: &str, lit: &str) -> Expr {
Expr::Eq {
lhs: Box::new(Expr::Path {
at: path.parse().expect("literal test path"),
}),
rhs: Box::new(Expr::Lit {
value: Value::String(lit.to_string()),
}),
}
}
fn branch(cond: Expr, then_: FlowNode, else_: FlowNode) -> FlowNode {
FlowNode::Branch {
cond,
then_: Box::new(then_),
else_: Box::new(else_),
}
}
fn body_contract(values: &[&str]) -> VerdictContract {
VerdictContract {
channel: VerdictChannel::Body,
values: values.iter().map(|v| v.to_string()).collect(),
}
}
fn part_contract(values: &[&str]) -> VerdictContract {
VerdictContract {
channel: VerdictChannel::Part,
values: values.iter().map(|v| v.to_string()).collect(),
}
}
#[test]
fn contract_with_correct_body_channel_and_value_compiles() {
let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
let flow = FlowNode::Seq {
children: vec![
step("gate", "$.verdict"),
branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
],
};
let bp = minimal_bp(agent, flow);
assert!(
Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
"a cond addressing the bare step output must match a channel: \"body\" contract"
);
}
#[test]
fn contract_with_correct_part_channel_and_value_compiles() {
let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
let flow = FlowNode::Seq {
children: vec![
step("gate", "$.gate"),
branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
],
};
let bp = minimal_bp(agent, flow);
assert!(
Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
"a cond addressing '<step>.parts.verdict' must match a channel: \"part\" contract"
);
}
#[test]
fn body_channel_contract_rejects_cond_addressing_parts_verdict() {
let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
let flow = FlowNode::Seq {
children: vec![
step("gate", "$.gate"),
branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
],
};
let bp = minimal_bp(agent, flow);
match Compiler::new(registry_with_echo()).compile(&bp) {
Err(CompileError::VerdictChannelMismatch {
where_,
agent,
expected_channel,
actual_shape,
}) => {
assert_eq!(agent, "gate");
assert_eq!(expected_channel, "body");
assert_eq!(actual_shape, "part");
assert!(where_.contains("Branch cond"), "where_: {where_}");
}
Err(other) => {
panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
}
Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
}
}
#[test]
fn part_channel_contract_rejects_cond_addressing_bare_output() {
let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
let flow = FlowNode::Seq {
children: vec![
step("gate", "$.verdict"),
branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
],
};
let bp = minimal_bp(agent, flow);
match Compiler::new(registry_with_echo()).compile(&bp) {
Err(CompileError::VerdictChannelMismatch {
agent,
expected_channel,
actual_shape,
..
}) => {
assert_eq!(agent, "gate");
assert_eq!(expected_channel, "part");
assert_eq!(actual_shape, "body");
}
Err(other) => {
panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
}
Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
}
}
#[test]
fn contract_rejects_lit_outside_declared_values() {
let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
let flow = FlowNode::Seq {
children: vec![
step("gate", "$.verdict"),
branch(eq_cond("$.verdict", "UNKNOWN"), noop(), noop()),
],
};
let bp = minimal_bp(agent, flow);
match Compiler::new(registry_with_echo()).compile(&bp) {
Err(CompileError::VerdictValueNotInContract {
agent,
value,
values,
..
}) => {
assert_eq!(agent, "gate");
assert_eq!(value, "UNKNOWN");
assert_eq!(values, vec!["PASS".to_string(), "BLOCKED".to_string()]);
}
Err(other) => {
panic!("expected VerdictValueNotInContract, got a different CompileError: {other}")
}
Ok(_) => panic!("expected compile-time rejection for a Lit outside declared values"),
}
}
#[test]
fn undeclared_agent_referenced_by_cond_compiles_with_warning_only() {
let agent = gate_agent(None);
let flow = FlowNode::Seq {
children: vec![
step("gate", "$.verdict"),
branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
],
};
let bp = minimal_bp(agent, flow);
assert!(
Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
"an undeclared verdict contract must never reject compile (opt-in, back-compat)"
);
}
#[test]
fn in_expr_with_lit_haystack_members_compiles() {
let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
let cond = Expr::In {
needle: Box::new(Expr::Path {
at: "$.verdict".parse().expect("literal test path"),
}),
haystack: Box::new(Expr::Lit {
value: serde_json::json!(["PASS", "BLOCKED"]),
}),
};
let flow = FlowNode::Seq {
children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
};
let bp = minimal_bp(agent, flow);
assert!(
Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
"an `In` haystack whose every Lit is a declared value must compile"
);
}
#[test]
fn strict_mode_rejects_unhandled_declared_value() {
let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
let flow = FlowNode::Seq {
children: vec![
step("gate", "$.verdict"),
branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
],
};
let mut bp = minimal_bp(agent, flow);
bp.metadata.strict_verdict_handling = Some(true);
match Compiler::new(registry_with_echo()).compile(&bp) {
Err(CompileError::VerdictValueUnhandled {
agent,
value,
declared_values,
step_ref,
}) => {
assert_eq!(agent, "gate");
assert_eq!(value, "PASS");
assert_eq!(
declared_values,
vec!["PASS".to_string(), "BLOCKED".to_string()]
);
assert_eq!(step_ref, "gate");
}
Err(other) => {
panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
}
Ok(_) => panic!(
"expected compile-time rejection for a declared verdict value with no \
downstream handler under strict_verdict_handling=Some(true)"
),
}
}
#[test]
fn default_mode_permits_unhandled_declared_value() {
let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
let flow = FlowNode::Seq {
children: vec![
step("gate", "$.verdict"),
branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
],
};
let bp = minimal_bp(agent, flow);
assert!(
Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
"default mode must never reject a Blueprint for unhandled declared values \
(opt-in, back-compat with GH #50)"
);
}
#[test]
fn strict_mode_accepts_all_declared_values_handled() {
let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
let flow = FlowNode::Seq {
children: vec![
step("gate", "$.verdict"),
branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
],
};
let mut bp = minimal_bp(agent, flow);
bp.metadata.strict_verdict_handling = Some(true);
assert!(
Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
"strict mode must accept a Blueprint that handles every declared value"
);
}
#[test]
fn strict_mode_accepts_declared_values_covered_by_in_expr() {
let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
let cond = Expr::In {
needle: Box::new(Expr::Path {
at: "$.verdict".parse().expect("literal test path"),
}),
haystack: Box::new(Expr::Lit {
value: serde_json::json!(["PASS", "BLOCKED"]),
}),
};
let flow = FlowNode::Seq {
children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
};
let mut bp = minimal_bp(agent, flow);
bp.metadata.strict_verdict_handling = Some(true);
assert!(
Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
"strict mode must accept an `In` haystack that covers every declared value"
);
}
#[test]
fn strict_mode_rejects_unhandled_part_channel_value() {
let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
let flow = FlowNode::Seq {
children: vec![
step("gate", "$.gate"),
branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
],
};
let mut bp = minimal_bp(agent, flow);
bp.metadata.strict_verdict_handling = Some(true);
match Compiler::new(registry_with_echo()).compile(&bp) {
Err(CompileError::VerdictValueUnhandled {
agent,
value,
step_ref,
..
}) => {
assert_eq!(agent, "gate");
assert_eq!(value, "PASS");
assert_eq!(step_ref, "gate");
}
Err(other) => {
panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
}
Ok(_) => panic!(
"expected compile-time rejection for a declared verdict value with no \
downstream handler (part channel) under strict_verdict_handling=Some(true)"
),
}
}
#[test]
fn verdict_omitted_blueprint_compiles_unchanged_with_empty_contracts() {
let agent = gate_agent(None);
let flow = FlowNode::Seq {
children: vec![
step("gate", "$.verdict"),
FlowNode::Loop {
counter: Expr::Path {
at: "$.n".parse().expect("literal test path"),
},
cond: eq_cond("$.verdict", "BLOCKED"),
body: Box::new(step("gate", "$.verdict")),
max: 3,
},
branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
],
};
let bp = minimal_bp(agent, flow);
let compiled = Compiler::new(registry_with_echo())
.compile(&bp)
.expect("a verdict-omitted Blueprint must compile unchanged");
assert!(
compiled.router.verdict_contracts.is_empty(),
"no agent declared a verdict contract"
);
}
#[test]
fn every_compile_error_diagnostic_kind_is_a_declared_lint() {
let kinds = [
"bound-agent-resolution",
"unknown-agent-kind",
"invalid-agent-spec",
"worker-binding-missing",
"unresolved-agent-ref",
"duplicate-agent-name",
"unresolved-operator-ref",
"unresolved-meta-ref",
"step-naming-collision",
"invalid-projection-placement",
"unresolved-audit-agent",
"verdict-channel-mismatch",
"verdict-value-not-in-contract",
"verdict-value-unhandled",
];
for kind in kinds {
assert!(
mlua_swarm_diag::lint_decl(kind).is_some(),
"kind '{kind}' emitted by From<&CompileError> has no LINT_DECLS entry"
);
}
}
#[test]
fn invalid_spec_with_worker_binding_prefix_specializes_the_diagnostic_kind() {
let err = CompileError::InvalidSpec {
name: "greeter".into(),
msg: format!("{WORKER_BINDING_REQUIRED_MSG_PREFIX}. Fix by either: (a) ..."),
};
let d = mlua_swarm_diag::Diagnostic::from(&err);
assert_eq!(d.kind, "worker-binding-missing");
assert_eq!(d.level, mlua_swarm_diag::DiagLevel::Error);
assert!(matches!(d.stage, mlua_swarm_diag::DiagStage::CompileLint));
assert!(d.message.contains("greeter"));
let suggestion = d
.suggestion
.expect("specialized arm must carry a suggestion");
assert!(suggestion.patch.contains("backend = \"ws_operator\""));
assert_eq!(
suggestion.applicability,
mlua_swarm_diag::Applicability::HasPlaceholders
);
assert_eq!(
d.docs_ref.expect("docs_ref must be set").uri,
"mse://guides/bp-dsl-templates"
);
match d.span.expect("span must be set").element {
mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "greeter"),
other => panic!("expected Agent span, got {other:?}"),
}
}
#[test]
fn generic_invalid_spec_maps_to_the_generic_kind() {
let err = CompileError::InvalidSpec {
name: "solo".into(),
msg: "operator spec: 'operator_ref' (string) required".into(),
};
let d = mlua_swarm_diag::Diagnostic::from(&err);
assert_eq!(d.kind, "invalid-agent-spec");
assert!(
d.suggestion.is_none(),
"generic arm carries no canned patch"
);
}
#[test]
fn verdict_value_not_in_contract_diagnostic_carries_suggestion_and_span() {
let err = CompileError::VerdictValueNotInContract {
where_: "Branch cond".into(),
agent: "review".into(),
value: "NOT_DECLARED".into(),
values: vec!["PASS".into(), "BLOCKED".into()],
};
let d = mlua_swarm_diag::Diagnostic::from(&err);
assert_eq!(d.kind, "verdict-value-not-in-contract");
assert!(d.message.contains("NOT_DECLARED"));
assert!(d.suggestion.is_some());
match d.span.expect("span must be set").element {
mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "review"),
other => panic!("expected Agent span, got {other:?}"),
}
}
}
#[cfg(test)]
mod subprocess_embed_compile_tests {
use super::*;
use mlua_swarm_schema::{current_schema_version, SubprocessDef, SubprocessOverrides};
fn subprocess_agent(name: &str, runner: Option<Runner>) -> AgentDef {
AgentDef {
name: name.to_string(),
kind: AgentKind::Subprocess,
spec: serde_json::json!({}),
profile: Some(AgentProfile {
system_prompt: "you are a headless worker".to_string(),
model: Some("profile-model".to_string()),
tools: vec!["Read".to_string()],
..Default::default()
}),
meta: None,
runner,
runner_ref: None,
verdict: None,
}
}
fn echo_def(name: &str) -> SubprocessDef {
SubprocessDef {
name: name.to_string(),
argv: vec!["sh".to_string(), "-c".to_string(), "cat".to_string()],
stdin: Some("{prompt}".to_string()),
env: Default::default(),
cwd: None,
output: None,
stream_mode: None,
}
}
fn bp_with(agents: Vec<AgentDef>, subprocesses: Vec<SubprocessDef>) -> Blueprint {
Blueprint {
schema_version: current_schema_version(),
id: "gh83-ut".into(),
flow: FlowNode::Seq { children: vec![] },
agents,
operators: vec![],
metas: vec![],
hints: Default::default(),
strategy: Default::default(),
metadata: BlueprintMetadata::default(),
spawner_hints: Default::default(),
default_agent_kind: AgentKind::Operator,
default_operator_kind: None,
default_init_ctx: None,
default_agent_ctx: None,
default_context_policy: None,
projection_placement: None,
audits: vec![],
degradation_policy: None,
runners: vec![],
default_runner: None,
subprocesses,
check_policy: None,
blueprint_ref_includes: vec![],
}
}
fn subprocess_runner(template: &str) -> Runner {
Runner::Subprocess {
template: template.to_string(),
overrides: SubprocessOverrides::default(),
}
}
#[test]
fn validate_placeholders_accepts_closed_set_and_json_braces() {
for ok in [
"{system} {system_file} {prompt} {model} {tools_csv} {work_dir} {task_id} {attempt}",
r#"echo '{"result": "ok", "nested": {"a": 1}}'"#,
"no placeholders at all",
"unmatched { brace",
] {
validate_embed_placeholders(ok, "ut").expect("must be accepted");
}
}
#[test]
fn validate_placeholders_rejects_unknown_token() {
let err = validate_embed_placeholders("--flag {evil}", "argv[1]").unwrap_err();
assert!(err.contains("'{evil}'"), "token named: {err}");
assert!(err.contains("closed set"), "closed set listed: {err}");
}
#[test]
fn validate_placeholders_descends_into_literal_braces() {
validate_embed_placeholders(r#"{"task": "{prompt}"}"#, "stdin")
.expect("nested closed-set token must be accepted");
let err = validate_embed_placeholders(r#"{"task": "{evil}"}"#, "stdin").unwrap_err();
assert!(
err.contains("'{evil}'"),
"nested unknown token caught: {err}"
);
}
#[test]
fn hint_resolution_finds_declared_template() {
let agent = subprocess_agent("headless", Some(subprocess_runner("echo")));
let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
let hint = resolve_subprocess_template_hint(&bp, &agent)
.expect("resolves")
.expect("Runner::Subprocess must synthesize a hint");
assert_eq!(hint[SUBPROCESS_TEMPLATE_HINT_KEY]["name"], "echo");
assert!(hint.get(SUBPROCESS_OVERRIDES_HINT_KEY).is_some());
}
#[test]
fn hint_resolution_unknown_template_is_invalid_spec() {
let agent = subprocess_agent("headless", Some(subprocess_runner("nope")));
let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
let err = resolve_subprocess_template_hint(&bp, &agent).unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("'nope'"), "missing template named: {msg}");
assert!(msg.contains("echo"), "defined templates listed: {msg}");
}
#[test]
fn hint_resolution_none_without_subprocess_runner() {
let agent = subprocess_agent("headless", None);
let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
let hint = resolve_subprocess_template_hint(&bp, &agent).expect("resolves");
assert!(hint.is_none(), "spec-based agents keep the historical path");
}
#[test]
fn build_embed_rejects_unknown_placeholder() {
let agent = subprocess_agent("headless", None);
let mut def = echo_def("echo");
def.argv.push("--x={evil}".to_string());
let err = SubprocessProcessSpawnerFactory::build_embed(
&agent,
&serde_json::to_value(&def).unwrap(),
None,
)
.unwrap_err();
assert!(format!("{err}").contains("'{evil}'"));
}
#[test]
fn build_embed_rejects_output_with_stream_mode() {
let agent = subprocess_agent("headless", None);
let mut def = echo_def("echo");
def.stream_mode = Some("ndjson_lines".to_string());
def.output = Some(mlua_swarm_schema::SubprocessOutput {
format: Some("json".to_string()),
result_ptr: None,
ok_from: None,
stats: None,
});
let err = SubprocessProcessSpawnerFactory::build_embed(
&agent,
&serde_json::to_value(&def).unwrap(),
None,
)
.unwrap_err();
assert!(format!("{err}").contains("plain-mode"));
}
#[test]
fn build_embed_rejects_malformed_result_ptr_and_ok_from() {
let agent = subprocess_agent("headless", None);
let mut def = echo_def("echo");
def.output = Some(mlua_swarm_schema::SubprocessOutput {
format: None,
result_ptr: Some("result".to_string()),
ok_from: None,
stats: None,
});
let err = SubprocessProcessSpawnerFactory::build_embed(
&agent,
&serde_json::to_value(&def).unwrap(),
None,
)
.unwrap_err();
assert!(format!("{err}").contains("JSON Pointer"));
let mut def = echo_def("echo");
def.output = Some(mlua_swarm_schema::SubprocessOutput {
format: None,
result_ptr: None,
ok_from: Some("status".to_string()),
stats: None,
});
let err = SubprocessProcessSpawnerFactory::build_embed(
&agent,
&serde_json::to_value(&def).unwrap(),
None,
)
.unwrap_err();
assert!(format!("{err}").contains("exit_code"));
}
#[test]
fn build_embed_bakes_profile_with_override_precedence() {
let agent = subprocess_agent("headless", None);
let def = echo_def("echo");
let overrides = SubprocessOverrides {
model: Some("override-model".to_string()),
tools: vec!["Bash".to_string(), "Write".to_string()],
cwd: Some("/tmp/override-wd".to_string()),
};
let sp = SubprocessProcessSpawnerFactory::build_embed(
&agent,
&serde_json::to_value(&def).unwrap(),
Some(&serde_json::to_value(&overrides).unwrap()),
)
.expect("builds");
let embed = sp.embed.as_ref().expect("embed template baked");
assert_eq!(embed.model.as_deref(), Some("override-model"));
assert_eq!(embed.tools_csv, "Bash,Write");
assert_eq!(embed.cwd.as_deref(), Some("/tmp/override-wd"));
assert_eq!(
embed.system_prompt.as_deref(),
Some("you are a headless worker")
);
}
}