1use crate::blueprint::{
31 resolve_bound_agents, AgentDef, AgentKind, AgentProfile, Blueprint, BlueprintMetadata,
32 BoundAgent, BoundAgentResolveError, Runner,
33};
34use crate::core::ctx::Ctx;
35use crate::core::engine::Engine;
36use crate::core::projection_placement::{ProjectionPlacement, ProjectionPlacementError};
37use crate::core::step_naming::{StepNaming, StepNamingError};
38use crate::operator::{Operator, OperatorSlotResolver, OperatorSpawner, WorkerBinding};
39use crate::types::{CapToken, StepId};
40use crate::worker::adapter::{InProcSpawner, SpawnError, SpawnerAdapter, WorkerFn};
41use crate::worker::process_spawner::{ProcessSpawner, StreamMode};
42use crate::worker::Worker;
43use async_trait::async_trait;
44use mlua_flow_ir::{Expr, Node as FlowNode, Path};
45use mlua_swarm_schema::{VerdictChannel, VerdictContract};
46use serde_json::Value;
47use std::collections::{BTreeMap, HashMap};
48use std::sync::Arc;
49use thiserror::Error;
50
51#[derive(Debug, Error)]
56pub enum CompileError {
57 #[error("bound agent resolution: {0}")]
59 BoundAgent(#[from] BoundAgentResolveError),
60 #[error("unknown agent kind in SpawnerRegistry: {0:?}")]
63 UnknownKind(AgentKind),
64 #[error("agent '{name}' spec invalid: {msg}")]
67 InvalidSpec {
68 name: String,
70 msg: String,
72 },
73 #[error("flow references agent '{0}' but no AgentDef matches")]
76 UnresolvedRef(String),
77 #[error("duplicate AgentDef name: {0}")]
79 DuplicateAgent(String),
80 #[error("agent '{agent}' operator_ref '{op_ref}' does not match any OperatorDef.name in Blueprint.operators (defined: {defined:?})")]
83 UnresolvedOperatorRef {
84 agent: String,
86 op_ref: String,
88 defined: Vec<String>,
91 },
92 #[error("{where_} names an undefined MetaDef: '{meta_ref}' (defined: {defined:?})")]
96 UnresolvedMetaRef {
97 where_: String,
101 meta_ref: String,
103 defined: Vec<String>,
106 },
107 #[error("StepNaming collision: {0}")]
113 StepNamingCollision(#[from] StepNamingError),
114 #[error("invalid projection_placement: {0}")]
121 InvalidProjectionPlacement(#[from] ProjectionPlacementError),
122 #[error("audits[].agent '{agent}' does not match any AgentDef.name in Blueprint.agents (defined: {defined:?})")]
127 UnresolvedAuditAgent {
128 agent: String,
130 defined: Vec<String>,
133 },
134 #[error(
143 "agent '{agent}' declares verdict channel '{expected_channel}' but {where_} \
144 addresses it as '{actual_shape}' output — see the \"Returning verdicts to drive \
145 BP flow\" guide's Pattern A (channel: \"body\") / Pattern B (channel: \"part\")"
146 )]
147 VerdictChannelMismatch {
148 where_: String,
151 agent: String,
153 expected_channel: String,
155 actual_shape: String,
158 },
159 #[error(
164 "agent '{agent}' verdict Lit '{value}' at {where_} is not a member of the declared \
165 values {values:?}"
166 )]
167 VerdictValueNotInContract {
168 where_: String,
171 agent: String,
174 value: String,
179 values: Vec<String>,
182 },
183 #[error(
195 "agent '{agent}' declares verdict value '{value}' but no downstream Branch/Loop \
196 cond references it (declared: {declared_values:?}, at step '{step_ref}') — either \
197 handle the value downstream or drop it from `verdict.values`"
198 )]
199 VerdictValueUnhandled {
200 agent: String,
203 value: String,
205 declared_values: Vec<String>,
208 step_ref: String,
213 },
214}
215
216pub const WORKER_BINDING_REQUIRED_MSG_PREFIX: &str =
225 "profile.worker_binding is required for this operator backend";
226
227impl From<&CompileError> for mlua_swarm_diag::Diagnostic {
245 fn from(err: &CompileError) -> Self {
246 use mlua_swarm_diag::{
247 Applicability, DiagElement, DiagLevel, DiagSpan, DiagStage, Diagnostic, DocsRef,
248 Suggestion,
249 };
250 let base = |kind: &'static str| {
251 Diagnostic::new(
252 kind,
253 DiagStage::CompileLint,
254 DiagLevel::Error,
255 err.to_string(),
256 )
257 };
258 let agent_span = |name: &str| DiagSpan {
259 element: DiagElement::Agent {
260 name: name.to_string(),
261 },
262 json_path: Some(format!("$.agents[?(@.name=='{name}')]")),
263 };
264 match err {
265 CompileError::BoundAgent(_) => base("bound-agent-resolution"),
266 CompileError::UnknownKind(_) => base("unknown-agent-kind").with_help(
267 "register a SpawnerFactory for this kind, or disable strategy.strict_kind",
268 ),
269 CompileError::InvalidSpec { name, msg }
270 if msg.starts_with(WORKER_BINDING_REQUIRED_MSG_PREFIX) =>
271 {
272 Diagnostic::new(
273 "worker-binding-missing",
274 DiagStage::CompileLint,
275 DiagLevel::Error,
276 format!(
277 "operator agent '{name}' has no explicit Runner or legacy \
278 `profile.worker_binding`"
279 ),
280 )
281 .with_note(msg.clone())
282 .with_suggestion(Suggestion {
283 msg: "add an explicit Runner (or legacy profile.worker_binding)".into(),
284 patch: "runner = { backend = \"ws_operator\", variant = \"claude\", \
285 tools = {} }"
286 .into(),
287 applicability: Applicability::HasPlaceholders,
288 })
289 .with_docs_ref(DocsRef {
290 uri: "mse://guides/bp-dsl-templates",
291 anchor: None,
292 })
293 .with_span(agent_span(name))
294 }
295 CompileError::InvalidSpec { name, .. } => {
296 base("invalid-agent-spec").with_span(agent_span(name))
297 }
298 CompileError::UnresolvedRef(ref_) => base("unresolved-agent-ref").with_span(DiagSpan {
299 element: DiagElement::Step { ref_: ref_.clone() },
300 json_path: None,
301 }),
302 CompileError::DuplicateAgent(name) => {
303 base("duplicate-agent-name").with_span(agent_span(name))
304 }
305 CompileError::UnresolvedOperatorRef { agent, defined, .. } => {
306 base("unresolved-operator-ref")
307 .with_note(format!("declared OperatorDef names: {defined:?}"))
308 .with_span(agent_span(agent))
309 }
310 CompileError::UnresolvedMetaRef { defined, .. } => base("unresolved-meta-ref")
311 .with_note(format!("declared MetaDef names: {defined:?}")),
312 CompileError::StepNamingCollision(_) => base("step-naming-collision"),
313 CompileError::InvalidProjectionPlacement(_) => base("invalid-projection-placement")
314 .with_span(DiagSpan {
315 element: DiagElement::BlueprintRoot,
316 json_path: Some("$.projection_placement".into()),
317 }),
318 CompileError::UnresolvedAuditAgent { defined, .. } => base("unresolved-audit-agent")
319 .with_note(format!("declared AgentDef names: {defined:?}"))
320 .with_span(DiagSpan {
321 element: DiagElement::BlueprintRoot,
322 json_path: Some("$.audits".into()),
323 }),
324 CompileError::VerdictChannelMismatch { agent, .. } => base("verdict-channel-mismatch")
325 .with_help(
326 "see the \"Returning verdicts to drive BP flow\" guide's Pattern A \
327 (channel: \"body\") / Pattern B (channel: \"part\")",
328 )
329 .with_docs_ref(DocsRef {
330 uri: "mse://guides/blueprint-authoring",
331 anchor: None,
332 })
333 .with_span(agent_span(agent)),
334 CompileError::VerdictValueNotInContract { agent, .. } => {
335 base("verdict-value-not-in-contract")
336 .with_suggestion(Suggestion {
342 msg: "align the cond literal with the agent's declared verdict \
343 contract"
344 .into(),
345 patch: "either add the cond's literal to `agents[N].verdict.values`, \
346 or change the cond to a value that is already declared"
347 .into(),
348 applicability: Applicability::MaybeIncorrect,
349 })
350 .with_docs_ref(DocsRef {
351 uri: "mse://guides/blueprint-authoring",
352 anchor: None,
353 })
354 .with_span(agent_span(agent))
355 }
356 CompileError::VerdictValueUnhandled {
357 agent,
358 declared_values,
359 ..
360 } => base("verdict-value-unhandled")
361 .with_note(format!("declared verdict.values: {declared_values:?}"))
362 .with_help(
363 "either handle the value in a downstream Branch/Loop cond, or drop it \
364 from verdict.values",
365 )
366 .with_span(agent_span(agent)),
367 }
368 }
369}
370
371pub trait SpawnerFactory: Send + Sync {
383 fn build(
386 &self,
387 agent_def: &AgentDef,
388 hint: Option<&Value>,
389 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError>;
390}
391
392pub trait SpawnerFactoryKind: SpawnerFactory {
408 const KIND: AgentKind;
411 type Worker: crate::worker::Worker;
418}
419
420#[derive(Clone)]
423pub struct SpawnerRegistry {
424 factories: HashMap<AgentKind, Arc<dyn SpawnerFactory>>,
425}
426
427impl SpawnerRegistry {
428 pub fn new() -> Self {
430 Self {
431 factories: HashMap::new(),
432 }
433 }
434 pub fn register<F: SpawnerFactoryKind + 'static>(&mut self, factory: Arc<F>) -> &mut Self {
443 let f: Arc<dyn SpawnerFactory> = factory;
444 self.factories.insert(F::KIND, f);
445 self
446 }
447}
448
449impl Default for SpawnerRegistry {
450 fn default() -> Self {
451 Self::new()
452 }
453}
454
455pub struct Compiler {
462 registry: SpawnerRegistry,
463 default_spawner: Option<Arc<dyn SpawnerAdapter>>,
464}
465
466pub struct CompiledBlueprint {
470 pub router: Arc<CompiledAgentTable>,
472 pub flow: FlowNode,
474 pub metadata: BlueprintMetadata,
476 pub step_naming: Arc<StepNaming>,
481 pub projection_placement: Arc<ProjectionPlacement>,
487}
488
489fn project_bound_agent_for_legacy_factories(bound: &BoundAgent) -> AgentDef {
490 let mut agent = bound.agent.clone();
491 match &bound.runner {
492 Some(Runner::WsOperator { variant, tools })
493 | Some(Runner::WsClaudeCode { variant, tools }) => {
494 let profile = agent.profile.get_or_insert_with(AgentProfile::default);
495 profile.worker_binding = Some(variant.clone());
496 profile.tools = tools.clone();
497 }
498 Some(Runner::AgentBlockInProcess { tools }) => {
499 let profile = agent.profile.get_or_insert_with(AgentProfile::default);
500 profile.worker_binding = None;
501 profile.tools = tools.clone();
502 }
503 Some(Runner::Subprocess { .. }) => {}
508 None => {}
509 }
510 let meta = agent.meta.get_or_insert_with(Default::default);
511 meta.context_policy = bound.context_policy.clone();
512 agent
513}
514
515pub(crate) fn materialize_bound_blueprint(
518 bp: &Blueprint,
519 bound_agents: &[BoundAgent],
520) -> Blueprint {
521 let mut effective = bp.clone();
522 effective.agents = bound_agents
523 .iter()
524 .map(project_bound_agent_for_legacy_factories)
525 .collect();
526 effective.default_context_policy = None;
529 effective
530}
531
532impl Compiler {
533 pub fn new(registry: SpawnerRegistry) -> Self {
537 Self {
538 registry,
539 default_spawner: None,
540 }
541 }
542
543 pub fn with_default(mut self, sp: Arc<dyn SpawnerAdapter>) -> Self {
547 self.default_spawner = Some(sp);
548 self
549 }
550
551 pub fn compile(&self, bp: &Blueprint) -> Result<CompiledBlueprint, CompileError> {
556 let bound_agents = resolve_bound_agents(bp)?;
557 self.compile_bound(bp, &bound_agents)
558 }
559
560 pub fn compile_bound(
573 &self,
574 bp: &Blueprint,
575 bound_agents: &[BoundAgent],
576 ) -> Result<CompiledBlueprint, CompileError> {
577 let effective = materialize_bound_blueprint(bp, bound_agents);
578 self.compile_resolved(&effective)
579 }
580
581 fn compile_resolved(&self, bp: &Blueprint) -> Result<CompiledBlueprint, CompileError> {
582 let mut routes: HashMap<String, Arc<dyn SpawnerAdapter>> = HashMap::new();
583 let mut seen: HashMap<String, ()> = HashMap::new();
584 let mut verdict_contracts: HashMap<String, VerdictContract> = HashMap::new();
590
591 let defined: Vec<String> = bp.operators.iter().map(|o| o.name.clone()).collect();
597 for ad in &bp.agents {
598 if !matches!(ad.kind, AgentKind::Operator) {
599 continue;
600 }
601 let op_ref = ad.spec.get("operator_ref").and_then(|v| v.as_str());
602 if let Some(op_ref) = op_ref {
603 if !defined.iter().any(|n| n == op_ref) {
604 return Err(CompileError::UnresolvedOperatorRef {
605 agent: ad.name.clone(),
606 op_ref: op_ref.to_string(),
607 defined: defined.clone(),
608 });
609 }
610 }
611 }
613
614 let metas_defined: Vec<String> = bp.metas.iter().map(|m| m.name.clone()).collect();
618 for ad in &bp.agents {
619 let meta_ref = ad.meta.as_ref().and_then(|m| m.meta_ref.as_ref());
620 if let Some(meta_ref) = meta_ref {
621 if !metas_defined.iter().any(|n| n == meta_ref) {
622 return Err(CompileError::UnresolvedMetaRef {
623 where_: format!("AgentMeta.meta_ref of agent '{}'", ad.name),
624 meta_ref: meta_ref.clone(),
625 defined: metas_defined.clone(),
626 });
627 }
628 }
629 }
630 let mut static_step_meta_refs: Vec<(String, String)> = Vec::new();
636 collect_step_meta_refs(&bp.flow, &mut static_step_meta_refs);
637 for (where_, meta_ref) in static_step_meta_refs {
638 if !metas_defined.iter().any(|n| n == &meta_ref) {
639 return Err(CompileError::UnresolvedMetaRef {
640 where_,
641 meta_ref,
642 defined: metas_defined.clone(),
643 });
644 }
645 }
646
647 let agents_defined: Vec<String> = bp.agents.iter().map(|a| a.name.clone()).collect();
652 for audit in &bp.audits {
653 if !agents_defined.iter().any(|n| n == &audit.agent) {
654 return Err(CompileError::UnresolvedAuditAgent {
655 agent: audit.agent.clone(),
656 defined: agents_defined.clone(),
657 });
658 }
659 }
660
661 for ad in &bp.agents {
662 if seen.contains_key(&ad.name) {
663 return Err(CompileError::DuplicateAgent(ad.name.clone()));
664 }
665 seen.insert(ad.name.clone(), ());
666
667 if let Some(contract) = &ad.verdict {
673 verdict_contracts.insert(ad.name.clone(), contract.clone());
674 }
675
676 let factory = match self.registry.factories.get(&ad.kind) {
677 Some(f) => f.clone(),
678 None => {
679 if bp.strategy.strict_kind {
680 return Err(CompileError::UnknownKind(ad.kind.clone()));
681 } else {
682 tracing::warn!(
683 agent = %ad.name,
684 kind = ?ad.kind,
685 "no spawner factory registered for agent kind; \
686 dropping agent from routing table (strict_kind=false)"
687 );
688 continue;
689 }
690 }
691 };
692 let hint = bp.hints.per_agent.get(&ad.name);
693 let subprocess_hint = if ad.kind == AgentKind::Subprocess {
705 resolve_subprocess_template_hint(bp, ad)?
706 } else {
707 None
708 };
709 let spawner = factory.build(ad, subprocess_hint.as_ref().or(hint))?;
710 routes.insert(ad.name.clone(), spawner);
711 }
712
713 let unhandled_gates = resolve_unhandled_verdict_gates(bp);
733 verify_verdict_conds(&bp.flow, &verdict_contracts, &unhandled_gates)?;
734
735 if bp.strategy.strict_refs {
736 verify_refs(&bp.flow, &routes, self.default_spawner.is_some())?;
737 }
738
739 let (step_naming, step_naming_warnings) = StepNaming::from_blueprint(bp)?;
755 for warning in &step_naming_warnings {
756 tracing::warn!(
757 name = %warning.name,
758 first_step_ref = %warning.first_step_ref,
759 second_step_ref = %warning.second_step_ref,
760 "StepNaming: undeclared steps' canonical/alias names collide; \
761 the step whose own ref matches the name keeps it (data-plane priority)"
762 );
763 }
764
765 let projection_placement =
773 ProjectionPlacement::from_spec(bp.projection_placement.as_ref())?;
774
775 let router = Arc::new(CompiledAgentTable {
776 routes,
777 default: self.default_spawner.clone(),
778 verdict_contracts,
779 });
780 Ok(CompiledBlueprint {
781 router,
782 flow: bp.flow.clone(),
783 metadata: bp.metadata.clone(),
784 step_naming: Arc::new(step_naming),
785 projection_placement: Arc::new(projection_placement),
786 })
787 }
788}
789
790fn verify_refs(
793 node: &FlowNode,
794 routes: &HashMap<String, Arc<dyn SpawnerAdapter>>,
795 has_default: bool,
796) -> Result<(), CompileError> {
797 let mut refs: Vec<String> = Vec::new();
798 collect_refs(node, &mut refs);
799 for r in refs {
800 if !routes.contains_key(&r) && !has_default {
801 return Err(CompileError::UnresolvedRef(r));
802 }
803 }
804 Ok(())
805}
806
807fn collect_refs(node: &FlowNode, out: &mut Vec<String>) {
808 match node {
809 FlowNode::Step { ref_, .. } => out.push(ref_.clone()),
810 FlowNode::Seq { children } => {
811 for c in children {
812 collect_refs(c, out);
813 }
814 }
815 FlowNode::Branch { then_, else_, .. } => {
816 collect_refs(then_, out);
817 collect_refs(else_, out);
818 }
819 FlowNode::Fanout { body, .. } => collect_refs(body, out),
820 FlowNode::Loop { body, .. } => collect_refs(body, out),
821 FlowNode::Try { body, catch, .. } => {
822 collect_refs(body, out);
823 collect_refs(catch, out);
824 }
825 FlowNode::Assign { .. } => {} }
827}
828
829fn collect_step_meta_refs(node: &FlowNode, out: &mut Vec<(String, String)>) {
837 match node {
838 FlowNode::Step { ref_, in_, .. } => {
839 if let Expr::Lit { value } = in_ {
840 if let Some(meta_ref) = static_step_meta_ref(value) {
841 out.push((format!("Step '{ref_}' $step_meta.ref"), meta_ref));
842 }
843 }
844 }
845 FlowNode::Seq { children } => {
846 for c in children {
847 collect_step_meta_refs(c, out);
848 }
849 }
850 FlowNode::Branch { then_, else_, .. } => {
851 collect_step_meta_refs(then_, out);
852 collect_step_meta_refs(else_, out);
853 }
854 FlowNode::Fanout { body, .. } => collect_step_meta_refs(body, out),
855 FlowNode::Loop { body, .. } => collect_step_meta_refs(body, out),
856 FlowNode::Try { body, catch, .. } => {
857 collect_step_meta_refs(body, out);
858 collect_step_meta_refs(catch, out);
859 }
860 FlowNode::Assign { .. } => {} }
862}
863
864fn static_step_meta_ref(value: &Value) -> Option<String> {
871 value
872 .as_object()?
873 .get("$step_meta")?
874 .as_object()?
875 .get("ref")?
876 .as_str()
877 .map(str::to_string)
878}
879
880const UNHANDLED_VERDICT_LINT_KIND: &str = "verdict-value-unhandled";
889
890#[derive(Debug, Clone, Copy, PartialEq, Eq)]
892enum UnhandledVerdictGate {
893 Deny,
895 Warn,
897 Silence,
899}
900
901#[derive(Debug, Clone, PartialEq, Eq)]
910struct UnhandledVerdictGates {
911 per_agent: HashMap<String, UnhandledVerdictGate>,
913 blueprint: UnhandledVerdictGate,
915}
916
917impl UnhandledVerdictGates {
918 fn for_agent(&self, agent: &str) -> UnhandledVerdictGate {
921 self.per_agent.get(agent).copied().unwrap_or(self.blueprint)
922 }
923
924 fn all_silent(&self) -> bool {
927 self.blueprint == UnhandledVerdictGate::Silence
928 && self
929 .per_agent
930 .values()
931 .all(|g| *g == UnhandledVerdictGate::Silence)
932 }
933}
934
935fn resolve_unhandled_verdict_gates(bp: &Blueprint) -> UnhandledVerdictGates {
943 let strict = bp.metadata.strict_verdict_handling.unwrap_or(false);
944 let blueprint = resolve_unhandled_verdict_gate(&bp.metadata);
945 let per_agent = bp
946 .agents
947 .iter()
948 .filter_map(|ad| {
949 let declared = declared_unhandled_verdict_setting(&ad.lints)?;
950 Some((
951 ad.name.clone(),
952 unhandled_verdict_gate(strict, Some(declared)),
953 ))
954 })
955 .collect();
956 UnhandledVerdictGates {
957 per_agent,
958 blueprint,
959 }
960}
961
962fn resolve_unhandled_verdict_gate(metadata: &BlueprintMetadata) -> UnhandledVerdictGate {
965 unhandled_verdict_gate(
966 metadata.strict_verdict_handling.unwrap_or(false),
967 declared_unhandled_verdict_setting(&metadata.lints),
968 )
969}
970
971fn declared_unhandled_verdict_setting(
981 lints: &Option<BTreeMap<String, mlua_swarm_schema::LintSetting>>,
982) -> Option<mlua_swarm_diag::LintSetting> {
983 use mlua_swarm_diag::{lint_decl, LintConfig};
984
985 let cfg = LintConfig::from_pairs(
986 lints
987 .as_ref()?
988 .iter()
989 .map(|(key, setting)| (key.clone(), diag_lint_setting(*setting))),
990 );
991 cfg.setting_for(lint_decl(UNHANDLED_VERDICT_LINT_KIND)?)
992}
993
994fn unhandled_verdict_gate(
1002 strict: bool,
1003 declared: Option<mlua_swarm_diag::LintSetting>,
1004) -> UnhandledVerdictGate {
1005 use mlua_swarm_diag::LintSetting;
1006
1007 match declared {
1008 _ if strict => UnhandledVerdictGate::Deny,
1009 Some(LintSetting::Deny) => UnhandledVerdictGate::Deny,
1010 Some(LintSetting::Allow) => UnhandledVerdictGate::Silence,
1011 Some(LintSetting::Warn) | None => UnhandledVerdictGate::Warn,
1012 }
1013}
1014
1015fn diag_lint_setting(setting: mlua_swarm_schema::LintSetting) -> mlua_swarm_diag::LintSetting {
1020 match setting {
1021 mlua_swarm_schema::LintSetting::Allow => mlua_swarm_diag::LintSetting::Allow,
1022 mlua_swarm_schema::LintSetting::Warn => mlua_swarm_diag::LintSetting::Warn,
1023 mlua_swarm_schema::LintSetting::Deny => mlua_swarm_diag::LintSetting::Deny,
1024 }
1025}
1026
1027fn verify_verdict_conds(
1037 flow: &FlowNode,
1038 verdict_contracts: &HashMap<String, VerdictContract>,
1039 unhandled_gates: &UnhandledVerdictGates,
1040) -> Result<(), CompileError> {
1041 let mut step_outputs: HashMap<String, String> = HashMap::new();
1042 let mut step_agents: HashMap<String, String> = HashMap::new();
1043 collect_step_outputs_and_agents(flow, &mut step_outputs, &mut step_agents);
1044
1045 let mut errors: Vec<CompileError> = Vec::new();
1046 let mut referenced_values: HashMap<String, std::collections::HashSet<String>> = HashMap::new();
1047 collect_verdict_conds(
1048 flow,
1049 &step_outputs,
1050 verdict_contracts,
1051 &mut referenced_values,
1052 &mut errors,
1053 );
1054 check_unhandled_verdict_values(
1055 verdict_contracts,
1056 &referenced_values,
1057 &step_agents,
1058 unhandled_gates,
1059 &mut errors,
1060 );
1061 match errors.into_iter().next() {
1062 Some(e) => Err(e),
1063 None => Ok(()),
1064 }
1065}
1066
1067fn collect_step_outputs_and_agents(
1082 node: &FlowNode,
1083 out: &mut HashMap<String, String>,
1084 step_agents: &mut HashMap<String, String>,
1085) {
1086 match node {
1087 FlowNode::Step {
1088 ref_,
1089 out: out_expr,
1090 ..
1091 } => {
1092 if let Expr::Path { at } = out_expr {
1093 out.insert(at.to_string(), ref_.clone());
1094 }
1095 step_agents
1096 .entry(ref_.clone())
1097 .or_insert_with(|| ref_.clone());
1098 }
1099 FlowNode::Seq { children } => {
1100 for c in children {
1101 collect_step_outputs_and_agents(c, out, step_agents);
1102 }
1103 }
1104 FlowNode::Branch { then_, else_, .. } => {
1105 collect_step_outputs_and_agents(then_, out, step_agents);
1106 collect_step_outputs_and_agents(else_, out, step_agents);
1107 }
1108 FlowNode::Fanout { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
1109 FlowNode::Loop { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
1110 FlowNode::Try { body, catch, .. } => {
1111 collect_step_outputs_and_agents(body, out, step_agents);
1112 collect_step_outputs_and_agents(catch, out, step_agents);
1113 }
1114 FlowNode::Assign { .. } => {} }
1116}
1117
1118fn collect_verdict_conds(
1123 node: &FlowNode,
1124 step_outputs: &HashMap<String, String>,
1125 verdict_contracts: &HashMap<String, VerdictContract>,
1126 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
1127 errors: &mut Vec<CompileError>,
1128) {
1129 match node {
1130 FlowNode::Branch { cond, then_, else_ } => {
1131 lint_cond_expr(
1132 cond,
1133 "Branch cond",
1134 step_outputs,
1135 verdict_contracts,
1136 referenced_values,
1137 errors,
1138 );
1139 collect_verdict_conds(
1140 then_,
1141 step_outputs,
1142 verdict_contracts,
1143 referenced_values,
1144 errors,
1145 );
1146 collect_verdict_conds(
1147 else_,
1148 step_outputs,
1149 verdict_contracts,
1150 referenced_values,
1151 errors,
1152 );
1153 }
1154 FlowNode::Loop { cond, body, .. } => {
1155 lint_cond_expr(
1156 cond,
1157 "Loop cond",
1158 step_outputs,
1159 verdict_contracts,
1160 referenced_values,
1161 errors,
1162 );
1163 collect_verdict_conds(
1164 body,
1165 step_outputs,
1166 verdict_contracts,
1167 referenced_values,
1168 errors,
1169 );
1170 }
1171 FlowNode::Seq { children } => {
1172 for c in children {
1173 collect_verdict_conds(
1174 c,
1175 step_outputs,
1176 verdict_contracts,
1177 referenced_values,
1178 errors,
1179 );
1180 }
1181 }
1182 FlowNode::Fanout { body, .. } => collect_verdict_conds(
1183 body,
1184 step_outputs,
1185 verdict_contracts,
1186 referenced_values,
1187 errors,
1188 ),
1189 FlowNode::Try { body, catch, .. } => {
1190 collect_verdict_conds(
1191 body,
1192 step_outputs,
1193 verdict_contracts,
1194 referenced_values,
1195 errors,
1196 );
1197 collect_verdict_conds(
1198 catch,
1199 step_outputs,
1200 verdict_contracts,
1201 referenced_values,
1202 errors,
1203 );
1204 }
1205 FlowNode::Step { .. } | FlowNode::Assign { .. } => {}
1206 }
1207}
1208
1209fn lint_cond_expr(
1218 expr: &Expr,
1219 where_: &str,
1220 step_outputs: &HashMap<String, String>,
1221 verdict_contracts: &HashMap<String, VerdictContract>,
1222 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
1223 errors: &mut Vec<CompileError>,
1224) {
1225 match expr {
1226 Expr::Eq { lhs, rhs } | Expr::Ne { lhs, rhs } => {
1227 if let Some((path, lit)) = path_lit_operands(lhs, rhs) {
1228 resolve_and_check(
1229 path,
1230 &[lit],
1231 where_,
1232 step_outputs,
1233 verdict_contracts,
1234 referenced_values,
1235 errors,
1236 );
1237 }
1238 }
1239 Expr::In { needle, haystack } => {
1240 if let (
1241 Expr::Path { at },
1242 Expr::Lit {
1243 value: Value::Array(items),
1244 },
1245 ) = (needle.as_ref(), haystack.as_ref())
1246 {
1247 let lits: Vec<&Value> = items.iter().collect();
1248 resolve_and_check(
1249 at,
1250 &lits,
1251 where_,
1252 step_outputs,
1253 verdict_contracts,
1254 referenced_values,
1255 errors,
1256 );
1257 }
1258 }
1259 Expr::And { args } | Expr::Or { args } => {
1260 for a in args {
1261 lint_cond_expr(
1262 a,
1263 where_,
1264 step_outputs,
1265 verdict_contracts,
1266 referenced_values,
1267 errors,
1268 );
1269 }
1270 }
1271 Expr::Not { arg } => lint_cond_expr(
1272 arg,
1273 where_,
1274 step_outputs,
1275 verdict_contracts,
1276 referenced_values,
1277 errors,
1278 ),
1279 _ => {}
1280 }
1281}
1282
1283fn path_lit_operands<'a>(lhs: &'a Expr, rhs: &'a Expr) -> Option<(&'a Path, &'a Value)> {
1289 match (lhs, rhs) {
1290 (Expr::Path { at }, Expr::Lit { value }) => Some((at, value)),
1291 (Expr::Lit { value }, Expr::Path { at }) => Some((at, value)),
1292 _ => None,
1293 }
1294}
1295
1296fn resolve_and_check(
1311 path: &Path,
1312 lits: &[&Value],
1313 where_: &str,
1314 step_outputs: &HashMap<String, String>,
1315 verdict_contracts: &HashMap<String, VerdictContract>,
1316 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
1317 errors: &mut Vec<CompileError>,
1318) {
1319 let path_str = path.to_string();
1320 let (agent, actual_shape) = if let Some(agent) = step_outputs.get(&path_str) {
1321 (agent, "body")
1322 } else if let Some(stripped) = path_str.strip_suffix(".parts.verdict") {
1323 match step_outputs.get(stripped) {
1324 Some(agent) => (agent, "part"),
1325 None => return,
1326 }
1327 } else {
1328 return;
1329 };
1330
1331 let Some(contract) = verdict_contracts.get(agent) else {
1332 tracing::warn!(
1333 agent = %agent,
1334 where_ = %where_,
1335 "cond references agent output but no verdict contract declared"
1336 );
1337 return;
1338 };
1339
1340 let expected_channel = match contract.channel {
1341 VerdictChannel::Body => "body",
1342 VerdictChannel::Part => "part",
1343 };
1344 if expected_channel != actual_shape {
1345 errors.push(CompileError::VerdictChannelMismatch {
1346 where_: where_.to_string(),
1347 agent: agent.clone(),
1348 expected_channel: expected_channel.to_string(),
1349 actual_shape: actual_shape.to_string(),
1350 });
1351 return;
1352 }
1353
1354 for lit in lits {
1355 let value_str = lit
1356 .as_str()
1357 .map(str::to_string)
1358 .unwrap_or_else(|| lit.to_string());
1359 if !contract.values.iter().any(|v| v == &value_str) {
1360 errors.push(CompileError::VerdictValueNotInContract {
1361 where_: where_.to_string(),
1362 agent: agent.clone(),
1363 value: value_str.clone(),
1364 values: contract.values.clone(),
1365 });
1366 }
1367 referenced_values
1374 .entry(agent.clone())
1375 .or_default()
1376 .insert(value_str);
1377 }
1378}
1379
1380fn check_unhandled_verdict_values(
1404 verdict_contracts: &HashMap<String, VerdictContract>,
1405 referenced_values: &HashMap<String, std::collections::HashSet<String>>,
1406 step_agents: &HashMap<String, String>,
1407 unhandled_gates: &UnhandledVerdictGates,
1408 errors: &mut Vec<CompileError>,
1409) {
1410 if unhandled_gates.all_silent() {
1411 return;
1412 }
1413 for finding in fold_unhandled_verdict_values(verdict_contracts, referenced_values, step_agents)
1414 {
1415 let gate = unhandled_gates.for_agent(&finding.agent);
1416 match gate {
1417 UnhandledVerdictGate::Deny => errors.push(CompileError::VerdictValueUnhandled {
1418 agent: finding.agent,
1419 value: finding.value,
1420 declared_values: finding.declared_values,
1421 step_ref: finding.step_ref,
1422 }),
1423 UnhandledVerdictGate::Warn => tracing::warn!(
1424 agent = %finding.agent,
1425 value = %finding.value,
1426 step_ref = %finding.step_ref,
1427 "declared verdict value has no downstream cond handler; \
1428 declare `metadata.lints = {{\"verdict-value-unhandled\": \"deny\"}}` \
1429 to reject at compile"
1430 ),
1431 UnhandledVerdictGate::Silence => {}
1435 }
1436 }
1437}
1438
1439#[derive(Debug, Clone, PartialEq, Eq)]
1450pub struct UnhandledVerdictValue {
1451 pub agent: String,
1453 pub value: String,
1455 pub declared_values: Vec<String>,
1457 pub step_ref: String,
1459}
1460
1461pub fn unhandled_verdict_values(
1478 flow: &FlowNode,
1479 verdict_contracts: &HashMap<String, VerdictContract>,
1480) -> Vec<UnhandledVerdictValue> {
1481 let mut step_outputs: HashMap<String, String> = HashMap::new();
1482 let mut step_agents: HashMap<String, String> = HashMap::new();
1483 collect_step_outputs_and_agents(flow, &mut step_outputs, &mut step_agents);
1484
1485 let mut referenced_values: HashMap<String, std::collections::HashSet<String>> = HashMap::new();
1486 let mut discarded_errors: Vec<CompileError> = Vec::new();
1487 collect_verdict_conds(
1488 flow,
1489 &step_outputs,
1490 verdict_contracts,
1491 &mut referenced_values,
1492 &mut discarded_errors,
1493 );
1494 fold_unhandled_verdict_values(verdict_contracts, &referenced_values, &step_agents)
1495}
1496
1497#[derive(Debug, Clone, PartialEq, Eq)]
1511pub struct AgentContractUnread {
1512 pub agent: String,
1514 pub declared_values: Vec<String>,
1516 pub step_ref: String,
1518}
1519
1520pub fn agents_with_all_verdict_values_unread(
1531 flow: &FlowNode,
1532 verdict_contracts: &HashMap<String, VerdictContract>,
1533) -> Vec<AgentContractUnread> {
1534 let per_value = unhandled_verdict_values(flow, verdict_contracts);
1535 let mut unread_counts: HashMap<String, usize> = HashMap::new();
1536 for finding in &per_value {
1537 *unread_counts.entry(finding.agent.clone()).or_default() += 1;
1538 }
1539 let mut agents: Vec<&String> = verdict_contracts.keys().collect();
1540 agents.sort();
1541 let mut out = Vec::new();
1542 for agent in agents {
1543 let contract = &verdict_contracts[agent];
1544 let declared = contract.values.len();
1545 if declared == 0 {
1546 continue;
1547 }
1548 let unread = unread_counts.get(agent).copied().unwrap_or(0);
1549 if unread != declared {
1550 continue;
1551 }
1552 let step_ref = per_value
1556 .iter()
1557 .find(|f| &f.agent == agent)
1558 .map(|f| f.step_ref.clone())
1559 .unwrap_or_else(|| agent.clone());
1560 out.push(AgentContractUnread {
1561 agent: agent.clone(),
1562 declared_values: contract.values.clone(),
1563 step_ref,
1564 });
1565 }
1566 out
1567}
1568
1569fn fold_unhandled_verdict_values(
1580 verdict_contracts: &HashMap<String, VerdictContract>,
1581 referenced_values: &HashMap<String, std::collections::HashSet<String>>,
1582 step_agents: &HashMap<String, String>,
1583) -> Vec<UnhandledVerdictValue> {
1584 let mut agents: Vec<&String> = verdict_contracts.keys().collect();
1585 agents.sort();
1586 let mut findings = Vec::new();
1587 for agent in agents {
1588 let contract = &verdict_contracts[agent];
1589 let referenced = referenced_values.get(agent);
1590 let step_ref = step_agents
1591 .get(agent)
1592 .cloned()
1593 .unwrap_or_else(|| agent.clone());
1594 for value in &contract.values {
1595 let handled = referenced.map(|set| set.contains(value)).unwrap_or(false);
1596 if handled {
1597 continue;
1598 }
1599 findings.push(UnhandledVerdictValue {
1600 agent: agent.clone(),
1601 value: value.clone(),
1602 declared_values: contract.values.clone(),
1603 step_ref: step_ref.clone(),
1604 });
1605 }
1606 }
1607 findings
1608}
1609
1610pub struct CompiledAgentTable {
1623 pub(crate) routes: HashMap<String, Arc<dyn SpawnerAdapter>>,
1624 pub(crate) default: Option<Arc<dyn SpawnerAdapter>>,
1625 pub(crate) verdict_contracts: HashMap<String, VerdictContract>,
1629}
1630
1631impl CompiledAgentTable {
1632 pub fn has_route(&self, agent: &str) -> bool {
1635 self.routes.contains_key(agent)
1636 }
1637 pub fn routed_agents(&self) -> Vec<String> {
1639 self.routes.keys().cloned().collect()
1640 }
1641 pub fn verdict_contract_for(&self, agent: &str) -> Option<&VerdictContract> {
1645 self.verdict_contracts.get(agent)
1646 }
1647}
1648
1649#[async_trait]
1650impl SpawnerAdapter for CompiledAgentTable {
1651 async fn spawn(
1652 &self,
1653 engine: &Engine,
1654 ctx: &Ctx,
1655 task_id: StepId,
1656 attempt: u32,
1657 token: CapToken,
1658 ) -> Result<Box<dyn Worker>, SpawnError> {
1659 let sp = self
1660 .routes
1661 .get(&ctx.agent)
1662 .cloned()
1663 .or_else(|| self.default.clone())
1664 .ok_or_else(|| SpawnError::NotRegistered(ctx.agent.clone()))?;
1665 sp.spawn(engine, ctx, task_id, attempt, token).await
1666 }
1667}
1668
1669pub struct SubprocessProcessSpawnerFactory;
1700
1701impl SpawnerFactoryKind for SubprocessProcessSpawnerFactory {
1702 const KIND: AgentKind = AgentKind::Subprocess;
1703 type Worker = crate::worker::process_spawner::ProcessWorker;
1704}
1705
1706pub const SUBPROCESS_TEMPLATE_HINT_KEY: &str = "subprocess_template";
1709pub const SUBPROCESS_OVERRIDES_HINT_KEY: &str = "subprocess_overrides";
1711
1712fn validate_embed_placeholders(s: &str, where_: &str) -> Result<(), String> {
1726 let mut rest = s;
1727 while let Some(start) = rest.find('{') {
1728 let after = &rest[start + 1..];
1729 let Some(end) = after.find('}') else {
1730 break;
1731 };
1732 let token = &after[..end];
1733 let is_candidate =
1734 !token.is_empty() && token.chars().all(|c| c.is_ascii_lowercase() || c == '_');
1735 if is_candidate {
1736 if !crate::worker::process_spawner::EMBED_PLACEHOLDERS.contains(&token) {
1737 return Err(format!(
1738 "unknown placeholder '{{{token}}}' in {where_}; closed set is \
1739 {{system, system_file, prompt, model, tools_csv, work_dir, task_id, attempt}}"
1740 ));
1741 }
1742 rest = &after[end + 1..];
1743 } else {
1744 rest = after;
1749 }
1750 }
1751 Ok(())
1752}
1753
1754fn resolve_subprocess_template_hint(
1760 bp: &Blueprint,
1761 ad: &AgentDef,
1762) -> Result<Option<Value>, CompileError> {
1763 let invalid = |msg: String| CompileError::InvalidSpec {
1764 name: ad.name.clone(),
1765 msg,
1766 };
1767 let runner = mlua_swarm_schema::resolve_runner(bp, ad).map_err(|e| invalid(e.to_string()))?;
1768 let Some(Runner::Subprocess {
1769 template,
1770 overrides,
1771 }) = runner
1772 else {
1773 return Ok(None);
1774 };
1775 let def = bp
1776 .subprocesses
1777 .iter()
1778 .find(|d| d.name == template)
1779 .ok_or_else(|| {
1780 let mut names: Vec<&str> = bp.subprocesses.iter().map(|d| d.name.as_str()).collect();
1781 names.sort_unstable();
1782 invalid(format!(
1783 "Runner::Subprocess template '{template}' not found in \
1784 Blueprint.subprocesses (defined: [{}])",
1785 names.join(", ")
1786 ))
1787 })?;
1788 Ok(Some(serde_json::json!({
1789 SUBPROCESS_TEMPLATE_HINT_KEY: def,
1790 SUBPROCESS_OVERRIDES_HINT_KEY: overrides,
1791 })))
1792}
1793
1794impl SubprocessProcessSpawnerFactory {
1795 fn build_embed(
1800 agent_def: &AgentDef,
1801 template: &Value,
1802 overrides: Option<&Value>,
1803 ) -> Result<ProcessSpawner, CompileError> {
1804 use crate::worker::process_spawner::EmbedTemplate;
1805 use mlua_swarm_schema::{SubprocessDef, SubprocessOverrides};
1806
1807 let agent_name = &agent_def.name;
1808 let invalid = |msg: String| CompileError::InvalidSpec {
1809 name: agent_name.to_string(),
1810 msg,
1811 };
1812 let def: SubprocessDef = serde_json::from_value(template.clone())
1813 .map_err(|e| invalid(format!("subprocess_template hint: {e}")))?;
1814 let overrides: SubprocessOverrides = match overrides {
1815 Some(v) => serde_json::from_value(v.clone())
1816 .map_err(|e| invalid(format!("subprocess_overrides hint: {e}")))?,
1817 None => SubprocessOverrides::default(),
1818 };
1819
1820 if def.argv.is_empty() {
1821 return Err(invalid(format!(
1822 "SubprocessDef '{}': argv must not be empty",
1823 def.name
1824 )));
1825 }
1826 for (i, a) in def.argv.iter().enumerate() {
1828 validate_embed_placeholders(a, &format!("argv[{i}]")).map_err(&invalid)?;
1829 }
1830 if let Some(stdin) = &def.stdin {
1831 validate_embed_placeholders(stdin, "stdin").map_err(&invalid)?;
1832 }
1833 for (k, v) in &def.env {
1834 validate_embed_placeholders(v, &format!("env['{k}']")).map_err(&invalid)?;
1835 }
1836 if let Some(cwd) = &def.cwd {
1837 validate_embed_placeholders(cwd, "cwd").map_err(&invalid)?;
1838 }
1839 let stream_mode = match def.stream_mode.as_deref() {
1840 Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
1841 Some("sse_events") => Some(StreamMode::SseEvents),
1842 Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
1843 Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
1844 None => None,
1845 };
1846 if let Some(output) = &def.output {
1847 if stream_mode.is_some() {
1848 return Err(invalid(format!(
1849 "SubprocessDef '{}': output normalization is a plain-mode \
1850 declaration; remove either `output` or `stream_mode`",
1851 def.name
1852 )));
1853 }
1854 if let Some(format) = output.format.as_deref() {
1855 if format != "json" {
1856 return Err(invalid(format!(
1857 "SubprocessDef '{}': unknown output.format '{format}' \
1858 (supported: \"json\")",
1859 def.name
1860 )));
1861 }
1862 }
1863 if let Some(ptr) = output.result_ptr.as_deref() {
1864 if !ptr.starts_with('/') {
1865 return Err(invalid(format!(
1866 "SubprocessDef '{}': output.result_ptr '{ptr}' is not a \
1867 JSON Pointer (RFC 6901 — must start with '/')",
1868 def.name
1869 )));
1870 }
1871 }
1872 if let Some(ok_from) = output.ok_from.as_deref() {
1873 if ok_from != "exit_code" && !ok_from.starts_with('/') {
1874 return Err(invalid(format!(
1875 "SubprocessDef '{}': output.ok_from '{ok_from}' must be \
1876 \"exit_code\" or a JSON Pointer (starting with '/')",
1877 def.name
1878 )));
1879 }
1880 }
1881 }
1882
1883 let profile = agent_def.profile.as_ref();
1886 let system_prompt = profile
1887 .map(|p| p.system_prompt.clone())
1888 .filter(|s| !s.is_empty());
1889 let model = overrides
1890 .model
1891 .clone()
1892 .or_else(|| profile.and_then(|p| p.model.clone()));
1893 let tools: Vec<String> = if overrides.tools.is_empty() {
1894 profile.map(|p| p.tools.clone()).unwrap_or_default()
1895 } else {
1896 overrides.tools.clone()
1897 };
1898 let cwd = overrides.cwd.clone().or_else(|| def.cwd.clone());
1900 if let Some(c) = &cwd {
1901 validate_embed_placeholders(c, "overrides.cwd").map_err(&invalid)?;
1902 }
1903
1904 let program = def.argv[0].clone();
1905 let sp = ProcessSpawner {
1906 program,
1907 args: Vec::new(),
1908 use_stdin: def.stdin.is_some(),
1909 stream_mode,
1910 embed: Some(EmbedTemplate {
1911 argv: def.argv,
1912 stdin: def.stdin,
1913 env: def.env,
1914 cwd,
1915 output: def.output,
1916 system_prompt,
1917 model,
1918 tools_csv: tools.join(","),
1919 }),
1920 };
1921 Ok(sp)
1922 }
1923}
1924
1925impl SpawnerFactory for SubprocessProcessSpawnerFactory {
1926 fn build(
1927 &self,
1928 agent_def: &AgentDef,
1929 hint: Option<&Value>,
1930 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1931 if let Some(template) = hint.and_then(|h| h.get(SUBPROCESS_TEMPLATE_HINT_KEY)) {
1935 let overrides = hint.and_then(|h| h.get(SUBPROCESS_OVERRIDES_HINT_KEY));
1936 return Self::build_embed(agent_def, template, overrides).map(|sp| {
1937 let arc: Arc<dyn SpawnerAdapter> = Arc::new(sp);
1938 arc
1939 });
1940 }
1941 let agent_name = &agent_def.name;
1942 let spec = &agent_def.spec;
1943 let invalid = |msg: String| CompileError::InvalidSpec {
1944 name: agent_name.to_string(),
1945 msg,
1946 };
1947 let program = spec
1948 .get("program")
1949 .and_then(|v| v.as_str())
1950 .ok_or_else(|| invalid("shell spec: 'program' (string) required".into()))?
1951 .to_string();
1952 let args: Vec<String> = spec
1953 .get("args")
1954 .and_then(|v| v.as_array())
1955 .map(|a| {
1956 a.iter()
1957 .filter_map(|x| x.as_str().map(|s| s.to_string()))
1958 .collect()
1959 })
1960 .unwrap_or_default();
1961 let use_stdin = spec
1962 .get("use_stdin")
1963 .and_then(|v| v.as_bool())
1964 .unwrap_or(true);
1965 let stream_mode = match spec.get("stream_mode").and_then(|v| v.as_str()) {
1966 Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
1967 Some("sse_events") => Some(StreamMode::SseEvents),
1968 Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
1969 Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
1970 None => None,
1971 };
1972
1973 let mut sp = ProcessSpawner {
1974 program,
1975 args,
1976 use_stdin,
1977 stream_mode,
1978 embed: None,
1979 };
1980 if let Some(mode) = sp.stream_mode.clone() {
1981 sp = sp.stream_mode(mode);
1982 }
1983 Ok(Arc::new(sp))
1984 }
1985}
1986
1987pub struct LuaInProcessSpawnerFactory {
2018 registry: HashMap<String, WorkerFn>,
2019 bridges: HashMap<String, HostBridge>,
2020}
2021
2022#[derive(Clone)]
2034pub struct HostBridge(
2035 Arc<dyn Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync>,
2036);
2037
2038impl HostBridge {
2039 pub fn new<F>(f: F) -> Self
2041 where
2042 F: Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync + 'static,
2043 {
2044 Self(Arc::new(f))
2045 }
2046
2047 pub fn call(&self, arg: serde_json::Value) -> Result<serde_json::Value, String> {
2051 (self.0)(arg)
2052 }
2053}
2054
2055#[derive(Clone)]
2062pub struct LuaScriptSource {
2063 pub source: String,
2065 pub label: String,
2068}
2069
2070impl LuaScriptSource {
2071 pub fn new(source: impl Into<String>, label: impl Into<String>) -> Self {
2073 Self {
2074 source: source.into(),
2075 label: label.into(),
2076 }
2077 }
2078}
2079
2080impl LuaInProcessSpawnerFactory {
2081 pub fn new() -> Self {
2083 Self {
2084 registry: HashMap::new(),
2085 bridges: HashMap::new(),
2086 }
2087 }
2088
2089 pub fn with_bridge(mut self, name: impl Into<String>, bridge: HostBridge) -> Self {
2096 self.bridges.insert(name.into(), bridge);
2097 self
2098 }
2099
2100 pub fn register_lua(mut self, fn_id: impl Into<String>, source: LuaScriptSource) -> Self {
2118 let source = Arc::new(source);
2119 let bridges = Arc::new(self.bridges.clone());
2120 let wrapped: WorkerFn = Arc::new(move |inv| {
2121 let source = source.clone();
2122 let bridges = bridges.clone();
2123 Box::pin(run_lua_worker(source, bridges, inv))
2124 });
2125 self.registry.insert(fn_id.into(), wrapped);
2126 self
2127 }
2128}
2129
2130async fn run_lua_worker(
2132 source: Arc<LuaScriptSource>,
2133 bridges: Arc<HashMap<String, HostBridge>>,
2134 inv: crate::worker::adapter::WorkerInvocation,
2135) -> Result<crate::worker::adapter::WorkerResult, crate::worker::adapter::WorkerError> {
2136 use crate::worker::adapter::WorkerError;
2137 use mlua::LuaSerdeExt;
2138
2139 let label = source.label.clone();
2140 let outcome =
2141 tokio::task::spawn_blocking(move || -> Result<(serde_json::Value, bool), String> {
2142 let lua = mlua::Lua::new();
2143 let g = lua.globals();
2144
2145 g.set("_PROMPT", inv.prompt.clone())
2147 .map_err(|e| format!("set _PROMPT: {e}"))?;
2148 g.set("_AGENT", inv.agent.clone())
2149 .map_err(|e| format!("set _AGENT: {e}"))?;
2150 g.set("_TASK_ID", inv.task_id.to_string())
2151 .map_err(|e| format!("set _TASK_ID: {e}"))?;
2152 g.set("_ATTEMPT", inv.attempt as i64)
2153 .map_err(|e| format!("set _ATTEMPT: {e}"))?;
2154
2155 for (name, value) in
2164 crate::worker::agent_block::runtime::context_globals(inv.context.as_ref())
2165 {
2166 let lua_val = lua
2167 .to_value(&value)
2168 .map_err(|e| format!("{name} to_value: {e}"))?;
2169 g.set(name.as_str(), lua_val)
2170 .map_err(|e| format!("set {name}: {e}"))?;
2171 }
2172
2173 if let Ok(json_val) = serde_json::from_str::<serde_json::Value>(&inv.prompt) {
2175 let lua_val = lua
2176 .to_value(&json_val)
2177 .map_err(|e| format!("_CTX to_value: {e}"))?;
2178 g.set("_CTX", lua_val)
2179 .map_err(|e| format!("set _CTX: {e}"))?;
2180 }
2181
2182 if !bridges.is_empty() {
2184 let host = lua
2185 .create_table()
2186 .map_err(|e| format!("create host table: {e}"))?;
2187 for (name, bridge) in bridges.iter() {
2188 let bridge = bridge.clone();
2189 let bname = name.clone();
2190 let f = lua
2191 .create_function(move |lua, arg: mlua::Value| {
2192 let json_arg: serde_json::Value = lua.from_value(arg).map_err(|e| {
2193 mlua::Error::external(format!("bridge {bname} arg → json: {e}"))
2194 })?;
2195 let result_json =
2196 bridge.call(json_arg).map_err(mlua::Error::external)?;
2197 lua.to_value(&result_json).map_err(|e| {
2198 mlua::Error::external(format!("bridge {bname} ret → lua: {e}"))
2199 })
2200 })
2201 .map_err(|e| format!("create_function {name}: {e}"))?;
2202 host.set(name.as_str(), f)
2203 .map_err(|e| format!("host.{name} set: {e}"))?;
2204 }
2205 g.set("host", host).map_err(|e| format!("set host: {e}"))?;
2206 }
2207
2208 let result: mlua::Value = lua
2210 .load(&source.source)
2211 .set_name(&source.label)
2212 .eval()
2213 .map_err(|e| format!("lua eval [{}]: {e}", source.label))?;
2214
2215 let json_result: serde_json::Value = lua
2217 .from_value(result)
2218 .map_err(|e| format!("lua → json [{}]: {e}", source.label))?;
2219
2220 let (value, ok) = match &json_result {
2221 serde_json::Value::Object(map)
2222 if map.contains_key("value") || map.contains_key("ok") =>
2223 {
2224 let ok = map.get("ok").and_then(|v| v.as_bool()).unwrap_or(true);
2225 let value = map.get("value").cloned().unwrap_or(json_result.clone());
2226 (value, ok)
2227 }
2228 _ => (json_result, true),
2229 };
2230 Ok((value, ok))
2231 })
2232 .await
2233 .map_err(|e| WorkerError::Failed(format!("spawn_blocking join [{label}]: {e}")))?
2234 .map_err(WorkerError::Failed)?;
2235
2236 Ok(crate::worker::adapter::WorkerResult {
2237 value: outcome.0,
2238 ok: outcome.1,
2239 stats: None,
2240 }
2241 .ensure_worker_kind("lua"))
2242}
2243
2244impl Default for LuaInProcessSpawnerFactory {
2245 fn default() -> Self {
2246 Self::new()
2247 }
2248}
2249
2250impl SpawnerFactoryKind for LuaInProcessSpawnerFactory {
2251 const KIND: AgentKind = AgentKind::Lua;
2252 type Worker = LuaWorker;
2253}
2254
2255impl SpawnerFactory for LuaInProcessSpawnerFactory {
2256 fn build(
2257 &self,
2258 agent_def: &AgentDef,
2259 _hint: Option<&Value>,
2260 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2261 if let Some(source) = agent_def.spec.get("source").and_then(|v| v.as_str()) {
2267 let label = agent_def
2268 .spec
2269 .get("label")
2270 .and_then(|v| v.as_str())
2271 .map(str::to_string)
2272 .unwrap_or_else(|| format!("{}.lua", agent_def.name));
2273 let script = Arc::new(LuaScriptSource::new(source.to_string(), label));
2274 let bridges = Arc::new(self.bridges.clone());
2275 let wrapped: WorkerFn = Arc::new(move |inv| {
2276 let source = script.clone();
2277 let bridges = bridges.clone();
2278 Box::pin(run_lua_worker(source, bridges, inv))
2279 });
2280 let mut sp: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
2281 sp.registry.insert(agent_def.name.to_string(), wrapped);
2282 return Ok(Arc::new(sp));
2283 }
2284 build_inproc_from_registry::<LuaWorker>(&self.registry, agent_def, "lua")
2285 }
2286}
2287
2288pub struct RustFnInProcessSpawnerFactory {
2302 registry: HashMap<String, WorkerFn>,
2303}
2304
2305impl RustFnInProcessSpawnerFactory {
2306 pub fn new() -> Self {
2308 Self {
2309 registry: HashMap::new(),
2310 }
2311 }
2312
2313 pub fn register_fn<F, Fut>(mut self, fn_id: impl Into<String>, f: F) -> Self
2316 where
2317 F: Fn(crate::worker::adapter::WorkerInvocation) -> Fut + Send + Sync + 'static,
2318 Fut: std::future::Future<
2319 Output = Result<
2320 crate::worker::adapter::WorkerResult,
2321 crate::worker::adapter::WorkerError,
2322 >,
2323 > + Send
2324 + 'static,
2325 {
2326 let f = Arc::new(f);
2327 let wrapped: WorkerFn = Arc::new(move |inv| {
2328 let f = f.clone();
2329 Box::pin(f(inv))
2330 });
2331 self.registry.insert(fn_id.into(), wrapped);
2332 self
2333 }
2334}
2335
2336impl Default for RustFnInProcessSpawnerFactory {
2337 fn default() -> Self {
2338 Self::new()
2339 }
2340}
2341
2342impl SpawnerFactoryKind for RustFnInProcessSpawnerFactory {
2343 const KIND: AgentKind = AgentKind::RustFn;
2344 type Worker = RustFnWorker;
2345}
2346
2347impl SpawnerFactory for RustFnInProcessSpawnerFactory {
2348 fn build(
2349 &self,
2350 agent_def: &AgentDef,
2351 _hint: Option<&Value>,
2352 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2353 build_inproc_from_registry::<RustFnWorker>(&self.registry, agent_def, "rust_fn")
2354 }
2355}
2356
2357fn build_inproc_from_registry<W>(
2363 registry: &HashMap<String, WorkerFn>,
2364 agent_def: &AgentDef,
2365 kind_label: &str,
2366) -> Result<Arc<dyn SpawnerAdapter>, CompileError>
2367where
2368 W: crate::worker::Worker + From<crate::worker::WorkerJoinHandler> + Send + Sync + 'static,
2369{
2370 let agent_name = &agent_def.name;
2371 let spec = &agent_def.spec;
2372 let invalid = |msg: String| CompileError::InvalidSpec {
2373 name: agent_name.to_string(),
2374 msg,
2375 };
2376 let fn_id = spec
2377 .get("fn_id")
2378 .and_then(|v| v.as_str())
2379 .ok_or_else(|| invalid(format!("{kind_label} spec: 'fn_id' (string) required")))?;
2380 let f = registry
2381 .get(fn_id)
2382 .cloned()
2383 .ok_or_else(|| invalid(format!("fn_id '{fn_id}' not registered in factory")))?;
2384 let mut sp: InProcSpawner<W> = InProcSpawner::<W>::typed();
2385 sp.registry.insert(agent_name.to_string(), f);
2389 Ok(Arc::new(sp))
2390}
2391
2392pub struct LuaWorker {
2397 pub handler: crate::worker::WorkerJoinHandler,
2399}
2400
2401impl From<crate::worker::WorkerJoinHandler> for LuaWorker {
2402 fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
2403 Self { handler }
2404 }
2405}
2406
2407#[async_trait::async_trait]
2408impl crate::worker::Worker for LuaWorker {
2409 fn id(&self) -> &crate::types::WorkerId {
2410 &self.handler.worker_id
2411 }
2412 fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
2413 self.handler.cancel.clone()
2414 }
2415 async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
2416 self.handler.await_completion().await
2417 }
2418}
2419
2420pub struct RustFnWorker {
2425 pub handler: crate::worker::WorkerJoinHandler,
2427}
2428
2429impl From<crate::worker::WorkerJoinHandler> for RustFnWorker {
2430 fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
2431 Self { handler }
2432 }
2433}
2434
2435#[async_trait::async_trait]
2436impl crate::worker::Worker for RustFnWorker {
2437 fn id(&self) -> &crate::types::WorkerId {
2438 &self.handler.worker_id
2439 }
2440 fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
2441 self.handler.cancel.clone()
2442 }
2443 async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
2444 self.handler.await_completion().await
2445 }
2446}
2447
2448pub struct OperatorSpawnerFactory {
2521 operators: Arc<std::sync::RwLock<HashMap<String, Arc<dyn Operator>>>>,
2522 slot_resolver: Arc<std::sync::RwLock<Option<Arc<dyn OperatorSlotResolver>>>>,
2525}
2526
2527impl OperatorSpawnerFactory {
2528 pub fn new() -> Self {
2530 Self {
2531 operators: Arc::new(std::sync::RwLock::new(HashMap::new())),
2532 slot_resolver: Arc::new(std::sync::RwLock::new(None)),
2533 }
2534 }
2535
2536 pub fn set_slot_resolver(&self, resolver: Arc<dyn OperatorSlotResolver>) -> &Self {
2545 *self
2546 .slot_resolver
2547 .write()
2548 .expect("OperatorSpawnerFactory.slot_resolver RwLock poisoned") = Some(resolver);
2549 self
2550 }
2551
2552 pub fn resolve_operator(
2560 &self,
2561 slot: &str,
2562 agent: &str,
2563 ) -> Result<Arc<dyn Operator>, CompileError> {
2564 let invalid = |msg: String| CompileError::InvalidSpec {
2565 name: agent.to_string(),
2566 msg,
2567 };
2568 let resolver = self
2569 .slot_resolver
2570 .read()
2571 .expect("OperatorSpawnerFactory.slot_resolver RwLock poisoned")
2572 .clone();
2573 if let Some(resolver) = resolver {
2574 return resolver.resolve(slot).ok_or_else(|| {
2575 invalid(format!(
2576 "operator_ref '{slot}': the installed OperatorSlotResolver serves no such \
2577 Operator seat. The seat is declared by Blueprint.operators[]; nothing is \
2578 resolved from the factory's own registry here, because falling back to it \
2579 would dispatch this agent to a backend the seat does not name."
2580 ))
2581 });
2582 }
2583 let operators = self
2584 .operators
2585 .read()
2586 .expect("OperatorSpawnerFactory.operators RwLock poisoned");
2587 operators.get(slot).cloned().ok_or_else(|| {
2588 let mut names: Vec<String> = operators.keys().cloned().collect();
2589 names.sort();
2590 let names_list = if names.is_empty() {
2591 "<none>".to_string()
2592 } else {
2593 names.join(", ")
2594 };
2595 invalid(format!(
2596 "operator_ref '{slot}' not registered in factory. \
2597 Registered sids: [{names_list}]. \
2598 Hint: call mse_operator_join(roles=[...]) to mint the sid first."
2599 ))
2600 })
2601 }
2602
2603 pub fn register_operator(&self, id: impl Into<String>, op: Arc<dyn Operator>) -> &Self {
2609 self.operators
2610 .write()
2611 .expect("OperatorSpawnerFactory.operators RwLock poisoned")
2612 .insert(id.into(), op);
2613 self
2614 }
2615
2616 pub fn unregister_operator(&self, id: &str) -> &Self {
2619 self.operators
2620 .write()
2621 .expect("OperatorSpawnerFactory.operators RwLock poisoned")
2622 .remove(id);
2623 self
2624 }
2625}
2626
2627impl Default for OperatorSpawnerFactory {
2628 fn default() -> Self {
2629 Self::new()
2630 }
2631}
2632
2633impl SpawnerFactoryKind for OperatorSpawnerFactory {
2634 const KIND: AgentKind = AgentKind::Operator;
2635 type Worker = crate::operator::OperatorWorker;
2636}
2637
2638impl SpawnerFactory for OperatorSpawnerFactory {
2639 fn build(
2644 &self,
2645 agent_def: &AgentDef,
2646 _hint: Option<&Value>,
2647 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2648 let agent_name = &agent_def.name;
2649 let spec = &agent_def.spec;
2650 let system_prompt = agent_def.profile.as_ref().map(|p| p.system_prompt.clone());
2656 let invalid = |msg: String| CompileError::InvalidSpec {
2657 name: agent_name.to_string(),
2658 msg,
2659 };
2660 let op_ref = spec
2661 .get("operator_ref")
2662 .and_then(|v| v.as_str())
2663 .ok_or_else(|| invalid("operator spec: 'operator_ref' (string) required".into()))?;
2664 let op = self.resolve_operator(op_ref, agent_name)?;
2668
2669 let worker_binding = agent_def
2676 .profile
2677 .as_ref()
2678 .and_then(|p| p.worker_binding.as_ref())
2679 .map(|variant| WorkerBinding {
2680 variant: variant.clone(),
2681 tools: agent_def
2682 .profile
2683 .as_ref()
2684 .map(|p| p.tools.clone())
2685 .unwrap_or_default(),
2686 request_digest: None,
2690 requested_model: None,
2691 });
2692 if op.requires_worker_binding() && worker_binding.is_none() {
2693 return Err(invalid(format!(
2700 "{WORKER_BINDING_REQUIRED_MSG_PREFIX}. \
2701 Fix by either: \
2702 (a) if authoring the Blueprint JSON directly, add \
2703 `agents[N].profile.worker_binding: \"<subagent-type>\"` \
2704 to the JSON literal; or \
2705 (b) if using an $agent_md file ref, add \
2706 `worker_binding: <subagent-type>` to the agent .md frontmatter."
2707 )));
2708 }
2709 Ok(Arc::new(OperatorSpawner::new(
2710 op,
2711 system_prompt,
2712 worker_binding,
2713 )))
2714 }
2715}
2716
2717#[cfg(test)]
2718mod operator_spawner_factory_worker_binding_tests {
2719 use super::*;
2720 use crate::blueprint::AgentProfile;
2721 use crate::core::ctx::Ctx;
2722 use crate::types::CapToken;
2723 use crate::worker::adapter::{WorkerError, WorkerResult};
2724
2725 struct StubOperator {
2730 requires_binding: bool,
2731 }
2732
2733 #[async_trait]
2734 impl Operator for StubOperator {
2735 async fn execute(
2736 &self,
2737 _ctx: &Ctx,
2738 _system: Option<String>,
2739 _prompt: Value,
2740 _worker: Option<WorkerBinding>,
2741 _worker_token: CapToken,
2742 ) -> Result<WorkerResult, WorkerError> {
2743 Ok(WorkerResult {
2744 value: Value::Null,
2745 ok: true,
2746 stats: None,
2747 })
2748 }
2749
2750 fn requires_worker_binding(&self) -> bool {
2751 self.requires_binding
2752 }
2753 }
2754
2755 fn agent_def_with(profile: Option<AgentProfile>) -> AgentDef {
2756 AgentDef {
2757 name: "test-agent".to_string(),
2758 kind: AgentKind::Operator,
2759 spec: serde_json::json!({ "operator_ref": "op1" }),
2760 profile,
2761 meta: None,
2762 runner: None,
2763 runner_ref: None,
2764 verdict: None,
2765 lints: None,
2766 }
2767 }
2768
2769 #[test]
2770 fn build_fails_loud_when_binding_required_but_absent() {
2771 let factory = OperatorSpawnerFactory::new();
2772 factory.register_operator(
2773 "op1",
2774 Arc::new(StubOperator {
2775 requires_binding: true,
2776 }) as Arc<dyn Operator>,
2777 );
2778 let def = agent_def_with(Some(AgentProfile::default()));
2779 match factory.build(&def, None) {
2780 Err(CompileError::InvalidSpec { name, msg }) => {
2781 assert_eq!(name, "test-agent");
2782 assert!(
2783 msg.contains("worker_binding is required"),
2784 "unexpected message: {msg}"
2785 );
2786 assert!(
2790 msg.contains("agents[N].profile.worker_binding"),
2791 "message missing JSON-direct hint (issue #9): {msg}"
2792 );
2793 assert!(
2794 msg.contains("agent .md frontmatter"),
2795 "message missing $agent_md hint: {msg}"
2796 );
2797 }
2798 Err(other) => panic!("expected InvalidSpec, got: {other:?}"),
2799 Ok(_) => panic!("expected compile-time failure, got Ok"),
2800 }
2801 }
2802
2803 #[test]
2810 fn factory_error_message_carries_the_shared_prefix_and_specializes_the_diagnostic() {
2811 let factory = OperatorSpawnerFactory::new();
2812 factory.register_operator(
2813 "op1",
2814 Arc::new(StubOperator {
2815 requires_binding: true,
2816 }) as Arc<dyn Operator>,
2817 );
2818 let def = agent_def_with(Some(AgentProfile::default()));
2819 let err = match factory.build(&def, None) {
2820 Err(err) => err,
2821 Ok(_) => panic!("expected compile-time failure, got Ok"),
2822 };
2823 match &err {
2824 CompileError::InvalidSpec { msg, .. } => {
2825 assert!(
2826 msg.starts_with(WORKER_BINDING_REQUIRED_MSG_PREFIX),
2827 "factory message must start with the shared prefix, got: {msg}"
2828 );
2829 }
2830 other => panic!("expected InvalidSpec, got: {other:?}"),
2831 }
2832 let d = mlua_swarm_diag::Diagnostic::from(&err);
2833 assert_eq!(d.kind, "worker-binding-missing");
2834 }
2835
2836 #[test]
2837 fn build_succeeds_when_binding_required_and_present() {
2838 let factory = OperatorSpawnerFactory::new();
2839 factory.register_operator(
2840 "op1",
2841 Arc::new(StubOperator {
2842 requires_binding: true,
2843 }) as Arc<dyn Operator>,
2844 );
2845 let profile = AgentProfile {
2846 worker_binding: Some("code-worker".to_string()),
2847 tools: vec!["Read".to_string(), "Edit".to_string()],
2848 ..Default::default()
2849 };
2850 let def = agent_def_with(Some(profile));
2851 assert!(
2852 factory.build(&def, None).is_ok(),
2853 "expected Ok when worker_binding is declared"
2854 );
2855 }
2856
2857 #[test]
2858 fn build_succeeds_when_binding_not_required_and_absent() {
2859 let factory = OperatorSpawnerFactory::new();
2860 factory.register_operator(
2861 "op1",
2862 Arc::new(StubOperator {
2863 requires_binding: false,
2864 }) as Arc<dyn Operator>,
2865 );
2866 let def = agent_def_with(Some(AgentProfile::default()));
2867 assert!(
2868 factory.build(&def, None).is_ok(),
2869 "backends that don't require a binding must not be gated by its absence"
2870 );
2871 }
2872}
2873
2874#[cfg(test)]
2882mod lua_inline_source_tests {
2883 use super::*;
2884 use crate::types::{CapToken, Role, StepId};
2885
2886 fn agent(name: &str, spec: Value) -> AgentDef {
2887 AgentDef {
2888 name: name.to_string(),
2889 kind: AgentKind::Lua,
2890 spec,
2891 profile: None,
2892 meta: None,
2893 runner: None,
2894 runner_ref: None,
2895 verdict: None,
2896 lints: None,
2897 }
2898 }
2899
2900 fn test_invocation(prompt: &str) -> crate::worker::adapter::WorkerInvocation {
2901 crate::worker::adapter::WorkerInvocation::new(
2902 CapToken {
2903 agent_id: "a".into(),
2904 role: Role::Worker,
2905 scopes: vec!["*".into()],
2906 issued_at: 0,
2907 expire_at: u64::MAX / 2,
2908 max_uses: None,
2909 nonce: "test-nonce".into(),
2910 sig_hex: "".into(),
2911 },
2912 StepId::parse("ST-test").expect("StepId parse"),
2913 1,
2914 "g",
2915 prompt,
2916 )
2917 }
2918
2919 #[test]
2920 fn build_accepts_inline_source_without_pre_registration() {
2921 let factory = LuaInProcessSpawnerFactory::new();
2922 let def = agent(
2923 "g",
2924 serde_json::json!({ "source": "return { value = 42, ok = true }" }),
2925 );
2926 assert!(
2927 factory.build(&def, None).is_ok(),
2928 "inline spec.source must build without a pre-registered fn_id"
2929 );
2930 }
2931
2932 #[test]
2933 fn build_rejects_when_neither_source_nor_fn_id_is_present() {
2934 let factory = LuaInProcessSpawnerFactory::new();
2935 let def = agent("g", serde_json::json!({}));
2936 match factory.build(&def, None) {
2937 Err(CompileError::InvalidSpec { msg, .. }) => {
2938 assert!(
2939 msg.contains("fn_id"),
2940 "empty spec must still surface the fn_id-required message: {msg}"
2941 );
2942 }
2943 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
2944 Ok(_) => panic!("expected InvalidSpec, got Ok(SpawnerAdapter)"),
2947 }
2948 }
2949
2950 #[tokio::test]
2954 async fn inline_source_evaluates_and_marshals_result() {
2955 let source =
2956 LuaScriptSource::new("return { value = _PROMPT .. '!', ok = true }", "smoke.lua");
2957 let out = run_lua_worker(
2958 std::sync::Arc::new(source),
2959 std::sync::Arc::new(HashMap::new()),
2960 test_invocation("hello"),
2961 )
2962 .await
2963 .expect("lua worker ok");
2964 assert_eq!(out.value, serde_json::json!("hello!"));
2965 assert!(out.ok);
2966 }
2967
2968 #[tokio::test]
2969 async fn inline_source_can_signal_agent_level_failure() {
2970 let source = LuaScriptSource::new("return { value = 'nope', ok = false }", "gate.lua");
2973 let out = run_lua_worker(
2974 std::sync::Arc::new(source),
2975 std::sync::Arc::new(HashMap::new()),
2976 test_invocation("input"),
2977 )
2978 .await
2979 .expect("lua worker ok");
2980 assert_eq!(out.value, serde_json::json!("nope"));
2981 assert!(!out.ok);
2982 }
2983}
2984
2985#[cfg(test)]
2988mod meta_ref_validation_tests {
2989 use super::*;
2990 use crate::blueprint::{AgentMeta, MetaDef};
2991 use crate::worker::adapter::WorkerResult;
2992
2993 fn registry_with_echo() -> SpawnerRegistry {
2994 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2995 Ok(WorkerResult {
2996 value: Value::String(inv.prompt),
2997 ok: true,
2998 stats: None,
2999 })
3000 });
3001 let mut reg = SpawnerRegistry::new();
3002 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
3003 reg
3004 }
3005
3006 fn rustfn_agent(name: &str) -> AgentDef {
3007 AgentDef {
3008 name: name.to_string(),
3009 kind: AgentKind::RustFn,
3010 spec: serde_json::json!({ "fn_id": "echo" }),
3011 profile: None,
3012 meta: None,
3013 runner: None,
3014 runner_ref: None,
3015 verdict: None,
3016 lints: None,
3017 }
3018 }
3019
3020 fn simple_flow(agent_ref: &str, in_: Expr) -> FlowNode {
3021 FlowNode::Step {
3022 ref_: agent_ref.to_string(),
3023 in_,
3024 out: Expr::Path {
3025 at: "$.output".parse().expect("literal test path: $.output"),
3026 },
3027 }
3028 }
3029
3030 fn minimal_bp(agents: Vec<AgentDef>, metas: Vec<MetaDef>, flow: FlowNode) -> Blueprint {
3031 Blueprint {
3032 schema_version: crate::blueprint::current_schema_version(),
3033 id: "meta-ref-ut".into(),
3034 flow,
3035 agents,
3036 operators: vec![],
3037 metas,
3038 hints: Default::default(),
3039 strategy: Default::default(),
3040 metadata: BlueprintMetadata::default(),
3041 spawner_hints: Default::default(),
3042 default_agent_kind: AgentKind::Operator,
3043 default_operator_kind: None,
3044 default_init_ctx: None,
3045 default_agent_ctx: None,
3046 default_context_policy: None,
3047 projection_placement: None,
3048 audits: vec![],
3049 degradation_policy: None,
3050 runners: vec![],
3051 default_runner: None,
3052 subprocesses: vec![],
3053 check_policy: None,
3054 blueprint_ref_includes: Vec::new(),
3055 }
3056 }
3057
3058 #[test]
3059 fn valid_meta_ref_compiles() {
3060 let mut agent = rustfn_agent("worker");
3061 agent.meta = Some(AgentMeta {
3062 meta_ref: Some("shared".to_string()),
3063 ..Default::default()
3064 });
3065 let bp = minimal_bp(
3066 vec![agent],
3067 vec![MetaDef {
3068 name: "shared".into(),
3069 ctx: serde_json::json!({ "k": "v" }),
3070 }],
3071 simple_flow(
3072 "worker",
3073 Expr::Path {
3074 at: "$.input".parse().expect("literal test path: $.input"),
3075 },
3076 ),
3077 );
3078 let compiler = Compiler::new(registry_with_echo());
3079 assert!(
3080 compiler.compile(&bp).is_ok(),
3081 "a resolvable AgentMeta.meta_ref must compile"
3082 );
3083 }
3084
3085 #[test]
3086 fn unknown_agent_meta_ref_is_unresolved_meta_ref() {
3087 let mut agent = rustfn_agent("worker");
3088 agent.meta = Some(AgentMeta {
3089 meta_ref: Some("missing".to_string()),
3090 ..Default::default()
3091 });
3092 let bp = minimal_bp(
3093 vec![agent],
3094 vec![],
3095 simple_flow(
3096 "worker",
3097 Expr::Path {
3098 at: "$.input".parse().expect("literal test path: $.input"),
3099 },
3100 ),
3101 );
3102 let compiler = Compiler::new(registry_with_echo());
3103 match compiler.compile(&bp) {
3104 Err(CompileError::UnresolvedMetaRef {
3105 where_,
3106 meta_ref,
3107 defined,
3108 }) => {
3109 assert!(
3110 where_.contains("worker"),
3111 "where_ must name the agent: {where_}"
3112 );
3113 assert_eq!(meta_ref, "missing");
3114 assert!(defined.is_empty());
3115 }
3116 Err(other) => {
3117 panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
3118 }
3119 Ok(_) => panic!("expected compile-time failure, got Ok"),
3120 }
3121 }
3122
3123 #[test]
3124 fn unknown_static_step_meta_ref_in_lit_is_unresolved_meta_ref() {
3125 let agent = rustfn_agent("worker");
3126 let in_ = Expr::Lit {
3127 value: serde_json::json!({ "$step_meta": { "ref": "missing" }, "$in": "go" }),
3128 };
3129 let bp = minimal_bp(vec![agent], vec![], simple_flow("worker", in_));
3130 let compiler = Compiler::new(registry_with_echo());
3131 match compiler.compile(&bp) {
3132 Err(CompileError::UnresolvedMetaRef {
3133 where_, meta_ref, ..
3134 }) => {
3135 assert!(
3136 where_.contains("worker"),
3137 "where_ must name the offending step: {where_}"
3138 );
3139 assert_eq!(meta_ref, "missing");
3140 }
3141 Err(other) => {
3142 panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
3143 }
3144 Ok(_) => panic!("expected compile-time failure, got Ok"),
3145 }
3146 }
3147
3148 #[test]
3149 fn path_op_input_with_no_static_envelope_compiles_fine() {
3150 let agent = rustfn_agent("worker");
3151 let bp = minimal_bp(
3152 vec![agent],
3153 vec![],
3154 simple_flow(
3155 "worker",
3156 Expr::Path {
3157 at: "$.input".parse().expect("literal test path: $.input"),
3158 },
3159 ),
3160 );
3161 let compiler = Compiler::new(registry_with_echo());
3162 assert!(
3163 compiler.compile(&bp).is_ok(),
3164 "a non-Lit Step.in must not trigger the best-effort static $step_meta check"
3165 );
3166 }
3167}
3168
3169#[cfg(test)]
3171mod audit_agent_validation_tests {
3172 use super::*;
3173 use crate::worker::adapter::WorkerResult;
3174 use mlua_swarm_schema::{AuditDef, AuditMode};
3175
3176 fn registry_with_echo() -> SpawnerRegistry {
3177 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
3178 Ok(WorkerResult {
3179 value: Value::String(inv.prompt),
3180 ok: true,
3181 stats: None,
3182 })
3183 });
3184 let mut reg = SpawnerRegistry::new();
3185 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
3186 reg
3187 }
3188
3189 fn rustfn_agent(name: &str) -> AgentDef {
3190 AgentDef {
3191 name: name.to_string(),
3192 kind: AgentKind::RustFn,
3193 spec: serde_json::json!({ "fn_id": "echo" }),
3194 profile: None,
3195 meta: None,
3196 runner: None,
3197 runner_ref: None,
3198 verdict: None,
3199 lints: None,
3200 }
3201 }
3202
3203 fn minimal_bp(agents: Vec<AgentDef>, audits: Vec<AuditDef>) -> Blueprint {
3204 Blueprint {
3205 schema_version: crate::blueprint::current_schema_version(),
3206 id: "audit-ref-ut".into(),
3207 flow: FlowNode::Step {
3208 ref_: "worker".to_string(),
3209 in_: Expr::Path {
3210 at: "$.input".parse().expect("literal test path: $.input"),
3211 },
3212 out: Expr::Path {
3213 at: "$.output".parse().expect("literal test path: $.output"),
3214 },
3215 },
3216 agents,
3217 operators: vec![],
3218 metas: vec![],
3219 hints: Default::default(),
3220 strategy: Default::default(),
3221 metadata: BlueprintMetadata::default(),
3222 spawner_hints: Default::default(),
3223 default_agent_kind: AgentKind::Operator,
3224 default_operator_kind: None,
3225 default_init_ctx: None,
3226 default_agent_ctx: None,
3227 default_context_policy: None,
3228 projection_placement: None,
3229 audits,
3230 degradation_policy: None,
3231 runners: vec![],
3232 default_runner: None,
3233 subprocesses: vec![],
3234 check_policy: None,
3235 blueprint_ref_includes: Vec::new(),
3236 }
3237 }
3238
3239 #[test]
3240 fn unresolved_audit_agent_is_a_loud_compile_error() {
3241 let bp = minimal_bp(
3242 vec![rustfn_agent("worker")],
3243 vec![AuditDef {
3244 agent: "missing-auditor".to_string(),
3245 steps: None,
3246 mode: AuditMode::default(),
3247 }],
3248 );
3249 let compiler = Compiler::new(registry_with_echo());
3250 match compiler.compile(&bp) {
3251 Err(CompileError::UnresolvedAuditAgent { agent, defined }) => {
3252 assert_eq!(agent, "missing-auditor");
3253 assert_eq!(defined, vec!["worker".to_string()]);
3254 }
3255 Err(other) => {
3256 panic!("expected UnresolvedAuditAgent, got a different CompileError: {other}")
3257 }
3258 Ok(_) => panic!("expected compile-time failure, got Ok"),
3259 }
3260 }
3261
3262 #[test]
3263 fn resolved_audit_agent_compiles_fine() {
3264 let bp = minimal_bp(
3265 vec![rustfn_agent("worker"), rustfn_agent("auditor")],
3266 vec![AuditDef {
3267 agent: "auditor".to_string(),
3268 steps: None,
3269 mode: AuditMode::default(),
3270 }],
3271 );
3272 let compiler = Compiler::new(registry_with_echo());
3273 assert!(
3274 compiler.compile(&bp).is_ok(),
3275 "an audits[].agent that names a declared AgentDef must compile"
3276 );
3277 }
3278}
3279
3280#[cfg(test)]
3289mod operator_ref_resolution_tests {
3290 use super::*;
3291 use crate::core::ctx::Ctx;
3292 use crate::types::CapToken;
3293 use crate::worker::adapter::{WorkerError, WorkerResult};
3294 use std::sync::Mutex;
3295
3296 type Seen = Arc<Mutex<Vec<(String, Option<Value>)>>>;
3298
3299 struct RecordingOperatorFactory {
3303 seen: Seen,
3304 }
3305
3306 impl SpawnerFactory for RecordingOperatorFactory {
3307 fn build(
3308 &self,
3309 agent_def: &AgentDef,
3310 hint: Option<&Value>,
3311 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
3312 self.seen
3313 .lock()
3314 .expect("RecordingOperatorFactory.seen poisoned")
3315 .push((agent_def.name.clone(), hint.cloned()));
3316 let mut spawner: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
3317 let worker: WorkerFn = Arc::new(|_inv| {
3318 Box::pin(async move {
3319 Ok(WorkerResult {
3320 value: Value::Null,
3321 ok: true,
3322 stats: None,
3323 })
3324 })
3325 });
3326 spawner.registry.insert(agent_def.name.clone(), worker);
3327 Ok(Arc::new(spawner))
3328 }
3329 }
3330
3331 impl SpawnerFactoryKind for RecordingOperatorFactory {
3332 const KIND: AgentKind = AgentKind::Operator;
3333 type Worker = crate::operator::OperatorWorker;
3334 }
3335
3336 struct RecordingLuaFactory {
3339 seen: Seen,
3340 }
3341
3342 impl SpawnerFactory for RecordingLuaFactory {
3343 fn build(
3344 &self,
3345 agent_def: &AgentDef,
3346 hint: Option<&Value>,
3347 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
3348 self.seen
3349 .lock()
3350 .expect("RecordingLuaFactory.seen poisoned")
3351 .push((agent_def.name.clone(), hint.cloned()));
3352 let mut spawner: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
3353 let worker: WorkerFn = Arc::new(|_inv| {
3354 Box::pin(async move {
3355 Ok(WorkerResult {
3356 value: Value::Null,
3357 ok: true,
3358 stats: None,
3359 })
3360 })
3361 });
3362 spawner.registry.insert(agent_def.name.clone(), worker);
3363 Ok(Arc::new(spawner))
3364 }
3365 }
3366
3367 impl SpawnerFactoryKind for RecordingLuaFactory {
3368 const KIND: AgentKind = AgentKind::Lua;
3369 type Worker = LuaWorker;
3370 }
3371
3372 fn recording_compiler() -> (Compiler, Seen, Seen) {
3373 let operator_seen: Seen = Arc::new(Mutex::new(Vec::new()));
3374 let lua_seen: Seen = Arc::new(Mutex::new(Vec::new()));
3375 let mut registry = SpawnerRegistry::new();
3376 registry.register::<RecordingOperatorFactory>(Arc::new(RecordingOperatorFactory {
3377 seen: operator_seen.clone(),
3378 }));
3379 registry.register::<RecordingLuaFactory>(Arc::new(RecordingLuaFactory {
3380 seen: lua_seen.clone(),
3381 }));
3382 (Compiler::new(registry), operator_seen, lua_seen)
3383 }
3384
3385 fn bp_with_operator_and_lua_agents() -> Blueprint {
3389 serde_json::from_value(serde_json::json!({
3390 "schema_version": crate::blueprint::current_schema_version(),
3391 "id": "operator-pin-ut",
3392 "flow": {
3393 "kind": "step",
3394 "ref": "planner",
3395 "in": { "op": "path", "at": "$.input" },
3396 "out": { "op": "path", "at": "$.output" }
3397 },
3398 "agents": [
3399 {
3400 "name": "planner",
3401 "kind": "operator",
3402 "spec": { "operator_ref": "main-ai" }
3403 },
3404 {
3405 "name": "scorer",
3406 "kind": "lua",
3407 "spec": { "source": "return { value = 1, ok = true }" }
3408 }
3409 ],
3410 "operators": [{ "name": "main-ai" }],
3411 "hints": { "per_agent": { "planner": { "authored": "keep-me" } } },
3412 "strategy": { "strict_refs": false }
3413 }))
3414 .expect("test Blueprint literal")
3415 }
3416
3417 fn hint_for(seen: &Seen, agent: &str) -> Option<Value> {
3418 seen.lock()
3419 .expect("seen poisoned")
3420 .iter()
3421 .find(|(name, _)| name == agent)
3422 .map(|(_, hint)| hint.clone())
3423 .expect("agent was never built")
3424 }
3425
3426 #[test]
3431 fn the_compile_hands_the_factory_the_authored_hint_untouched() {
3432 let (compiler, operator_seen, lua_seen) = recording_compiler();
3433 let bp = bp_with_operator_and_lua_agents();
3434 let bound = resolve_bound_agents(&bp).expect("resolve bound agents");
3435 compiler.compile_bound(&bp, &bound).expect("compile");
3436
3437 assert_eq!(
3438 hint_for(&operator_seen, "planner"),
3439 Some(serde_json::json!({ "authored": "keep-me" })),
3440 "the compile must hand over the authored hint verbatim"
3441 );
3442 assert_eq!(
3443 hint_for(&lua_seen, "scorer"),
3444 None,
3445 "an agent with no authored hint must still be built with None"
3446 );
3447 }
3448
3449 #[test]
3452 fn a_non_object_authored_hint_is_none_of_the_compilers_business() {
3453 let (compiler, _operator_seen, _lua_seen) = recording_compiler();
3454 let mut bp = bp_with_operator_and_lua_agents();
3455 bp.hints
3456 .per_agent
3457 .insert("planner".to_string(), Value::String("not-an-object".into()));
3458 let bound = resolve_bound_agents(&bp).expect("resolve bound agents");
3459 assert!(
3460 compiler.compile_bound(&bp, &bound).is_ok(),
3461 "the compile must accept whatever hint shape the author declared"
3462 );
3463 }
3464
3465 struct StubOperator {
3471 requires_binding: bool,
3472 }
3473
3474 #[async_trait]
3475 impl Operator for StubOperator {
3476 async fn execute(
3477 &self,
3478 _ctx: &Ctx,
3479 _system: Option<String>,
3480 _prompt: Value,
3481 _worker: Option<WorkerBinding>,
3482 _worker_token: CapToken,
3483 ) -> Result<WorkerResult, WorkerError> {
3484 Ok(WorkerResult {
3485 value: Value::Null,
3486 ok: true,
3487 stats: None,
3488 })
3489 }
3490
3491 fn requires_worker_binding(&self) -> bool {
3492 self.requires_binding
3493 }
3494 }
3495
3496 fn operator_agent() -> AgentDef {
3497 AgentDef {
3498 name: "planner".to_string(),
3499 kind: AgentKind::Operator,
3500 spec: serde_json::json!({ "operator_ref": "main-ai" }),
3501 profile: None,
3502 meta: None,
3503 runner: None,
3504 runner_ref: None,
3505 verdict: None,
3506 lints: None,
3507 }
3508 }
3509
3510 struct StubResolver {
3514 seats: Vec<&'static str>,
3515 asked: Mutex<Vec<String>>,
3516 }
3517
3518 impl OperatorSlotResolver for StubResolver {
3519 fn resolve(&self, slot: &str) -> Option<Arc<dyn Operator>> {
3520 self.asked
3521 .lock()
3522 .expect("StubResolver.asked poisoned")
3523 .push(slot.to_string());
3524 self.seats.contains(&slot).then(|| {
3525 Arc::new(StubOperator {
3526 requires_binding: false,
3527 }) as Arc<dyn Operator>
3528 })
3529 }
3530 }
3531
3532 #[test]
3537 fn an_installed_resolver_answers_the_seat_and_the_registry_is_not_consulted() {
3538 let factory = OperatorSpawnerFactory::new();
3539 factory.register_operator(
3540 "main-ai",
3541 Arc::new(StubOperator {
3542 requires_binding: true,
3543 }) as Arc<dyn Operator>,
3544 );
3545 let resolver = Arc::new(StubResolver {
3546 seats: vec!["main-ai"],
3547 asked: Mutex::new(Vec::new()),
3548 });
3549 factory.set_slot_resolver(resolver.clone());
3550
3551 assert!(
3552 factory.build(&operator_agent(), None).is_ok(),
3553 "the installed resolver must answer the seat, not the registry entry \
3554 registered under the same name"
3555 );
3556 assert_eq!(
3557 *resolver.asked.lock().expect("asked"),
3558 vec!["main-ai".to_string()],
3559 "the resolver is asked for the seat the AgentDef declares"
3560 );
3561 }
3562
3563 #[test]
3567 fn a_resolver_miss_fails_loud_and_never_falls_back_to_the_registry() {
3568 let factory = OperatorSpawnerFactory::new();
3569 factory.register_operator(
3570 "main-ai",
3571 Arc::new(StubOperator {
3572 requires_binding: false,
3573 }) as Arc<dyn Operator>,
3574 );
3575 factory.set_slot_resolver(Arc::new(StubResolver {
3576 seats: vec!["some-other-seat"],
3577 asked: Mutex::new(Vec::new()),
3578 }));
3579
3580 match factory.build(&operator_agent(), None) {
3581 Err(CompileError::InvalidSpec { name, msg }) => {
3582 assert_eq!(name, "planner");
3583 assert!(
3584 msg.contains("main-ai"),
3585 "message must name the seat that went unserved: {msg}"
3586 );
3587 assert!(
3588 msg.contains("OperatorSlotResolver"),
3589 "message must say which side refused, so the wiring is the \
3590 obvious suspect: {msg}"
3591 );
3592 }
3593 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
3594 Ok(_) => panic!(
3595 "an unserved seat must fail the compile, not silently resolve the \
3596 registry entry"
3597 ),
3598 }
3599 }
3600
3601 #[test]
3604 fn without_a_resolver_the_registry_answers_with_the_historical_message() {
3605 let factory = OperatorSpawnerFactory::new();
3606 match factory.build(&operator_agent(), None) {
3607 Err(CompileError::InvalidSpec { msg, .. }) => {
3608 assert!(
3609 msg.contains("operator_ref 'main-ai' not registered in factory"),
3610 "the registry-side message must stay the historical one: {msg}"
3611 );
3612 }
3613 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
3614 Ok(_) => panic!("an unregistered seat must still fail"),
3615 }
3616 factory.register_operator(
3617 "main-ai",
3618 Arc::new(StubOperator {
3619 requires_binding: false,
3620 }) as Arc<dyn Operator>,
3621 );
3622 assert!(
3623 factory.build(&operator_agent(), None).is_ok(),
3624 "a registered backend must still resolve the seat directly"
3625 );
3626 }
3627}
3628
3629#[cfg(test)]
3632mod projection_placement_compile_tests {
3633 use super::*;
3634 use crate::core::projection_placement::{ProjectionPlacement, RootPreference};
3635 use crate::worker::adapter::WorkerResult;
3636 use mlua_swarm_schema::ProjectionPlacementSpec;
3637
3638 fn registry_with_echo() -> SpawnerRegistry {
3639 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
3640 Ok(WorkerResult {
3641 value: Value::String(inv.prompt),
3642 ok: true,
3643 stats: None,
3644 })
3645 });
3646 let mut reg = SpawnerRegistry::new();
3647 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
3648 reg
3649 }
3650
3651 fn minimal_bp(projection_placement: Option<ProjectionPlacementSpec>) -> Blueprint {
3652 Blueprint {
3653 schema_version: crate::blueprint::current_schema_version(),
3654 id: "projection-placement-ut".into(),
3655 flow: FlowNode::Step {
3656 ref_: "worker".to_string(),
3657 in_: Expr::Path {
3658 at: "$.input".parse().expect("literal test path: $.input"),
3659 },
3660 out: Expr::Path {
3661 at: "$.output".parse().expect("literal test path: $.output"),
3662 },
3663 },
3664 agents: vec![AgentDef {
3665 name: "worker".to_string(),
3666 kind: AgentKind::RustFn,
3667 spec: serde_json::json!({ "fn_id": "echo" }),
3668 profile: None,
3669 meta: None,
3670 runner: None,
3671 runner_ref: None,
3672 verdict: None,
3673 lints: None,
3674 }],
3675 operators: vec![],
3676 metas: vec![],
3677 hints: Default::default(),
3678 strategy: Default::default(),
3679 metadata: BlueprintMetadata::default(),
3680 spawner_hints: Default::default(),
3681 default_agent_kind: AgentKind::Operator,
3682 default_operator_kind: None,
3683 default_init_ctx: None,
3684 default_agent_ctx: None,
3685 default_context_policy: None,
3686 projection_placement,
3687 audits: vec![],
3688 degradation_policy: None,
3689 runners: vec![],
3690 default_runner: None,
3691 subprocesses: vec![],
3692 check_policy: None,
3693 blueprint_ref_includes: Vec::new(),
3694 }
3695 }
3696
3697 #[test]
3698 fn undeclared_projection_placement_compiles_to_byte_compat_default() {
3699 let bp = minimal_bp(None);
3700 let compiled = Compiler::new(registry_with_echo())
3701 .compile(&bp)
3702 .expect("undeclared projection_placement compiles");
3703 assert_eq!(
3704 *compiled.projection_placement,
3705 ProjectionPlacement::default()
3706 );
3707 }
3708
3709 #[test]
3710 fn declared_valid_projection_placement_compiles_to_matching_resolver() {
3711 let bp = minimal_bp(Some(ProjectionPlacementSpec {
3712 root: Some("project_root".to_string()),
3713 dir_template: Some("custom/{task_id}/out".to_string()),
3714 }));
3715 let compiled = Compiler::new(registry_with_echo())
3716 .compile(&bp)
3717 .expect("valid projection_placement compiles");
3718 assert_eq!(
3719 compiled.projection_placement.root_preference,
3720 RootPreference::ProjectRoot
3721 );
3722 assert_eq!(
3723 compiled.projection_placement.dir_template,
3724 "custom/{task_id}/out"
3725 );
3726 }
3727
3728 #[test]
3729 fn declared_invalid_dir_template_rejects_compile() {
3730 let bp = minimal_bp(Some(ProjectionPlacementSpec {
3731 root: None,
3732 dir_template: Some("workspace/tasks/ctx".to_string()), }));
3734 match Compiler::new(registry_with_echo()).compile(&bp) {
3735 Err(CompileError::InvalidProjectionPlacement(_)) => {}
3736 Err(other) => {
3737 panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
3738 }
3739 Ok(_) => {
3740 panic!("expected compile-time rejection for a missing {{task_id}} placeholder")
3741 }
3742 }
3743 }
3744
3745 #[test]
3746 fn declared_invalid_root_literal_rejects_compile() {
3747 let bp = minimal_bp(Some(ProjectionPlacementSpec {
3748 root: Some("nope".to_string()),
3749 dir_template: None,
3750 }));
3751 match Compiler::new(registry_with_echo()).compile(&bp) {
3752 Err(CompileError::InvalidProjectionPlacement(_)) => {}
3753 Err(other) => {
3754 panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
3755 }
3756 Ok(_) => panic!("expected compile-time rejection for an invalid root literal"),
3757 }
3758 }
3759}
3760
3761#[cfg(test)]
3763mod verdict_contract_lint_tests {
3764 use super::*;
3765 use crate::worker::adapter::WorkerResult;
3766
3767 fn registry_with_echo() -> SpawnerRegistry {
3768 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
3769 Ok(WorkerResult {
3770 value: Value::String(inv.prompt),
3771 ok: true,
3772 stats: None,
3773 })
3774 });
3775 let mut reg = SpawnerRegistry::new();
3776 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
3777 reg
3778 }
3779
3780 fn gate_agent(verdict: Option<VerdictContract>) -> AgentDef {
3781 AgentDef {
3782 name: "gate".to_string(),
3783 kind: AgentKind::RustFn,
3784 spec: serde_json::json!({ "fn_id": "echo" }),
3785 profile: None,
3786 meta: None,
3787 runner: None,
3788 runner_ref: None,
3789 verdict,
3790 lints: None,
3791 }
3792 }
3793
3794 fn minimal_bp(agent: AgentDef, flow: FlowNode) -> Blueprint {
3795 Blueprint {
3796 schema_version: crate::blueprint::current_schema_version(),
3797 id: "verdict-contract-ut".into(),
3798 flow,
3799 agents: vec![agent],
3800 operators: vec![],
3801 metas: vec![],
3802 hints: Default::default(),
3803 strategy: Default::default(),
3804 metadata: BlueprintMetadata::default(),
3805 spawner_hints: Default::default(),
3806 default_agent_kind: AgentKind::Operator,
3807 default_operator_kind: None,
3808 default_init_ctx: None,
3809 default_agent_ctx: None,
3810 default_context_policy: None,
3811 projection_placement: None,
3812 audits: vec![],
3813 degradation_policy: None,
3814 runners: vec![],
3815 default_runner: None,
3816 subprocesses: vec![],
3817 check_policy: None,
3818 blueprint_ref_includes: Vec::new(),
3819 }
3820 }
3821
3822 fn step(ref_: &str, out_path: &str) -> FlowNode {
3823 FlowNode::Step {
3824 ref_: ref_.to_string(),
3825 in_: Expr::Lit { value: Value::Null },
3826 out: Expr::Path {
3827 at: out_path.parse().expect("literal test path"),
3828 },
3829 }
3830 }
3831
3832 fn noop() -> FlowNode {
3833 FlowNode::Seq { children: vec![] }
3834 }
3835
3836 fn eq_cond(path: &str, lit: &str) -> Expr {
3837 Expr::Eq {
3838 lhs: Box::new(Expr::Path {
3839 at: path.parse().expect("literal test path"),
3840 }),
3841 rhs: Box::new(Expr::Lit {
3842 value: Value::String(lit.to_string()),
3843 }),
3844 }
3845 }
3846
3847 fn branch(cond: Expr, then_: FlowNode, else_: FlowNode) -> FlowNode {
3848 FlowNode::Branch {
3849 cond,
3850 then_: Box::new(then_),
3851 else_: Box::new(else_),
3852 }
3853 }
3854
3855 fn body_contract(values: &[&str]) -> VerdictContract {
3856 VerdictContract {
3857 channel: VerdictChannel::Body,
3858 values: values.iter().map(|v| v.to_string()).collect(),
3859 }
3860 }
3861
3862 fn part_contract(values: &[&str]) -> VerdictContract {
3863 VerdictContract {
3864 channel: VerdictChannel::Part,
3865 values: values.iter().map(|v| v.to_string()).collect(),
3866 }
3867 }
3868
3869 #[test]
3870 fn contract_with_correct_body_channel_and_value_compiles() {
3871 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3872 let flow = FlowNode::Seq {
3873 children: vec![
3874 step("gate", "$.verdict"),
3875 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3876 ],
3877 };
3878 let bp = minimal_bp(agent, flow);
3879 assert!(
3880 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3881 "a cond addressing the bare step output must match a channel: \"body\" contract"
3882 );
3883 }
3884
3885 #[test]
3886 fn contract_with_correct_part_channel_and_value_compiles() {
3887 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3888 let flow = FlowNode::Seq {
3889 children: vec![
3890 step("gate", "$.gate"),
3891 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3892 ],
3893 };
3894 let bp = minimal_bp(agent, flow);
3895 assert!(
3896 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3897 "a cond addressing '<step>.parts.verdict' must match a channel: \"part\" contract"
3898 );
3899 }
3900
3901 #[test]
3902 fn body_channel_contract_rejects_cond_addressing_parts_verdict() {
3903 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3907 let flow = FlowNode::Seq {
3908 children: vec![
3909 step("gate", "$.gate"),
3910 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3911 ],
3912 };
3913 let bp = minimal_bp(agent, flow);
3914 match Compiler::new(registry_with_echo()).compile(&bp) {
3915 Err(CompileError::VerdictChannelMismatch {
3916 where_,
3917 agent,
3918 expected_channel,
3919 actual_shape,
3920 }) => {
3921 assert_eq!(agent, "gate");
3922 assert_eq!(expected_channel, "body");
3923 assert_eq!(actual_shape, "part");
3924 assert!(where_.contains("Branch cond"), "where_: {where_}");
3925 }
3926 Err(other) => {
3927 panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
3928 }
3929 Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
3930 }
3931 }
3932
3933 #[test]
3934 fn part_channel_contract_rejects_cond_addressing_bare_output() {
3935 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3938 let flow = FlowNode::Seq {
3939 children: vec![
3940 step("gate", "$.verdict"),
3941 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3942 ],
3943 };
3944 let bp = minimal_bp(agent, flow);
3945 match Compiler::new(registry_with_echo()).compile(&bp) {
3946 Err(CompileError::VerdictChannelMismatch {
3947 agent,
3948 expected_channel,
3949 actual_shape,
3950 ..
3951 }) => {
3952 assert_eq!(agent, "gate");
3953 assert_eq!(expected_channel, "part");
3954 assert_eq!(actual_shape, "body");
3955 }
3956 Err(other) => {
3957 panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
3958 }
3959 Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
3960 }
3961 }
3962
3963 #[test]
3964 fn contract_rejects_lit_outside_declared_values() {
3965 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3966 let flow = FlowNode::Seq {
3967 children: vec![
3968 step("gate", "$.verdict"),
3969 branch(eq_cond("$.verdict", "UNKNOWN"), noop(), noop()),
3970 ],
3971 };
3972 let bp = minimal_bp(agent, flow);
3973 match Compiler::new(registry_with_echo()).compile(&bp) {
3974 Err(CompileError::VerdictValueNotInContract {
3975 agent,
3976 value,
3977 values,
3978 ..
3979 }) => {
3980 assert_eq!(agent, "gate");
3981 assert_eq!(value, "UNKNOWN");
3982 assert_eq!(values, vec!["PASS".to_string(), "BLOCKED".to_string()]);
3983 }
3984 Err(other) => {
3985 panic!("expected VerdictValueNotInContract, got a different CompileError: {other}")
3986 }
3987 Ok(_) => panic!("expected compile-time rejection for a Lit outside declared values"),
3988 }
3989 }
3990
3991 #[test]
3992 fn undeclared_agent_referenced_by_cond_compiles_with_warning_only() {
3993 let agent = gate_agent(None);
3994 let flow = FlowNode::Seq {
3995 children: vec![
3996 step("gate", "$.verdict"),
3997 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3998 ],
3999 };
4000 let bp = minimal_bp(agent, flow);
4001 assert!(
4002 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4003 "an undeclared verdict contract must never reject compile (opt-in, back-compat)"
4004 );
4005 }
4006
4007 #[test]
4008 fn in_expr_with_lit_haystack_members_compiles() {
4009 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4010 let cond = Expr::In {
4011 needle: Box::new(Expr::Path {
4012 at: "$.verdict".parse().expect("literal test path"),
4013 }),
4014 haystack: Box::new(Expr::Lit {
4015 value: serde_json::json!(["PASS", "BLOCKED"]),
4016 }),
4017 };
4018 let flow = FlowNode::Seq {
4019 children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
4020 };
4021 let bp = minimal_bp(agent, flow);
4022 assert!(
4023 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4024 "an `In` haystack whose every Lit is a declared value must compile"
4025 );
4026 }
4027
4028 #[test]
4035 fn strict_mode_rejects_unhandled_declared_value() {
4036 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4037 let flow = FlowNode::Seq {
4038 children: vec![
4039 step("gate", "$.verdict"),
4040 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
4041 ],
4042 };
4043 let mut bp = minimal_bp(agent, flow);
4044 bp.metadata.strict_verdict_handling = Some(true);
4045 match Compiler::new(registry_with_echo()).compile(&bp) {
4046 Err(CompileError::VerdictValueUnhandled {
4047 agent,
4048 value,
4049 declared_values,
4050 step_ref,
4051 }) => {
4052 assert_eq!(agent, "gate");
4053 assert_eq!(value, "PASS");
4054 assert_eq!(
4055 declared_values,
4056 vec!["PASS".to_string(), "BLOCKED".to_string()]
4057 );
4058 assert_eq!(step_ref, "gate");
4059 }
4060 Err(other) => {
4061 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4062 }
4063 Ok(_) => panic!(
4064 "expected compile-time rejection for a declared verdict value with no \
4065 downstream handler under strict_verdict_handling=Some(true)"
4066 ),
4067 }
4068 }
4069
4070 #[test]
4077 fn default_mode_permits_unhandled_declared_value() {
4078 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4079 let flow = FlowNode::Seq {
4080 children: vec![
4081 step("gate", "$.verdict"),
4082 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
4083 ],
4084 };
4085 let bp = minimal_bp(agent, flow);
4086 assert!(
4088 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4089 "default mode must never reject a Blueprint for unhandled declared values \
4090 (opt-in, back-compat with GH #50)"
4091 );
4092 }
4093
4094 #[test]
4099 fn strict_mode_accepts_all_declared_values_handled() {
4100 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4101 let flow = FlowNode::Seq {
4104 children: vec![
4105 step("gate", "$.verdict"),
4106 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
4107 branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
4108 ],
4109 };
4110 let mut bp = minimal_bp(agent, flow);
4111 bp.metadata.strict_verdict_handling = Some(true);
4112 assert!(
4113 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4114 "strict mode must accept a Blueprint that handles every declared value"
4115 );
4116 }
4117
4118 #[test]
4122 fn strict_mode_accepts_declared_values_covered_by_in_expr() {
4123 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4124 let cond = Expr::In {
4125 needle: Box::new(Expr::Path {
4126 at: "$.verdict".parse().expect("literal test path"),
4127 }),
4128 haystack: Box::new(Expr::Lit {
4129 value: serde_json::json!(["PASS", "BLOCKED"]),
4130 }),
4131 };
4132 let flow = FlowNode::Seq {
4133 children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
4134 };
4135 let mut bp = minimal_bp(agent, flow);
4136 bp.metadata.strict_verdict_handling = Some(true);
4137 assert!(
4138 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4139 "strict mode must accept an `In` haystack that covers every declared value"
4140 );
4141 }
4142
4143 #[test]
4147 fn strict_mode_rejects_unhandled_part_channel_value() {
4148 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
4149 let flow = FlowNode::Seq {
4150 children: vec![
4151 step("gate", "$.gate"),
4152 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
4153 ],
4154 };
4155 let mut bp = minimal_bp(agent, flow);
4156 bp.metadata.strict_verdict_handling = Some(true);
4157 match Compiler::new(registry_with_echo()).compile(&bp) {
4158 Err(CompileError::VerdictValueUnhandled {
4159 agent,
4160 value,
4161 step_ref,
4162 ..
4163 }) => {
4164 assert_eq!(agent, "gate");
4165 assert_eq!(value, "PASS");
4166 assert_eq!(step_ref, "gate");
4167 }
4168 Err(other) => {
4169 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4170 }
4171 Ok(_) => panic!(
4172 "expected compile-time rejection for a declared verdict value with no \
4173 downstream handler (part channel) under strict_verdict_handling=Some(true)"
4174 ),
4175 }
4176 }
4177
4178 fn lints(
4184 pairs: &[(&str, mlua_swarm_schema::LintSetting)],
4185 ) -> Option<std::collections::BTreeMap<String, mlua_swarm_schema::LintSetting>> {
4186 Some(
4187 pairs
4188 .iter()
4189 .map(|(key, setting)| ((*key).to_string(), *setting))
4190 .collect(),
4191 )
4192 }
4193
4194 fn bp_with_unhandled_value() -> Blueprint {
4198 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4199 let flow = FlowNode::Seq {
4200 children: vec![
4201 step("gate", "$.verdict"),
4202 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
4203 ],
4204 };
4205 minimal_bp(agent, flow)
4206 }
4207
4208 fn named_agent(name: &str, verdict: Option<VerdictContract>) -> AgentDef {
4212 AgentDef {
4213 name: name.to_string(),
4214 ..gate_agent(verdict)
4215 }
4216 }
4217
4218 fn bp_with_two_unhandled_agents() -> Blueprint {
4222 let flow = FlowNode::Seq {
4223 children: vec![
4224 step("researcher", "$.researcher_verdict"),
4225 step("reviewer", "$.reviewer_verdict"),
4226 branch(eq_cond("$.researcher_verdict", "BLOCKED"), noop(), noop()),
4227 branch(eq_cond("$.reviewer_verdict", "BLOCKED"), noop(), noop()),
4228 ],
4229 };
4230 let mut bp = minimal_bp(
4231 named_agent("researcher", Some(body_contract(&["PASS", "BLOCKED"]))),
4232 flow,
4233 );
4234 bp.agents.push(named_agent(
4235 "reviewer",
4236 Some(body_contract(&["PASS", "BLOCKED"])),
4237 ));
4238 bp
4239 }
4240
4241 #[test]
4246 fn agent_lints_deny_rejects_only_the_declaring_agent() {
4247 let mut bp = bp_with_two_unhandled_agents();
4248 bp.agents[0].lints = lints(&[(
4249 "verdict-value-unhandled",
4250 mlua_swarm_schema::LintSetting::Deny,
4251 )]);
4252 match Compiler::new(registry_with_echo()).compile(&bp) {
4253 Err(CompileError::VerdictValueUnhandled { agent, value, .. }) => {
4254 assert_eq!(agent, "researcher", "the sibling only warns");
4255 assert_eq!(value, "PASS");
4256 }
4257 Err(other) => {
4258 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4259 }
4260 Ok(_) => panic!(
4261 "expected compile-time rejection under \
4262 agents[0].lints = {{\"verdict-value-unhandled\": \"deny\"}}"
4263 ),
4264 }
4265 }
4266
4267 #[test]
4271 fn agent_allow_beats_blueprint_deny_for_that_agent() {
4272 let mut bp = bp_with_two_unhandled_agents();
4273 bp.metadata.lints = lints(&[(
4274 "verdict-value-unhandled",
4275 mlua_swarm_schema::LintSetting::Deny,
4276 )]);
4277 bp.agents[0].lints = lints(&[(
4278 "verdict-value-unhandled",
4279 mlua_swarm_schema::LintSetting::Allow,
4280 )]);
4281 match Compiler::new(registry_with_echo()).compile(&bp) {
4282 Err(CompileError::VerdictValueUnhandled { agent, .. }) => {
4283 assert_eq!(
4284 agent, "reviewer",
4285 "the allowing agent is silenced; the sibling still denies"
4286 );
4287 }
4288 Err(other) => {
4289 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4290 }
4291 Ok(_) => panic!("the sibling agent's Blueprint-level deny must still reject"),
4292 }
4293 }
4294
4295 #[test]
4299 fn strict_flag_wins_over_agent_lints_allow() {
4300 let mut bp = bp_with_unhandled_value();
4301 bp.metadata.strict_verdict_handling = Some(true);
4302 bp.agents[0].lints = lints(&[("all", mlua_swarm_schema::LintSetting::Allow)]);
4303 match Compiler::new(registry_with_echo()).compile(&bp) {
4304 Err(CompileError::VerdictValueUnhandled { agent, .. }) => assert_eq!(agent, "gate"),
4305 Err(other) => {
4306 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4307 }
4308 Ok(_) => panic!(
4309 "strict_verdict_handling=Some(true) must still reject under an agent-level allow"
4310 ),
4311 }
4312 }
4313
4314 #[test]
4317 fn agent_category_key_reaches_the_kind() {
4318 let mut bp = bp_with_two_unhandled_agents();
4319 bp.agents[0].lints =
4320 lints(&[("category:suspicious", mlua_swarm_schema::LintSetting::Deny)]);
4321 match Compiler::new(registry_with_echo()).compile(&bp) {
4322 Err(CompileError::VerdictValueUnhandled { agent, .. }) => assert_eq!(
4323 agent, "researcher",
4324 "a category: group deny must reach the kind it covers, on the declaring agent"
4325 ),
4326 Err(other) => {
4327 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4328 }
4329 Ok(_) => panic!("expected compile-time rejection under an agent-level category deny"),
4330 }
4331 }
4332
4333 #[test]
4336 fn agent_without_lints_inherits_the_blueprint_layer() {
4337 let mut bp = bp_with_two_unhandled_agents();
4338 bp.metadata.lints = lints(&[("all", mlua_swarm_schema::LintSetting::Allow)]);
4339 assert!(
4340 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4341 "a Blueprint-wide allow covers every agent that declares nothing"
4342 );
4343
4344 let gates = resolve_unhandled_verdict_gates(&bp);
4345 assert_eq!(gates.for_agent("reviewer"), UnhandledVerdictGate::Silence);
4346 assert!(gates.all_silent());
4347
4348 bp.agents[0].lints = lints(&[(
4349 "verdict-value-unhandled",
4350 mlua_swarm_schema::LintSetting::Warn,
4351 )]);
4352 let gates = resolve_unhandled_verdict_gates(&bp);
4353 assert_eq!(
4354 gates.for_agent("researcher"),
4355 UnhandledVerdictGate::Warn,
4356 "the agent's own layer wins over the Blueprint's allow"
4357 );
4358 assert_eq!(
4359 gates.for_agent("reviewer"),
4360 UnhandledVerdictGate::Silence,
4361 "the sibling keeps the Blueprint layer"
4362 );
4363 assert!(!gates.all_silent());
4364 }
4365
4366 #[test]
4370 fn lints_deny_rejects_unhandled_declared_value() {
4371 let mut bp = bp_with_unhandled_value();
4372 bp.metadata.lints = lints(&[(
4373 "verdict-value-unhandled",
4374 mlua_swarm_schema::LintSetting::Deny,
4375 )]);
4376 match Compiler::new(registry_with_echo()).compile(&bp) {
4377 Err(CompileError::VerdictValueUnhandled { agent, value, .. }) => {
4378 assert_eq!(agent, "gate");
4379 assert_eq!(value, "PASS");
4380 }
4381 Err(other) => {
4382 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4383 }
4384 Ok(_) => panic!(
4385 "expected compile-time rejection under \
4386 metadata.lints = {{\"verdict-value-unhandled\": \"deny\"}}"
4387 ),
4388 }
4389 }
4390
4391 #[test]
4394 fn lints_category_deny_rejects_unhandled_declared_value() {
4395 let mut bp = bp_with_unhandled_value();
4396 bp.metadata.lints = lints(&[("category:suspicious", mlua_swarm_schema::LintSetting::Deny)]);
4397 assert!(
4398 matches!(
4399 Compiler::new(registry_with_echo()).compile(&bp),
4400 Err(CompileError::VerdictValueUnhandled { .. })
4401 ),
4402 "a category: group deny must reach the kind it covers"
4403 );
4404 }
4405
4406 #[test]
4410 fn lints_allow_compiles_and_silences_the_warn() {
4411 let mut bp = bp_with_unhandled_value();
4412 bp.metadata.lints = lints(&[(
4413 "verdict-value-unhandled",
4414 mlua_swarm_schema::LintSetting::Allow,
4415 )]);
4416 assert!(
4417 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4418 "an allowed lint must never reject the compile"
4419 );
4420 assert_eq!(
4421 resolve_unhandled_verdict_gate(&bp.metadata),
4422 UnhandledVerdictGate::Silence
4423 );
4424 }
4425
4426 #[test]
4430 fn strict_flag_wins_over_lints_allow() {
4431 let mut bp = bp_with_unhandled_value();
4432 bp.metadata.strict_verdict_handling = Some(true);
4433 bp.metadata.lints = lints(&[("all", mlua_swarm_schema::LintSetting::Allow)]);
4434 assert!(
4435 matches!(
4436 Compiler::new(registry_with_echo()).compile(&bp),
4437 Err(CompileError::VerdictValueUnhandled { .. })
4438 ),
4439 "strict_verdict_handling=Some(true) must still reject under a lints allow"
4440 );
4441 }
4442
4443 #[test]
4446 fn unhandled_verdict_gate_resolution_table() {
4447 use mlua_swarm_schema::LintSetting;
4448
4449 let gate = |strict, map| {
4450 resolve_unhandled_verdict_gate(&BlueprintMetadata {
4451 strict_verdict_handling: strict,
4452 lints: map,
4453 ..Default::default()
4454 })
4455 };
4456 let kind = "verdict-value-unhandled";
4457
4458 assert_eq!(gate(None, None), UnhandledVerdictGate::Warn);
4459 assert_eq!(gate(Some(false), None), UnhandledVerdictGate::Warn);
4460 assert_eq!(gate(Some(true), None), UnhandledVerdictGate::Deny);
4461 assert_eq!(
4462 gate(None, lints(&[(kind, LintSetting::Deny)])),
4463 UnhandledVerdictGate::Deny
4464 );
4465 assert_eq!(
4466 gate(None, lints(&[(kind, LintSetting::Warn)])),
4467 UnhandledVerdictGate::Warn
4468 );
4469 assert_eq!(
4470 gate(None, lints(&[(kind, LintSetting::Allow)])),
4471 UnhandledVerdictGate::Silence
4472 );
4473 assert_eq!(
4474 gate(Some(true), lints(&[(kind, LintSetting::Allow)])),
4475 UnhandledVerdictGate::Deny,
4476 "strict wins over allow"
4477 );
4478 assert_eq!(
4480 gate(
4481 None,
4482 lints(&[
4483 (kind, LintSetting::Allow),
4484 ("category:suspicious", LintSetting::Deny),
4485 ])
4486 ),
4487 UnhandledVerdictGate::Silence
4488 );
4489 assert_eq!(
4492 gate(None, lints(&[("no-such-lint", LintSetting::Deny)])),
4493 UnhandledVerdictGate::Warn
4494 );
4495 }
4496
4497 #[test]
4501 fn lints_never_soften_other_compile_errors() {
4502 let mut bp = bp_with_unhandled_value();
4503 bp.agents.push(gate_agent(None));
4504 bp.metadata.lints = lints(&[("all", mlua_swarm_schema::LintSetting::Allow)]);
4505 assert!(
4506 matches!(
4507 Compiler::new(registry_with_echo()).compile(&bp),
4508 Err(CompileError::DuplicateAgent(name)) if name == "gate"
4509 ),
4510 "an `all` allow must not suppress a compile hard error"
4511 );
4512 }
4513
4514 #[test]
4521 fn verdict_omitted_blueprint_compiles_unchanged_with_empty_contracts() {
4522 let agent = gate_agent(None);
4523 let flow = FlowNode::Seq {
4524 children: vec![
4525 step("gate", "$.verdict"),
4526 FlowNode::Loop {
4527 counter: Expr::Path {
4528 at: "$.n".parse().expect("literal test path"),
4529 },
4530 cond: eq_cond("$.verdict", "BLOCKED"),
4531 body: Box::new(step("gate", "$.verdict")),
4532 max: 3,
4533 },
4534 branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
4535 ],
4536 };
4537 let bp = minimal_bp(agent, flow);
4538 let compiled = Compiler::new(registry_with_echo())
4539 .compile(&bp)
4540 .expect("a verdict-omitted Blueprint must compile unchanged");
4541 assert!(
4542 compiled.router.verdict_contracts.is_empty(),
4543 "no agent declared a verdict contract"
4544 );
4545 }
4546
4547 #[test]
4554 fn every_compile_error_diagnostic_kind_is_a_declared_lint() {
4555 let kinds = [
4556 "bound-agent-resolution",
4557 "unknown-agent-kind",
4558 "invalid-agent-spec",
4559 "worker-binding-missing",
4560 "unresolved-agent-ref",
4561 "duplicate-agent-name",
4562 "unresolved-operator-ref",
4563 "unresolved-meta-ref",
4564 "step-naming-collision",
4565 "invalid-projection-placement",
4566 "unresolved-audit-agent",
4567 "verdict-channel-mismatch",
4568 "verdict-value-not-in-contract",
4569 "verdict-value-unhandled",
4570 ];
4571 for kind in kinds {
4572 assert!(
4573 mlua_swarm_diag::lint_decl(kind).is_some(),
4574 "kind '{kind}' emitted by From<&CompileError> has no LINT_DECLS entry"
4575 );
4576 }
4577 }
4578
4579 #[test]
4580 fn invalid_spec_with_worker_binding_prefix_specializes_the_diagnostic_kind() {
4581 let err = CompileError::InvalidSpec {
4585 name: "greeter".into(),
4586 msg: format!("{WORKER_BINDING_REQUIRED_MSG_PREFIX}. Fix by either: (a) ..."),
4587 };
4588 let d = mlua_swarm_diag::Diagnostic::from(&err);
4589 assert_eq!(d.kind, "worker-binding-missing");
4590 assert_eq!(d.level, mlua_swarm_diag::DiagLevel::Error);
4591 assert!(matches!(d.stage, mlua_swarm_diag::DiagStage::CompileLint));
4592 assert!(d.message.contains("greeter"));
4593 let suggestion = d
4594 .suggestion
4595 .expect("specialized arm must carry a suggestion");
4596 assert!(suggestion.patch.contains("backend = \"ws_operator\""));
4597 assert_eq!(
4598 suggestion.applicability,
4599 mlua_swarm_diag::Applicability::HasPlaceholders
4600 );
4601 assert_eq!(
4602 d.docs_ref.expect("docs_ref must be set").uri,
4603 "mse://guides/bp-dsl-templates"
4604 );
4605 match d.span.expect("span must be set").element {
4606 mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "greeter"),
4607 other => panic!("expected Agent span, got {other:?}"),
4608 }
4609 }
4610
4611 #[test]
4612 fn generic_invalid_spec_maps_to_the_generic_kind() {
4613 let err = CompileError::InvalidSpec {
4614 name: "solo".into(),
4615 msg: "operator spec: 'operator_ref' (string) required".into(),
4616 };
4617 let d = mlua_swarm_diag::Diagnostic::from(&err);
4618 assert_eq!(d.kind, "invalid-agent-spec");
4619 assert!(
4620 d.suggestion.is_none(),
4621 "generic arm carries no canned patch"
4622 );
4623 }
4624
4625 #[test]
4626 fn verdict_value_not_in_contract_diagnostic_carries_suggestion_and_span() {
4627 let err = CompileError::VerdictValueNotInContract {
4628 where_: "Branch cond".into(),
4629 agent: "review".into(),
4630 value: "NOT_DECLARED".into(),
4631 values: vec!["PASS".into(), "BLOCKED".into()],
4632 };
4633 let d = mlua_swarm_diag::Diagnostic::from(&err);
4634 assert_eq!(d.kind, "verdict-value-not-in-contract");
4635 assert!(d.message.contains("NOT_DECLARED"));
4636 assert!(d.suggestion.is_some());
4637 match d.span.expect("span must be set").element {
4638 mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "review"),
4639 other => panic!("expected Agent span, got {other:?}"),
4640 }
4641 }
4642}
4643
4644#[cfg(test)]
4646mod subprocess_embed_compile_tests {
4647 use super::*;
4648 use mlua_swarm_schema::{current_schema_version, SubprocessDef, SubprocessOverrides};
4649
4650 fn subprocess_agent(name: &str, runner: Option<Runner>) -> AgentDef {
4651 AgentDef {
4652 name: name.to_string(),
4653 kind: AgentKind::Subprocess,
4654 spec: serde_json::json!({}),
4655 profile: Some(AgentProfile {
4656 system_prompt: "you are a headless worker".to_string(),
4657 model: Some("profile-model".to_string()),
4658 tools: vec!["Read".to_string()],
4659 ..Default::default()
4660 }),
4661 meta: None,
4662 runner,
4663 runner_ref: None,
4664 verdict: None,
4665 lints: None,
4666 }
4667 }
4668
4669 fn echo_def(name: &str) -> SubprocessDef {
4670 SubprocessDef {
4671 name: name.to_string(),
4672 argv: vec!["sh".to_string(), "-c".to_string(), "cat".to_string()],
4673 stdin: Some("{prompt}".to_string()),
4674 env: Default::default(),
4675 cwd: None,
4676 output: None,
4677 stream_mode: None,
4678 }
4679 }
4680
4681 fn bp_with(agents: Vec<AgentDef>, subprocesses: Vec<SubprocessDef>) -> Blueprint {
4682 Blueprint {
4683 schema_version: current_schema_version(),
4684 id: "gh83-ut".into(),
4685 flow: FlowNode::Seq { children: vec![] },
4686 agents,
4687 operators: vec![],
4688 metas: vec![],
4689 hints: Default::default(),
4690 strategy: Default::default(),
4691 metadata: BlueprintMetadata::default(),
4692 spawner_hints: Default::default(),
4693 default_agent_kind: AgentKind::Operator,
4694 default_operator_kind: None,
4695 default_init_ctx: None,
4696 default_agent_ctx: None,
4697 default_context_policy: None,
4698 projection_placement: None,
4699 audits: vec![],
4700 degradation_policy: None,
4701 runners: vec![],
4702 default_runner: None,
4703 subprocesses,
4704 check_policy: None,
4705 blueprint_ref_includes: vec![],
4706 }
4707 }
4708
4709 fn subprocess_runner(template: &str) -> Runner {
4710 Runner::Subprocess {
4711 template: template.to_string(),
4712 overrides: SubprocessOverrides::default(),
4713 }
4714 }
4715
4716 #[test]
4717 fn validate_placeholders_accepts_closed_set_and_json_braces() {
4718 for ok in [
4719 "{system} {system_file} {prompt} {model} {tools_csv} {work_dir} {task_id} {attempt}",
4720 r#"echo '{"result": "ok", "nested": {"a": 1}}'"#,
4721 "no placeholders at all",
4722 "unmatched { brace",
4723 ] {
4724 validate_embed_placeholders(ok, "ut").expect("must be accepted");
4725 }
4726 }
4727
4728 #[test]
4729 fn validate_placeholders_rejects_unknown_token() {
4730 let err = validate_embed_placeholders("--flag {evil}", "argv[1]").unwrap_err();
4731 assert!(err.contains("'{evil}'"), "token named: {err}");
4732 assert!(err.contains("closed set"), "closed set listed: {err}");
4733 }
4734
4735 #[test]
4739 fn validate_placeholders_descends_into_literal_braces() {
4740 validate_embed_placeholders(r#"{"task": "{prompt}"}"#, "stdin")
4741 .expect("nested closed-set token must be accepted");
4742 let err = validate_embed_placeholders(r#"{"task": "{evil}"}"#, "stdin").unwrap_err();
4743 assert!(
4744 err.contains("'{evil}'"),
4745 "nested unknown token caught: {err}"
4746 );
4747 }
4748
4749 #[test]
4750 fn hint_resolution_finds_declared_template() {
4751 let agent = subprocess_agent("headless", Some(subprocess_runner("echo")));
4752 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
4753 let hint = resolve_subprocess_template_hint(&bp, &agent)
4754 .expect("resolves")
4755 .expect("Runner::Subprocess must synthesize a hint");
4756 assert_eq!(hint[SUBPROCESS_TEMPLATE_HINT_KEY]["name"], "echo");
4757 assert!(hint.get(SUBPROCESS_OVERRIDES_HINT_KEY).is_some());
4758 }
4759
4760 #[test]
4761 fn hint_resolution_unknown_template_is_invalid_spec() {
4762 let agent = subprocess_agent("headless", Some(subprocess_runner("nope")));
4763 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
4764 let err = resolve_subprocess_template_hint(&bp, &agent).unwrap_err();
4765 let msg = format!("{err}");
4766 assert!(msg.contains("'nope'"), "missing template named: {msg}");
4767 assert!(msg.contains("echo"), "defined templates listed: {msg}");
4768 }
4769
4770 #[test]
4771 fn hint_resolution_none_without_subprocess_runner() {
4772 let agent = subprocess_agent("headless", None);
4773 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
4774 let hint = resolve_subprocess_template_hint(&bp, &agent).expect("resolves");
4775 assert!(hint.is_none(), "spec-based agents keep the historical path");
4776 }
4777
4778 fn agent_block_agent(name: &str, runner: Option<Runner>, profile_tools: &[&str]) -> AgentDef {
4785 AgentDef {
4786 name: name.to_string(),
4787 kind: AgentKind::AgentBlock,
4788 spec: serde_json::json!({}),
4789 profile: Some(AgentProfile {
4790 system_prompt: "you are an in-process auditor".to_string(),
4791 tools: profile_tools.iter().map(|t| t.to_string()).collect(),
4792 ..Default::default()
4793 }),
4794 meta: None,
4795 runner,
4796 runner_ref: None,
4797 verdict: None,
4798 lints: None,
4799 }
4800 }
4801
4802 fn agent_block_runner(tools: &[&str]) -> Runner {
4803 Runner::AgentBlockInProcess {
4804 tools: tools.iter().map(|t| t.to_string()).collect(),
4805 }
4806 }
4807
4808 #[test]
4814 fn agent_block_runner_tools_are_projected_over_profile_tools() {
4815 let agent = agent_block_agent(
4816 "auditor",
4817 Some(agent_block_runner(&["mcp__outline__list_docs"])),
4818 &["Read"],
4819 );
4820 let bp = bp_with(vec![agent], vec![]);
4821 let bound = resolve_bound_agents(&bp).expect("binds");
4822 let effective = materialize_bound_blueprint(&bp, &bound);
4823 assert_eq!(
4824 effective.agents[0].profile.as_ref().unwrap().tools,
4825 vec!["mcp__outline__list_docs".to_string()],
4826 "the declared Runner tools replace profile.tools (['Read'])"
4827 );
4828 }
4829
4830 #[test]
4834 fn agent_block_projection_distinguishes_declared_empty_from_absent() {
4835 let declared = agent_block_agent("auditor", Some(agent_block_runner(&[])), &["Read"]);
4836 let bp = bp_with(vec![declared], vec![]);
4837 let bound = resolve_bound_agents(&bp).expect("binds");
4838 let effective = materialize_bound_blueprint(&bp, &bound);
4839 assert!(
4840 effective.agents[0]
4841 .profile
4842 .as_ref()
4843 .unwrap()
4844 .tools
4845 .is_empty(),
4846 "empty means enforced-empty, not 'unset'"
4847 );
4848
4849 let absent = agent_block_agent("auditor", None, &["Read"]);
4850 let bp = bp_with(vec![absent], vec![]);
4851 let bound = resolve_bound_agents(&bp).expect("binds");
4852 let effective = materialize_bound_blueprint(&bp, &bound);
4853 assert_eq!(
4854 effective.agents[0].profile.as_ref().unwrap().tools,
4855 vec!["Read".to_string()],
4856 "no Runner declared → the agent.md tools line stands"
4857 );
4858 }
4859
4860 #[test]
4867 fn compile_rejects_script_mode_with_a_declared_mcp_grant() {
4868 let mut agent = agent_block_agent(
4869 "auditor",
4870 Some(agent_block_runner(&["mcp__outline__list_docs"])),
4871 &[],
4872 );
4873 agent.spec = serde_json::json!({ "script_path": "gate.lua" });
4874 let mut bp = bp_with(vec![agent], vec![]);
4875 bp.strategy.strict_refs = false;
4876
4877 let mut registry = SpawnerRegistry::new();
4878 registry.register::<crate::worker::agent_block::AgentBlockInProcessSpawnerFactory>(
4879 Arc::new(crate::worker::agent_block::AgentBlockInProcessSpawnerFactory::new()),
4880 );
4881 let err = match Compiler::new(registry).compile(&bp) {
4883 Err(e) => e,
4884 Ok(_) => panic!("script mode + declared MCP grant must not compile"),
4885 };
4886 let msg = format!("{err}");
4887 assert!(msg.contains("script_path"), "names the trigger: {msg}");
4888 assert!(
4889 msg.contains("mcp__outline__list_docs"),
4890 "names the unenforceable tools: {msg}"
4891 );
4892 }
4893
4894 #[test]
4898 fn compile_accepts_script_mode_with_only_inert_tools() {
4899 let mut agent = agent_block_agent("auditor", None, &["Read", "WebSearch"]);
4900 agent.spec = serde_json::json!({ "script_path": "gate.lua" });
4901 let mut bp = bp_with(vec![agent], vec![]);
4902 bp.strategy.strict_refs = false;
4903
4904 let mut registry = SpawnerRegistry::new();
4905 registry.register::<crate::worker::agent_block::AgentBlockInProcessSpawnerFactory>(
4906 Arc::new(crate::worker::agent_block::AgentBlockInProcessSpawnerFactory::new()),
4907 );
4908 if let Err(e) = Compiler::new(registry).compile(&bp) {
4909 panic!("inert tools must not trip the MCP-grant guard: {e}");
4910 }
4911 }
4912
4913 #[test]
4914 fn build_embed_rejects_unknown_placeholder() {
4915 let agent = subprocess_agent("headless", None);
4916 let mut def = echo_def("echo");
4917 def.argv.push("--x={evil}".to_string());
4918 let err = SubprocessProcessSpawnerFactory::build_embed(
4919 &agent,
4920 &serde_json::to_value(&def).unwrap(),
4921 None,
4922 )
4923 .unwrap_err();
4924 assert!(format!("{err}").contains("'{evil}'"));
4925 }
4926
4927 #[test]
4928 fn build_embed_rejects_output_with_stream_mode() {
4929 let agent = subprocess_agent("headless", None);
4930 let mut def = echo_def("echo");
4931 def.stream_mode = Some("ndjson_lines".to_string());
4932 def.output = Some(mlua_swarm_schema::SubprocessOutput {
4933 format: Some("json".to_string()),
4934 result_ptr: None,
4935 ok_from: None,
4936 stats: None,
4937 });
4938 let err = SubprocessProcessSpawnerFactory::build_embed(
4939 &agent,
4940 &serde_json::to_value(&def).unwrap(),
4941 None,
4942 )
4943 .unwrap_err();
4944 assert!(format!("{err}").contains("plain-mode"));
4945 }
4946
4947 #[test]
4948 fn build_embed_rejects_malformed_result_ptr_and_ok_from() {
4949 let agent = subprocess_agent("headless", None);
4950 let mut def = echo_def("echo");
4951 def.output = Some(mlua_swarm_schema::SubprocessOutput {
4952 format: None,
4953 result_ptr: Some("result".to_string()),
4954 ok_from: None,
4955 stats: None,
4956 });
4957 let err = SubprocessProcessSpawnerFactory::build_embed(
4958 &agent,
4959 &serde_json::to_value(&def).unwrap(),
4960 None,
4961 )
4962 .unwrap_err();
4963 assert!(format!("{err}").contains("JSON Pointer"));
4964
4965 let mut def = echo_def("echo");
4966 def.output = Some(mlua_swarm_schema::SubprocessOutput {
4967 format: None,
4968 result_ptr: None,
4969 ok_from: Some("status".to_string()),
4970 stats: None,
4971 });
4972 let err = SubprocessProcessSpawnerFactory::build_embed(
4973 &agent,
4974 &serde_json::to_value(&def).unwrap(),
4975 None,
4976 )
4977 .unwrap_err();
4978 assert!(format!("{err}").contains("exit_code"));
4979 }
4980
4981 #[test]
4982 fn build_embed_bakes_profile_with_override_precedence() {
4983 let agent = subprocess_agent("headless", None);
4984 let def = echo_def("echo");
4985 let overrides = SubprocessOverrides {
4986 model: Some("override-model".to_string()),
4987 tools: vec!["Bash".to_string(), "Write".to_string()],
4988 cwd: Some("/tmp/override-wd".to_string()),
4989 };
4990 let sp = SubprocessProcessSpawnerFactory::build_embed(
4991 &agent,
4992 &serde_json::to_value(&def).unwrap(),
4993 Some(&serde_json::to_value(&overrides).unwrap()),
4994 )
4995 .expect("builds");
4996 let embed = sp.embed.as_ref().expect("embed template baked");
4997 assert_eq!(embed.model.as_deref(), Some("override-model"));
4998 assert_eq!(embed.tools_csv, "Bash,Write");
4999 assert_eq!(embed.cwd.as_deref(), Some("/tmp/override-wd"));
5000 assert_eq!(
5001 embed.system_prompt.as_deref(),
5002 Some("you are a headless worker")
5003 );
5004 }
5005}