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::{BTreeMap, HashMap};
48use std::sync::Arc;
49use thiserror::Error;
50
51#[derive(Debug, Error)]
56pub enum CompileError {
57 #[error("bound agent resolution: {0}")]
59 BoundAgent(#[from] BoundAgentResolveError),
60 #[error("unknown agent kind in SpawnerRegistry: {0:?}")]
63 UnknownKind(AgentKind),
64 #[error("agent '{name}' spec invalid: {msg}")]
67 InvalidSpec {
68 name: String,
70 msg: String,
72 },
73 #[error("flow references agent '{0}' but no AgentDef matches")]
76 UnresolvedRef(String),
77 #[error("duplicate AgentDef name: {0}")]
79 DuplicateAgent(String),
80 #[error("agent '{agent}' operator_ref '{op_ref}' does not match any OperatorDef.name in Blueprint.operators (defined: {defined:?})")]
83 UnresolvedOperatorRef {
84 agent: String,
86 op_ref: String,
88 defined: Vec<String>,
91 },
92 #[error("{where_} names an undefined MetaDef: '{meta_ref}' (defined: {defined:?})")]
96 UnresolvedMetaRef {
97 where_: String,
101 meta_ref: String,
103 defined: Vec<String>,
106 },
107 #[error("StepNaming collision: {0}")]
113 StepNamingCollision(#[from] StepNamingError),
114 #[error("invalid projection_placement: {0}")]
121 InvalidProjectionPlacement(#[from] ProjectionPlacementError),
122 #[error("audits[].agent '{agent}' does not match any AgentDef.name in Blueprint.agents (defined: {defined:?})")]
127 UnresolvedAuditAgent {
128 agent: String,
130 defined: Vec<String>,
133 },
134 #[error(
143 "agent '{agent}' declares verdict channel '{expected_channel}' but {where_} \
144 addresses it as '{actual_shape}' output — see the \"Returning verdicts to drive \
145 BP flow\" guide's Pattern A (channel: \"body\") / Pattern B (channel: \"part\")"
146 )]
147 VerdictChannelMismatch {
148 where_: String,
151 agent: String,
153 expected_channel: String,
155 actual_shape: String,
158 },
159 #[error(
164 "agent '{agent}' verdict Lit '{value}' at {where_} is not a member of the declared \
165 values {values:?}"
166 )]
167 VerdictValueNotInContract {
168 where_: String,
171 agent: String,
174 value: String,
179 values: Vec<String>,
182 },
183 #[error(
195 "agent '{agent}' declares verdict value '{value}' but no downstream Branch/Loop \
196 cond references it (declared: {declared_values:?}, at step '{step_ref}') — either \
197 handle the value downstream or drop it from `verdict.values`"
198 )]
199 VerdictValueUnhandled {
200 agent: String,
203 value: String,
205 declared_values: Vec<String>,
208 step_ref: String,
213 },
214}
215
216pub const WORKER_BINDING_REQUIRED_MSG_PREFIX: &str =
225 "profile.worker_binding is required for this operator backend";
226
227impl From<&CompileError> for mlua_swarm_diag::Diagnostic {
245 fn from(err: &CompileError) -> Self {
246 use mlua_swarm_diag::{
247 Applicability, DiagElement, DiagLevel, DiagSpan, DiagStage, Diagnostic, DocsRef,
248 Suggestion,
249 };
250 let base = |kind: &'static str| {
251 Diagnostic::new(
252 kind,
253 DiagStage::CompileLint,
254 DiagLevel::Error,
255 err.to_string(),
256 )
257 };
258 let agent_span = |name: &str| DiagSpan {
259 element: DiagElement::Agent {
260 name: name.to_string(),
261 },
262 json_path: Some(format!("$.agents[?(@.name=='{name}')]")),
263 };
264 match err {
265 CompileError::BoundAgent(_) => base("bound-agent-resolution"),
266 CompileError::UnknownKind(_) => base("unknown-agent-kind").with_help(
267 "register a SpawnerFactory for this kind, or disable strategy.strict_kind",
268 ),
269 CompileError::InvalidSpec { name, msg }
270 if msg.starts_with(WORKER_BINDING_REQUIRED_MSG_PREFIX) =>
271 {
272 Diagnostic::new(
273 "worker-binding-missing",
274 DiagStage::CompileLint,
275 DiagLevel::Error,
276 format!(
277 "operator agent '{name}' has no explicit Runner or legacy \
278 `profile.worker_binding`"
279 ),
280 )
281 .with_note(msg.clone())
282 .with_suggestion(Suggestion {
283 msg: "add an explicit Runner (or legacy profile.worker_binding)".into(),
284 patch: "runner = { backend = \"ws_operator\", variant = \"claude\", \
285 tools = {} }"
286 .into(),
287 applicability: Applicability::HasPlaceholders,
288 })
289 .with_docs_ref(DocsRef {
290 uri: "mse://guides/bp-dsl-templates",
291 anchor: None,
292 })
293 .with_span(agent_span(name))
294 }
295 CompileError::InvalidSpec { name, .. } => {
296 base("invalid-agent-spec").with_span(agent_span(name))
297 }
298 CompileError::UnresolvedRef(ref_) => base("unresolved-agent-ref").with_span(DiagSpan {
299 element: DiagElement::Step { ref_: ref_.clone() },
300 json_path: None,
301 }),
302 CompileError::DuplicateAgent(name) => {
303 base("duplicate-agent-name").with_span(agent_span(name))
304 }
305 CompileError::UnresolvedOperatorRef { agent, defined, .. } => {
306 base("unresolved-operator-ref")
307 .with_note(format!("declared OperatorDef names: {defined:?}"))
308 .with_span(agent_span(agent))
309 }
310 CompileError::UnresolvedMetaRef { defined, .. } => base("unresolved-meta-ref")
311 .with_note(format!("declared MetaDef names: {defined:?}")),
312 CompileError::StepNamingCollision(_) => base("step-naming-collision"),
313 CompileError::InvalidProjectionPlacement(_) => base("invalid-projection-placement")
314 .with_span(DiagSpan {
315 element: DiagElement::BlueprintRoot,
316 json_path: Some("$.projection_placement".into()),
317 }),
318 CompileError::UnresolvedAuditAgent { defined, .. } => base("unresolved-audit-agent")
319 .with_note(format!("declared AgentDef names: {defined:?}"))
320 .with_span(DiagSpan {
321 element: DiagElement::BlueprintRoot,
322 json_path: Some("$.audits".into()),
323 }),
324 CompileError::VerdictChannelMismatch { agent, .. } => base("verdict-channel-mismatch")
325 .with_help(
326 "see the \"Returning verdicts to drive BP flow\" guide's Pattern A \
327 (channel: \"body\") / Pattern B (channel: \"part\")",
328 )
329 .with_docs_ref(DocsRef {
330 uri: "mse://guides/blueprint-authoring",
331 anchor: None,
332 })
333 .with_span(agent_span(agent)),
334 CompileError::VerdictValueNotInContract { agent, .. } => {
335 base("verdict-value-not-in-contract")
336 .with_suggestion(Suggestion {
342 msg: "align the cond literal with the agent's declared verdict \
343 contract"
344 .into(),
345 patch: "either add the cond's literal to `agents[N].verdict.values`, \
346 or change the cond to a value that is already declared"
347 .into(),
348 applicability: Applicability::MaybeIncorrect,
349 })
350 .with_docs_ref(DocsRef {
351 uri: "mse://guides/blueprint-authoring",
352 anchor: None,
353 })
354 .with_span(agent_span(agent))
355 }
356 CompileError::VerdictValueUnhandled {
357 agent,
358 declared_values,
359 ..
360 } => base("verdict-value-unhandled")
361 .with_note(format!("declared verdict.values: {declared_values:?}"))
362 .with_help(
363 "either handle the value in a downstream Branch/Loop cond, or drop it \
364 from verdict.values",
365 )
366 .with_span(agent_span(agent)),
367 }
368 }
369}
370
371pub trait SpawnerFactory: Send + Sync {
383 fn build(
386 &self,
387 agent_def: &AgentDef,
388 hint: Option<&Value>,
389 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError>;
390}
391
392pub trait SpawnerFactoryKind: SpawnerFactory {
408 const KIND: AgentKind;
411 type Worker: crate::worker::Worker;
418}
419
420#[derive(Clone)]
423pub struct SpawnerRegistry {
424 factories: HashMap<AgentKind, Arc<dyn SpawnerFactory>>,
425}
426
427impl SpawnerRegistry {
428 pub fn new() -> Self {
430 Self {
431 factories: HashMap::new(),
432 }
433 }
434 pub fn register<F: SpawnerFactoryKind + 'static>(&mut self, factory: Arc<F>) -> &mut Self {
443 let f: Arc<dyn SpawnerFactory> = factory;
444 self.factories.insert(F::KIND, f);
445 self
446 }
447}
448
449impl Default for SpawnerRegistry {
450 fn default() -> Self {
451 Self::new()
452 }
453}
454
455pub struct Compiler {
462 registry: SpawnerRegistry,
463 default_spawner: Option<Arc<dyn SpawnerAdapter>>,
464}
465
466pub struct CompiledBlueprint {
470 pub router: Arc<CompiledAgentTable>,
472 pub flow: FlowNode,
474 pub metadata: BlueprintMetadata,
476 pub step_naming: Arc<StepNaming>,
481 pub projection_placement: Arc<ProjectionPlacement>,
487}
488
489fn project_bound_agent_for_legacy_factories(bound: &BoundAgent) -> AgentDef {
490 let mut agent = bound.agent.clone();
491 match &bound.runner {
492 Some(Runner::WsOperator { variant, tools })
493 | Some(Runner::WsClaudeCode { variant, tools }) => {
494 let profile = agent.profile.get_or_insert_with(AgentProfile::default);
495 profile.worker_binding = Some(variant.clone());
496 profile.tools = tools.clone();
497 }
498 Some(Runner::AgentBlockInProcess { tools }) => {
499 let profile = agent.profile.get_or_insert_with(AgentProfile::default);
500 profile.worker_binding = None;
501 profile.tools = tools.clone();
502 }
503 Some(Runner::Subprocess { .. }) => {}
508 None => {}
509 }
510 let meta = agent.meta.get_or_insert_with(Default::default);
511 meta.context_policy = bound.context_policy.clone();
512 agent
513}
514
515pub(crate) fn materialize_bound_blueprint(
518 bp: &Blueprint,
519 bound_agents: &[BoundAgent],
520) -> Blueprint {
521 let mut effective = bp.clone();
522 effective.agents = bound_agents
523 .iter()
524 .map(project_bound_agent_for_legacy_factories)
525 .collect();
526 effective.default_context_policy = None;
529 effective
530}
531
532impl Compiler {
533 pub fn new(registry: SpawnerRegistry) -> Self {
537 Self {
538 registry,
539 default_spawner: None,
540 }
541 }
542
543 pub fn with_default(mut self, sp: Arc<dyn SpawnerAdapter>) -> Self {
547 self.default_spawner = Some(sp);
548 self
549 }
550
551 pub fn compile(&self, bp: &Blueprint) -> Result<CompiledBlueprint, CompileError> {
556 let bound_agents = resolve_bound_agents(bp)?;
557 self.compile_bound_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 unhandled_gates = resolve_unhandled_verdict_gates(bp);
771 verify_verdict_conds(&bp.flow, &verdict_contracts, &unhandled_gates)?;
772
773 if bp.strategy.strict_refs {
774 verify_refs(&bp.flow, &routes, self.default_spawner.is_some())?;
775 }
776
777 let (step_naming, step_naming_warnings) = StepNaming::from_blueprint(bp)?;
793 for warning in &step_naming_warnings {
794 tracing::warn!(
795 name = %warning.name,
796 first_step_ref = %warning.first_step_ref,
797 second_step_ref = %warning.second_step_ref,
798 "StepNaming: undeclared steps' canonical/alias names collide; \
799 the step whose own ref matches the name keeps it (data-plane priority)"
800 );
801 }
802
803 let projection_placement =
811 ProjectionPlacement::from_spec(bp.projection_placement.as_ref())?;
812
813 let router = Arc::new(CompiledAgentTable {
814 routes,
815 default: self.default_spawner.clone(),
816 verdict_contracts,
817 });
818 Ok(CompiledBlueprint {
819 router,
820 flow: bp.flow.clone(),
821 metadata: bp.metadata.clone(),
822 step_naming: Arc::new(step_naming),
823 projection_placement: Arc::new(projection_placement),
824 })
825 }
826}
827
828fn verify_refs(
831 node: &FlowNode,
832 routes: &HashMap<String, Arc<dyn SpawnerAdapter>>,
833 has_default: bool,
834) -> Result<(), CompileError> {
835 let mut refs: Vec<String> = Vec::new();
836 collect_refs(node, &mut refs);
837 for r in refs {
838 if !routes.contains_key(&r) && !has_default {
839 return Err(CompileError::UnresolvedRef(r));
840 }
841 }
842 Ok(())
843}
844
845fn collect_refs(node: &FlowNode, out: &mut Vec<String>) {
846 match node {
847 FlowNode::Step { ref_, .. } => out.push(ref_.clone()),
848 FlowNode::Seq { children } => {
849 for c in children {
850 collect_refs(c, out);
851 }
852 }
853 FlowNode::Branch { then_, else_, .. } => {
854 collect_refs(then_, out);
855 collect_refs(else_, out);
856 }
857 FlowNode::Fanout { body, .. } => collect_refs(body, out),
858 FlowNode::Loop { body, .. } => collect_refs(body, out),
859 FlowNode::Try { body, catch, .. } => {
860 collect_refs(body, out);
861 collect_refs(catch, out);
862 }
863 FlowNode::Assign { .. } => {} }
865}
866
867fn collect_step_meta_refs(node: &FlowNode, out: &mut Vec<(String, String)>) {
875 match node {
876 FlowNode::Step { ref_, in_, .. } => {
877 if let Expr::Lit { value } = in_ {
878 if let Some(meta_ref) = static_step_meta_ref(value) {
879 out.push((format!("Step '{ref_}' $step_meta.ref"), meta_ref));
880 }
881 }
882 }
883 FlowNode::Seq { children } => {
884 for c in children {
885 collect_step_meta_refs(c, out);
886 }
887 }
888 FlowNode::Branch { then_, else_, .. } => {
889 collect_step_meta_refs(then_, out);
890 collect_step_meta_refs(else_, out);
891 }
892 FlowNode::Fanout { body, .. } => collect_step_meta_refs(body, out),
893 FlowNode::Loop { body, .. } => collect_step_meta_refs(body, out),
894 FlowNode::Try { body, catch, .. } => {
895 collect_step_meta_refs(body, out);
896 collect_step_meta_refs(catch, out);
897 }
898 FlowNode::Assign { .. } => {} }
900}
901
902fn static_step_meta_ref(value: &Value) -> Option<String> {
909 value
910 .as_object()?
911 .get("$step_meta")?
912 .as_object()?
913 .get("ref")?
914 .as_str()
915 .map(str::to_string)
916}
917
918const UNHANDLED_VERDICT_LINT_KIND: &str = "verdict-value-unhandled";
927
928#[derive(Debug, Clone, Copy, PartialEq, Eq)]
930enum UnhandledVerdictGate {
931 Deny,
933 Warn,
935 Silence,
937}
938
939#[derive(Debug, Clone, PartialEq, Eq)]
948struct UnhandledVerdictGates {
949 per_agent: HashMap<String, UnhandledVerdictGate>,
951 blueprint: UnhandledVerdictGate,
953}
954
955impl UnhandledVerdictGates {
956 fn for_agent(&self, agent: &str) -> UnhandledVerdictGate {
959 self.per_agent.get(agent).copied().unwrap_or(self.blueprint)
960 }
961
962 fn all_silent(&self) -> bool {
965 self.blueprint == UnhandledVerdictGate::Silence
966 && self
967 .per_agent
968 .values()
969 .all(|g| *g == UnhandledVerdictGate::Silence)
970 }
971}
972
973fn resolve_unhandled_verdict_gates(bp: &Blueprint) -> UnhandledVerdictGates {
981 let strict = bp.metadata.strict_verdict_handling.unwrap_or(false);
982 let blueprint = resolve_unhandled_verdict_gate(&bp.metadata);
983 let per_agent = bp
984 .agents
985 .iter()
986 .filter_map(|ad| {
987 let declared = declared_unhandled_verdict_setting(&ad.lints)?;
988 Some((
989 ad.name.clone(),
990 unhandled_verdict_gate(strict, Some(declared)),
991 ))
992 })
993 .collect();
994 UnhandledVerdictGates {
995 per_agent,
996 blueprint,
997 }
998}
999
1000fn resolve_unhandled_verdict_gate(metadata: &BlueprintMetadata) -> UnhandledVerdictGate {
1003 unhandled_verdict_gate(
1004 metadata.strict_verdict_handling.unwrap_or(false),
1005 declared_unhandled_verdict_setting(&metadata.lints),
1006 )
1007}
1008
1009fn declared_unhandled_verdict_setting(
1019 lints: &Option<BTreeMap<String, mlua_swarm_schema::LintSetting>>,
1020) -> Option<mlua_swarm_diag::LintSetting> {
1021 use mlua_swarm_diag::{lint_decl, LintConfig};
1022
1023 let cfg = LintConfig::from_pairs(
1024 lints
1025 .as_ref()?
1026 .iter()
1027 .map(|(key, setting)| (key.clone(), diag_lint_setting(*setting))),
1028 );
1029 cfg.setting_for(lint_decl(UNHANDLED_VERDICT_LINT_KIND)?)
1030}
1031
1032fn unhandled_verdict_gate(
1040 strict: bool,
1041 declared: Option<mlua_swarm_diag::LintSetting>,
1042) -> UnhandledVerdictGate {
1043 use mlua_swarm_diag::LintSetting;
1044
1045 match declared {
1046 _ if strict => UnhandledVerdictGate::Deny,
1047 Some(LintSetting::Deny) => UnhandledVerdictGate::Deny,
1048 Some(LintSetting::Allow) => UnhandledVerdictGate::Silence,
1049 Some(LintSetting::Warn) | None => UnhandledVerdictGate::Warn,
1050 }
1051}
1052
1053fn diag_lint_setting(setting: mlua_swarm_schema::LintSetting) -> mlua_swarm_diag::LintSetting {
1058 match setting {
1059 mlua_swarm_schema::LintSetting::Allow => mlua_swarm_diag::LintSetting::Allow,
1060 mlua_swarm_schema::LintSetting::Warn => mlua_swarm_diag::LintSetting::Warn,
1061 mlua_swarm_schema::LintSetting::Deny => mlua_swarm_diag::LintSetting::Deny,
1062 }
1063}
1064
1065fn verify_verdict_conds(
1075 flow: &FlowNode,
1076 verdict_contracts: &HashMap<String, VerdictContract>,
1077 unhandled_gates: &UnhandledVerdictGates,
1078) -> Result<(), CompileError> {
1079 let mut step_outputs: HashMap<String, String> = HashMap::new();
1080 let mut step_agents: HashMap<String, String> = HashMap::new();
1081 collect_step_outputs_and_agents(flow, &mut step_outputs, &mut step_agents);
1082
1083 let mut errors: Vec<CompileError> = Vec::new();
1084 let mut referenced_values: HashMap<String, std::collections::HashSet<String>> = HashMap::new();
1085 collect_verdict_conds(
1086 flow,
1087 &step_outputs,
1088 verdict_contracts,
1089 &mut referenced_values,
1090 &mut errors,
1091 );
1092 check_unhandled_verdict_values(
1093 verdict_contracts,
1094 &referenced_values,
1095 &step_agents,
1096 unhandled_gates,
1097 &mut errors,
1098 );
1099 match errors.into_iter().next() {
1100 Some(e) => Err(e),
1101 None => Ok(()),
1102 }
1103}
1104
1105fn collect_step_outputs_and_agents(
1120 node: &FlowNode,
1121 out: &mut HashMap<String, String>,
1122 step_agents: &mut HashMap<String, String>,
1123) {
1124 match node {
1125 FlowNode::Step {
1126 ref_,
1127 out: out_expr,
1128 ..
1129 } => {
1130 if let Expr::Path { at } = out_expr {
1131 out.insert(at.to_string(), ref_.clone());
1132 }
1133 step_agents
1134 .entry(ref_.clone())
1135 .or_insert_with(|| ref_.clone());
1136 }
1137 FlowNode::Seq { children } => {
1138 for c in children {
1139 collect_step_outputs_and_agents(c, out, step_agents);
1140 }
1141 }
1142 FlowNode::Branch { then_, else_, .. } => {
1143 collect_step_outputs_and_agents(then_, out, step_agents);
1144 collect_step_outputs_and_agents(else_, out, step_agents);
1145 }
1146 FlowNode::Fanout { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
1147 FlowNode::Loop { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
1148 FlowNode::Try { body, catch, .. } => {
1149 collect_step_outputs_and_agents(body, out, step_agents);
1150 collect_step_outputs_and_agents(catch, out, step_agents);
1151 }
1152 FlowNode::Assign { .. } => {} }
1154}
1155
1156fn collect_verdict_conds(
1161 node: &FlowNode,
1162 step_outputs: &HashMap<String, String>,
1163 verdict_contracts: &HashMap<String, VerdictContract>,
1164 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
1165 errors: &mut Vec<CompileError>,
1166) {
1167 match node {
1168 FlowNode::Branch { cond, then_, else_ } => {
1169 lint_cond_expr(
1170 cond,
1171 "Branch cond",
1172 step_outputs,
1173 verdict_contracts,
1174 referenced_values,
1175 errors,
1176 );
1177 collect_verdict_conds(
1178 then_,
1179 step_outputs,
1180 verdict_contracts,
1181 referenced_values,
1182 errors,
1183 );
1184 collect_verdict_conds(
1185 else_,
1186 step_outputs,
1187 verdict_contracts,
1188 referenced_values,
1189 errors,
1190 );
1191 }
1192 FlowNode::Loop { cond, body, .. } => {
1193 lint_cond_expr(
1194 cond,
1195 "Loop cond",
1196 step_outputs,
1197 verdict_contracts,
1198 referenced_values,
1199 errors,
1200 );
1201 collect_verdict_conds(
1202 body,
1203 step_outputs,
1204 verdict_contracts,
1205 referenced_values,
1206 errors,
1207 );
1208 }
1209 FlowNode::Seq { children } => {
1210 for c in children {
1211 collect_verdict_conds(
1212 c,
1213 step_outputs,
1214 verdict_contracts,
1215 referenced_values,
1216 errors,
1217 );
1218 }
1219 }
1220 FlowNode::Fanout { body, .. } => collect_verdict_conds(
1221 body,
1222 step_outputs,
1223 verdict_contracts,
1224 referenced_values,
1225 errors,
1226 ),
1227 FlowNode::Try { body, catch, .. } => {
1228 collect_verdict_conds(
1229 body,
1230 step_outputs,
1231 verdict_contracts,
1232 referenced_values,
1233 errors,
1234 );
1235 collect_verdict_conds(
1236 catch,
1237 step_outputs,
1238 verdict_contracts,
1239 referenced_values,
1240 errors,
1241 );
1242 }
1243 FlowNode::Step { .. } | FlowNode::Assign { .. } => {}
1244 }
1245}
1246
1247fn lint_cond_expr(
1256 expr: &Expr,
1257 where_: &str,
1258 step_outputs: &HashMap<String, String>,
1259 verdict_contracts: &HashMap<String, VerdictContract>,
1260 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
1261 errors: &mut Vec<CompileError>,
1262) {
1263 match expr {
1264 Expr::Eq { lhs, rhs } | Expr::Ne { lhs, rhs } => {
1265 if let Some((path, lit)) = path_lit_operands(lhs, rhs) {
1266 resolve_and_check(
1267 path,
1268 &[lit],
1269 where_,
1270 step_outputs,
1271 verdict_contracts,
1272 referenced_values,
1273 errors,
1274 );
1275 }
1276 }
1277 Expr::In { needle, haystack } => {
1278 if let (
1279 Expr::Path { at },
1280 Expr::Lit {
1281 value: Value::Array(items),
1282 },
1283 ) = (needle.as_ref(), haystack.as_ref())
1284 {
1285 let lits: Vec<&Value> = items.iter().collect();
1286 resolve_and_check(
1287 at,
1288 &lits,
1289 where_,
1290 step_outputs,
1291 verdict_contracts,
1292 referenced_values,
1293 errors,
1294 );
1295 }
1296 }
1297 Expr::And { args } | Expr::Or { args } => {
1298 for a in args {
1299 lint_cond_expr(
1300 a,
1301 where_,
1302 step_outputs,
1303 verdict_contracts,
1304 referenced_values,
1305 errors,
1306 );
1307 }
1308 }
1309 Expr::Not { arg } => lint_cond_expr(
1310 arg,
1311 where_,
1312 step_outputs,
1313 verdict_contracts,
1314 referenced_values,
1315 errors,
1316 ),
1317 _ => {}
1318 }
1319}
1320
1321fn path_lit_operands<'a>(lhs: &'a Expr, rhs: &'a Expr) -> Option<(&'a Path, &'a Value)> {
1327 match (lhs, rhs) {
1328 (Expr::Path { at }, Expr::Lit { value }) => Some((at, value)),
1329 (Expr::Lit { value }, Expr::Path { at }) => Some((at, value)),
1330 _ => None,
1331 }
1332}
1333
1334fn resolve_and_check(
1349 path: &Path,
1350 lits: &[&Value],
1351 where_: &str,
1352 step_outputs: &HashMap<String, String>,
1353 verdict_contracts: &HashMap<String, VerdictContract>,
1354 referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
1355 errors: &mut Vec<CompileError>,
1356) {
1357 let path_str = path.to_string();
1358 let (agent, actual_shape) = if let Some(agent) = step_outputs.get(&path_str) {
1359 (agent, "body")
1360 } else if let Some(stripped) = path_str.strip_suffix(".parts.verdict") {
1361 match step_outputs.get(stripped) {
1362 Some(agent) => (agent, "part"),
1363 None => return,
1364 }
1365 } else {
1366 return;
1367 };
1368
1369 let Some(contract) = verdict_contracts.get(agent) else {
1370 tracing::warn!(
1371 agent = %agent,
1372 where_ = %where_,
1373 "cond references agent output but no verdict contract declared"
1374 );
1375 return;
1376 };
1377
1378 let expected_channel = match contract.channel {
1379 VerdictChannel::Body => "body",
1380 VerdictChannel::Part => "part",
1381 };
1382 if expected_channel != actual_shape {
1383 errors.push(CompileError::VerdictChannelMismatch {
1384 where_: where_.to_string(),
1385 agent: agent.clone(),
1386 expected_channel: expected_channel.to_string(),
1387 actual_shape: actual_shape.to_string(),
1388 });
1389 return;
1390 }
1391
1392 for lit in lits {
1393 let value_str = lit
1394 .as_str()
1395 .map(str::to_string)
1396 .unwrap_or_else(|| lit.to_string());
1397 if !contract.values.iter().any(|v| v == &value_str) {
1398 errors.push(CompileError::VerdictValueNotInContract {
1399 where_: where_.to_string(),
1400 agent: agent.clone(),
1401 value: value_str.clone(),
1402 values: contract.values.clone(),
1403 });
1404 }
1405 referenced_values
1412 .entry(agent.clone())
1413 .or_default()
1414 .insert(value_str);
1415 }
1416}
1417
1418fn check_unhandled_verdict_values(
1442 verdict_contracts: &HashMap<String, VerdictContract>,
1443 referenced_values: &HashMap<String, std::collections::HashSet<String>>,
1444 step_agents: &HashMap<String, String>,
1445 unhandled_gates: &UnhandledVerdictGates,
1446 errors: &mut Vec<CompileError>,
1447) {
1448 if unhandled_gates.all_silent() {
1449 return;
1450 }
1451 for finding in fold_unhandled_verdict_values(verdict_contracts, referenced_values, step_agents)
1452 {
1453 let gate = unhandled_gates.for_agent(&finding.agent);
1454 match gate {
1455 UnhandledVerdictGate::Deny => errors.push(CompileError::VerdictValueUnhandled {
1456 agent: finding.agent,
1457 value: finding.value,
1458 declared_values: finding.declared_values,
1459 step_ref: finding.step_ref,
1460 }),
1461 UnhandledVerdictGate::Warn => tracing::warn!(
1462 agent = %finding.agent,
1463 value = %finding.value,
1464 step_ref = %finding.step_ref,
1465 "declared verdict value has no downstream cond handler; \
1466 declare `metadata.lints = {{\"verdict-value-unhandled\": \"deny\"}}` \
1467 to reject at compile"
1468 ),
1469 UnhandledVerdictGate::Silence => {}
1473 }
1474 }
1475}
1476
1477#[derive(Debug, Clone, PartialEq, Eq)]
1488pub struct UnhandledVerdictValue {
1489 pub agent: String,
1491 pub value: String,
1493 pub declared_values: Vec<String>,
1495 pub step_ref: String,
1497}
1498
1499pub fn unhandled_verdict_values(
1516 flow: &FlowNode,
1517 verdict_contracts: &HashMap<String, VerdictContract>,
1518) -> Vec<UnhandledVerdictValue> {
1519 let mut step_outputs: HashMap<String, String> = HashMap::new();
1520 let mut step_agents: HashMap<String, String> = HashMap::new();
1521 collect_step_outputs_and_agents(flow, &mut step_outputs, &mut step_agents);
1522
1523 let mut referenced_values: HashMap<String, std::collections::HashSet<String>> = HashMap::new();
1524 let mut discarded_errors: Vec<CompileError> = Vec::new();
1525 collect_verdict_conds(
1526 flow,
1527 &step_outputs,
1528 verdict_contracts,
1529 &mut referenced_values,
1530 &mut discarded_errors,
1531 );
1532 fold_unhandled_verdict_values(verdict_contracts, &referenced_values, &step_agents)
1533}
1534
1535#[derive(Debug, Clone, PartialEq, Eq)]
1549pub struct AgentContractUnread {
1550 pub agent: String,
1552 pub declared_values: Vec<String>,
1554 pub step_ref: String,
1556}
1557
1558pub fn agents_with_all_verdict_values_unread(
1569 flow: &FlowNode,
1570 verdict_contracts: &HashMap<String, VerdictContract>,
1571) -> Vec<AgentContractUnread> {
1572 let per_value = unhandled_verdict_values(flow, verdict_contracts);
1573 let mut unread_counts: HashMap<String, usize> = HashMap::new();
1574 for finding in &per_value {
1575 *unread_counts.entry(finding.agent.clone()).or_default() += 1;
1576 }
1577 let mut agents: Vec<&String> = verdict_contracts.keys().collect();
1578 agents.sort();
1579 let mut out = Vec::new();
1580 for agent in agents {
1581 let contract = &verdict_contracts[agent];
1582 let declared = contract.values.len();
1583 if declared == 0 {
1584 continue;
1585 }
1586 let unread = unread_counts.get(agent).copied().unwrap_or(0);
1587 if unread != declared {
1588 continue;
1589 }
1590 let step_ref = per_value
1594 .iter()
1595 .find(|f| &f.agent == agent)
1596 .map(|f| f.step_ref.clone())
1597 .unwrap_or_else(|| agent.clone());
1598 out.push(AgentContractUnread {
1599 agent: agent.clone(),
1600 declared_values: contract.values.clone(),
1601 step_ref,
1602 });
1603 }
1604 out
1605}
1606
1607fn fold_unhandled_verdict_values(
1618 verdict_contracts: &HashMap<String, VerdictContract>,
1619 referenced_values: &HashMap<String, std::collections::HashSet<String>>,
1620 step_agents: &HashMap<String, String>,
1621) -> Vec<UnhandledVerdictValue> {
1622 let mut agents: Vec<&String> = verdict_contracts.keys().collect();
1623 agents.sort();
1624 let mut findings = Vec::new();
1625 for agent in agents {
1626 let contract = &verdict_contracts[agent];
1627 let referenced = referenced_values.get(agent);
1628 let step_ref = step_agents
1629 .get(agent)
1630 .cloned()
1631 .unwrap_or_else(|| agent.clone());
1632 for value in &contract.values {
1633 let handled = referenced.map(|set| set.contains(value)).unwrap_or(false);
1634 if handled {
1635 continue;
1636 }
1637 findings.push(UnhandledVerdictValue {
1638 agent: agent.clone(),
1639 value: value.clone(),
1640 declared_values: contract.values.clone(),
1641 step_ref: step_ref.clone(),
1642 });
1643 }
1644 }
1645 findings
1646}
1647
1648pub struct CompiledAgentTable {
1661 pub(crate) routes: HashMap<String, Arc<dyn SpawnerAdapter>>,
1662 pub(crate) default: Option<Arc<dyn SpawnerAdapter>>,
1663 pub(crate) verdict_contracts: HashMap<String, VerdictContract>,
1667}
1668
1669impl CompiledAgentTable {
1670 pub fn has_route(&self, agent: &str) -> bool {
1673 self.routes.contains_key(agent)
1674 }
1675 pub fn routed_agents(&self) -> Vec<String> {
1677 self.routes.keys().cloned().collect()
1678 }
1679 pub fn verdict_contract_for(&self, agent: &str) -> Option<&VerdictContract> {
1683 self.verdict_contracts.get(agent)
1684 }
1685}
1686
1687#[async_trait]
1688impl SpawnerAdapter for CompiledAgentTable {
1689 async fn spawn(
1690 &self,
1691 engine: &Engine,
1692 ctx: &Ctx,
1693 task_id: StepId,
1694 attempt: u32,
1695 token: CapToken,
1696 ) -> Result<Box<dyn Worker>, SpawnError> {
1697 let sp = self
1698 .routes
1699 .get(&ctx.agent)
1700 .cloned()
1701 .or_else(|| self.default.clone())
1702 .ok_or_else(|| SpawnError::NotRegistered(ctx.agent.clone()))?;
1703 sp.spawn(engine, ctx, task_id, attempt, token).await
1704 }
1705}
1706
1707pub struct SubprocessProcessSpawnerFactory;
1738
1739impl SpawnerFactoryKind for SubprocessProcessSpawnerFactory {
1740 const KIND: AgentKind = AgentKind::Subprocess;
1741 type Worker = crate::worker::process_spawner::ProcessWorker;
1742}
1743
1744pub const SUBPROCESS_TEMPLATE_HINT_KEY: &str = "subprocess_template";
1747pub const SUBPROCESS_OVERRIDES_HINT_KEY: &str = "subprocess_overrides";
1749
1750fn validate_embed_placeholders(s: &str, where_: &str) -> Result<(), String> {
1764 let mut rest = s;
1765 while let Some(start) = rest.find('{') {
1766 let after = &rest[start + 1..];
1767 let Some(end) = after.find('}') else {
1768 break;
1769 };
1770 let token = &after[..end];
1771 let is_candidate =
1772 !token.is_empty() && token.chars().all(|c| c.is_ascii_lowercase() || c == '_');
1773 if is_candidate {
1774 if !crate::worker::process_spawner::EMBED_PLACEHOLDERS.contains(&token) {
1775 return Err(format!(
1776 "unknown placeholder '{{{token}}}' in {where_}; closed set is \
1777 {{system, system_file, prompt, model, tools_csv, work_dir, task_id, attempt}}"
1778 ));
1779 }
1780 rest = &after[end + 1..];
1781 } else {
1782 rest = after;
1787 }
1788 }
1789 Ok(())
1790}
1791
1792fn resolve_subprocess_template_hint(
1798 bp: &Blueprint,
1799 ad: &AgentDef,
1800) -> Result<Option<Value>, CompileError> {
1801 let invalid = |msg: String| CompileError::InvalidSpec {
1802 name: ad.name.clone(),
1803 msg,
1804 };
1805 let runner = mlua_swarm_schema::resolve_runner(bp, ad).map_err(|e| invalid(e.to_string()))?;
1806 let Some(Runner::Subprocess {
1807 template,
1808 overrides,
1809 }) = runner
1810 else {
1811 return Ok(None);
1812 };
1813 let def = bp
1814 .subprocesses
1815 .iter()
1816 .find(|d| d.name == template)
1817 .ok_or_else(|| {
1818 let mut names: Vec<&str> = bp.subprocesses.iter().map(|d| d.name.as_str()).collect();
1819 names.sort_unstable();
1820 invalid(format!(
1821 "Runner::Subprocess template '{template}' not found in \
1822 Blueprint.subprocesses (defined: [{}])",
1823 names.join(", ")
1824 ))
1825 })?;
1826 Ok(Some(serde_json::json!({
1827 SUBPROCESS_TEMPLATE_HINT_KEY: def,
1828 SUBPROCESS_OVERRIDES_HINT_KEY: overrides,
1829 })))
1830}
1831
1832pub const OPERATOR_SID_PIN_HINT_KEY: &str = "operator_sid_pin";
1836
1837fn merge_operator_pin_hint(
1844 hint: Option<&Value>,
1845 pin: &str,
1846 agent: &str,
1847) -> Result<Value, CompileError> {
1848 let mut object = match hint {
1849 None => serde_json::Map::new(),
1850 Some(Value::Object(map)) => map.clone(),
1851 Some(other) => {
1852 return Err(CompileError::InvalidSpec {
1853 name: agent.to_string(),
1854 msg: format!(
1855 "hints.per_agent['{agent}'] must be a JSON object to carry the \
1856 run-scoped operator pin (got {other})"
1857 ),
1858 });
1859 }
1860 };
1861 object.insert(
1862 OPERATOR_SID_PIN_HINT_KEY.to_string(),
1863 Value::String(pin.to_string()),
1864 );
1865 Ok(Value::Object(object))
1866}
1867
1868impl SubprocessProcessSpawnerFactory {
1869 fn build_embed(
1874 agent_def: &AgentDef,
1875 template: &Value,
1876 overrides: Option<&Value>,
1877 ) -> Result<ProcessSpawner, CompileError> {
1878 use crate::worker::process_spawner::EmbedTemplate;
1879 use mlua_swarm_schema::{SubprocessDef, SubprocessOverrides};
1880
1881 let agent_name = &agent_def.name;
1882 let invalid = |msg: String| CompileError::InvalidSpec {
1883 name: agent_name.to_string(),
1884 msg,
1885 };
1886 let def: SubprocessDef = serde_json::from_value(template.clone())
1887 .map_err(|e| invalid(format!("subprocess_template hint: {e}")))?;
1888 let overrides: SubprocessOverrides = match overrides {
1889 Some(v) => serde_json::from_value(v.clone())
1890 .map_err(|e| invalid(format!("subprocess_overrides hint: {e}")))?,
1891 None => SubprocessOverrides::default(),
1892 };
1893
1894 if def.argv.is_empty() {
1895 return Err(invalid(format!(
1896 "SubprocessDef '{}': argv must not be empty",
1897 def.name
1898 )));
1899 }
1900 for (i, a) in def.argv.iter().enumerate() {
1902 validate_embed_placeholders(a, &format!("argv[{i}]")).map_err(&invalid)?;
1903 }
1904 if let Some(stdin) = &def.stdin {
1905 validate_embed_placeholders(stdin, "stdin").map_err(&invalid)?;
1906 }
1907 for (k, v) in &def.env {
1908 validate_embed_placeholders(v, &format!("env['{k}']")).map_err(&invalid)?;
1909 }
1910 if let Some(cwd) = &def.cwd {
1911 validate_embed_placeholders(cwd, "cwd").map_err(&invalid)?;
1912 }
1913 let stream_mode = match def.stream_mode.as_deref() {
1914 Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
1915 Some("sse_events") => Some(StreamMode::SseEvents),
1916 Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
1917 Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
1918 None => None,
1919 };
1920 if let Some(output) = &def.output {
1921 if stream_mode.is_some() {
1922 return Err(invalid(format!(
1923 "SubprocessDef '{}': output normalization is a plain-mode \
1924 declaration; remove either `output` or `stream_mode`",
1925 def.name
1926 )));
1927 }
1928 if let Some(format) = output.format.as_deref() {
1929 if format != "json" {
1930 return Err(invalid(format!(
1931 "SubprocessDef '{}': unknown output.format '{format}' \
1932 (supported: \"json\")",
1933 def.name
1934 )));
1935 }
1936 }
1937 if let Some(ptr) = output.result_ptr.as_deref() {
1938 if !ptr.starts_with('/') {
1939 return Err(invalid(format!(
1940 "SubprocessDef '{}': output.result_ptr '{ptr}' is not a \
1941 JSON Pointer (RFC 6901 — must start with '/')",
1942 def.name
1943 )));
1944 }
1945 }
1946 if let Some(ok_from) = output.ok_from.as_deref() {
1947 if ok_from != "exit_code" && !ok_from.starts_with('/') {
1948 return Err(invalid(format!(
1949 "SubprocessDef '{}': output.ok_from '{ok_from}' must be \
1950 \"exit_code\" or a JSON Pointer (starting with '/')",
1951 def.name
1952 )));
1953 }
1954 }
1955 }
1956
1957 let profile = agent_def.profile.as_ref();
1960 let system_prompt = profile
1961 .map(|p| p.system_prompt.clone())
1962 .filter(|s| !s.is_empty());
1963 let model = overrides
1964 .model
1965 .clone()
1966 .or_else(|| profile.and_then(|p| p.model.clone()));
1967 let tools: Vec<String> = if overrides.tools.is_empty() {
1968 profile.map(|p| p.tools.clone()).unwrap_or_default()
1969 } else {
1970 overrides.tools.clone()
1971 };
1972 let cwd = overrides.cwd.clone().or_else(|| def.cwd.clone());
1974 if let Some(c) = &cwd {
1975 validate_embed_placeholders(c, "overrides.cwd").map_err(&invalid)?;
1976 }
1977
1978 let program = def.argv[0].clone();
1979 let sp = ProcessSpawner {
1980 program,
1981 args: Vec::new(),
1982 use_stdin: def.stdin.is_some(),
1983 stream_mode,
1984 embed: Some(EmbedTemplate {
1985 argv: def.argv,
1986 stdin: def.stdin,
1987 env: def.env,
1988 cwd,
1989 output: def.output,
1990 system_prompt,
1991 model,
1992 tools_csv: tools.join(","),
1993 }),
1994 };
1995 Ok(sp)
1996 }
1997}
1998
1999impl SpawnerFactory for SubprocessProcessSpawnerFactory {
2000 fn build(
2001 &self,
2002 agent_def: &AgentDef,
2003 hint: Option<&Value>,
2004 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2005 if let Some(template) = hint.and_then(|h| h.get(SUBPROCESS_TEMPLATE_HINT_KEY)) {
2009 let overrides = hint.and_then(|h| h.get(SUBPROCESS_OVERRIDES_HINT_KEY));
2010 return Self::build_embed(agent_def, template, overrides).map(|sp| {
2011 let arc: Arc<dyn SpawnerAdapter> = Arc::new(sp);
2012 arc
2013 });
2014 }
2015 let agent_name = &agent_def.name;
2016 let spec = &agent_def.spec;
2017 let invalid = |msg: String| CompileError::InvalidSpec {
2018 name: agent_name.to_string(),
2019 msg,
2020 };
2021 let program = spec
2022 .get("program")
2023 .and_then(|v| v.as_str())
2024 .ok_or_else(|| invalid("shell spec: 'program' (string) required".into()))?
2025 .to_string();
2026 let args: Vec<String> = spec
2027 .get("args")
2028 .and_then(|v| v.as_array())
2029 .map(|a| {
2030 a.iter()
2031 .filter_map(|x| x.as_str().map(|s| s.to_string()))
2032 .collect()
2033 })
2034 .unwrap_or_default();
2035 let use_stdin = spec
2036 .get("use_stdin")
2037 .and_then(|v| v.as_bool())
2038 .unwrap_or(true);
2039 let stream_mode = match spec.get("stream_mode").and_then(|v| v.as_str()) {
2040 Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
2041 Some("sse_events") => Some(StreamMode::SseEvents),
2042 Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
2043 Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
2044 None => None,
2045 };
2046
2047 let mut sp = ProcessSpawner {
2048 program,
2049 args,
2050 use_stdin,
2051 stream_mode,
2052 embed: None,
2053 };
2054 if let Some(mode) = sp.stream_mode.clone() {
2055 sp = sp.stream_mode(mode);
2056 }
2057 Ok(Arc::new(sp))
2058 }
2059}
2060
2061pub struct LuaInProcessSpawnerFactory {
2092 registry: HashMap<String, WorkerFn>,
2093 bridges: HashMap<String, HostBridge>,
2094}
2095
2096#[derive(Clone)]
2108pub struct HostBridge(
2109 Arc<dyn Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync>,
2110);
2111
2112impl HostBridge {
2113 pub fn new<F>(f: F) -> Self
2115 where
2116 F: Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync + 'static,
2117 {
2118 Self(Arc::new(f))
2119 }
2120
2121 pub fn call(&self, arg: serde_json::Value) -> Result<serde_json::Value, String> {
2125 (self.0)(arg)
2126 }
2127}
2128
2129#[derive(Clone)]
2136pub struct LuaScriptSource {
2137 pub source: String,
2139 pub label: String,
2142}
2143
2144impl LuaScriptSource {
2145 pub fn new(source: impl Into<String>, label: impl Into<String>) -> Self {
2147 Self {
2148 source: source.into(),
2149 label: label.into(),
2150 }
2151 }
2152}
2153
2154impl LuaInProcessSpawnerFactory {
2155 pub fn new() -> Self {
2157 Self {
2158 registry: HashMap::new(),
2159 bridges: HashMap::new(),
2160 }
2161 }
2162
2163 pub fn with_bridge(mut self, name: impl Into<String>, bridge: HostBridge) -> Self {
2170 self.bridges.insert(name.into(), bridge);
2171 self
2172 }
2173
2174 pub fn register_lua(mut self, fn_id: impl Into<String>, source: LuaScriptSource) -> Self {
2192 let source = Arc::new(source);
2193 let bridges = Arc::new(self.bridges.clone());
2194 let wrapped: WorkerFn = Arc::new(move |inv| {
2195 let source = source.clone();
2196 let bridges = bridges.clone();
2197 Box::pin(run_lua_worker(source, bridges, inv))
2198 });
2199 self.registry.insert(fn_id.into(), wrapped);
2200 self
2201 }
2202}
2203
2204async fn run_lua_worker(
2206 source: Arc<LuaScriptSource>,
2207 bridges: Arc<HashMap<String, HostBridge>>,
2208 inv: crate::worker::adapter::WorkerInvocation,
2209) -> Result<crate::worker::adapter::WorkerResult, crate::worker::adapter::WorkerError> {
2210 use crate::worker::adapter::WorkerError;
2211 use mlua::LuaSerdeExt;
2212
2213 let label = source.label.clone();
2214 let outcome =
2215 tokio::task::spawn_blocking(move || -> Result<(serde_json::Value, bool), String> {
2216 let lua = mlua::Lua::new();
2217 let g = lua.globals();
2218
2219 g.set("_PROMPT", inv.prompt.clone())
2221 .map_err(|e| format!("set _PROMPT: {e}"))?;
2222 g.set("_AGENT", inv.agent.clone())
2223 .map_err(|e| format!("set _AGENT: {e}"))?;
2224 g.set("_TASK_ID", inv.task_id.to_string())
2225 .map_err(|e| format!("set _TASK_ID: {e}"))?;
2226 g.set("_ATTEMPT", inv.attempt as i64)
2227 .map_err(|e| format!("set _ATTEMPT: {e}"))?;
2228
2229 for (name, value) in
2238 crate::worker::agent_block::runtime::context_globals(inv.context.as_ref())
2239 {
2240 let lua_val = lua
2241 .to_value(&value)
2242 .map_err(|e| format!("{name} to_value: {e}"))?;
2243 g.set(name.as_str(), lua_val)
2244 .map_err(|e| format!("set {name}: {e}"))?;
2245 }
2246
2247 if let Ok(json_val) = serde_json::from_str::<serde_json::Value>(&inv.prompt) {
2249 let lua_val = lua
2250 .to_value(&json_val)
2251 .map_err(|e| format!("_CTX to_value: {e}"))?;
2252 g.set("_CTX", lua_val)
2253 .map_err(|e| format!("set _CTX: {e}"))?;
2254 }
2255
2256 if !bridges.is_empty() {
2258 let host = lua
2259 .create_table()
2260 .map_err(|e| format!("create host table: {e}"))?;
2261 for (name, bridge) in bridges.iter() {
2262 let bridge = bridge.clone();
2263 let bname = name.clone();
2264 let f = lua
2265 .create_function(move |lua, arg: mlua::Value| {
2266 let json_arg: serde_json::Value = lua.from_value(arg).map_err(|e| {
2267 mlua::Error::external(format!("bridge {bname} arg → json: {e}"))
2268 })?;
2269 let result_json =
2270 bridge.call(json_arg).map_err(mlua::Error::external)?;
2271 lua.to_value(&result_json).map_err(|e| {
2272 mlua::Error::external(format!("bridge {bname} ret → lua: {e}"))
2273 })
2274 })
2275 .map_err(|e| format!("create_function {name}: {e}"))?;
2276 host.set(name.as_str(), f)
2277 .map_err(|e| format!("host.{name} set: {e}"))?;
2278 }
2279 g.set("host", host).map_err(|e| format!("set host: {e}"))?;
2280 }
2281
2282 let result: mlua::Value = lua
2284 .load(&source.source)
2285 .set_name(&source.label)
2286 .eval()
2287 .map_err(|e| format!("lua eval [{}]: {e}", source.label))?;
2288
2289 let json_result: serde_json::Value = lua
2291 .from_value(result)
2292 .map_err(|e| format!("lua → json [{}]: {e}", source.label))?;
2293
2294 let (value, ok) = match &json_result {
2295 serde_json::Value::Object(map)
2296 if map.contains_key("value") || map.contains_key("ok") =>
2297 {
2298 let ok = map.get("ok").and_then(|v| v.as_bool()).unwrap_or(true);
2299 let value = map.get("value").cloned().unwrap_or(json_result.clone());
2300 (value, ok)
2301 }
2302 _ => (json_result, true),
2303 };
2304 Ok((value, ok))
2305 })
2306 .await
2307 .map_err(|e| WorkerError::Failed(format!("spawn_blocking join [{label}]: {e}")))?
2308 .map_err(WorkerError::Failed)?;
2309
2310 Ok(crate::worker::adapter::WorkerResult {
2311 value: outcome.0,
2312 ok: outcome.1,
2313 stats: None,
2314 }
2315 .ensure_worker_kind("lua"))
2316}
2317
2318impl Default for LuaInProcessSpawnerFactory {
2319 fn default() -> Self {
2320 Self::new()
2321 }
2322}
2323
2324impl SpawnerFactoryKind for LuaInProcessSpawnerFactory {
2325 const KIND: AgentKind = AgentKind::Lua;
2326 type Worker = LuaWorker;
2327}
2328
2329impl SpawnerFactory for LuaInProcessSpawnerFactory {
2330 fn build(
2331 &self,
2332 agent_def: &AgentDef,
2333 _hint: Option<&Value>,
2334 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2335 if let Some(source) = agent_def.spec.get("source").and_then(|v| v.as_str()) {
2341 let label = agent_def
2342 .spec
2343 .get("label")
2344 .and_then(|v| v.as_str())
2345 .map(str::to_string)
2346 .unwrap_or_else(|| format!("{}.lua", agent_def.name));
2347 let script = Arc::new(LuaScriptSource::new(source.to_string(), label));
2348 let bridges = Arc::new(self.bridges.clone());
2349 let wrapped: WorkerFn = Arc::new(move |inv| {
2350 let source = script.clone();
2351 let bridges = bridges.clone();
2352 Box::pin(run_lua_worker(source, bridges, inv))
2353 });
2354 let mut sp: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
2355 sp.registry.insert(agent_def.name.to_string(), wrapped);
2356 return Ok(Arc::new(sp));
2357 }
2358 build_inproc_from_registry::<LuaWorker>(&self.registry, agent_def, "lua")
2359 }
2360}
2361
2362pub struct RustFnInProcessSpawnerFactory {
2376 registry: HashMap<String, WorkerFn>,
2377}
2378
2379impl RustFnInProcessSpawnerFactory {
2380 pub fn new() -> Self {
2382 Self {
2383 registry: HashMap::new(),
2384 }
2385 }
2386
2387 pub fn register_fn<F, Fut>(mut self, fn_id: impl Into<String>, f: F) -> Self
2390 where
2391 F: Fn(crate::worker::adapter::WorkerInvocation) -> Fut + Send + Sync + 'static,
2392 Fut: std::future::Future<
2393 Output = Result<
2394 crate::worker::adapter::WorkerResult,
2395 crate::worker::adapter::WorkerError,
2396 >,
2397 > + Send
2398 + 'static,
2399 {
2400 let f = Arc::new(f);
2401 let wrapped: WorkerFn = Arc::new(move |inv| {
2402 let f = f.clone();
2403 Box::pin(f(inv))
2404 });
2405 self.registry.insert(fn_id.into(), wrapped);
2406 self
2407 }
2408}
2409
2410impl Default for RustFnInProcessSpawnerFactory {
2411 fn default() -> Self {
2412 Self::new()
2413 }
2414}
2415
2416impl SpawnerFactoryKind for RustFnInProcessSpawnerFactory {
2417 const KIND: AgentKind = AgentKind::RustFn;
2418 type Worker = RustFnWorker;
2419}
2420
2421impl SpawnerFactory for RustFnInProcessSpawnerFactory {
2422 fn build(
2423 &self,
2424 agent_def: &AgentDef,
2425 _hint: Option<&Value>,
2426 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2427 build_inproc_from_registry::<RustFnWorker>(&self.registry, agent_def, "rust_fn")
2428 }
2429}
2430
2431fn build_inproc_from_registry<W>(
2437 registry: &HashMap<String, WorkerFn>,
2438 agent_def: &AgentDef,
2439 kind_label: &str,
2440) -> Result<Arc<dyn SpawnerAdapter>, CompileError>
2441where
2442 W: crate::worker::Worker + From<crate::worker::WorkerJoinHandler> + Send + Sync + 'static,
2443{
2444 let agent_name = &agent_def.name;
2445 let spec = &agent_def.spec;
2446 let invalid = |msg: String| CompileError::InvalidSpec {
2447 name: agent_name.to_string(),
2448 msg,
2449 };
2450 let fn_id = spec
2451 .get("fn_id")
2452 .and_then(|v| v.as_str())
2453 .ok_or_else(|| invalid(format!("{kind_label} spec: 'fn_id' (string) required")))?;
2454 let f = registry
2455 .get(fn_id)
2456 .cloned()
2457 .ok_or_else(|| invalid(format!("fn_id '{fn_id}' not registered in factory")))?;
2458 let mut sp: InProcSpawner<W> = InProcSpawner::<W>::typed();
2459 sp.registry.insert(agent_name.to_string(), f);
2463 Ok(Arc::new(sp))
2464}
2465
2466pub struct LuaWorker {
2471 pub handler: crate::worker::WorkerJoinHandler,
2473}
2474
2475impl From<crate::worker::WorkerJoinHandler> for LuaWorker {
2476 fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
2477 Self { handler }
2478 }
2479}
2480
2481#[async_trait::async_trait]
2482impl crate::worker::Worker for LuaWorker {
2483 fn id(&self) -> &crate::types::WorkerId {
2484 &self.handler.worker_id
2485 }
2486 fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
2487 self.handler.cancel.clone()
2488 }
2489 async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
2490 self.handler.await_completion().await
2491 }
2492}
2493
2494pub struct RustFnWorker {
2499 pub handler: crate::worker::WorkerJoinHandler,
2501}
2502
2503impl From<crate::worker::WorkerJoinHandler> for RustFnWorker {
2504 fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
2505 Self { handler }
2506 }
2507}
2508
2509#[async_trait::async_trait]
2510impl crate::worker::Worker for RustFnWorker {
2511 fn id(&self) -> &crate::types::WorkerId {
2512 &self.handler.worker_id
2513 }
2514 fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
2515 self.handler.cancel.clone()
2516 }
2517 async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
2518 self.handler.await_completion().await
2519 }
2520}
2521
2522pub struct OperatorSpawnerFactory {
2586 operators: Arc<std::sync::RwLock<HashMap<String, Arc<dyn Operator>>>>,
2587}
2588
2589impl OperatorSpawnerFactory {
2590 pub fn new() -> Self {
2592 Self {
2593 operators: Arc::new(std::sync::RwLock::new(HashMap::new())),
2594 }
2595 }
2596
2597 pub fn register_operator(&self, id: impl Into<String>, op: Arc<dyn Operator>) -> &Self {
2603 self.operators
2604 .write()
2605 .expect("OperatorSpawnerFactory.operators RwLock poisoned")
2606 .insert(id.into(), op);
2607 self
2608 }
2609
2610 pub fn unregister_operator(&self, id: &str) -> &Self {
2613 self.operators
2614 .write()
2615 .expect("OperatorSpawnerFactory.operators RwLock poisoned")
2616 .remove(id);
2617 self
2618 }
2619}
2620
2621impl Default for OperatorSpawnerFactory {
2622 fn default() -> Self {
2623 Self::new()
2624 }
2625}
2626
2627impl SpawnerFactoryKind for OperatorSpawnerFactory {
2628 const KIND: AgentKind = AgentKind::Operator;
2629 type Worker = crate::operator::OperatorWorker;
2630}
2631
2632impl SpawnerFactory for OperatorSpawnerFactory {
2633 fn build(
2634 &self,
2635 agent_def: &AgentDef,
2636 hint: Option<&Value>,
2637 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
2638 let agent_name = &agent_def.name;
2639 let spec = &agent_def.spec;
2640 let system_prompt = agent_def.profile.as_ref().map(|p| p.system_prompt.clone());
2646 let invalid = |msg: String| CompileError::InvalidSpec {
2647 name: agent_name.to_string(),
2648 msg,
2649 };
2650 let op_ref = spec
2651 .get("operator_ref")
2652 .and_then(|v| v.as_str())
2653 .ok_or_else(|| invalid("operator spec: 'operator_ref' (string) required".into()))?;
2654 let pin = hint
2661 .and_then(|h| h.get(OPERATOR_SID_PIN_HINT_KEY))
2662 .and_then(|v| v.as_str());
2663 let lookup_key = pin.unwrap_or(op_ref);
2664 let operators = self
2665 .operators
2666 .read()
2667 .expect("OperatorSpawnerFactory.operators RwLock poisoned");
2668 let op = operators.get(lookup_key).cloned().ok_or_else(|| {
2669 let mut names: Vec<String> = operators.keys().cloned().collect();
2670 names.sort();
2671 let names_list = if names.is_empty() {
2672 "<none>".to_string()
2673 } else {
2674 names.join(", ")
2675 };
2676 match pin {
2677 Some(pin) => invalid(format!(
2678 "run-scoped operator pin '{pin}' (declared operator_ref '{op_ref}') \
2679 is not registered in factory. \
2680 Registered ids: [{names_list}]. \
2681 The launch pinned this run to that session; resolving \
2682 '{op_ref}' through whichever session currently holds the \
2683 role would send this run's Spawn frames somewhere the \
2684 caller did not ask for, so the launch fails instead. \
2685 Hint: the pinned session must be joined (and still live) \
2686 before launching."
2687 )),
2688 None => invalid(format!(
2689 "operator_ref '{op_ref}' not registered in factory. \
2690 Registered sids: [{names_list}]. \
2691 Hint: call mse_operator_join(roles=[...]) to mint the sid first."
2692 )),
2693 }
2694 })?;
2695 drop(operators);
2696
2697 let worker_binding = agent_def
2704 .profile
2705 .as_ref()
2706 .and_then(|p| p.worker_binding.as_ref())
2707 .map(|variant| WorkerBinding {
2708 variant: variant.clone(),
2709 tools: agent_def
2710 .profile
2711 .as_ref()
2712 .map(|p| p.tools.clone())
2713 .unwrap_or_default(),
2714 request_digest: None,
2718 requested_model: None,
2719 });
2720 if op.requires_worker_binding() && worker_binding.is_none() {
2721 return Err(invalid(format!(
2728 "{WORKER_BINDING_REQUIRED_MSG_PREFIX}. \
2729 Fix by either: \
2730 (a) if authoring the Blueprint JSON directly, add \
2731 `agents[N].profile.worker_binding: \"<subagent-type>\"` \
2732 to the JSON literal; or \
2733 (b) if using an $agent_md file ref, add \
2734 `worker_binding: <subagent-type>` to the agent .md frontmatter."
2735 )));
2736 }
2737 Ok(Arc::new(OperatorSpawner::new(
2738 op,
2739 system_prompt,
2740 worker_binding,
2741 )))
2742 }
2743}
2744
2745#[cfg(test)]
2746mod operator_spawner_factory_worker_binding_tests {
2747 use super::*;
2748 use crate::blueprint::AgentProfile;
2749 use crate::core::ctx::Ctx;
2750 use crate::types::CapToken;
2751 use crate::worker::adapter::{WorkerError, WorkerResult};
2752
2753 struct StubOperator {
2758 requires_binding: bool,
2759 }
2760
2761 #[async_trait]
2762 impl Operator for StubOperator {
2763 async fn execute(
2764 &self,
2765 _ctx: &Ctx,
2766 _system: Option<String>,
2767 _prompt: Value,
2768 _worker: Option<WorkerBinding>,
2769 _worker_token: CapToken,
2770 ) -> Result<WorkerResult, WorkerError> {
2771 Ok(WorkerResult {
2772 value: Value::Null,
2773 ok: true,
2774 stats: None,
2775 })
2776 }
2777
2778 fn requires_worker_binding(&self) -> bool {
2779 self.requires_binding
2780 }
2781 }
2782
2783 fn agent_def_with(profile: Option<AgentProfile>) -> AgentDef {
2784 AgentDef {
2785 name: "test-agent".to_string(),
2786 kind: AgentKind::Operator,
2787 spec: serde_json::json!({ "operator_ref": "op1" }),
2788 profile,
2789 meta: None,
2790 runner: None,
2791 runner_ref: None,
2792 verdict: None,
2793 lints: None,
2794 }
2795 }
2796
2797 #[test]
2798 fn build_fails_loud_when_binding_required_but_absent() {
2799 let factory = OperatorSpawnerFactory::new();
2800 factory.register_operator(
2801 "op1",
2802 Arc::new(StubOperator {
2803 requires_binding: true,
2804 }) as Arc<dyn Operator>,
2805 );
2806 let def = agent_def_with(Some(AgentProfile::default()));
2807 match factory.build(&def, None) {
2808 Err(CompileError::InvalidSpec { name, msg }) => {
2809 assert_eq!(name, "test-agent");
2810 assert!(
2811 msg.contains("worker_binding is required"),
2812 "unexpected message: {msg}"
2813 );
2814 assert!(
2818 msg.contains("agents[N].profile.worker_binding"),
2819 "message missing JSON-direct hint (issue #9): {msg}"
2820 );
2821 assert!(
2822 msg.contains("agent .md frontmatter"),
2823 "message missing $agent_md hint: {msg}"
2824 );
2825 }
2826 Err(other) => panic!("expected InvalidSpec, got: {other:?}"),
2827 Ok(_) => panic!("expected compile-time failure, got Ok"),
2828 }
2829 }
2830
2831 #[test]
2838 fn factory_error_message_carries_the_shared_prefix_and_specializes_the_diagnostic() {
2839 let factory = OperatorSpawnerFactory::new();
2840 factory.register_operator(
2841 "op1",
2842 Arc::new(StubOperator {
2843 requires_binding: true,
2844 }) as Arc<dyn Operator>,
2845 );
2846 let def = agent_def_with(Some(AgentProfile::default()));
2847 let err = match factory.build(&def, None) {
2848 Err(err) => err,
2849 Ok(_) => panic!("expected compile-time failure, got Ok"),
2850 };
2851 match &err {
2852 CompileError::InvalidSpec { msg, .. } => {
2853 assert!(
2854 msg.starts_with(WORKER_BINDING_REQUIRED_MSG_PREFIX),
2855 "factory message must start with the shared prefix, got: {msg}"
2856 );
2857 }
2858 other => panic!("expected InvalidSpec, got: {other:?}"),
2859 }
2860 let d = mlua_swarm_diag::Diagnostic::from(&err);
2861 assert_eq!(d.kind, "worker-binding-missing");
2862 }
2863
2864 #[test]
2865 fn build_succeeds_when_binding_required_and_present() {
2866 let factory = OperatorSpawnerFactory::new();
2867 factory.register_operator(
2868 "op1",
2869 Arc::new(StubOperator {
2870 requires_binding: true,
2871 }) as Arc<dyn Operator>,
2872 );
2873 let profile = AgentProfile {
2874 worker_binding: Some("code-worker".to_string()),
2875 tools: vec!["Read".to_string(), "Edit".to_string()],
2876 ..Default::default()
2877 };
2878 let def = agent_def_with(Some(profile));
2879 assert!(
2880 factory.build(&def, None).is_ok(),
2881 "expected Ok when worker_binding is declared"
2882 );
2883 }
2884
2885 #[test]
2886 fn build_succeeds_when_binding_not_required_and_absent() {
2887 let factory = OperatorSpawnerFactory::new();
2888 factory.register_operator(
2889 "op1",
2890 Arc::new(StubOperator {
2891 requires_binding: false,
2892 }) as Arc<dyn Operator>,
2893 );
2894 let def = agent_def_with(Some(AgentProfile::default()));
2895 assert!(
2896 factory.build(&def, None).is_ok(),
2897 "backends that don't require a binding must not be gated by its absence"
2898 );
2899 }
2900}
2901
2902#[cfg(test)]
2910mod lua_inline_source_tests {
2911 use super::*;
2912 use crate::types::{CapToken, Role, StepId};
2913
2914 fn agent(name: &str, spec: Value) -> AgentDef {
2915 AgentDef {
2916 name: name.to_string(),
2917 kind: AgentKind::Lua,
2918 spec,
2919 profile: None,
2920 meta: None,
2921 runner: None,
2922 runner_ref: None,
2923 verdict: None,
2924 lints: None,
2925 }
2926 }
2927
2928 fn test_invocation(prompt: &str) -> crate::worker::adapter::WorkerInvocation {
2929 crate::worker::adapter::WorkerInvocation::new(
2930 CapToken {
2931 agent_id: "a".into(),
2932 role: Role::Worker,
2933 scopes: vec!["*".into()],
2934 issued_at: 0,
2935 expire_at: u64::MAX / 2,
2936 max_uses: None,
2937 nonce: "test-nonce".into(),
2938 sig_hex: "".into(),
2939 },
2940 StepId::parse("ST-test").expect("StepId parse"),
2941 1,
2942 "g",
2943 prompt,
2944 )
2945 }
2946
2947 #[test]
2948 fn build_accepts_inline_source_without_pre_registration() {
2949 let factory = LuaInProcessSpawnerFactory::new();
2950 let def = agent(
2951 "g",
2952 serde_json::json!({ "source": "return { value = 42, ok = true }" }),
2953 );
2954 assert!(
2955 factory.build(&def, None).is_ok(),
2956 "inline spec.source must build without a pre-registered fn_id"
2957 );
2958 }
2959
2960 #[test]
2961 fn build_rejects_when_neither_source_nor_fn_id_is_present() {
2962 let factory = LuaInProcessSpawnerFactory::new();
2963 let def = agent("g", serde_json::json!({}));
2964 match factory.build(&def, None) {
2965 Err(CompileError::InvalidSpec { msg, .. }) => {
2966 assert!(
2967 msg.contains("fn_id"),
2968 "empty spec must still surface the fn_id-required message: {msg}"
2969 );
2970 }
2971 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
2972 Ok(_) => panic!("expected InvalidSpec, got Ok(SpawnerAdapter)"),
2975 }
2976 }
2977
2978 #[tokio::test]
2982 async fn inline_source_evaluates_and_marshals_result() {
2983 let source =
2984 LuaScriptSource::new("return { value = _PROMPT .. '!', ok = true }", "smoke.lua");
2985 let out = run_lua_worker(
2986 std::sync::Arc::new(source),
2987 std::sync::Arc::new(HashMap::new()),
2988 test_invocation("hello"),
2989 )
2990 .await
2991 .expect("lua worker ok");
2992 assert_eq!(out.value, serde_json::json!("hello!"));
2993 assert!(out.ok);
2994 }
2995
2996 #[tokio::test]
2997 async fn inline_source_can_signal_agent_level_failure() {
2998 let source = LuaScriptSource::new("return { value = 'nope', ok = false }", "gate.lua");
3001 let out = run_lua_worker(
3002 std::sync::Arc::new(source),
3003 std::sync::Arc::new(HashMap::new()),
3004 test_invocation("input"),
3005 )
3006 .await
3007 .expect("lua worker ok");
3008 assert_eq!(out.value, serde_json::json!("nope"));
3009 assert!(!out.ok);
3010 }
3011}
3012
3013#[cfg(test)]
3016mod meta_ref_validation_tests {
3017 use super::*;
3018 use crate::blueprint::{AgentMeta, MetaDef};
3019 use crate::worker::adapter::WorkerResult;
3020
3021 fn registry_with_echo() -> SpawnerRegistry {
3022 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
3023 Ok(WorkerResult {
3024 value: Value::String(inv.prompt),
3025 ok: true,
3026 stats: None,
3027 })
3028 });
3029 let mut reg = SpawnerRegistry::new();
3030 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
3031 reg
3032 }
3033
3034 fn rustfn_agent(name: &str) -> AgentDef {
3035 AgentDef {
3036 name: name.to_string(),
3037 kind: AgentKind::RustFn,
3038 spec: serde_json::json!({ "fn_id": "echo" }),
3039 profile: None,
3040 meta: None,
3041 runner: None,
3042 runner_ref: None,
3043 verdict: None,
3044 lints: None,
3045 }
3046 }
3047
3048 fn simple_flow(agent_ref: &str, in_: Expr) -> FlowNode {
3049 FlowNode::Step {
3050 ref_: agent_ref.to_string(),
3051 in_,
3052 out: Expr::Path {
3053 at: "$.output".parse().expect("literal test path: $.output"),
3054 },
3055 }
3056 }
3057
3058 fn minimal_bp(agents: Vec<AgentDef>, metas: Vec<MetaDef>, flow: FlowNode) -> Blueprint {
3059 Blueprint {
3060 schema_version: crate::blueprint::current_schema_version(),
3061 id: "meta-ref-ut".into(),
3062 flow,
3063 agents,
3064 operators: vec![],
3065 metas,
3066 hints: Default::default(),
3067 strategy: Default::default(),
3068 metadata: BlueprintMetadata::default(),
3069 spawner_hints: Default::default(),
3070 default_agent_kind: AgentKind::Operator,
3071 default_operator_kind: None,
3072 default_init_ctx: None,
3073 default_agent_ctx: None,
3074 default_context_policy: None,
3075 projection_placement: None,
3076 audits: vec![],
3077 degradation_policy: None,
3078 runners: vec![],
3079 default_runner: None,
3080 subprocesses: vec![],
3081 check_policy: None,
3082 blueprint_ref_includes: Vec::new(),
3083 }
3084 }
3085
3086 #[test]
3087 fn valid_meta_ref_compiles() {
3088 let mut agent = rustfn_agent("worker");
3089 agent.meta = Some(AgentMeta {
3090 meta_ref: Some("shared".to_string()),
3091 ..Default::default()
3092 });
3093 let bp = minimal_bp(
3094 vec![agent],
3095 vec![MetaDef {
3096 name: "shared".into(),
3097 ctx: serde_json::json!({ "k": "v" }),
3098 }],
3099 simple_flow(
3100 "worker",
3101 Expr::Path {
3102 at: "$.input".parse().expect("literal test path: $.input"),
3103 },
3104 ),
3105 );
3106 let compiler = Compiler::new(registry_with_echo());
3107 assert!(
3108 compiler.compile(&bp).is_ok(),
3109 "a resolvable AgentMeta.meta_ref must compile"
3110 );
3111 }
3112
3113 #[test]
3114 fn unknown_agent_meta_ref_is_unresolved_meta_ref() {
3115 let mut agent = rustfn_agent("worker");
3116 agent.meta = Some(AgentMeta {
3117 meta_ref: Some("missing".to_string()),
3118 ..Default::default()
3119 });
3120 let bp = minimal_bp(
3121 vec![agent],
3122 vec![],
3123 simple_flow(
3124 "worker",
3125 Expr::Path {
3126 at: "$.input".parse().expect("literal test path: $.input"),
3127 },
3128 ),
3129 );
3130 let compiler = Compiler::new(registry_with_echo());
3131 match compiler.compile(&bp) {
3132 Err(CompileError::UnresolvedMetaRef {
3133 where_,
3134 meta_ref,
3135 defined,
3136 }) => {
3137 assert!(
3138 where_.contains("worker"),
3139 "where_ must name the agent: {where_}"
3140 );
3141 assert_eq!(meta_ref, "missing");
3142 assert!(defined.is_empty());
3143 }
3144 Err(other) => {
3145 panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
3146 }
3147 Ok(_) => panic!("expected compile-time failure, got Ok"),
3148 }
3149 }
3150
3151 #[test]
3152 fn unknown_static_step_meta_ref_in_lit_is_unresolved_meta_ref() {
3153 let agent = rustfn_agent("worker");
3154 let in_ = Expr::Lit {
3155 value: serde_json::json!({ "$step_meta": { "ref": "missing" }, "$in": "go" }),
3156 };
3157 let bp = minimal_bp(vec![agent], vec![], simple_flow("worker", in_));
3158 let compiler = Compiler::new(registry_with_echo());
3159 match compiler.compile(&bp) {
3160 Err(CompileError::UnresolvedMetaRef {
3161 where_, meta_ref, ..
3162 }) => {
3163 assert!(
3164 where_.contains("worker"),
3165 "where_ must name the offending step: {where_}"
3166 );
3167 assert_eq!(meta_ref, "missing");
3168 }
3169 Err(other) => {
3170 panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
3171 }
3172 Ok(_) => panic!("expected compile-time failure, got Ok"),
3173 }
3174 }
3175
3176 #[test]
3177 fn path_op_input_with_no_static_envelope_compiles_fine() {
3178 let agent = rustfn_agent("worker");
3179 let bp = minimal_bp(
3180 vec![agent],
3181 vec![],
3182 simple_flow(
3183 "worker",
3184 Expr::Path {
3185 at: "$.input".parse().expect("literal test path: $.input"),
3186 },
3187 ),
3188 );
3189 let compiler = Compiler::new(registry_with_echo());
3190 assert!(
3191 compiler.compile(&bp).is_ok(),
3192 "a non-Lit Step.in must not trigger the best-effort static $step_meta check"
3193 );
3194 }
3195}
3196
3197#[cfg(test)]
3199mod audit_agent_validation_tests {
3200 use super::*;
3201 use crate::worker::adapter::WorkerResult;
3202 use mlua_swarm_schema::{AuditDef, AuditMode};
3203
3204 fn registry_with_echo() -> SpawnerRegistry {
3205 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
3206 Ok(WorkerResult {
3207 value: Value::String(inv.prompt),
3208 ok: true,
3209 stats: None,
3210 })
3211 });
3212 let mut reg = SpawnerRegistry::new();
3213 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
3214 reg
3215 }
3216
3217 fn rustfn_agent(name: &str) -> AgentDef {
3218 AgentDef {
3219 name: name.to_string(),
3220 kind: AgentKind::RustFn,
3221 spec: serde_json::json!({ "fn_id": "echo" }),
3222 profile: None,
3223 meta: None,
3224 runner: None,
3225 runner_ref: None,
3226 verdict: None,
3227 lints: None,
3228 }
3229 }
3230
3231 fn minimal_bp(agents: Vec<AgentDef>, audits: Vec<AuditDef>) -> Blueprint {
3232 Blueprint {
3233 schema_version: crate::blueprint::current_schema_version(),
3234 id: "audit-ref-ut".into(),
3235 flow: FlowNode::Step {
3236 ref_: "worker".to_string(),
3237 in_: Expr::Path {
3238 at: "$.input".parse().expect("literal test path: $.input"),
3239 },
3240 out: Expr::Path {
3241 at: "$.output".parse().expect("literal test path: $.output"),
3242 },
3243 },
3244 agents,
3245 operators: vec![],
3246 metas: vec![],
3247 hints: Default::default(),
3248 strategy: Default::default(),
3249 metadata: BlueprintMetadata::default(),
3250 spawner_hints: Default::default(),
3251 default_agent_kind: AgentKind::Operator,
3252 default_operator_kind: None,
3253 default_init_ctx: None,
3254 default_agent_ctx: None,
3255 default_context_policy: None,
3256 projection_placement: None,
3257 audits,
3258 degradation_policy: None,
3259 runners: vec![],
3260 default_runner: None,
3261 subprocesses: vec![],
3262 check_policy: None,
3263 blueprint_ref_includes: Vec::new(),
3264 }
3265 }
3266
3267 #[test]
3268 fn unresolved_audit_agent_is_a_loud_compile_error() {
3269 let bp = minimal_bp(
3270 vec![rustfn_agent("worker")],
3271 vec![AuditDef {
3272 agent: "missing-auditor".to_string(),
3273 steps: None,
3274 mode: AuditMode::default(),
3275 }],
3276 );
3277 let compiler = Compiler::new(registry_with_echo());
3278 match compiler.compile(&bp) {
3279 Err(CompileError::UnresolvedAuditAgent { agent, defined }) => {
3280 assert_eq!(agent, "missing-auditor");
3281 assert_eq!(defined, vec!["worker".to_string()]);
3282 }
3283 Err(other) => {
3284 panic!("expected UnresolvedAuditAgent, got a different CompileError: {other}")
3285 }
3286 Ok(_) => panic!("expected compile-time failure, got Ok"),
3287 }
3288 }
3289
3290 #[test]
3291 fn resolved_audit_agent_compiles_fine() {
3292 let bp = minimal_bp(
3293 vec![rustfn_agent("worker"), rustfn_agent("auditor")],
3294 vec![AuditDef {
3295 agent: "auditor".to_string(),
3296 steps: None,
3297 mode: AuditMode::default(),
3298 }],
3299 );
3300 let compiler = Compiler::new(registry_with_echo());
3301 assert!(
3302 compiler.compile(&bp).is_ok(),
3303 "an audits[].agent that names a declared AgentDef must compile"
3304 );
3305 }
3306}
3307
3308#[cfg(test)]
3316mod operator_run_pin_tests {
3317 use super::*;
3318 use crate::core::ctx::Ctx;
3319 use crate::types::CapToken;
3320 use crate::worker::adapter::{WorkerError, WorkerResult};
3321 use std::sync::Mutex;
3322
3323 type Seen = Arc<Mutex<Vec<(String, Option<Value>)>>>;
3325
3326 struct RecordingOperatorFactory {
3330 seen: Seen,
3331 }
3332
3333 impl SpawnerFactory for RecordingOperatorFactory {
3334 fn build(
3335 &self,
3336 agent_def: &AgentDef,
3337 hint: Option<&Value>,
3338 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
3339 self.seen
3340 .lock()
3341 .expect("RecordingOperatorFactory.seen poisoned")
3342 .push((agent_def.name.clone(), hint.cloned()));
3343 let mut spawner: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
3344 let worker: WorkerFn = Arc::new(|_inv| {
3345 Box::pin(async move {
3346 Ok(WorkerResult {
3347 value: Value::Null,
3348 ok: true,
3349 stats: None,
3350 })
3351 })
3352 });
3353 spawner.registry.insert(agent_def.name.clone(), worker);
3354 Ok(Arc::new(spawner))
3355 }
3356 }
3357
3358 impl SpawnerFactoryKind for RecordingOperatorFactory {
3359 const KIND: AgentKind = AgentKind::Operator;
3360 type Worker = crate::operator::OperatorWorker;
3361 }
3362
3363 struct RecordingLuaFactory {
3366 seen: Seen,
3367 }
3368
3369 impl SpawnerFactory for RecordingLuaFactory {
3370 fn build(
3371 &self,
3372 agent_def: &AgentDef,
3373 hint: Option<&Value>,
3374 ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
3375 self.seen
3376 .lock()
3377 .expect("RecordingLuaFactory.seen poisoned")
3378 .push((agent_def.name.clone(), hint.cloned()));
3379 let mut spawner: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
3380 let worker: WorkerFn = Arc::new(|_inv| {
3381 Box::pin(async move {
3382 Ok(WorkerResult {
3383 value: Value::Null,
3384 ok: true,
3385 stats: None,
3386 })
3387 })
3388 });
3389 spawner.registry.insert(agent_def.name.clone(), worker);
3390 Ok(Arc::new(spawner))
3391 }
3392 }
3393
3394 impl SpawnerFactoryKind for RecordingLuaFactory {
3395 const KIND: AgentKind = AgentKind::Lua;
3396 type Worker = LuaWorker;
3397 }
3398
3399 fn recording_compiler() -> (Compiler, Seen, Seen) {
3400 let operator_seen: Seen = Arc::new(Mutex::new(Vec::new()));
3401 let lua_seen: Seen = Arc::new(Mutex::new(Vec::new()));
3402 let mut registry = SpawnerRegistry::new();
3403 registry.register::<RecordingOperatorFactory>(Arc::new(RecordingOperatorFactory {
3404 seen: operator_seen.clone(),
3405 }));
3406 registry.register::<RecordingLuaFactory>(Arc::new(RecordingLuaFactory {
3407 seen: lua_seen.clone(),
3408 }));
3409 (Compiler::new(registry), operator_seen, lua_seen)
3410 }
3411
3412 fn bp_with_operator_and_lua_agents() -> Blueprint {
3416 serde_json::from_value(serde_json::json!({
3417 "schema_version": crate::blueprint::current_schema_version(),
3418 "id": "operator-pin-ut",
3419 "flow": {
3420 "kind": "step",
3421 "ref": "planner",
3422 "in": { "op": "path", "at": "$.input" },
3423 "out": { "op": "path", "at": "$.output" }
3424 },
3425 "agents": [
3426 {
3427 "name": "planner",
3428 "kind": "operator",
3429 "spec": { "operator_ref": "main-ai" }
3430 },
3431 {
3432 "name": "scorer",
3433 "kind": "lua",
3434 "spec": { "source": "return { value = 1, ok = true }" }
3435 }
3436 ],
3437 "operators": [{ "name": "main-ai" }],
3438 "hints": { "per_agent": { "planner": { "authored": "keep-me" } } },
3439 "strategy": { "strict_refs": false }
3440 }))
3441 .expect("test Blueprint literal")
3442 }
3443
3444 fn hint_for(seen: &Seen, agent: &str) -> Option<Value> {
3445 seen.lock()
3446 .expect("seen poisoned")
3447 .iter()
3448 .find(|(name, _)| name == agent)
3449 .map(|(_, hint)| hint.clone())
3450 .expect("agent was never built")
3451 }
3452
3453 #[test]
3456 fn pinned_compile_hands_the_operator_factory_the_pinned_sid() {
3457 let (compiler, operator_seen, _lua_seen) = recording_compiler();
3458 let bp = bp_with_operator_and_lua_agents();
3459 let bound = resolve_bound_agents(&bp).expect("resolve bound agents");
3460 compiler
3461 .compile_bound_pinned(&bp, &bound, Some("S-pinned"))
3462 .expect("pinned compile");
3463
3464 let hint = hint_for(&operator_seen, "planner").expect("planner hint");
3465 assert_eq!(
3466 hint.get(OPERATOR_SID_PIN_HINT_KEY),
3467 Some(&Value::String("S-pinned".to_string())),
3468 "the pin must reach the factory as a build hint: {hint}"
3469 );
3470 assert_eq!(
3471 hint.get("authored"),
3472 Some(&Value::String("keep-me".to_string())),
3473 "merging the pin must not drop the author's declared hint: {hint}"
3474 );
3475 }
3476
3477 #[test]
3481 fn unpinned_compile_passes_the_authored_hint_through_untouched() {
3482 let (compiler, operator_seen, lua_seen) = recording_compiler();
3483 let bp = bp_with_operator_and_lua_agents();
3484 let bound = resolve_bound_agents(&bp).expect("resolve bound agents");
3485 compiler
3486 .compile_bound(&bp, &bound)
3487 .expect("unpinned compile");
3488
3489 assert_eq!(
3490 hint_for(&operator_seen, "planner"),
3491 Some(serde_json::json!({ "authored": "keep-me" })),
3492 "an unpinned compile must hand over the authored hint verbatim"
3493 );
3494 assert_eq!(
3495 hint_for(&lua_seen, "scorer"),
3496 None,
3497 "an agent with no authored hint must still be built with None"
3498 );
3499 }
3500
3501 #[test]
3505 fn pin_does_not_leak_onto_non_operator_agents() {
3506 let (compiler, _operator_seen, lua_seen) = recording_compiler();
3507 let bp = bp_with_operator_and_lua_agents();
3508 let bound = resolve_bound_agents(&bp).expect("resolve bound agents");
3509 compiler
3510 .compile_bound_pinned(&bp, &bound, Some("S-pinned"))
3511 .expect("pinned compile");
3512
3513 assert_eq!(
3514 hint_for(&lua_seen, "scorer"),
3515 None,
3516 "a non-Operator agent must not receive the operator pin hint"
3517 );
3518 }
3519
3520 #[test]
3524 fn non_object_authored_hint_fails_the_pinned_compile() {
3525 let (compiler, _operator_seen, _lua_seen) = recording_compiler();
3526 let mut bp = bp_with_operator_and_lua_agents();
3527 bp.hints
3528 .per_agent
3529 .insert("planner".to_string(), Value::String("not-an-object".into()));
3530 let bound = resolve_bound_agents(&bp).expect("resolve bound agents");
3531 match compiler.compile_bound_pinned(&bp, &bound, Some("S-pinned")) {
3532 Err(CompileError::InvalidSpec { name, msg }) => {
3533 assert_eq!(name, "planner");
3534 assert!(
3535 msg.contains("run-scoped operator pin"),
3536 "message must explain why the hint blocks the pin: {msg}"
3537 );
3538 }
3539 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
3540 Ok(_) => panic!("expected the non-object hint to fail the pinned compile"),
3541 }
3542 assert!(
3544 compiler.compile_bound(&bp, &bound).is_ok(),
3545 "an unpinned compile must keep accepting whatever hint shape the author declared"
3546 );
3547 }
3548
3549 struct StubOperator {
3555 requires_binding: bool,
3556 }
3557
3558 #[async_trait]
3559 impl Operator for StubOperator {
3560 async fn execute(
3561 &self,
3562 _ctx: &Ctx,
3563 _system: Option<String>,
3564 _prompt: Value,
3565 _worker: Option<WorkerBinding>,
3566 _worker_token: CapToken,
3567 ) -> Result<WorkerResult, WorkerError> {
3568 Ok(WorkerResult {
3569 value: Value::Null,
3570 ok: true,
3571 stats: None,
3572 })
3573 }
3574
3575 fn requires_worker_binding(&self) -> bool {
3576 self.requires_binding
3577 }
3578 }
3579
3580 fn operator_agent() -> AgentDef {
3581 AgentDef {
3582 name: "planner".to_string(),
3583 kind: AgentKind::Operator,
3584 spec: serde_json::json!({ "operator_ref": "main-ai" }),
3585 profile: None,
3586 meta: None,
3587 runner: None,
3588 runner_ref: None,
3589 verdict: None,
3590 lints: None,
3591 }
3592 }
3593
3594 fn pin_hint(sid: &str) -> Value {
3595 serde_json::json!({ OPERATOR_SID_PIN_HINT_KEY: sid })
3596 }
3597
3598 #[test]
3602 fn pin_resolves_the_pinned_session_not_the_role_holder() {
3603 let factory = OperatorSpawnerFactory::new();
3604 factory.register_operator(
3609 "main-ai",
3610 Arc::new(StubOperator {
3611 requires_binding: true,
3612 }) as Arc<dyn Operator>,
3613 );
3614 factory.register_operator(
3615 "S-pinned",
3616 Arc::new(StubOperator {
3617 requires_binding: false,
3618 }) as Arc<dyn Operator>,
3619 );
3620 assert!(
3621 factory
3622 .build(&operator_agent(), Some(&pin_hint("S-pinned")))
3623 .is_ok(),
3624 "the pinned session must resolve the spawner, not the role's holder"
3625 );
3626 assert!(
3629 factory.build(&operator_agent(), None).is_err(),
3630 "without a pin the role's holder must still be the resolution"
3631 );
3632 }
3633
3634 #[test]
3637 fn pin_miss_fails_loud_and_never_falls_back_to_the_role() {
3638 let factory = OperatorSpawnerFactory::new();
3639 factory.register_operator(
3640 "main-ai",
3641 Arc::new(StubOperator {
3642 requires_binding: false,
3643 }) as Arc<dyn Operator>,
3644 );
3645 match factory.build(&operator_agent(), Some(&pin_hint("S-gone"))) {
3646 Err(CompileError::InvalidSpec { name, msg }) => {
3647 assert_eq!(name, "planner");
3648 assert!(msg.contains("S-gone"), "message must name the pin: {msg}");
3649 assert!(
3650 msg.contains("main-ai"),
3651 "message must name the declared role it refused to fall back to: {msg}"
3652 );
3653 }
3654 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
3655 Ok(_) => panic!(
3656 "a pin naming no live session must fail the launch, not silently resolve the role"
3657 ),
3658 }
3659 }
3660
3661 #[test]
3664 fn unpinned_build_keeps_the_historical_role_lookup_and_message() {
3665 let factory = OperatorSpawnerFactory::new();
3666 match factory.build(&operator_agent(), None) {
3667 Err(CompileError::InvalidSpec { msg, .. }) => {
3668 assert!(
3669 msg.contains("operator_ref 'main-ai' not registered in factory"),
3670 "unpinned message must stay the historical one: {msg}"
3671 );
3672 assert!(
3673 !msg.contains("run-scoped operator pin"),
3674 "an unpinned failure must not mention the pin: {msg}"
3675 );
3676 }
3677 Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
3678 Ok(_) => panic!("an unregistered role must still fail"),
3679 }
3680 }
3681}
3682
3683#[cfg(test)]
3686mod projection_placement_compile_tests {
3687 use super::*;
3688 use crate::core::projection_placement::{ProjectionPlacement, RootPreference};
3689 use crate::worker::adapter::WorkerResult;
3690 use mlua_swarm_schema::ProjectionPlacementSpec;
3691
3692 fn registry_with_echo() -> SpawnerRegistry {
3693 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
3694 Ok(WorkerResult {
3695 value: Value::String(inv.prompt),
3696 ok: true,
3697 stats: None,
3698 })
3699 });
3700 let mut reg = SpawnerRegistry::new();
3701 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
3702 reg
3703 }
3704
3705 fn minimal_bp(projection_placement: Option<ProjectionPlacementSpec>) -> Blueprint {
3706 Blueprint {
3707 schema_version: crate::blueprint::current_schema_version(),
3708 id: "projection-placement-ut".into(),
3709 flow: FlowNode::Step {
3710 ref_: "worker".to_string(),
3711 in_: Expr::Path {
3712 at: "$.input".parse().expect("literal test path: $.input"),
3713 },
3714 out: Expr::Path {
3715 at: "$.output".parse().expect("literal test path: $.output"),
3716 },
3717 },
3718 agents: vec![AgentDef {
3719 name: "worker".to_string(),
3720 kind: AgentKind::RustFn,
3721 spec: serde_json::json!({ "fn_id": "echo" }),
3722 profile: None,
3723 meta: None,
3724 runner: None,
3725 runner_ref: None,
3726 verdict: None,
3727 lints: None,
3728 }],
3729 operators: vec![],
3730 metas: vec![],
3731 hints: Default::default(),
3732 strategy: Default::default(),
3733 metadata: BlueprintMetadata::default(),
3734 spawner_hints: Default::default(),
3735 default_agent_kind: AgentKind::Operator,
3736 default_operator_kind: None,
3737 default_init_ctx: None,
3738 default_agent_ctx: None,
3739 default_context_policy: None,
3740 projection_placement,
3741 audits: vec![],
3742 degradation_policy: None,
3743 runners: vec![],
3744 default_runner: None,
3745 subprocesses: vec![],
3746 check_policy: None,
3747 blueprint_ref_includes: Vec::new(),
3748 }
3749 }
3750
3751 #[test]
3752 fn undeclared_projection_placement_compiles_to_byte_compat_default() {
3753 let bp = minimal_bp(None);
3754 let compiled = Compiler::new(registry_with_echo())
3755 .compile(&bp)
3756 .expect("undeclared projection_placement compiles");
3757 assert_eq!(
3758 *compiled.projection_placement,
3759 ProjectionPlacement::default()
3760 );
3761 }
3762
3763 #[test]
3764 fn declared_valid_projection_placement_compiles_to_matching_resolver() {
3765 let bp = minimal_bp(Some(ProjectionPlacementSpec {
3766 root: Some("project_root".to_string()),
3767 dir_template: Some("custom/{task_id}/out".to_string()),
3768 }));
3769 let compiled = Compiler::new(registry_with_echo())
3770 .compile(&bp)
3771 .expect("valid projection_placement compiles");
3772 assert_eq!(
3773 compiled.projection_placement.root_preference,
3774 RootPreference::ProjectRoot
3775 );
3776 assert_eq!(
3777 compiled.projection_placement.dir_template,
3778 "custom/{task_id}/out"
3779 );
3780 }
3781
3782 #[test]
3783 fn declared_invalid_dir_template_rejects_compile() {
3784 let bp = minimal_bp(Some(ProjectionPlacementSpec {
3785 root: None,
3786 dir_template: Some("workspace/tasks/ctx".to_string()), }));
3788 match Compiler::new(registry_with_echo()).compile(&bp) {
3789 Err(CompileError::InvalidProjectionPlacement(_)) => {}
3790 Err(other) => {
3791 panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
3792 }
3793 Ok(_) => {
3794 panic!("expected compile-time rejection for a missing {{task_id}} placeholder")
3795 }
3796 }
3797 }
3798
3799 #[test]
3800 fn declared_invalid_root_literal_rejects_compile() {
3801 let bp = minimal_bp(Some(ProjectionPlacementSpec {
3802 root: Some("nope".to_string()),
3803 dir_template: None,
3804 }));
3805 match Compiler::new(registry_with_echo()).compile(&bp) {
3806 Err(CompileError::InvalidProjectionPlacement(_)) => {}
3807 Err(other) => {
3808 panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
3809 }
3810 Ok(_) => panic!("expected compile-time rejection for an invalid root literal"),
3811 }
3812 }
3813}
3814
3815#[cfg(test)]
3817mod verdict_contract_lint_tests {
3818 use super::*;
3819 use crate::worker::adapter::WorkerResult;
3820
3821 fn registry_with_echo() -> SpawnerRegistry {
3822 let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
3823 Ok(WorkerResult {
3824 value: Value::String(inv.prompt),
3825 ok: true,
3826 stats: None,
3827 })
3828 });
3829 let mut reg = SpawnerRegistry::new();
3830 reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
3831 reg
3832 }
3833
3834 fn gate_agent(verdict: Option<VerdictContract>) -> AgentDef {
3835 AgentDef {
3836 name: "gate".to_string(),
3837 kind: AgentKind::RustFn,
3838 spec: serde_json::json!({ "fn_id": "echo" }),
3839 profile: None,
3840 meta: None,
3841 runner: None,
3842 runner_ref: None,
3843 verdict,
3844 lints: None,
3845 }
3846 }
3847
3848 fn minimal_bp(agent: AgentDef, flow: FlowNode) -> Blueprint {
3849 Blueprint {
3850 schema_version: crate::blueprint::current_schema_version(),
3851 id: "verdict-contract-ut".into(),
3852 flow,
3853 agents: vec![agent],
3854 operators: vec![],
3855 metas: vec![],
3856 hints: Default::default(),
3857 strategy: Default::default(),
3858 metadata: BlueprintMetadata::default(),
3859 spawner_hints: Default::default(),
3860 default_agent_kind: AgentKind::Operator,
3861 default_operator_kind: None,
3862 default_init_ctx: None,
3863 default_agent_ctx: None,
3864 default_context_policy: None,
3865 projection_placement: None,
3866 audits: vec![],
3867 degradation_policy: None,
3868 runners: vec![],
3869 default_runner: None,
3870 subprocesses: vec![],
3871 check_policy: None,
3872 blueprint_ref_includes: Vec::new(),
3873 }
3874 }
3875
3876 fn step(ref_: &str, out_path: &str) -> FlowNode {
3877 FlowNode::Step {
3878 ref_: ref_.to_string(),
3879 in_: Expr::Lit { value: Value::Null },
3880 out: Expr::Path {
3881 at: out_path.parse().expect("literal test path"),
3882 },
3883 }
3884 }
3885
3886 fn noop() -> FlowNode {
3887 FlowNode::Seq { children: vec![] }
3888 }
3889
3890 fn eq_cond(path: &str, lit: &str) -> Expr {
3891 Expr::Eq {
3892 lhs: Box::new(Expr::Path {
3893 at: path.parse().expect("literal test path"),
3894 }),
3895 rhs: Box::new(Expr::Lit {
3896 value: Value::String(lit.to_string()),
3897 }),
3898 }
3899 }
3900
3901 fn branch(cond: Expr, then_: FlowNode, else_: FlowNode) -> FlowNode {
3902 FlowNode::Branch {
3903 cond,
3904 then_: Box::new(then_),
3905 else_: Box::new(else_),
3906 }
3907 }
3908
3909 fn body_contract(values: &[&str]) -> VerdictContract {
3910 VerdictContract {
3911 channel: VerdictChannel::Body,
3912 values: values.iter().map(|v| v.to_string()).collect(),
3913 }
3914 }
3915
3916 fn part_contract(values: &[&str]) -> VerdictContract {
3917 VerdictContract {
3918 channel: VerdictChannel::Part,
3919 values: values.iter().map(|v| v.to_string()).collect(),
3920 }
3921 }
3922
3923 #[test]
3924 fn contract_with_correct_body_channel_and_value_compiles() {
3925 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3926 let flow = FlowNode::Seq {
3927 children: vec![
3928 step("gate", "$.verdict"),
3929 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3930 ],
3931 };
3932 let bp = minimal_bp(agent, flow);
3933 assert!(
3934 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3935 "a cond addressing the bare step output must match a channel: \"body\" contract"
3936 );
3937 }
3938
3939 #[test]
3940 fn contract_with_correct_part_channel_and_value_compiles() {
3941 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3942 let flow = FlowNode::Seq {
3943 children: vec![
3944 step("gate", "$.gate"),
3945 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3946 ],
3947 };
3948 let bp = minimal_bp(agent, flow);
3949 assert!(
3950 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
3951 "a cond addressing '<step>.parts.verdict' must match a channel: \"part\" contract"
3952 );
3953 }
3954
3955 #[test]
3956 fn body_channel_contract_rejects_cond_addressing_parts_verdict() {
3957 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
3961 let flow = FlowNode::Seq {
3962 children: vec![
3963 step("gate", "$.gate"),
3964 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
3965 ],
3966 };
3967 let bp = minimal_bp(agent, flow);
3968 match Compiler::new(registry_with_echo()).compile(&bp) {
3969 Err(CompileError::VerdictChannelMismatch {
3970 where_,
3971 agent,
3972 expected_channel,
3973 actual_shape,
3974 }) => {
3975 assert_eq!(agent, "gate");
3976 assert_eq!(expected_channel, "body");
3977 assert_eq!(actual_shape, "part");
3978 assert!(where_.contains("Branch cond"), "where_: {where_}");
3979 }
3980 Err(other) => {
3981 panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
3982 }
3983 Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
3984 }
3985 }
3986
3987 #[test]
3988 fn part_channel_contract_rejects_cond_addressing_bare_output() {
3989 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
3992 let flow = FlowNode::Seq {
3993 children: vec![
3994 step("gate", "$.verdict"),
3995 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
3996 ],
3997 };
3998 let bp = minimal_bp(agent, flow);
3999 match Compiler::new(registry_with_echo()).compile(&bp) {
4000 Err(CompileError::VerdictChannelMismatch {
4001 agent,
4002 expected_channel,
4003 actual_shape,
4004 ..
4005 }) => {
4006 assert_eq!(agent, "gate");
4007 assert_eq!(expected_channel, "part");
4008 assert_eq!(actual_shape, "body");
4009 }
4010 Err(other) => {
4011 panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
4012 }
4013 Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
4014 }
4015 }
4016
4017 #[test]
4018 fn contract_rejects_lit_outside_declared_values() {
4019 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4020 let flow = FlowNode::Seq {
4021 children: vec![
4022 step("gate", "$.verdict"),
4023 branch(eq_cond("$.verdict", "UNKNOWN"), noop(), noop()),
4024 ],
4025 };
4026 let bp = minimal_bp(agent, flow);
4027 match Compiler::new(registry_with_echo()).compile(&bp) {
4028 Err(CompileError::VerdictValueNotInContract {
4029 agent,
4030 value,
4031 values,
4032 ..
4033 }) => {
4034 assert_eq!(agent, "gate");
4035 assert_eq!(value, "UNKNOWN");
4036 assert_eq!(values, vec!["PASS".to_string(), "BLOCKED".to_string()]);
4037 }
4038 Err(other) => {
4039 panic!("expected VerdictValueNotInContract, got a different CompileError: {other}")
4040 }
4041 Ok(_) => panic!("expected compile-time rejection for a Lit outside declared values"),
4042 }
4043 }
4044
4045 #[test]
4046 fn undeclared_agent_referenced_by_cond_compiles_with_warning_only() {
4047 let agent = gate_agent(None);
4048 let flow = FlowNode::Seq {
4049 children: vec![
4050 step("gate", "$.verdict"),
4051 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
4052 ],
4053 };
4054 let bp = minimal_bp(agent, flow);
4055 assert!(
4056 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4057 "an undeclared verdict contract must never reject compile (opt-in, back-compat)"
4058 );
4059 }
4060
4061 #[test]
4062 fn in_expr_with_lit_haystack_members_compiles() {
4063 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4064 let cond = Expr::In {
4065 needle: Box::new(Expr::Path {
4066 at: "$.verdict".parse().expect("literal test path"),
4067 }),
4068 haystack: Box::new(Expr::Lit {
4069 value: serde_json::json!(["PASS", "BLOCKED"]),
4070 }),
4071 };
4072 let flow = FlowNode::Seq {
4073 children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
4074 };
4075 let bp = minimal_bp(agent, flow);
4076 assert!(
4077 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4078 "an `In` haystack whose every Lit is a declared value must compile"
4079 );
4080 }
4081
4082 #[test]
4089 fn strict_mode_rejects_unhandled_declared_value() {
4090 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4091 let flow = FlowNode::Seq {
4092 children: vec![
4093 step("gate", "$.verdict"),
4094 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
4095 ],
4096 };
4097 let mut bp = minimal_bp(agent, flow);
4098 bp.metadata.strict_verdict_handling = Some(true);
4099 match Compiler::new(registry_with_echo()).compile(&bp) {
4100 Err(CompileError::VerdictValueUnhandled {
4101 agent,
4102 value,
4103 declared_values,
4104 step_ref,
4105 }) => {
4106 assert_eq!(agent, "gate");
4107 assert_eq!(value, "PASS");
4108 assert_eq!(
4109 declared_values,
4110 vec!["PASS".to_string(), "BLOCKED".to_string()]
4111 );
4112 assert_eq!(step_ref, "gate");
4113 }
4114 Err(other) => {
4115 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4116 }
4117 Ok(_) => panic!(
4118 "expected compile-time rejection for a declared verdict value with no \
4119 downstream handler under strict_verdict_handling=Some(true)"
4120 ),
4121 }
4122 }
4123
4124 #[test]
4131 fn default_mode_permits_unhandled_declared_value() {
4132 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4133 let flow = FlowNode::Seq {
4134 children: vec![
4135 step("gate", "$.verdict"),
4136 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
4137 ],
4138 };
4139 let bp = minimal_bp(agent, flow);
4140 assert!(
4142 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4143 "default mode must never reject a Blueprint for unhandled declared values \
4144 (opt-in, back-compat with GH #50)"
4145 );
4146 }
4147
4148 #[test]
4153 fn strict_mode_accepts_all_declared_values_handled() {
4154 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4155 let flow = FlowNode::Seq {
4158 children: vec![
4159 step("gate", "$.verdict"),
4160 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
4161 branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
4162 ],
4163 };
4164 let mut bp = minimal_bp(agent, flow);
4165 bp.metadata.strict_verdict_handling = Some(true);
4166 assert!(
4167 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4168 "strict mode must accept a Blueprint that handles every declared value"
4169 );
4170 }
4171
4172 #[test]
4176 fn strict_mode_accepts_declared_values_covered_by_in_expr() {
4177 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4178 let cond = Expr::In {
4179 needle: Box::new(Expr::Path {
4180 at: "$.verdict".parse().expect("literal test path"),
4181 }),
4182 haystack: Box::new(Expr::Lit {
4183 value: serde_json::json!(["PASS", "BLOCKED"]),
4184 }),
4185 };
4186 let flow = FlowNode::Seq {
4187 children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
4188 };
4189 let mut bp = minimal_bp(agent, flow);
4190 bp.metadata.strict_verdict_handling = Some(true);
4191 assert!(
4192 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4193 "strict mode must accept an `In` haystack that covers every declared value"
4194 );
4195 }
4196
4197 #[test]
4201 fn strict_mode_rejects_unhandled_part_channel_value() {
4202 let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
4203 let flow = FlowNode::Seq {
4204 children: vec![
4205 step("gate", "$.gate"),
4206 branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
4207 ],
4208 };
4209 let mut bp = minimal_bp(agent, flow);
4210 bp.metadata.strict_verdict_handling = Some(true);
4211 match Compiler::new(registry_with_echo()).compile(&bp) {
4212 Err(CompileError::VerdictValueUnhandled {
4213 agent,
4214 value,
4215 step_ref,
4216 ..
4217 }) => {
4218 assert_eq!(agent, "gate");
4219 assert_eq!(value, "PASS");
4220 assert_eq!(step_ref, "gate");
4221 }
4222 Err(other) => {
4223 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4224 }
4225 Ok(_) => panic!(
4226 "expected compile-time rejection for a declared verdict value with no \
4227 downstream handler (part channel) under strict_verdict_handling=Some(true)"
4228 ),
4229 }
4230 }
4231
4232 fn lints(
4238 pairs: &[(&str, mlua_swarm_schema::LintSetting)],
4239 ) -> Option<std::collections::BTreeMap<String, mlua_swarm_schema::LintSetting>> {
4240 Some(
4241 pairs
4242 .iter()
4243 .map(|(key, setting)| ((*key).to_string(), *setting))
4244 .collect(),
4245 )
4246 }
4247
4248 fn bp_with_unhandled_value() -> Blueprint {
4252 let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
4253 let flow = FlowNode::Seq {
4254 children: vec![
4255 step("gate", "$.verdict"),
4256 branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
4257 ],
4258 };
4259 minimal_bp(agent, flow)
4260 }
4261
4262 fn named_agent(name: &str, verdict: Option<VerdictContract>) -> AgentDef {
4266 AgentDef {
4267 name: name.to_string(),
4268 ..gate_agent(verdict)
4269 }
4270 }
4271
4272 fn bp_with_two_unhandled_agents() -> Blueprint {
4276 let flow = FlowNode::Seq {
4277 children: vec![
4278 step("researcher", "$.researcher_verdict"),
4279 step("reviewer", "$.reviewer_verdict"),
4280 branch(eq_cond("$.researcher_verdict", "BLOCKED"), noop(), noop()),
4281 branch(eq_cond("$.reviewer_verdict", "BLOCKED"), noop(), noop()),
4282 ],
4283 };
4284 let mut bp = minimal_bp(
4285 named_agent("researcher", Some(body_contract(&["PASS", "BLOCKED"]))),
4286 flow,
4287 );
4288 bp.agents.push(named_agent(
4289 "reviewer",
4290 Some(body_contract(&["PASS", "BLOCKED"])),
4291 ));
4292 bp
4293 }
4294
4295 #[test]
4300 fn agent_lints_deny_rejects_only_the_declaring_agent() {
4301 let mut bp = bp_with_two_unhandled_agents();
4302 bp.agents[0].lints = lints(&[(
4303 "verdict-value-unhandled",
4304 mlua_swarm_schema::LintSetting::Deny,
4305 )]);
4306 match Compiler::new(registry_with_echo()).compile(&bp) {
4307 Err(CompileError::VerdictValueUnhandled { agent, value, .. }) => {
4308 assert_eq!(agent, "researcher", "the sibling only warns");
4309 assert_eq!(value, "PASS");
4310 }
4311 Err(other) => {
4312 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4313 }
4314 Ok(_) => panic!(
4315 "expected compile-time rejection under \
4316 agents[0].lints = {{\"verdict-value-unhandled\": \"deny\"}}"
4317 ),
4318 }
4319 }
4320
4321 #[test]
4325 fn agent_allow_beats_blueprint_deny_for_that_agent() {
4326 let mut bp = bp_with_two_unhandled_agents();
4327 bp.metadata.lints = lints(&[(
4328 "verdict-value-unhandled",
4329 mlua_swarm_schema::LintSetting::Deny,
4330 )]);
4331 bp.agents[0].lints = lints(&[(
4332 "verdict-value-unhandled",
4333 mlua_swarm_schema::LintSetting::Allow,
4334 )]);
4335 match Compiler::new(registry_with_echo()).compile(&bp) {
4336 Err(CompileError::VerdictValueUnhandled { agent, .. }) => {
4337 assert_eq!(
4338 agent, "reviewer",
4339 "the allowing agent is silenced; the sibling still denies"
4340 );
4341 }
4342 Err(other) => {
4343 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4344 }
4345 Ok(_) => panic!("the sibling agent's Blueprint-level deny must still reject"),
4346 }
4347 }
4348
4349 #[test]
4353 fn strict_flag_wins_over_agent_lints_allow() {
4354 let mut bp = bp_with_unhandled_value();
4355 bp.metadata.strict_verdict_handling = Some(true);
4356 bp.agents[0].lints = lints(&[("all", mlua_swarm_schema::LintSetting::Allow)]);
4357 match Compiler::new(registry_with_echo()).compile(&bp) {
4358 Err(CompileError::VerdictValueUnhandled { agent, .. }) => assert_eq!(agent, "gate"),
4359 Err(other) => {
4360 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4361 }
4362 Ok(_) => panic!(
4363 "strict_verdict_handling=Some(true) must still reject under an agent-level allow"
4364 ),
4365 }
4366 }
4367
4368 #[test]
4371 fn agent_category_key_reaches_the_kind() {
4372 let mut bp = bp_with_two_unhandled_agents();
4373 bp.agents[0].lints =
4374 lints(&[("category:suspicious", mlua_swarm_schema::LintSetting::Deny)]);
4375 match Compiler::new(registry_with_echo()).compile(&bp) {
4376 Err(CompileError::VerdictValueUnhandled { agent, .. }) => assert_eq!(
4377 agent, "researcher",
4378 "a category: group deny must reach the kind it covers, on the declaring agent"
4379 ),
4380 Err(other) => {
4381 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4382 }
4383 Ok(_) => panic!("expected compile-time rejection under an agent-level category deny"),
4384 }
4385 }
4386
4387 #[test]
4390 fn agent_without_lints_inherits_the_blueprint_layer() {
4391 let mut bp = bp_with_two_unhandled_agents();
4392 bp.metadata.lints = lints(&[("all", mlua_swarm_schema::LintSetting::Allow)]);
4393 assert!(
4394 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4395 "a Blueprint-wide allow covers every agent that declares nothing"
4396 );
4397
4398 let gates = resolve_unhandled_verdict_gates(&bp);
4399 assert_eq!(gates.for_agent("reviewer"), UnhandledVerdictGate::Silence);
4400 assert!(gates.all_silent());
4401
4402 bp.agents[0].lints = lints(&[(
4403 "verdict-value-unhandled",
4404 mlua_swarm_schema::LintSetting::Warn,
4405 )]);
4406 let gates = resolve_unhandled_verdict_gates(&bp);
4407 assert_eq!(
4408 gates.for_agent("researcher"),
4409 UnhandledVerdictGate::Warn,
4410 "the agent's own layer wins over the Blueprint's allow"
4411 );
4412 assert_eq!(
4413 gates.for_agent("reviewer"),
4414 UnhandledVerdictGate::Silence,
4415 "the sibling keeps the Blueprint layer"
4416 );
4417 assert!(!gates.all_silent());
4418 }
4419
4420 #[test]
4424 fn lints_deny_rejects_unhandled_declared_value() {
4425 let mut bp = bp_with_unhandled_value();
4426 bp.metadata.lints = lints(&[(
4427 "verdict-value-unhandled",
4428 mlua_swarm_schema::LintSetting::Deny,
4429 )]);
4430 match Compiler::new(registry_with_echo()).compile(&bp) {
4431 Err(CompileError::VerdictValueUnhandled { agent, value, .. }) => {
4432 assert_eq!(agent, "gate");
4433 assert_eq!(value, "PASS");
4434 }
4435 Err(other) => {
4436 panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
4437 }
4438 Ok(_) => panic!(
4439 "expected compile-time rejection under \
4440 metadata.lints = {{\"verdict-value-unhandled\": \"deny\"}}"
4441 ),
4442 }
4443 }
4444
4445 #[test]
4448 fn lints_category_deny_rejects_unhandled_declared_value() {
4449 let mut bp = bp_with_unhandled_value();
4450 bp.metadata.lints = lints(&[("category:suspicious", mlua_swarm_schema::LintSetting::Deny)]);
4451 assert!(
4452 matches!(
4453 Compiler::new(registry_with_echo()).compile(&bp),
4454 Err(CompileError::VerdictValueUnhandled { .. })
4455 ),
4456 "a category: group deny must reach the kind it covers"
4457 );
4458 }
4459
4460 #[test]
4464 fn lints_allow_compiles_and_silences_the_warn() {
4465 let mut bp = bp_with_unhandled_value();
4466 bp.metadata.lints = lints(&[(
4467 "verdict-value-unhandled",
4468 mlua_swarm_schema::LintSetting::Allow,
4469 )]);
4470 assert!(
4471 Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
4472 "an allowed lint must never reject the compile"
4473 );
4474 assert_eq!(
4475 resolve_unhandled_verdict_gate(&bp.metadata),
4476 UnhandledVerdictGate::Silence
4477 );
4478 }
4479
4480 #[test]
4484 fn strict_flag_wins_over_lints_allow() {
4485 let mut bp = bp_with_unhandled_value();
4486 bp.metadata.strict_verdict_handling = Some(true);
4487 bp.metadata.lints = lints(&[("all", mlua_swarm_schema::LintSetting::Allow)]);
4488 assert!(
4489 matches!(
4490 Compiler::new(registry_with_echo()).compile(&bp),
4491 Err(CompileError::VerdictValueUnhandled { .. })
4492 ),
4493 "strict_verdict_handling=Some(true) must still reject under a lints allow"
4494 );
4495 }
4496
4497 #[test]
4500 fn unhandled_verdict_gate_resolution_table() {
4501 use mlua_swarm_schema::LintSetting;
4502
4503 let gate = |strict, map| {
4504 resolve_unhandled_verdict_gate(&BlueprintMetadata {
4505 strict_verdict_handling: strict,
4506 lints: map,
4507 ..Default::default()
4508 })
4509 };
4510 let kind = "verdict-value-unhandled";
4511
4512 assert_eq!(gate(None, None), UnhandledVerdictGate::Warn);
4513 assert_eq!(gate(Some(false), None), UnhandledVerdictGate::Warn);
4514 assert_eq!(gate(Some(true), None), UnhandledVerdictGate::Deny);
4515 assert_eq!(
4516 gate(None, lints(&[(kind, LintSetting::Deny)])),
4517 UnhandledVerdictGate::Deny
4518 );
4519 assert_eq!(
4520 gate(None, lints(&[(kind, LintSetting::Warn)])),
4521 UnhandledVerdictGate::Warn
4522 );
4523 assert_eq!(
4524 gate(None, lints(&[(kind, LintSetting::Allow)])),
4525 UnhandledVerdictGate::Silence
4526 );
4527 assert_eq!(
4528 gate(Some(true), lints(&[(kind, LintSetting::Allow)])),
4529 UnhandledVerdictGate::Deny,
4530 "strict wins over allow"
4531 );
4532 assert_eq!(
4534 gate(
4535 None,
4536 lints(&[
4537 (kind, LintSetting::Allow),
4538 ("category:suspicious", LintSetting::Deny),
4539 ])
4540 ),
4541 UnhandledVerdictGate::Silence
4542 );
4543 assert_eq!(
4546 gate(None, lints(&[("no-such-lint", LintSetting::Deny)])),
4547 UnhandledVerdictGate::Warn
4548 );
4549 }
4550
4551 #[test]
4555 fn lints_never_soften_other_compile_errors() {
4556 let mut bp = bp_with_unhandled_value();
4557 bp.agents.push(gate_agent(None));
4558 bp.metadata.lints = lints(&[("all", mlua_swarm_schema::LintSetting::Allow)]);
4559 assert!(
4560 matches!(
4561 Compiler::new(registry_with_echo()).compile(&bp),
4562 Err(CompileError::DuplicateAgent(name)) if name == "gate"
4563 ),
4564 "an `all` allow must not suppress a compile hard error"
4565 );
4566 }
4567
4568 #[test]
4575 fn verdict_omitted_blueprint_compiles_unchanged_with_empty_contracts() {
4576 let agent = gate_agent(None);
4577 let flow = FlowNode::Seq {
4578 children: vec![
4579 step("gate", "$.verdict"),
4580 FlowNode::Loop {
4581 counter: Expr::Path {
4582 at: "$.n".parse().expect("literal test path"),
4583 },
4584 cond: eq_cond("$.verdict", "BLOCKED"),
4585 body: Box::new(step("gate", "$.verdict")),
4586 max: 3,
4587 },
4588 branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
4589 ],
4590 };
4591 let bp = minimal_bp(agent, flow);
4592 let compiled = Compiler::new(registry_with_echo())
4593 .compile(&bp)
4594 .expect("a verdict-omitted Blueprint must compile unchanged");
4595 assert!(
4596 compiled.router.verdict_contracts.is_empty(),
4597 "no agent declared a verdict contract"
4598 );
4599 }
4600
4601 #[test]
4608 fn every_compile_error_diagnostic_kind_is_a_declared_lint() {
4609 let kinds = [
4610 "bound-agent-resolution",
4611 "unknown-agent-kind",
4612 "invalid-agent-spec",
4613 "worker-binding-missing",
4614 "unresolved-agent-ref",
4615 "duplicate-agent-name",
4616 "unresolved-operator-ref",
4617 "unresolved-meta-ref",
4618 "step-naming-collision",
4619 "invalid-projection-placement",
4620 "unresolved-audit-agent",
4621 "verdict-channel-mismatch",
4622 "verdict-value-not-in-contract",
4623 "verdict-value-unhandled",
4624 ];
4625 for kind in kinds {
4626 assert!(
4627 mlua_swarm_diag::lint_decl(kind).is_some(),
4628 "kind '{kind}' emitted by From<&CompileError> has no LINT_DECLS entry"
4629 );
4630 }
4631 }
4632
4633 #[test]
4634 fn invalid_spec_with_worker_binding_prefix_specializes_the_diagnostic_kind() {
4635 let err = CompileError::InvalidSpec {
4639 name: "greeter".into(),
4640 msg: format!("{WORKER_BINDING_REQUIRED_MSG_PREFIX}. Fix by either: (a) ..."),
4641 };
4642 let d = mlua_swarm_diag::Diagnostic::from(&err);
4643 assert_eq!(d.kind, "worker-binding-missing");
4644 assert_eq!(d.level, mlua_swarm_diag::DiagLevel::Error);
4645 assert!(matches!(d.stage, mlua_swarm_diag::DiagStage::CompileLint));
4646 assert!(d.message.contains("greeter"));
4647 let suggestion = d
4648 .suggestion
4649 .expect("specialized arm must carry a suggestion");
4650 assert!(suggestion.patch.contains("backend = \"ws_operator\""));
4651 assert_eq!(
4652 suggestion.applicability,
4653 mlua_swarm_diag::Applicability::HasPlaceholders
4654 );
4655 assert_eq!(
4656 d.docs_ref.expect("docs_ref must be set").uri,
4657 "mse://guides/bp-dsl-templates"
4658 );
4659 match d.span.expect("span must be set").element {
4660 mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "greeter"),
4661 other => panic!("expected Agent span, got {other:?}"),
4662 }
4663 }
4664
4665 #[test]
4666 fn generic_invalid_spec_maps_to_the_generic_kind() {
4667 let err = CompileError::InvalidSpec {
4668 name: "solo".into(),
4669 msg: "operator spec: 'operator_ref' (string) required".into(),
4670 };
4671 let d = mlua_swarm_diag::Diagnostic::from(&err);
4672 assert_eq!(d.kind, "invalid-agent-spec");
4673 assert!(
4674 d.suggestion.is_none(),
4675 "generic arm carries no canned patch"
4676 );
4677 }
4678
4679 #[test]
4680 fn verdict_value_not_in_contract_diagnostic_carries_suggestion_and_span() {
4681 let err = CompileError::VerdictValueNotInContract {
4682 where_: "Branch cond".into(),
4683 agent: "review".into(),
4684 value: "NOT_DECLARED".into(),
4685 values: vec!["PASS".into(), "BLOCKED".into()],
4686 };
4687 let d = mlua_swarm_diag::Diagnostic::from(&err);
4688 assert_eq!(d.kind, "verdict-value-not-in-contract");
4689 assert!(d.message.contains("NOT_DECLARED"));
4690 assert!(d.suggestion.is_some());
4691 match d.span.expect("span must be set").element {
4692 mlua_swarm_diag::DiagElement::Agent { name } => assert_eq!(name, "review"),
4693 other => panic!("expected Agent span, got {other:?}"),
4694 }
4695 }
4696}
4697
4698#[cfg(test)]
4700mod subprocess_embed_compile_tests {
4701 use super::*;
4702 use mlua_swarm_schema::{current_schema_version, SubprocessDef, SubprocessOverrides};
4703
4704 fn subprocess_agent(name: &str, runner: Option<Runner>) -> AgentDef {
4705 AgentDef {
4706 name: name.to_string(),
4707 kind: AgentKind::Subprocess,
4708 spec: serde_json::json!({}),
4709 profile: Some(AgentProfile {
4710 system_prompt: "you are a headless worker".to_string(),
4711 model: Some("profile-model".to_string()),
4712 tools: vec!["Read".to_string()],
4713 ..Default::default()
4714 }),
4715 meta: None,
4716 runner,
4717 runner_ref: None,
4718 verdict: None,
4719 lints: None,
4720 }
4721 }
4722
4723 fn echo_def(name: &str) -> SubprocessDef {
4724 SubprocessDef {
4725 name: name.to_string(),
4726 argv: vec!["sh".to_string(), "-c".to_string(), "cat".to_string()],
4727 stdin: Some("{prompt}".to_string()),
4728 env: Default::default(),
4729 cwd: None,
4730 output: None,
4731 stream_mode: None,
4732 }
4733 }
4734
4735 fn bp_with(agents: Vec<AgentDef>, subprocesses: Vec<SubprocessDef>) -> Blueprint {
4736 Blueprint {
4737 schema_version: current_schema_version(),
4738 id: "gh83-ut".into(),
4739 flow: FlowNode::Seq { children: vec![] },
4740 agents,
4741 operators: vec![],
4742 metas: vec![],
4743 hints: Default::default(),
4744 strategy: Default::default(),
4745 metadata: BlueprintMetadata::default(),
4746 spawner_hints: Default::default(),
4747 default_agent_kind: AgentKind::Operator,
4748 default_operator_kind: None,
4749 default_init_ctx: None,
4750 default_agent_ctx: None,
4751 default_context_policy: None,
4752 projection_placement: None,
4753 audits: vec![],
4754 degradation_policy: None,
4755 runners: vec![],
4756 default_runner: None,
4757 subprocesses,
4758 check_policy: None,
4759 blueprint_ref_includes: vec![],
4760 }
4761 }
4762
4763 fn subprocess_runner(template: &str) -> Runner {
4764 Runner::Subprocess {
4765 template: template.to_string(),
4766 overrides: SubprocessOverrides::default(),
4767 }
4768 }
4769
4770 #[test]
4771 fn validate_placeholders_accepts_closed_set_and_json_braces() {
4772 for ok in [
4773 "{system} {system_file} {prompt} {model} {tools_csv} {work_dir} {task_id} {attempt}",
4774 r#"echo '{"result": "ok", "nested": {"a": 1}}'"#,
4775 "no placeholders at all",
4776 "unmatched { brace",
4777 ] {
4778 validate_embed_placeholders(ok, "ut").expect("must be accepted");
4779 }
4780 }
4781
4782 #[test]
4783 fn validate_placeholders_rejects_unknown_token() {
4784 let err = validate_embed_placeholders("--flag {evil}", "argv[1]").unwrap_err();
4785 assert!(err.contains("'{evil}'"), "token named: {err}");
4786 assert!(err.contains("closed set"), "closed set listed: {err}");
4787 }
4788
4789 #[test]
4793 fn validate_placeholders_descends_into_literal_braces() {
4794 validate_embed_placeholders(r#"{"task": "{prompt}"}"#, "stdin")
4795 .expect("nested closed-set token must be accepted");
4796 let err = validate_embed_placeholders(r#"{"task": "{evil}"}"#, "stdin").unwrap_err();
4797 assert!(
4798 err.contains("'{evil}'"),
4799 "nested unknown token caught: {err}"
4800 );
4801 }
4802
4803 #[test]
4804 fn hint_resolution_finds_declared_template() {
4805 let agent = subprocess_agent("headless", Some(subprocess_runner("echo")));
4806 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
4807 let hint = resolve_subprocess_template_hint(&bp, &agent)
4808 .expect("resolves")
4809 .expect("Runner::Subprocess must synthesize a hint");
4810 assert_eq!(hint[SUBPROCESS_TEMPLATE_HINT_KEY]["name"], "echo");
4811 assert!(hint.get(SUBPROCESS_OVERRIDES_HINT_KEY).is_some());
4812 }
4813
4814 #[test]
4815 fn hint_resolution_unknown_template_is_invalid_spec() {
4816 let agent = subprocess_agent("headless", Some(subprocess_runner("nope")));
4817 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
4818 let err = resolve_subprocess_template_hint(&bp, &agent).unwrap_err();
4819 let msg = format!("{err}");
4820 assert!(msg.contains("'nope'"), "missing template named: {msg}");
4821 assert!(msg.contains("echo"), "defined templates listed: {msg}");
4822 }
4823
4824 #[test]
4825 fn hint_resolution_none_without_subprocess_runner() {
4826 let agent = subprocess_agent("headless", None);
4827 let bp = bp_with(vec![agent.clone()], vec![echo_def("echo")]);
4828 let hint = resolve_subprocess_template_hint(&bp, &agent).expect("resolves");
4829 assert!(hint.is_none(), "spec-based agents keep the historical path");
4830 }
4831
4832 fn agent_block_agent(name: &str, runner: Option<Runner>, profile_tools: &[&str]) -> AgentDef {
4839 AgentDef {
4840 name: name.to_string(),
4841 kind: AgentKind::AgentBlock,
4842 spec: serde_json::json!({}),
4843 profile: Some(AgentProfile {
4844 system_prompt: "you are an in-process auditor".to_string(),
4845 tools: profile_tools.iter().map(|t| t.to_string()).collect(),
4846 ..Default::default()
4847 }),
4848 meta: None,
4849 runner,
4850 runner_ref: None,
4851 verdict: None,
4852 lints: None,
4853 }
4854 }
4855
4856 fn agent_block_runner(tools: &[&str]) -> Runner {
4857 Runner::AgentBlockInProcess {
4858 tools: tools.iter().map(|t| t.to_string()).collect(),
4859 }
4860 }
4861
4862 #[test]
4868 fn agent_block_runner_tools_are_projected_over_profile_tools() {
4869 let agent = agent_block_agent(
4870 "auditor",
4871 Some(agent_block_runner(&["mcp__outline__list_docs"])),
4872 &["Read"],
4873 );
4874 let bp = bp_with(vec![agent], vec![]);
4875 let bound = resolve_bound_agents(&bp).expect("binds");
4876 let effective = materialize_bound_blueprint(&bp, &bound);
4877 assert_eq!(
4878 effective.agents[0].profile.as_ref().unwrap().tools,
4879 vec!["mcp__outline__list_docs".to_string()],
4880 "the declared Runner tools replace profile.tools (['Read'])"
4881 );
4882 }
4883
4884 #[test]
4888 fn agent_block_projection_distinguishes_declared_empty_from_absent() {
4889 let declared = agent_block_agent("auditor", Some(agent_block_runner(&[])), &["Read"]);
4890 let bp = bp_with(vec![declared], vec![]);
4891 let bound = resolve_bound_agents(&bp).expect("binds");
4892 let effective = materialize_bound_blueprint(&bp, &bound);
4893 assert!(
4894 effective.agents[0]
4895 .profile
4896 .as_ref()
4897 .unwrap()
4898 .tools
4899 .is_empty(),
4900 "empty means enforced-empty, not 'unset'"
4901 );
4902
4903 let absent = agent_block_agent("auditor", None, &["Read"]);
4904 let bp = bp_with(vec![absent], vec![]);
4905 let bound = resolve_bound_agents(&bp).expect("binds");
4906 let effective = materialize_bound_blueprint(&bp, &bound);
4907 assert_eq!(
4908 effective.agents[0].profile.as_ref().unwrap().tools,
4909 vec!["Read".to_string()],
4910 "no Runner declared → the agent.md tools line stands"
4911 );
4912 }
4913
4914 #[test]
4921 fn compile_rejects_script_mode_with_a_declared_mcp_grant() {
4922 let mut agent = agent_block_agent(
4923 "auditor",
4924 Some(agent_block_runner(&["mcp__outline__list_docs"])),
4925 &[],
4926 );
4927 agent.spec = serde_json::json!({ "script_path": "gate.lua" });
4928 let mut bp = bp_with(vec![agent], vec![]);
4929 bp.strategy.strict_refs = false;
4930
4931 let mut registry = SpawnerRegistry::new();
4932 registry.register::<crate::worker::agent_block::AgentBlockInProcessSpawnerFactory>(
4933 Arc::new(crate::worker::agent_block::AgentBlockInProcessSpawnerFactory::new()),
4934 );
4935 let err = match Compiler::new(registry).compile(&bp) {
4937 Err(e) => e,
4938 Ok(_) => panic!("script mode + declared MCP grant must not compile"),
4939 };
4940 let msg = format!("{err}");
4941 assert!(msg.contains("script_path"), "names the trigger: {msg}");
4942 assert!(
4943 msg.contains("mcp__outline__list_docs"),
4944 "names the unenforceable tools: {msg}"
4945 );
4946 }
4947
4948 #[test]
4952 fn compile_accepts_script_mode_with_only_inert_tools() {
4953 let mut agent = agent_block_agent("auditor", None, &["Read", "WebSearch"]);
4954 agent.spec = serde_json::json!({ "script_path": "gate.lua" });
4955 let mut bp = bp_with(vec![agent], vec![]);
4956 bp.strategy.strict_refs = false;
4957
4958 let mut registry = SpawnerRegistry::new();
4959 registry.register::<crate::worker::agent_block::AgentBlockInProcessSpawnerFactory>(
4960 Arc::new(crate::worker::agent_block::AgentBlockInProcessSpawnerFactory::new()),
4961 );
4962 if let Err(e) = Compiler::new(registry).compile(&bp) {
4963 panic!("inert tools must not trip the MCP-grant guard: {e}");
4964 }
4965 }
4966
4967 #[test]
4968 fn build_embed_rejects_unknown_placeholder() {
4969 let agent = subprocess_agent("headless", None);
4970 let mut def = echo_def("echo");
4971 def.argv.push("--x={evil}".to_string());
4972 let err = SubprocessProcessSpawnerFactory::build_embed(
4973 &agent,
4974 &serde_json::to_value(&def).unwrap(),
4975 None,
4976 )
4977 .unwrap_err();
4978 assert!(format!("{err}").contains("'{evil}'"));
4979 }
4980
4981 #[test]
4982 fn build_embed_rejects_output_with_stream_mode() {
4983 let agent = subprocess_agent("headless", None);
4984 let mut def = echo_def("echo");
4985 def.stream_mode = Some("ndjson_lines".to_string());
4986 def.output = Some(mlua_swarm_schema::SubprocessOutput {
4987 format: Some("json".to_string()),
4988 result_ptr: None,
4989 ok_from: None,
4990 stats: None,
4991 });
4992 let err = SubprocessProcessSpawnerFactory::build_embed(
4993 &agent,
4994 &serde_json::to_value(&def).unwrap(),
4995 None,
4996 )
4997 .unwrap_err();
4998 assert!(format!("{err}").contains("plain-mode"));
4999 }
5000
5001 #[test]
5002 fn build_embed_rejects_malformed_result_ptr_and_ok_from() {
5003 let agent = subprocess_agent("headless", None);
5004 let mut def = echo_def("echo");
5005 def.output = Some(mlua_swarm_schema::SubprocessOutput {
5006 format: None,
5007 result_ptr: Some("result".to_string()),
5008 ok_from: None,
5009 stats: None,
5010 });
5011 let err = SubprocessProcessSpawnerFactory::build_embed(
5012 &agent,
5013 &serde_json::to_value(&def).unwrap(),
5014 None,
5015 )
5016 .unwrap_err();
5017 assert!(format!("{err}").contains("JSON Pointer"));
5018
5019 let mut def = echo_def("echo");
5020 def.output = Some(mlua_swarm_schema::SubprocessOutput {
5021 format: None,
5022 result_ptr: None,
5023 ok_from: Some("status".to_string()),
5024 stats: None,
5025 });
5026 let err = SubprocessProcessSpawnerFactory::build_embed(
5027 &agent,
5028 &serde_json::to_value(&def).unwrap(),
5029 None,
5030 )
5031 .unwrap_err();
5032 assert!(format!("{err}").contains("exit_code"));
5033 }
5034
5035 #[test]
5036 fn build_embed_bakes_profile_with_override_precedence() {
5037 let agent = subprocess_agent("headless", None);
5038 let def = echo_def("echo");
5039 let overrides = SubprocessOverrides {
5040 model: Some("override-model".to_string()),
5041 tools: vec!["Bash".to_string(), "Write".to_string()],
5042 cwd: Some("/tmp/override-wd".to_string()),
5043 };
5044 let sp = SubprocessProcessSpawnerFactory::build_embed(
5045 &agent,
5046 &serde_json::to_value(&def).unwrap(),
5047 Some(&serde_json::to_value(&overrides).unwrap()),
5048 )
5049 .expect("builds");
5050 let embed = sp.embed.as_ref().expect("embed template baked");
5051 assert_eq!(embed.model.as_deref(), Some("override-model"));
5052 assert_eq!(embed.tools_csv, "Bash,Write");
5053 assert_eq!(embed.cwd.as_deref(), Some("/tmp/override-wd"));
5054 assert_eq!(
5055 embed.system_prompt.as_deref(),
5056 Some("you are a headless worker")
5057 );
5058 }
5059}