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 {
697 resolve_subprocess_template_hint(bp, ad)?
698 } else {
699 None
700 };
701 let spawner = factory.build(ad, subprocess_hint.as_ref().or(hint))?;
702 routes.insert(ad.name.clone(), spawner);
703 }
704
705 let strict_verdict_handling = bp.metadata.strict_verdict_handling.unwrap_or(false);
722 verify_verdict_conds(&bp.flow, &verdict_contracts, strict_verdict_handling)?;
723
724 if bp.strategy.strict_refs {
725 verify_refs(&bp.flow, &routes, self.default_spawner.is_some())?;
726 }
727
728 let (step_naming, step_naming_warnings) = StepNaming::from_blueprint(bp)?;
736 for warning in &step_naming_warnings {
737 tracing::warn!(
738 name = %warning.name,
739 first_step_ref = %warning.first_step_ref,
740 second_step_ref = %warning.second_step_ref,
741 "StepNaming: undeclared steps' canonical/alias names collide; \
742 the step whose own ref matches the name keeps it (data-plane priority)"
743 );
744 }
745
746 let projection_placement =
754 ProjectionPlacement::from_spec(bp.projection_placement.as_ref())?;
755
756 let router = Arc::new(CompiledAgentTable {
757 routes,
758 default: self.default_spawner.clone(),
759 verdict_contracts,
760 });
761 Ok(CompiledBlueprint {
762 router,
763 flow: bp.flow.clone(),
764 metadata: bp.metadata.clone(),
765 step_naming: Arc::new(step_naming),
766 projection_placement: Arc::new(projection_placement),
767 })
768 }
769}
770
771fn verify_refs(
774 node: &FlowNode,
775 routes: &HashMap<String, Arc<dyn SpawnerAdapter>>,
776 has_default: bool,
777) -> Result<(), CompileError> {
778 let mut refs: Vec<String> = Vec::new();
779 collect_refs(node, &mut refs);
780 for r in refs {
781 if !routes.contains_key(&r) && !has_default {
782 return Err(CompileError::UnresolvedRef(r));
783 }
784 }
785 Ok(())
786}
787
788fn collect_refs(node: &FlowNode, out: &mut Vec<String>) {
789 match node {
790 FlowNode::Step { ref_, .. } => out.push(ref_.clone()),
791 FlowNode::Seq { children } => {
792 for c in children {
793 collect_refs(c, out);
794 }
795 }
796 FlowNode::Branch { then_, else_, .. } => {
797 collect_refs(then_, out);
798 collect_refs(else_, out);
799 }
800 FlowNode::Fanout { body, .. } => collect_refs(body, out),
801 FlowNode::Loop { body, .. } => collect_refs(body, out),
802 FlowNode::Try { body, catch, .. } => {
803 collect_refs(body, out);
804 collect_refs(catch, out);
805 }
806 FlowNode::Assign { .. } => {} }
808}
809
810fn collect_step_meta_refs(node: &FlowNode, out: &mut Vec<(String, String)>) {
818 match node {
819 FlowNode::Step { ref_, in_, .. } => {
820 if let Expr::Lit { value } = in_ {
821 if let Some(meta_ref) = static_step_meta_ref(value) {
822 out.push((format!("Step '{ref_}' $step_meta.ref"), meta_ref));
823 }
824 }
825 }
826 FlowNode::Seq { children } => {
827 for c in children {
828 collect_step_meta_refs(c, out);
829 }
830 }
831 FlowNode::Branch { then_, else_, .. } => {
832 collect_step_meta_refs(then_, out);
833 collect_step_meta_refs(else_, out);
834 }
835 FlowNode::Fanout { body, .. } => collect_step_meta_refs(body, out),
836 FlowNode::Loop { body, .. } => collect_step_meta_refs(body, out),
837 FlowNode::Try { body, catch, .. } => {
838 collect_step_meta_refs(body, out);
839 collect_step_meta_refs(catch, out);
840 }
841 FlowNode::Assign { .. } => {} }
843}
844
845fn static_step_meta_ref(value: &Value) -> Option<String> {
852 value
853 .as_object()?
854 .get("$step_meta")?
855 .as_object()?
856 .get("ref")?
857 .as_str()
858 .map(str::to_string)
859}
860
861fn verify_verdict_conds(
873 flow: &FlowNode,
874 verdict_contracts: &HashMap<String, VerdictContract>,
875 strict_verdict_handling: bool,
876) -> Result<(), CompileError> {
877 let mut step_outputs: HashMap<String, String> = HashMap::new();
878 let mut step_agents: HashMap<String, String> = HashMap::new();
879 collect_step_outputs_and_agents(flow, &mut step_outputs, &mut step_agents);
880
881 let mut errors: Vec<CompileError> = Vec::new();
882 let mut referenced_values: HashMap<String, std::collections::HashSet<String>> = HashMap::new();
883 collect_verdict_conds(
884 flow,
885 &step_outputs,
886 verdict_contracts,
887 &mut referenced_values,
888 &mut errors,
889 );
890 check_unhandled_verdict_values(
891 verdict_contracts,
892 &referenced_values,
893 &step_agents,
894 strict_verdict_handling,
895 &mut errors,
896 );
897 match errors.into_iter().next() {
898 Some(e) => Err(e),
899 None => Ok(()),
900 }
901}
902
903fn collect_step_outputs_and_agents(
918 node: &FlowNode,
919 out: &mut HashMap<String, String>,
920 step_agents: &mut HashMap<String, String>,
921) {
922 match node {
923 FlowNode::Step {
924 ref_,
925 out: out_expr,
926 ..
927 } => {
928 if let Expr::Path { at } = out_expr {
929 out.insert(at.to_string(), ref_.clone());
930 }
931 step_agents
932 .entry(ref_.clone())
933 .or_insert_with(|| ref_.clone());
934 }
935 FlowNode::Seq { children } => {
936 for c in children {
937 collect_step_outputs_and_agents(c, out, step_agents);
938 }
939 }
940 FlowNode::Branch { then_, else_, .. } => {
941 collect_step_outputs_and_agents(then_, out, step_agents);
942 collect_step_outputs_and_agents(else_, out, step_agents);
943 }
944 FlowNode::Fanout { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
945 FlowNode::Loop { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
946 FlowNode::Try { body, catch, .. } => {
947 collect_step_outputs_and_agents(body, out, step_agents);
948 collect_step_outputs_and_agents(catch, out, step_agents);
949 }
950 FlowNode::Assign { .. } => {} }
952}
953
954fn collect_verdict_conds(
959 node: &FlowNode,
960 step_outputs: &HashMap<String, String>,
961 verdict_contracts: &HashMap<String, VerdictContract>,
962 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
963 errors: &mut Vec<CompileError>,
964) {
965 match node {
966 FlowNode::Branch { cond, then_, else_ } => {
967 lint_cond_expr(
968 cond,
969 "Branch cond",
970 step_outputs,
971 verdict_contracts,
972 referenced_values,
973 errors,
974 );
975 collect_verdict_conds(
976 then_,
977 step_outputs,
978 verdict_contracts,
979 referenced_values,
980 errors,
981 );
982 collect_verdict_conds(
983 else_,
984 step_outputs,
985 verdict_contracts,
986 referenced_values,
987 errors,
988 );
989 }
990 FlowNode::Loop { cond, body, .. } => {
991 lint_cond_expr(
992 cond,
993 "Loop cond",
994 step_outputs,
995 verdict_contracts,
996 referenced_values,
997 errors,
998 );
999 collect_verdict_conds(
1000 body,
1001 step_outputs,
1002 verdict_contracts,
1003 referenced_values,
1004 errors,
1005 );
1006 }
1007 FlowNode::Seq { children } => {
1008 for c in children {
1009 collect_verdict_conds(
1010 c,
1011 step_outputs,
1012 verdict_contracts,
1013 referenced_values,
1014 errors,
1015 );
1016 }
1017 }
1018 FlowNode::Fanout { body, .. } => collect_verdict_conds(
1019 body,
1020 step_outputs,
1021 verdict_contracts,
1022 referenced_values,
1023 errors,
1024 ),
1025 FlowNode::Try { body, catch, .. } => {
1026 collect_verdict_conds(
1027 body,
1028 step_outputs,
1029 verdict_contracts,
1030 referenced_values,
1031 errors,
1032 );
1033 collect_verdict_conds(
1034 catch,
1035 step_outputs,
1036 verdict_contracts,
1037 referenced_values,
1038 errors,
1039 );
1040 }
1041 FlowNode::Step { .. } | FlowNode::Assign { .. } => {}
1042 }
1043}
1044
1045fn lint_cond_expr(
1054 expr: &Expr,
1055 where_: &str,
1056 step_outputs: &HashMap<String, String>,
1057 verdict_contracts: &HashMap<String, VerdictContract>,
1058 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
1059 errors: &mut Vec<CompileError>,
1060) {
1061 match expr {
1062 Expr::Eq { lhs, rhs } | Expr::Ne { lhs, rhs } => {
1063 if let Some((path, lit)) = path_lit_operands(lhs, rhs) {
1064 resolve_and_check(
1065 path,
1066 &[lit],
1067 where_,
1068 step_outputs,
1069 verdict_contracts,
1070 referenced_values,
1071 errors,
1072 );
1073 }
1074 }
1075 Expr::In { needle, haystack } => {
1076 if let (
1077 Expr::Path { at },
1078 Expr::Lit {
1079 value: Value::Array(items),
1080 },
1081 ) = (needle.as_ref(), haystack.as_ref())
1082 {
1083 let lits: Vec<&Value> = items.iter().collect();
1084 resolve_and_check(
1085 at,
1086 &lits,
1087 where_,
1088 step_outputs,
1089 verdict_contracts,
1090 referenced_values,
1091 errors,
1092 );
1093 }
1094 }
1095 Expr::And { args } | Expr::Or { args } => {
1096 for a in args {
1097 lint_cond_expr(
1098 a,
1099 where_,
1100 step_outputs,
1101 verdict_contracts,
1102 referenced_values,
1103 errors,
1104 );
1105 }
1106 }
1107 Expr::Not { arg } => lint_cond_expr(
1108 arg,
1109 where_,
1110 step_outputs,
1111 verdict_contracts,
1112 referenced_values,
1113 errors,
1114 ),
1115 _ => {}
1116 }
1117}
1118
1119fn path_lit_operands<'a>(lhs: &'a Expr, rhs: &'a Expr) -> Option<(&'a Path, &'a Value)> {
1125 match (lhs, rhs) {
1126 (Expr::Path { at }, Expr::Lit { value }) => Some((at, value)),
1127 (Expr::Lit { value }, Expr::Path { at }) => Some((at, value)),
1128 _ => None,
1129 }
1130}
1131
1132fn resolve_and_check(
1147 path: &Path,
1148 lits: &[&Value],
1149 where_: &str,
1150 step_outputs: &HashMap<String, String>,
1151 verdict_contracts: &HashMap<String, VerdictContract>,
1152 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
1153 errors: &mut Vec<CompileError>,
1154) {
1155 let path_str = path.to_string();
1156 let (agent, actual_shape) = if let Some(agent) = step_outputs.get(&path_str) {
1157 (agent, "body")
1158 } else if let Some(stripped) = path_str.strip_suffix(".parts.verdict") {
1159 match step_outputs.get(stripped) {
1160 Some(agent) => (agent, "part"),
1161 None => return,
1162 }
1163 } else {
1164 return;
1165 };
1166
1167 let Some(contract) = verdict_contracts.get(agent) else {
1168 tracing::warn!(
1169 agent = %agent,
1170 where_ = %where_,
1171 "cond references agent output but no verdict contract declared"
1172 );
1173 return;
1174 };
1175
1176 let expected_channel = match contract.channel {
1177 VerdictChannel::Body => "body",
1178 VerdictChannel::Part => "part",
1179 };
1180 if expected_channel != actual_shape {
1181 errors.push(CompileError::VerdictChannelMismatch {
1182 where_: where_.to_string(),
1183 agent: agent.clone(),
1184 expected_channel: expected_channel.to_string(),
1185 actual_shape: actual_shape.to_string(),
1186 });
1187 return;
1188 }
1189
1190 for lit in lits {
1191 let value_str = lit
1192 .as_str()
1193 .map(str::to_string)
1194 .unwrap_or_else(|| lit.to_string());
1195 if !contract.values.iter().any(|v| v == &value_str) {
1196 errors.push(CompileError::VerdictValueNotInContract {
1197 where_: where_.to_string(),
1198 agent: agent.clone(),
1199 value: value_str.clone(),
1200 values: contract.values.clone(),
1201 });
1202 }
1203 referenced_values
1210 .entry(agent.clone())
1211 .or_default()
1212 .insert(value_str);
1213 }
1214}
1215
1216fn check_unhandled_verdict_values(
1234 verdict_contracts: &HashMap<String, VerdictContract>,
1235 referenced_values: &HashMap<String, std::collections::HashSet<String>>,
1236 step_agents: &HashMap<String, String>,
1237 strict_verdict_handling: bool,
1238 errors: &mut Vec<CompileError>,
1239) {
1240 for finding in fold_unhandled_verdict_values(verdict_contracts, referenced_values, step_agents)
1241 {
1242 if strict_verdict_handling {
1243 errors.push(CompileError::VerdictValueUnhandled {
1244 agent: finding.agent,
1245 value: finding.value,
1246 declared_values: finding.declared_values,
1247 step_ref: finding.step_ref,
1248 });
1249 } else {
1250 tracing::warn!(
1251 agent = %finding.agent,
1252 value = %finding.value,
1253 step_ref = %finding.step_ref,
1254 "declared verdict value has no downstream cond handler; \
1255 opt in to `metadata.strict_verdict_handling` to reject at compile"
1256 );
1257 }
1258 }
1259}
1260
1261#[derive(Debug, Clone, PartialEq, Eq)]
1272pub struct UnhandledVerdictValue {
1273 pub agent: String,
1275 pub value: String,
1277 pub declared_values: Vec<String>,
1279 pub step_ref: String,
1281}
1282
1283pub fn unhandled_verdict_values(
1300 flow: &FlowNode,
1301 verdict_contracts: &HashMap<String, VerdictContract>,
1302) -> Vec<UnhandledVerdictValue> {
1303 let mut step_outputs: HashMap<String, String> = HashMap::new();
1304 let mut step_agents: HashMap<String, String> = HashMap::new();
1305 collect_step_outputs_and_agents(flow, &mut step_outputs, &mut step_agents);
1306
1307 let mut referenced_values: HashMap<String, std::collections::HashSet<String>> = HashMap::new();
1308 let mut discarded_errors: Vec<CompileError> = Vec::new();
1309 collect_verdict_conds(
1310 flow,
1311 &step_outputs,
1312 verdict_contracts,
1313 &mut referenced_values,
1314 &mut discarded_errors,
1315 );
1316 fold_unhandled_verdict_values(verdict_contracts, &referenced_values, &step_agents)
1317}
1318
1319#[derive(Debug, Clone, PartialEq, Eq)]
1333pub struct AgentContractUnread {
1334 pub agent: String,
1336 pub declared_values: Vec<String>,
1338 pub step_ref: String,
1340}
1341
1342pub fn agents_with_all_verdict_values_unread(
1353 flow: &FlowNode,
1354 verdict_contracts: &HashMap<String, VerdictContract>,
1355) -> Vec<AgentContractUnread> {
1356 let per_value = unhandled_verdict_values(flow, verdict_contracts);
1357 let mut unread_counts: HashMap<String, usize> = HashMap::new();
1358 for finding in &per_value {
1359 *unread_counts.entry(finding.agent.clone()).or_default() += 1;
1360 }
1361 let mut agents: Vec<&String> = verdict_contracts.keys().collect();
1362 agents.sort();
1363 let mut out = Vec::new();
1364 for agent in agents {
1365 let contract = &verdict_contracts[agent];
1366 let declared = contract.values.len();
1367 if declared == 0 {
1368 continue;
1369 }
1370 let unread = unread_counts.get(agent).copied().unwrap_or(0);
1371 if unread != declared {
1372 continue;
1373 }
1374 let step_ref = per_value
1378 .iter()
1379 .find(|f| &f.agent == agent)
1380 .map(|f| f.step_ref.clone())
1381 .unwrap_or_else(|| agent.clone());
1382 out.push(AgentContractUnread {
1383 agent: agent.clone(),
1384 declared_values: contract.values.clone(),
1385 step_ref,
1386 });
1387 }
1388 out
1389}
1390
1391fn fold_unhandled_verdict_values(
1402 verdict_contracts: &HashMap<String, VerdictContract>,
1403 referenced_values: &HashMap<String, std::collections::HashSet<String>>,
1404 step_agents: &HashMap<String, String>,
1405) -> Vec<UnhandledVerdictValue> {
1406 let mut agents: Vec<&String> = verdict_contracts.keys().collect();
1407 agents.sort();
1408 let mut findings = Vec::new();
1409 for agent in agents {
1410 let contract = &verdict_contracts[agent];
1411 let referenced = referenced_values.get(agent);
1412 let step_ref = step_agents
1413 .get(agent)
1414 .cloned()
1415 .unwrap_or_else(|| agent.clone());
1416 for value in &contract.values {
1417 let handled = referenced.map(|set| set.contains(value)).unwrap_or(false);
1418 if handled {
1419 continue;
1420 }
1421 findings.push(UnhandledVerdictValue {
1422 agent: agent.clone(),
1423 value: value.clone(),
1424 declared_values: contract.values.clone(),
1425 step_ref: step_ref.clone(),
1426 });
1427 }
1428 }
1429 findings
1430}
1431
1432pub struct CompiledAgentTable {
1445 pub(crate) routes: HashMap<String, Arc<dyn SpawnerAdapter>>,
1446 pub(crate) default: Option<Arc<dyn SpawnerAdapter>>,
1447 pub(crate) verdict_contracts: HashMap<String, VerdictContract>,
1451}
1452
1453impl CompiledAgentTable {
1454 pub fn has_route(&self, agent: &str) -> bool {
1457 self.routes.contains_key(agent)
1458 }
1459 pub fn routed_agents(&self) -> Vec<String> {
1461 self.routes.keys().cloned().collect()
1462 }
1463 pub fn verdict_contract_for(&self, agent: &str) -> Option<&VerdictContract> {
1467 self.verdict_contracts.get(agent)
1468 }
1469}
1470
1471#[async_trait]
1472impl SpawnerAdapter for CompiledAgentTable {
1473 async fn spawn(
1474 &self,
1475 engine: &Engine,
1476 ctx: &Ctx,
1477 task_id: StepId,
1478 attempt: u32,
1479 token: CapToken,
1480 ) -> Result<Box<dyn Worker>, SpawnError> {
1481 let sp = self
1482 .routes
1483 .get(&ctx.agent)
1484 .cloned()
1485 .or_else(|| self.default.clone())
1486 .ok_or_else(|| SpawnError::NotRegistered(ctx.agent.clone()))?;
1487 sp.spawn(engine, ctx, task_id, attempt, token).await
1488 }
1489}
1490
1491pub struct SubprocessProcessSpawnerFactory;
1522
1523impl SpawnerFactoryKind for SubprocessProcessSpawnerFactory {
1524 const KIND: AgentKind = AgentKind::Subprocess;
1525 type Worker = crate::worker::process_spawner::ProcessWorker;
1526}
1527
1528pub const SUBPROCESS_TEMPLATE_HINT_KEY: &str = "subprocess_template";
1531pub const SUBPROCESS_OVERRIDES_HINT_KEY: &str = "subprocess_overrides";
1533
1534fn validate_embed_placeholders(s: &str, where_: &str) -> Result<(), String> {
1548 let mut rest = s;
1549 while let Some(start) = rest.find('{') {
1550 let after = &rest[start + 1..];
1551 let Some(end) = after.find('}') else {
1552 break;
1553 };
1554 let token = &after[..end];
1555 let is_candidate =
1556 !token.is_empty() && token.chars().all(|c| c.is_ascii_lowercase() || c == '_');
1557 if is_candidate {
1558 if !crate::worker::process_spawner::EMBED_PLACEHOLDERS.contains(&token) {
1559 return Err(format!(
1560 "unknown placeholder '{{{token}}}' in {where_}; closed set is \
1561 {{system, system_file, prompt, model, tools_csv, work_dir, task_id, attempt}}"
1562 ));
1563 }
1564 rest = &after[end + 1..];
1565 } else {
1566 rest = after;
1571 }
1572 }
1573 Ok(())
1574}
1575
1576fn resolve_subprocess_template_hint(
1582 bp: &Blueprint,
1583 ad: &AgentDef,
1584) -> Result<Option<Value>, CompileError> {
1585 let invalid = |msg: String| CompileError::InvalidSpec {
1586 name: ad.name.clone(),
1587 msg,
1588 };
1589 let runner = mlua_swarm_schema::resolve_runner(bp, ad).map_err(|e| invalid(e.to_string()))?;
1590 let Some(Runner::Subprocess {
1591 template,
1592 overrides,
1593 }) = runner
1594 else {
1595 return Ok(None);
1596 };
1597 let def = bp
1598 .subprocesses
1599 .iter()
1600 .find(|d| d.name == template)
1601 .ok_or_else(|| {
1602 let mut names: Vec<&str> = bp.subprocesses.iter().map(|d| d.name.as_str()).collect();
1603 names.sort_unstable();
1604 invalid(format!(
1605 "Runner::Subprocess template '{template}' not found in \
1606 Blueprint.subprocesses (defined: [{}])",
1607 names.join(", ")
1608 ))
1609 })?;
1610 Ok(Some(serde_json::json!({
1611 SUBPROCESS_TEMPLATE_HINT_KEY: def,
1612 SUBPROCESS_OVERRIDES_HINT_KEY: overrides,
1613 })))
1614}
1615
1616impl SubprocessProcessSpawnerFactory {
1617 fn build_embed(
1622 agent_def: &AgentDef,
1623 template: &Value,
1624 overrides: Option<&Value>,
1625 ) -> Result<ProcessSpawner, CompileError> {
1626 use crate::worker::process_spawner::EmbedTemplate;
1627 use mlua_swarm_schema::{SubprocessDef, SubprocessOverrides};
1628
1629 let agent_name = &agent_def.name;
1630 let invalid = |msg: String| CompileError::InvalidSpec {
1631 name: agent_name.to_string(),
1632 msg,
1633 };
1634 let def: SubprocessDef = serde_json::from_value(template.clone())
1635 .map_err(|e| invalid(format!("subprocess_template hint: {e}")))?;
1636 let overrides: SubprocessOverrides = match overrides {
1637 Some(v) => serde_json::from_value(v.clone())
1638 .map_err(|e| invalid(format!("subprocess_overrides hint: {e}")))?,
1639 None => SubprocessOverrides::default(),
1640 };
1641
1642 if def.argv.is_empty() {
1643 return Err(invalid(format!(
1644 "SubprocessDef '{}': argv must not be empty",
1645 def.name
1646 )));
1647 }
1648 for (i, a) in def.argv.iter().enumerate() {
1650 validate_embed_placeholders(a, &format!("argv[{i}]")).map_err(&invalid)?;
1651 }
1652 if let Some(stdin) = &def.stdin {
1653 validate_embed_placeholders(stdin, "stdin").map_err(&invalid)?;
1654 }
1655 for (k, v) in &def.env {
1656 validate_embed_placeholders(v, &format!("env['{k}']")).map_err(&invalid)?;
1657 }
1658 if let Some(cwd) = &def.cwd {
1659 validate_embed_placeholders(cwd, "cwd").map_err(&invalid)?;
1660 }
1661 let stream_mode = match def.stream_mode.as_deref() {
1662 Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
1663 Some("sse_events") => Some(StreamMode::SseEvents),
1664 Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
1665 Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
1666 None => None,
1667 };
1668 if let Some(output) = &def.output {
1669 if stream_mode.is_some() {
1670 return Err(invalid(format!(
1671 "SubprocessDef '{}': output normalization is a plain-mode \
1672 declaration; remove either `output` or `stream_mode`",
1673 def.name
1674 )));
1675 }
1676 if let Some(format) = output.format.as_deref() {
1677 if format != "json" {
1678 return Err(invalid(format!(
1679 "SubprocessDef '{}': unknown output.format '{format}' \
1680 (supported: \"json\")",
1681 def.name
1682 )));
1683 }
1684 }
1685 if let Some(ptr) = output.result_ptr.as_deref() {
1686 if !ptr.starts_with('/') {
1687 return Err(invalid(format!(
1688 "SubprocessDef '{}': output.result_ptr '{ptr}' is not a \
1689 JSON Pointer (RFC 6901 — must start with '/')",
1690 def.name
1691 )));
1692 }
1693 }
1694 if let Some(ok_from) = output.ok_from.as_deref() {
1695 if ok_from != "exit_code" && !ok_from.starts_with('/') {
1696 return Err(invalid(format!(
1697 "SubprocessDef '{}': output.ok_from '{ok_from}' must be \
1698 \"exit_code\" or a JSON Pointer (starting with '/')",
1699 def.name
1700 )));
1701 }
1702 }
1703 }
1704
1705 let profile = agent_def.profile.as_ref();
1708 let system_prompt = profile
1709 .map(|p| p.system_prompt.clone())
1710 .filter(|s| !s.is_empty());
1711 let model = overrides
1712 .model
1713 .clone()
1714 .or_else(|| profile.and_then(|p| p.model.clone()));
1715 let tools: Vec<String> = if overrides.tools.is_empty() {
1716 profile.map(|p| p.tools.clone()).unwrap_or_default()
1717 } else {
1718 overrides.tools.clone()
1719 };
1720 let cwd = overrides.cwd.clone().or_else(|| def.cwd.clone());
1722 if let Some(c) = &cwd {
1723 validate_embed_placeholders(c, "overrides.cwd").map_err(&invalid)?;
1724 }
1725
1726 let program = def.argv[0].clone();
1727 let sp = ProcessSpawner {
1728 program,
1729 args: Vec::new(),
1730 use_stdin: def.stdin.is_some(),
1731 stream_mode,
1732 embed: Some(EmbedTemplate {
1733 argv: def.argv,
1734 stdin: def.stdin,
1735 env: def.env,
1736 cwd,
1737 output: def.output,
1738 system_prompt,
1739 model,
1740 tools_csv: tools.join(","),
1741 }),
1742 };
1743 Ok(sp)
1744 }
1745}
1746
1747impl SpawnerFactory for SubprocessProcessSpawnerFactory {
1748 fn build(
1749 &self,
1750 agent_def: &AgentDef,
1751 hint: Option<&Value>,
1752 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1753 if let Some(template) = hint.and_then(|h| h.get(SUBPROCESS_TEMPLATE_HINT_KEY)) {
1757 let overrides = hint.and_then(|h| h.get(SUBPROCESS_OVERRIDES_HINT_KEY));
1758 return Self::build_embed(agent_def, template, overrides).map(|sp| {
1759 let arc: Arc<dyn SpawnerAdapter> = Arc::new(sp);
1760 arc
1761 });
1762 }
1763 let agent_name = &agent_def.name;
1764 let spec = &agent_def.spec;
1765 let invalid = |msg: String| CompileError::InvalidSpec {
1766 name: agent_name.to_string(),
1767 msg,
1768 };
1769 let program = spec
1770 .get("program")
1771 .and_then(|v| v.as_str())
1772 .ok_or_else(|| invalid("shell spec: 'program' (string) required".into()))?
1773 .to_string();
1774 let args: Vec<String> = spec
1775 .get("args")
1776 .and_then(|v| v.as_array())
1777 .map(|a| {
1778 a.iter()
1779 .filter_map(|x| x.as_str().map(|s| s.to_string()))
1780 .collect()
1781 })
1782 .unwrap_or_default();
1783 let use_stdin = spec
1784 .get("use_stdin")
1785 .and_then(|v| v.as_bool())
1786 .unwrap_or(true);
1787 let stream_mode = match spec.get("stream_mode").and_then(|v| v.as_str()) {
1788 Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
1789 Some("sse_events") => Some(StreamMode::SseEvents),
1790 Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
1791 Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
1792 None => None,
1793 };
1794
1795 let mut sp = ProcessSpawner {
1796 program,
1797 args,
1798 use_stdin,
1799 stream_mode,
1800 embed: None,
1801 };
1802 if let Some(mode) = sp.stream_mode.clone() {
1803 sp = sp.stream_mode(mode);
1804 }
1805 Ok(Arc::new(sp))
1806 }
1807}
1808
1809pub struct LuaInProcessSpawnerFactory {
1840 registry: HashMap<String, WorkerFn>,
1841 bridges: HashMap<String, HostBridge>,
1842}
1843
1844#[derive(Clone)]
1856pub struct HostBridge(
1857 Arc<dyn Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync>,
1858);
1859
1860impl HostBridge {
1861 pub fn new<F>(f: F) -> Self
1863 where
1864 F: Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync + 'static,
1865 {
1866 Self(Arc::new(f))
1867 }
1868
1869 pub fn call(&self, arg: serde_json::Value) -> Result<serde_json::Value, String> {
1873 (self.0)(arg)
1874 }
1875}
1876
1877#[derive(Clone)]
1884pub struct LuaScriptSource {
1885 pub source: String,
1887 pub label: String,
1890}
1891
1892impl LuaScriptSource {
1893 pub fn new(source: impl Into<String>, label: impl Into<String>) -> Self {
1895 Self {
1896 source: source.into(),
1897 label: label.into(),
1898 }
1899 }
1900}
1901
1902impl LuaInProcessSpawnerFactory {
1903 pub fn new() -> Self {
1905 Self {
1906 registry: HashMap::new(),
1907 bridges: HashMap::new(),
1908 }
1909 }
1910
1911 pub fn with_bridge(mut self, name: impl Into<String>, bridge: HostBridge) -> Self {
1918 self.bridges.insert(name.into(), bridge);
1919 self
1920 }
1921
1922 pub fn register_lua(mut self, fn_id: impl Into<String>, source: LuaScriptSource) -> Self {
1940 let source = Arc::new(source);
1941 let bridges = Arc::new(self.bridges.clone());
1942 let wrapped: WorkerFn = Arc::new(move |inv| {
1943 let source = source.clone();
1944 let bridges = bridges.clone();
1945 Box::pin(run_lua_worker(source, bridges, inv))
1946 });
1947 self.registry.insert(fn_id.into(), wrapped);
1948 self
1949 }
1950}
1951
1952async fn run_lua_worker(
1954 source: Arc<LuaScriptSource>,
1955 bridges: Arc<HashMap<String, HostBridge>>,
1956 inv: crate::worker::adapter::WorkerInvocation,
1957) -> Result<crate::worker::adapter::WorkerResult, crate::worker::adapter::WorkerError> {
1958 use crate::worker::adapter::WorkerError;
1959 use mlua::LuaSerdeExt;
1960
1961 let label = source.label.clone();
1962 let outcome =
1963 tokio::task::spawn_blocking(move || -> Result<(serde_json::Value, bool), String> {
1964 let lua = mlua::Lua::new();
1965 let g = lua.globals();
1966
1967 g.set("_PROMPT", inv.prompt.clone())
1969 .map_err(|e| format!("set _PROMPT: {e}"))?;
1970 g.set("_AGENT", inv.agent.clone())
1971 .map_err(|e| format!("set _AGENT: {e}"))?;
1972 g.set("_TASK_ID", inv.task_id.to_string())
1973 .map_err(|e| format!("set _TASK_ID: {e}"))?;
1974 g.set("_ATTEMPT", inv.attempt as i64)
1975 .map_err(|e| format!("set _ATTEMPT: {e}"))?;
1976
1977 for (name, value) in
1986 crate::worker::agent_block::runtime::context_globals(inv.context.as_ref())
1987 {
1988 let lua_val = lua
1989 .to_value(&value)
1990 .map_err(|e| format!("{name} to_value: {e}"))?;
1991 g.set(name.as_str(), lua_val)
1992 .map_err(|e| format!("set {name}: {e}"))?;
1993 }
1994
1995 if let Ok(json_val) = serde_json::from_str::<serde_json::Value>(&inv.prompt) {
1997 let lua_val = lua
1998 .to_value(&json_val)
1999 .map_err(|e| format!("_CTX to_value: {e}"))?;
2000 g.set("_CTX", lua_val)
2001 .map_err(|e| format!("set _CTX: {e}"))?;
2002 }
2003
2004 if !bridges.is_empty() {
2006 let host = lua
2007 .create_table()
2008 .map_err(|e| format!("create host table: {e}"))?;
2009 for (name, bridge) in bridges.iter() {
2010 let bridge = bridge.clone();
2011 let bname = name.clone();
2012 let f = lua
2013 .create_function(move |lua, arg: mlua::Value| {
2014 let json_arg: serde_json::Value = lua.from_value(arg).map_err(|e| {
2015 mlua::Error::external(format!("bridge {bname} arg → json: {e}"))
2016 })?;
2017 let result_json =
2018 bridge.call(json_arg).map_err(mlua::Error::external)?;
2019 lua.to_value(&result_json).map_err(|e| {
2020 mlua::Error::external(format!("bridge {bname} ret → lua: {e}"))
2021 })
2022 })
2023 .map_err(|e| format!("create_function {name}: {e}"))?;
2024 host.set(name.as_str(), f)
2025 .map_err(|e| format!("host.{name} set: {e}"))?;
2026 }
2027 g.set("host", host).map_err(|e| format!("set host: {e}"))?;
2028 }
2029
2030 let result: mlua::Value = lua
2032 .load(&source.source)
2033 .set_name(&source.label)
2034 .eval()
2035 .map_err(|e| format!("lua eval [{}]: {e}", source.label))?;
2036
2037 let json_result: serde_json::Value = lua
2039 .from_value(result)
2040 .map_err(|e| format!("lua → json [{}]: {e}", source.label))?;
2041
2042 let (value, ok) = match &json_result {
2043 serde_json::Value::Object(map)
2044 if map.contains_key("value") || map.contains_key("ok") =>
2045 {
2046 let ok = map.get("ok").and_then(|v| v.as_bool()).unwrap_or(true);
2047 let value = map.get("value").cloned().unwrap_or(json_result.clone());
2048 (value, ok)
2049 }
2050 _ => (json_result, true),
2051 };
2052 Ok((value, ok))
2053 })
2054 .await
2055 .map_err(|e| WorkerError::Failed(format!("spawn_blocking join [{label}]: {e}")))?
2056 .map_err(WorkerError::Failed)?;
2057
2058 Ok(crate::worker::adapter::WorkerResult {
2059 value: outcome.0,
2060 ok: outcome.1,
2061 stats: None,
2062 }
2063 .ensure_worker_kind("lua"))
2064}
2065
2066impl Default for LuaInProcessSpawnerFactory {
2067 fn default() -> Self {
2068 Self::new()
2069 }
2070}
2071
2072impl SpawnerFactoryKind for LuaInProcessSpawnerFactory {
2073 const KIND: AgentKind = AgentKind::Lua;
2074 type Worker = LuaWorker;
2075}
2076
2077impl SpawnerFactory for LuaInProcessSpawnerFactory {
2078 fn build(
2079 &self,
2080 agent_def: &AgentDef,
2081 _hint: Option<&Value>,
2082 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2083 if let Some(source) = agent_def.spec.get("source").and_then(|v| v.as_str()) {
2089 let label = agent_def
2090 .spec
2091 .get("label")
2092 .and_then(|v| v.as_str())
2093 .map(str::to_string)
2094 .unwrap_or_else(|| format!("{}.lua", agent_def.name));
2095 let script = Arc::new(LuaScriptSource::new(source.to_string(), label));
2096 let bridges = Arc::new(self.bridges.clone());
2097 let wrapped: WorkerFn = Arc::new(move |inv| {
2098 let source = script.clone();
2099 let bridges = bridges.clone();
2100 Box::pin(run_lua_worker(source, bridges, inv))
2101 });
2102 let mut sp: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
2103 sp.registry.insert(agent_def.name.to_string(), wrapped);
2104 return Ok(Arc::new(sp));
2105 }
2106 build_inproc_from_registry::<LuaWorker>(&self.registry, agent_def, "lua")
2107 }
2108}
2109
2110pub struct RustFnInProcessSpawnerFactory {
2124 registry: HashMap<String, WorkerFn>,
2125}
2126
2127impl RustFnInProcessSpawnerFactory {
2128 pub fn new() -> Self {
2130 Self {
2131 registry: HashMap::new(),
2132 }
2133 }
2134
2135 pub fn register_fn<F, Fut>(mut self, fn_id: impl Into<String>, f: F) -> Self
2138 where
2139 F: Fn(crate::worker::adapter::WorkerInvocation) -> Fut + Send + Sync + 'static,
2140 Fut: std::future::Future<
2141 Output = Result<
2142 crate::worker::adapter::WorkerResult,
2143 crate::worker::adapter::WorkerError,
2144 >,
2145 > + Send
2146 + 'static,
2147 {
2148 let f = Arc::new(f);
2149 let wrapped: WorkerFn = Arc::new(move |inv| {
2150 let f = f.clone();
2151 Box::pin(f(inv))
2152 });
2153 self.registry.insert(fn_id.into(), wrapped);
2154 self
2155 }
2156}
2157
2158impl Default for RustFnInProcessSpawnerFactory {
2159 fn default() -> Self {
2160 Self::new()
2161 }
2162}
2163
2164impl SpawnerFactoryKind for RustFnInProcessSpawnerFactory {
2165 const KIND: AgentKind = AgentKind::RustFn;
2166 type Worker = RustFnWorker;
2167}
2168
2169impl SpawnerFactory for RustFnInProcessSpawnerFactory {
2170 fn build(
2171 &self,
2172 agent_def: &AgentDef,
2173 _hint: Option<&Value>,
2174 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2175 build_inproc_from_registry::<RustFnWorker>(&self.registry, agent_def, "rust_fn")
2176 }
2177}
2178
2179fn build_inproc_from_registry<W>(
2185 registry: &HashMap<String, WorkerFn>,
2186 agent_def: &AgentDef,
2187 kind_label: &str,
2188) -> Result<Arc<dyn SpawnerAdapter>, CompileError>
2189where
2190 W: crate::worker::Worker + From<crate::worker::WorkerJoinHandler> + Send + Sync + 'static,
2191{
2192 let agent_name = &agent_def.name;
2193 let spec = &agent_def.spec;
2194 let invalid = |msg: String| CompileError::InvalidSpec {
2195 name: agent_name.to_string(),
2196 msg,
2197 };
2198 let fn_id = spec
2199 .get("fn_id")
2200 .and_then(|v| v.as_str())
2201 .ok_or_else(|| invalid(format!("{kind_label} spec: 'fn_id' (string) required")))?;
2202 let f = registry
2203 .get(fn_id)
2204 .cloned()
2205 .ok_or_else(|| invalid(format!("fn_id '{fn_id}' not registered in factory")))?;
2206 let mut sp: InProcSpawner<W> = InProcSpawner::<W>::typed();
2207 sp.registry.insert(agent_name.to_string(), f);
2211 Ok(Arc::new(sp))
2212}
2213
2214pub struct LuaWorker {
2219 pub handler: crate::worker::WorkerJoinHandler,
2221}
2222
2223impl From<crate::worker::WorkerJoinHandler> for LuaWorker {
2224 fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
2225 Self { handler }
2226 }
2227}
2228
2229#[async_trait::async_trait]
2230impl crate::worker::Worker for LuaWorker {
2231 fn id(&self) -> &crate::types::WorkerId {
2232 &self.handler.worker_id
2233 }
2234 fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
2235 self.handler.cancel.clone()
2236 }
2237 async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
2238 self.handler.await_completion().await
2239 }
2240}
2241
2242pub struct RustFnWorker {
2247 pub handler: crate::worker::WorkerJoinHandler,
2249}
2250
2251impl From<crate::worker::WorkerJoinHandler> for RustFnWorker {
2252 fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
2253 Self { handler }
2254 }
2255}
2256
2257#[async_trait::async_trait]
2258impl crate::worker::Worker for RustFnWorker {
2259 fn id(&self) -> &crate::types::WorkerId {
2260 &self.handler.worker_id
2261 }
2262 fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
2263 self.handler.cancel.clone()
2264 }
2265 async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
2266 self.handler.await_completion().await
2267 }
2268}
2269
2270pub struct OperatorSpawnerFactory {
2323 operators: Arc<std::sync::RwLock<HashMap<String, Arc<dyn Operator>>>>,
2324}
2325
2326impl OperatorSpawnerFactory {
2327 pub fn new() -> Self {
2329 Self {
2330 operators: Arc::new(std::sync::RwLock::new(HashMap::new())),
2331 }
2332 }
2333
2334 pub fn register_operator(&self, id: impl Into<String>, op: Arc<dyn Operator>) -> &Self {
2340 self.operators
2341 .write()
2342 .expect("OperatorSpawnerFactory.operators RwLock poisoned")
2343 .insert(id.into(), op);
2344 self
2345 }
2346
2347 pub fn unregister_operator(&self, id: &str) -> &Self {
2350 self.operators
2351 .write()
2352 .expect("OperatorSpawnerFactory.operators RwLock poisoned")
2353 .remove(id);
2354 self
2355 }
2356}
2357
2358impl Default for OperatorSpawnerFactory {
2359 fn default() -> Self {
2360 Self::new()
2361 }
2362}
2363
2364impl SpawnerFactoryKind for OperatorSpawnerFactory {
2365 const KIND: AgentKind = AgentKind::Operator;
2366 type Worker = crate::operator::OperatorWorker;
2367}
2368
2369impl SpawnerFactory for OperatorSpawnerFactory {
2370 fn build(
2371 &self,
2372 agent_def: &AgentDef,
2373 _hint: Option<&Value>,
2374 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2375 let agent_name = &agent_def.name;
2376 let spec = &agent_def.spec;
2377 let system_prompt = agent_def.profile.as_ref().map(|p| p.system_prompt.clone());
2383 let invalid = |msg: String| CompileError::InvalidSpec {
2384 name: agent_name.to_string(),
2385 msg,
2386 };
2387 let op_ref = spec
2388 .get("operator_ref")
2389 .and_then(|v| v.as_str())
2390 .ok_or_else(|| invalid("operator spec: 'operator_ref' (string) required".into()))?;
2391 let operators = self
2392 .operators
2393 .read()
2394 .expect("OperatorSpawnerFactory.operators RwLock poisoned");
2395 let op = operators.get(op_ref).cloned().ok_or_else(|| {
2396 let mut names: Vec<String> = operators.keys().cloned().collect();
2397 names.sort();
2398 let names_list = if names.is_empty() {
2399 "<none>".to_string()
2400 } else {
2401 names.join(", ")
2402 };
2403 invalid(format!(
2404 "operator_ref '{op_ref}' not registered in factory. \
2405 Registered sids: [{names_list}]. \
2406 Hint: call mse_operator_join(roles=[...]) to mint the sid first."
2407 ))
2408 })?;
2409 drop(operators);
2410
2411 let worker_binding = agent_def
2418 .profile
2419 .as_ref()
2420 .and_then(|p| p.worker_binding.as_ref())
2421 .map(|variant| WorkerBinding {
2422 variant: variant.clone(),
2423 tools: agent_def
2424 .profile
2425 .as_ref()
2426 .map(|p| p.tools.clone())
2427 .unwrap_or_default(),
2428 request_digest: None,
2432 requested_model: None,
2433 });
2434 if op.requires_worker_binding() && worker_binding.is_none() {
2435 return Err(invalid(format!(
2442 "{WORKER_BINDING_REQUIRED_MSG_PREFIX}. \
2443 Fix by either: \
2444 (a) if authoring the Blueprint JSON directly, add \
2445 `agents[N].profile.worker_binding: \"<subagent-type>\"` \
2446 to the JSON literal; or \
2447 (b) if using an $agent_md file ref, add \
2448 `worker_binding: <subagent-type>` to the agent .md frontmatter."
2449 )));
2450 }
2451 Ok(Arc::new(OperatorSpawner::new(
2452 op,
2453 system_prompt,
2454 worker_binding,
2455 )))
2456 }
2457}
2458
2459#[cfg(test)]
2460mod operator_spawner_factory_worker_binding_tests {
2461 use super::*;
2462 use crate::blueprint::AgentProfile;
2463 use crate::core::ctx::Ctx;
2464 use crate::types::CapToken;
2465 use crate::worker::adapter::{WorkerError, WorkerResult};
2466
2467 struct StubOperator {
2472 requires_binding: bool,
2473 }
2474
2475 #[async_trait]
2476 impl Operator for StubOperator {
2477 async fn execute(
2478 &self,
2479 _ctx: &Ctx,
2480 _system: Option<String>,
2481 _prompt: Value,
2482 _worker: Option<WorkerBinding>,
2483 _worker_token: CapToken,
2484 ) -> Result<WorkerResult, WorkerError> {
2485 Ok(WorkerResult {
2486 value: Value::Null,
2487 ok: true,
2488 stats: None,
2489 })
2490 }
2491
2492 fn requires_worker_binding(&self) -> bool {
2493 self.requires_binding
2494 }
2495 }
2496
2497 fn agent_def_with(profile: Option<AgentProfile>) -> AgentDef {
2498 AgentDef {
2499 name: "test-agent".to_string(),
2500 kind: AgentKind::Operator,
2501 spec: serde_json::json!({ "operator_ref": "op1" }),
2502 profile,
2503 meta: None,
2504 runner: None,
2505 runner_ref: None,
2506 verdict: None,
2507 }
2508 }
2509
2510 #[test]
2511 fn build_fails_loud_when_binding_required_but_absent() {
2512 let factory = OperatorSpawnerFactory::new();
2513 factory.register_operator(
2514 "op1",
2515 Arc::new(StubOperator {
2516 requires_binding: true,
2517 }) as Arc<dyn Operator>,
2518 );
2519 let def = agent_def_with(Some(AgentProfile::default()));
2520 match factory.build(&def, None) {
2521 Err(CompileError::InvalidSpec { name, msg }) => {
2522 assert_eq!(name, "test-agent");
2523 assert!(
2524 msg.contains("worker_binding is required"),
2525 "unexpected message: {msg}"
2526 );
2527 assert!(
2531 msg.contains("agents[N].profile.worker_binding"),
2532 "message missing JSON-direct hint (issue #9): {msg}"
2533 );
2534 assert!(
2535 msg.contains("agent .md frontmatter"),
2536 "message missing $agent_md hint: {msg}"
2537 );
2538 }
2539 Err(other) => panic!("expected InvalidSpec, got: {other:?}"),
2540 Ok(_) => panic!("expected compile-time failure, got Ok"),
2541 }
2542 }
2543
2544 #[test]
2551 fn factory_error_message_carries_the_shared_prefix_and_specializes_the_diagnostic() {
2552 let factory = OperatorSpawnerFactory::new();
2553 factory.register_operator(
2554 "op1",
2555 Arc::new(StubOperator {
2556 requires_binding: true,
2557 }) as Arc<dyn Operator>,
2558 );
2559 let def = agent_def_with(Some(AgentProfile::default()));
2560 let err = match factory.build(&def, None) {
2561 Err(err) => err,
2562 Ok(_) => panic!("expected compile-time failure, got Ok"),
2563 };
2564 match &err {
2565 CompileError::InvalidSpec { msg, .. } => {
2566 assert!(
2567 msg.starts_with(WORKER_BINDING_REQUIRED_MSG_PREFIX),
2568 "factory message must start with the shared prefix, got: {msg}"
2569 );
2570 }
2571 other => panic!("expected InvalidSpec, got: {other:?}"),
2572 }
2573 let d = mlua_swarm_diag::Diagnostic::from(&err);
2574 assert_eq!(d.kind, "worker-binding-missing");
2575 }
2576
2577 #[test]
2578 fn build_succeeds_when_binding_required_and_present() {
2579 let factory = OperatorSpawnerFactory::new();
2580 factory.register_operator(
2581 "op1",
2582 Arc::new(StubOperator {
2583 requires_binding: true,
2584 }) as Arc<dyn Operator>,
2585 );
2586 let profile = AgentProfile {
2587 worker_binding: Some("code-worker".to_string()),
2588 tools: vec!["Read".to_string(), "Edit".to_string()],
2589 ..Default::default()
2590 };
2591 let def = agent_def_with(Some(profile));
2592 assert!(
2593 factory.build(&def, None).is_ok(),
2594 "expected Ok when worker_binding is declared"
2595 );
2596 }
2597
2598 #[test]
2599 fn build_succeeds_when_binding_not_required_and_absent() {
2600 let factory = OperatorSpawnerFactory::new();
2601 factory.register_operator(
2602 "op1",
2603 Arc::new(StubOperator {
2604 requires_binding: false,
2605 }) as Arc<dyn Operator>,
2606 );
2607 let def = agent_def_with(Some(AgentProfile::default()));
2608 assert!(
2609 factory.build(&def, None).is_ok(),
2610 "backends that don't require a binding must not be gated by its absence"
2611 );
2612 }
2613}
2614
2615#[cfg(test)]
2623mod lua_inline_source_tests {
2624 use super::*;
2625 use crate::types::{CapToken, Role, StepId};
2626
2627 fn agent(name: &str, spec: Value) -> AgentDef {
2628 AgentDef {
2629 name: name.to_string(),
2630 kind: AgentKind::Lua,
2631 spec,
2632 profile: None,
2633 meta: None,
2634 runner: None,
2635 runner_ref: None,
2636 verdict: None,
2637 }
2638 }
2639
2640 fn test_invocation(prompt: &str) -> crate::worker::adapter::WorkerInvocation {
2641 crate::worker::adapter::WorkerInvocation::new(
2642 CapToken {
2643 agent_id: "a".into(),
2644 role: Role::Worker,
2645 scopes: vec!["*".into()],
2646 issued_at: 0,
2647 expire_at: u64::MAX / 2,
2648 max_uses: None,
2649 nonce: "test-nonce".into(),
2650 sig_hex: "".into(),
2651 },
2652 StepId::parse("ST-test").expect("StepId parse"),
2653 1,
2654 "g",
2655 prompt,
2656 )
2657 }
2658
2659 #[test]
2660 fn build_accepts_inline_source_without_pre_registration() {
2661 let factory = LuaInProcessSpawnerFactory::new();
2662 let def = agent(
2663 "g",
2664 serde_json::json!({ "source": "return { value = 42, ok = true }" }),
2665 );
2666 assert!(
2667 factory.build(&def, None).is_ok(),
2668 "inline spec.source must build without a pre-registered fn_id"
2669 );
2670 }
2671
2672 #[test]
2673 fn build_rejects_when_neither_source_nor_fn_id_is_present() {
2674 let factory = LuaInProcessSpawnerFactory::new();
2675 let def = agent("g", serde_json::json!({}));
2676 match factory.build(&def, None) {
2677 Err(CompileError::InvalidSpec { msg, .. }) => {
2678 assert!(
2679 msg.contains("fn_id"),
2680 "empty spec must still surface the fn_id-required message: {msg}"
2681 );
2682 }
2683 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
2684 Ok(_) => panic!("expected InvalidSpec, got Ok(SpawnerAdapter)"),
2687 }
2688 }
2689
2690 #[tokio::test]
2694 async fn inline_source_evaluates_and_marshals_result() {
2695 let source =
2696 LuaScriptSource::new("return { value = _PROMPT .. '!', ok = true }", "smoke.lua");
2697 let out = run_lua_worker(
2698 std::sync::Arc::new(source),
2699 std::sync::Arc::new(HashMap::new()),
2700 test_invocation("hello"),
2701 )
2702 .await
2703 .expect("lua worker ok");
2704 assert_eq!(out.value, serde_json::json!("hello!"));
2705 assert!(out.ok);
2706 }
2707
2708 #[tokio::test]
2709 async fn inline_source_can_signal_agent_level_failure() {
2710 let source = LuaScriptSource::new("return { value = 'nope', ok = false }", "gate.lua");
2713 let out = run_lua_worker(
2714 std::sync::Arc::new(source),
2715 std::sync::Arc::new(HashMap::new()),
2716 test_invocation("input"),
2717 )
2718 .await
2719 .expect("lua worker ok");
2720 assert_eq!(out.value, serde_json::json!("nope"));
2721 assert!(!out.ok);
2722 }
2723}
2724
2725#[cfg(test)]
2728mod meta_ref_validation_tests {
2729 use super::*;
2730 use crate::blueprint::{AgentMeta, MetaDef};
2731 use crate::worker::adapter::WorkerResult;
2732
2733 fn registry_with_echo() -> SpawnerRegistry {
2734 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2735 Ok(WorkerResult {
2736 value: Value::String(inv.prompt),
2737 ok: true,
2738 stats: None,
2739 })
2740 });
2741 let mut reg = SpawnerRegistry::new();
2742 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2743 reg
2744 }
2745
2746 fn rustfn_agent(name: &str) -> AgentDef {
2747 AgentDef {
2748 name: name.to_string(),
2749 kind: AgentKind::RustFn,
2750 spec: serde_json::json!({ "fn_id": "echo" }),
2751 profile: None,
2752 meta: None,
2753 runner: None,
2754 runner_ref: None,
2755 verdict: None,
2756 }
2757 }
2758
2759 fn simple_flow(agent_ref: &str, in_: Expr) -> FlowNode {
2760 FlowNode::Step {
2761 ref_: agent_ref.to_string(),
2762 in_,
2763 out: Expr::Path {
2764 at: "$.output".parse().expect("literal test path: $.output"),
2765 },
2766 }
2767 }
2768
2769 fn minimal_bp(agents: Vec<AgentDef>, metas: Vec<MetaDef>, flow: FlowNode) -> Blueprint {
2770 Blueprint {
2771 schema_version: crate::blueprint::current_schema_version(),
2772 id: "meta-ref-ut".into(),
2773 flow,
2774 agents,
2775 operators: vec![],
2776 metas,
2777 hints: Default::default(),
2778 strategy: Default::default(),
2779 metadata: BlueprintMetadata::default(),
2780 spawner_hints: Default::default(),
2781 default_agent_kind: AgentKind::Operator,
2782 default_operator_kind: None,
2783 default_init_ctx: None,
2784 default_agent_ctx: None,
2785 default_context_policy: None,
2786 projection_placement: None,
2787 audits: vec![],
2788 degradation_policy: None,
2789 runners: vec![],
2790 default_runner: None,
2791 subprocesses: vec![],
2792 check_policy: None,
2793 blueprint_ref_includes: Vec::new(),
2794 }
2795 }
2796
2797 #[test]
2798 fn valid_meta_ref_compiles() {
2799 let mut agent = rustfn_agent("worker");
2800 agent.meta = Some(AgentMeta {
2801 meta_ref: Some("shared".to_string()),
2802 ..Default::default()
2803 });
2804 let bp = minimal_bp(
2805 vec![agent],
2806 vec![MetaDef {
2807 name: "shared".into(),
2808 ctx: serde_json::json!({ "k": "v" }),
2809 }],
2810 simple_flow(
2811 "worker",
2812 Expr::Path {
2813 at: "$.input".parse().expect("literal test path: $.input"),
2814 },
2815 ),
2816 );
2817 let compiler = Compiler::new(registry_with_echo());
2818 assert!(
2819 compiler.compile(&bp).is_ok(),
2820 "a resolvable AgentMeta.meta_ref must compile"
2821 );
2822 }
2823
2824 #[test]
2825 fn unknown_agent_meta_ref_is_unresolved_meta_ref() {
2826 let mut agent = rustfn_agent("worker");
2827 agent.meta = Some(AgentMeta {
2828 meta_ref: Some("missing".to_string()),
2829 ..Default::default()
2830 });
2831 let bp = minimal_bp(
2832 vec![agent],
2833 vec![],
2834 simple_flow(
2835 "worker",
2836 Expr::Path {
2837 at: "$.input".parse().expect("literal test path: $.input"),
2838 },
2839 ),
2840 );
2841 let compiler = Compiler::new(registry_with_echo());
2842 match compiler.compile(&bp) {
2843 Err(CompileError::UnresolvedMetaRef {
2844 where_,
2845 meta_ref,
2846 defined,
2847 }) => {
2848 assert!(
2849 where_.contains("worker"),
2850 "where_ must name the agent: {where_}"
2851 );
2852 assert_eq!(meta_ref, "missing");
2853 assert!(defined.is_empty());
2854 }
2855 Err(other) => {
2856 panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
2857 }
2858 Ok(_) => panic!("expected compile-time failure, got Ok"),
2859 }
2860 }
2861
2862 #[test]
2863 fn unknown_static_step_meta_ref_in_lit_is_unresolved_meta_ref() {
2864 let agent = rustfn_agent("worker");
2865 let in_ = Expr::Lit {
2866 value: serde_json::json!({ "$step_meta": { "ref": "missing" }, "$in": "go" }),
2867 };
2868 let bp = minimal_bp(vec![agent], vec![], simple_flow("worker", in_));
2869 let compiler = Compiler::new(registry_with_echo());
2870 match compiler.compile(&bp) {
2871 Err(CompileError::UnresolvedMetaRef {
2872 where_, meta_ref, ..
2873 }) => {
2874 assert!(
2875 where_.contains("worker"),
2876 "where_ must name the offending step: {where_}"
2877 );
2878 assert_eq!(meta_ref, "missing");
2879 }
2880 Err(other) => {
2881 panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
2882 }
2883 Ok(_) => panic!("expected compile-time failure, got Ok"),
2884 }
2885 }
2886
2887 #[test]
2888 fn path_op_input_with_no_static_envelope_compiles_fine() {
2889 let agent = rustfn_agent("worker");
2890 let bp = minimal_bp(
2891 vec![agent],
2892 vec![],
2893 simple_flow(
2894 "worker",
2895 Expr::Path {
2896 at: "$.input".parse().expect("literal test path: $.input"),
2897 },
2898 ),
2899 );
2900 let compiler = Compiler::new(registry_with_echo());
2901 assert!(
2902 compiler.compile(&bp).is_ok(),
2903 "a non-Lit Step.in must not trigger the best-effort static $step_meta check"
2904 );
2905 }
2906}
2907
2908#[cfg(test)]
2910mod audit_agent_validation_tests {
2911 use super::*;
2912 use crate::worker::adapter::WorkerResult;
2913 use mlua_swarm_schema::{AuditDef, AuditMode};
2914
2915 fn registry_with_echo() -> SpawnerRegistry {
2916 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2917 Ok(WorkerResult {
2918 value: Value::String(inv.prompt),
2919 ok: true,
2920 stats: None,
2921 })
2922 });
2923 let mut reg = SpawnerRegistry::new();
2924 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2925 reg
2926 }
2927
2928 fn rustfn_agent(name: &str) -> AgentDef {
2929 AgentDef {
2930 name: name.to_string(),
2931 kind: AgentKind::RustFn,
2932 spec: serde_json::json!({ "fn_id": "echo" }),
2933 profile: None,
2934 meta: None,
2935 runner: None,
2936 runner_ref: None,
2937 verdict: None,
2938 }
2939 }
2940
2941 fn minimal_bp(agents: Vec<AgentDef>, audits: Vec<AuditDef>) -> Blueprint {
2942 Blueprint {
2943 schema_version: crate::blueprint::current_schema_version(),
2944 id: "audit-ref-ut".into(),
2945 flow: FlowNode::Step {
2946 ref_: "worker".to_string(),
2947 in_: Expr::Path {
2948 at: "$.input".parse().expect("literal test path: $.input"),
2949 },
2950 out: Expr::Path {
2951 at: "$.output".parse().expect("literal test path: $.output"),
2952 },
2953 },
2954 agents,
2955 operators: vec![],
2956 metas: vec![],
2957 hints: Default::default(),
2958 strategy: Default::default(),
2959 metadata: BlueprintMetadata::default(),
2960 spawner_hints: Default::default(),
2961 default_agent_kind: AgentKind::Operator,
2962 default_operator_kind: None,
2963 default_init_ctx: None,
2964 default_agent_ctx: None,
2965 default_context_policy: None,
2966 projection_placement: None,
2967 audits,
2968 degradation_policy: None,
2969 runners: vec![],
2970 default_runner: None,
2971 subprocesses: vec![],
2972 check_policy: None,
2973 blueprint_ref_includes: Vec::new(),
2974 }
2975 }
2976
2977 #[test]
2978 fn unresolved_audit_agent_is_a_loud_compile_error() {
2979 let bp = minimal_bp(
2980 vec![rustfn_agent("worker")],
2981 vec![AuditDef {
2982 agent: "missing-auditor".to_string(),
2983 steps: None,
2984 mode: AuditMode::default(),
2985 }],
2986 );
2987 let compiler = Compiler::new(registry_with_echo());
2988 match compiler.compile(&bp) {
2989 Err(CompileError::UnresolvedAuditAgent { agent, defined }) => {
2990 assert_eq!(agent, "missing-auditor");
2991 assert_eq!(defined, vec!["worker".to_string()]);
2992 }
2993 Err(other) => {
2994 panic!("expected UnresolvedAuditAgent, got a different CompileError: {other}")
2995 }
2996 Ok(_) => panic!("expected compile-time failure, got Ok"),
2997 }
2998 }
2999
3000 #[test]
3001 fn resolved_audit_agent_compiles_fine() {
3002 let bp = minimal_bp(
3003 vec![rustfn_agent("worker"), rustfn_agent("auditor")],
3004 vec![AuditDef {
3005 agent: "auditor".to_string(),
3006 steps: None,
3007 mode: AuditMode::default(),
3008 }],
3009 );
3010 let compiler = Compiler::new(registry_with_echo());
3011 assert!(
3012 compiler.compile(&bp).is_ok(),
3013 "an audits[].agent that names a declared AgentDef must compile"
3014 );
3015 }
3016}
3017
3018#[cfg(test)]
3021mod projection_placement_compile_tests {
3022 use super::*;
3023 use crate::core::projection_placement::{ProjectionPlacement, RootPreference};
3024 use crate::worker::adapter::WorkerResult;
3025 use mlua_swarm_schema::ProjectionPlacementSpec;
3026
3027 fn registry_with_echo() -> SpawnerRegistry {
3028 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
3029 Ok(WorkerResult {
3030 value: Value::String(inv.prompt),
3031 ok: true,
3032 stats: None,
3033 })
3034 });
3035 let mut reg = SpawnerRegistry::new();
3036 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
3037 reg
3038 }
3039
3040 fn minimal_bp(projection_placement: Option<ProjectionPlacementSpec>) -> Blueprint {
3041 Blueprint {
3042 schema_version: crate::blueprint::current_schema_version(),
3043 id: "projection-placement-ut".into(),
3044 flow: FlowNode::Step {
3045 ref_: "worker".to_string(),
3046 in_: Expr::Path {
3047 at: "$.input".parse().expect("literal test path: $.input"),
3048 },
3049 out: Expr::Path {
3050 at: "$.output".parse().expect("literal test path: $.output"),
3051 },
3052 },
3053 agents: vec![AgentDef {
3054 name: "worker".to_string(),
3055 kind: AgentKind::RustFn,
3056 spec: serde_json::json!({ "fn_id": "echo" }),
3057 profile: None,
3058 meta: None,
3059 runner: None,
3060 runner_ref: None,
3061 verdict: None,
3062 }],
3063 operators: vec![],
3064 metas: vec![],
3065 hints: Default::default(),
3066 strategy: Default::default(),
3067 metadata: BlueprintMetadata::default(),
3068 spawner_hints: Default::default(),
3069 default_agent_kind: AgentKind::Operator,
3070 default_operator_kind: None,
3071 default_init_ctx: None,
3072 default_agent_ctx: None,
3073 default_context_policy: None,
3074 projection_placement,
3075 audits: vec![],
3076 degradation_policy: None,
3077 runners: vec![],
3078 default_runner: None,
3079 subprocesses: vec![],
3080 check_policy: None,
3081 blueprint_ref_includes: Vec::new(),
3082 }
3083 }
3084
3085 #[test]
3086 fn undeclared_projection_placement_compiles_to_byte_compat_default() {
3087 let bp = minimal_bp(None);
3088 let compiled = Compiler::new(registry_with_echo())
3089 .compile(&bp)
3090 .expect("undeclared projection_placement compiles");
3091 assert_eq!(
3092 *compiled.projection_placement,
3093 ProjectionPlacement::default()
3094 );
3095 }
3096
3097 #[test]
3098 fn declared_valid_projection_placement_compiles_to_matching_resolver() {
3099 let bp = minimal_bp(Some(ProjectionPlacementSpec {
3100 root: Some("project_root".to_string()),
3101 dir_template: Some("custom/{task_id}/out".to_string()),
3102 }));
3103 let compiled = Compiler::new(registry_with_echo())
3104 .compile(&bp)
3105 .expect("valid projection_placement compiles");
3106 assert_eq!(
3107 compiled.projection_placement.root_preference,
3108 RootPreference::ProjectRoot
3109 );
3110 assert_eq!(
3111 compiled.projection_placement.dir_template,
3112 "custom/{task_id}/out"
3113 );
3114 }
3115
3116 #[test]
3117 fn declared_invalid_dir_template_rejects_compile() {
3118 let bp = minimal_bp(Some(ProjectionPlacementSpec {
3119 root: None,
3120 dir_template: Some("workspace/tasks/ctx".to_string()), }));
3122 match Compiler::new(registry_with_echo()).compile(&bp) {
3123 Err(CompileError::InvalidProjectionPlacement(_)) => {}
3124 Err(other) => {
3125 panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
3126 }
3127 Ok(_) => {
3128 panic!("expected compile-time rejection for a missing {{task_id}} placeholder")
3129 }
3130 }
3131 }
3132
3133 #[test]
3134 fn declared_invalid_root_literal_rejects_compile() {
3135 let bp = minimal_bp(Some(ProjectionPlacementSpec {
3136 root: Some("nope".to_string()),
3137 dir_template: None,
3138 }));
3139 match Compiler::new(registry_with_echo()).compile(&bp) {
3140 Err(CompileError::InvalidProjectionPlacement(_)) => {}
3141 Err(other) => {
3142 panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
3143 }
3144 Ok(_) => panic!("expected compile-time rejection for an invalid root literal"),
3145 }
3146 }
3147}
3148
3149#[cfg(test)]
3151mod verdict_contract_lint_tests {
3152 use super::*;
3153 use crate::worker::adapter::WorkerResult;
3154
3155 fn registry_with_echo() -> SpawnerRegistry {
3156 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
3157 Ok(WorkerResult {
3158 value: Value::String(inv.prompt),
3159 ok: true,
3160 stats: None,
3161 })
3162 });
3163 let mut reg = SpawnerRegistry::new();
3164 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
3165 reg
3166 }
3167
3168 fn gate_agent(verdict: Option<VerdictContract>) -> AgentDef {
3169 AgentDef {
3170 name: "gate".to_string(),
3171 kind: AgentKind::RustFn,
3172 spec: serde_json::json!({ "fn_id": "echo" }),
3173 profile: None,
3174 meta: None,
3175 runner: None,
3176 runner_ref: None,
3177 verdict,
3178 }
3179 }
3180
3181 fn minimal_bp(agent: AgentDef, flow: FlowNode) -> Blueprint {
3182 Blueprint {
3183 schema_version: crate::blueprint::current_schema_version(),
3184 id: "verdict-contract-ut".into(),
3185 flow,
3186 agents: vec![agent],
3187 operators: vec![],
3188 metas: vec![],
3189 hints: Default::default(),
3190 strategy: Default::default(),
3191 metadata: BlueprintMetadata::default(),
3192 spawner_hints: Default::default(),
3193 default_agent_kind: AgentKind::Operator,
3194 default_operator_kind: None,
3195 default_init_ctx: None,
3196 default_agent_ctx: None,
3197 default_context_policy: None,
3198 projection_placement: None,
3199 audits: vec![],
3200 degradation_policy: None,
3201 runners: vec![],
3202 default_runner: None,
3203 subprocesses: vec![],
3204 check_policy: None,
3205 blueprint_ref_includes: Vec::new(),
3206 }
3207 }
3208
3209 fn step(ref_: &str, out_path: &str) -> FlowNode {
3210 FlowNode::Step {
3211 ref_: ref_.to_string(),
3212 in_: Expr::Lit { value: Value::Null },
3213 out: Expr::Path {
3214 at: out_path.parse().expect("literal test path"),
3215 },
3216 }
3217 }
3218
3219 fn noop() -> FlowNode {
3220 FlowNode::Seq { children: vec![] }
3221 }
3222
3223 fn eq_cond(path: &str, lit: &str) -> Expr {
3224 Expr::Eq {
3225 lhs: Box::new(Expr::Path {
3226 at: path.parse().expect("literal test path"),
3227 }),
3228 rhs: Box::new(Expr::Lit {
3229 value: Value::String(lit.to_string()),
3230 }),
3231 }
3232 }
3233
3234 fn branch(cond: Expr, then_: FlowNode, else_: FlowNode) -> FlowNode {
3235 FlowNode::Branch {
3236 cond,
3237 then_: Box::new(then_),
3238 else_: Box::new(else_),
3239 }
3240 }
3241
3242 fn body_contract(values: &[&str]) -> VerdictContract {
3243 VerdictContract {
3244 channel: VerdictChannel::Body,
3245 values: values.iter().map(|v| v.to_string()).collect(),
3246 }
3247 }
3248
3249 fn part_contract(values: &[&str]) -> VerdictContract {
3250 VerdictContract {
3251 channel: VerdictChannel::Part,
3252 values: values.iter().map(|v| v.to_string()).collect(),
3253 }
3254 }
3255
3256 #[test]
3257 fn contract_with_correct_body_channel_and_value_compiles() {
3258 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3259 let flow = FlowNode::Seq {
3260 children: vec![
3261 step("gate", "$.verdict"),
3262 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3263 ],
3264 };
3265 let bp = minimal_bp(agent, flow);
3266 assert!(
3267 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3268 "a cond addressing the bare step output must match a channel: \"body\" contract"
3269 );
3270 }
3271
3272 #[test]
3273 fn contract_with_correct_part_channel_and_value_compiles() {
3274 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3275 let flow = FlowNode::Seq {
3276 children: vec![
3277 step("gate", "$.gate"),
3278 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3279 ],
3280 };
3281 let bp = minimal_bp(agent, flow);
3282 assert!(
3283 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3284 "a cond addressing '<step>.parts.verdict' must match a channel: \"part\" contract"
3285 );
3286 }
3287
3288 #[test]
3289 fn body_channel_contract_rejects_cond_addressing_parts_verdict() {
3290 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3294 let flow = FlowNode::Seq {
3295 children: vec![
3296 step("gate", "$.gate"),
3297 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3298 ],
3299 };
3300 let bp = minimal_bp(agent, flow);
3301 match Compiler::new(registry_with_echo()).compile(&bp) {
3302 Err(CompileError::VerdictChannelMismatch {
3303 where_,
3304 agent,
3305 expected_channel,
3306 actual_shape,
3307 }) => {
3308 assert_eq!(agent, "gate");
3309 assert_eq!(expected_channel, "body");
3310 assert_eq!(actual_shape, "part");
3311 assert!(where_.contains("Branch cond"), "where_: {where_}");
3312 }
3313 Err(other) => {
3314 panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
3315 }
3316 Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
3317 }
3318 }
3319
3320 #[test]
3321 fn part_channel_contract_rejects_cond_addressing_bare_output() {
3322 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3325 let flow = FlowNode::Seq {
3326 children: vec![
3327 step("gate", "$.verdict"),
3328 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3329 ],
3330 };
3331 let bp = minimal_bp(agent, flow);
3332 match Compiler::new(registry_with_echo()).compile(&bp) {
3333 Err(CompileError::VerdictChannelMismatch {
3334 agent,
3335 expected_channel,
3336 actual_shape,
3337 ..
3338 }) => {
3339 assert_eq!(agent, "gate");
3340 assert_eq!(expected_channel, "part");
3341 assert_eq!(actual_shape, "body");
3342 }
3343 Err(other) => {
3344 panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
3345 }
3346 Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
3347 }
3348 }
3349
3350 #[test]
3351 fn contract_rejects_lit_outside_declared_values() {
3352 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3353 let flow = FlowNode::Seq {
3354 children: vec![
3355 step("gate", "$.verdict"),
3356 branch(eq_cond("$.verdict", "UNKNOWN"), noop(), noop()),
3357 ],
3358 };
3359 let bp = minimal_bp(agent, flow);
3360 match Compiler::new(registry_with_echo()).compile(&bp) {
3361 Err(CompileError::VerdictValueNotInContract {
3362 agent,
3363 value,
3364 values,
3365 ..
3366 }) => {
3367 assert_eq!(agent, "gate");
3368 assert_eq!(value, "UNKNOWN");
3369 assert_eq!(values, vec!["PASS".to_string(), "BLOCKED".to_string()]);
3370 }
3371 Err(other) => {
3372 panic!("expected VerdictValueNotInContract, got a different CompileError: {other}")
3373 }
3374 Ok(_) => panic!("expected compile-time rejection for a Lit outside declared values"),
3375 }
3376 }
3377
3378 #[test]
3379 fn undeclared_agent_referenced_by_cond_compiles_with_warning_only() {
3380 let agent = gate_agent(None);
3381 let flow = FlowNode::Seq {
3382 children: vec![
3383 step("gate", "$.verdict"),
3384 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3385 ],
3386 };
3387 let bp = minimal_bp(agent, flow);
3388 assert!(
3389 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3390 "an undeclared verdict contract must never reject compile (opt-in, back-compat)"
3391 );
3392 }
3393
3394 #[test]
3395 fn in_expr_with_lit_haystack_members_compiles() {
3396 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3397 let cond = Expr::In {
3398 needle: Box::new(Expr::Path {
3399 at: "$.verdict".parse().expect("literal test path"),
3400 }),
3401 haystack: Box::new(Expr::Lit {
3402 value: serde_json::json!(["PASS", "BLOCKED"]),
3403 }),
3404 };
3405 let flow = FlowNode::Seq {
3406 children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
3407 };
3408 let bp = minimal_bp(agent, flow);
3409 assert!(
3410 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3411 "an `In` haystack whose every Lit is a declared value must compile"
3412 );
3413 }
3414
3415 #[test]
3422 fn strict_mode_rejects_unhandled_declared_value() {
3423 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3424 let flow = FlowNode::Seq {
3425 children: vec![
3426 step("gate", "$.verdict"),
3427 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3428 ],
3429 };
3430 let mut bp = minimal_bp(agent, flow);
3431 bp.metadata.strict_verdict_handling = Some(true);
3432 match Compiler::new(registry_with_echo()).compile(&bp) {
3433 Err(CompileError::VerdictValueUnhandled {
3434 agent,
3435 value,
3436 declared_values,
3437 step_ref,
3438 }) => {
3439 assert_eq!(agent, "gate");
3440 assert_eq!(value, "PASS");
3441 assert_eq!(
3442 declared_values,
3443 vec!["PASS".to_string(), "BLOCKED".to_string()]
3444 );
3445 assert_eq!(step_ref, "gate");
3446 }
3447 Err(other) => {
3448 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
3449 }
3450 Ok(_) => panic!(
3451 "expected compile-time rejection for a declared verdict value with no \
3452 downstream handler under strict_verdict_handling=Some(true)"
3453 ),
3454 }
3455 }
3456
3457 #[test]
3464 fn default_mode_permits_unhandled_declared_value() {
3465 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3466 let flow = FlowNode::Seq {
3467 children: vec![
3468 step("gate", "$.verdict"),
3469 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3470 ],
3471 };
3472 let bp = minimal_bp(agent, flow);
3473 assert!(
3475 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3476 "default mode must never reject a Blueprint for unhandled declared values \
3477 (opt-in, back-compat with GH #50)"
3478 );
3479 }
3480
3481 #[test]
3486 fn strict_mode_accepts_all_declared_values_handled() {
3487 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3488 let flow = FlowNode::Seq {
3491 children: vec![
3492 step("gate", "$.verdict"),
3493 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3494 branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
3495 ],
3496 };
3497 let mut bp = minimal_bp(agent, flow);
3498 bp.metadata.strict_verdict_handling = Some(true);
3499 assert!(
3500 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3501 "strict mode must accept a Blueprint that handles every declared value"
3502 );
3503 }
3504
3505 #[test]
3509 fn strict_mode_accepts_declared_values_covered_by_in_expr() {
3510 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3511 let cond = Expr::In {
3512 needle: Box::new(Expr::Path {
3513 at: "$.verdict".parse().expect("literal test path"),
3514 }),
3515 haystack: Box::new(Expr::Lit {
3516 value: serde_json::json!(["PASS", "BLOCKED"]),
3517 }),
3518 };
3519 let flow = FlowNode::Seq {
3520 children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
3521 };
3522 let mut bp = minimal_bp(agent, flow);
3523 bp.metadata.strict_verdict_handling = Some(true);
3524 assert!(
3525 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3526 "strict mode must accept an `In` haystack that covers every declared value"
3527 );
3528 }
3529
3530 #[test]
3534 fn strict_mode_rejects_unhandled_part_channel_value() {
3535 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3536 let flow = FlowNode::Seq {
3537 children: vec![
3538 step("gate", "$.gate"),
3539 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3540 ],
3541 };
3542 let mut bp = minimal_bp(agent, flow);
3543 bp.metadata.strict_verdict_handling = Some(true);
3544 match Compiler::new(registry_with_echo()).compile(&bp) {
3545 Err(CompileError::VerdictValueUnhandled {
3546 agent,
3547 value,
3548 step_ref,
3549 ..
3550 }) => {
3551 assert_eq!(agent, "gate");
3552 assert_eq!(value, "PASS");
3553 assert_eq!(step_ref, "gate");
3554 }
3555 Err(other) => {
3556 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
3557 }
3558 Ok(_) => panic!(
3559 "expected compile-time rejection for a declared verdict value with no \
3560 downstream handler (part channel) under strict_verdict_handling=Some(true)"
3561 ),
3562 }
3563 }
3564
3565 #[test]
3572 fn verdict_omitted_blueprint_compiles_unchanged_with_empty_contracts() {
3573 let agent = gate_agent(None);
3574 let flow = FlowNode::Seq {
3575 children: vec![
3576 step("gate", "$.verdict"),
3577 FlowNode::Loop {
3578 counter: Expr::Path {
3579 at: "$.n".parse().expect("literal test path"),
3580 },
3581 cond: eq_cond("$.verdict", "BLOCKED"),
3582 body: Box::new(step("gate", "$.verdict")),
3583 max: 3,
3584 },
3585 branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
3586 ],
3587 };
3588 let bp = minimal_bp(agent, flow);
3589 let compiled = Compiler::new(registry_with_echo())
3590 .compile(&bp)
3591 .expect("a verdict-omitted Blueprint must compile unchanged");
3592 assert!(
3593 compiled.router.verdict_contracts.is_empty(),
3594 "no agent declared a verdict contract"
3595 );
3596 }
3597
3598 #[test]
3605 fn every_compile_error_diagnostic_kind_is_a_declared_lint() {
3606 let kinds = [
3607 "bound-agent-resolution",
3608 "unknown-agent-kind",
3609 "invalid-agent-spec",
3610 "worker-binding-missing",
3611 "unresolved-agent-ref",
3612 "duplicate-agent-name",
3613 "unresolved-operator-ref",
3614 "unresolved-meta-ref",
3615 "step-naming-collision",
3616 "invalid-projection-placement",
3617 "unresolved-audit-agent",
3618 "verdict-channel-mismatch",
3619 "verdict-value-not-in-contract",
3620 "verdict-value-unhandled",
3621 ];
3622 for kind in kinds {
3623 assert!(
3624 mlua_swarm_diag::lint_decl(kind).is_some(),
3625 "kind '{kind}' emitted by From<&CompileError> has no LINT_DECLS entry"
3626 );
3627 }
3628 }
3629
3630 #[test]
3631 fn invalid_spec_with_worker_binding_prefix_specializes_the_diagnostic_kind() {
3632 let err = CompileError::InvalidSpec {
3636 name: "greeter".into(),
3637 msg: format!("{WORKER_BINDING_REQUIRED_MSG_PREFIX}. Fix by either: (a) ..."),
3638 };
3639 let d = mlua_swarm_diag::Diagnostic::from(&err);
3640 assert_eq!(d.kind, "worker-binding-missing");
3641 assert_eq!(d.level, mlua_swarm_diag::DiagLevel::Error);
3642 assert!(matches!(d.stage, mlua_swarm_diag::DiagStage::CompileLint));
3643 assert!(d.message.contains("greeter"));
3644 let suggestion = d
3645 .suggestion
3646 .expect("specialized arm must carry a suggestion");
3647 assert!(suggestion.patch.contains("backend = \"ws_operator\""));
3648 assert_eq!(
3649 suggestion.applicability,
3650 mlua_swarm_diag::Applicability::HasPlaceholders
3651 );
3652 assert_eq!(
3653 d.docs_ref.expect("docs_ref must be set").uri,
3654 "mse://guides/bp-dsl-templates"
3655 );
3656 match d.span.expect("span must be set").element {
3657 mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "greeter"),
3658 other => panic!("expected Agent span, got {other:?}"),
3659 }
3660 }
3661
3662 #[test]
3663 fn generic_invalid_spec_maps_to_the_generic_kind() {
3664 let err = CompileError::InvalidSpec {
3665 name: "solo".into(),
3666 msg: "operator spec: 'operator_ref' (string) required".into(),
3667 };
3668 let d = mlua_swarm_diag::Diagnostic::from(&err);
3669 assert_eq!(d.kind, "invalid-agent-spec");
3670 assert!(
3671 d.suggestion.is_none(),
3672 "generic arm carries no canned patch"
3673 );
3674 }
3675
3676 #[test]
3677 fn verdict_value_not_in_contract_diagnostic_carries_suggestion_and_span() {
3678 let err = CompileError::VerdictValueNotInContract {
3679 where_: "Branch cond".into(),
3680 agent: "review".into(),
3681 value: "NOT_DECLARED".into(),
3682 values: vec!["PASS".into(), "BLOCKED".into()],
3683 };
3684 let d = mlua_swarm_diag::Diagnostic::from(&err);
3685 assert_eq!(d.kind, "verdict-value-not-in-contract");
3686 assert!(d.message.contains("NOT_DECLARED"));
3687 assert!(d.suggestion.is_some());
3688 match d.span.expect("span must be set").element {
3689 mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "review"),
3690 other => panic!("expected Agent span, got {other:?}"),
3691 }
3692 }
3693}
3694
3695#[cfg(test)]
3697mod subprocess_embed_compile_tests {
3698 use super::*;
3699 use mlua_swarm_schema::{current_schema_version, SubprocessDef, SubprocessOverrides};
3700
3701 fn subprocess_agent(name: &str, runner: Option<Runner>) -> AgentDef {
3702 AgentDef {
3703 name: name.to_string(),
3704 kind: AgentKind::Subprocess,
3705 spec: serde_json::json!({}),
3706 profile: Some(AgentProfile {
3707 system_prompt: "you are a headless worker".to_string(),
3708 model: Some("profile-model".to_string()),
3709 tools: vec!["Read".to_string()],
3710 ..Default::default()
3711 }),
3712 meta: None,
3713 runner,
3714 runner_ref: None,
3715 verdict: None,
3716 }
3717 }
3718
3719 fn echo_def(name: &str) -> SubprocessDef {
3720 SubprocessDef {
3721 name: name.to_string(),
3722 argv: vec!["sh".to_string(), "-c".to_string(), "cat".to_string()],
3723 stdin: Some("{prompt}".to_string()),
3724 env: Default::default(),
3725 cwd: None,
3726 output: None,
3727 stream_mode: None,
3728 }
3729 }
3730
3731 fn bp_with(agents: Vec<AgentDef>, subprocesses: Vec<SubprocessDef>) -> Blueprint {
3732 Blueprint {
3733 schema_version: current_schema_version(),
3734 id: "gh83-ut".into(),
3735 flow: FlowNode::Seq { children: vec![] },
3736 agents,
3737 operators: vec![],
3738 metas: vec![],
3739 hints: Default::default(),
3740 strategy: Default::default(),
3741 metadata: BlueprintMetadata::default(),
3742 spawner_hints: Default::default(),
3743 default_agent_kind: AgentKind::Operator,
3744 default_operator_kind: None,
3745 default_init_ctx: None,
3746 default_agent_ctx: None,
3747 default_context_policy: None,
3748 projection_placement: None,
3749 audits: vec![],
3750 degradation_policy: None,
3751 runners: vec![],
3752 default_runner: None,
3753 subprocesses,
3754 check_policy: None,
3755 blueprint_ref_includes: vec![],
3756 }
3757 }
3758
3759 fn subprocess_runner(template: &str) -> Runner {
3760 Runner::Subprocess {
3761 template: template.to_string(),
3762 overrides: SubprocessOverrides::default(),
3763 }
3764 }
3765
3766 #[test]
3767 fn validate_placeholders_accepts_closed_set_and_json_braces() {
3768 for ok in [
3769 "{system} {system_file} {prompt} {model} {tools_csv} {work_dir} {task_id} {attempt}",
3770 r#"echo '{"result": "ok", "nested": {"a": 1}}'"#,
3771 "no placeholders at all",
3772 "unmatched { brace",
3773 ] {
3774 validate_embed_placeholders(ok, "ut").expect("must be accepted");
3775 }
3776 }
3777
3778 #[test]
3779 fn validate_placeholders_rejects_unknown_token() {
3780 let err = validate_embed_placeholders("--flag {evil}", "argv[1]").unwrap_err();
3781 assert!(err.contains("'{evil}'"), "token named: {err}");
3782 assert!(err.contains("closed set"), "closed set listed: {err}");
3783 }
3784
3785 #[test]
3789 fn validate_placeholders_descends_into_literal_braces() {
3790 validate_embed_placeholders(r#"{"task": "{prompt}"}"#, "stdin")
3791 .expect("nested closed-set token must be accepted");
3792 let err = validate_embed_placeholders(r#"{"task": "{evil}"}"#, "stdin").unwrap_err();
3793 assert!(
3794 err.contains("'{evil}'"),
3795 "nested unknown token caught: {err}"
3796 );
3797 }
3798
3799 #[test]
3800 fn hint_resolution_finds_declared_template() {
3801 let agent = subprocess_agent("headless", Some(subprocess_runner("echo")));
3802 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
3803 let hint = resolve_subprocess_template_hint(&bp, &agent)
3804 .expect("resolves")
3805 .expect("Runner::Subprocess must synthesize a hint");
3806 assert_eq!(hint[SUBPROCESS_TEMPLATE_HINT_KEY]["name"], "echo");
3807 assert!(hint.get(SUBPROCESS_OVERRIDES_HINT_KEY).is_some());
3808 }
3809
3810 #[test]
3811 fn hint_resolution_unknown_template_is_invalid_spec() {
3812 let agent = subprocess_agent("headless", Some(subprocess_runner("nope")));
3813 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
3814 let err = resolve_subprocess_template_hint(&bp, &agent).unwrap_err();
3815 let msg = format!("{err}");
3816 assert!(msg.contains("'nope'"), "missing template named: {msg}");
3817 assert!(msg.contains("echo"), "defined templates listed: {msg}");
3818 }
3819
3820 #[test]
3821 fn hint_resolution_none_without_subprocess_runner() {
3822 let agent = subprocess_agent("headless", None);
3823 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
3824 let hint = resolve_subprocess_template_hint(&bp, &agent).expect("resolves");
3825 assert!(hint.is_none(), "spec-based agents keep the historical path");
3826 }
3827
3828 fn agent_block_agent(name: &str, runner: Option<Runner>, profile_tools: &[&str]) -> AgentDef {
3835 AgentDef {
3836 name: name.to_string(),
3837 kind: AgentKind::AgentBlock,
3838 spec: serde_json::json!({}),
3839 profile: Some(AgentProfile {
3840 system_prompt: "you are an in-process auditor".to_string(),
3841 tools: profile_tools.iter().map(|t| t.to_string()).collect(),
3842 ..Default::default()
3843 }),
3844 meta: None,
3845 runner,
3846 runner_ref: None,
3847 verdict: None,
3848 }
3849 }
3850
3851 fn agent_block_runner(tools: &[&str]) -> Runner {
3852 Runner::AgentBlockInProcess {
3853 tools: tools.iter().map(|t| t.to_string()).collect(),
3854 }
3855 }
3856
3857 #[test]
3863 fn agent_block_runner_tools_are_projected_over_profile_tools() {
3864 let agent = agent_block_agent(
3865 "auditor",
3866 Some(agent_block_runner(&["mcp__outline__list_docs"])),
3867 &["Read"],
3868 );
3869 let bp = bp_with(vec![agent], vec![]);
3870 let bound = resolve_bound_agents(&bp).expect("binds");
3871 let effective = materialize_bound_blueprint(&bp, &bound);
3872 assert_eq!(
3873 effective.agents[0].profile.as_ref().unwrap().tools,
3874 vec!["mcp__outline__list_docs".to_string()],
3875 "the declared Runner tools replace profile.tools (['Read'])"
3876 );
3877 }
3878
3879 #[test]
3883 fn agent_block_projection_distinguishes_declared_empty_from_absent() {
3884 let declared = agent_block_agent("auditor", Some(agent_block_runner(&[])), &["Read"]);
3885 let bp = bp_with(vec![declared], vec![]);
3886 let bound = resolve_bound_agents(&bp).expect("binds");
3887 let effective = materialize_bound_blueprint(&bp, &bound);
3888 assert!(
3889 effective.agents[0]
3890 .profile
3891 .as_ref()
3892 .unwrap()
3893 .tools
3894 .is_empty(),
3895 "empty means enforced-empty, not 'unset'"
3896 );
3897
3898 let absent = agent_block_agent("auditor", None, &["Read"]);
3899 let bp = bp_with(vec![absent], vec![]);
3900 let bound = resolve_bound_agents(&bp).expect("binds");
3901 let effective = materialize_bound_blueprint(&bp, &bound);
3902 assert_eq!(
3903 effective.agents[0].profile.as_ref().unwrap().tools,
3904 vec!["Read".to_string()],
3905 "no Runner declared → the agent.md tools line stands"
3906 );
3907 }
3908
3909 #[test]
3916 fn compile_rejects_script_mode_with_a_declared_mcp_grant() {
3917 let mut agent = agent_block_agent(
3918 "auditor",
3919 Some(agent_block_runner(&["mcp__outline__list_docs"])),
3920 &[],
3921 );
3922 agent.spec = serde_json::json!({ "script_path": "gate.lua" });
3923 let mut bp = bp_with(vec![agent], vec![]);
3924 bp.strategy.strict_refs = false;
3925
3926 let mut registry = SpawnerRegistry::new();
3927 registry.register::<crate::worker::agent_block::AgentBlockInProcessSpawnerFactory>(
3928 Arc::new(crate::worker::agent_block::AgentBlockInProcessSpawnerFactory::new()),
3929 );
3930 let err = match Compiler::new(registry).compile(&bp) {
3932 Err(e) => e,
3933 Ok(_) => panic!("script mode + declared MCP grant must not compile"),
3934 };
3935 let msg = format!("{err}");
3936 assert!(msg.contains("script_path"), "names the trigger: {msg}");
3937 assert!(
3938 msg.contains("mcp__outline__list_docs"),
3939 "names the unenforceable tools: {msg}"
3940 );
3941 }
3942
3943 #[test]
3947 fn compile_accepts_script_mode_with_only_inert_tools() {
3948 let mut agent = agent_block_agent("auditor", None, &["Read", "WebSearch"]);
3949 agent.spec = serde_json::json!({ "script_path": "gate.lua" });
3950 let mut bp = bp_with(vec![agent], vec![]);
3951 bp.strategy.strict_refs = false;
3952
3953 let mut registry = SpawnerRegistry::new();
3954 registry.register::<crate::worker::agent_block::AgentBlockInProcessSpawnerFactory>(
3955 Arc::new(crate::worker::agent_block::AgentBlockInProcessSpawnerFactory::new()),
3956 );
3957 if let Err(e) = Compiler::new(registry).compile(&bp) {
3958 panic!("inert tools must not trip the MCP-grant guard: {e}");
3959 }
3960 }
3961
3962 #[test]
3963 fn build_embed_rejects_unknown_placeholder() {
3964 let agent = subprocess_agent("headless", None);
3965 let mut def = echo_def("echo");
3966 def.argv.push("--x={evil}".to_string());
3967 let err = SubprocessProcessSpawnerFactory::build_embed(
3968 &agent,
3969 &serde_json::to_value(&def).unwrap(),
3970 None,
3971 )
3972 .unwrap_err();
3973 assert!(format!("{err}").contains("'{evil}'"));
3974 }
3975
3976 #[test]
3977 fn build_embed_rejects_output_with_stream_mode() {
3978 let agent = subprocess_agent("headless", None);
3979 let mut def = echo_def("echo");
3980 def.stream_mode = Some("ndjson_lines".to_string());
3981 def.output = Some(mlua_swarm_schema::SubprocessOutput {
3982 format: Some("json".to_string()),
3983 result_ptr: None,
3984 ok_from: None,
3985 stats: None,
3986 });
3987 let err = SubprocessProcessSpawnerFactory::build_embed(
3988 &agent,
3989 &serde_json::to_value(&def).unwrap(),
3990 None,
3991 )
3992 .unwrap_err();
3993 assert!(format!("{err}").contains("plain-mode"));
3994 }
3995
3996 #[test]
3997 fn build_embed_rejects_malformed_result_ptr_and_ok_from() {
3998 let agent = subprocess_agent("headless", None);
3999 let mut def = echo_def("echo");
4000 def.output = Some(mlua_swarm_schema::SubprocessOutput {
4001 format: None,
4002 result_ptr: Some("result".to_string()),
4003 ok_from: None,
4004 stats: None,
4005 });
4006 let err = SubprocessProcessSpawnerFactory::build_embed(
4007 &agent,
4008 &serde_json::to_value(&def).unwrap(),
4009 None,
4010 )
4011 .unwrap_err();
4012 assert!(format!("{err}").contains("JSON Pointer"));
4013
4014 let mut def = echo_def("echo");
4015 def.output = Some(mlua_swarm_schema::SubprocessOutput {
4016 format: None,
4017 result_ptr: None,
4018 ok_from: Some("status".to_string()),
4019 stats: None,
4020 });
4021 let err = SubprocessProcessSpawnerFactory::build_embed(
4022 &agent,
4023 &serde_json::to_value(&def).unwrap(),
4024 None,
4025 )
4026 .unwrap_err();
4027 assert!(format!("{err}").contains("exit_code"));
4028 }
4029
4030 #[test]
4031 fn build_embed_bakes_profile_with_override_precedence() {
4032 let agent = subprocess_agent("headless", None);
4033 let def = echo_def("echo");
4034 let overrides = SubprocessOverrides {
4035 model: Some("override-model".to_string()),
4036 tools: vec!["Bash".to_string(), "Write".to_string()],
4037 cwd: Some("/tmp/override-wd".to_string()),
4038 };
4039 let sp = SubprocessProcessSpawnerFactory::build_embed(
4040 &agent,
4041 &serde_json::to_value(&def).unwrap(),
4042 Some(&serde_json::to_value(&overrides).unwrap()),
4043 )
4044 .expect("builds");
4045 let embed = sp.embed.as_ref().expect("embed template baked");
4046 assert_eq!(embed.model.as_deref(), Some("override-model"));
4047 assert_eq!(embed.tools_csv, "Bash,Write");
4048 assert_eq!(embed.cwd.as_deref(), Some("/tmp/override-wd"));
4049 assert_eq!(
4050 embed.system_prompt.as_deref(),
4051 Some("you are a headless worker")
4052 );
4053 }
4054}