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, 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::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(
565 &self,
566 bp: &Blueprint,
567 bound_agents: &[BoundAgent],
568 ) -> Result<CompiledBlueprint, CompileError> {
569 let effective = materialize_bound_blueprint(bp, bound_agents);
570 self.compile_resolved(&effective)
571 }
572
573 fn compile_resolved(&self, bp: &Blueprint) -> Result<CompiledBlueprint, CompileError> {
574 let mut routes: HashMap<String, Arc<dyn SpawnerAdapter>> = HashMap::new();
575 let mut seen: HashMap<String, ()> = HashMap::new();
576 let mut verdict_contracts: HashMap<String, VerdictContract> = HashMap::new();
582
583 let defined: Vec<String> = bp.operators.iter().map(|o| o.name.clone()).collect();
589 for ad in &bp.agents {
590 if !matches!(ad.kind, AgentKind::Operator) {
591 continue;
592 }
593 let op_ref = ad.spec.get("operator_ref").and_then(|v| v.as_str());
594 if let Some(op_ref) = op_ref {
595 if !defined.iter().any(|n| n == op_ref) {
596 return Err(CompileError::UnresolvedOperatorRef {
597 agent: ad.name.clone(),
598 op_ref: op_ref.to_string(),
599 defined: defined.clone(),
600 });
601 }
602 }
603 }
605
606 let metas_defined: Vec<String> = bp.metas.iter().map(|m| m.name.clone()).collect();
610 for ad in &bp.agents {
611 let meta_ref = ad.meta.as_ref().and_then(|m| m.meta_ref.as_ref());
612 if let Some(meta_ref) = meta_ref {
613 if !metas_defined.iter().any(|n| n == meta_ref) {
614 return Err(CompileError::UnresolvedMetaRef {
615 where_: format!("AgentMeta.meta_ref of agent '{}'", ad.name),
616 meta_ref: meta_ref.clone(),
617 defined: metas_defined.clone(),
618 });
619 }
620 }
621 }
622 let mut static_step_meta_refs: Vec<(String, String)> = Vec::new();
628 collect_step_meta_refs(&bp.flow, &mut static_step_meta_refs);
629 for (where_, meta_ref) in static_step_meta_refs {
630 if !metas_defined.iter().any(|n| n == &meta_ref) {
631 return Err(CompileError::UnresolvedMetaRef {
632 where_,
633 meta_ref,
634 defined: metas_defined.clone(),
635 });
636 }
637 }
638
639 let agents_defined: Vec<String> = bp.agents.iter().map(|a| a.name.clone()).collect();
644 for audit in &bp.audits {
645 if !agents_defined.iter().any(|n| n == &audit.agent) {
646 return Err(CompileError::UnresolvedAuditAgent {
647 agent: audit.agent.clone(),
648 defined: agents_defined.clone(),
649 });
650 }
651 }
652
653 for ad in &bp.agents {
654 if seen.contains_key(&ad.name) {
655 return Err(CompileError::DuplicateAgent(ad.name.clone()));
656 }
657 seen.insert(ad.name.clone(), ());
658
659 if let Some(contract) = &ad.verdict {
665 verdict_contracts.insert(ad.name.clone(), contract.clone());
666 }
667
668 let factory = match self.registry.factories.get(&ad.kind) {
669 Some(f) => f.clone(),
670 None => {
671 if bp.strategy.strict_kind {
672 return Err(CompileError::UnknownKind(ad.kind.clone()));
673 } else {
674 tracing::warn!(
675 agent = %ad.name,
676 kind = ?ad.kind,
677 "no spawner factory registered for agent kind; \
678 dropping agent from routing table (strict_kind=false)"
679 );
680 continue;
681 }
682 }
683 };
684 let hint = bp.hints.per_agent.get(&ad.name);
685 let subprocess_hint = if ad.kind == AgentKind::Subprocess {
691 resolve_subprocess_template_hint(bp, ad)?
692 } else {
693 None
694 };
695 let spawner = factory.build(ad, subprocess_hint.as_ref().or(hint))?;
696 routes.insert(ad.name.clone(), spawner);
697 }
698
699 let strict_verdict_handling = bp.metadata.strict_verdict_handling.unwrap_or(false);
716 verify_verdict_conds(&bp.flow, &verdict_contracts, strict_verdict_handling)?;
717
718 if bp.strategy.strict_refs {
719 verify_refs(&bp.flow, &routes, self.default_spawner.is_some())?;
720 }
721
722 let (step_naming, step_naming_warnings) = StepNaming::from_blueprint(bp)?;
730 for warning in &step_naming_warnings {
731 tracing::warn!(
732 name = %warning.name,
733 first_step_ref = %warning.first_step_ref,
734 second_step_ref = %warning.second_step_ref,
735 "StepNaming: undeclared steps' canonical/alias names collide; \
736 the step whose own ref matches the name keeps it (data-plane priority)"
737 );
738 }
739
740 let projection_placement =
748 ProjectionPlacement::from_spec(bp.projection_placement.as_ref())?;
749
750 let router = Arc::new(CompiledAgentTable {
751 routes,
752 default: self.default_spawner.clone(),
753 verdict_contracts,
754 });
755 Ok(CompiledBlueprint {
756 router,
757 flow: bp.flow.clone(),
758 metadata: bp.metadata.clone(),
759 step_naming: Arc::new(step_naming),
760 projection_placement: Arc::new(projection_placement),
761 })
762 }
763}
764
765fn verify_refs(
768 node: &FlowNode,
769 routes: &HashMap<String, Arc<dyn SpawnerAdapter>>,
770 has_default: bool,
771) -> Result<(), CompileError> {
772 let mut refs: Vec<String> = Vec::new();
773 collect_refs(node, &mut refs);
774 for r in refs {
775 if !routes.contains_key(&r) && !has_default {
776 return Err(CompileError::UnresolvedRef(r));
777 }
778 }
779 Ok(())
780}
781
782fn collect_refs(node: &FlowNode, out: &mut Vec<String>) {
783 match node {
784 FlowNode::Step { ref_, .. } => out.push(ref_.clone()),
785 FlowNode::Seq { children } => {
786 for c in children {
787 collect_refs(c, out);
788 }
789 }
790 FlowNode::Branch { then_, else_, .. } => {
791 collect_refs(then_, out);
792 collect_refs(else_, out);
793 }
794 FlowNode::Fanout { body, .. } => collect_refs(body, out),
795 FlowNode::Loop { body, .. } => collect_refs(body, out),
796 FlowNode::Try { body, catch, .. } => {
797 collect_refs(body, out);
798 collect_refs(catch, out);
799 }
800 FlowNode::Assign { .. } => {} }
802}
803
804fn collect_step_meta_refs(node: &FlowNode, out: &mut Vec<(String, String)>) {
812 match node {
813 FlowNode::Step { ref_, in_, .. } => {
814 if let Expr::Lit { value } = in_ {
815 if let Some(meta_ref) = static_step_meta_ref(value) {
816 out.push((format!("Step '{ref_}' $step_meta.ref"), meta_ref));
817 }
818 }
819 }
820 FlowNode::Seq { children } => {
821 for c in children {
822 collect_step_meta_refs(c, out);
823 }
824 }
825 FlowNode::Branch { then_, else_, .. } => {
826 collect_step_meta_refs(then_, out);
827 collect_step_meta_refs(else_, out);
828 }
829 FlowNode::Fanout { body, .. } => collect_step_meta_refs(body, out),
830 FlowNode::Loop { body, .. } => collect_step_meta_refs(body, out),
831 FlowNode::Try { body, catch, .. } => {
832 collect_step_meta_refs(body, out);
833 collect_step_meta_refs(catch, out);
834 }
835 FlowNode::Assign { .. } => {} }
837}
838
839fn static_step_meta_ref(value: &Value) -> Option<String> {
846 value
847 .as_object()?
848 .get("$step_meta")?
849 .as_object()?
850 .get("ref")?
851 .as_str()
852 .map(str::to_string)
853}
854
855fn verify_verdict_conds(
867 flow: &FlowNode,
868 verdict_contracts: &HashMap<String, VerdictContract>,
869 strict_verdict_handling: bool,
870) -> Result<(), CompileError> {
871 let mut step_outputs: HashMap<String, String> = HashMap::new();
872 let mut step_agents: HashMap<String, String> = HashMap::new();
873 collect_step_outputs_and_agents(flow, &mut step_outputs, &mut step_agents);
874
875 let mut errors: Vec<CompileError> = Vec::new();
876 let mut referenced_values: HashMap<String, std::collections::HashSet<String>> = HashMap::new();
877 collect_verdict_conds(
878 flow,
879 &step_outputs,
880 verdict_contracts,
881 &mut referenced_values,
882 &mut errors,
883 );
884 check_unhandled_verdict_values(
885 verdict_contracts,
886 &referenced_values,
887 &step_agents,
888 strict_verdict_handling,
889 &mut errors,
890 );
891 match errors.into_iter().next() {
892 Some(e) => Err(e),
893 None => Ok(()),
894 }
895}
896
897fn collect_step_outputs_and_agents(
912 node: &FlowNode,
913 out: &mut HashMap<String, String>,
914 step_agents: &mut HashMap<String, String>,
915) {
916 match node {
917 FlowNode::Step {
918 ref_,
919 out: out_expr,
920 ..
921 } => {
922 if let Expr::Path { at } = out_expr {
923 out.insert(at.to_string(), ref_.clone());
924 }
925 step_agents
926 .entry(ref_.clone())
927 .or_insert_with(|| ref_.clone());
928 }
929 FlowNode::Seq { children } => {
930 for c in children {
931 collect_step_outputs_and_agents(c, out, step_agents);
932 }
933 }
934 FlowNode::Branch { then_, else_, .. } => {
935 collect_step_outputs_and_agents(then_, out, step_agents);
936 collect_step_outputs_and_agents(else_, out, step_agents);
937 }
938 FlowNode::Fanout { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
939 FlowNode::Loop { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
940 FlowNode::Try { body, catch, .. } => {
941 collect_step_outputs_and_agents(body, out, step_agents);
942 collect_step_outputs_and_agents(catch, out, step_agents);
943 }
944 FlowNode::Assign { .. } => {} }
946}
947
948fn collect_verdict_conds(
953 node: &FlowNode,
954 step_outputs: &HashMap<String, String>,
955 verdict_contracts: &HashMap<String, VerdictContract>,
956 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
957 errors: &mut Vec<CompileError>,
958) {
959 match node {
960 FlowNode::Branch { cond, then_, else_ } => {
961 lint_cond_expr(
962 cond,
963 "Branch cond",
964 step_outputs,
965 verdict_contracts,
966 referenced_values,
967 errors,
968 );
969 collect_verdict_conds(
970 then_,
971 step_outputs,
972 verdict_contracts,
973 referenced_values,
974 errors,
975 );
976 collect_verdict_conds(
977 else_,
978 step_outputs,
979 verdict_contracts,
980 referenced_values,
981 errors,
982 );
983 }
984 FlowNode::Loop { cond, body, .. } => {
985 lint_cond_expr(
986 cond,
987 "Loop cond",
988 step_outputs,
989 verdict_contracts,
990 referenced_values,
991 errors,
992 );
993 collect_verdict_conds(
994 body,
995 step_outputs,
996 verdict_contracts,
997 referenced_values,
998 errors,
999 );
1000 }
1001 FlowNode::Seq { children } => {
1002 for c in children {
1003 collect_verdict_conds(
1004 c,
1005 step_outputs,
1006 verdict_contracts,
1007 referenced_values,
1008 errors,
1009 );
1010 }
1011 }
1012 FlowNode::Fanout { body, .. } => collect_verdict_conds(
1013 body,
1014 step_outputs,
1015 verdict_contracts,
1016 referenced_values,
1017 errors,
1018 ),
1019 FlowNode::Try { body, catch, .. } => {
1020 collect_verdict_conds(
1021 body,
1022 step_outputs,
1023 verdict_contracts,
1024 referenced_values,
1025 errors,
1026 );
1027 collect_verdict_conds(
1028 catch,
1029 step_outputs,
1030 verdict_contracts,
1031 referenced_values,
1032 errors,
1033 );
1034 }
1035 FlowNode::Step { .. } | FlowNode::Assign { .. } => {}
1036 }
1037}
1038
1039fn lint_cond_expr(
1048 expr: &Expr,
1049 where_: &str,
1050 step_outputs: &HashMap<String, String>,
1051 verdict_contracts: &HashMap<String, VerdictContract>,
1052 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
1053 errors: &mut Vec<CompileError>,
1054) {
1055 match expr {
1056 Expr::Eq { lhs, rhs } | Expr::Ne { lhs, rhs } => {
1057 if let Some((path, lit)) = path_lit_operands(lhs, rhs) {
1058 resolve_and_check(
1059 path,
1060 &[lit],
1061 where_,
1062 step_outputs,
1063 verdict_contracts,
1064 referenced_values,
1065 errors,
1066 );
1067 }
1068 }
1069 Expr::In { needle, haystack } => {
1070 if let (
1071 Expr::Path { at },
1072 Expr::Lit {
1073 value: Value::Array(items),
1074 },
1075 ) = (needle.as_ref(), haystack.as_ref())
1076 {
1077 let lits: Vec<&Value> = items.iter().collect();
1078 resolve_and_check(
1079 at,
1080 &lits,
1081 where_,
1082 step_outputs,
1083 verdict_contracts,
1084 referenced_values,
1085 errors,
1086 );
1087 }
1088 }
1089 Expr::And { args } | Expr::Or { args } => {
1090 for a in args {
1091 lint_cond_expr(
1092 a,
1093 where_,
1094 step_outputs,
1095 verdict_contracts,
1096 referenced_values,
1097 errors,
1098 );
1099 }
1100 }
1101 Expr::Not { arg } => lint_cond_expr(
1102 arg,
1103 where_,
1104 step_outputs,
1105 verdict_contracts,
1106 referenced_values,
1107 errors,
1108 ),
1109 _ => {}
1110 }
1111}
1112
1113fn path_lit_operands<'a>(lhs: &'a Expr, rhs: &'a Expr) -> Option<(&'a Path, &'a Value)> {
1119 match (lhs, rhs) {
1120 (Expr::Path { at }, Expr::Lit { value }) => Some((at, value)),
1121 (Expr::Lit { value }, Expr::Path { at }) => Some((at, value)),
1122 _ => None,
1123 }
1124}
1125
1126fn resolve_and_check(
1141 path: &Path,
1142 lits: &[&Value],
1143 where_: &str,
1144 step_outputs: &HashMap<String, String>,
1145 verdict_contracts: &HashMap<String, VerdictContract>,
1146 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
1147 errors: &mut Vec<CompileError>,
1148) {
1149 let path_str = path.to_string();
1150 let (agent, actual_shape) = if let Some(agent) = step_outputs.get(&path_str) {
1151 (agent, "body")
1152 } else if let Some(stripped) = path_str.strip_suffix(".parts.verdict") {
1153 match step_outputs.get(stripped) {
1154 Some(agent) => (agent, "part"),
1155 None => return,
1156 }
1157 } else {
1158 return;
1159 };
1160
1161 let Some(contract) = verdict_contracts.get(agent) else {
1162 tracing::warn!(
1163 agent = %agent,
1164 where_ = %where_,
1165 "cond references agent output but no verdict contract declared"
1166 );
1167 return;
1168 };
1169
1170 let expected_channel = match contract.channel {
1171 VerdictChannel::Body => "body",
1172 VerdictChannel::Part => "part",
1173 };
1174 if expected_channel != actual_shape {
1175 errors.push(CompileError::VerdictChannelMismatch {
1176 where_: where_.to_string(),
1177 agent: agent.clone(),
1178 expected_channel: expected_channel.to_string(),
1179 actual_shape: actual_shape.to_string(),
1180 });
1181 return;
1182 }
1183
1184 for lit in lits {
1185 let value_str = lit
1186 .as_str()
1187 .map(str::to_string)
1188 .unwrap_or_else(|| lit.to_string());
1189 if !contract.values.iter().any(|v| v == &value_str) {
1190 errors.push(CompileError::VerdictValueNotInContract {
1191 where_: where_.to_string(),
1192 agent: agent.clone(),
1193 value: value_str.clone(),
1194 values: contract.values.clone(),
1195 });
1196 }
1197 referenced_values
1204 .entry(agent.clone())
1205 .or_default()
1206 .insert(value_str);
1207 }
1208}
1209
1210fn check_unhandled_verdict_values(
1228 verdict_contracts: &HashMap<String, VerdictContract>,
1229 referenced_values: &HashMap<String, std::collections::HashSet<String>>,
1230 step_agents: &HashMap<String, String>,
1231 strict_verdict_handling: bool,
1232 errors: &mut Vec<CompileError>,
1233) {
1234 let mut agents: Vec<&String> = verdict_contracts.keys().collect();
1240 agents.sort();
1241 for agent in agents {
1242 let contract = &verdict_contracts[agent];
1243 let referenced = referenced_values.get(agent);
1244 let step_ref = step_agents
1245 .get(agent)
1246 .cloned()
1247 .unwrap_or_else(|| agent.clone());
1248 for value in &contract.values {
1249 let handled = referenced.map(|set| set.contains(value)).unwrap_or(false);
1250 if handled {
1251 continue;
1252 }
1253 if strict_verdict_handling {
1254 errors.push(CompileError::VerdictValueUnhandled {
1255 agent: agent.clone(),
1256 value: value.clone(),
1257 declared_values: contract.values.clone(),
1258 step_ref: step_ref.clone(),
1259 });
1260 } else {
1261 tracing::warn!(
1262 agent = %agent,
1263 value = %value,
1264 step_ref = %step_ref,
1265 "declared verdict value has no downstream cond handler; \
1266 opt in to `metadata.strict_verdict_handling` to reject at compile"
1267 );
1268 }
1269 }
1270 }
1271}
1272
1273pub struct CompiledAgentTable {
1286 pub(crate) routes: HashMap<String, Arc<dyn SpawnerAdapter>>,
1287 pub(crate) default: Option<Arc<dyn SpawnerAdapter>>,
1288 pub(crate) verdict_contracts: HashMap<String, VerdictContract>,
1292}
1293
1294impl CompiledAgentTable {
1295 pub fn has_route(&self, agent: &str) -> bool {
1298 self.routes.contains_key(agent)
1299 }
1300 pub fn routed_agents(&self) -> Vec<String> {
1302 self.routes.keys().cloned().collect()
1303 }
1304 pub fn verdict_contract_for(&self, agent: &str) -> Option<&VerdictContract> {
1308 self.verdict_contracts.get(agent)
1309 }
1310}
1311
1312#[async_trait]
1313impl SpawnerAdapter for CompiledAgentTable {
1314 async fn spawn(
1315 &self,
1316 engine: &Engine,
1317 ctx: &Ctx,
1318 task_id: StepId,
1319 attempt: u32,
1320 token: CapToken,
1321 ) -> Result<Box<dyn Worker>, SpawnError> {
1322 let sp = self
1323 .routes
1324 .get(&ctx.agent)
1325 .cloned()
1326 .or_else(|| self.default.clone())
1327 .ok_or_else(|| SpawnError::NotRegistered(ctx.agent.clone()))?;
1328 sp.spawn(engine, ctx, task_id, attempt, token).await
1329 }
1330}
1331
1332pub struct SubprocessProcessSpawnerFactory;
1363
1364impl SpawnerFactoryKind for SubprocessProcessSpawnerFactory {
1365 const KIND: AgentKind = AgentKind::Subprocess;
1366 type Worker = crate::worker::process_spawner::ProcessWorker;
1367}
1368
1369pub const SUBPROCESS_TEMPLATE_HINT_KEY: &str = "subprocess_template";
1372pub const SUBPROCESS_OVERRIDES_HINT_KEY: &str = "subprocess_overrides";
1374
1375fn validate_embed_placeholders(s: &str, where_: &str) -> Result<(), String> {
1380 let mut rest = s;
1381 while let Some(start) = rest.find('{') {
1382 let after = &rest[start + 1..];
1383 let Some(end) = after.find('}') else {
1384 break;
1385 };
1386 let token = &after[..end];
1387 let is_candidate =
1388 !token.is_empty() && token.chars().all(|c| c.is_ascii_lowercase() || c == '_');
1389 if is_candidate {
1390 if !crate::worker::process_spawner::EMBED_PLACEHOLDERS.contains(&token) {
1391 return Err(format!(
1392 "unknown placeholder '{{{token}}}' in {where_}; closed set is \
1393 {{system, system_file, prompt, model, tools_csv, work_dir, task_id, attempt}}"
1394 ));
1395 }
1396 rest = &after[end + 1..];
1397 } else {
1398 rest = after;
1403 }
1404 }
1405 Ok(())
1406}
1407
1408fn resolve_subprocess_template_hint(
1414 bp: &Blueprint,
1415 ad: &AgentDef,
1416) -> Result<Option<Value>, CompileError> {
1417 let invalid = |msg: String| CompileError::InvalidSpec {
1418 name: ad.name.clone(),
1419 msg,
1420 };
1421 let runner = mlua_swarm_schema::resolve_runner(bp, ad).map_err(|e| invalid(e.to_string()))?;
1422 let Some(Runner::Subprocess {
1423 template,
1424 overrides,
1425 }) = runner
1426 else {
1427 return Ok(None);
1428 };
1429 let def = bp
1430 .subprocesses
1431 .iter()
1432 .find(|d| d.name == template)
1433 .ok_or_else(|| {
1434 let mut names: Vec<&str> = bp.subprocesses.iter().map(|d| d.name.as_str()).collect();
1435 names.sort_unstable();
1436 invalid(format!(
1437 "Runner::Subprocess template '{template}' not found in \
1438 Blueprint.subprocesses (defined: [{}])",
1439 names.join(", ")
1440 ))
1441 })?;
1442 Ok(Some(serde_json::json!({
1443 SUBPROCESS_TEMPLATE_HINT_KEY: def,
1444 SUBPROCESS_OVERRIDES_HINT_KEY: overrides,
1445 })))
1446}
1447
1448impl SubprocessProcessSpawnerFactory {
1449 fn build_embed(
1454 agent_def: &AgentDef,
1455 template: &Value,
1456 overrides: Option<&Value>,
1457 ) -> Result<ProcessSpawner, CompileError> {
1458 use crate::worker::process_spawner::EmbedTemplate;
1459 use mlua_swarm_schema::{SubprocessDef, SubprocessOverrides};
1460
1461 let agent_name = &agent_def.name;
1462 let invalid = |msg: String| CompileError::InvalidSpec {
1463 name: agent_name.to_string(),
1464 msg,
1465 };
1466 let def: SubprocessDef = serde_json::from_value(template.clone())
1467 .map_err(|e| invalid(format!("subprocess_template hint: {e}")))?;
1468 let overrides: SubprocessOverrides = match overrides {
1469 Some(v) => serde_json::from_value(v.clone())
1470 .map_err(|e| invalid(format!("subprocess_overrides hint: {e}")))?,
1471 None => SubprocessOverrides::default(),
1472 };
1473
1474 if def.argv.is_empty() {
1475 return Err(invalid(format!(
1476 "SubprocessDef '{}': argv must not be empty",
1477 def.name
1478 )));
1479 }
1480 for (i, a) in def.argv.iter().enumerate() {
1482 validate_embed_placeholders(a, &format!("argv[{i}]")).map_err(&invalid)?;
1483 }
1484 if let Some(stdin) = &def.stdin {
1485 validate_embed_placeholders(stdin, "stdin").map_err(&invalid)?;
1486 }
1487 for (k, v) in &def.env {
1488 validate_embed_placeholders(v, &format!("env['{k}']")).map_err(&invalid)?;
1489 }
1490 if let Some(cwd) = &def.cwd {
1491 validate_embed_placeholders(cwd, "cwd").map_err(&invalid)?;
1492 }
1493 let stream_mode = match def.stream_mode.as_deref() {
1494 Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
1495 Some("sse_events") => Some(StreamMode::SseEvents),
1496 Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
1497 Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
1498 None => None,
1499 };
1500 if let Some(output) = &def.output {
1501 if stream_mode.is_some() {
1502 return Err(invalid(format!(
1503 "SubprocessDef '{}': output normalization is a plain-mode \
1504 declaration; remove either `output` or `stream_mode`",
1505 def.name
1506 )));
1507 }
1508 if let Some(format) = output.format.as_deref() {
1509 if format != "json" {
1510 return Err(invalid(format!(
1511 "SubprocessDef '{}': unknown output.format '{format}' \
1512 (supported: \"json\")",
1513 def.name
1514 )));
1515 }
1516 }
1517 if let Some(ptr) = output.result_ptr.as_deref() {
1518 if !ptr.starts_with('/') {
1519 return Err(invalid(format!(
1520 "SubprocessDef '{}': output.result_ptr '{ptr}' is not a \
1521 JSON Pointer (RFC 6901 — must start with '/')",
1522 def.name
1523 )));
1524 }
1525 }
1526 if let Some(ok_from) = output.ok_from.as_deref() {
1527 if ok_from != "exit_code" && !ok_from.starts_with('/') {
1528 return Err(invalid(format!(
1529 "SubprocessDef '{}': output.ok_from '{ok_from}' must be \
1530 \"exit_code\" or a JSON Pointer (starting with '/')",
1531 def.name
1532 )));
1533 }
1534 }
1535 }
1536
1537 let profile = agent_def.profile.as_ref();
1540 let system_prompt = profile
1541 .map(|p| p.system_prompt.clone())
1542 .filter(|s| !s.is_empty());
1543 let model = overrides
1544 .model
1545 .clone()
1546 .or_else(|| profile.and_then(|p| p.model.clone()));
1547 let tools: Vec<String> = if overrides.tools.is_empty() {
1548 profile.map(|p| p.tools.clone()).unwrap_or_default()
1549 } else {
1550 overrides.tools.clone()
1551 };
1552 let cwd = overrides.cwd.clone().or_else(|| def.cwd.clone());
1554 if let Some(c) = &cwd {
1555 validate_embed_placeholders(c, "overrides.cwd").map_err(&invalid)?;
1556 }
1557
1558 let program = def.argv[0].clone();
1559 let sp = ProcessSpawner {
1560 program,
1561 args: Vec::new(),
1562 use_stdin: def.stdin.is_some(),
1563 stream_mode,
1564 embed: Some(EmbedTemplate {
1565 argv: def.argv,
1566 stdin: def.stdin,
1567 env: def.env,
1568 cwd,
1569 output: def.output,
1570 system_prompt,
1571 model,
1572 tools_csv: tools.join(","),
1573 }),
1574 };
1575 Ok(sp)
1576 }
1577}
1578
1579impl SpawnerFactory for SubprocessProcessSpawnerFactory {
1580 fn build(
1581 &self,
1582 agent_def: &AgentDef,
1583 hint: Option<&Value>,
1584 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1585 if let Some(template) = hint.and_then(|h| h.get(SUBPROCESS_TEMPLATE_HINT_KEY)) {
1589 let overrides = hint.and_then(|h| h.get(SUBPROCESS_OVERRIDES_HINT_KEY));
1590 return Self::build_embed(agent_def, template, overrides).map(|sp| {
1591 let arc: Arc<dyn SpawnerAdapter> = Arc::new(sp);
1592 arc
1593 });
1594 }
1595 let agent_name = &agent_def.name;
1596 let spec = &agent_def.spec;
1597 let invalid = |msg: String| CompileError::InvalidSpec {
1598 name: agent_name.to_string(),
1599 msg,
1600 };
1601 let program = spec
1602 .get("program")
1603 .and_then(|v| v.as_str())
1604 .ok_or_else(|| invalid("shell spec: 'program' (string) required".into()))?
1605 .to_string();
1606 let args: Vec<String> = spec
1607 .get("args")
1608 .and_then(|v| v.as_array())
1609 .map(|a| {
1610 a.iter()
1611 .filter_map(|x| x.as_str().map(|s| s.to_string()))
1612 .collect()
1613 })
1614 .unwrap_or_default();
1615 let use_stdin = spec
1616 .get("use_stdin")
1617 .and_then(|v| v.as_bool())
1618 .unwrap_or(true);
1619 let stream_mode = match spec.get("stream_mode").and_then(|v| v.as_str()) {
1620 Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
1621 Some("sse_events") => Some(StreamMode::SseEvents),
1622 Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
1623 Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
1624 None => None,
1625 };
1626
1627 let mut sp = ProcessSpawner {
1628 program,
1629 args,
1630 use_stdin,
1631 stream_mode,
1632 embed: None,
1633 };
1634 if let Some(mode) = sp.stream_mode.clone() {
1635 sp = sp.stream_mode(mode);
1636 }
1637 Ok(Arc::new(sp))
1638 }
1639}
1640
1641pub struct LuaInProcessSpawnerFactory {
1672 registry: HashMap<String, WorkerFn>,
1673 bridges: HashMap<String, HostBridge>,
1674}
1675
1676#[derive(Clone)]
1688pub struct HostBridge(
1689 Arc<dyn Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync>,
1690);
1691
1692impl HostBridge {
1693 pub fn new<F>(f: F) -> Self
1695 where
1696 F: Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync + 'static,
1697 {
1698 Self(Arc::new(f))
1699 }
1700
1701 pub fn call(&self, arg: serde_json::Value) -> Result<serde_json::Value, String> {
1705 (self.0)(arg)
1706 }
1707}
1708
1709#[derive(Clone)]
1716pub struct LuaScriptSource {
1717 pub source: String,
1719 pub label: String,
1722}
1723
1724impl LuaScriptSource {
1725 pub fn new(source: impl Into<String>, label: impl Into<String>) -> Self {
1727 Self {
1728 source: source.into(),
1729 label: label.into(),
1730 }
1731 }
1732}
1733
1734impl LuaInProcessSpawnerFactory {
1735 pub fn new() -> Self {
1737 Self {
1738 registry: HashMap::new(),
1739 bridges: HashMap::new(),
1740 }
1741 }
1742
1743 pub fn with_bridge(mut self, name: impl Into<String>, bridge: HostBridge) -> Self {
1750 self.bridges.insert(name.into(), bridge);
1751 self
1752 }
1753
1754 pub fn register_lua(mut self, fn_id: impl Into<String>, source: LuaScriptSource) -> Self {
1772 let source = Arc::new(source);
1773 let bridges = Arc::new(self.bridges.clone());
1774 let wrapped: WorkerFn = Arc::new(move |inv| {
1775 let source = source.clone();
1776 let bridges = bridges.clone();
1777 Box::pin(run_lua_worker(source, bridges, inv))
1778 });
1779 self.registry.insert(fn_id.into(), wrapped);
1780 self
1781 }
1782}
1783
1784async fn run_lua_worker(
1786 source: Arc<LuaScriptSource>,
1787 bridges: Arc<HashMap<String, HostBridge>>,
1788 inv: crate::worker::adapter::WorkerInvocation,
1789) -> Result<crate::worker::adapter::WorkerResult, crate::worker::adapter::WorkerError> {
1790 use crate::worker::adapter::WorkerError;
1791 use mlua::LuaSerdeExt;
1792
1793 let label = source.label.clone();
1794 let outcome =
1795 tokio::task::spawn_blocking(move || -> Result<(serde_json::Value, bool), String> {
1796 let lua = mlua::Lua::new();
1797 let g = lua.globals();
1798
1799 g.set("_PROMPT", inv.prompt.clone())
1801 .map_err(|e| format!("set _PROMPT: {e}"))?;
1802 g.set("_AGENT", inv.agent.clone())
1803 .map_err(|e| format!("set _AGENT: {e}"))?;
1804 g.set("_TASK_ID", inv.task_id.to_string())
1805 .map_err(|e| format!("set _TASK_ID: {e}"))?;
1806 g.set("_ATTEMPT", inv.attempt as i64)
1807 .map_err(|e| format!("set _ATTEMPT: {e}"))?;
1808
1809 if let Ok(json_val) = serde_json::from_str::<serde_json::Value>(&inv.prompt) {
1811 let lua_val = lua
1812 .to_value(&json_val)
1813 .map_err(|e| format!("_CTX to_value: {e}"))?;
1814 g.set("_CTX", lua_val)
1815 .map_err(|e| format!("set _CTX: {e}"))?;
1816 }
1817
1818 if !bridges.is_empty() {
1820 let host = lua
1821 .create_table()
1822 .map_err(|e| format!("create host table: {e}"))?;
1823 for (name, bridge) in bridges.iter() {
1824 let bridge = bridge.clone();
1825 let bname = name.clone();
1826 let f = lua
1827 .create_function(move |lua, arg: mlua::Value| {
1828 let json_arg: serde_json::Value = lua.from_value(arg).map_err(|e| {
1829 mlua::Error::external(format!("bridge {bname} arg → json: {e}"))
1830 })?;
1831 let result_json =
1832 bridge.call(json_arg).map_err(mlua::Error::external)?;
1833 lua.to_value(&result_json).map_err(|e| {
1834 mlua::Error::external(format!("bridge {bname} ret → lua: {e}"))
1835 })
1836 })
1837 .map_err(|e| format!("create_function {name}: {e}"))?;
1838 host.set(name.as_str(), f)
1839 .map_err(|e| format!("host.{name} set: {e}"))?;
1840 }
1841 g.set("host", host).map_err(|e| format!("set host: {e}"))?;
1842 }
1843
1844 let result: mlua::Value = lua
1846 .load(&source.source)
1847 .set_name(&source.label)
1848 .eval()
1849 .map_err(|e| format!("lua eval [{}]: {e}", source.label))?;
1850
1851 let json_result: serde_json::Value = lua
1853 .from_value(result)
1854 .map_err(|e| format!("lua → json [{}]: {e}", source.label))?;
1855
1856 let (value, ok) = match &json_result {
1857 serde_json::Value::Object(map)
1858 if map.contains_key("value") || map.contains_key("ok") =>
1859 {
1860 let ok = map.get("ok").and_then(|v| v.as_bool()).unwrap_or(true);
1861 let value = map.get("value").cloned().unwrap_or(json_result.clone());
1862 (value, ok)
1863 }
1864 _ => (json_result, true),
1865 };
1866 Ok((value, ok))
1867 })
1868 .await
1869 .map_err(|e| WorkerError::Failed(format!("spawn_blocking join [{label}]: {e}")))?
1870 .map_err(WorkerError::Failed)?;
1871
1872 Ok(crate::worker::adapter::WorkerResult {
1873 value: outcome.0,
1874 ok: outcome.1,
1875 })
1876}
1877
1878impl Default for LuaInProcessSpawnerFactory {
1879 fn default() -> Self {
1880 Self::new()
1881 }
1882}
1883
1884impl SpawnerFactoryKind for LuaInProcessSpawnerFactory {
1885 const KIND: AgentKind = AgentKind::Lua;
1886 type Worker = LuaWorker;
1887}
1888
1889impl SpawnerFactory for LuaInProcessSpawnerFactory {
1890 fn build(
1891 &self,
1892 agent_def: &AgentDef,
1893 _hint: Option<&Value>,
1894 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1895 if let Some(source) = agent_def.spec.get("source").and_then(|v| v.as_str()) {
1901 let label = agent_def
1902 .spec
1903 .get("label")
1904 .and_then(|v| v.as_str())
1905 .map(str::to_string)
1906 .unwrap_or_else(|| format!("{}.lua", agent_def.name));
1907 let script = Arc::new(LuaScriptSource::new(source.to_string(), label));
1908 let bridges = Arc::new(self.bridges.clone());
1909 let wrapped: WorkerFn = Arc::new(move |inv| {
1910 let source = script.clone();
1911 let bridges = bridges.clone();
1912 Box::pin(run_lua_worker(source, bridges, inv))
1913 });
1914 let mut sp: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
1915 sp.registry.insert(agent_def.name.to_string(), wrapped);
1916 return Ok(Arc::new(sp));
1917 }
1918 build_inproc_from_registry::<LuaWorker>(&self.registry, agent_def, "lua")
1919 }
1920}
1921
1922pub struct RustFnInProcessSpawnerFactory {
1936 registry: HashMap<String, WorkerFn>,
1937}
1938
1939impl RustFnInProcessSpawnerFactory {
1940 pub fn new() -> Self {
1942 Self {
1943 registry: HashMap::new(),
1944 }
1945 }
1946
1947 pub fn register_fn<F, Fut>(mut self, fn_id: impl Into<String>, f: F) -> Self
1950 where
1951 F: Fn(crate::worker::adapter::WorkerInvocation) -> Fut + Send + Sync + 'static,
1952 Fut: std::future::Future<
1953 Output = Result<
1954 crate::worker::adapter::WorkerResult,
1955 crate::worker::adapter::WorkerError,
1956 >,
1957 > + Send
1958 + 'static,
1959 {
1960 let f = Arc::new(f);
1961 let wrapped: WorkerFn = Arc::new(move |inv| {
1962 let f = f.clone();
1963 Box::pin(f(inv))
1964 });
1965 self.registry.insert(fn_id.into(), wrapped);
1966 self
1967 }
1968}
1969
1970impl Default for RustFnInProcessSpawnerFactory {
1971 fn default() -> Self {
1972 Self::new()
1973 }
1974}
1975
1976impl SpawnerFactoryKind for RustFnInProcessSpawnerFactory {
1977 const KIND: AgentKind = AgentKind::RustFn;
1978 type Worker = RustFnWorker;
1979}
1980
1981impl SpawnerFactory for RustFnInProcessSpawnerFactory {
1982 fn build(
1983 &self,
1984 agent_def: &AgentDef,
1985 _hint: Option<&Value>,
1986 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1987 build_inproc_from_registry::<RustFnWorker>(&self.registry, agent_def, "rust_fn")
1988 }
1989}
1990
1991fn build_inproc_from_registry<W>(
1997 registry: &HashMap<String, WorkerFn>,
1998 agent_def: &AgentDef,
1999 kind_label: &str,
2000) -> Result<Arc<dyn SpawnerAdapter>, CompileError>
2001where
2002 W: crate::worker::Worker + From<crate::worker::WorkerJoinHandler> + Send + Sync + 'static,
2003{
2004 let agent_name = &agent_def.name;
2005 let spec = &agent_def.spec;
2006 let invalid = |msg: String| CompileError::InvalidSpec {
2007 name: agent_name.to_string(),
2008 msg,
2009 };
2010 let fn_id = spec
2011 .get("fn_id")
2012 .and_then(|v| v.as_str())
2013 .ok_or_else(|| invalid(format!("{kind_label} spec: 'fn_id' (string) required")))?;
2014 let f = registry
2015 .get(fn_id)
2016 .cloned()
2017 .ok_or_else(|| invalid(format!("fn_id '{fn_id}' not registered in factory")))?;
2018 let mut sp: InProcSpawner<W> = InProcSpawner::<W>::typed();
2019 sp.registry.insert(agent_name.to_string(), f);
2023 Ok(Arc::new(sp))
2024}
2025
2026pub struct LuaWorker {
2031 pub handler: crate::worker::WorkerJoinHandler,
2033}
2034
2035impl From<crate::worker::WorkerJoinHandler> for LuaWorker {
2036 fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
2037 Self { handler }
2038 }
2039}
2040
2041#[async_trait::async_trait]
2042impl crate::worker::Worker for LuaWorker {
2043 fn id(&self) -> &crate::types::WorkerId {
2044 &self.handler.worker_id
2045 }
2046 fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
2047 self.handler.cancel.clone()
2048 }
2049 async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
2050 self.handler.await_completion().await
2051 }
2052}
2053
2054pub struct RustFnWorker {
2059 pub handler: crate::worker::WorkerJoinHandler,
2061}
2062
2063impl From<crate::worker::WorkerJoinHandler> for RustFnWorker {
2064 fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
2065 Self { handler }
2066 }
2067}
2068
2069#[async_trait::async_trait]
2070impl crate::worker::Worker for RustFnWorker {
2071 fn id(&self) -> &crate::types::WorkerId {
2072 &self.handler.worker_id
2073 }
2074 fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
2075 self.handler.cancel.clone()
2076 }
2077 async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
2078 self.handler.await_completion().await
2079 }
2080}
2081
2082pub struct OperatorSpawnerFactory {
2135 operators: Arc<std::sync::RwLock<HashMap<String, Arc<dyn Operator>>>>,
2136}
2137
2138impl OperatorSpawnerFactory {
2139 pub fn new() -> Self {
2141 Self {
2142 operators: Arc::new(std::sync::RwLock::new(HashMap::new())),
2143 }
2144 }
2145
2146 pub fn register_operator(&self, id: impl Into<String>, op: Arc<dyn Operator>) -> &Self {
2152 self.operators
2153 .write()
2154 .expect("OperatorSpawnerFactory.operators RwLock poisoned")
2155 .insert(id.into(), op);
2156 self
2157 }
2158
2159 pub fn unregister_operator(&self, id: &str) -> &Self {
2162 self.operators
2163 .write()
2164 .expect("OperatorSpawnerFactory.operators RwLock poisoned")
2165 .remove(id);
2166 self
2167 }
2168}
2169
2170impl Default for OperatorSpawnerFactory {
2171 fn default() -> Self {
2172 Self::new()
2173 }
2174}
2175
2176impl SpawnerFactoryKind for OperatorSpawnerFactory {
2177 const KIND: AgentKind = AgentKind::Operator;
2178 type Worker = crate::operator::OperatorWorker;
2179}
2180
2181impl SpawnerFactory for OperatorSpawnerFactory {
2182 fn build(
2183 &self,
2184 agent_def: &AgentDef,
2185 _hint: Option<&Value>,
2186 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2187 let agent_name = &agent_def.name;
2188 let spec = &agent_def.spec;
2189 let system_prompt = agent_def.profile.as_ref().map(|p| p.system_prompt.clone());
2195 let invalid = |msg: String| CompileError::InvalidSpec {
2196 name: agent_name.to_string(),
2197 msg,
2198 };
2199 let op_ref = spec
2200 .get("operator_ref")
2201 .and_then(|v| v.as_str())
2202 .ok_or_else(|| invalid("operator spec: 'operator_ref' (string) required".into()))?;
2203 let operators = self
2204 .operators
2205 .read()
2206 .expect("OperatorSpawnerFactory.operators RwLock poisoned");
2207 let op = operators.get(op_ref).cloned().ok_or_else(|| {
2208 let mut names: Vec<String> = operators.keys().cloned().collect();
2209 names.sort();
2210 let names_list = if names.is_empty() {
2211 "<none>".to_string()
2212 } else {
2213 names.join(", ")
2214 };
2215 invalid(format!(
2216 "operator_ref '{op_ref}' not registered in factory. \
2217 Registered sids: [{names_list}]. \
2218 Hint: call mse_operator_join(roles=[...]) to mint the sid first."
2219 ))
2220 })?;
2221 drop(operators);
2222
2223 let worker_binding = agent_def
2230 .profile
2231 .as_ref()
2232 .and_then(|p| p.worker_binding.as_ref())
2233 .map(|variant| WorkerBinding {
2234 variant: variant.clone(),
2235 tools: agent_def
2236 .profile
2237 .as_ref()
2238 .map(|p| p.tools.clone())
2239 .unwrap_or_default(),
2240 request_digest: None,
2244 requested_model: None,
2245 });
2246 if op.requires_worker_binding() && worker_binding.is_none() {
2247 return Err(invalid(format!(
2254 "{WORKER_BINDING_REQUIRED_MSG_PREFIX}. \
2255 Fix by either: \
2256 (a) if authoring the Blueprint JSON directly, add \
2257 `agents[N].profile.worker_binding: \"<subagent-type>\"` \
2258 to the JSON literal; or \
2259 (b) if using an $agent_md file ref, add \
2260 `worker_binding: <subagent-type>` to the agent .md frontmatter."
2261 )));
2262 }
2263 Ok(Arc::new(OperatorSpawner::new(
2264 op,
2265 system_prompt,
2266 worker_binding,
2267 )))
2268 }
2269}
2270
2271#[cfg(test)]
2272mod operator_spawner_factory_worker_binding_tests {
2273 use super::*;
2274 use crate::blueprint::AgentProfile;
2275 use crate::core::ctx::Ctx;
2276 use crate::types::CapToken;
2277 use crate::worker::adapter::{WorkerError, WorkerResult};
2278
2279 struct StubOperator {
2284 requires_binding: bool,
2285 }
2286
2287 #[async_trait]
2288 impl Operator for StubOperator {
2289 async fn execute(
2290 &self,
2291 _ctx: &Ctx,
2292 _system: Option<String>,
2293 _prompt: Value,
2294 _worker: Option<WorkerBinding>,
2295 _worker_token: CapToken,
2296 ) -> Result<WorkerResult, WorkerError> {
2297 Ok(WorkerResult {
2298 value: Value::Null,
2299 ok: true,
2300 })
2301 }
2302
2303 fn requires_worker_binding(&self) -> bool {
2304 self.requires_binding
2305 }
2306 }
2307
2308 fn agent_def_with(profile: Option<AgentProfile>) -> AgentDef {
2309 AgentDef {
2310 name: "test-agent".to_string(),
2311 kind: AgentKind::Operator,
2312 spec: serde_json::json!({ "operator_ref": "op1" }),
2313 profile,
2314 meta: None,
2315 runner: None,
2316 runner_ref: None,
2317 verdict: None,
2318 }
2319 }
2320
2321 #[test]
2322 fn build_fails_loud_when_binding_required_but_absent() {
2323 let factory = OperatorSpawnerFactory::new();
2324 factory.register_operator(
2325 "op1",
2326 Arc::new(StubOperator {
2327 requires_binding: true,
2328 }) as Arc<dyn Operator>,
2329 );
2330 let def = agent_def_with(Some(AgentProfile::default()));
2331 match factory.build(&def, None) {
2332 Err(CompileError::InvalidSpec { name, msg }) => {
2333 assert_eq!(name, "test-agent");
2334 assert!(
2335 msg.contains("worker_binding is required"),
2336 "unexpected message: {msg}"
2337 );
2338 assert!(
2342 msg.contains("agents[N].profile.worker_binding"),
2343 "message missing JSON-direct hint (issue #9): {msg}"
2344 );
2345 assert!(
2346 msg.contains("agent .md frontmatter"),
2347 "message missing $agent_md hint: {msg}"
2348 );
2349 }
2350 Err(other) => panic!("expected InvalidSpec, got: {other:?}"),
2351 Ok(_) => panic!("expected compile-time failure, got Ok"),
2352 }
2353 }
2354
2355 #[test]
2362 fn factory_error_message_carries_the_shared_prefix_and_specializes_the_diagnostic() {
2363 let factory = OperatorSpawnerFactory::new();
2364 factory.register_operator(
2365 "op1",
2366 Arc::new(StubOperator {
2367 requires_binding: true,
2368 }) as Arc<dyn Operator>,
2369 );
2370 let def = agent_def_with(Some(AgentProfile::default()));
2371 let err = match factory.build(&def, None) {
2372 Err(err) => err,
2373 Ok(_) => panic!("expected compile-time failure, got Ok"),
2374 };
2375 match &err {
2376 CompileError::InvalidSpec { msg, .. } => {
2377 assert!(
2378 msg.starts_with(WORKER_BINDING_REQUIRED_MSG_PREFIX),
2379 "factory message must start with the shared prefix, got: {msg}"
2380 );
2381 }
2382 other => panic!("expected InvalidSpec, got: {other:?}"),
2383 }
2384 let d = mlua_swarm_diag::Diagnostic::from(&err);
2385 assert_eq!(d.kind, "worker-binding-missing");
2386 }
2387
2388 #[test]
2389 fn build_succeeds_when_binding_required_and_present() {
2390 let factory = OperatorSpawnerFactory::new();
2391 factory.register_operator(
2392 "op1",
2393 Arc::new(StubOperator {
2394 requires_binding: true,
2395 }) as Arc<dyn Operator>,
2396 );
2397 let profile = AgentProfile {
2398 worker_binding: Some("mse-worker-coder".to_string()),
2399 tools: vec!["Read".to_string(), "Edit".to_string()],
2400 ..Default::default()
2401 };
2402 let def = agent_def_with(Some(profile));
2403 assert!(
2404 factory.build(&def, None).is_ok(),
2405 "expected Ok when worker_binding is declared"
2406 );
2407 }
2408
2409 #[test]
2410 fn build_succeeds_when_binding_not_required_and_absent() {
2411 let factory = OperatorSpawnerFactory::new();
2412 factory.register_operator(
2413 "op1",
2414 Arc::new(StubOperator {
2415 requires_binding: false,
2416 }) as Arc<dyn Operator>,
2417 );
2418 let def = agent_def_with(Some(AgentProfile::default()));
2419 assert!(
2420 factory.build(&def, None).is_ok(),
2421 "backends that don't require a binding must not be gated by its absence"
2422 );
2423 }
2424}
2425
2426#[cfg(test)]
2434mod lua_inline_source_tests {
2435 use super::*;
2436 use crate::types::{CapToken, Role, StepId};
2437
2438 fn agent(name: &str, spec: Value) -> AgentDef {
2439 AgentDef {
2440 name: name.to_string(),
2441 kind: AgentKind::Lua,
2442 spec,
2443 profile: None,
2444 meta: None,
2445 runner: None,
2446 runner_ref: None,
2447 verdict: None,
2448 }
2449 }
2450
2451 fn test_invocation(prompt: &str) -> crate::worker::adapter::WorkerInvocation {
2452 crate::worker::adapter::WorkerInvocation {
2453 token: CapToken {
2454 agent_id: "a".into(),
2455 role: Role::Worker,
2456 scopes: vec!["*".into()],
2457 issued_at: 0,
2458 expire_at: u64::MAX / 2,
2459 max_uses: None,
2460 nonce: "test-nonce".into(),
2461 sig_hex: "".into(),
2462 },
2463 task_id: StepId::parse("ST-test").expect("StepId parse"),
2464 attempt: 1,
2465 agent: "g".into(),
2466 prompt: prompt.into(),
2467 sink: None,
2468 cancel_token: None,
2469 }
2470 }
2471
2472 #[test]
2473 fn build_accepts_inline_source_without_pre_registration() {
2474 let factory = LuaInProcessSpawnerFactory::new();
2475 let def = agent(
2476 "g",
2477 serde_json::json!({ "source": "return { value = 42, ok = true }" }),
2478 );
2479 assert!(
2480 factory.build(&def, None).is_ok(),
2481 "inline spec.source must build without a pre-registered fn_id"
2482 );
2483 }
2484
2485 #[test]
2486 fn build_rejects_when_neither_source_nor_fn_id_is_present() {
2487 let factory = LuaInProcessSpawnerFactory::new();
2488 let def = agent("g", serde_json::json!({}));
2489 match factory.build(&def, None) {
2490 Err(CompileError::InvalidSpec { msg, .. }) => {
2491 assert!(
2492 msg.contains("fn_id"),
2493 "empty spec must still surface the fn_id-required message: {msg}"
2494 );
2495 }
2496 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
2497 Ok(_) => panic!("expected InvalidSpec, got Ok(SpawnerAdapter)"),
2500 }
2501 }
2502
2503 #[tokio::test]
2507 async fn inline_source_evaluates_and_marshals_result() {
2508 let source =
2509 LuaScriptSource::new("return { value = _PROMPT .. '!', ok = true }", "smoke.lua");
2510 let out = run_lua_worker(
2511 std::sync::Arc::new(source),
2512 std::sync::Arc::new(HashMap::new()),
2513 test_invocation("hello"),
2514 )
2515 .await
2516 .expect("lua worker ok");
2517 assert_eq!(out.value, serde_json::json!("hello!"));
2518 assert!(out.ok);
2519 }
2520
2521 #[tokio::test]
2522 async fn inline_source_can_signal_agent_level_failure() {
2523 let source = LuaScriptSource::new("return { value = 'nope', ok = false }", "gate.lua");
2526 let out = run_lua_worker(
2527 std::sync::Arc::new(source),
2528 std::sync::Arc::new(HashMap::new()),
2529 test_invocation("input"),
2530 )
2531 .await
2532 .expect("lua worker ok");
2533 assert_eq!(out.value, serde_json::json!("nope"));
2534 assert!(!out.ok);
2535 }
2536}
2537
2538#[cfg(test)]
2541mod meta_ref_validation_tests {
2542 use super::*;
2543 use crate::blueprint::{AgentMeta, MetaDef};
2544 use crate::worker::adapter::WorkerResult;
2545
2546 fn registry_with_echo() -> SpawnerRegistry {
2547 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2548 Ok(WorkerResult {
2549 value: Value::String(inv.prompt),
2550 ok: true,
2551 })
2552 });
2553 let mut reg = SpawnerRegistry::new();
2554 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2555 reg
2556 }
2557
2558 fn rustfn_agent(name: &str) -> AgentDef {
2559 AgentDef {
2560 name: name.to_string(),
2561 kind: AgentKind::RustFn,
2562 spec: serde_json::json!({ "fn_id": "echo" }),
2563 profile: None,
2564 meta: None,
2565 runner: None,
2566 runner_ref: None,
2567 verdict: None,
2568 }
2569 }
2570
2571 fn simple_flow(agent_ref: &str, in_: Expr) -> FlowNode {
2572 FlowNode::Step {
2573 ref_: agent_ref.to_string(),
2574 in_,
2575 out: Expr::Path {
2576 at: "$.output".parse().expect("literal test path: $.output"),
2577 },
2578 }
2579 }
2580
2581 fn minimal_bp(agents: Vec<AgentDef>, metas: Vec<MetaDef>, flow: FlowNode) -> Blueprint {
2582 Blueprint {
2583 schema_version: crate::blueprint::current_schema_version(),
2584 id: "meta-ref-ut".into(),
2585 flow,
2586 agents,
2587 operators: vec![],
2588 metas,
2589 hints: Default::default(),
2590 strategy: Default::default(),
2591 metadata: BlueprintMetadata::default(),
2592 spawner_hints: Default::default(),
2593 default_agent_kind: AgentKind::Operator,
2594 default_operator_kind: None,
2595 default_init_ctx: None,
2596 default_agent_ctx: None,
2597 default_context_policy: None,
2598 projection_placement: None,
2599 audits: vec![],
2600 degradation_policy: None,
2601 runners: vec![],
2602 default_runner: None,
2603 subprocesses: vec![],
2604 check_policy: None,
2605 blueprint_ref_includes: Vec::new(),
2606 }
2607 }
2608
2609 #[test]
2610 fn valid_meta_ref_compiles() {
2611 let mut agent = rustfn_agent("worker");
2612 agent.meta = Some(AgentMeta {
2613 meta_ref: Some("shared".to_string()),
2614 ..Default::default()
2615 });
2616 let bp = minimal_bp(
2617 vec![agent],
2618 vec![MetaDef {
2619 name: "shared".into(),
2620 ctx: serde_json::json!({ "k": "v" }),
2621 }],
2622 simple_flow(
2623 "worker",
2624 Expr::Path {
2625 at: "$.input".parse().expect("literal test path: $.input"),
2626 },
2627 ),
2628 );
2629 let compiler = Compiler::new(registry_with_echo());
2630 assert!(
2631 compiler.compile(&bp).is_ok(),
2632 "a resolvable AgentMeta.meta_ref must compile"
2633 );
2634 }
2635
2636 #[test]
2637 fn unknown_agent_meta_ref_is_unresolved_meta_ref() {
2638 let mut agent = rustfn_agent("worker");
2639 agent.meta = Some(AgentMeta {
2640 meta_ref: Some("missing".to_string()),
2641 ..Default::default()
2642 });
2643 let bp = minimal_bp(
2644 vec![agent],
2645 vec![],
2646 simple_flow(
2647 "worker",
2648 Expr::Path {
2649 at: "$.input".parse().expect("literal test path: $.input"),
2650 },
2651 ),
2652 );
2653 let compiler = Compiler::new(registry_with_echo());
2654 match compiler.compile(&bp) {
2655 Err(CompileError::UnresolvedMetaRef {
2656 where_,
2657 meta_ref,
2658 defined,
2659 }) => {
2660 assert!(
2661 where_.contains("worker"),
2662 "where_ must name the agent: {where_}"
2663 );
2664 assert_eq!(meta_ref, "missing");
2665 assert!(defined.is_empty());
2666 }
2667 Err(other) => {
2668 panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
2669 }
2670 Ok(_) => panic!("expected compile-time failure, got Ok"),
2671 }
2672 }
2673
2674 #[test]
2675 fn unknown_static_step_meta_ref_in_lit_is_unresolved_meta_ref() {
2676 let agent = rustfn_agent("worker");
2677 let in_ = Expr::Lit {
2678 value: serde_json::json!({ "$step_meta": { "ref": "missing" }, "$in": "go" }),
2679 };
2680 let bp = minimal_bp(vec![agent], vec![], simple_flow("worker", in_));
2681 let compiler = Compiler::new(registry_with_echo());
2682 match compiler.compile(&bp) {
2683 Err(CompileError::UnresolvedMetaRef {
2684 where_, meta_ref, ..
2685 }) => {
2686 assert!(
2687 where_.contains("worker"),
2688 "where_ must name the offending step: {where_}"
2689 );
2690 assert_eq!(meta_ref, "missing");
2691 }
2692 Err(other) => {
2693 panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
2694 }
2695 Ok(_) => panic!("expected compile-time failure, got Ok"),
2696 }
2697 }
2698
2699 #[test]
2700 fn path_op_input_with_no_static_envelope_compiles_fine() {
2701 let agent = rustfn_agent("worker");
2702 let bp = minimal_bp(
2703 vec![agent],
2704 vec![],
2705 simple_flow(
2706 "worker",
2707 Expr::Path {
2708 at: "$.input".parse().expect("literal test path: $.input"),
2709 },
2710 ),
2711 );
2712 let compiler = Compiler::new(registry_with_echo());
2713 assert!(
2714 compiler.compile(&bp).is_ok(),
2715 "a non-Lit Step.in must not trigger the best-effort static $step_meta check"
2716 );
2717 }
2718}
2719
2720#[cfg(test)]
2722mod audit_agent_validation_tests {
2723 use super::*;
2724 use crate::worker::adapter::WorkerResult;
2725 use mlua_swarm_schema::{AuditDef, AuditMode};
2726
2727 fn registry_with_echo() -> SpawnerRegistry {
2728 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2729 Ok(WorkerResult {
2730 value: Value::String(inv.prompt),
2731 ok: true,
2732 })
2733 });
2734 let mut reg = SpawnerRegistry::new();
2735 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2736 reg
2737 }
2738
2739 fn rustfn_agent(name: &str) -> AgentDef {
2740 AgentDef {
2741 name: name.to_string(),
2742 kind: AgentKind::RustFn,
2743 spec: serde_json::json!({ "fn_id": "echo" }),
2744 profile: None,
2745 meta: None,
2746 runner: None,
2747 runner_ref: None,
2748 verdict: None,
2749 }
2750 }
2751
2752 fn minimal_bp(agents: Vec<AgentDef>, audits: Vec<AuditDef>) -> Blueprint {
2753 Blueprint {
2754 schema_version: crate::blueprint::current_schema_version(),
2755 id: "audit-ref-ut".into(),
2756 flow: FlowNode::Step {
2757 ref_: "worker".to_string(),
2758 in_: Expr::Path {
2759 at: "$.input".parse().expect("literal test path: $.input"),
2760 },
2761 out: Expr::Path {
2762 at: "$.output".parse().expect("literal test path: $.output"),
2763 },
2764 },
2765 agents,
2766 operators: vec![],
2767 metas: vec![],
2768 hints: Default::default(),
2769 strategy: Default::default(),
2770 metadata: BlueprintMetadata::default(),
2771 spawner_hints: Default::default(),
2772 default_agent_kind: AgentKind::Operator,
2773 default_operator_kind: None,
2774 default_init_ctx: None,
2775 default_agent_ctx: None,
2776 default_context_policy: None,
2777 projection_placement: None,
2778 audits,
2779 degradation_policy: None,
2780 runners: vec![],
2781 default_runner: None,
2782 subprocesses: vec![],
2783 check_policy: None,
2784 blueprint_ref_includes: Vec::new(),
2785 }
2786 }
2787
2788 #[test]
2789 fn unresolved_audit_agent_is_a_loud_compile_error() {
2790 let bp = minimal_bp(
2791 vec![rustfn_agent("worker")],
2792 vec![AuditDef {
2793 agent: "missing-auditor".to_string(),
2794 steps: None,
2795 mode: AuditMode::default(),
2796 }],
2797 );
2798 let compiler = Compiler::new(registry_with_echo());
2799 match compiler.compile(&bp) {
2800 Err(CompileError::UnresolvedAuditAgent { agent, defined }) => {
2801 assert_eq!(agent, "missing-auditor");
2802 assert_eq!(defined, vec!["worker".to_string()]);
2803 }
2804 Err(other) => {
2805 panic!("expected UnresolvedAuditAgent, got a different CompileError: {other}")
2806 }
2807 Ok(_) => panic!("expected compile-time failure, got Ok"),
2808 }
2809 }
2810
2811 #[test]
2812 fn resolved_audit_agent_compiles_fine() {
2813 let bp = minimal_bp(
2814 vec![rustfn_agent("worker"), rustfn_agent("auditor")],
2815 vec![AuditDef {
2816 agent: "auditor".to_string(),
2817 steps: None,
2818 mode: AuditMode::default(),
2819 }],
2820 );
2821 let compiler = Compiler::new(registry_with_echo());
2822 assert!(
2823 compiler.compile(&bp).is_ok(),
2824 "an audits[].agent that names a declared AgentDef must compile"
2825 );
2826 }
2827}
2828
2829#[cfg(test)]
2832mod projection_placement_compile_tests {
2833 use super::*;
2834 use crate::core::projection_placement::{ProjectionPlacement, RootPreference};
2835 use crate::worker::adapter::WorkerResult;
2836 use mlua_swarm_schema::ProjectionPlacementSpec;
2837
2838 fn registry_with_echo() -> SpawnerRegistry {
2839 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2840 Ok(WorkerResult {
2841 value: Value::String(inv.prompt),
2842 ok: true,
2843 })
2844 });
2845 let mut reg = SpawnerRegistry::new();
2846 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2847 reg
2848 }
2849
2850 fn minimal_bp(projection_placement: Option<ProjectionPlacementSpec>) -> Blueprint {
2851 Blueprint {
2852 schema_version: crate::blueprint::current_schema_version(),
2853 id: "projection-placement-ut".into(),
2854 flow: FlowNode::Step {
2855 ref_: "worker".to_string(),
2856 in_: Expr::Path {
2857 at: "$.input".parse().expect("literal test path: $.input"),
2858 },
2859 out: Expr::Path {
2860 at: "$.output".parse().expect("literal test path: $.output"),
2861 },
2862 },
2863 agents: vec![AgentDef {
2864 name: "worker".to_string(),
2865 kind: AgentKind::RustFn,
2866 spec: serde_json::json!({ "fn_id": "echo" }),
2867 profile: None,
2868 meta: None,
2869 runner: None,
2870 runner_ref: None,
2871 verdict: None,
2872 }],
2873 operators: vec![],
2874 metas: vec![],
2875 hints: Default::default(),
2876 strategy: Default::default(),
2877 metadata: BlueprintMetadata::default(),
2878 spawner_hints: Default::default(),
2879 default_agent_kind: AgentKind::Operator,
2880 default_operator_kind: None,
2881 default_init_ctx: None,
2882 default_agent_ctx: None,
2883 default_context_policy: None,
2884 projection_placement,
2885 audits: vec![],
2886 degradation_policy: None,
2887 runners: vec![],
2888 default_runner: None,
2889 subprocesses: vec![],
2890 check_policy: None,
2891 blueprint_ref_includes: Vec::new(),
2892 }
2893 }
2894
2895 #[test]
2896 fn undeclared_projection_placement_compiles_to_byte_compat_default() {
2897 let bp = minimal_bp(None);
2898 let compiled = Compiler::new(registry_with_echo())
2899 .compile(&bp)
2900 .expect("undeclared projection_placement compiles");
2901 assert_eq!(
2902 *compiled.projection_placement,
2903 ProjectionPlacement::default()
2904 );
2905 }
2906
2907 #[test]
2908 fn declared_valid_projection_placement_compiles_to_matching_resolver() {
2909 let bp = minimal_bp(Some(ProjectionPlacementSpec {
2910 root: Some("project_root".to_string()),
2911 dir_template: Some("custom/{task_id}/out".to_string()),
2912 }));
2913 let compiled = Compiler::new(registry_with_echo())
2914 .compile(&bp)
2915 .expect("valid projection_placement compiles");
2916 assert_eq!(
2917 compiled.projection_placement.root_preference,
2918 RootPreference::ProjectRoot
2919 );
2920 assert_eq!(
2921 compiled.projection_placement.dir_template,
2922 "custom/{task_id}/out"
2923 );
2924 }
2925
2926 #[test]
2927 fn declared_invalid_dir_template_rejects_compile() {
2928 let bp = minimal_bp(Some(ProjectionPlacementSpec {
2929 root: None,
2930 dir_template: Some("workspace/tasks/ctx".to_string()), }));
2932 match Compiler::new(registry_with_echo()).compile(&bp) {
2933 Err(CompileError::InvalidProjectionPlacement(_)) => {}
2934 Err(other) => {
2935 panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
2936 }
2937 Ok(_) => {
2938 panic!("expected compile-time rejection for a missing {{task_id}} placeholder")
2939 }
2940 }
2941 }
2942
2943 #[test]
2944 fn declared_invalid_root_literal_rejects_compile() {
2945 let bp = minimal_bp(Some(ProjectionPlacementSpec {
2946 root: Some("nope".to_string()),
2947 dir_template: None,
2948 }));
2949 match Compiler::new(registry_with_echo()).compile(&bp) {
2950 Err(CompileError::InvalidProjectionPlacement(_)) => {}
2951 Err(other) => {
2952 panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
2953 }
2954 Ok(_) => panic!("expected compile-time rejection for an invalid root literal"),
2955 }
2956 }
2957}
2958
2959#[cfg(test)]
2961mod verdict_contract_lint_tests {
2962 use super::*;
2963 use crate::worker::adapter::WorkerResult;
2964
2965 fn registry_with_echo() -> SpawnerRegistry {
2966 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2967 Ok(WorkerResult {
2968 value: Value::String(inv.prompt),
2969 ok: true,
2970 })
2971 });
2972 let mut reg = SpawnerRegistry::new();
2973 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2974 reg
2975 }
2976
2977 fn gate_agent(verdict: Option<VerdictContract>) -> AgentDef {
2978 AgentDef {
2979 name: "gate".to_string(),
2980 kind: AgentKind::RustFn,
2981 spec: serde_json::json!({ "fn_id": "echo" }),
2982 profile: None,
2983 meta: None,
2984 runner: None,
2985 runner_ref: None,
2986 verdict,
2987 }
2988 }
2989
2990 fn minimal_bp(agent: AgentDef, flow: FlowNode) -> Blueprint {
2991 Blueprint {
2992 schema_version: crate::blueprint::current_schema_version(),
2993 id: "verdict-contract-ut".into(),
2994 flow,
2995 agents: vec![agent],
2996 operators: vec![],
2997 metas: vec![],
2998 hints: Default::default(),
2999 strategy: Default::default(),
3000 metadata: BlueprintMetadata::default(),
3001 spawner_hints: Default::default(),
3002 default_agent_kind: AgentKind::Operator,
3003 default_operator_kind: None,
3004 default_init_ctx: None,
3005 default_agent_ctx: None,
3006 default_context_policy: None,
3007 projection_placement: None,
3008 audits: vec![],
3009 degradation_policy: None,
3010 runners: vec![],
3011 default_runner: None,
3012 subprocesses: vec![],
3013 check_policy: None,
3014 blueprint_ref_includes: Vec::new(),
3015 }
3016 }
3017
3018 fn step(ref_: &str, out_path: &str) -> FlowNode {
3019 FlowNode::Step {
3020 ref_: ref_.to_string(),
3021 in_: Expr::Lit { value: Value::Null },
3022 out: Expr::Path {
3023 at: out_path.parse().expect("literal test path"),
3024 },
3025 }
3026 }
3027
3028 fn noop() -> FlowNode {
3029 FlowNode::Seq { children: vec![] }
3030 }
3031
3032 fn eq_cond(path: &str, lit: &str) -> Expr {
3033 Expr::Eq {
3034 lhs: Box::new(Expr::Path {
3035 at: path.parse().expect("literal test path"),
3036 }),
3037 rhs: Box::new(Expr::Lit {
3038 value: Value::String(lit.to_string()),
3039 }),
3040 }
3041 }
3042
3043 fn branch(cond: Expr, then_: FlowNode, else_: FlowNode) -> FlowNode {
3044 FlowNode::Branch {
3045 cond,
3046 then_: Box::new(then_),
3047 else_: Box::new(else_),
3048 }
3049 }
3050
3051 fn body_contract(values: &[&str]) -> VerdictContract {
3052 VerdictContract {
3053 channel: VerdictChannel::Body,
3054 values: values.iter().map(|v| v.to_string()).collect(),
3055 }
3056 }
3057
3058 fn part_contract(values: &[&str]) -> VerdictContract {
3059 VerdictContract {
3060 channel: VerdictChannel::Part,
3061 values: values.iter().map(|v| v.to_string()).collect(),
3062 }
3063 }
3064
3065 #[test]
3066 fn contract_with_correct_body_channel_and_value_compiles() {
3067 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3068 let flow = FlowNode::Seq {
3069 children: vec![
3070 step("gate", "$.verdict"),
3071 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3072 ],
3073 };
3074 let bp = minimal_bp(agent, flow);
3075 assert!(
3076 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3077 "a cond addressing the bare step output must match a channel: \"body\" contract"
3078 );
3079 }
3080
3081 #[test]
3082 fn contract_with_correct_part_channel_and_value_compiles() {
3083 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3084 let flow = FlowNode::Seq {
3085 children: vec![
3086 step("gate", "$.gate"),
3087 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3088 ],
3089 };
3090 let bp = minimal_bp(agent, flow);
3091 assert!(
3092 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3093 "a cond addressing '<step>.parts.verdict' must match a channel: \"part\" contract"
3094 );
3095 }
3096
3097 #[test]
3098 fn body_channel_contract_rejects_cond_addressing_parts_verdict() {
3099 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3103 let flow = FlowNode::Seq {
3104 children: vec![
3105 step("gate", "$.gate"),
3106 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3107 ],
3108 };
3109 let bp = minimal_bp(agent, flow);
3110 match Compiler::new(registry_with_echo()).compile(&bp) {
3111 Err(CompileError::VerdictChannelMismatch {
3112 where_,
3113 agent,
3114 expected_channel,
3115 actual_shape,
3116 }) => {
3117 assert_eq!(agent, "gate");
3118 assert_eq!(expected_channel, "body");
3119 assert_eq!(actual_shape, "part");
3120 assert!(where_.contains("Branch cond"), "where_: {where_}");
3121 }
3122 Err(other) => {
3123 panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
3124 }
3125 Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
3126 }
3127 }
3128
3129 #[test]
3130 fn part_channel_contract_rejects_cond_addressing_bare_output() {
3131 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3134 let flow = FlowNode::Seq {
3135 children: vec![
3136 step("gate", "$.verdict"),
3137 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3138 ],
3139 };
3140 let bp = minimal_bp(agent, flow);
3141 match Compiler::new(registry_with_echo()).compile(&bp) {
3142 Err(CompileError::VerdictChannelMismatch {
3143 agent,
3144 expected_channel,
3145 actual_shape,
3146 ..
3147 }) => {
3148 assert_eq!(agent, "gate");
3149 assert_eq!(expected_channel, "part");
3150 assert_eq!(actual_shape, "body");
3151 }
3152 Err(other) => {
3153 panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
3154 }
3155 Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
3156 }
3157 }
3158
3159 #[test]
3160 fn contract_rejects_lit_outside_declared_values() {
3161 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3162 let flow = FlowNode::Seq {
3163 children: vec![
3164 step("gate", "$.verdict"),
3165 branch(eq_cond("$.verdict", "UNKNOWN"), noop(), noop()),
3166 ],
3167 };
3168 let bp = minimal_bp(agent, flow);
3169 match Compiler::new(registry_with_echo()).compile(&bp) {
3170 Err(CompileError::VerdictValueNotInContract {
3171 agent,
3172 value,
3173 values,
3174 ..
3175 }) => {
3176 assert_eq!(agent, "gate");
3177 assert_eq!(value, "UNKNOWN");
3178 assert_eq!(values, vec!["PASS".to_string(), "BLOCKED".to_string()]);
3179 }
3180 Err(other) => {
3181 panic!("expected VerdictValueNotInContract, got a different CompileError: {other}")
3182 }
3183 Ok(_) => panic!("expected compile-time rejection for a Lit outside declared values"),
3184 }
3185 }
3186
3187 #[test]
3188 fn undeclared_agent_referenced_by_cond_compiles_with_warning_only() {
3189 let agent = gate_agent(None);
3190 let flow = FlowNode::Seq {
3191 children: vec![
3192 step("gate", "$.verdict"),
3193 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3194 ],
3195 };
3196 let bp = minimal_bp(agent, flow);
3197 assert!(
3198 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3199 "an undeclared verdict contract must never reject compile (opt-in, back-compat)"
3200 );
3201 }
3202
3203 #[test]
3204 fn in_expr_with_lit_haystack_members_compiles() {
3205 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3206 let cond = Expr::In {
3207 needle: Box::new(Expr::Path {
3208 at: "$.verdict".parse().expect("literal test path"),
3209 }),
3210 haystack: Box::new(Expr::Lit {
3211 value: serde_json::json!(["PASS", "BLOCKED"]),
3212 }),
3213 };
3214 let flow = FlowNode::Seq {
3215 children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
3216 };
3217 let bp = minimal_bp(agent, flow);
3218 assert!(
3219 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3220 "an `In` haystack whose every Lit is a declared value must compile"
3221 );
3222 }
3223
3224 #[test]
3231 fn strict_mode_rejects_unhandled_declared_value() {
3232 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3233 let flow = FlowNode::Seq {
3234 children: vec![
3235 step("gate", "$.verdict"),
3236 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3237 ],
3238 };
3239 let mut bp = minimal_bp(agent, flow);
3240 bp.metadata.strict_verdict_handling = Some(true);
3241 match Compiler::new(registry_with_echo()).compile(&bp) {
3242 Err(CompileError::VerdictValueUnhandled {
3243 agent,
3244 value,
3245 declared_values,
3246 step_ref,
3247 }) => {
3248 assert_eq!(agent, "gate");
3249 assert_eq!(value, "PASS");
3250 assert_eq!(
3251 declared_values,
3252 vec!["PASS".to_string(), "BLOCKED".to_string()]
3253 );
3254 assert_eq!(step_ref, "gate");
3255 }
3256 Err(other) => {
3257 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
3258 }
3259 Ok(_) => panic!(
3260 "expected compile-time rejection for a declared verdict value with no \
3261 downstream handler under strict_verdict_handling=Some(true)"
3262 ),
3263 }
3264 }
3265
3266 #[test]
3273 fn default_mode_permits_unhandled_declared_value() {
3274 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3275 let flow = FlowNode::Seq {
3276 children: vec![
3277 step("gate", "$.verdict"),
3278 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3279 ],
3280 };
3281 let bp = minimal_bp(agent, flow);
3282 assert!(
3284 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3285 "default mode must never reject a Blueprint for unhandled declared values \
3286 (opt-in, back-compat with GH #50)"
3287 );
3288 }
3289
3290 #[test]
3295 fn strict_mode_accepts_all_declared_values_handled() {
3296 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3297 let flow = FlowNode::Seq {
3300 children: vec![
3301 step("gate", "$.verdict"),
3302 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3303 branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
3304 ],
3305 };
3306 let mut bp = minimal_bp(agent, flow);
3307 bp.metadata.strict_verdict_handling = Some(true);
3308 assert!(
3309 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3310 "strict mode must accept a Blueprint that handles every declared value"
3311 );
3312 }
3313
3314 #[test]
3318 fn strict_mode_accepts_declared_values_covered_by_in_expr() {
3319 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3320 let cond = Expr::In {
3321 needle: Box::new(Expr::Path {
3322 at: "$.verdict".parse().expect("literal test path"),
3323 }),
3324 haystack: Box::new(Expr::Lit {
3325 value: serde_json::json!(["PASS", "BLOCKED"]),
3326 }),
3327 };
3328 let flow = FlowNode::Seq {
3329 children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
3330 };
3331 let mut bp = minimal_bp(agent, flow);
3332 bp.metadata.strict_verdict_handling = Some(true);
3333 assert!(
3334 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3335 "strict mode must accept an `In` haystack that covers every declared value"
3336 );
3337 }
3338
3339 #[test]
3343 fn strict_mode_rejects_unhandled_part_channel_value() {
3344 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3345 let flow = FlowNode::Seq {
3346 children: vec![
3347 step("gate", "$.gate"),
3348 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3349 ],
3350 };
3351 let mut bp = minimal_bp(agent, flow);
3352 bp.metadata.strict_verdict_handling = Some(true);
3353 match Compiler::new(registry_with_echo()).compile(&bp) {
3354 Err(CompileError::VerdictValueUnhandled {
3355 agent,
3356 value,
3357 step_ref,
3358 ..
3359 }) => {
3360 assert_eq!(agent, "gate");
3361 assert_eq!(value, "PASS");
3362 assert_eq!(step_ref, "gate");
3363 }
3364 Err(other) => {
3365 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
3366 }
3367 Ok(_) => panic!(
3368 "expected compile-time rejection for a declared verdict value with no \
3369 downstream handler (part channel) under strict_verdict_handling=Some(true)"
3370 ),
3371 }
3372 }
3373
3374 #[test]
3381 fn verdict_omitted_blueprint_compiles_unchanged_with_empty_contracts() {
3382 let agent = gate_agent(None);
3383 let flow = FlowNode::Seq {
3384 children: vec![
3385 step("gate", "$.verdict"),
3386 FlowNode::Loop {
3387 counter: Expr::Path {
3388 at: "$.n".parse().expect("literal test path"),
3389 },
3390 cond: eq_cond("$.verdict", "BLOCKED"),
3391 body: Box::new(step("gate", "$.verdict")),
3392 max: 3,
3393 },
3394 branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
3395 ],
3396 };
3397 let bp = minimal_bp(agent, flow);
3398 let compiled = Compiler::new(registry_with_echo())
3399 .compile(&bp)
3400 .expect("a verdict-omitted Blueprint must compile unchanged");
3401 assert!(
3402 compiled.router.verdict_contracts.is_empty(),
3403 "no agent declared a verdict contract"
3404 );
3405 }
3406
3407 #[test]
3414 fn every_compile_error_diagnostic_kind_is_a_declared_lint() {
3415 let kinds = [
3416 "bound-agent-resolution",
3417 "unknown-agent-kind",
3418 "invalid-agent-spec",
3419 "worker-binding-missing",
3420 "unresolved-agent-ref",
3421 "duplicate-agent-name",
3422 "unresolved-operator-ref",
3423 "unresolved-meta-ref",
3424 "step-naming-collision",
3425 "invalid-projection-placement",
3426 "unresolved-audit-agent",
3427 "verdict-channel-mismatch",
3428 "verdict-value-not-in-contract",
3429 "verdict-value-unhandled",
3430 ];
3431 for kind in kinds {
3432 assert!(
3433 mlua_swarm_diag::lint_decl(kind).is_some(),
3434 "kind '{kind}' emitted by From<&CompileError> has no LINT_DECLS entry"
3435 );
3436 }
3437 }
3438
3439 #[test]
3440 fn invalid_spec_with_worker_binding_prefix_specializes_the_diagnostic_kind() {
3441 let err = CompileError::InvalidSpec {
3445 name: "greeter".into(),
3446 msg: format!("{WORKER_BINDING_REQUIRED_MSG_PREFIX}. Fix by either: (a) ..."),
3447 };
3448 let d = mlua_swarm_diag::Diagnostic::from(&err);
3449 assert_eq!(d.kind, "worker-binding-missing");
3450 assert_eq!(d.level, mlua_swarm_diag::DiagLevel::Error);
3451 assert!(matches!(d.stage, mlua_swarm_diag::DiagStage::CompileLint));
3452 assert!(d.message.contains("greeter"));
3453 let suggestion = d
3454 .suggestion
3455 .expect("specialized arm must carry a suggestion");
3456 assert!(suggestion.patch.contains("backend = \"ws_operator\""));
3457 assert_eq!(
3458 suggestion.applicability,
3459 mlua_swarm_diag::Applicability::HasPlaceholders
3460 );
3461 assert_eq!(
3462 d.docs_ref.expect("docs_ref must be set").uri,
3463 "mse://guides/bp-dsl-templates"
3464 );
3465 match d.span.expect("span must be set").element {
3466 mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "greeter"),
3467 other => panic!("expected Agent span, got {other:?}"),
3468 }
3469 }
3470
3471 #[test]
3472 fn generic_invalid_spec_maps_to_the_generic_kind() {
3473 let err = CompileError::InvalidSpec {
3474 name: "solo".into(),
3475 msg: "operator spec: 'operator_ref' (string) required".into(),
3476 };
3477 let d = mlua_swarm_diag::Diagnostic::from(&err);
3478 assert_eq!(d.kind, "invalid-agent-spec");
3479 assert!(
3480 d.suggestion.is_none(),
3481 "generic arm carries no canned patch"
3482 );
3483 }
3484
3485 #[test]
3486 fn verdict_value_not_in_contract_diagnostic_carries_suggestion_and_span() {
3487 let err = CompileError::VerdictValueNotInContract {
3488 where_: "Branch cond".into(),
3489 agent: "review".into(),
3490 value: "NOT_DECLARED".into(),
3491 values: vec!["PASS".into(), "BLOCKED".into()],
3492 };
3493 let d = mlua_swarm_diag::Diagnostic::from(&err);
3494 assert_eq!(d.kind, "verdict-value-not-in-contract");
3495 assert!(d.message.contains("NOT_DECLARED"));
3496 assert!(d.suggestion.is_some());
3497 match d.span.expect("span must be set").element {
3498 mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "review"),
3499 other => panic!("expected Agent span, got {other:?}"),
3500 }
3501 }
3502}
3503
3504#[cfg(test)]
3506mod subprocess_embed_compile_tests {
3507 use super::*;
3508 use mlua_swarm_schema::{current_schema_version, SubprocessDef, SubprocessOverrides};
3509
3510 fn subprocess_agent(name: &str, runner: Option<Runner>) -> AgentDef {
3511 AgentDef {
3512 name: name.to_string(),
3513 kind: AgentKind::Subprocess,
3514 spec: serde_json::json!({}),
3515 profile: Some(AgentProfile {
3516 system_prompt: "you are a headless worker".to_string(),
3517 model: Some("profile-model".to_string()),
3518 tools: vec!["Read".to_string()],
3519 ..Default::default()
3520 }),
3521 meta: None,
3522 runner,
3523 runner_ref: None,
3524 verdict: None,
3525 }
3526 }
3527
3528 fn echo_def(name: &str) -> SubprocessDef {
3529 SubprocessDef {
3530 name: name.to_string(),
3531 argv: vec!["sh".to_string(), "-c".to_string(), "cat".to_string()],
3532 stdin: Some("{prompt}".to_string()),
3533 env: Default::default(),
3534 cwd: None,
3535 output: None,
3536 stream_mode: None,
3537 }
3538 }
3539
3540 fn bp_with(agents: Vec<AgentDef>, subprocesses: Vec<SubprocessDef>) -> Blueprint {
3541 Blueprint {
3542 schema_version: current_schema_version(),
3543 id: "gh83-ut".into(),
3544 flow: FlowNode::Seq { children: vec![] },
3545 agents,
3546 operators: vec![],
3547 metas: vec![],
3548 hints: Default::default(),
3549 strategy: Default::default(),
3550 metadata: BlueprintMetadata::default(),
3551 spawner_hints: Default::default(),
3552 default_agent_kind: AgentKind::Operator,
3553 default_operator_kind: None,
3554 default_init_ctx: None,
3555 default_agent_ctx: None,
3556 default_context_policy: None,
3557 projection_placement: None,
3558 audits: vec![],
3559 degradation_policy: None,
3560 runners: vec![],
3561 default_runner: None,
3562 subprocesses,
3563 check_policy: None,
3564 blueprint_ref_includes: vec![],
3565 }
3566 }
3567
3568 fn subprocess_runner(template: &str) -> Runner {
3569 Runner::Subprocess {
3570 template: template.to_string(),
3571 overrides: SubprocessOverrides::default(),
3572 }
3573 }
3574
3575 #[test]
3576 fn validate_placeholders_accepts_closed_set_and_json_braces() {
3577 for ok in [
3578 "{system} {system_file} {prompt} {model} {tools_csv} {work_dir} {task_id} {attempt}",
3579 r#"echo '{"result": "ok", "nested": {"a": 1}}'"#,
3580 "no placeholders at all",
3581 "unmatched { brace",
3582 ] {
3583 validate_embed_placeholders(ok, "ut").expect("must be accepted");
3584 }
3585 }
3586
3587 #[test]
3588 fn validate_placeholders_rejects_unknown_token() {
3589 let err = validate_embed_placeholders("--flag {evil}", "argv[1]").unwrap_err();
3590 assert!(err.contains("'{evil}'"), "token named: {err}");
3591 assert!(err.contains("closed set"), "closed set listed: {err}");
3592 }
3593
3594 #[test]
3598 fn validate_placeholders_descends_into_literal_braces() {
3599 validate_embed_placeholders(r#"{"task": "{prompt}"}"#, "stdin")
3600 .expect("nested closed-set token must be accepted");
3601 let err = validate_embed_placeholders(r#"{"task": "{evil}"}"#, "stdin").unwrap_err();
3602 assert!(
3603 err.contains("'{evil}'"),
3604 "nested unknown token caught: {err}"
3605 );
3606 }
3607
3608 #[test]
3609 fn hint_resolution_finds_declared_template() {
3610 let agent = subprocess_agent("headless", Some(subprocess_runner("echo")));
3611 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
3612 let hint = resolve_subprocess_template_hint(&bp, &agent)
3613 .expect("resolves")
3614 .expect("Runner::Subprocess must synthesize a hint");
3615 assert_eq!(hint[SUBPROCESS_TEMPLATE_HINT_KEY]["name"], "echo");
3616 assert!(hint.get(SUBPROCESS_OVERRIDES_HINT_KEY).is_some());
3617 }
3618
3619 #[test]
3620 fn hint_resolution_unknown_template_is_invalid_spec() {
3621 let agent = subprocess_agent("headless", Some(subprocess_runner("nope")));
3622 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
3623 let err = resolve_subprocess_template_hint(&bp, &agent).unwrap_err();
3624 let msg = format!("{err}");
3625 assert!(msg.contains("'nope'"), "missing template named: {msg}");
3626 assert!(msg.contains("echo"), "defined templates listed: {msg}");
3627 }
3628
3629 #[test]
3630 fn hint_resolution_none_without_subprocess_runner() {
3631 let agent = subprocess_agent("headless", None);
3632 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
3633 let hint = resolve_subprocess_template_hint(&bp, &agent).expect("resolves");
3634 assert!(hint.is_none(), "spec-based agents keep the historical path");
3635 }
3636
3637 #[test]
3638 fn build_embed_rejects_unknown_placeholder() {
3639 let agent = subprocess_agent("headless", None);
3640 let mut def = echo_def("echo");
3641 def.argv.push("--x={evil}".to_string());
3642 let err = SubprocessProcessSpawnerFactory::build_embed(
3643 &agent,
3644 &serde_json::to_value(&def).unwrap(),
3645 None,
3646 )
3647 .unwrap_err();
3648 assert!(format!("{err}").contains("'{evil}'"));
3649 }
3650
3651 #[test]
3652 fn build_embed_rejects_output_with_stream_mode() {
3653 let agent = subprocess_agent("headless", None);
3654 let mut def = echo_def("echo");
3655 def.stream_mode = Some("ndjson_lines".to_string());
3656 def.output = Some(mlua_swarm_schema::SubprocessOutput {
3657 format: Some("json".to_string()),
3658 result_ptr: None,
3659 ok_from: None,
3660 });
3661 let err = SubprocessProcessSpawnerFactory::build_embed(
3662 &agent,
3663 &serde_json::to_value(&def).unwrap(),
3664 None,
3665 )
3666 .unwrap_err();
3667 assert!(format!("{err}").contains("plain-mode"));
3668 }
3669
3670 #[test]
3671 fn build_embed_rejects_malformed_result_ptr_and_ok_from() {
3672 let agent = subprocess_agent("headless", None);
3673 let mut def = echo_def("echo");
3674 def.output = Some(mlua_swarm_schema::SubprocessOutput {
3675 format: None,
3676 result_ptr: Some("result".to_string()),
3677 ok_from: None,
3678 });
3679 let err = SubprocessProcessSpawnerFactory::build_embed(
3680 &agent,
3681 &serde_json::to_value(&def).unwrap(),
3682 None,
3683 )
3684 .unwrap_err();
3685 assert!(format!("{err}").contains("JSON Pointer"));
3686
3687 let mut def = echo_def("echo");
3688 def.output = Some(mlua_swarm_schema::SubprocessOutput {
3689 format: None,
3690 result_ptr: None,
3691 ok_from: Some("status".to_string()),
3692 });
3693 let err = SubprocessProcessSpawnerFactory::build_embed(
3694 &agent,
3695 &serde_json::to_value(&def).unwrap(),
3696 None,
3697 )
3698 .unwrap_err();
3699 assert!(format!("{err}").contains("exit_code"));
3700 }
3701
3702 #[test]
3703 fn build_embed_bakes_profile_with_override_precedence() {
3704 let agent = subprocess_agent("headless", None);
3705 let def = echo_def("echo");
3706 let overrides = SubprocessOverrides {
3707 model: Some("override-model".to_string()),
3708 tools: vec!["Bash".to_string(), "Write".to_string()],
3709 cwd: Some("/tmp/override-wd".to_string()),
3710 };
3711 let sp = SubprocessProcessSpawnerFactory::build_embed(
3712 &agent,
3713 &serde_json::to_value(&def).unwrap(),
3714 Some(&serde_json::to_value(&overrides).unwrap()),
3715 )
3716 .expect("builds");
3717 let embed = sp.embed.as_ref().expect("embed template baked");
3718 assert_eq!(embed.model.as_deref(), Some("override-model"));
3719 assert_eq!(embed.tools_csv, "Bash,Write");
3720 assert_eq!(embed.cwd.as_deref(), Some("/tmp/override-wd"));
3721 assert_eq!(
3722 embed.system_prompt.as_deref(),
3723 Some("you are a headless worker")
3724 );
3725 }
3726}