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 stats: None,
1876 }
1877 .ensure_worker_kind("lua"))
1878}
1879
1880impl Default for LuaInProcessSpawnerFactory {
1881 fn default() -> Self {
1882 Self::new()
1883 }
1884}
1885
1886impl SpawnerFactoryKind for LuaInProcessSpawnerFactory {
1887 const KIND: AgentKind = AgentKind::Lua;
1888 type Worker = LuaWorker;
1889}
1890
1891impl SpawnerFactory for LuaInProcessSpawnerFactory {
1892 fn build(
1893 &self,
1894 agent_def: &AgentDef,
1895 _hint: Option<&Value>,
1896 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1897 if let Some(source) = agent_def.spec.get("source").and_then(|v| v.as_str()) {
1903 let label = agent_def
1904 .spec
1905 .get("label")
1906 .and_then(|v| v.as_str())
1907 .map(str::to_string)
1908 .unwrap_or_else(|| format!("{}.lua", agent_def.name));
1909 let script = Arc::new(LuaScriptSource::new(source.to_string(), label));
1910 let bridges = Arc::new(self.bridges.clone());
1911 let wrapped: WorkerFn = Arc::new(move |inv| {
1912 let source = script.clone();
1913 let bridges = bridges.clone();
1914 Box::pin(run_lua_worker(source, bridges, inv))
1915 });
1916 let mut sp: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
1917 sp.registry.insert(agent_def.name.to_string(), wrapped);
1918 return Ok(Arc::new(sp));
1919 }
1920 build_inproc_from_registry::<LuaWorker>(&self.registry, agent_def, "lua")
1921 }
1922}
1923
1924pub struct RustFnInProcessSpawnerFactory {
1938 registry: HashMap<String, WorkerFn>,
1939}
1940
1941impl RustFnInProcessSpawnerFactory {
1942 pub fn new() -> Self {
1944 Self {
1945 registry: HashMap::new(),
1946 }
1947 }
1948
1949 pub fn register_fn<F, Fut>(mut self, fn_id: impl Into<String>, f: F) -> Self
1952 where
1953 F: Fn(crate::worker::adapter::WorkerInvocation) -> Fut + Send + Sync + 'static,
1954 Fut: std::future::Future<
1955 Output = Result<
1956 crate::worker::adapter::WorkerResult,
1957 crate::worker::adapter::WorkerError,
1958 >,
1959 > + Send
1960 + 'static,
1961 {
1962 let f = Arc::new(f);
1963 let wrapped: WorkerFn = Arc::new(move |inv| {
1964 let f = f.clone();
1965 Box::pin(f(inv))
1966 });
1967 self.registry.insert(fn_id.into(), wrapped);
1968 self
1969 }
1970}
1971
1972impl Default for RustFnInProcessSpawnerFactory {
1973 fn default() -> Self {
1974 Self::new()
1975 }
1976}
1977
1978impl SpawnerFactoryKind for RustFnInProcessSpawnerFactory {
1979 const KIND: AgentKind = AgentKind::RustFn;
1980 type Worker = RustFnWorker;
1981}
1982
1983impl SpawnerFactory for RustFnInProcessSpawnerFactory {
1984 fn build(
1985 &self,
1986 agent_def: &AgentDef,
1987 _hint: Option<&Value>,
1988 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1989 build_inproc_from_registry::<RustFnWorker>(&self.registry, agent_def, "rust_fn")
1990 }
1991}
1992
1993fn build_inproc_from_registry<W>(
1999 registry: &HashMap<String, WorkerFn>,
2000 agent_def: &AgentDef,
2001 kind_label: &str,
2002) -> Result<Arc<dyn SpawnerAdapter>, CompileError>
2003where
2004 W: crate::worker::Worker + From<crate::worker::WorkerJoinHandler> + Send + Sync + 'static,
2005{
2006 let agent_name = &agent_def.name;
2007 let spec = &agent_def.spec;
2008 let invalid = |msg: String| CompileError::InvalidSpec {
2009 name: agent_name.to_string(),
2010 msg,
2011 };
2012 let fn_id = spec
2013 .get("fn_id")
2014 .and_then(|v| v.as_str())
2015 .ok_or_else(|| invalid(format!("{kind_label} spec: 'fn_id' (string) required")))?;
2016 let f = registry
2017 .get(fn_id)
2018 .cloned()
2019 .ok_or_else(|| invalid(format!("fn_id '{fn_id}' not registered in factory")))?;
2020 let mut sp: InProcSpawner<W> = InProcSpawner::<W>::typed();
2021 sp.registry.insert(agent_name.to_string(), f);
2025 Ok(Arc::new(sp))
2026}
2027
2028pub struct LuaWorker {
2033 pub handler: crate::worker::WorkerJoinHandler,
2035}
2036
2037impl From<crate::worker::WorkerJoinHandler> for LuaWorker {
2038 fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
2039 Self { handler }
2040 }
2041}
2042
2043#[async_trait::async_trait]
2044impl crate::worker::Worker for LuaWorker {
2045 fn id(&self) -> &crate::types::WorkerId {
2046 &self.handler.worker_id
2047 }
2048 fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
2049 self.handler.cancel.clone()
2050 }
2051 async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
2052 self.handler.await_completion().await
2053 }
2054}
2055
2056pub struct RustFnWorker {
2061 pub handler: crate::worker::WorkerJoinHandler,
2063}
2064
2065impl From<crate::worker::WorkerJoinHandler> for RustFnWorker {
2066 fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
2067 Self { handler }
2068 }
2069}
2070
2071#[async_trait::async_trait]
2072impl crate::worker::Worker for RustFnWorker {
2073 fn id(&self) -> &crate::types::WorkerId {
2074 &self.handler.worker_id
2075 }
2076 fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
2077 self.handler.cancel.clone()
2078 }
2079 async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
2080 self.handler.await_completion().await
2081 }
2082}
2083
2084pub struct OperatorSpawnerFactory {
2137 operators: Arc<std::sync::RwLock<HashMap<String, Arc<dyn Operator>>>>,
2138}
2139
2140impl OperatorSpawnerFactory {
2141 pub fn new() -> Self {
2143 Self {
2144 operators: Arc::new(std::sync::RwLock::new(HashMap::new())),
2145 }
2146 }
2147
2148 pub fn register_operator(&self, id: impl Into<String>, op: Arc<dyn Operator>) -> &Self {
2154 self.operators
2155 .write()
2156 .expect("OperatorSpawnerFactory.operators RwLock poisoned")
2157 .insert(id.into(), op);
2158 self
2159 }
2160
2161 pub fn unregister_operator(&self, id: &str) -> &Self {
2164 self.operators
2165 .write()
2166 .expect("OperatorSpawnerFactory.operators RwLock poisoned")
2167 .remove(id);
2168 self
2169 }
2170}
2171
2172impl Default for OperatorSpawnerFactory {
2173 fn default() -> Self {
2174 Self::new()
2175 }
2176}
2177
2178impl SpawnerFactoryKind for OperatorSpawnerFactory {
2179 const KIND: AgentKind = AgentKind::Operator;
2180 type Worker = crate::operator::OperatorWorker;
2181}
2182
2183impl SpawnerFactory for OperatorSpawnerFactory {
2184 fn build(
2185 &self,
2186 agent_def: &AgentDef,
2187 _hint: Option<&Value>,
2188 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2189 let agent_name = &agent_def.name;
2190 let spec = &agent_def.spec;
2191 let system_prompt = agent_def.profile.as_ref().map(|p| p.system_prompt.clone());
2197 let invalid = |msg: String| CompileError::InvalidSpec {
2198 name: agent_name.to_string(),
2199 msg,
2200 };
2201 let op_ref = spec
2202 .get("operator_ref")
2203 .and_then(|v| v.as_str())
2204 .ok_or_else(|| invalid("operator spec: 'operator_ref' (string) required".into()))?;
2205 let operators = self
2206 .operators
2207 .read()
2208 .expect("OperatorSpawnerFactory.operators RwLock poisoned");
2209 let op = operators.get(op_ref).cloned().ok_or_else(|| {
2210 let mut names: Vec<String> = operators.keys().cloned().collect();
2211 names.sort();
2212 let names_list = if names.is_empty() {
2213 "<none>".to_string()
2214 } else {
2215 names.join(", ")
2216 };
2217 invalid(format!(
2218 "operator_ref '{op_ref}' not registered in factory. \
2219 Registered sids: [{names_list}]. \
2220 Hint: call mse_operator_join(roles=[...]) to mint the sid first."
2221 ))
2222 })?;
2223 drop(operators);
2224
2225 let worker_binding = agent_def
2232 .profile
2233 .as_ref()
2234 .and_then(|p| p.worker_binding.as_ref())
2235 .map(|variant| WorkerBinding {
2236 variant: variant.clone(),
2237 tools: agent_def
2238 .profile
2239 .as_ref()
2240 .map(|p| p.tools.clone())
2241 .unwrap_or_default(),
2242 request_digest: None,
2246 requested_model: None,
2247 });
2248 if op.requires_worker_binding() && worker_binding.is_none() {
2249 return Err(invalid(format!(
2256 "{WORKER_BINDING_REQUIRED_MSG_PREFIX}. \
2257 Fix by either: \
2258 (a) if authoring the Blueprint JSON directly, add \
2259 `agents[N].profile.worker_binding: \"<subagent-type>\"` \
2260 to the JSON literal; or \
2261 (b) if using an $agent_md file ref, add \
2262 `worker_binding: <subagent-type>` to the agent .md frontmatter."
2263 )));
2264 }
2265 Ok(Arc::new(OperatorSpawner::new(
2266 op,
2267 system_prompt,
2268 worker_binding,
2269 )))
2270 }
2271}
2272
2273#[cfg(test)]
2274mod operator_spawner_factory_worker_binding_tests {
2275 use super::*;
2276 use crate::blueprint::AgentProfile;
2277 use crate::core::ctx::Ctx;
2278 use crate::types::CapToken;
2279 use crate::worker::adapter::{WorkerError, WorkerResult};
2280
2281 struct StubOperator {
2286 requires_binding: bool,
2287 }
2288
2289 #[async_trait]
2290 impl Operator for StubOperator {
2291 async fn execute(
2292 &self,
2293 _ctx: &Ctx,
2294 _system: Option<String>,
2295 _prompt: Value,
2296 _worker: Option<WorkerBinding>,
2297 _worker_token: CapToken,
2298 ) -> Result<WorkerResult, WorkerError> {
2299 Ok(WorkerResult {
2300 value: Value::Null,
2301 ok: true,
2302 stats: None,
2303 })
2304 }
2305
2306 fn requires_worker_binding(&self) -> bool {
2307 self.requires_binding
2308 }
2309 }
2310
2311 fn agent_def_with(profile: Option<AgentProfile>) -> AgentDef {
2312 AgentDef {
2313 name: "test-agent".to_string(),
2314 kind: AgentKind::Operator,
2315 spec: serde_json::json!({ "operator_ref": "op1" }),
2316 profile,
2317 meta: None,
2318 runner: None,
2319 runner_ref: None,
2320 verdict: None,
2321 }
2322 }
2323
2324 #[test]
2325 fn build_fails_loud_when_binding_required_but_absent() {
2326 let factory = OperatorSpawnerFactory::new();
2327 factory.register_operator(
2328 "op1",
2329 Arc::new(StubOperator {
2330 requires_binding: true,
2331 }) as Arc<dyn Operator>,
2332 );
2333 let def = agent_def_with(Some(AgentProfile::default()));
2334 match factory.build(&def, None) {
2335 Err(CompileError::InvalidSpec { name, msg }) => {
2336 assert_eq!(name, "test-agent");
2337 assert!(
2338 msg.contains("worker_binding is required"),
2339 "unexpected message: {msg}"
2340 );
2341 assert!(
2345 msg.contains("agents[N].profile.worker_binding"),
2346 "message missing JSON-direct hint (issue #9): {msg}"
2347 );
2348 assert!(
2349 msg.contains("agent .md frontmatter"),
2350 "message missing $agent_md hint: {msg}"
2351 );
2352 }
2353 Err(other) => panic!("expected InvalidSpec, got: {other:?}"),
2354 Ok(_) => panic!("expected compile-time failure, got Ok"),
2355 }
2356 }
2357
2358 #[test]
2365 fn factory_error_message_carries_the_shared_prefix_and_specializes_the_diagnostic() {
2366 let factory = OperatorSpawnerFactory::new();
2367 factory.register_operator(
2368 "op1",
2369 Arc::new(StubOperator {
2370 requires_binding: true,
2371 }) as Arc<dyn Operator>,
2372 );
2373 let def = agent_def_with(Some(AgentProfile::default()));
2374 let err = match factory.build(&def, None) {
2375 Err(err) => err,
2376 Ok(_) => panic!("expected compile-time failure, got Ok"),
2377 };
2378 match &err {
2379 CompileError::InvalidSpec { msg, .. } => {
2380 assert!(
2381 msg.starts_with(WORKER_BINDING_REQUIRED_MSG_PREFIX),
2382 "factory message must start with the shared prefix, got: {msg}"
2383 );
2384 }
2385 other => panic!("expected InvalidSpec, got: {other:?}"),
2386 }
2387 let d = mlua_swarm_diag::Diagnostic::from(&err);
2388 assert_eq!(d.kind, "worker-binding-missing");
2389 }
2390
2391 #[test]
2392 fn build_succeeds_when_binding_required_and_present() {
2393 let factory = OperatorSpawnerFactory::new();
2394 factory.register_operator(
2395 "op1",
2396 Arc::new(StubOperator {
2397 requires_binding: true,
2398 }) as Arc<dyn Operator>,
2399 );
2400 let profile = AgentProfile {
2401 worker_binding: Some("mse-worker-coder".to_string()),
2402 tools: vec!["Read".to_string(), "Edit".to_string()],
2403 ..Default::default()
2404 };
2405 let def = agent_def_with(Some(profile));
2406 assert!(
2407 factory.build(&def, None).is_ok(),
2408 "expected Ok when worker_binding is declared"
2409 );
2410 }
2411
2412 #[test]
2413 fn build_succeeds_when_binding_not_required_and_absent() {
2414 let factory = OperatorSpawnerFactory::new();
2415 factory.register_operator(
2416 "op1",
2417 Arc::new(StubOperator {
2418 requires_binding: false,
2419 }) as Arc<dyn Operator>,
2420 );
2421 let def = agent_def_with(Some(AgentProfile::default()));
2422 assert!(
2423 factory.build(&def, None).is_ok(),
2424 "backends that don't require a binding must not be gated by its absence"
2425 );
2426 }
2427}
2428
2429#[cfg(test)]
2437mod lua_inline_source_tests {
2438 use super::*;
2439 use crate::types::{CapToken, Role, StepId};
2440
2441 fn agent(name: &str, spec: Value) -> AgentDef {
2442 AgentDef {
2443 name: name.to_string(),
2444 kind: AgentKind::Lua,
2445 spec,
2446 profile: None,
2447 meta: None,
2448 runner: None,
2449 runner_ref: None,
2450 verdict: None,
2451 }
2452 }
2453
2454 fn test_invocation(prompt: &str) -> crate::worker::adapter::WorkerInvocation {
2455 crate::worker::adapter::WorkerInvocation {
2456 token: CapToken {
2457 agent_id: "a".into(),
2458 role: Role::Worker,
2459 scopes: vec!["*".into()],
2460 issued_at: 0,
2461 expire_at: u64::MAX / 2,
2462 max_uses: None,
2463 nonce: "test-nonce".into(),
2464 sig_hex: "".into(),
2465 },
2466 task_id: StepId::parse("ST-test").expect("StepId parse"),
2467 attempt: 1,
2468 agent: "g".into(),
2469 prompt: prompt.into(),
2470 sink: None,
2471 cancel_token: None,
2472 }
2473 }
2474
2475 #[test]
2476 fn build_accepts_inline_source_without_pre_registration() {
2477 let factory = LuaInProcessSpawnerFactory::new();
2478 let def = agent(
2479 "g",
2480 serde_json::json!({ "source": "return { value = 42, ok = true }" }),
2481 );
2482 assert!(
2483 factory.build(&def, None).is_ok(),
2484 "inline spec.source must build without a pre-registered fn_id"
2485 );
2486 }
2487
2488 #[test]
2489 fn build_rejects_when_neither_source_nor_fn_id_is_present() {
2490 let factory = LuaInProcessSpawnerFactory::new();
2491 let def = agent("g", serde_json::json!({}));
2492 match factory.build(&def, None) {
2493 Err(CompileError::InvalidSpec { msg, .. }) => {
2494 assert!(
2495 msg.contains("fn_id"),
2496 "empty spec must still surface the fn_id-required message: {msg}"
2497 );
2498 }
2499 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
2500 Ok(_) => panic!("expected InvalidSpec, got Ok(SpawnerAdapter)"),
2503 }
2504 }
2505
2506 #[tokio::test]
2510 async fn inline_source_evaluates_and_marshals_result() {
2511 let source =
2512 LuaScriptSource::new("return { value = _PROMPT .. '!', ok = true }", "smoke.lua");
2513 let out = run_lua_worker(
2514 std::sync::Arc::new(source),
2515 std::sync::Arc::new(HashMap::new()),
2516 test_invocation("hello"),
2517 )
2518 .await
2519 .expect("lua worker ok");
2520 assert_eq!(out.value, serde_json::json!("hello!"));
2521 assert!(out.ok);
2522 }
2523
2524 #[tokio::test]
2525 async fn inline_source_can_signal_agent_level_failure() {
2526 let source = LuaScriptSource::new("return { value = 'nope', ok = false }", "gate.lua");
2529 let out = run_lua_worker(
2530 std::sync::Arc::new(source),
2531 std::sync::Arc::new(HashMap::new()),
2532 test_invocation("input"),
2533 )
2534 .await
2535 .expect("lua worker ok");
2536 assert_eq!(out.value, serde_json::json!("nope"));
2537 assert!(!out.ok);
2538 }
2539}
2540
2541#[cfg(test)]
2544mod meta_ref_validation_tests {
2545 use super::*;
2546 use crate::blueprint::{AgentMeta, MetaDef};
2547 use crate::worker::adapter::WorkerResult;
2548
2549 fn registry_with_echo() -> SpawnerRegistry {
2550 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2551 Ok(WorkerResult {
2552 value: Value::String(inv.prompt),
2553 ok: true,
2554 stats: None,
2555 })
2556 });
2557 let mut reg = SpawnerRegistry::new();
2558 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2559 reg
2560 }
2561
2562 fn rustfn_agent(name: &str) -> AgentDef {
2563 AgentDef {
2564 name: name.to_string(),
2565 kind: AgentKind::RustFn,
2566 spec: serde_json::json!({ "fn_id": "echo" }),
2567 profile: None,
2568 meta: None,
2569 runner: None,
2570 runner_ref: None,
2571 verdict: None,
2572 }
2573 }
2574
2575 fn simple_flow(agent_ref: &str, in_: Expr) -> FlowNode {
2576 FlowNode::Step {
2577 ref_: agent_ref.to_string(),
2578 in_,
2579 out: Expr::Path {
2580 at: "$.output".parse().expect("literal test path: $.output"),
2581 },
2582 }
2583 }
2584
2585 fn minimal_bp(agents: Vec<AgentDef>, metas: Vec<MetaDef>, flow: FlowNode) -> Blueprint {
2586 Blueprint {
2587 schema_version: crate::blueprint::current_schema_version(),
2588 id: "meta-ref-ut".into(),
2589 flow,
2590 agents,
2591 operators: vec![],
2592 metas,
2593 hints: Default::default(),
2594 strategy: Default::default(),
2595 metadata: BlueprintMetadata::default(),
2596 spawner_hints: Default::default(),
2597 default_agent_kind: AgentKind::Operator,
2598 default_operator_kind: None,
2599 default_init_ctx: None,
2600 default_agent_ctx: None,
2601 default_context_policy: None,
2602 projection_placement: None,
2603 audits: vec![],
2604 degradation_policy: None,
2605 runners: vec![],
2606 default_runner: None,
2607 subprocesses: vec![],
2608 check_policy: None,
2609 blueprint_ref_includes: Vec::new(),
2610 }
2611 }
2612
2613 #[test]
2614 fn valid_meta_ref_compiles() {
2615 let mut agent = rustfn_agent("worker");
2616 agent.meta = Some(AgentMeta {
2617 meta_ref: Some("shared".to_string()),
2618 ..Default::default()
2619 });
2620 let bp = minimal_bp(
2621 vec![agent],
2622 vec![MetaDef {
2623 name: "shared".into(),
2624 ctx: serde_json::json!({ "k": "v" }),
2625 }],
2626 simple_flow(
2627 "worker",
2628 Expr::Path {
2629 at: "$.input".parse().expect("literal test path: $.input"),
2630 },
2631 ),
2632 );
2633 let compiler = Compiler::new(registry_with_echo());
2634 assert!(
2635 compiler.compile(&bp).is_ok(),
2636 "a resolvable AgentMeta.meta_ref must compile"
2637 );
2638 }
2639
2640 #[test]
2641 fn unknown_agent_meta_ref_is_unresolved_meta_ref() {
2642 let mut agent = rustfn_agent("worker");
2643 agent.meta = Some(AgentMeta {
2644 meta_ref: Some("missing".to_string()),
2645 ..Default::default()
2646 });
2647 let bp = minimal_bp(
2648 vec![agent],
2649 vec![],
2650 simple_flow(
2651 "worker",
2652 Expr::Path {
2653 at: "$.input".parse().expect("literal test path: $.input"),
2654 },
2655 ),
2656 );
2657 let compiler = Compiler::new(registry_with_echo());
2658 match compiler.compile(&bp) {
2659 Err(CompileError::UnresolvedMetaRef {
2660 where_,
2661 meta_ref,
2662 defined,
2663 }) => {
2664 assert!(
2665 where_.contains("worker"),
2666 "where_ must name the agent: {where_}"
2667 );
2668 assert_eq!(meta_ref, "missing");
2669 assert!(defined.is_empty());
2670 }
2671 Err(other) => {
2672 panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
2673 }
2674 Ok(_) => panic!("expected compile-time failure, got Ok"),
2675 }
2676 }
2677
2678 #[test]
2679 fn unknown_static_step_meta_ref_in_lit_is_unresolved_meta_ref() {
2680 let agent = rustfn_agent("worker");
2681 let in_ = Expr::Lit {
2682 value: serde_json::json!({ "$step_meta": { "ref": "missing" }, "$in": "go" }),
2683 };
2684 let bp = minimal_bp(vec![agent], vec![], simple_flow("worker", in_));
2685 let compiler = Compiler::new(registry_with_echo());
2686 match compiler.compile(&bp) {
2687 Err(CompileError::UnresolvedMetaRef {
2688 where_, meta_ref, ..
2689 }) => {
2690 assert!(
2691 where_.contains("worker"),
2692 "where_ must name the offending step: {where_}"
2693 );
2694 assert_eq!(meta_ref, "missing");
2695 }
2696 Err(other) => {
2697 panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
2698 }
2699 Ok(_) => panic!("expected compile-time failure, got Ok"),
2700 }
2701 }
2702
2703 #[test]
2704 fn path_op_input_with_no_static_envelope_compiles_fine() {
2705 let agent = rustfn_agent("worker");
2706 let bp = minimal_bp(
2707 vec![agent],
2708 vec![],
2709 simple_flow(
2710 "worker",
2711 Expr::Path {
2712 at: "$.input".parse().expect("literal test path: $.input"),
2713 },
2714 ),
2715 );
2716 let compiler = Compiler::new(registry_with_echo());
2717 assert!(
2718 compiler.compile(&bp).is_ok(),
2719 "a non-Lit Step.in must not trigger the best-effort static $step_meta check"
2720 );
2721 }
2722}
2723
2724#[cfg(test)]
2726mod audit_agent_validation_tests {
2727 use super::*;
2728 use crate::worker::adapter::WorkerResult;
2729 use mlua_swarm_schema::{AuditDef, AuditMode};
2730
2731 fn registry_with_echo() -> SpawnerRegistry {
2732 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2733 Ok(WorkerResult {
2734 value: Value::String(inv.prompt),
2735 ok: true,
2736 stats: None,
2737 })
2738 });
2739 let mut reg = SpawnerRegistry::new();
2740 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2741 reg
2742 }
2743
2744 fn rustfn_agent(name: &str) -> AgentDef {
2745 AgentDef {
2746 name: name.to_string(),
2747 kind: AgentKind::RustFn,
2748 spec: serde_json::json!({ "fn_id": "echo" }),
2749 profile: None,
2750 meta: None,
2751 runner: None,
2752 runner_ref: None,
2753 verdict: None,
2754 }
2755 }
2756
2757 fn minimal_bp(agents: Vec<AgentDef>, audits: Vec<AuditDef>) -> Blueprint {
2758 Blueprint {
2759 schema_version: crate::blueprint::current_schema_version(),
2760 id: "audit-ref-ut".into(),
2761 flow: FlowNode::Step {
2762 ref_: "worker".to_string(),
2763 in_: Expr::Path {
2764 at: "$.input".parse().expect("literal test path: $.input"),
2765 },
2766 out: Expr::Path {
2767 at: "$.output".parse().expect("literal test path: $.output"),
2768 },
2769 },
2770 agents,
2771 operators: vec![],
2772 metas: vec![],
2773 hints: Default::default(),
2774 strategy: Default::default(),
2775 metadata: BlueprintMetadata::default(),
2776 spawner_hints: Default::default(),
2777 default_agent_kind: AgentKind::Operator,
2778 default_operator_kind: None,
2779 default_init_ctx: None,
2780 default_agent_ctx: None,
2781 default_context_policy: None,
2782 projection_placement: None,
2783 audits,
2784 degradation_policy: None,
2785 runners: vec![],
2786 default_runner: None,
2787 subprocesses: vec![],
2788 check_policy: None,
2789 blueprint_ref_includes: Vec::new(),
2790 }
2791 }
2792
2793 #[test]
2794 fn unresolved_audit_agent_is_a_loud_compile_error() {
2795 let bp = minimal_bp(
2796 vec![rustfn_agent("worker")],
2797 vec![AuditDef {
2798 agent: "missing-auditor".to_string(),
2799 steps: None,
2800 mode: AuditMode::default(),
2801 }],
2802 );
2803 let compiler = Compiler::new(registry_with_echo());
2804 match compiler.compile(&bp) {
2805 Err(CompileError::UnresolvedAuditAgent { agent, defined }) => {
2806 assert_eq!(agent, "missing-auditor");
2807 assert_eq!(defined, vec!["worker".to_string()]);
2808 }
2809 Err(other) => {
2810 panic!("expected UnresolvedAuditAgent, got a different CompileError: {other}")
2811 }
2812 Ok(_) => panic!("expected compile-time failure, got Ok"),
2813 }
2814 }
2815
2816 #[test]
2817 fn resolved_audit_agent_compiles_fine() {
2818 let bp = minimal_bp(
2819 vec![rustfn_agent("worker"), rustfn_agent("auditor")],
2820 vec![AuditDef {
2821 agent: "auditor".to_string(),
2822 steps: None,
2823 mode: AuditMode::default(),
2824 }],
2825 );
2826 let compiler = Compiler::new(registry_with_echo());
2827 assert!(
2828 compiler.compile(&bp).is_ok(),
2829 "an audits[].agent that names a declared AgentDef must compile"
2830 );
2831 }
2832}
2833
2834#[cfg(test)]
2837mod projection_placement_compile_tests {
2838 use super::*;
2839 use crate::core::projection_placement::{ProjectionPlacement, RootPreference};
2840 use crate::worker::adapter::WorkerResult;
2841 use mlua_swarm_schema::ProjectionPlacementSpec;
2842
2843 fn registry_with_echo() -> SpawnerRegistry {
2844 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2845 Ok(WorkerResult {
2846 value: Value::String(inv.prompt),
2847 ok: true,
2848 stats: None,
2849 })
2850 });
2851 let mut reg = SpawnerRegistry::new();
2852 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2853 reg
2854 }
2855
2856 fn minimal_bp(projection_placement: Option<ProjectionPlacementSpec>) -> Blueprint {
2857 Blueprint {
2858 schema_version: crate::blueprint::current_schema_version(),
2859 id: "projection-placement-ut".into(),
2860 flow: FlowNode::Step {
2861 ref_: "worker".to_string(),
2862 in_: Expr::Path {
2863 at: "$.input".parse().expect("literal test path: $.input"),
2864 },
2865 out: Expr::Path {
2866 at: "$.output".parse().expect("literal test path: $.output"),
2867 },
2868 },
2869 agents: vec![AgentDef {
2870 name: "worker".to_string(),
2871 kind: AgentKind::RustFn,
2872 spec: serde_json::json!({ "fn_id": "echo" }),
2873 profile: None,
2874 meta: None,
2875 runner: None,
2876 runner_ref: None,
2877 verdict: None,
2878 }],
2879 operators: vec![],
2880 metas: vec![],
2881 hints: Default::default(),
2882 strategy: Default::default(),
2883 metadata: BlueprintMetadata::default(),
2884 spawner_hints: Default::default(),
2885 default_agent_kind: AgentKind::Operator,
2886 default_operator_kind: None,
2887 default_init_ctx: None,
2888 default_agent_ctx: None,
2889 default_context_policy: None,
2890 projection_placement,
2891 audits: vec![],
2892 degradation_policy: None,
2893 runners: vec![],
2894 default_runner: None,
2895 subprocesses: vec![],
2896 check_policy: None,
2897 blueprint_ref_includes: Vec::new(),
2898 }
2899 }
2900
2901 #[test]
2902 fn undeclared_projection_placement_compiles_to_byte_compat_default() {
2903 let bp = minimal_bp(None);
2904 let compiled = Compiler::new(registry_with_echo())
2905 .compile(&bp)
2906 .expect("undeclared projection_placement compiles");
2907 assert_eq!(
2908 *compiled.projection_placement,
2909 ProjectionPlacement::default()
2910 );
2911 }
2912
2913 #[test]
2914 fn declared_valid_projection_placement_compiles_to_matching_resolver() {
2915 let bp = minimal_bp(Some(ProjectionPlacementSpec {
2916 root: Some("project_root".to_string()),
2917 dir_template: Some("custom/{task_id}/out".to_string()),
2918 }));
2919 let compiled = Compiler::new(registry_with_echo())
2920 .compile(&bp)
2921 .expect("valid projection_placement compiles");
2922 assert_eq!(
2923 compiled.projection_placement.root_preference,
2924 RootPreference::ProjectRoot
2925 );
2926 assert_eq!(
2927 compiled.projection_placement.dir_template,
2928 "custom/{task_id}/out"
2929 );
2930 }
2931
2932 #[test]
2933 fn declared_invalid_dir_template_rejects_compile() {
2934 let bp = minimal_bp(Some(ProjectionPlacementSpec {
2935 root: None,
2936 dir_template: Some("workspace/tasks/ctx".to_string()), }));
2938 match Compiler::new(registry_with_echo()).compile(&bp) {
2939 Err(CompileError::InvalidProjectionPlacement(_)) => {}
2940 Err(other) => {
2941 panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
2942 }
2943 Ok(_) => {
2944 panic!("expected compile-time rejection for a missing {{task_id}} placeholder")
2945 }
2946 }
2947 }
2948
2949 #[test]
2950 fn declared_invalid_root_literal_rejects_compile() {
2951 let bp = minimal_bp(Some(ProjectionPlacementSpec {
2952 root: Some("nope".to_string()),
2953 dir_template: None,
2954 }));
2955 match Compiler::new(registry_with_echo()).compile(&bp) {
2956 Err(CompileError::InvalidProjectionPlacement(_)) => {}
2957 Err(other) => {
2958 panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
2959 }
2960 Ok(_) => panic!("expected compile-time rejection for an invalid root literal"),
2961 }
2962 }
2963}
2964
2965#[cfg(test)]
2967mod verdict_contract_lint_tests {
2968 use super::*;
2969 use crate::worker::adapter::WorkerResult;
2970
2971 fn registry_with_echo() -> SpawnerRegistry {
2972 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2973 Ok(WorkerResult {
2974 value: Value::String(inv.prompt),
2975 ok: true,
2976 stats: None,
2977 })
2978 });
2979 let mut reg = SpawnerRegistry::new();
2980 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2981 reg
2982 }
2983
2984 fn gate_agent(verdict: Option<VerdictContract>) -> AgentDef {
2985 AgentDef {
2986 name: "gate".to_string(),
2987 kind: AgentKind::RustFn,
2988 spec: serde_json::json!({ "fn_id": "echo" }),
2989 profile: None,
2990 meta: None,
2991 runner: None,
2992 runner_ref: None,
2993 verdict,
2994 }
2995 }
2996
2997 fn minimal_bp(agent: AgentDef, flow: FlowNode) -> Blueprint {
2998 Blueprint {
2999 schema_version: crate::blueprint::current_schema_version(),
3000 id: "verdict-contract-ut".into(),
3001 flow,
3002 agents: vec![agent],
3003 operators: vec![],
3004 metas: vec![],
3005 hints: Default::default(),
3006 strategy: Default::default(),
3007 metadata: BlueprintMetadata::default(),
3008 spawner_hints: Default::default(),
3009 default_agent_kind: AgentKind::Operator,
3010 default_operator_kind: None,
3011 default_init_ctx: None,
3012 default_agent_ctx: None,
3013 default_context_policy: None,
3014 projection_placement: None,
3015 audits: vec![],
3016 degradation_policy: None,
3017 runners: vec![],
3018 default_runner: None,
3019 subprocesses: vec![],
3020 check_policy: None,
3021 blueprint_ref_includes: Vec::new(),
3022 }
3023 }
3024
3025 fn step(ref_: &str, out_path: &str) -> FlowNode {
3026 FlowNode::Step {
3027 ref_: ref_.to_string(),
3028 in_: Expr::Lit { value: Value::Null },
3029 out: Expr::Path {
3030 at: out_path.parse().expect("literal test path"),
3031 },
3032 }
3033 }
3034
3035 fn noop() -> FlowNode {
3036 FlowNode::Seq { children: vec![] }
3037 }
3038
3039 fn eq_cond(path: &str, lit: &str) -> Expr {
3040 Expr::Eq {
3041 lhs: Box::new(Expr::Path {
3042 at: path.parse().expect("literal test path"),
3043 }),
3044 rhs: Box::new(Expr::Lit {
3045 value: Value::String(lit.to_string()),
3046 }),
3047 }
3048 }
3049
3050 fn branch(cond: Expr, then_: FlowNode, else_: FlowNode) -> FlowNode {
3051 FlowNode::Branch {
3052 cond,
3053 then_: Box::new(then_),
3054 else_: Box::new(else_),
3055 }
3056 }
3057
3058 fn body_contract(values: &[&str]) -> VerdictContract {
3059 VerdictContract {
3060 channel: VerdictChannel::Body,
3061 values: values.iter().map(|v| v.to_string()).collect(),
3062 }
3063 }
3064
3065 fn part_contract(values: &[&str]) -> VerdictContract {
3066 VerdictContract {
3067 channel: VerdictChannel::Part,
3068 values: values.iter().map(|v| v.to_string()).collect(),
3069 }
3070 }
3071
3072 #[test]
3073 fn contract_with_correct_body_channel_and_value_compiles() {
3074 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3075 let flow = FlowNode::Seq {
3076 children: vec![
3077 step("gate", "$.verdict"),
3078 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3079 ],
3080 };
3081 let bp = minimal_bp(agent, flow);
3082 assert!(
3083 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3084 "a cond addressing the bare step output must match a channel: \"body\" contract"
3085 );
3086 }
3087
3088 #[test]
3089 fn contract_with_correct_part_channel_and_value_compiles() {
3090 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3091 let flow = FlowNode::Seq {
3092 children: vec![
3093 step("gate", "$.gate"),
3094 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3095 ],
3096 };
3097 let bp = minimal_bp(agent, flow);
3098 assert!(
3099 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3100 "a cond addressing '<step>.parts.verdict' must match a channel: \"part\" contract"
3101 );
3102 }
3103
3104 #[test]
3105 fn body_channel_contract_rejects_cond_addressing_parts_verdict() {
3106 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3110 let flow = FlowNode::Seq {
3111 children: vec![
3112 step("gate", "$.gate"),
3113 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3114 ],
3115 };
3116 let bp = minimal_bp(agent, flow);
3117 match Compiler::new(registry_with_echo()).compile(&bp) {
3118 Err(CompileError::VerdictChannelMismatch {
3119 where_,
3120 agent,
3121 expected_channel,
3122 actual_shape,
3123 }) => {
3124 assert_eq!(agent, "gate");
3125 assert_eq!(expected_channel, "body");
3126 assert_eq!(actual_shape, "part");
3127 assert!(where_.contains("Branch cond"), "where_: {where_}");
3128 }
3129 Err(other) => {
3130 panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
3131 }
3132 Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
3133 }
3134 }
3135
3136 #[test]
3137 fn part_channel_contract_rejects_cond_addressing_bare_output() {
3138 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3141 let flow = FlowNode::Seq {
3142 children: vec![
3143 step("gate", "$.verdict"),
3144 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3145 ],
3146 };
3147 let bp = minimal_bp(agent, flow);
3148 match Compiler::new(registry_with_echo()).compile(&bp) {
3149 Err(CompileError::VerdictChannelMismatch {
3150 agent,
3151 expected_channel,
3152 actual_shape,
3153 ..
3154 }) => {
3155 assert_eq!(agent, "gate");
3156 assert_eq!(expected_channel, "part");
3157 assert_eq!(actual_shape, "body");
3158 }
3159 Err(other) => {
3160 panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
3161 }
3162 Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
3163 }
3164 }
3165
3166 #[test]
3167 fn contract_rejects_lit_outside_declared_values() {
3168 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3169 let flow = FlowNode::Seq {
3170 children: vec![
3171 step("gate", "$.verdict"),
3172 branch(eq_cond("$.verdict", "UNKNOWN"), noop(), noop()),
3173 ],
3174 };
3175 let bp = minimal_bp(agent, flow);
3176 match Compiler::new(registry_with_echo()).compile(&bp) {
3177 Err(CompileError::VerdictValueNotInContract {
3178 agent,
3179 value,
3180 values,
3181 ..
3182 }) => {
3183 assert_eq!(agent, "gate");
3184 assert_eq!(value, "UNKNOWN");
3185 assert_eq!(values, vec!["PASS".to_string(), "BLOCKED".to_string()]);
3186 }
3187 Err(other) => {
3188 panic!("expected VerdictValueNotInContract, got a different CompileError: {other}")
3189 }
3190 Ok(_) => panic!("expected compile-time rejection for a Lit outside declared values"),
3191 }
3192 }
3193
3194 #[test]
3195 fn undeclared_agent_referenced_by_cond_compiles_with_warning_only() {
3196 let agent = gate_agent(None);
3197 let flow = FlowNode::Seq {
3198 children: vec![
3199 step("gate", "$.verdict"),
3200 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3201 ],
3202 };
3203 let bp = minimal_bp(agent, flow);
3204 assert!(
3205 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3206 "an undeclared verdict contract must never reject compile (opt-in, back-compat)"
3207 );
3208 }
3209
3210 #[test]
3211 fn in_expr_with_lit_haystack_members_compiles() {
3212 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3213 let cond = Expr::In {
3214 needle: Box::new(Expr::Path {
3215 at: "$.verdict".parse().expect("literal test path"),
3216 }),
3217 haystack: Box::new(Expr::Lit {
3218 value: serde_json::json!(["PASS", "BLOCKED"]),
3219 }),
3220 };
3221 let flow = FlowNode::Seq {
3222 children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
3223 };
3224 let bp = minimal_bp(agent, flow);
3225 assert!(
3226 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3227 "an `In` haystack whose every Lit is a declared value must compile"
3228 );
3229 }
3230
3231 #[test]
3238 fn strict_mode_rejects_unhandled_declared_value() {
3239 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3240 let flow = FlowNode::Seq {
3241 children: vec![
3242 step("gate", "$.verdict"),
3243 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3244 ],
3245 };
3246 let mut bp = minimal_bp(agent, flow);
3247 bp.metadata.strict_verdict_handling = Some(true);
3248 match Compiler::new(registry_with_echo()).compile(&bp) {
3249 Err(CompileError::VerdictValueUnhandled {
3250 agent,
3251 value,
3252 declared_values,
3253 step_ref,
3254 }) => {
3255 assert_eq!(agent, "gate");
3256 assert_eq!(value, "PASS");
3257 assert_eq!(
3258 declared_values,
3259 vec!["PASS".to_string(), "BLOCKED".to_string()]
3260 );
3261 assert_eq!(step_ref, "gate");
3262 }
3263 Err(other) => {
3264 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
3265 }
3266 Ok(_) => panic!(
3267 "expected compile-time rejection for a declared verdict value with no \
3268 downstream handler under strict_verdict_handling=Some(true)"
3269 ),
3270 }
3271 }
3272
3273 #[test]
3280 fn default_mode_permits_unhandled_declared_value() {
3281 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3282 let flow = FlowNode::Seq {
3283 children: vec![
3284 step("gate", "$.verdict"),
3285 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3286 ],
3287 };
3288 let bp = minimal_bp(agent, flow);
3289 assert!(
3291 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3292 "default mode must never reject a Blueprint for unhandled declared values \
3293 (opt-in, back-compat with GH #50)"
3294 );
3295 }
3296
3297 #[test]
3302 fn strict_mode_accepts_all_declared_values_handled() {
3303 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3304 let flow = FlowNode::Seq {
3307 children: vec![
3308 step("gate", "$.verdict"),
3309 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3310 branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
3311 ],
3312 };
3313 let mut bp = minimal_bp(agent, flow);
3314 bp.metadata.strict_verdict_handling = Some(true);
3315 assert!(
3316 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3317 "strict mode must accept a Blueprint that handles every declared value"
3318 );
3319 }
3320
3321 #[test]
3325 fn strict_mode_accepts_declared_values_covered_by_in_expr() {
3326 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3327 let cond = Expr::In {
3328 needle: Box::new(Expr::Path {
3329 at: "$.verdict".parse().expect("literal test path"),
3330 }),
3331 haystack: Box::new(Expr::Lit {
3332 value: serde_json::json!(["PASS", "BLOCKED"]),
3333 }),
3334 };
3335 let flow = FlowNode::Seq {
3336 children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
3337 };
3338 let mut bp = minimal_bp(agent, flow);
3339 bp.metadata.strict_verdict_handling = Some(true);
3340 assert!(
3341 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3342 "strict mode must accept an `In` haystack that covers every declared value"
3343 );
3344 }
3345
3346 #[test]
3350 fn strict_mode_rejects_unhandled_part_channel_value() {
3351 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3352 let flow = FlowNode::Seq {
3353 children: vec![
3354 step("gate", "$.gate"),
3355 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3356 ],
3357 };
3358 let mut bp = minimal_bp(agent, flow);
3359 bp.metadata.strict_verdict_handling = Some(true);
3360 match Compiler::new(registry_with_echo()).compile(&bp) {
3361 Err(CompileError::VerdictValueUnhandled {
3362 agent,
3363 value,
3364 step_ref,
3365 ..
3366 }) => {
3367 assert_eq!(agent, "gate");
3368 assert_eq!(value, "PASS");
3369 assert_eq!(step_ref, "gate");
3370 }
3371 Err(other) => {
3372 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
3373 }
3374 Ok(_) => panic!(
3375 "expected compile-time rejection for a declared verdict value with no \
3376 downstream handler (part channel) under strict_verdict_handling=Some(true)"
3377 ),
3378 }
3379 }
3380
3381 #[test]
3388 fn verdict_omitted_blueprint_compiles_unchanged_with_empty_contracts() {
3389 let agent = gate_agent(None);
3390 let flow = FlowNode::Seq {
3391 children: vec![
3392 step("gate", "$.verdict"),
3393 FlowNode::Loop {
3394 counter: Expr::Path {
3395 at: "$.n".parse().expect("literal test path"),
3396 },
3397 cond: eq_cond("$.verdict", "BLOCKED"),
3398 body: Box::new(step("gate", "$.verdict")),
3399 max: 3,
3400 },
3401 branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
3402 ],
3403 };
3404 let bp = minimal_bp(agent, flow);
3405 let compiled = Compiler::new(registry_with_echo())
3406 .compile(&bp)
3407 .expect("a verdict-omitted Blueprint must compile unchanged");
3408 assert!(
3409 compiled.router.verdict_contracts.is_empty(),
3410 "no agent declared a verdict contract"
3411 );
3412 }
3413
3414 #[test]
3421 fn every_compile_error_diagnostic_kind_is_a_declared_lint() {
3422 let kinds = [
3423 "bound-agent-resolution",
3424 "unknown-agent-kind",
3425 "invalid-agent-spec",
3426 "worker-binding-missing",
3427 "unresolved-agent-ref",
3428 "duplicate-agent-name",
3429 "unresolved-operator-ref",
3430 "unresolved-meta-ref",
3431 "step-naming-collision",
3432 "invalid-projection-placement",
3433 "unresolved-audit-agent",
3434 "verdict-channel-mismatch",
3435 "verdict-value-not-in-contract",
3436 "verdict-value-unhandled",
3437 ];
3438 for kind in kinds {
3439 assert!(
3440 mlua_swarm_diag::lint_decl(kind).is_some(),
3441 "kind '{kind}' emitted by From<&CompileError> has no LINT_DECLS entry"
3442 );
3443 }
3444 }
3445
3446 #[test]
3447 fn invalid_spec_with_worker_binding_prefix_specializes_the_diagnostic_kind() {
3448 let err = CompileError::InvalidSpec {
3452 name: "greeter".into(),
3453 msg: format!("{WORKER_BINDING_REQUIRED_MSG_PREFIX}. Fix by either: (a) ..."),
3454 };
3455 let d = mlua_swarm_diag::Diagnostic::from(&err);
3456 assert_eq!(d.kind, "worker-binding-missing");
3457 assert_eq!(d.level, mlua_swarm_diag::DiagLevel::Error);
3458 assert!(matches!(d.stage, mlua_swarm_diag::DiagStage::CompileLint));
3459 assert!(d.message.contains("greeter"));
3460 let suggestion = d
3461 .suggestion
3462 .expect("specialized arm must carry a suggestion");
3463 assert!(suggestion.patch.contains("backend = \"ws_operator\""));
3464 assert_eq!(
3465 suggestion.applicability,
3466 mlua_swarm_diag::Applicability::HasPlaceholders
3467 );
3468 assert_eq!(
3469 d.docs_ref.expect("docs_ref must be set").uri,
3470 "mse://guides/bp-dsl-templates"
3471 );
3472 match d.span.expect("span must be set").element {
3473 mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "greeter"),
3474 other => panic!("expected Agent span, got {other:?}"),
3475 }
3476 }
3477
3478 #[test]
3479 fn generic_invalid_spec_maps_to_the_generic_kind() {
3480 let err = CompileError::InvalidSpec {
3481 name: "solo".into(),
3482 msg: "operator spec: 'operator_ref' (string) required".into(),
3483 };
3484 let d = mlua_swarm_diag::Diagnostic::from(&err);
3485 assert_eq!(d.kind, "invalid-agent-spec");
3486 assert!(
3487 d.suggestion.is_none(),
3488 "generic arm carries no canned patch"
3489 );
3490 }
3491
3492 #[test]
3493 fn verdict_value_not_in_contract_diagnostic_carries_suggestion_and_span() {
3494 let err = CompileError::VerdictValueNotInContract {
3495 where_: "Branch cond".into(),
3496 agent: "review".into(),
3497 value: "NOT_DECLARED".into(),
3498 values: vec!["PASS".into(), "BLOCKED".into()],
3499 };
3500 let d = mlua_swarm_diag::Diagnostic::from(&err);
3501 assert_eq!(d.kind, "verdict-value-not-in-contract");
3502 assert!(d.message.contains("NOT_DECLARED"));
3503 assert!(d.suggestion.is_some());
3504 match d.span.expect("span must be set").element {
3505 mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "review"),
3506 other => panic!("expected Agent span, got {other:?}"),
3507 }
3508 }
3509}
3510
3511#[cfg(test)]
3513mod subprocess_embed_compile_tests {
3514 use super::*;
3515 use mlua_swarm_schema::{current_schema_version, SubprocessDef, SubprocessOverrides};
3516
3517 fn subprocess_agent(name: &str, runner: Option<Runner>) -> AgentDef {
3518 AgentDef {
3519 name: name.to_string(),
3520 kind: AgentKind::Subprocess,
3521 spec: serde_json::json!({}),
3522 profile: Some(AgentProfile {
3523 system_prompt: "you are a headless worker".to_string(),
3524 model: Some("profile-model".to_string()),
3525 tools: vec!["Read".to_string()],
3526 ..Default::default()
3527 }),
3528 meta: None,
3529 runner,
3530 runner_ref: None,
3531 verdict: None,
3532 }
3533 }
3534
3535 fn echo_def(name: &str) -> SubprocessDef {
3536 SubprocessDef {
3537 name: name.to_string(),
3538 argv: vec!["sh".to_string(), "-c".to_string(), "cat".to_string()],
3539 stdin: Some("{prompt}".to_string()),
3540 env: Default::default(),
3541 cwd: None,
3542 output: None,
3543 stream_mode: None,
3544 }
3545 }
3546
3547 fn bp_with(agents: Vec<AgentDef>, subprocesses: Vec<SubprocessDef>) -> Blueprint {
3548 Blueprint {
3549 schema_version: current_schema_version(),
3550 id: "gh83-ut".into(),
3551 flow: FlowNode::Seq { children: vec![] },
3552 agents,
3553 operators: vec![],
3554 metas: vec![],
3555 hints: Default::default(),
3556 strategy: Default::default(),
3557 metadata: BlueprintMetadata::default(),
3558 spawner_hints: Default::default(),
3559 default_agent_kind: AgentKind::Operator,
3560 default_operator_kind: None,
3561 default_init_ctx: None,
3562 default_agent_ctx: None,
3563 default_context_policy: None,
3564 projection_placement: None,
3565 audits: vec![],
3566 degradation_policy: None,
3567 runners: vec![],
3568 default_runner: None,
3569 subprocesses,
3570 check_policy: None,
3571 blueprint_ref_includes: vec![],
3572 }
3573 }
3574
3575 fn subprocess_runner(template: &str) -> Runner {
3576 Runner::Subprocess {
3577 template: template.to_string(),
3578 overrides: SubprocessOverrides::default(),
3579 }
3580 }
3581
3582 #[test]
3583 fn validate_placeholders_accepts_closed_set_and_json_braces() {
3584 for ok in [
3585 "{system} {system_file} {prompt} {model} {tools_csv} {work_dir} {task_id} {attempt}",
3586 r#"echo '{"result": "ok", "nested": {"a": 1}}'"#,
3587 "no placeholders at all",
3588 "unmatched { brace",
3589 ] {
3590 validate_embed_placeholders(ok, "ut").expect("must be accepted");
3591 }
3592 }
3593
3594 #[test]
3595 fn validate_placeholders_rejects_unknown_token() {
3596 let err = validate_embed_placeholders("--flag {evil}", "argv[1]").unwrap_err();
3597 assert!(err.contains("'{evil}'"), "token named: {err}");
3598 assert!(err.contains("closed set"), "closed set listed: {err}");
3599 }
3600
3601 #[test]
3605 fn validate_placeholders_descends_into_literal_braces() {
3606 validate_embed_placeholders(r#"{"task": "{prompt}"}"#, "stdin")
3607 .expect("nested closed-set token must be accepted");
3608 let err = validate_embed_placeholders(r#"{"task": "{evil}"}"#, "stdin").unwrap_err();
3609 assert!(
3610 err.contains("'{evil}'"),
3611 "nested unknown token caught: {err}"
3612 );
3613 }
3614
3615 #[test]
3616 fn hint_resolution_finds_declared_template() {
3617 let agent = subprocess_agent("headless", Some(subprocess_runner("echo")));
3618 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
3619 let hint = resolve_subprocess_template_hint(&bp, &agent)
3620 .expect("resolves")
3621 .expect("Runner::Subprocess must synthesize a hint");
3622 assert_eq!(hint[SUBPROCESS_TEMPLATE_HINT_KEY]["name"], "echo");
3623 assert!(hint.get(SUBPROCESS_OVERRIDES_HINT_KEY).is_some());
3624 }
3625
3626 #[test]
3627 fn hint_resolution_unknown_template_is_invalid_spec() {
3628 let agent = subprocess_agent("headless", Some(subprocess_runner("nope")));
3629 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
3630 let err = resolve_subprocess_template_hint(&bp, &agent).unwrap_err();
3631 let msg = format!("{err}");
3632 assert!(msg.contains("'nope'"), "missing template named: {msg}");
3633 assert!(msg.contains("echo"), "defined templates listed: {msg}");
3634 }
3635
3636 #[test]
3637 fn hint_resolution_none_without_subprocess_runner() {
3638 let agent = subprocess_agent("headless", None);
3639 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
3640 let hint = resolve_subprocess_template_hint(&bp, &agent).expect("resolves");
3641 assert!(hint.is_none(), "spec-based agents keep the historical path");
3642 }
3643
3644 #[test]
3645 fn build_embed_rejects_unknown_placeholder() {
3646 let agent = subprocess_agent("headless", None);
3647 let mut def = echo_def("echo");
3648 def.argv.push("--x={evil}".to_string());
3649 let err = SubprocessProcessSpawnerFactory::build_embed(
3650 &agent,
3651 &serde_json::to_value(&def).unwrap(),
3652 None,
3653 )
3654 .unwrap_err();
3655 assert!(format!("{err}").contains("'{evil}'"));
3656 }
3657
3658 #[test]
3659 fn build_embed_rejects_output_with_stream_mode() {
3660 let agent = subprocess_agent("headless", None);
3661 let mut def = echo_def("echo");
3662 def.stream_mode = Some("ndjson_lines".to_string());
3663 def.output = Some(mlua_swarm_schema::SubprocessOutput {
3664 format: Some("json".to_string()),
3665 result_ptr: None,
3666 ok_from: None,
3667 stats: None,
3668 });
3669 let err = SubprocessProcessSpawnerFactory::build_embed(
3670 &agent,
3671 &serde_json::to_value(&def).unwrap(),
3672 None,
3673 )
3674 .unwrap_err();
3675 assert!(format!("{err}").contains("plain-mode"));
3676 }
3677
3678 #[test]
3679 fn build_embed_rejects_malformed_result_ptr_and_ok_from() {
3680 let agent = subprocess_agent("headless", None);
3681 let mut def = echo_def("echo");
3682 def.output = Some(mlua_swarm_schema::SubprocessOutput {
3683 format: None,
3684 result_ptr: Some("result".to_string()),
3685 ok_from: None,
3686 stats: None,
3687 });
3688 let err = SubprocessProcessSpawnerFactory::build_embed(
3689 &agent,
3690 &serde_json::to_value(&def).unwrap(),
3691 None,
3692 )
3693 .unwrap_err();
3694 assert!(format!("{err}").contains("JSON Pointer"));
3695
3696 let mut def = echo_def("echo");
3697 def.output = Some(mlua_swarm_schema::SubprocessOutput {
3698 format: None,
3699 result_ptr: None,
3700 ok_from: Some("status".to_string()),
3701 stats: None,
3702 });
3703 let err = SubprocessProcessSpawnerFactory::build_embed(
3704 &agent,
3705 &serde_json::to_value(&def).unwrap(),
3706 None,
3707 )
3708 .unwrap_err();
3709 assert!(format!("{err}").contains("exit_code"));
3710 }
3711
3712 #[test]
3713 fn build_embed_bakes_profile_with_override_precedence() {
3714 let agent = subprocess_agent("headless", None);
3715 let def = echo_def("echo");
3716 let overrides = SubprocessOverrides {
3717 model: Some("override-model".to_string()),
3718 tools: vec!["Bash".to_string(), "Write".to_string()],
3719 cwd: Some("/tmp/override-wd".to_string()),
3720 };
3721 let sp = SubprocessProcessSpawnerFactory::build_embed(
3722 &agent,
3723 &serde_json::to_value(&def).unwrap(),
3724 Some(&serde_json::to_value(&overrides).unwrap()),
3725 )
3726 .expect("builds");
3727 let embed = sp.embed.as_ref().expect("embed template baked");
3728 assert_eq!(embed.model.as_deref(), Some("override-model"));
3729 assert_eq!(embed.tools_csv, "Bash,Write");
3730 assert_eq!(embed.cwd.as_deref(), Some("/tmp/override-wd"));
3731 assert_eq!(
3732 embed.system_prompt.as_deref(),
3733 Some("you are a headless worker")
3734 );
3735 }
3736}