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_pinned(bp, &bound_agents, None)
558 }
559
560 pub fn compile_bound(
565 &self,
566 bp: &Blueprint,
567 bound_agents: &[BoundAgent],
568 ) -> Result<CompiledBlueprint, CompileError> {
569 self.compile_bound_pinned(bp, bound_agents, None)
570 }
571
572 pub fn compile_bound_pinned(
587 &self,
588 bp: &Blueprint,
589 bound_agents: &[BoundAgent],
590 operator_pin: Option<&str>,
591 ) -> Result<CompiledBlueprint, CompileError> {
592 let effective = materialize_bound_blueprint(bp, bound_agents);
593 self.compile_resolved(&effective, operator_pin)
594 }
595
596 fn compile_resolved(
597 &self,
598 bp: &Blueprint,
599 operator_pin: Option<&str>,
600 ) -> Result<CompiledBlueprint, CompileError> {
601 let mut routes: HashMap<String, Arc<dyn SpawnerAdapter>> = HashMap::new();
602 let mut seen: HashMap<String, ()> = HashMap::new();
603 let mut verdict_contracts: HashMap<String, VerdictContract> = HashMap::new();
609
610 let defined: Vec<String> = bp.operators.iter().map(|o| o.name.clone()).collect();
616 for ad in &bp.agents {
617 if !matches!(ad.kind, AgentKind::Operator) {
618 continue;
619 }
620 let op_ref = ad.spec.get("operator_ref").and_then(|v| v.as_str());
621 if let Some(op_ref) = op_ref {
622 if !defined.iter().any(|n| n == op_ref) {
623 return Err(CompileError::UnresolvedOperatorRef {
624 agent: ad.name.clone(),
625 op_ref: op_ref.to_string(),
626 defined: defined.clone(),
627 });
628 }
629 }
630 }
632
633 let metas_defined: Vec<String> = bp.metas.iter().map(|m| m.name.clone()).collect();
637 for ad in &bp.agents {
638 let meta_ref = ad.meta.as_ref().and_then(|m| m.meta_ref.as_ref());
639 if let Some(meta_ref) = meta_ref {
640 if !metas_defined.iter().any(|n| n == meta_ref) {
641 return Err(CompileError::UnresolvedMetaRef {
642 where_: format!("AgentMeta.meta_ref of agent '{}'", ad.name),
643 meta_ref: meta_ref.clone(),
644 defined: metas_defined.clone(),
645 });
646 }
647 }
648 }
649 let mut static_step_meta_refs: Vec<(String, String)> = Vec::new();
655 collect_step_meta_refs(&bp.flow, &mut static_step_meta_refs);
656 for (where_, meta_ref) in static_step_meta_refs {
657 if !metas_defined.iter().any(|n| n == &meta_ref) {
658 return Err(CompileError::UnresolvedMetaRef {
659 where_,
660 meta_ref,
661 defined: metas_defined.clone(),
662 });
663 }
664 }
665
666 let agents_defined: Vec<String> = bp.agents.iter().map(|a| a.name.clone()).collect();
671 for audit in &bp.audits {
672 if !agents_defined.iter().any(|n| n == &audit.agent) {
673 return Err(CompileError::UnresolvedAuditAgent {
674 agent: audit.agent.clone(),
675 defined: agents_defined.clone(),
676 });
677 }
678 }
679
680 for ad in &bp.agents {
681 if seen.contains_key(&ad.name) {
682 return Err(CompileError::DuplicateAgent(ad.name.clone()));
683 }
684 seen.insert(ad.name.clone(), ());
685
686 if let Some(contract) = &ad.verdict {
692 verdict_contracts.insert(ad.name.clone(), contract.clone());
693 }
694
695 let factory = match self.registry.factories.get(&ad.kind) {
696 Some(f) => f.clone(),
697 None => {
698 if bp.strategy.strict_kind {
699 return Err(CompileError::UnknownKind(ad.kind.clone()));
700 } else {
701 tracing::warn!(
702 agent = %ad.name,
703 kind = ?ad.kind,
704 "no spawner factory registered for agent kind; \
705 dropping agent from routing table (strict_kind=false)"
706 );
707 continue;
708 }
709 }
710 };
711 let hint = bp.hints.per_agent.get(&ad.name);
712 let subprocess_hint = if ad.kind == AgentKind::Subprocess {
724 resolve_subprocess_template_hint(bp, ad)?
725 } else {
726 None
727 };
728 let operator_pin_hint = match operator_pin {
736 Some(pin) if ad.kind == AgentKind::Operator => {
737 Some(merge_operator_pin_hint(hint, pin, &ad.name)?)
738 }
739 _ => None,
740 };
741 let spawner = factory.build(
742 ad,
743 operator_pin_hint
744 .as_ref()
745 .or(subprocess_hint.as_ref())
746 .or(hint),
747 )?;
748 routes.insert(ad.name.clone(), spawner);
749 }
750
751 let strict_verdict_handling = bp.metadata.strict_verdict_handling.unwrap_or(false);
768 verify_verdict_conds(&bp.flow, &verdict_contracts, strict_verdict_handling)?;
769
770 if bp.strategy.strict_refs {
771 verify_refs(&bp.flow, &routes, self.default_spawner.is_some())?;
772 }
773
774 let (step_naming, step_naming_warnings) = StepNaming::from_blueprint(bp)?;
790 for warning in &step_naming_warnings {
791 tracing::warn!(
792 name = %warning.name,
793 first_step_ref = %warning.first_step_ref,
794 second_step_ref = %warning.second_step_ref,
795 "StepNaming: undeclared steps' canonical/alias names collide; \
796 the step whose own ref matches the name keeps it (data-plane priority)"
797 );
798 }
799
800 let projection_placement =
808 ProjectionPlacement::from_spec(bp.projection_placement.as_ref())?;
809
810 let router = Arc::new(CompiledAgentTable {
811 routes,
812 default: self.default_spawner.clone(),
813 verdict_contracts,
814 });
815 Ok(CompiledBlueprint {
816 router,
817 flow: bp.flow.clone(),
818 metadata: bp.metadata.clone(),
819 step_naming: Arc::new(step_naming),
820 projection_placement: Arc::new(projection_placement),
821 })
822 }
823}
824
825fn verify_refs(
828 node: &FlowNode,
829 routes: &HashMap<String, Arc<dyn SpawnerAdapter>>,
830 has_default: bool,
831) -> Result<(), CompileError> {
832 let mut refs: Vec<String> = Vec::new();
833 collect_refs(node, &mut refs);
834 for r in refs {
835 if !routes.contains_key(&r) && !has_default {
836 return Err(CompileError::UnresolvedRef(r));
837 }
838 }
839 Ok(())
840}
841
842fn collect_refs(node: &FlowNode, out: &mut Vec<String>) {
843 match node {
844 FlowNode::Step { ref_, .. } => out.push(ref_.clone()),
845 FlowNode::Seq { children } => {
846 for c in children {
847 collect_refs(c, out);
848 }
849 }
850 FlowNode::Branch { then_, else_, .. } => {
851 collect_refs(then_, out);
852 collect_refs(else_, out);
853 }
854 FlowNode::Fanout { body, .. } => collect_refs(body, out),
855 FlowNode::Loop { body, .. } => collect_refs(body, out),
856 FlowNode::Try { body, catch, .. } => {
857 collect_refs(body, out);
858 collect_refs(catch, out);
859 }
860 FlowNode::Assign { .. } => {} }
862}
863
864fn collect_step_meta_refs(node: &FlowNode, out: &mut Vec<(String, String)>) {
872 match node {
873 FlowNode::Step { ref_, in_, .. } => {
874 if let Expr::Lit { value } = in_ {
875 if let Some(meta_ref) = static_step_meta_ref(value) {
876 out.push((format!("Step '{ref_}' $step_meta.ref"), meta_ref));
877 }
878 }
879 }
880 FlowNode::Seq { children } => {
881 for c in children {
882 collect_step_meta_refs(c, out);
883 }
884 }
885 FlowNode::Branch { then_, else_, .. } => {
886 collect_step_meta_refs(then_, out);
887 collect_step_meta_refs(else_, out);
888 }
889 FlowNode::Fanout { body, .. } => collect_step_meta_refs(body, out),
890 FlowNode::Loop { body, .. } => collect_step_meta_refs(body, out),
891 FlowNode::Try { body, catch, .. } => {
892 collect_step_meta_refs(body, out);
893 collect_step_meta_refs(catch, out);
894 }
895 FlowNode::Assign { .. } => {} }
897}
898
899fn static_step_meta_ref(value: &Value) -> Option<String> {
906 value
907 .as_object()?
908 .get("$step_meta")?
909 .as_object()?
910 .get("ref")?
911 .as_str()
912 .map(str::to_string)
913}
914
915fn verify_verdict_conds(
927 flow: &FlowNode,
928 verdict_contracts: &HashMap<String, VerdictContract>,
929 strict_verdict_handling: bool,
930) -> Result<(), CompileError> {
931 let mut step_outputs: HashMap<String, String> = HashMap::new();
932 let mut step_agents: HashMap<String, String> = HashMap::new();
933 collect_step_outputs_and_agents(flow, &mut step_outputs, &mut step_agents);
934
935 let mut errors: Vec<CompileError> = Vec::new();
936 let mut referenced_values: HashMap<String, std::collections::HashSet<String>> = HashMap::new();
937 collect_verdict_conds(
938 flow,
939 &step_outputs,
940 verdict_contracts,
941 &mut referenced_values,
942 &mut errors,
943 );
944 check_unhandled_verdict_values(
945 verdict_contracts,
946 &referenced_values,
947 &step_agents,
948 strict_verdict_handling,
949 &mut errors,
950 );
951 match errors.into_iter().next() {
952 Some(e) => Err(e),
953 None => Ok(()),
954 }
955}
956
957fn collect_step_outputs_and_agents(
972 node: &FlowNode,
973 out: &mut HashMap<String, String>,
974 step_agents: &mut HashMap<String, String>,
975) {
976 match node {
977 FlowNode::Step {
978 ref_,
979 out: out_expr,
980 ..
981 } => {
982 if let Expr::Path { at } = out_expr {
983 out.insert(at.to_string(), ref_.clone());
984 }
985 step_agents
986 .entry(ref_.clone())
987 .or_insert_with(|| ref_.clone());
988 }
989 FlowNode::Seq { children } => {
990 for c in children {
991 collect_step_outputs_and_agents(c, out, step_agents);
992 }
993 }
994 FlowNode::Branch { then_, else_, .. } => {
995 collect_step_outputs_and_agents(then_, out, step_agents);
996 collect_step_outputs_and_agents(else_, out, step_agents);
997 }
998 FlowNode::Fanout { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
999 FlowNode::Loop { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
1000 FlowNode::Try { body, catch, .. } => {
1001 collect_step_outputs_and_agents(body, out, step_agents);
1002 collect_step_outputs_and_agents(catch, out, step_agents);
1003 }
1004 FlowNode::Assign { .. } => {} }
1006}
1007
1008fn collect_verdict_conds(
1013 node: &FlowNode,
1014 step_outputs: &HashMap<String, String>,
1015 verdict_contracts: &HashMap<String, VerdictContract>,
1016 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
1017 errors: &mut Vec<CompileError>,
1018) {
1019 match node {
1020 FlowNode::Branch { cond, then_, else_ } => {
1021 lint_cond_expr(
1022 cond,
1023 "Branch cond",
1024 step_outputs,
1025 verdict_contracts,
1026 referenced_values,
1027 errors,
1028 );
1029 collect_verdict_conds(
1030 then_,
1031 step_outputs,
1032 verdict_contracts,
1033 referenced_values,
1034 errors,
1035 );
1036 collect_verdict_conds(
1037 else_,
1038 step_outputs,
1039 verdict_contracts,
1040 referenced_values,
1041 errors,
1042 );
1043 }
1044 FlowNode::Loop { cond, body, .. } => {
1045 lint_cond_expr(
1046 cond,
1047 "Loop cond",
1048 step_outputs,
1049 verdict_contracts,
1050 referenced_values,
1051 errors,
1052 );
1053 collect_verdict_conds(
1054 body,
1055 step_outputs,
1056 verdict_contracts,
1057 referenced_values,
1058 errors,
1059 );
1060 }
1061 FlowNode::Seq { children } => {
1062 for c in children {
1063 collect_verdict_conds(
1064 c,
1065 step_outputs,
1066 verdict_contracts,
1067 referenced_values,
1068 errors,
1069 );
1070 }
1071 }
1072 FlowNode::Fanout { body, .. } => collect_verdict_conds(
1073 body,
1074 step_outputs,
1075 verdict_contracts,
1076 referenced_values,
1077 errors,
1078 ),
1079 FlowNode::Try { body, catch, .. } => {
1080 collect_verdict_conds(
1081 body,
1082 step_outputs,
1083 verdict_contracts,
1084 referenced_values,
1085 errors,
1086 );
1087 collect_verdict_conds(
1088 catch,
1089 step_outputs,
1090 verdict_contracts,
1091 referenced_values,
1092 errors,
1093 );
1094 }
1095 FlowNode::Step { .. } | FlowNode::Assign { .. } => {}
1096 }
1097}
1098
1099fn lint_cond_expr(
1108 expr: &Expr,
1109 where_: &str,
1110 step_outputs: &HashMap<String, String>,
1111 verdict_contracts: &HashMap<String, VerdictContract>,
1112 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
1113 errors: &mut Vec<CompileError>,
1114) {
1115 match expr {
1116 Expr::Eq { lhs, rhs } | Expr::Ne { lhs, rhs } => {
1117 if let Some((path, lit)) = path_lit_operands(lhs, rhs) {
1118 resolve_and_check(
1119 path,
1120 &[lit],
1121 where_,
1122 step_outputs,
1123 verdict_contracts,
1124 referenced_values,
1125 errors,
1126 );
1127 }
1128 }
1129 Expr::In { needle, haystack } => {
1130 if let (
1131 Expr::Path { at },
1132 Expr::Lit {
1133 value: Value::Array(items),
1134 },
1135 ) = (needle.as_ref(), haystack.as_ref())
1136 {
1137 let lits: Vec<&Value> = items.iter().collect();
1138 resolve_and_check(
1139 at,
1140 &lits,
1141 where_,
1142 step_outputs,
1143 verdict_contracts,
1144 referenced_values,
1145 errors,
1146 );
1147 }
1148 }
1149 Expr::And { args } | Expr::Or { args } => {
1150 for a in args {
1151 lint_cond_expr(
1152 a,
1153 where_,
1154 step_outputs,
1155 verdict_contracts,
1156 referenced_values,
1157 errors,
1158 );
1159 }
1160 }
1161 Expr::Not { arg } => lint_cond_expr(
1162 arg,
1163 where_,
1164 step_outputs,
1165 verdict_contracts,
1166 referenced_values,
1167 errors,
1168 ),
1169 _ => {}
1170 }
1171}
1172
1173fn path_lit_operands<'a>(lhs: &'a Expr, rhs: &'a Expr) -> Option<(&'a Path, &'a Value)> {
1179 match (lhs, rhs) {
1180 (Expr::Path { at }, Expr::Lit { value }) => Some((at, value)),
1181 (Expr::Lit { value }, Expr::Path { at }) => Some((at, value)),
1182 _ => None,
1183 }
1184}
1185
1186fn resolve_and_check(
1201 path: &Path,
1202 lits: &[&Value],
1203 where_: &str,
1204 step_outputs: &HashMap<String, String>,
1205 verdict_contracts: &HashMap<String, VerdictContract>,
1206 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
1207 errors: &mut Vec<CompileError>,
1208) {
1209 let path_str = path.to_string();
1210 let (agent, actual_shape) = if let Some(agent) = step_outputs.get(&path_str) {
1211 (agent, "body")
1212 } else if let Some(stripped) = path_str.strip_suffix(".parts.verdict") {
1213 match step_outputs.get(stripped) {
1214 Some(agent) => (agent, "part"),
1215 None => return,
1216 }
1217 } else {
1218 return;
1219 };
1220
1221 let Some(contract) = verdict_contracts.get(agent) else {
1222 tracing::warn!(
1223 agent = %agent,
1224 where_ = %where_,
1225 "cond references agent output but no verdict contract declared"
1226 );
1227 return;
1228 };
1229
1230 let expected_channel = match contract.channel {
1231 VerdictChannel::Body => "body",
1232 VerdictChannel::Part => "part",
1233 };
1234 if expected_channel != actual_shape {
1235 errors.push(CompileError::VerdictChannelMismatch {
1236 where_: where_.to_string(),
1237 agent: agent.clone(),
1238 expected_channel: expected_channel.to_string(),
1239 actual_shape: actual_shape.to_string(),
1240 });
1241 return;
1242 }
1243
1244 for lit in lits {
1245 let value_str = lit
1246 .as_str()
1247 .map(str::to_string)
1248 .unwrap_or_else(|| lit.to_string());
1249 if !contract.values.iter().any(|v| v == &value_str) {
1250 errors.push(CompileError::VerdictValueNotInContract {
1251 where_: where_.to_string(),
1252 agent: agent.clone(),
1253 value: value_str.clone(),
1254 values: contract.values.clone(),
1255 });
1256 }
1257 referenced_values
1264 .entry(agent.clone())
1265 .or_default()
1266 .insert(value_str);
1267 }
1268}
1269
1270fn check_unhandled_verdict_values(
1288 verdict_contracts: &HashMap<String, VerdictContract>,
1289 referenced_values: &HashMap<String, std::collections::HashSet<String>>,
1290 step_agents: &HashMap<String, String>,
1291 strict_verdict_handling: bool,
1292 errors: &mut Vec<CompileError>,
1293) {
1294 for finding in fold_unhandled_verdict_values(verdict_contracts, referenced_values, step_agents)
1295 {
1296 if strict_verdict_handling {
1297 errors.push(CompileError::VerdictValueUnhandled {
1298 agent: finding.agent,
1299 value: finding.value,
1300 declared_values: finding.declared_values,
1301 step_ref: finding.step_ref,
1302 });
1303 } else {
1304 tracing::warn!(
1305 agent = %finding.agent,
1306 value = %finding.value,
1307 step_ref = %finding.step_ref,
1308 "declared verdict value has no downstream cond handler; \
1309 opt in to `metadata.strict_verdict_handling` to reject at compile"
1310 );
1311 }
1312 }
1313}
1314
1315#[derive(Debug, Clone, PartialEq, Eq)]
1326pub struct UnhandledVerdictValue {
1327 pub agent: String,
1329 pub value: String,
1331 pub declared_values: Vec<String>,
1333 pub step_ref: String,
1335}
1336
1337pub fn unhandled_verdict_values(
1354 flow: &FlowNode,
1355 verdict_contracts: &HashMap<String, VerdictContract>,
1356) -> Vec<UnhandledVerdictValue> {
1357 let mut step_outputs: HashMap<String, String> = HashMap::new();
1358 let mut step_agents: HashMap<String, String> = HashMap::new();
1359 collect_step_outputs_and_agents(flow, &mut step_outputs, &mut step_agents);
1360
1361 let mut referenced_values: HashMap<String, std::collections::HashSet<String>> = HashMap::new();
1362 let mut discarded_errors: Vec<CompileError> = Vec::new();
1363 collect_verdict_conds(
1364 flow,
1365 &step_outputs,
1366 verdict_contracts,
1367 &mut referenced_values,
1368 &mut discarded_errors,
1369 );
1370 fold_unhandled_verdict_values(verdict_contracts, &referenced_values, &step_agents)
1371}
1372
1373#[derive(Debug, Clone, PartialEq, Eq)]
1387pub struct AgentContractUnread {
1388 pub agent: String,
1390 pub declared_values: Vec<String>,
1392 pub step_ref: String,
1394}
1395
1396pub fn agents_with_all_verdict_values_unread(
1407 flow: &FlowNode,
1408 verdict_contracts: &HashMap<String, VerdictContract>,
1409) -> Vec<AgentContractUnread> {
1410 let per_value = unhandled_verdict_values(flow, verdict_contracts);
1411 let mut unread_counts: HashMap<String, usize> = HashMap::new();
1412 for finding in &per_value {
1413 *unread_counts.entry(finding.agent.clone()).or_default() += 1;
1414 }
1415 let mut agents: Vec<&String> = verdict_contracts.keys().collect();
1416 agents.sort();
1417 let mut out = Vec::new();
1418 for agent in agents {
1419 let contract = &verdict_contracts[agent];
1420 let declared = contract.values.len();
1421 if declared == 0 {
1422 continue;
1423 }
1424 let unread = unread_counts.get(agent).copied().unwrap_or(0);
1425 if unread != declared {
1426 continue;
1427 }
1428 let step_ref = per_value
1432 .iter()
1433 .find(|f| &f.agent == agent)
1434 .map(|f| f.step_ref.clone())
1435 .unwrap_or_else(|| agent.clone());
1436 out.push(AgentContractUnread {
1437 agent: agent.clone(),
1438 declared_values: contract.values.clone(),
1439 step_ref,
1440 });
1441 }
1442 out
1443}
1444
1445fn fold_unhandled_verdict_values(
1456 verdict_contracts: &HashMap<String, VerdictContract>,
1457 referenced_values: &HashMap<String, std::collections::HashSet<String>>,
1458 step_agents: &HashMap<String, String>,
1459) -> Vec<UnhandledVerdictValue> {
1460 let mut agents: Vec<&String> = verdict_contracts.keys().collect();
1461 agents.sort();
1462 let mut findings = Vec::new();
1463 for agent in agents {
1464 let contract = &verdict_contracts[agent];
1465 let referenced = referenced_values.get(agent);
1466 let step_ref = step_agents
1467 .get(agent)
1468 .cloned()
1469 .unwrap_or_else(|| agent.clone());
1470 for value in &contract.values {
1471 let handled = referenced.map(|set| set.contains(value)).unwrap_or(false);
1472 if handled {
1473 continue;
1474 }
1475 findings.push(UnhandledVerdictValue {
1476 agent: agent.clone(),
1477 value: value.clone(),
1478 declared_values: contract.values.clone(),
1479 step_ref: step_ref.clone(),
1480 });
1481 }
1482 }
1483 findings
1484}
1485
1486pub struct CompiledAgentTable {
1499 pub(crate) routes: HashMap<String, Arc<dyn SpawnerAdapter>>,
1500 pub(crate) default: Option<Arc<dyn SpawnerAdapter>>,
1501 pub(crate) verdict_contracts: HashMap<String, VerdictContract>,
1505}
1506
1507impl CompiledAgentTable {
1508 pub fn has_route(&self, agent: &str) -> bool {
1511 self.routes.contains_key(agent)
1512 }
1513 pub fn routed_agents(&self) -> Vec<String> {
1515 self.routes.keys().cloned().collect()
1516 }
1517 pub fn verdict_contract_for(&self, agent: &str) -> Option<&VerdictContract> {
1521 self.verdict_contracts.get(agent)
1522 }
1523}
1524
1525#[async_trait]
1526impl SpawnerAdapter for CompiledAgentTable {
1527 async fn spawn(
1528 &self,
1529 engine: &Engine,
1530 ctx: &Ctx,
1531 task_id: StepId,
1532 attempt: u32,
1533 token: CapToken,
1534 ) -> Result<Box<dyn Worker>, SpawnError> {
1535 let sp = self
1536 .routes
1537 .get(&ctx.agent)
1538 .cloned()
1539 .or_else(|| self.default.clone())
1540 .ok_or_else(|| SpawnError::NotRegistered(ctx.agent.clone()))?;
1541 sp.spawn(engine, ctx, task_id, attempt, token).await
1542 }
1543}
1544
1545pub struct SubprocessProcessSpawnerFactory;
1576
1577impl SpawnerFactoryKind for SubprocessProcessSpawnerFactory {
1578 const KIND: AgentKind = AgentKind::Subprocess;
1579 type Worker = crate::worker::process_spawner::ProcessWorker;
1580}
1581
1582pub const SUBPROCESS_TEMPLATE_HINT_KEY: &str = "subprocess_template";
1585pub const SUBPROCESS_OVERRIDES_HINT_KEY: &str = "subprocess_overrides";
1587
1588fn validate_embed_placeholders(s: &str, where_: &str) -> Result<(), String> {
1602 let mut rest = s;
1603 while let Some(start) = rest.find('{') {
1604 let after = &rest[start + 1..];
1605 let Some(end) = after.find('}') else {
1606 break;
1607 };
1608 let token = &after[..end];
1609 let is_candidate =
1610 !token.is_empty() && token.chars().all(|c| c.is_ascii_lowercase() || c == '_');
1611 if is_candidate {
1612 if !crate::worker::process_spawner::EMBED_PLACEHOLDERS.contains(&token) {
1613 return Err(format!(
1614 "unknown placeholder '{{{token}}}' in {where_}; closed set is \
1615 {{system, system_file, prompt, model, tools_csv, work_dir, task_id, attempt}}"
1616 ));
1617 }
1618 rest = &after[end + 1..];
1619 } else {
1620 rest = after;
1625 }
1626 }
1627 Ok(())
1628}
1629
1630fn resolve_subprocess_template_hint(
1636 bp: &Blueprint,
1637 ad: &AgentDef,
1638) -> Result<Option<Value>, CompileError> {
1639 let invalid = |msg: String| CompileError::InvalidSpec {
1640 name: ad.name.clone(),
1641 msg,
1642 };
1643 let runner = mlua_swarm_schema::resolve_runner(bp, ad).map_err(|e| invalid(e.to_string()))?;
1644 let Some(Runner::Subprocess {
1645 template,
1646 overrides,
1647 }) = runner
1648 else {
1649 return Ok(None);
1650 };
1651 let def = bp
1652 .subprocesses
1653 .iter()
1654 .find(|d| d.name == template)
1655 .ok_or_else(|| {
1656 let mut names: Vec<&str> = bp.subprocesses.iter().map(|d| d.name.as_str()).collect();
1657 names.sort_unstable();
1658 invalid(format!(
1659 "Runner::Subprocess template '{template}' not found in \
1660 Blueprint.subprocesses (defined: [{}])",
1661 names.join(", ")
1662 ))
1663 })?;
1664 Ok(Some(serde_json::json!({
1665 SUBPROCESS_TEMPLATE_HINT_KEY: def,
1666 SUBPROCESS_OVERRIDES_HINT_KEY: overrides,
1667 })))
1668}
1669
1670pub const OPERATOR_SID_PIN_HINT_KEY: &str = "operator_sid_pin";
1674
1675fn merge_operator_pin_hint(
1682 hint: Option<&Value>,
1683 pin: &str,
1684 agent: &str,
1685) -> Result<Value, CompileError> {
1686 let mut object = match hint {
1687 None => serde_json::Map::new(),
1688 Some(Value::Object(map)) => map.clone(),
1689 Some(other) => {
1690 return Err(CompileError::InvalidSpec {
1691 name: agent.to_string(),
1692 msg: format!(
1693 "hints.per_agent['{agent}'] must be a JSON object to carry the \
1694 run-scoped operator pin (got {other})"
1695 ),
1696 });
1697 }
1698 };
1699 object.insert(
1700 OPERATOR_SID_PIN_HINT_KEY.to_string(),
1701 Value::String(pin.to_string()),
1702 );
1703 Ok(Value::Object(object))
1704}
1705
1706impl SubprocessProcessSpawnerFactory {
1707 fn build_embed(
1712 agent_def: &AgentDef,
1713 template: &Value,
1714 overrides: Option<&Value>,
1715 ) -> Result<ProcessSpawner, CompileError> {
1716 use crate::worker::process_spawner::EmbedTemplate;
1717 use mlua_swarm_schema::{SubprocessDef, SubprocessOverrides};
1718
1719 let agent_name = &agent_def.name;
1720 let invalid = |msg: String| CompileError::InvalidSpec {
1721 name: agent_name.to_string(),
1722 msg,
1723 };
1724 let def: SubprocessDef = serde_json::from_value(template.clone())
1725 .map_err(|e| invalid(format!("subprocess_template hint: {e}")))?;
1726 let overrides: SubprocessOverrides = match overrides {
1727 Some(v) => serde_json::from_value(v.clone())
1728 .map_err(|e| invalid(format!("subprocess_overrides hint: {e}")))?,
1729 None => SubprocessOverrides::default(),
1730 };
1731
1732 if def.argv.is_empty() {
1733 return Err(invalid(format!(
1734 "SubprocessDef '{}': argv must not be empty",
1735 def.name
1736 )));
1737 }
1738 for (i, a) in def.argv.iter().enumerate() {
1740 validate_embed_placeholders(a, &format!("argv[{i}]")).map_err(&invalid)?;
1741 }
1742 if let Some(stdin) = &def.stdin {
1743 validate_embed_placeholders(stdin, "stdin").map_err(&invalid)?;
1744 }
1745 for (k, v) in &def.env {
1746 validate_embed_placeholders(v, &format!("env['{k}']")).map_err(&invalid)?;
1747 }
1748 if let Some(cwd) = &def.cwd {
1749 validate_embed_placeholders(cwd, "cwd").map_err(&invalid)?;
1750 }
1751 let stream_mode = match def.stream_mode.as_deref() {
1752 Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
1753 Some("sse_events") => Some(StreamMode::SseEvents),
1754 Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
1755 Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
1756 None => None,
1757 };
1758 if let Some(output) = &def.output {
1759 if stream_mode.is_some() {
1760 return Err(invalid(format!(
1761 "SubprocessDef '{}': output normalization is a plain-mode \
1762 declaration; remove either `output` or `stream_mode`",
1763 def.name
1764 )));
1765 }
1766 if let Some(format) = output.format.as_deref() {
1767 if format != "json" {
1768 return Err(invalid(format!(
1769 "SubprocessDef '{}': unknown output.format '{format}' \
1770 (supported: \"json\")",
1771 def.name
1772 )));
1773 }
1774 }
1775 if let Some(ptr) = output.result_ptr.as_deref() {
1776 if !ptr.starts_with('/') {
1777 return Err(invalid(format!(
1778 "SubprocessDef '{}': output.result_ptr '{ptr}' is not a \
1779 JSON Pointer (RFC 6901 — must start with '/')",
1780 def.name
1781 )));
1782 }
1783 }
1784 if let Some(ok_from) = output.ok_from.as_deref() {
1785 if ok_from != "exit_code" && !ok_from.starts_with('/') {
1786 return Err(invalid(format!(
1787 "SubprocessDef '{}': output.ok_from '{ok_from}' must be \
1788 \"exit_code\" or a JSON Pointer (starting with '/')",
1789 def.name
1790 )));
1791 }
1792 }
1793 }
1794
1795 let profile = agent_def.profile.as_ref();
1798 let system_prompt = profile
1799 .map(|p| p.system_prompt.clone())
1800 .filter(|s| !s.is_empty());
1801 let model = overrides
1802 .model
1803 .clone()
1804 .or_else(|| profile.and_then(|p| p.model.clone()));
1805 let tools: Vec<String> = if overrides.tools.is_empty() {
1806 profile.map(|p| p.tools.clone()).unwrap_or_default()
1807 } else {
1808 overrides.tools.clone()
1809 };
1810 let cwd = overrides.cwd.clone().or_else(|| def.cwd.clone());
1812 if let Some(c) = &cwd {
1813 validate_embed_placeholders(c, "overrides.cwd").map_err(&invalid)?;
1814 }
1815
1816 let program = def.argv[0].clone();
1817 let sp = ProcessSpawner {
1818 program,
1819 args: Vec::new(),
1820 use_stdin: def.stdin.is_some(),
1821 stream_mode,
1822 embed: Some(EmbedTemplate {
1823 argv: def.argv,
1824 stdin: def.stdin,
1825 env: def.env,
1826 cwd,
1827 output: def.output,
1828 system_prompt,
1829 model,
1830 tools_csv: tools.join(","),
1831 }),
1832 };
1833 Ok(sp)
1834 }
1835}
1836
1837impl SpawnerFactory for SubprocessProcessSpawnerFactory {
1838 fn build(
1839 &self,
1840 agent_def: &AgentDef,
1841 hint: Option<&Value>,
1842 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1843 if let Some(template) = hint.and_then(|h| h.get(SUBPROCESS_TEMPLATE_HINT_KEY)) {
1847 let overrides = hint.and_then(|h| h.get(SUBPROCESS_OVERRIDES_HINT_KEY));
1848 return Self::build_embed(agent_def, template, overrides).map(|sp| {
1849 let arc: Arc<dyn SpawnerAdapter> = Arc::new(sp);
1850 arc
1851 });
1852 }
1853 let agent_name = &agent_def.name;
1854 let spec = &agent_def.spec;
1855 let invalid = |msg: String| CompileError::InvalidSpec {
1856 name: agent_name.to_string(),
1857 msg,
1858 };
1859 let program = spec
1860 .get("program")
1861 .and_then(|v| v.as_str())
1862 .ok_or_else(|| invalid("shell spec: 'program' (string) required".into()))?
1863 .to_string();
1864 let args: Vec<String> = spec
1865 .get("args")
1866 .and_then(|v| v.as_array())
1867 .map(|a| {
1868 a.iter()
1869 .filter_map(|x| x.as_str().map(|s| s.to_string()))
1870 .collect()
1871 })
1872 .unwrap_or_default();
1873 let use_stdin = spec
1874 .get("use_stdin")
1875 .and_then(|v| v.as_bool())
1876 .unwrap_or(true);
1877 let stream_mode = match spec.get("stream_mode").and_then(|v| v.as_str()) {
1878 Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
1879 Some("sse_events") => Some(StreamMode::SseEvents),
1880 Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
1881 Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
1882 None => None,
1883 };
1884
1885 let mut sp = ProcessSpawner {
1886 program,
1887 args,
1888 use_stdin,
1889 stream_mode,
1890 embed: None,
1891 };
1892 if let Some(mode) = sp.stream_mode.clone() {
1893 sp = sp.stream_mode(mode);
1894 }
1895 Ok(Arc::new(sp))
1896 }
1897}
1898
1899pub struct LuaInProcessSpawnerFactory {
1930 registry: HashMap<String, WorkerFn>,
1931 bridges: HashMap<String, HostBridge>,
1932}
1933
1934#[derive(Clone)]
1946pub struct HostBridge(
1947 Arc<dyn Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync>,
1948);
1949
1950impl HostBridge {
1951 pub fn new<F>(f: F) -> Self
1953 where
1954 F: Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync + 'static,
1955 {
1956 Self(Arc::new(f))
1957 }
1958
1959 pub fn call(&self, arg: serde_json::Value) -> Result<serde_json::Value, String> {
1963 (self.0)(arg)
1964 }
1965}
1966
1967#[derive(Clone)]
1974pub struct LuaScriptSource {
1975 pub source: String,
1977 pub label: String,
1980}
1981
1982impl LuaScriptSource {
1983 pub fn new(source: impl Into<String>, label: impl Into<String>) -> Self {
1985 Self {
1986 source: source.into(),
1987 label: label.into(),
1988 }
1989 }
1990}
1991
1992impl LuaInProcessSpawnerFactory {
1993 pub fn new() -> Self {
1995 Self {
1996 registry: HashMap::new(),
1997 bridges: HashMap::new(),
1998 }
1999 }
2000
2001 pub fn with_bridge(mut self, name: impl Into<String>, bridge: HostBridge) -> Self {
2008 self.bridges.insert(name.into(), bridge);
2009 self
2010 }
2011
2012 pub fn register_lua(mut self, fn_id: impl Into<String>, source: LuaScriptSource) -> Self {
2030 let source = Arc::new(source);
2031 let bridges = Arc::new(self.bridges.clone());
2032 let wrapped: WorkerFn = Arc::new(move |inv| {
2033 let source = source.clone();
2034 let bridges = bridges.clone();
2035 Box::pin(run_lua_worker(source, bridges, inv))
2036 });
2037 self.registry.insert(fn_id.into(), wrapped);
2038 self
2039 }
2040}
2041
2042async fn run_lua_worker(
2044 source: Arc<LuaScriptSource>,
2045 bridges: Arc<HashMap<String, HostBridge>>,
2046 inv: crate::worker::adapter::WorkerInvocation,
2047) -> Result<crate::worker::adapter::WorkerResult, crate::worker::adapter::WorkerError> {
2048 use crate::worker::adapter::WorkerError;
2049 use mlua::LuaSerdeExt;
2050
2051 let label = source.label.clone();
2052 let outcome =
2053 tokio::task::spawn_blocking(move || -> Result<(serde_json::Value, bool), String> {
2054 let lua = mlua::Lua::new();
2055 let g = lua.globals();
2056
2057 g.set("_PROMPT", inv.prompt.clone())
2059 .map_err(|e| format!("set _PROMPT: {e}"))?;
2060 g.set("_AGENT", inv.agent.clone())
2061 .map_err(|e| format!("set _AGENT: {e}"))?;
2062 g.set("_TASK_ID", inv.task_id.to_string())
2063 .map_err(|e| format!("set _TASK_ID: {e}"))?;
2064 g.set("_ATTEMPT", inv.attempt as i64)
2065 .map_err(|e| format!("set _ATTEMPT: {e}"))?;
2066
2067 for (name, value) in
2076 crate::worker::agent_block::runtime::context_globals(inv.context.as_ref())
2077 {
2078 let lua_val = lua
2079 .to_value(&value)
2080 .map_err(|e| format!("{name} to_value: {e}"))?;
2081 g.set(name.as_str(), lua_val)
2082 .map_err(|e| format!("set {name}: {e}"))?;
2083 }
2084
2085 if let Ok(json_val) = serde_json::from_str::<serde_json::Value>(&inv.prompt) {
2087 let lua_val = lua
2088 .to_value(&json_val)
2089 .map_err(|e| format!("_CTX to_value: {e}"))?;
2090 g.set("_CTX", lua_val)
2091 .map_err(|e| format!("set _CTX: {e}"))?;
2092 }
2093
2094 if !bridges.is_empty() {
2096 let host = lua
2097 .create_table()
2098 .map_err(|e| format!("create host table: {e}"))?;
2099 for (name, bridge) in bridges.iter() {
2100 let bridge = bridge.clone();
2101 let bname = name.clone();
2102 let f = lua
2103 .create_function(move |lua, arg: mlua::Value| {
2104 let json_arg: serde_json::Value = lua.from_value(arg).map_err(|e| {
2105 mlua::Error::external(format!("bridge {bname} arg → json: {e}"))
2106 })?;
2107 let result_json =
2108 bridge.call(json_arg).map_err(mlua::Error::external)?;
2109 lua.to_value(&result_json).map_err(|e| {
2110 mlua::Error::external(format!("bridge {bname} ret → lua: {e}"))
2111 })
2112 })
2113 .map_err(|e| format!("create_function {name}: {e}"))?;
2114 host.set(name.as_str(), f)
2115 .map_err(|e| format!("host.{name} set: {e}"))?;
2116 }
2117 g.set("host", host).map_err(|e| format!("set host: {e}"))?;
2118 }
2119
2120 let result: mlua::Value = lua
2122 .load(&source.source)
2123 .set_name(&source.label)
2124 .eval()
2125 .map_err(|e| format!("lua eval [{}]: {e}", source.label))?;
2126
2127 let json_result: serde_json::Value = lua
2129 .from_value(result)
2130 .map_err(|e| format!("lua → json [{}]: {e}", source.label))?;
2131
2132 let (value, ok) = match &json_result {
2133 serde_json::Value::Object(map)
2134 if map.contains_key("value") || map.contains_key("ok") =>
2135 {
2136 let ok = map.get("ok").and_then(|v| v.as_bool()).unwrap_or(true);
2137 let value = map.get("value").cloned().unwrap_or(json_result.clone());
2138 (value, ok)
2139 }
2140 _ => (json_result, true),
2141 };
2142 Ok((value, ok))
2143 })
2144 .await
2145 .map_err(|e| WorkerError::Failed(format!("spawn_blocking join [{label}]: {e}")))?
2146 .map_err(WorkerError::Failed)?;
2147
2148 Ok(crate::worker::adapter::WorkerResult {
2149 value: outcome.0,
2150 ok: outcome.1,
2151 stats: None,
2152 }
2153 .ensure_worker_kind("lua"))
2154}
2155
2156impl Default for LuaInProcessSpawnerFactory {
2157 fn default() -> Self {
2158 Self::new()
2159 }
2160}
2161
2162impl SpawnerFactoryKind for LuaInProcessSpawnerFactory {
2163 const KIND: AgentKind = AgentKind::Lua;
2164 type Worker = LuaWorker;
2165}
2166
2167impl SpawnerFactory for LuaInProcessSpawnerFactory {
2168 fn build(
2169 &self,
2170 agent_def: &AgentDef,
2171 _hint: Option<&Value>,
2172 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2173 if let Some(source) = agent_def.spec.get("source").and_then(|v| v.as_str()) {
2179 let label = agent_def
2180 .spec
2181 .get("label")
2182 .and_then(|v| v.as_str())
2183 .map(str::to_string)
2184 .unwrap_or_else(|| format!("{}.lua", agent_def.name));
2185 let script = Arc::new(LuaScriptSource::new(source.to_string(), label));
2186 let bridges = Arc::new(self.bridges.clone());
2187 let wrapped: WorkerFn = Arc::new(move |inv| {
2188 let source = script.clone();
2189 let bridges = bridges.clone();
2190 Box::pin(run_lua_worker(source, bridges, inv))
2191 });
2192 let mut sp: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
2193 sp.registry.insert(agent_def.name.to_string(), wrapped);
2194 return Ok(Arc::new(sp));
2195 }
2196 build_inproc_from_registry::<LuaWorker>(&self.registry, agent_def, "lua")
2197 }
2198}
2199
2200pub struct RustFnInProcessSpawnerFactory {
2214 registry: HashMap<String, WorkerFn>,
2215}
2216
2217impl RustFnInProcessSpawnerFactory {
2218 pub fn new() -> Self {
2220 Self {
2221 registry: HashMap::new(),
2222 }
2223 }
2224
2225 pub fn register_fn<F, Fut>(mut self, fn_id: impl Into<String>, f: F) -> Self
2228 where
2229 F: Fn(crate::worker::adapter::WorkerInvocation) -> Fut + Send + Sync + 'static,
2230 Fut: std::future::Future<
2231 Output = Result<
2232 crate::worker::adapter::WorkerResult,
2233 crate::worker::adapter::WorkerError,
2234 >,
2235 > + Send
2236 + 'static,
2237 {
2238 let f = Arc::new(f);
2239 let wrapped: WorkerFn = Arc::new(move |inv| {
2240 let f = f.clone();
2241 Box::pin(f(inv))
2242 });
2243 self.registry.insert(fn_id.into(), wrapped);
2244 self
2245 }
2246}
2247
2248impl Default for RustFnInProcessSpawnerFactory {
2249 fn default() -> Self {
2250 Self::new()
2251 }
2252}
2253
2254impl SpawnerFactoryKind for RustFnInProcessSpawnerFactory {
2255 const KIND: AgentKind = AgentKind::RustFn;
2256 type Worker = RustFnWorker;
2257}
2258
2259impl SpawnerFactory for RustFnInProcessSpawnerFactory {
2260 fn build(
2261 &self,
2262 agent_def: &AgentDef,
2263 _hint: Option<&Value>,
2264 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2265 build_inproc_from_registry::<RustFnWorker>(&self.registry, agent_def, "rust_fn")
2266 }
2267}
2268
2269fn build_inproc_from_registry<W>(
2275 registry: &HashMap<String, WorkerFn>,
2276 agent_def: &AgentDef,
2277 kind_label: &str,
2278) -> Result<Arc<dyn SpawnerAdapter>, CompileError>
2279where
2280 W: crate::worker::Worker + From<crate::worker::WorkerJoinHandler> + Send + Sync + 'static,
2281{
2282 let agent_name = &agent_def.name;
2283 let spec = &agent_def.spec;
2284 let invalid = |msg: String| CompileError::InvalidSpec {
2285 name: agent_name.to_string(),
2286 msg,
2287 };
2288 let fn_id = spec
2289 .get("fn_id")
2290 .and_then(|v| v.as_str())
2291 .ok_or_else(|| invalid(format!("{kind_label} spec: 'fn_id' (string) required")))?;
2292 let f = registry
2293 .get(fn_id)
2294 .cloned()
2295 .ok_or_else(|| invalid(format!("fn_id '{fn_id}' not registered in factory")))?;
2296 let mut sp: InProcSpawner<W> = InProcSpawner::<W>::typed();
2297 sp.registry.insert(agent_name.to_string(), f);
2301 Ok(Arc::new(sp))
2302}
2303
2304pub struct LuaWorker {
2309 pub handler: crate::worker::WorkerJoinHandler,
2311}
2312
2313impl From<crate::worker::WorkerJoinHandler> for LuaWorker {
2314 fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
2315 Self { handler }
2316 }
2317}
2318
2319#[async_trait::async_trait]
2320impl crate::worker::Worker for LuaWorker {
2321 fn id(&self) -> &crate::types::WorkerId {
2322 &self.handler.worker_id
2323 }
2324 fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
2325 self.handler.cancel.clone()
2326 }
2327 async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
2328 self.handler.await_completion().await
2329 }
2330}
2331
2332pub struct RustFnWorker {
2337 pub handler: crate::worker::WorkerJoinHandler,
2339}
2340
2341impl From<crate::worker::WorkerJoinHandler> for RustFnWorker {
2342 fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
2343 Self { handler }
2344 }
2345}
2346
2347#[async_trait::async_trait]
2348impl crate::worker::Worker for RustFnWorker {
2349 fn id(&self) -> &crate::types::WorkerId {
2350 &self.handler.worker_id
2351 }
2352 fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
2353 self.handler.cancel.clone()
2354 }
2355 async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
2356 self.handler.await_completion().await
2357 }
2358}
2359
2360pub struct OperatorSpawnerFactory {
2424 operators: Arc<std::sync::RwLock<HashMap<String, Arc<dyn Operator>>>>,
2425}
2426
2427impl OperatorSpawnerFactory {
2428 pub fn new() -> Self {
2430 Self {
2431 operators: Arc::new(std::sync::RwLock::new(HashMap::new())),
2432 }
2433 }
2434
2435 pub fn register_operator(&self, id: impl Into<String>, op: Arc<dyn Operator>) -> &Self {
2441 self.operators
2442 .write()
2443 .expect("OperatorSpawnerFactory.operators RwLock poisoned")
2444 .insert(id.into(), op);
2445 self
2446 }
2447
2448 pub fn unregister_operator(&self, id: &str) -> &Self {
2451 self.operators
2452 .write()
2453 .expect("OperatorSpawnerFactory.operators RwLock poisoned")
2454 .remove(id);
2455 self
2456 }
2457}
2458
2459impl Default for OperatorSpawnerFactory {
2460 fn default() -> Self {
2461 Self::new()
2462 }
2463}
2464
2465impl SpawnerFactoryKind for OperatorSpawnerFactory {
2466 const KIND: AgentKind = AgentKind::Operator;
2467 type Worker = crate::operator::OperatorWorker;
2468}
2469
2470impl SpawnerFactory for OperatorSpawnerFactory {
2471 fn build(
2472 &self,
2473 agent_def: &AgentDef,
2474 hint: Option<&Value>,
2475 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2476 let agent_name = &agent_def.name;
2477 let spec = &agent_def.spec;
2478 let system_prompt = agent_def.profile.as_ref().map(|p| p.system_prompt.clone());
2484 let invalid = |msg: String| CompileError::InvalidSpec {
2485 name: agent_name.to_string(),
2486 msg,
2487 };
2488 let op_ref = spec
2489 .get("operator_ref")
2490 .and_then(|v| v.as_str())
2491 .ok_or_else(|| invalid("operator spec: 'operator_ref' (string) required".into()))?;
2492 let pin = hint
2499 .and_then(|h| h.get(OPERATOR_SID_PIN_HINT_KEY))
2500 .and_then(|v| v.as_str());
2501 let lookup_key = pin.unwrap_or(op_ref);
2502 let operators = self
2503 .operators
2504 .read()
2505 .expect("OperatorSpawnerFactory.operators RwLock poisoned");
2506 let op = operators.get(lookup_key).cloned().ok_or_else(|| {
2507 let mut names: Vec<String> = operators.keys().cloned().collect();
2508 names.sort();
2509 let names_list = if names.is_empty() {
2510 "<none>".to_string()
2511 } else {
2512 names.join(", ")
2513 };
2514 match pin {
2515 Some(pin) => invalid(format!(
2516 "run-scoped operator pin '{pin}' (declared operator_ref '{op_ref}') \
2517 is not registered in factory. \
2518 Registered ids: [{names_list}]. \
2519 The launch pinned this run to that session; resolving \
2520 '{op_ref}' through whichever session currently holds the \
2521 role would send this run's Spawn frames somewhere the \
2522 caller did not ask for, so the launch fails instead. \
2523 Hint: the pinned session must be joined (and still live) \
2524 before launching."
2525 )),
2526 None => invalid(format!(
2527 "operator_ref '{op_ref}' not registered in factory. \
2528 Registered sids: [{names_list}]. \
2529 Hint: call mse_operator_join(roles=[...]) to mint the sid first."
2530 )),
2531 }
2532 })?;
2533 drop(operators);
2534
2535 let worker_binding = agent_def
2542 .profile
2543 .as_ref()
2544 .and_then(|p| p.worker_binding.as_ref())
2545 .map(|variant| WorkerBinding {
2546 variant: variant.clone(),
2547 tools: agent_def
2548 .profile
2549 .as_ref()
2550 .map(|p| p.tools.clone())
2551 .unwrap_or_default(),
2552 request_digest: None,
2556 requested_model: None,
2557 });
2558 if op.requires_worker_binding() && worker_binding.is_none() {
2559 return Err(invalid(format!(
2566 "{WORKER_BINDING_REQUIRED_MSG_PREFIX}. \
2567 Fix by either: \
2568 (a) if authoring the Blueprint JSON directly, add \
2569 `agents[N].profile.worker_binding: \"<subagent-type>\"` \
2570 to the JSON literal; or \
2571 (b) if using an $agent_md file ref, add \
2572 `worker_binding: <subagent-type>` to the agent .md frontmatter."
2573 )));
2574 }
2575 Ok(Arc::new(OperatorSpawner::new(
2576 op,
2577 system_prompt,
2578 worker_binding,
2579 )))
2580 }
2581}
2582
2583#[cfg(test)]
2584mod operator_spawner_factory_worker_binding_tests {
2585 use super::*;
2586 use crate::blueprint::AgentProfile;
2587 use crate::core::ctx::Ctx;
2588 use crate::types::CapToken;
2589 use crate::worker::adapter::{WorkerError, WorkerResult};
2590
2591 struct StubOperator {
2596 requires_binding: bool,
2597 }
2598
2599 #[async_trait]
2600 impl Operator for StubOperator {
2601 async fn execute(
2602 &self,
2603 _ctx: &Ctx,
2604 _system: Option<String>,
2605 _prompt: Value,
2606 _worker: Option<WorkerBinding>,
2607 _worker_token: CapToken,
2608 ) -> Result<WorkerResult, WorkerError> {
2609 Ok(WorkerResult {
2610 value: Value::Null,
2611 ok: true,
2612 stats: None,
2613 })
2614 }
2615
2616 fn requires_worker_binding(&self) -> bool {
2617 self.requires_binding
2618 }
2619 }
2620
2621 fn agent_def_with(profile: Option<AgentProfile>) -> AgentDef {
2622 AgentDef {
2623 name: "test-agent".to_string(),
2624 kind: AgentKind::Operator,
2625 spec: serde_json::json!({ "operator_ref": "op1" }),
2626 profile,
2627 meta: None,
2628 runner: None,
2629 runner_ref: None,
2630 verdict: None,
2631 }
2632 }
2633
2634 #[test]
2635 fn build_fails_loud_when_binding_required_but_absent() {
2636 let factory = OperatorSpawnerFactory::new();
2637 factory.register_operator(
2638 "op1",
2639 Arc::new(StubOperator {
2640 requires_binding: true,
2641 }) as Arc<dyn Operator>,
2642 );
2643 let def = agent_def_with(Some(AgentProfile::default()));
2644 match factory.build(&def, None) {
2645 Err(CompileError::InvalidSpec { name, msg }) => {
2646 assert_eq!(name, "test-agent");
2647 assert!(
2648 msg.contains("worker_binding is required"),
2649 "unexpected message: {msg}"
2650 );
2651 assert!(
2655 msg.contains("agents[N].profile.worker_binding"),
2656 "message missing JSON-direct hint (issue #9): {msg}"
2657 );
2658 assert!(
2659 msg.contains("agent .md frontmatter"),
2660 "message missing $agent_md hint: {msg}"
2661 );
2662 }
2663 Err(other) => panic!("expected InvalidSpec, got: {other:?}"),
2664 Ok(_) => panic!("expected compile-time failure, got Ok"),
2665 }
2666 }
2667
2668 #[test]
2675 fn factory_error_message_carries_the_shared_prefix_and_specializes_the_diagnostic() {
2676 let factory = OperatorSpawnerFactory::new();
2677 factory.register_operator(
2678 "op1",
2679 Arc::new(StubOperator {
2680 requires_binding: true,
2681 }) as Arc<dyn Operator>,
2682 );
2683 let def = agent_def_with(Some(AgentProfile::default()));
2684 let err = match factory.build(&def, None) {
2685 Err(err) => err,
2686 Ok(_) => panic!("expected compile-time failure, got Ok"),
2687 };
2688 match &err {
2689 CompileError::InvalidSpec { msg, .. } => {
2690 assert!(
2691 msg.starts_with(WORKER_BINDING_REQUIRED_MSG_PREFIX),
2692 "factory message must start with the shared prefix, got: {msg}"
2693 );
2694 }
2695 other => panic!("expected InvalidSpec, got: {other:?}"),
2696 }
2697 let d = mlua_swarm_diag::Diagnostic::from(&err);
2698 assert_eq!(d.kind, "worker-binding-missing");
2699 }
2700
2701 #[test]
2702 fn build_succeeds_when_binding_required_and_present() {
2703 let factory = OperatorSpawnerFactory::new();
2704 factory.register_operator(
2705 "op1",
2706 Arc::new(StubOperator {
2707 requires_binding: true,
2708 }) as Arc<dyn Operator>,
2709 );
2710 let profile = AgentProfile {
2711 worker_binding: Some("code-worker".to_string()),
2712 tools: vec!["Read".to_string(), "Edit".to_string()],
2713 ..Default::default()
2714 };
2715 let def = agent_def_with(Some(profile));
2716 assert!(
2717 factory.build(&def, None).is_ok(),
2718 "expected Ok when worker_binding is declared"
2719 );
2720 }
2721
2722 #[test]
2723 fn build_succeeds_when_binding_not_required_and_absent() {
2724 let factory = OperatorSpawnerFactory::new();
2725 factory.register_operator(
2726 "op1",
2727 Arc::new(StubOperator {
2728 requires_binding: false,
2729 }) as Arc<dyn Operator>,
2730 );
2731 let def = agent_def_with(Some(AgentProfile::default()));
2732 assert!(
2733 factory.build(&def, None).is_ok(),
2734 "backends that don't require a binding must not be gated by its absence"
2735 );
2736 }
2737}
2738
2739#[cfg(test)]
2747mod lua_inline_source_tests {
2748 use super::*;
2749 use crate::types::{CapToken, Role, StepId};
2750
2751 fn agent(name: &str, spec: Value) -> AgentDef {
2752 AgentDef {
2753 name: name.to_string(),
2754 kind: AgentKind::Lua,
2755 spec,
2756 profile: None,
2757 meta: None,
2758 runner: None,
2759 runner_ref: None,
2760 verdict: None,
2761 }
2762 }
2763
2764 fn test_invocation(prompt: &str) -> crate::worker::adapter::WorkerInvocation {
2765 crate::worker::adapter::WorkerInvocation::new(
2766 CapToken {
2767 agent_id: "a".into(),
2768 role: Role::Worker,
2769 scopes: vec!["*".into()],
2770 issued_at: 0,
2771 expire_at: u64::MAX / 2,
2772 max_uses: None,
2773 nonce: "test-nonce".into(),
2774 sig_hex: "".into(),
2775 },
2776 StepId::parse("ST-test").expect("StepId parse"),
2777 1,
2778 "g",
2779 prompt,
2780 )
2781 }
2782
2783 #[test]
2784 fn build_accepts_inline_source_without_pre_registration() {
2785 let factory = LuaInProcessSpawnerFactory::new();
2786 let def = agent(
2787 "g",
2788 serde_json::json!({ "source": "return { value = 42, ok = true }" }),
2789 );
2790 assert!(
2791 factory.build(&def, None).is_ok(),
2792 "inline spec.source must build without a pre-registered fn_id"
2793 );
2794 }
2795
2796 #[test]
2797 fn build_rejects_when_neither_source_nor_fn_id_is_present() {
2798 let factory = LuaInProcessSpawnerFactory::new();
2799 let def = agent("g", serde_json::json!({}));
2800 match factory.build(&def, None) {
2801 Err(CompileError::InvalidSpec { msg, .. }) => {
2802 assert!(
2803 msg.contains("fn_id"),
2804 "empty spec must still surface the fn_id-required message: {msg}"
2805 );
2806 }
2807 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
2808 Ok(_) => panic!("expected InvalidSpec, got Ok(SpawnerAdapter)"),
2811 }
2812 }
2813
2814 #[tokio::test]
2818 async fn inline_source_evaluates_and_marshals_result() {
2819 let source =
2820 LuaScriptSource::new("return { value = _PROMPT .. '!', ok = true }", "smoke.lua");
2821 let out = run_lua_worker(
2822 std::sync::Arc::new(source),
2823 std::sync::Arc::new(HashMap::new()),
2824 test_invocation("hello"),
2825 )
2826 .await
2827 .expect("lua worker ok");
2828 assert_eq!(out.value, serde_json::json!("hello!"));
2829 assert!(out.ok);
2830 }
2831
2832 #[tokio::test]
2833 async fn inline_source_can_signal_agent_level_failure() {
2834 let source = LuaScriptSource::new("return { value = 'nope', ok = false }", "gate.lua");
2837 let out = run_lua_worker(
2838 std::sync::Arc::new(source),
2839 std::sync::Arc::new(HashMap::new()),
2840 test_invocation("input"),
2841 )
2842 .await
2843 .expect("lua worker ok");
2844 assert_eq!(out.value, serde_json::json!("nope"));
2845 assert!(!out.ok);
2846 }
2847}
2848
2849#[cfg(test)]
2852mod meta_ref_validation_tests {
2853 use super::*;
2854 use crate::blueprint::{AgentMeta, MetaDef};
2855 use crate::worker::adapter::WorkerResult;
2856
2857 fn registry_with_echo() -> SpawnerRegistry {
2858 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2859 Ok(WorkerResult {
2860 value: Value::String(inv.prompt),
2861 ok: true,
2862 stats: None,
2863 })
2864 });
2865 let mut reg = SpawnerRegistry::new();
2866 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2867 reg
2868 }
2869
2870 fn rustfn_agent(name: &str) -> AgentDef {
2871 AgentDef {
2872 name: name.to_string(),
2873 kind: AgentKind::RustFn,
2874 spec: serde_json::json!({ "fn_id": "echo" }),
2875 profile: None,
2876 meta: None,
2877 runner: None,
2878 runner_ref: None,
2879 verdict: None,
2880 }
2881 }
2882
2883 fn simple_flow(agent_ref: &str, in_: Expr) -> FlowNode {
2884 FlowNode::Step {
2885 ref_: agent_ref.to_string(),
2886 in_,
2887 out: Expr::Path {
2888 at: "$.output".parse().expect("literal test path: $.output"),
2889 },
2890 }
2891 }
2892
2893 fn minimal_bp(agents: Vec<AgentDef>, metas: Vec<MetaDef>, flow: FlowNode) -> Blueprint {
2894 Blueprint {
2895 schema_version: crate::blueprint::current_schema_version(),
2896 id: "meta-ref-ut".into(),
2897 flow,
2898 agents,
2899 operators: vec![],
2900 metas,
2901 hints: Default::default(),
2902 strategy: Default::default(),
2903 metadata: BlueprintMetadata::default(),
2904 spawner_hints: Default::default(),
2905 default_agent_kind: AgentKind::Operator,
2906 default_operator_kind: None,
2907 default_init_ctx: None,
2908 default_agent_ctx: None,
2909 default_context_policy: None,
2910 projection_placement: None,
2911 audits: vec![],
2912 degradation_policy: None,
2913 runners: vec![],
2914 default_runner: None,
2915 subprocesses: vec![],
2916 check_policy: None,
2917 blueprint_ref_includes: Vec::new(),
2918 }
2919 }
2920
2921 #[test]
2922 fn valid_meta_ref_compiles() {
2923 let mut agent = rustfn_agent("worker");
2924 agent.meta = Some(AgentMeta {
2925 meta_ref: Some("shared".to_string()),
2926 ..Default::default()
2927 });
2928 let bp = minimal_bp(
2929 vec![agent],
2930 vec![MetaDef {
2931 name: "shared".into(),
2932 ctx: serde_json::json!({ "k": "v" }),
2933 }],
2934 simple_flow(
2935 "worker",
2936 Expr::Path {
2937 at: "$.input".parse().expect("literal test path: $.input"),
2938 },
2939 ),
2940 );
2941 let compiler = Compiler::new(registry_with_echo());
2942 assert!(
2943 compiler.compile(&bp).is_ok(),
2944 "a resolvable AgentMeta.meta_ref must compile"
2945 );
2946 }
2947
2948 #[test]
2949 fn unknown_agent_meta_ref_is_unresolved_meta_ref() {
2950 let mut agent = rustfn_agent("worker");
2951 agent.meta = Some(AgentMeta {
2952 meta_ref: Some("missing".to_string()),
2953 ..Default::default()
2954 });
2955 let bp = minimal_bp(
2956 vec![agent],
2957 vec![],
2958 simple_flow(
2959 "worker",
2960 Expr::Path {
2961 at: "$.input".parse().expect("literal test path: $.input"),
2962 },
2963 ),
2964 );
2965 let compiler = Compiler::new(registry_with_echo());
2966 match compiler.compile(&bp) {
2967 Err(CompileError::UnresolvedMetaRef {
2968 where_,
2969 meta_ref,
2970 defined,
2971 }) => {
2972 assert!(
2973 where_.contains("worker"),
2974 "where_ must name the agent: {where_}"
2975 );
2976 assert_eq!(meta_ref, "missing");
2977 assert!(defined.is_empty());
2978 }
2979 Err(other) => {
2980 panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
2981 }
2982 Ok(_) => panic!("expected compile-time failure, got Ok"),
2983 }
2984 }
2985
2986 #[test]
2987 fn unknown_static_step_meta_ref_in_lit_is_unresolved_meta_ref() {
2988 let agent = rustfn_agent("worker");
2989 let in_ = Expr::Lit {
2990 value: serde_json::json!({ "$step_meta": { "ref": "missing" }, "$in": "go" }),
2991 };
2992 let bp = minimal_bp(vec![agent], vec![], simple_flow("worker", in_));
2993 let compiler = Compiler::new(registry_with_echo());
2994 match compiler.compile(&bp) {
2995 Err(CompileError::UnresolvedMetaRef {
2996 where_, meta_ref, ..
2997 }) => {
2998 assert!(
2999 where_.contains("worker"),
3000 "where_ must name the offending step: {where_}"
3001 );
3002 assert_eq!(meta_ref, "missing");
3003 }
3004 Err(other) => {
3005 panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
3006 }
3007 Ok(_) => panic!("expected compile-time failure, got Ok"),
3008 }
3009 }
3010
3011 #[test]
3012 fn path_op_input_with_no_static_envelope_compiles_fine() {
3013 let agent = rustfn_agent("worker");
3014 let bp = minimal_bp(
3015 vec![agent],
3016 vec![],
3017 simple_flow(
3018 "worker",
3019 Expr::Path {
3020 at: "$.input".parse().expect("literal test path: $.input"),
3021 },
3022 ),
3023 );
3024 let compiler = Compiler::new(registry_with_echo());
3025 assert!(
3026 compiler.compile(&bp).is_ok(),
3027 "a non-Lit Step.in must not trigger the best-effort static $step_meta check"
3028 );
3029 }
3030}
3031
3032#[cfg(test)]
3034mod audit_agent_validation_tests {
3035 use super::*;
3036 use crate::worker::adapter::WorkerResult;
3037 use mlua_swarm_schema::{AuditDef, AuditMode};
3038
3039 fn registry_with_echo() -> SpawnerRegistry {
3040 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
3041 Ok(WorkerResult {
3042 value: Value::String(inv.prompt),
3043 ok: true,
3044 stats: None,
3045 })
3046 });
3047 let mut reg = SpawnerRegistry::new();
3048 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
3049 reg
3050 }
3051
3052 fn rustfn_agent(name: &str) -> AgentDef {
3053 AgentDef {
3054 name: name.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 }
3064
3065 fn minimal_bp(agents: Vec<AgentDef>, audits: Vec<AuditDef>) -> Blueprint {
3066 Blueprint {
3067 schema_version: crate::blueprint::current_schema_version(),
3068 id: "audit-ref-ut".into(),
3069 flow: FlowNode::Step {
3070 ref_: "worker".to_string(),
3071 in_: Expr::Path {
3072 at: "$.input".parse().expect("literal test path: $.input"),
3073 },
3074 out: Expr::Path {
3075 at: "$.output".parse().expect("literal test path: $.output"),
3076 },
3077 },
3078 agents,
3079 operators: vec![],
3080 metas: vec![],
3081 hints: Default::default(),
3082 strategy: Default::default(),
3083 metadata: BlueprintMetadata::default(),
3084 spawner_hints: Default::default(),
3085 default_agent_kind: AgentKind::Operator,
3086 default_operator_kind: None,
3087 default_init_ctx: None,
3088 default_agent_ctx: None,
3089 default_context_policy: None,
3090 projection_placement: None,
3091 audits,
3092 degradation_policy: None,
3093 runners: vec![],
3094 default_runner: None,
3095 subprocesses: vec![],
3096 check_policy: None,
3097 blueprint_ref_includes: Vec::new(),
3098 }
3099 }
3100
3101 #[test]
3102 fn unresolved_audit_agent_is_a_loud_compile_error() {
3103 let bp = minimal_bp(
3104 vec![rustfn_agent("worker")],
3105 vec![AuditDef {
3106 agent: "missing-auditor".to_string(),
3107 steps: None,
3108 mode: AuditMode::default(),
3109 }],
3110 );
3111 let compiler = Compiler::new(registry_with_echo());
3112 match compiler.compile(&bp) {
3113 Err(CompileError::UnresolvedAuditAgent { agent, defined }) => {
3114 assert_eq!(agent, "missing-auditor");
3115 assert_eq!(defined, vec!["worker".to_string()]);
3116 }
3117 Err(other) => {
3118 panic!("expected UnresolvedAuditAgent, got a different CompileError: {other}")
3119 }
3120 Ok(_) => panic!("expected compile-time failure, got Ok"),
3121 }
3122 }
3123
3124 #[test]
3125 fn resolved_audit_agent_compiles_fine() {
3126 let bp = minimal_bp(
3127 vec![rustfn_agent("worker"), rustfn_agent("auditor")],
3128 vec![AuditDef {
3129 agent: "auditor".to_string(),
3130 steps: None,
3131 mode: AuditMode::default(),
3132 }],
3133 );
3134 let compiler = Compiler::new(registry_with_echo());
3135 assert!(
3136 compiler.compile(&bp).is_ok(),
3137 "an audits[].agent that names a declared AgentDef must compile"
3138 );
3139 }
3140}
3141
3142#[cfg(test)]
3150mod operator_run_pin_tests {
3151 use super::*;
3152 use crate::core::ctx::Ctx;
3153 use crate::types::CapToken;
3154 use crate::worker::adapter::{WorkerError, WorkerResult};
3155 use std::sync::Mutex;
3156
3157 type Seen = Arc<Mutex<Vec<(String, Option<Value>)>>>;
3159
3160 struct RecordingOperatorFactory {
3164 seen: Seen,
3165 }
3166
3167 impl SpawnerFactory for RecordingOperatorFactory {
3168 fn build(
3169 &self,
3170 agent_def: &AgentDef,
3171 hint: Option<&Value>,
3172 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
3173 self.seen
3174 .lock()
3175 .expect("RecordingOperatorFactory.seen poisoned")
3176 .push((agent_def.name.clone(), hint.cloned()));
3177 let mut spawner: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
3178 let worker: WorkerFn = Arc::new(|_inv| {
3179 Box::pin(async move {
3180 Ok(WorkerResult {
3181 value: Value::Null,
3182 ok: true,
3183 stats: None,
3184 })
3185 })
3186 });
3187 spawner.registry.insert(agent_def.name.clone(), worker);
3188 Ok(Arc::new(spawner))
3189 }
3190 }
3191
3192 impl SpawnerFactoryKind for RecordingOperatorFactory {
3193 const KIND: AgentKind = AgentKind::Operator;
3194 type Worker = crate::operator::OperatorWorker;
3195 }
3196
3197 struct RecordingLuaFactory {
3200 seen: Seen,
3201 }
3202
3203 impl SpawnerFactory for RecordingLuaFactory {
3204 fn build(
3205 &self,
3206 agent_def: &AgentDef,
3207 hint: Option<&Value>,
3208 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
3209 self.seen
3210 .lock()
3211 .expect("RecordingLuaFactory.seen poisoned")
3212 .push((agent_def.name.clone(), hint.cloned()));
3213 let mut spawner: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
3214 let worker: WorkerFn = Arc::new(|_inv| {
3215 Box::pin(async move {
3216 Ok(WorkerResult {
3217 value: Value::Null,
3218 ok: true,
3219 stats: None,
3220 })
3221 })
3222 });
3223 spawner.registry.insert(agent_def.name.clone(), worker);
3224 Ok(Arc::new(spawner))
3225 }
3226 }
3227
3228 impl SpawnerFactoryKind for RecordingLuaFactory {
3229 const KIND: AgentKind = AgentKind::Lua;
3230 type Worker = LuaWorker;
3231 }
3232
3233 fn recording_compiler() -> (Compiler, Seen, Seen) {
3234 let operator_seen: Seen = Arc::new(Mutex::new(Vec::new()));
3235 let lua_seen: Seen = Arc::new(Mutex::new(Vec::new()));
3236 let mut registry = SpawnerRegistry::new();
3237 registry.register::<RecordingOperatorFactory>(Arc::new(RecordingOperatorFactory {
3238 seen: operator_seen.clone(),
3239 }));
3240 registry.register::<RecordingLuaFactory>(Arc::new(RecordingLuaFactory {
3241 seen: lua_seen.clone(),
3242 }));
3243 (Compiler::new(registry), operator_seen, lua_seen)
3244 }
3245
3246 fn bp_with_operator_and_lua_agents() -> Blueprint {
3250 serde_json::from_value(serde_json::json!({
3251 "schema_version": crate::blueprint::current_schema_version(),
3252 "id": "operator-pin-ut",
3253 "flow": {
3254 "kind": "step",
3255 "ref": "planner",
3256 "in": { "op": "path", "at": "$.input" },
3257 "out": { "op": "path", "at": "$.output" }
3258 },
3259 "agents": [
3260 {
3261 "name": "planner",
3262 "kind": "operator",
3263 "spec": { "operator_ref": "main-ai" }
3264 },
3265 {
3266 "name": "scorer",
3267 "kind": "lua",
3268 "spec": { "source": "return { value = 1, ok = true }" }
3269 }
3270 ],
3271 "operators": [{ "name": "main-ai" }],
3272 "hints": { "per_agent": { "planner": { "authored": "keep-me" } } },
3273 "strategy": { "strict_refs": false }
3274 }))
3275 .expect("test Blueprint literal")
3276 }
3277
3278 fn hint_for(seen: &Seen, agent: &str) -> Option<Value> {
3279 seen.lock()
3280 .expect("seen poisoned")
3281 .iter()
3282 .find(|(name, _)| name == agent)
3283 .map(|(_, hint)| hint.clone())
3284 .expect("agent was never built")
3285 }
3286
3287 #[test]
3290 fn pinned_compile_hands_the_operator_factory_the_pinned_sid() {
3291 let (compiler, operator_seen, _lua_seen) = recording_compiler();
3292 let bp = bp_with_operator_and_lua_agents();
3293 let bound = resolve_bound_agents(&bp).expect("resolve bound agents");
3294 compiler
3295 .compile_bound_pinned(&bp, &bound, Some("S-pinned"))
3296 .expect("pinned compile");
3297
3298 let hint = hint_for(&operator_seen, "planner").expect("planner hint");
3299 assert_eq!(
3300 hint.get(OPERATOR_SID_PIN_HINT_KEY),
3301 Some(&Value::String("S-pinned".to_string())),
3302 "the pin must reach the factory as a build hint: {hint}"
3303 );
3304 assert_eq!(
3305 hint.get("authored"),
3306 Some(&Value::String("keep-me".to_string())),
3307 "merging the pin must not drop the author's declared hint: {hint}"
3308 );
3309 }
3310
3311 #[test]
3315 fn unpinned_compile_passes_the_authored_hint_through_untouched() {
3316 let (compiler, operator_seen, lua_seen) = recording_compiler();
3317 let bp = bp_with_operator_and_lua_agents();
3318 let bound = resolve_bound_agents(&bp).expect("resolve bound agents");
3319 compiler
3320 .compile_bound(&bp, &bound)
3321 .expect("unpinned compile");
3322
3323 assert_eq!(
3324 hint_for(&operator_seen, "planner"),
3325 Some(serde_json::json!({ "authored": "keep-me" })),
3326 "an unpinned compile must hand over the authored hint verbatim"
3327 );
3328 assert_eq!(
3329 hint_for(&lua_seen, "scorer"),
3330 None,
3331 "an agent with no authored hint must still be built with None"
3332 );
3333 }
3334
3335 #[test]
3339 fn pin_does_not_leak_onto_non_operator_agents() {
3340 let (compiler, _operator_seen, lua_seen) = recording_compiler();
3341 let bp = bp_with_operator_and_lua_agents();
3342 let bound = resolve_bound_agents(&bp).expect("resolve bound agents");
3343 compiler
3344 .compile_bound_pinned(&bp, &bound, Some("S-pinned"))
3345 .expect("pinned compile");
3346
3347 assert_eq!(
3348 hint_for(&lua_seen, "scorer"),
3349 None,
3350 "a non-Operator agent must not receive the operator pin hint"
3351 );
3352 }
3353
3354 #[test]
3358 fn non_object_authored_hint_fails_the_pinned_compile() {
3359 let (compiler, _operator_seen, _lua_seen) = recording_compiler();
3360 let mut bp = bp_with_operator_and_lua_agents();
3361 bp.hints
3362 .per_agent
3363 .insert("planner".to_string(), Value::String("not-an-object".into()));
3364 let bound = resolve_bound_agents(&bp).expect("resolve bound agents");
3365 match compiler.compile_bound_pinned(&bp, &bound, Some("S-pinned")) {
3366 Err(CompileError::InvalidSpec { name, msg }) => {
3367 assert_eq!(name, "planner");
3368 assert!(
3369 msg.contains("run-scoped operator pin"),
3370 "message must explain why the hint blocks the pin: {msg}"
3371 );
3372 }
3373 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
3374 Ok(_) => panic!("expected the non-object hint to fail the pinned compile"),
3375 }
3376 assert!(
3378 compiler.compile_bound(&bp, &bound).is_ok(),
3379 "an unpinned compile must keep accepting whatever hint shape the author declared"
3380 );
3381 }
3382
3383 struct StubOperator {
3389 requires_binding: bool,
3390 }
3391
3392 #[async_trait]
3393 impl Operator for StubOperator {
3394 async fn execute(
3395 &self,
3396 _ctx: &Ctx,
3397 _system: Option<String>,
3398 _prompt: Value,
3399 _worker: Option<WorkerBinding>,
3400 _worker_token: CapToken,
3401 ) -> Result<WorkerResult, WorkerError> {
3402 Ok(WorkerResult {
3403 value: Value::Null,
3404 ok: true,
3405 stats: None,
3406 })
3407 }
3408
3409 fn requires_worker_binding(&self) -> bool {
3410 self.requires_binding
3411 }
3412 }
3413
3414 fn operator_agent() -> AgentDef {
3415 AgentDef {
3416 name: "planner".to_string(),
3417 kind: AgentKind::Operator,
3418 spec: serde_json::json!({ "operator_ref": "main-ai" }),
3419 profile: None,
3420 meta: None,
3421 runner: None,
3422 runner_ref: None,
3423 verdict: None,
3424 }
3425 }
3426
3427 fn pin_hint(sid: &str) -> Value {
3428 serde_json::json!({ OPERATOR_SID_PIN_HINT_KEY: sid })
3429 }
3430
3431 #[test]
3435 fn pin_resolves_the_pinned_session_not_the_role_holder() {
3436 let factory = OperatorSpawnerFactory::new();
3437 factory.register_operator(
3442 "main-ai",
3443 Arc::new(StubOperator {
3444 requires_binding: true,
3445 }) as Arc<dyn Operator>,
3446 );
3447 factory.register_operator(
3448 "S-pinned",
3449 Arc::new(StubOperator {
3450 requires_binding: false,
3451 }) as Arc<dyn Operator>,
3452 );
3453 assert!(
3454 factory
3455 .build(&operator_agent(), Some(&pin_hint("S-pinned")))
3456 .is_ok(),
3457 "the pinned session must resolve the spawner, not the role's holder"
3458 );
3459 assert!(
3462 factory.build(&operator_agent(), None).is_err(),
3463 "without a pin the role's holder must still be the resolution"
3464 );
3465 }
3466
3467 #[test]
3470 fn pin_miss_fails_loud_and_never_falls_back_to_the_role() {
3471 let factory = OperatorSpawnerFactory::new();
3472 factory.register_operator(
3473 "main-ai",
3474 Arc::new(StubOperator {
3475 requires_binding: false,
3476 }) as Arc<dyn Operator>,
3477 );
3478 match factory.build(&operator_agent(), Some(&pin_hint("S-gone"))) {
3479 Err(CompileError::InvalidSpec { name, msg }) => {
3480 assert_eq!(name, "planner");
3481 assert!(msg.contains("S-gone"), "message must name the pin: {msg}");
3482 assert!(
3483 msg.contains("main-ai"),
3484 "message must name the declared role it refused to fall back to: {msg}"
3485 );
3486 }
3487 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
3488 Ok(_) => panic!(
3489 "a pin naming no live session must fail the launch, not silently resolve the role"
3490 ),
3491 }
3492 }
3493
3494 #[test]
3497 fn unpinned_build_keeps_the_historical_role_lookup_and_message() {
3498 let factory = OperatorSpawnerFactory::new();
3499 match factory.build(&operator_agent(), None) {
3500 Err(CompileError::InvalidSpec { msg, .. }) => {
3501 assert!(
3502 msg.contains("operator_ref 'main-ai' not registered in factory"),
3503 "unpinned message must stay the historical one: {msg}"
3504 );
3505 assert!(
3506 !msg.contains("run-scoped operator pin"),
3507 "an unpinned failure must not mention the pin: {msg}"
3508 );
3509 }
3510 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
3511 Ok(_) => panic!("an unregistered role must still fail"),
3512 }
3513 }
3514}
3515
3516#[cfg(test)]
3519mod projection_placement_compile_tests {
3520 use super::*;
3521 use crate::core::projection_placement::{ProjectionPlacement, RootPreference};
3522 use crate::worker::adapter::WorkerResult;
3523 use mlua_swarm_schema::ProjectionPlacementSpec;
3524
3525 fn registry_with_echo() -> SpawnerRegistry {
3526 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
3527 Ok(WorkerResult {
3528 value: Value::String(inv.prompt),
3529 ok: true,
3530 stats: None,
3531 })
3532 });
3533 let mut reg = SpawnerRegistry::new();
3534 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
3535 reg
3536 }
3537
3538 fn minimal_bp(projection_placement: Option<ProjectionPlacementSpec>) -> Blueprint {
3539 Blueprint {
3540 schema_version: crate::blueprint::current_schema_version(),
3541 id: "projection-placement-ut".into(),
3542 flow: FlowNode::Step {
3543 ref_: "worker".to_string(),
3544 in_: Expr::Path {
3545 at: "$.input".parse().expect("literal test path: $.input"),
3546 },
3547 out: Expr::Path {
3548 at: "$.output".parse().expect("literal test path: $.output"),
3549 },
3550 },
3551 agents: vec![AgentDef {
3552 name: "worker".to_string(),
3553 kind: AgentKind::RustFn,
3554 spec: serde_json::json!({ "fn_id": "echo" }),
3555 profile: None,
3556 meta: None,
3557 runner: None,
3558 runner_ref: None,
3559 verdict: None,
3560 }],
3561 operators: vec![],
3562 metas: vec![],
3563 hints: Default::default(),
3564 strategy: Default::default(),
3565 metadata: BlueprintMetadata::default(),
3566 spawner_hints: Default::default(),
3567 default_agent_kind: AgentKind::Operator,
3568 default_operator_kind: None,
3569 default_init_ctx: None,
3570 default_agent_ctx: None,
3571 default_context_policy: None,
3572 projection_placement,
3573 audits: vec![],
3574 degradation_policy: None,
3575 runners: vec![],
3576 default_runner: None,
3577 subprocesses: vec![],
3578 check_policy: None,
3579 blueprint_ref_includes: Vec::new(),
3580 }
3581 }
3582
3583 #[test]
3584 fn undeclared_projection_placement_compiles_to_byte_compat_default() {
3585 let bp = minimal_bp(None);
3586 let compiled = Compiler::new(registry_with_echo())
3587 .compile(&bp)
3588 .expect("undeclared projection_placement compiles");
3589 assert_eq!(
3590 *compiled.projection_placement,
3591 ProjectionPlacement::default()
3592 );
3593 }
3594
3595 #[test]
3596 fn declared_valid_projection_placement_compiles_to_matching_resolver() {
3597 let bp = minimal_bp(Some(ProjectionPlacementSpec {
3598 root: Some("project_root".to_string()),
3599 dir_template: Some("custom/{task_id}/out".to_string()),
3600 }));
3601 let compiled = Compiler::new(registry_with_echo())
3602 .compile(&bp)
3603 .expect("valid projection_placement compiles");
3604 assert_eq!(
3605 compiled.projection_placement.root_preference,
3606 RootPreference::ProjectRoot
3607 );
3608 assert_eq!(
3609 compiled.projection_placement.dir_template,
3610 "custom/{task_id}/out"
3611 );
3612 }
3613
3614 #[test]
3615 fn declared_invalid_dir_template_rejects_compile() {
3616 let bp = minimal_bp(Some(ProjectionPlacementSpec {
3617 root: None,
3618 dir_template: Some("workspace/tasks/ctx".to_string()), }));
3620 match Compiler::new(registry_with_echo()).compile(&bp) {
3621 Err(CompileError::InvalidProjectionPlacement(_)) => {}
3622 Err(other) => {
3623 panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
3624 }
3625 Ok(_) => {
3626 panic!("expected compile-time rejection for a missing {{task_id}} placeholder")
3627 }
3628 }
3629 }
3630
3631 #[test]
3632 fn declared_invalid_root_literal_rejects_compile() {
3633 let bp = minimal_bp(Some(ProjectionPlacementSpec {
3634 root: Some("nope".to_string()),
3635 dir_template: None,
3636 }));
3637 match Compiler::new(registry_with_echo()).compile(&bp) {
3638 Err(CompileError::InvalidProjectionPlacement(_)) => {}
3639 Err(other) => {
3640 panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
3641 }
3642 Ok(_) => panic!("expected compile-time rejection for an invalid root literal"),
3643 }
3644 }
3645}
3646
3647#[cfg(test)]
3649mod verdict_contract_lint_tests {
3650 use super::*;
3651 use crate::worker::adapter::WorkerResult;
3652
3653 fn registry_with_echo() -> SpawnerRegistry {
3654 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
3655 Ok(WorkerResult {
3656 value: Value::String(inv.prompt),
3657 ok: true,
3658 stats: None,
3659 })
3660 });
3661 let mut reg = SpawnerRegistry::new();
3662 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
3663 reg
3664 }
3665
3666 fn gate_agent(verdict: Option<VerdictContract>) -> AgentDef {
3667 AgentDef {
3668 name: "gate".to_string(),
3669 kind: AgentKind::RustFn,
3670 spec: serde_json::json!({ "fn_id": "echo" }),
3671 profile: None,
3672 meta: None,
3673 runner: None,
3674 runner_ref: None,
3675 verdict,
3676 }
3677 }
3678
3679 fn minimal_bp(agent: AgentDef, flow: FlowNode) -> Blueprint {
3680 Blueprint {
3681 schema_version: crate::blueprint::current_schema_version(),
3682 id: "verdict-contract-ut".into(),
3683 flow,
3684 agents: vec![agent],
3685 operators: vec![],
3686 metas: vec![],
3687 hints: Default::default(),
3688 strategy: Default::default(),
3689 metadata: BlueprintMetadata::default(),
3690 spawner_hints: Default::default(),
3691 default_agent_kind: AgentKind::Operator,
3692 default_operator_kind: None,
3693 default_init_ctx: None,
3694 default_agent_ctx: None,
3695 default_context_policy: None,
3696 projection_placement: None,
3697 audits: vec![],
3698 degradation_policy: None,
3699 runners: vec![],
3700 default_runner: None,
3701 subprocesses: vec![],
3702 check_policy: None,
3703 blueprint_ref_includes: Vec::new(),
3704 }
3705 }
3706
3707 fn step(ref_: &str, out_path: &str) -> FlowNode {
3708 FlowNode::Step {
3709 ref_: ref_.to_string(),
3710 in_: Expr::Lit { value: Value::Null },
3711 out: Expr::Path {
3712 at: out_path.parse().expect("literal test path"),
3713 },
3714 }
3715 }
3716
3717 fn noop() -> FlowNode {
3718 FlowNode::Seq { children: vec![] }
3719 }
3720
3721 fn eq_cond(path: &str, lit: &str) -> Expr {
3722 Expr::Eq {
3723 lhs: Box::new(Expr::Path {
3724 at: path.parse().expect("literal test path"),
3725 }),
3726 rhs: Box::new(Expr::Lit {
3727 value: Value::String(lit.to_string()),
3728 }),
3729 }
3730 }
3731
3732 fn branch(cond: Expr, then_: FlowNode, else_: FlowNode) -> FlowNode {
3733 FlowNode::Branch {
3734 cond,
3735 then_: Box::new(then_),
3736 else_: Box::new(else_),
3737 }
3738 }
3739
3740 fn body_contract(values: &[&str]) -> VerdictContract {
3741 VerdictContract {
3742 channel: VerdictChannel::Body,
3743 values: values.iter().map(|v| v.to_string()).collect(),
3744 }
3745 }
3746
3747 fn part_contract(values: &[&str]) -> VerdictContract {
3748 VerdictContract {
3749 channel: VerdictChannel::Part,
3750 values: values.iter().map(|v| v.to_string()).collect(),
3751 }
3752 }
3753
3754 #[test]
3755 fn contract_with_correct_body_channel_and_value_compiles() {
3756 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3757 let flow = FlowNode::Seq {
3758 children: vec![
3759 step("gate", "$.verdict"),
3760 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3761 ],
3762 };
3763 let bp = minimal_bp(agent, flow);
3764 assert!(
3765 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3766 "a cond addressing the bare step output must match a channel: \"body\" contract"
3767 );
3768 }
3769
3770 #[test]
3771 fn contract_with_correct_part_channel_and_value_compiles() {
3772 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3773 let flow = FlowNode::Seq {
3774 children: vec![
3775 step("gate", "$.gate"),
3776 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3777 ],
3778 };
3779 let bp = minimal_bp(agent, flow);
3780 assert!(
3781 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3782 "a cond addressing '<step>.parts.verdict' must match a channel: \"part\" contract"
3783 );
3784 }
3785
3786 #[test]
3787 fn body_channel_contract_rejects_cond_addressing_parts_verdict() {
3788 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3792 let flow = FlowNode::Seq {
3793 children: vec![
3794 step("gate", "$.gate"),
3795 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3796 ],
3797 };
3798 let bp = minimal_bp(agent, flow);
3799 match Compiler::new(registry_with_echo()).compile(&bp) {
3800 Err(CompileError::VerdictChannelMismatch {
3801 where_,
3802 agent,
3803 expected_channel,
3804 actual_shape,
3805 }) => {
3806 assert_eq!(agent, "gate");
3807 assert_eq!(expected_channel, "body");
3808 assert_eq!(actual_shape, "part");
3809 assert!(where_.contains("Branch cond"), "where_: {where_}");
3810 }
3811 Err(other) => {
3812 panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
3813 }
3814 Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
3815 }
3816 }
3817
3818 #[test]
3819 fn part_channel_contract_rejects_cond_addressing_bare_output() {
3820 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3823 let flow = FlowNode::Seq {
3824 children: vec![
3825 step("gate", "$.verdict"),
3826 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3827 ],
3828 };
3829 let bp = minimal_bp(agent, flow);
3830 match Compiler::new(registry_with_echo()).compile(&bp) {
3831 Err(CompileError::VerdictChannelMismatch {
3832 agent,
3833 expected_channel,
3834 actual_shape,
3835 ..
3836 }) => {
3837 assert_eq!(agent, "gate");
3838 assert_eq!(expected_channel, "part");
3839 assert_eq!(actual_shape, "body");
3840 }
3841 Err(other) => {
3842 panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
3843 }
3844 Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
3845 }
3846 }
3847
3848 #[test]
3849 fn contract_rejects_lit_outside_declared_values() {
3850 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3851 let flow = FlowNode::Seq {
3852 children: vec![
3853 step("gate", "$.verdict"),
3854 branch(eq_cond("$.verdict", "UNKNOWN"), noop(), noop()),
3855 ],
3856 };
3857 let bp = minimal_bp(agent, flow);
3858 match Compiler::new(registry_with_echo()).compile(&bp) {
3859 Err(CompileError::VerdictValueNotInContract {
3860 agent,
3861 value,
3862 values,
3863 ..
3864 }) => {
3865 assert_eq!(agent, "gate");
3866 assert_eq!(value, "UNKNOWN");
3867 assert_eq!(values, vec!["PASS".to_string(), "BLOCKED".to_string()]);
3868 }
3869 Err(other) => {
3870 panic!("expected VerdictValueNotInContract, got a different CompileError: {other}")
3871 }
3872 Ok(_) => panic!("expected compile-time rejection for a Lit outside declared values"),
3873 }
3874 }
3875
3876 #[test]
3877 fn undeclared_agent_referenced_by_cond_compiles_with_warning_only() {
3878 let agent = gate_agent(None);
3879 let flow = FlowNode::Seq {
3880 children: vec![
3881 step("gate", "$.verdict"),
3882 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3883 ],
3884 };
3885 let bp = minimal_bp(agent, flow);
3886 assert!(
3887 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3888 "an undeclared verdict contract must never reject compile (opt-in, back-compat)"
3889 );
3890 }
3891
3892 #[test]
3893 fn in_expr_with_lit_haystack_members_compiles() {
3894 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3895 let cond = Expr::In {
3896 needle: Box::new(Expr::Path {
3897 at: "$.verdict".parse().expect("literal test path"),
3898 }),
3899 haystack: Box::new(Expr::Lit {
3900 value: serde_json::json!(["PASS", "BLOCKED"]),
3901 }),
3902 };
3903 let flow = FlowNode::Seq {
3904 children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
3905 };
3906 let bp = minimal_bp(agent, flow);
3907 assert!(
3908 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3909 "an `In` haystack whose every Lit is a declared value must compile"
3910 );
3911 }
3912
3913 #[test]
3920 fn strict_mode_rejects_unhandled_declared_value() {
3921 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3922 let flow = FlowNode::Seq {
3923 children: vec![
3924 step("gate", "$.verdict"),
3925 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3926 ],
3927 };
3928 let mut bp = minimal_bp(agent, flow);
3929 bp.metadata.strict_verdict_handling = Some(true);
3930 match Compiler::new(registry_with_echo()).compile(&bp) {
3931 Err(CompileError::VerdictValueUnhandled {
3932 agent,
3933 value,
3934 declared_values,
3935 step_ref,
3936 }) => {
3937 assert_eq!(agent, "gate");
3938 assert_eq!(value, "PASS");
3939 assert_eq!(
3940 declared_values,
3941 vec!["PASS".to_string(), "BLOCKED".to_string()]
3942 );
3943 assert_eq!(step_ref, "gate");
3944 }
3945 Err(other) => {
3946 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
3947 }
3948 Ok(_) => panic!(
3949 "expected compile-time rejection for a declared verdict value with no \
3950 downstream handler under strict_verdict_handling=Some(true)"
3951 ),
3952 }
3953 }
3954
3955 #[test]
3962 fn default_mode_permits_unhandled_declared_value() {
3963 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3964 let flow = FlowNode::Seq {
3965 children: vec![
3966 step("gate", "$.verdict"),
3967 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3968 ],
3969 };
3970 let bp = minimal_bp(agent, flow);
3971 assert!(
3973 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3974 "default mode must never reject a Blueprint for unhandled declared values \
3975 (opt-in, back-compat with GH #50)"
3976 );
3977 }
3978
3979 #[test]
3984 fn strict_mode_accepts_all_declared_values_handled() {
3985 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3986 let flow = FlowNode::Seq {
3989 children: vec![
3990 step("gate", "$.verdict"),
3991 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3992 branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
3993 ],
3994 };
3995 let mut bp = minimal_bp(agent, flow);
3996 bp.metadata.strict_verdict_handling = Some(true);
3997 assert!(
3998 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3999 "strict mode must accept a Blueprint that handles every declared value"
4000 );
4001 }
4002
4003 #[test]
4007 fn strict_mode_accepts_declared_values_covered_by_in_expr() {
4008 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4009 let cond = Expr::In {
4010 needle: Box::new(Expr::Path {
4011 at: "$.verdict".parse().expect("literal test path"),
4012 }),
4013 haystack: Box::new(Expr::Lit {
4014 value: serde_json::json!(["PASS", "BLOCKED"]),
4015 }),
4016 };
4017 let flow = FlowNode::Seq {
4018 children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
4019 };
4020 let mut bp = minimal_bp(agent, flow);
4021 bp.metadata.strict_verdict_handling = Some(true);
4022 assert!(
4023 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4024 "strict mode must accept an `In` haystack that covers every declared value"
4025 );
4026 }
4027
4028 #[test]
4032 fn strict_mode_rejects_unhandled_part_channel_value() {
4033 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
4034 let flow = FlowNode::Seq {
4035 children: vec![
4036 step("gate", "$.gate"),
4037 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
4038 ],
4039 };
4040 let mut bp = minimal_bp(agent, flow);
4041 bp.metadata.strict_verdict_handling = Some(true);
4042 match Compiler::new(registry_with_echo()).compile(&bp) {
4043 Err(CompileError::VerdictValueUnhandled {
4044 agent,
4045 value,
4046 step_ref,
4047 ..
4048 }) => {
4049 assert_eq!(agent, "gate");
4050 assert_eq!(value, "PASS");
4051 assert_eq!(step_ref, "gate");
4052 }
4053 Err(other) => {
4054 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4055 }
4056 Ok(_) => panic!(
4057 "expected compile-time rejection for a declared verdict value with no \
4058 downstream handler (part channel) under strict_verdict_handling=Some(true)"
4059 ),
4060 }
4061 }
4062
4063 #[test]
4070 fn verdict_omitted_blueprint_compiles_unchanged_with_empty_contracts() {
4071 let agent = gate_agent(None);
4072 let flow = FlowNode::Seq {
4073 children: vec![
4074 step("gate", "$.verdict"),
4075 FlowNode::Loop {
4076 counter: Expr::Path {
4077 at: "$.n".parse().expect("literal test path"),
4078 },
4079 cond: eq_cond("$.verdict", "BLOCKED"),
4080 body: Box::new(step("gate", "$.verdict")),
4081 max: 3,
4082 },
4083 branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
4084 ],
4085 };
4086 let bp = minimal_bp(agent, flow);
4087 let compiled = Compiler::new(registry_with_echo())
4088 .compile(&bp)
4089 .expect("a verdict-omitted Blueprint must compile unchanged");
4090 assert!(
4091 compiled.router.verdict_contracts.is_empty(),
4092 "no agent declared a verdict contract"
4093 );
4094 }
4095
4096 #[test]
4103 fn every_compile_error_diagnostic_kind_is_a_declared_lint() {
4104 let kinds = [
4105 "bound-agent-resolution",
4106 "unknown-agent-kind",
4107 "invalid-agent-spec",
4108 "worker-binding-missing",
4109 "unresolved-agent-ref",
4110 "duplicate-agent-name",
4111 "unresolved-operator-ref",
4112 "unresolved-meta-ref",
4113 "step-naming-collision",
4114 "invalid-projection-placement",
4115 "unresolved-audit-agent",
4116 "verdict-channel-mismatch",
4117 "verdict-value-not-in-contract",
4118 "verdict-value-unhandled",
4119 ];
4120 for kind in kinds {
4121 assert!(
4122 mlua_swarm_diag::lint_decl(kind).is_some(),
4123 "kind '{kind}' emitted by From<&CompileError> has no LINT_DECLS entry"
4124 );
4125 }
4126 }
4127
4128 #[test]
4129 fn invalid_spec_with_worker_binding_prefix_specializes_the_diagnostic_kind() {
4130 let err = CompileError::InvalidSpec {
4134 name: "greeter".into(),
4135 msg: format!("{WORKER_BINDING_REQUIRED_MSG_PREFIX}. Fix by either: (a) ..."),
4136 };
4137 let d = mlua_swarm_diag::Diagnostic::from(&err);
4138 assert_eq!(d.kind, "worker-binding-missing");
4139 assert_eq!(d.level, mlua_swarm_diag::DiagLevel::Error);
4140 assert!(matches!(d.stage, mlua_swarm_diag::DiagStage::CompileLint));
4141 assert!(d.message.contains("greeter"));
4142 let suggestion = d
4143 .suggestion
4144 .expect("specialized arm must carry a suggestion");
4145 assert!(suggestion.patch.contains("backend = \"ws_operator\""));
4146 assert_eq!(
4147 suggestion.applicability,
4148 mlua_swarm_diag::Applicability::HasPlaceholders
4149 );
4150 assert_eq!(
4151 d.docs_ref.expect("docs_ref must be set").uri,
4152 "mse://guides/bp-dsl-templates"
4153 );
4154 match d.span.expect("span must be set").element {
4155 mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "greeter"),
4156 other => panic!("expected Agent span, got {other:?}"),
4157 }
4158 }
4159
4160 #[test]
4161 fn generic_invalid_spec_maps_to_the_generic_kind() {
4162 let err = CompileError::InvalidSpec {
4163 name: "solo".into(),
4164 msg: "operator spec: 'operator_ref' (string) required".into(),
4165 };
4166 let d = mlua_swarm_diag::Diagnostic::from(&err);
4167 assert_eq!(d.kind, "invalid-agent-spec");
4168 assert!(
4169 d.suggestion.is_none(),
4170 "generic arm carries no canned patch"
4171 );
4172 }
4173
4174 #[test]
4175 fn verdict_value_not_in_contract_diagnostic_carries_suggestion_and_span() {
4176 let err = CompileError::VerdictValueNotInContract {
4177 where_: "Branch cond".into(),
4178 agent: "review".into(),
4179 value: "NOT_DECLARED".into(),
4180 values: vec!["PASS".into(), "BLOCKED".into()],
4181 };
4182 let d = mlua_swarm_diag::Diagnostic::from(&err);
4183 assert_eq!(d.kind, "verdict-value-not-in-contract");
4184 assert!(d.message.contains("NOT_DECLARED"));
4185 assert!(d.suggestion.is_some());
4186 match d.span.expect("span must be set").element {
4187 mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "review"),
4188 other => panic!("expected Agent span, got {other:?}"),
4189 }
4190 }
4191}
4192
4193#[cfg(test)]
4195mod subprocess_embed_compile_tests {
4196 use super::*;
4197 use mlua_swarm_schema::{current_schema_version, SubprocessDef, SubprocessOverrides};
4198
4199 fn subprocess_agent(name: &str, runner: Option<Runner>) -> AgentDef {
4200 AgentDef {
4201 name: name.to_string(),
4202 kind: AgentKind::Subprocess,
4203 spec: serde_json::json!({}),
4204 profile: Some(AgentProfile {
4205 system_prompt: "you are a headless worker".to_string(),
4206 model: Some("profile-model".to_string()),
4207 tools: vec!["Read".to_string()],
4208 ..Default::default()
4209 }),
4210 meta: None,
4211 runner,
4212 runner_ref: None,
4213 verdict: None,
4214 }
4215 }
4216
4217 fn echo_def(name: &str) -> SubprocessDef {
4218 SubprocessDef {
4219 name: name.to_string(),
4220 argv: vec!["sh".to_string(), "-c".to_string(), "cat".to_string()],
4221 stdin: Some("{prompt}".to_string()),
4222 env: Default::default(),
4223 cwd: None,
4224 output: None,
4225 stream_mode: None,
4226 }
4227 }
4228
4229 fn bp_with(agents: Vec<AgentDef>, subprocesses: Vec<SubprocessDef>) -> Blueprint {
4230 Blueprint {
4231 schema_version: current_schema_version(),
4232 id: "gh83-ut".into(),
4233 flow: FlowNode::Seq { children: vec![] },
4234 agents,
4235 operators: vec![],
4236 metas: vec![],
4237 hints: Default::default(),
4238 strategy: Default::default(),
4239 metadata: BlueprintMetadata::default(),
4240 spawner_hints: Default::default(),
4241 default_agent_kind: AgentKind::Operator,
4242 default_operator_kind: None,
4243 default_init_ctx: None,
4244 default_agent_ctx: None,
4245 default_context_policy: None,
4246 projection_placement: None,
4247 audits: vec![],
4248 degradation_policy: None,
4249 runners: vec![],
4250 default_runner: None,
4251 subprocesses,
4252 check_policy: None,
4253 blueprint_ref_includes: vec![],
4254 }
4255 }
4256
4257 fn subprocess_runner(template: &str) -> Runner {
4258 Runner::Subprocess {
4259 template: template.to_string(),
4260 overrides: SubprocessOverrides::default(),
4261 }
4262 }
4263
4264 #[test]
4265 fn validate_placeholders_accepts_closed_set_and_json_braces() {
4266 for ok in [
4267 "{system} {system_file} {prompt} {model} {tools_csv} {work_dir} {task_id} {attempt}",
4268 r#"echo '{"result": "ok", "nested": {"a": 1}}'"#,
4269 "no placeholders at all",
4270 "unmatched { brace",
4271 ] {
4272 validate_embed_placeholders(ok, "ut").expect("must be accepted");
4273 }
4274 }
4275
4276 #[test]
4277 fn validate_placeholders_rejects_unknown_token() {
4278 let err = validate_embed_placeholders("--flag {evil}", "argv[1]").unwrap_err();
4279 assert!(err.contains("'{evil}'"), "token named: {err}");
4280 assert!(err.contains("closed set"), "closed set listed: {err}");
4281 }
4282
4283 #[test]
4287 fn validate_placeholders_descends_into_literal_braces() {
4288 validate_embed_placeholders(r#"{"task": "{prompt}"}"#, "stdin")
4289 .expect("nested closed-set token must be accepted");
4290 let err = validate_embed_placeholders(r#"{"task": "{evil}"}"#, "stdin").unwrap_err();
4291 assert!(
4292 err.contains("'{evil}'"),
4293 "nested unknown token caught: {err}"
4294 );
4295 }
4296
4297 #[test]
4298 fn hint_resolution_finds_declared_template() {
4299 let agent = subprocess_agent("headless", Some(subprocess_runner("echo")));
4300 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
4301 let hint = resolve_subprocess_template_hint(&bp, &agent)
4302 .expect("resolves")
4303 .expect("Runner::Subprocess must synthesize a hint");
4304 assert_eq!(hint[SUBPROCESS_TEMPLATE_HINT_KEY]["name"], "echo");
4305 assert!(hint.get(SUBPROCESS_OVERRIDES_HINT_KEY).is_some());
4306 }
4307
4308 #[test]
4309 fn hint_resolution_unknown_template_is_invalid_spec() {
4310 let agent = subprocess_agent("headless", Some(subprocess_runner("nope")));
4311 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
4312 let err = resolve_subprocess_template_hint(&bp, &agent).unwrap_err();
4313 let msg = format!("{err}");
4314 assert!(msg.contains("'nope'"), "missing template named: {msg}");
4315 assert!(msg.contains("echo"), "defined templates listed: {msg}");
4316 }
4317
4318 #[test]
4319 fn hint_resolution_none_without_subprocess_runner() {
4320 let agent = subprocess_agent("headless", None);
4321 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
4322 let hint = resolve_subprocess_template_hint(&bp, &agent).expect("resolves");
4323 assert!(hint.is_none(), "spec-based agents keep the historical path");
4324 }
4325
4326 fn agent_block_agent(name: &str, runner: Option<Runner>, profile_tools: &[&str]) -> AgentDef {
4333 AgentDef {
4334 name: name.to_string(),
4335 kind: AgentKind::AgentBlock,
4336 spec: serde_json::json!({}),
4337 profile: Some(AgentProfile {
4338 system_prompt: "you are an in-process auditor".to_string(),
4339 tools: profile_tools.iter().map(|t| t.to_string()).collect(),
4340 ..Default::default()
4341 }),
4342 meta: None,
4343 runner,
4344 runner_ref: None,
4345 verdict: None,
4346 }
4347 }
4348
4349 fn agent_block_runner(tools: &[&str]) -> Runner {
4350 Runner::AgentBlockInProcess {
4351 tools: tools.iter().map(|t| t.to_string()).collect(),
4352 }
4353 }
4354
4355 #[test]
4361 fn agent_block_runner_tools_are_projected_over_profile_tools() {
4362 let agent = agent_block_agent(
4363 "auditor",
4364 Some(agent_block_runner(&["mcp__outline__list_docs"])),
4365 &["Read"],
4366 );
4367 let bp = bp_with(vec![agent], vec![]);
4368 let bound = resolve_bound_agents(&bp).expect("binds");
4369 let effective = materialize_bound_blueprint(&bp, &bound);
4370 assert_eq!(
4371 effective.agents[0].profile.as_ref().unwrap().tools,
4372 vec!["mcp__outline__list_docs".to_string()],
4373 "the declared Runner tools replace profile.tools (['Read'])"
4374 );
4375 }
4376
4377 #[test]
4381 fn agent_block_projection_distinguishes_declared_empty_from_absent() {
4382 let declared = agent_block_agent("auditor", Some(agent_block_runner(&[])), &["Read"]);
4383 let bp = bp_with(vec![declared], vec![]);
4384 let bound = resolve_bound_agents(&bp).expect("binds");
4385 let effective = materialize_bound_blueprint(&bp, &bound);
4386 assert!(
4387 effective.agents[0]
4388 .profile
4389 .as_ref()
4390 .unwrap()
4391 .tools
4392 .is_empty(),
4393 "empty means enforced-empty, not 'unset'"
4394 );
4395
4396 let absent = agent_block_agent("auditor", None, &["Read"]);
4397 let bp = bp_with(vec![absent], vec![]);
4398 let bound = resolve_bound_agents(&bp).expect("binds");
4399 let effective = materialize_bound_blueprint(&bp, &bound);
4400 assert_eq!(
4401 effective.agents[0].profile.as_ref().unwrap().tools,
4402 vec!["Read".to_string()],
4403 "no Runner declared → the agent.md tools line stands"
4404 );
4405 }
4406
4407 #[test]
4414 fn compile_rejects_script_mode_with_a_declared_mcp_grant() {
4415 let mut agent = agent_block_agent(
4416 "auditor",
4417 Some(agent_block_runner(&["mcp__outline__list_docs"])),
4418 &[],
4419 );
4420 agent.spec = serde_json::json!({ "script_path": "gate.lua" });
4421 let mut bp = bp_with(vec![agent], vec![]);
4422 bp.strategy.strict_refs = false;
4423
4424 let mut registry = SpawnerRegistry::new();
4425 registry.register::<crate::worker::agent_block::AgentBlockInProcessSpawnerFactory>(
4426 Arc::new(crate::worker::agent_block::AgentBlockInProcessSpawnerFactory::new()),
4427 );
4428 let err = match Compiler::new(registry).compile(&bp) {
4430 Err(e) => e,
4431 Ok(_) => panic!("script mode + declared MCP grant must not compile"),
4432 };
4433 let msg = format!("{err}");
4434 assert!(msg.contains("script_path"), "names the trigger: {msg}");
4435 assert!(
4436 msg.contains("mcp__outline__list_docs"),
4437 "names the unenforceable tools: {msg}"
4438 );
4439 }
4440
4441 #[test]
4445 fn compile_accepts_script_mode_with_only_inert_tools() {
4446 let mut agent = agent_block_agent("auditor", None, &["Read", "WebSearch"]);
4447 agent.spec = serde_json::json!({ "script_path": "gate.lua" });
4448 let mut bp = bp_with(vec![agent], vec![]);
4449 bp.strategy.strict_refs = false;
4450
4451 let mut registry = SpawnerRegistry::new();
4452 registry.register::<crate::worker::agent_block::AgentBlockInProcessSpawnerFactory>(
4453 Arc::new(crate::worker::agent_block::AgentBlockInProcessSpawnerFactory::new()),
4454 );
4455 if let Err(e) = Compiler::new(registry).compile(&bp) {
4456 panic!("inert tools must not trip the MCP-grant guard: {e}");
4457 }
4458 }
4459
4460 #[test]
4461 fn build_embed_rejects_unknown_placeholder() {
4462 let agent = subprocess_agent("headless", None);
4463 let mut def = echo_def("echo");
4464 def.argv.push("--x={evil}".to_string());
4465 let err = SubprocessProcessSpawnerFactory::build_embed(
4466 &agent,
4467 &serde_json::to_value(&def).unwrap(),
4468 None,
4469 )
4470 .unwrap_err();
4471 assert!(format!("{err}").contains("'{evil}'"));
4472 }
4473
4474 #[test]
4475 fn build_embed_rejects_output_with_stream_mode() {
4476 let agent = subprocess_agent("headless", None);
4477 let mut def = echo_def("echo");
4478 def.stream_mode = Some("ndjson_lines".to_string());
4479 def.output = Some(mlua_swarm_schema::SubprocessOutput {
4480 format: Some("json".to_string()),
4481 result_ptr: None,
4482 ok_from: None,
4483 stats: None,
4484 });
4485 let err = SubprocessProcessSpawnerFactory::build_embed(
4486 &agent,
4487 &serde_json::to_value(&def).unwrap(),
4488 None,
4489 )
4490 .unwrap_err();
4491 assert!(format!("{err}").contains("plain-mode"));
4492 }
4493
4494 #[test]
4495 fn build_embed_rejects_malformed_result_ptr_and_ok_from() {
4496 let agent = subprocess_agent("headless", None);
4497 let mut def = echo_def("echo");
4498 def.output = Some(mlua_swarm_schema::SubprocessOutput {
4499 format: None,
4500 result_ptr: Some("result".to_string()),
4501 ok_from: None,
4502 stats: None,
4503 });
4504 let err = SubprocessProcessSpawnerFactory::build_embed(
4505 &agent,
4506 &serde_json::to_value(&def).unwrap(),
4507 None,
4508 )
4509 .unwrap_err();
4510 assert!(format!("{err}").contains("JSON Pointer"));
4511
4512 let mut def = echo_def("echo");
4513 def.output = Some(mlua_swarm_schema::SubprocessOutput {
4514 format: None,
4515 result_ptr: None,
4516 ok_from: Some("status".to_string()),
4517 stats: None,
4518 });
4519 let err = SubprocessProcessSpawnerFactory::build_embed(
4520 &agent,
4521 &serde_json::to_value(&def).unwrap(),
4522 None,
4523 )
4524 .unwrap_err();
4525 assert!(format!("{err}").contains("exit_code"));
4526 }
4527
4528 #[test]
4529 fn build_embed_bakes_profile_with_override_precedence() {
4530 let agent = subprocess_agent("headless", None);
4531 let def = echo_def("echo");
4532 let overrides = SubprocessOverrides {
4533 model: Some("override-model".to_string()),
4534 tools: vec!["Bash".to_string(), "Write".to_string()],
4535 cwd: Some("/tmp/override-wd".to_string()),
4536 };
4537 let sp = SubprocessProcessSpawnerFactory::build_embed(
4538 &agent,
4539 &serde_json::to_value(&def).unwrap(),
4540 Some(&serde_json::to_value(&overrides).unwrap()),
4541 )
4542 .expect("builds");
4543 let embed = sp.embed.as_ref().expect("embed template baked");
4544 assert_eq!(embed.model.as_deref(), Some("override-model"));
4545 assert_eq!(embed.tools_csv, "Bash,Write");
4546 assert_eq!(embed.cwd.as_deref(), Some("/tmp/override-wd"));
4547 assert_eq!(
4548 embed.system_prompt.as_deref(),
4549 Some("you are a headless worker")
4550 );
4551 }
4552}