Skip to main content

mlua_swarm/blueprint/
compiler.rs

1//! Blueprint `Compiler`, `CompiledAgentTable`, and the three default
2//! `SpawnerFactory` implementations.
3//!
4//! ## Pipeline
5//!
6//! ```text
7//! Blueprint (= flow + agents + hints + strategy + spawner_hints)
8//!     │
9//!     │ Compiler.compile(&bp)          ← this module (AgentDef → SpawnerAdapter table)
10//!     ▼
11//! CompiledBlueprint {
12//!     router: Arc<CompiledAgentTable>, // ctx.agent → SpawnerAdapter lookup
13//!     flow:   FlowNode,                // the flow.ir source (evaluated via EngineDispatcher)
14//!     metadata: BlueprintMetadata,
15//! }
16//!     │
17//!     │ service::linker::link(router, blueprint.spawner_hints.layers, &engine)
18//!     ▼                                   ↑ Layer wrapping is done separately (src/service/linker.rs)
19//! `Arc<dyn SpawnerAdapter>`            (already wrapped with base + hint SpawnerLayers)
20//!     │
21//!     ▼ EngineDispatcher::with_spawner → engine.dispatch_attempt_with
22//! ```
23//!
24//! `CompiledAgentTable` is a thin table: it looks up `routes[name]` by
25//! `ctx.agent` and hands the spawn off to the matching `SpawnerAdapter`.
26//! The `routes` map is built at compile time through `SpawnerFactory`
27//! implementations. Layer wrapping is not part of this module — it lives
28//! in `service::linker::link`.
29
30use crate::blueprint::{
31    resolve_bound_agents, AgentDef, AgentKind, AgentProfile, Blueprint, BlueprintMetadata,
32    BoundAgent, BoundAgentResolveError, Runner,
33};
34use crate::core::ctx::Ctx;
35use crate::core::engine::Engine;
36use crate::core::projection_placement::{ProjectionPlacement, ProjectionPlacementError};
37use crate::core::step_naming::{StepNaming, StepNamingError};
38use crate::operator::{Operator, OperatorSpawner, WorkerBinding};
39use crate::types::{CapToken, StepId};
40use crate::worker::adapter::{InProcSpawner, SpawnError, SpawnerAdapter, WorkerFn};
41use crate::worker::process_spawner::{ProcessSpawner, StreamMode};
42use crate::worker::Worker;
43use async_trait::async_trait;
44use mlua_flow_ir::{Expr, Node as FlowNode, Path};
45use mlua_swarm_schema::{VerdictChannel, VerdictContract};
46use serde_json::Value;
47use std::collections::HashMap;
48use std::sync::Arc;
49use thiserror::Error;
50
51// ─── error ───────────────────────────────────────────────────────────────
52
53/// Everything that can go wrong while `Compiler::compile` turns a
54/// `Blueprint` into a `CompiledBlueprint`.
55#[derive(Debug, Error)]
56pub enum CompileError {
57    /// Runner / Agent / Context binding failed before any spawner was built.
58    #[error("bound agent resolution: {0}")]
59    BoundAgent(#[from] BoundAgentResolveError),
60    /// An `AgentDef.kind` has no matching entry in the `SpawnerRegistry`
61    /// and `Blueprint.strategy.strict_kind` is set.
62    #[error("unknown agent kind in SpawnerRegistry: {0:?}")]
63    UnknownKind(AgentKind),
64    /// The `AgentDef.spec` shape did not match what the factory for its
65    /// kind requires (missing/mistyped field, etc.).
66    #[error("agent '{name}' spec invalid: {msg}")]
67    InvalidSpec {
68        /// The offending agent's name.
69        name: String,
70        /// Human-readable description of what was wrong with the spec.
71        msg: String,
72    },
73    /// The flow references an agent name that has no corresponding
74    /// `AgentDef` (and no default spawner is configured).
75    #[error("flow references agent '{0}' but no AgentDef matches")]
76    UnresolvedRef(String),
77    /// Two `AgentDef`s in the same `Blueprint` share a name.
78    #[error("duplicate AgentDef name: {0}")]
79    DuplicateAgent(String),
80    /// A `kind = Operator` agent's `spec.operator_ref` does not match
81    /// any `OperatorDef.name` declared in `Blueprint.operators`.
82    #[error("agent '{agent}' operator_ref '{op_ref}' does not match any OperatorDef.name in Blueprint.operators (defined: {defined:?})")]
83    UnresolvedOperatorRef {
84        /// The agent whose `operator_ref` didn't resolve.
85        agent: String,
86        /// The `operator_ref` value that was looked up.
87        op_ref: String,
88        /// The `OperatorDef.name`s that *are* declared, for the error
89        /// message.
90        defined: Vec<String>,
91    },
92    /// GH #21 Phase 2: an `AgentMeta.meta_ref` or a statically-visible
93    /// `$step_meta.ref` (inside a `Step.in` **Lit** expr) does not match
94    /// any `MetaDef.name` declared in `Blueprint.metas`.
95    #[error("{where_} names an undefined MetaDef: '{meta_ref}' (defined: {defined:?})")]
96    UnresolvedMetaRef {
97        /// Human-readable description of where the reference was found
98        /// (e.g. `"AgentMeta.meta_ref of agent 'planner'"` or `"Step
99        /// 'scout' $step_meta.ref"`).
100        where_: String,
101        /// The `meta_ref` value that was looked up.
102        meta_ref: String,
103        /// The `MetaDef.name`s that *are* declared, for the error
104        /// message.
105        defined: Vec<String>,
106    },
107    /// GH #23: two Steps' canonical/alias projection names collide and at
108    /// least one side declared `AgentMeta.projection_name` — see
109    /// [`crate::core::step_naming::StepNaming::from_blueprint`]'s doc for
110    /// the full resolution rule (an undeclared/undeclared clash is a soft
111    /// warning instead, logged but not rejected).
112    #[error("StepNaming collision: {0}")]
113    StepNamingCollision(#[from] StepNamingError),
114    /// GH #27 (follow-up to #23): `Blueprint.projection_placement` failed
115    /// validation — see
116    /// [`crate::core::projection_placement::ProjectionPlacement::from_spec`]'s
117    /// doc for the rejection rules (`dir_template` empty / missing the
118    /// `{task_id}` placeholder / absolute / containing a `..` segment, or
119    /// `root` not `"work_dir"`/`"project_root"`).
120    #[error("invalid projection_placement: {0}")]
121    InvalidProjectionPlacement(#[from] ProjectionPlacementError),
122    /// GH #34: an `audits[].agent` name does not match any `AgentDef.name`
123    /// declared in `Blueprint.agents` — mirrors the `operator_ref`
124    /// validation above (same "design-time reference must resolve"
125    /// discipline).
126    #[error("audits[].agent '{agent}' does not match any AgentDef.name in Blueprint.agents (defined: {defined:?})")]
127    UnresolvedAuditAgent {
128        /// The `audits[].agent` value that was looked up.
129        agent: String,
130        /// The `AgentDef.name`s that *are* declared, for the error
131        /// message.
132        defined: Vec<String>,
133    },
134    /// GH #50: a `Branch`/`Loop` `cond` compares a contract-bearing
135    /// agent's output using the wrong OUTPUT channel — e.g. the agent
136    /// declares `channel: "part"` (verdict staged as the named part
137    /// `"verdict"`, addressed `$.<step>.parts.verdict`) but the cond
138    /// addresses the bare step output (`$.<step>`) instead, or vice
139    /// versa. See the `blueprint-authoring.md` guide's "Returning
140    /// verdicts to drive BP flow" section for Pattern A (`channel:
141    /// "body"`) vs Pattern B (`channel: "part"`).
142    #[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        /// Human-readable description of where the offending cond was
149        /// found (e.g. `"Branch cond"` / `"Loop cond"`).
150        where_: String,
151        /// The agent whose declared `verdict.channel` didn't match.
152        agent: String,
153        /// The agent's declared channel (`"body"` or `"part"`).
154        expected_channel: String,
155        /// The channel shape the cond's `Path` actually addressed
156        /// (`"body"` or `"part"`).
157        actual_shape: String,
158    },
159    /// GH #50: a `Branch`/`Loop` `cond`'s `Lit` operand (or, for `In`, one
160    /// of the `Lit` haystack's array elements) is not a member of a
161    /// contract-bearing agent's declared `verdict.values` closed token
162    /// set.
163    #[error(
164        "agent '{agent}' verdict Lit '{value}' at {where_} is not a member of the declared \
165         values {values:?}"
166    )]
167    VerdictValueNotInContract {
168        /// Human-readable description of where the offending cond was
169        /// found (e.g. `"Branch cond"` / `"Loop cond"`).
170        where_: String,
171        /// The agent whose declared `verdict.values` didn't contain
172        /// `value`.
173        agent: String,
174        /// The offending `Lit` value, rendered as a string (the raw JSON
175        /// representation when it is not itself a JSON string — a
176        /// non-string `Lit` can never be a member of `values: Vec<String>`
177        /// either way).
178        value: String,
179        /// The agent's declared `verdict.values` closed token set, for the
180        /// error message.
181        values: Vec<String>,
182    },
183    /// GH #50 follow-up (issue `33bc825b`): a contract-bearing agent
184    /// declares `verdict.values = [...]` but at least one member of that
185    /// closed token set is never referenced by any downstream
186    /// `Branch`/`Loop` `cond` `Lit` — the flow author declared a verdict
187    /// value they never wrote a handler for. Emitted only when the
188    /// Blueprint opts in via
189    /// [`BlueprintMetadata::strict_verdict_handling`]`= Some(true)`; under
190    /// the default (`None`/`Some(false)`) unhandled values surface as
191    /// `tracing::warn!` only and compilation succeeds (back-compat with
192    /// Blueprints that intentionally leave some verdict values as
193    /// silent-pass informational tokens).
194    #[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        /// The agent whose declared `verdict.values` entry lacks a
201        /// downstream handler.
202        agent: String,
203        /// The declared value that has no downstream `cond` reference.
204        value: String,
205        /// The agent's full declared `verdict.values` closed token set,
206        /// for the error message.
207        declared_values: Vec<String>,
208        /// The `Step.ref_` where this agent is invoked. When the same
209        /// agent is invoked at multiple sites, the first one encountered
210        /// during flow walk is reported (best-effort — the diagnostic
211        /// still identifies the offending agent uniquely).
212        step_ref: String,
213    },
214}
215
216// ─── SpawnerFactory + Registry ───────────────────────────────────────────
217
218/// Factory trait that interprets an `AgentDef` and builds the concrete
219/// `SpawnerAdapter`. Register one per kind. Parsing the spec,
220/// validating it, and baking the profile are the implementation's job.
221///
222/// The signature was widened in v9 from `(name, spec, hint)` to
223/// `(&AgentDef, hint)` so the profile can be passed through. Most
224/// implementations still just pull `&agent_def.name` and
225/// `&agent_def.spec`, but Operator-backend factories consume
226/// `agent_def.profile` to bake the persona in.
227pub trait SpawnerFactory: Send + Sync {
228    /// Build the concrete `SpawnerAdapter` for one `AgentDef`. `hint` is
229    /// the matching entry (if any) from `Blueprint.hints.per_agent`.
230    fn build(
231        &self,
232        agent_def: &AgentDef,
233        hint: Option<&Value>,
234    ) -> Result<Arc<dyn SpawnerAdapter>, CompileError>;
235}
236
237/// Companion trait that carries the **type-side source of truth** for
238/// the Adapter ↔ `AgentKind` correspondence.
239///
240/// The base [`SpawnerFactory`] trait deliberately does not carry an
241/// associated const so it stays dyn-compatible — that is, so it can be
242/// stored and dispatched as `Arc<dyn SpawnerFactory>`. This companion
243/// trait splits `const KIND: AgentKind` out, and
244/// [`SpawnerRegistry::register`] uses `F::KIND` as the `HashMap` key.
245/// That physically removes the string-lookup failure mode at the type
246/// layer.
247///
248/// The three built-in factories (`Shell` / `InProc` / `Operator`)
249/// implement this. Extension backends (say, `AgentBlockSpawnerFactory`)
250/// follow the same explicit two-step recipe: add a new `AgentKind`
251/// variant and implement this trait.
252pub trait SpawnerFactoryKind: SpawnerFactory {
253    /// The `AgentKind` this factory handles — used as the `HashMap` key
254    /// by `SpawnerRegistry::register`.
255    const KIND: AgentKind;
256    /// The concrete Worker type produced by this `AgentKind` — this
257    /// binds the type chain all the way from `AgentKind` down to `Worker`.
258    /// Every factory declares it so the `AgentKind → Worker` mapping is
259    /// explicit across all four layers. It is the source of truth for
260    /// preserving the concrete type right up until `SpawnerAdapter::spawn`
261    /// erases it into `Box<dyn Worker>`.
262    type Worker: crate::worker::Worker;
263}
264
265/// `AgentKind → SpawnerFactory` mapping. The compiler looks entries up
266/// during `compile()`.
267#[derive(Clone)]
268pub struct SpawnerRegistry {
269    factories: HashMap<AgentKind, Arc<dyn SpawnerFactory>>,
270}
271
272impl SpawnerRegistry {
273    /// Start with an empty `AgentKind → SpawnerFactory` mapping.
274    pub fn new() -> Self {
275        Self {
276            factories: HashMap::new(),
277        }
278    }
279    /// **Type-driven registration** — takes `F::KIND` and uses it as the
280    /// `HashMap` key.
281    ///
282    /// Callers use the form
283    /// `reg.register::<SubprocessProcessSpawnerFactory>(Arc::new(...))`
284    /// and never have to pass an `AgentKind` literal. The Adapter ↔ Kind
285    /// correspondence is enforced at the type layer, physically removing
286    /// the string / enum-literal lookup failure mode.
287    pub fn register<F: SpawnerFactoryKind + 'static>(&mut self, factory: Arc<F>) -> &mut Self {
288        let f: Arc<dyn SpawnerFactory> = factory;
289        self.factories.insert(F::KIND, f);
290        self
291    }
292}
293
294impl Default for SpawnerRegistry {
295    fn default() -> Self {
296        Self::new()
297    }
298}
299
300// ─── Compiler ────────────────────────────────────────────────────────────
301
302/// Turns a `Blueprint` into a `CompiledBlueprint` by resolving every
303/// `AgentDef` against a `SpawnerRegistry`. One-shot: build a fresh
304/// `Compiler` per `compile()` call (or reuse it — it holds no
305/// per-compile state).
306pub struct Compiler {
307    registry: SpawnerRegistry,
308    default_spawner: Option<Arc<dyn SpawnerAdapter>>,
309}
310
311/// The result of `Compiler::compile` — a routing table plus the
312/// unmodified flow and metadata, ready to hand to
313/// `EngineDispatcher::with_spawner` / `mlua_flow_ir::eval_async`.
314pub struct CompiledBlueprint {
315    /// `ctx.agent → SpawnerAdapter` lookup table.
316    pub router: Arc<CompiledAgentTable>,
317    /// The flow.ir source, copied verbatim from `Blueprint.flow`.
318    pub flow: FlowNode,
319    /// Copied verbatim from `Blueprint.metadata`.
320    pub metadata: BlueprintMetadata,
321    /// GH #23: the Blueprint's [`StepNaming`] addressing-space table,
322    /// built once here (the sole construction site — see
323    /// [`StepNaming::from_blueprint`]'s doc) and threaded through
324    /// `EngineDispatcher::with_step_naming` for `EngineState` storage.
325    pub step_naming: Arc<StepNaming>,
326    /// GH #27 (follow-up to #23): the Blueprint's [`ProjectionPlacement`]
327    /// resolver, built once here (the sole construction site — see
328    /// [`ProjectionPlacement::from_spec`]'s doc) and threaded through
329    /// `EngineDispatcher::with_projection_placement` for `EngineState`
330    /// storage.
331    pub projection_placement: Arc<ProjectionPlacement>,
332}
333
334fn project_bound_agent_for_legacy_factories(bound: &BoundAgent) -> AgentDef {
335    let mut agent = bound.agent.clone();
336    match &bound.runner {
337        Some(Runner::WsOperator { variant, tools })
338        | Some(Runner::WsClaudeCode { variant, tools }) => {
339            let profile = agent.profile.get_or_insert_with(AgentProfile::default);
340            profile.worker_binding = Some(variant.clone());
341            profile.tools = tools.clone();
342        }
343        Some(Runner::AgentBlockInProcess { tools }) => {
344            let profile = agent.profile.get_or_insert_with(AgentProfile::default);
345            profile.worker_binding = None;
346            profile.tools = tools.clone();
347        }
348        None => {}
349    }
350    let meta = agent.meta.get_or_insert_with(Default::default);
351    meta.context_policy = bound.context_policy.clone();
352    agent
353}
354
355/// Rebuild a Blueprint's Agent/Context layers from an immutable binding
356/// snapshot while leaving its flow and non-binding metadata untouched.
357pub(crate) fn materialize_bound_blueprint(
358    bp: &Blueprint,
359    bound_agents: &[BoundAgent],
360) -> Blueprint {
361    let mut effective = bp.clone();
362    effective.agents = bound_agents
363        .iter()
364        .map(project_bound_agent_for_legacy_factories)
365        .collect();
366    // Each effective policy is now pinned on its AgentDef; retaining a
367    // mutable BP-global default would reintroduce registry drift on resume.
368    effective.default_context_policy = None;
369    effective
370}
371
372impl Compiler {
373    /// Build a `Compiler` around the given `SpawnerRegistry`, with no
374    /// default spawner (unresolved flow refs are an error unless
375    /// `with_default` is chained on).
376    pub fn new(registry: SpawnerRegistry) -> Self {
377        Self {
378            registry,
379            default_spawner: None,
380        }
381    }
382
383    /// Set a default spawner — used for flow refs (and unregistered
384    /// `AgentKind`s under non-strict strategy) that don't resolve
385    /// against any `AgentDef`/`SpawnerRegistry` entry.
386    pub fn with_default(mut self, sp: Arc<dyn SpawnerAdapter>) -> Self {
387        self.default_spawner = Some(sp);
388        self
389    }
390
391    /// Resolve every `Blueprint.agents` entry through the registry,
392    /// validate `operator_ref`s and flow refs per `Blueprint.strategy`,
393    /// and return the routing table alongside the untouched flow and
394    /// metadata.
395    pub fn compile(&self, bp: &Blueprint) -> Result<CompiledBlueprint, CompileError> {
396        let bound_agents = resolve_bound_agents(bp)?;
397        self.compile_bound(bp, &bound_agents)
398    }
399
400    /// Compile with an already-resolved immutable binding snapshot. Resume
401    /// paths use this entry point so a mutable Blueprint registry cannot
402    /// silently change the Runner, prompt, contract, or static context policy
403    /// between the original Run and its continuation.
404    pub fn compile_bound(
405        &self,
406        bp: &Blueprint,
407        bound_agents: &[BoundAgent],
408    ) -> Result<CompiledBlueprint, CompileError> {
409        let effective = materialize_bound_blueprint(bp, bound_agents);
410        self.compile_resolved(&effective)
411    }
412
413    fn compile_resolved(&self, bp: &Blueprint) -> Result<CompiledBlueprint, CompileError> {
414        let mut routes: HashMap<String, Arc<dyn SpawnerAdapter>> = HashMap::new();
415        let mut seen: HashMap<String, ()> = HashMap::new();
416        // GH #50: `AgentDef.name` → declared `VerdictContract`, collected
417        // alongside `routes` below (every `verdict: Some(...)` agent, kind
418        // resolution notwithstanding). Consumed by the cond↔output-shape
419        // lint right after the loop, and carried into
420        // `CompiledAgentTable.verdict_contracts`.
421        let mut verdict_contracts: HashMap<String, VerdictContract> = HashMap::new();
422
423        // Design-time validation (OperatorDef as a first-class value):
424        // every `kind = Operator` agent's `spec.operator_ref` must point at
425        // one of `bp.operators[].name`. A Blueprint with any Operator agent
426        // must therefore declare its operators up front; the empty-operators
427        // backward-compat bypass is retired.
428        let defined: Vec<String> = bp.operators.iter().map(|o| o.name.clone()).collect();
429        for ad in &bp.agents {
430            if !matches!(ad.kind, AgentKind::Operator) {
431                continue;
432            }
433            let op_ref = ad.spec.get("operator_ref").and_then(|v| v.as_str());
434            if let Some(op_ref) = op_ref {
435                if !defined.iter().any(|n| n == op_ref) {
436                    return Err(CompileError::UnresolvedOperatorRef {
437                        agent: ad.name.clone(),
438                        op_ref: op_ref.to_string(),
439                        defined: defined.clone(),
440                    });
441                }
442            }
443            // A missing `op_ref` is reported through OperatorSpawnerFactory.build under a different error.
444        }
445
446        // GH #21 Phase 2: named `MetaDef` pool (`Blueprint.metas`) —
447        // validate every reference against it, mirroring the
448        // `operator_ref` validation above.
449        let metas_defined: Vec<String> = bp.metas.iter().map(|m| m.name.clone()).collect();
450        for ad in &bp.agents {
451            let meta_ref = ad.meta.as_ref().and_then(|m| m.meta_ref.as_ref());
452            if let Some(meta_ref) = meta_ref {
453                if !metas_defined.iter().any(|n| n == meta_ref) {
454                    return Err(CompileError::UnresolvedMetaRef {
455                        where_: format!("AgentMeta.meta_ref of agent '{}'", ad.name),
456                        meta_ref: meta_ref.clone(),
457                        defined: metas_defined.clone(),
458                    });
459                }
460            }
461        }
462        // Best-effort static walk of the flow for `$step_meta.ref`
463        // envelopes embedded in a Step's **Lit** `in` expr — this is a
464        // design-time hint only: a non-`Lit` `Step.in` (e.g. `Path`) is
465        // invisible here and skipped silently; `EngineDispatcher::dispatch`
466        // is the authoritative, loud validation line for those.
467        let mut static_step_meta_refs: Vec<(String, String)> = Vec::new();
468        collect_step_meta_refs(&bp.flow, &mut static_step_meta_refs);
469        for (where_, meta_ref) in static_step_meta_refs {
470            if !metas_defined.iter().any(|n| n == &meta_ref) {
471                return Err(CompileError::UnresolvedMetaRef {
472                    where_,
473                    meta_ref,
474                    defined: metas_defined.clone(),
475                });
476            }
477        }
478
479        // GH #34: `audits[].agent` must name an entry in `Blueprint.agents`
480        // — mirrors the `operator_ref` validation above (design-time
481        // reference must resolve at compile time, before any spawner is
482        // built).
483        let agents_defined: Vec<String> = bp.agents.iter().map(|a| a.name.clone()).collect();
484        for audit in &bp.audits {
485            if !agents_defined.iter().any(|n| n == &audit.agent) {
486                return Err(CompileError::UnresolvedAuditAgent {
487                    agent: audit.agent.clone(),
488                    defined: agents_defined.clone(),
489                });
490            }
491        }
492
493        for ad in &bp.agents {
494            if seen.contains_key(&ad.name) {
495                return Err(CompileError::DuplicateAgent(ad.name.clone()));
496            }
497            seen.insert(ad.name.clone(), ());
498
499            // GH #50: contract registration is orthogonal to spawner
500            // resolution (an agent may declare `verdict` regardless of
501            // whether its `kind` resolves), so it happens unconditionally
502            // here, before the kind-resolution branch below that may
503            // `continue`.
504            if let Some(contract) = &ad.verdict {
505                verdict_contracts.insert(ad.name.clone(), contract.clone());
506            }
507
508            let factory = match self.registry.factories.get(&ad.kind) {
509                Some(f) => f.clone(),
510                None => {
511                    if bp.strategy.strict_kind {
512                        return Err(CompileError::UnknownKind(ad.kind.clone()));
513                    } else {
514                        tracing::warn!(
515                            agent = %ad.name,
516                            kind = ?ad.kind,
517                            "no spawner factory registered for agent kind; \
518                             dropping agent from routing table (strict_kind=false)"
519                        );
520                        continue;
521                    }
522                }
523            };
524            let hint = bp.hints.per_agent.get(&ad.name);
525            let spawner = factory.build(ad, hint)?;
526            routes.insert(ad.name.clone(), spawner);
527        }
528
529        // GH #50: `Branch`/`Loop` cond↔output-shape lint. A contract-
530        // bearing agent's output must be compared the way its declared
531        // `verdict.channel` requires and its `Lit` value(s) must be
532        // members of its declared `verdict.values`; an agent referenced by
533        // a cond but declaring no contract only gets a `tracing::warn!`
534        // (opt-in, back-compat — see `AgentDef::verdict`'s doc). Read-only
535        // inspection of `bp.flow` — no rewriting, no new `Expr` forms.
536        //
537        // GH #50 follow-up (issue `33bc825b`): the reverse-direction lint
538        // — declared `verdict.values` entries that no downstream cond
539        // references — runs in the same walk. Under
540        // `BlueprintMetadata.strict_verdict_handling = Some(true)` it
541        // rejects the compile; otherwise it only surfaces
542        // `tracing::warn!` so existing Blueprints that intentionally leave
543        // some verdict values as silent-pass informational tokens keep
544        // compiling unchanged.
545        let strict_verdict_handling = bp.metadata.strict_verdict_handling.unwrap_or(false);
546        verify_verdict_conds(&bp.flow, &verdict_contracts, strict_verdict_handling)?;
547
548        if bp.strategy.strict_refs {
549            verify_refs(&bp.flow, &routes, self.default_spawner.is_some())?;
550        }
551
552        // GH #23: build the StepNaming addressing-space table once, here
553        // (the sole construction site). A hard collision (either side
554        // declares `AgentMeta.projection_name`) rejects the compile via
555        // `?` (`StepNamingError` → `CompileError::StepNamingCollision`,
556        // same family as the other Blueprint validation checks above); a
557        // soft undeclared/undeclared collision is logged and compilation
558        // proceeds (pre-GH-#23 union-rule behavior preserved).
559        let (step_naming, step_naming_warnings) = StepNaming::from_blueprint(bp)?;
560        for warning in &step_naming_warnings {
561            tracing::warn!(
562                name = %warning.name,
563                first_step_ref = %warning.first_step_ref,
564                second_step_ref = %warning.second_step_ref,
565                "StepNaming: undeclared steps' canonical/alias names collide; \
566                 the step whose own ref matches the name keeps it (data-plane priority)"
567            );
568        }
569
570        // GH #27 (follow-up to #23): build the ProjectionPlacement resolver
571        // once, here (the sole construction site) — an invalid
572        // `dir_template` / `root` literal rejects the compile via `?`
573        // (`ProjectionPlacementError` → `CompileError::InvalidProjectionPlacement`,
574        // same family as the other Blueprint validation checks above). No
575        // declared `projection_placement` (the pre-#27 default) resolves
576        // to `ProjectionPlacement::default()` unchanged.
577        let projection_placement =
578            ProjectionPlacement::from_spec(bp.projection_placement.as_ref())?;
579
580        let router = Arc::new(CompiledAgentTable {
581            routes,
582            default: self.default_spawner.clone(),
583            verdict_contracts,
584        });
585        Ok(CompiledBlueprint {
586            router,
587            flow: bp.flow.clone(),
588            metadata: bp.metadata.clone(),
589            step_naming: Arc::new(step_naming),
590            projection_placement: Arc::new(projection_placement),
591        })
592    }
593}
594
595/// Walk the flow `Node`, collect every `Step.ref`, and check that no ref
596/// is unresolved against `routes` (or the default, when one exists).
597fn verify_refs(
598    node: &FlowNode,
599    routes: &HashMap<String, Arc<dyn SpawnerAdapter>>,
600    has_default: bool,
601) -> Result<(), CompileError> {
602    let mut refs: Vec<String> = Vec::new();
603    collect_refs(node, &mut refs);
604    for r in refs {
605        if !routes.contains_key(&r) && !has_default {
606            return Err(CompileError::UnresolvedRef(r));
607        }
608    }
609    Ok(())
610}
611
612fn collect_refs(node: &FlowNode, out: &mut Vec<String>) {
613    match node {
614        FlowNode::Step { ref_, .. } => out.push(ref_.clone()),
615        FlowNode::Seq { children } => {
616            for c in children {
617                collect_refs(c, out);
618            }
619        }
620        FlowNode::Branch { then_, else_, .. } => {
621            collect_refs(then_, out);
622            collect_refs(else_, out);
623        }
624        FlowNode::Fanout { body, .. } => collect_refs(body, out),
625        FlowNode::Loop { body, .. } => collect_refs(body, out),
626        FlowNode::Try { body, catch, .. } => {
627            collect_refs(body, out);
628            collect_refs(catch, out);
629        }
630        FlowNode::Assign { .. } => {} // The Assign node carries no ref.
631    }
632}
633
634/// GH #21 Phase 2: walk the flow `Node` (same recursion shape as
635/// [`collect_refs`]) and collect every statically-visible `$step_meta.ref`
636/// found inside a Step's `in` **Lit** expr, as `(where_, meta_ref)` pairs
637/// for [`CompileError::UnresolvedMetaRef`] reporting. Non-`Lit` `in`
638/// exprs (e.g. `Expr::Path`) cannot be inspected statically and are
639/// silently skipped — `EngineDispatcher::dispatch` (the `mlua-swarm` core
640/// crate) is the authoritative, loud validation line for those.
641fn collect_step_meta_refs(node: &FlowNode, out: &mut Vec<(String, String)>) {
642    match node {
643        FlowNode::Step { ref_, in_, .. } => {
644            if let Expr::Lit { value } = in_ {
645                if let Some(meta_ref) = static_step_meta_ref(value) {
646                    out.push((format!("Step '{ref_}' $step_meta.ref"), meta_ref));
647                }
648            }
649        }
650        FlowNode::Seq { children } => {
651            for c in children {
652                collect_step_meta_refs(c, out);
653            }
654        }
655        FlowNode::Branch { then_, else_, .. } => {
656            collect_step_meta_refs(then_, out);
657            collect_step_meta_refs(else_, out);
658        }
659        FlowNode::Fanout { body, .. } => collect_step_meta_refs(body, out),
660        FlowNode::Loop { body, .. } => collect_step_meta_refs(body, out),
661        FlowNode::Try { body, catch, .. } => {
662            collect_step_meta_refs(body, out);
663            collect_step_meta_refs(catch, out);
664        }
665        FlowNode::Assign { .. } => {} // The Assign node carries no `in`.
666    }
667}
668
669/// Extract the `$step_meta.ref` string out of a literal `Step.in` value,
670/// if present and well-formed: `{"$step_meta": {"ref": "<name>", ...},
671/// ...}`. Any other shape (no `$step_meta` key, `ref` absent/null, `ref`
672/// not a string) yields `None` — this is a best-effort static hint only;
673/// a malformed envelope is caught loudly at dispatch time instead (see
674/// `EngineDispatcher::dispatch`'s doc in the `mlua-swarm` core crate).
675fn static_step_meta_ref(value: &Value) -> Option<String> {
676    value
677        .as_object()?
678        .get("$step_meta")?
679        .as_object()?
680        .get("ref")?
681        .as_str()
682        .map(str::to_string)
683}
684
685// ─── GH #50: verdict contract cond↔output-shape lint ───────────────────────
686
687/// GH #50: `Blueprint.agents[].verdict` cond↔output-shape lint, run from
688/// `Compiler::compile` after the routing table is built. Two-pass, same
689/// shape as [`collect_step_meta_refs`]'s best-effort static walk: Pass 1
690/// ([`collect_step_outputs`]) builds `Step.out` `Path` string → producing
691/// `Step.ref_`; Pass 2 ([`collect_verdict_conds`]) walks every
692/// `Branch`/`Loop` `cond` and resolves each `Eq`/`Ne`/`In` `Path`+`Lit`
693/// comparison back through the Pass 1 map. Collects every violation before
694/// returning, then surfaces the first one (mirrors the other
695/// `Compiler::compile` validation blocks' `Result::Err`-via-`?` pattern).
696fn verify_verdict_conds(
697    flow: &FlowNode,
698    verdict_contracts: &HashMap<String, VerdictContract>,
699    strict_verdict_handling: bool,
700) -> Result<(), CompileError> {
701    let mut step_outputs: HashMap<String, String> = HashMap::new();
702    let mut step_agents: HashMap<String, String> = HashMap::new();
703    collect_step_outputs_and_agents(flow, &mut step_outputs, &mut step_agents);
704
705    let mut errors: Vec<CompileError> = Vec::new();
706    let mut referenced_values: HashMap<String, std::collections::HashSet<String>> = HashMap::new();
707    collect_verdict_conds(
708        flow,
709        &step_outputs,
710        verdict_contracts,
711        &mut referenced_values,
712        &mut errors,
713    );
714    check_unhandled_verdict_values(
715        verdict_contracts,
716        &referenced_values,
717        &step_agents,
718        strict_verdict_handling,
719        &mut errors,
720    );
721    match errors.into_iter().next() {
722        Some(e) => Err(e),
723        None => Ok(()),
724    }
725}
726
727/// Pass 1 of [`verify_verdict_conds`]: `Step.out` `Path` (rendered via its
728/// canonical `Display` string) → the producing `Step.ref_` — mirrors
729/// [`collect_refs`]'s `Step.ref_` ↔ `AgentDef.name` correspondence (a
730/// `Step.ref_` directly indexes `Blueprint.agents[].name`, per
731/// `verify_refs`). Only `Step` nodes produce agent output; `Fanout`'s
732/// joined-array `out` and `Assign`'s computed `at` are not attributed to
733/// any single agent and are not inserted here.
734///
735/// GH #50 follow-up (issue `33bc825b`): `step_agents` additionally maps
736/// each `Step.ref_` (= agent name) to the first-seen `Step.ref_` literal,
737/// so [`check_unhandled_verdict_values`] can attribute a diagnostic to a
738/// concrete step site. When the same agent is invoked at multiple sites,
739/// the first-encountered site is retained (best-effort — the diagnostic
740/// still identifies the offending agent uniquely).
741fn collect_step_outputs_and_agents(
742    node: &FlowNode,
743    out: &mut HashMap<String, String>,
744    step_agents: &mut HashMap<String, String>,
745) {
746    match node {
747        FlowNode::Step {
748            ref_,
749            out: out_expr,
750            ..
751        } => {
752            if let Expr::Path { at } = out_expr {
753                out.insert(at.to_string(), ref_.clone());
754            }
755            step_agents
756                .entry(ref_.clone())
757                .or_insert_with(|| ref_.clone());
758        }
759        FlowNode::Seq { children } => {
760            for c in children {
761                collect_step_outputs_and_agents(c, out, step_agents);
762            }
763        }
764        FlowNode::Branch { then_, else_, .. } => {
765            collect_step_outputs_and_agents(then_, out, step_agents);
766            collect_step_outputs_and_agents(else_, out, step_agents);
767        }
768        FlowNode::Fanout { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
769        FlowNode::Loop { body, .. } => collect_step_outputs_and_agents(body, out, step_agents),
770        FlowNode::Try { body, catch, .. } => {
771            collect_step_outputs_and_agents(body, out, step_agents);
772            collect_step_outputs_and_agents(catch, out, step_agents);
773        }
774        FlowNode::Assign { .. } => {} // The Assign node produces no agent output.
775    }
776}
777
778/// Pass 2 of [`verify_verdict_conds`]: recurse through the flow the same
779/// way [`collect_refs`] does, and for every `Branch`/`Loop` node lint its
780/// own `cond` field via [`lint_cond_expr`] (in addition to recursing into
781/// `then_`/`else_`/`body`).
782fn collect_verdict_conds(
783    node: &FlowNode,
784    step_outputs: &HashMap<String, String>,
785    verdict_contracts: &HashMap<String, VerdictContract>,
786    referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
787    errors: &mut Vec<CompileError>,
788) {
789    match node {
790        FlowNode::Branch { cond, then_, else_ } => {
791            lint_cond_expr(
792                cond,
793                "Branch cond",
794                step_outputs,
795                verdict_contracts,
796                referenced_values,
797                errors,
798            );
799            collect_verdict_conds(
800                then_,
801                step_outputs,
802                verdict_contracts,
803                referenced_values,
804                errors,
805            );
806            collect_verdict_conds(
807                else_,
808                step_outputs,
809                verdict_contracts,
810                referenced_values,
811                errors,
812            );
813        }
814        FlowNode::Loop { cond, body, .. } => {
815            lint_cond_expr(
816                cond,
817                "Loop cond",
818                step_outputs,
819                verdict_contracts,
820                referenced_values,
821                errors,
822            );
823            collect_verdict_conds(
824                body,
825                step_outputs,
826                verdict_contracts,
827                referenced_values,
828                errors,
829            );
830        }
831        FlowNode::Seq { children } => {
832            for c in children {
833                collect_verdict_conds(
834                    c,
835                    step_outputs,
836                    verdict_contracts,
837                    referenced_values,
838                    errors,
839                );
840            }
841        }
842        FlowNode::Fanout { body, .. } => collect_verdict_conds(
843            body,
844            step_outputs,
845            verdict_contracts,
846            referenced_values,
847            errors,
848        ),
849        FlowNode::Try { body, catch, .. } => {
850            collect_verdict_conds(
851                body,
852                step_outputs,
853                verdict_contracts,
854                referenced_values,
855                errors,
856            );
857            collect_verdict_conds(
858                catch,
859                step_outputs,
860                verdict_contracts,
861                referenced_values,
862                errors,
863            );
864        }
865        FlowNode::Step { .. } | FlowNode::Assign { .. } => {}
866    }
867}
868
869/// Lint one `cond` `Expr` tree for [`collect_verdict_conds`]: recurses into
870/// `And`/`Or`/`Not` (the only boolean combinators a verdict comparison can
871/// be nested under) and, for every `Eq`/`Ne` leaf whose operands are a
872/// `Path` + `Lit` pair (either order — see [`path_lit_operands`]), or every
873/// `In` leaf whose `needle` is a `Path` and `haystack` is a `Lit` JSON
874/// array, resolves + validates via [`resolve_and_check`]. Any other `Expr`
875/// shape (arithmetic, `Exists`, `CallExtern`, a non-`Path`/`Lit` `Eq`/`Ne`
876/// pair, ...) is not a verdict comparison and is skipped.
877fn lint_cond_expr(
878    expr: &Expr,
879    where_: &str,
880    step_outputs: &HashMap<String, String>,
881    verdict_contracts: &HashMap<String, VerdictContract>,
882    referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
883    errors: &mut Vec<CompileError>,
884) {
885    match expr {
886        Expr::Eq { lhs, rhs } | Expr::Ne { lhs, rhs } => {
887            if let Some((path, lit)) = path_lit_operands(lhs, rhs) {
888                resolve_and_check(
889                    path,
890                    &[lit],
891                    where_,
892                    step_outputs,
893                    verdict_contracts,
894                    referenced_values,
895                    errors,
896                );
897            }
898        }
899        Expr::In { needle, haystack } => {
900            if let (
901                Expr::Path { at },
902                Expr::Lit {
903                    value: Value::Array(items),
904                },
905            ) = (needle.as_ref(), haystack.as_ref())
906            {
907                let lits: Vec<&Value> = items.iter().collect();
908                resolve_and_check(
909                    at,
910                    &lits,
911                    where_,
912                    step_outputs,
913                    verdict_contracts,
914                    referenced_values,
915                    errors,
916                );
917            }
918        }
919        Expr::And { args } | Expr::Or { args } => {
920            for a in args {
921                lint_cond_expr(
922                    a,
923                    where_,
924                    step_outputs,
925                    verdict_contracts,
926                    referenced_values,
927                    errors,
928                );
929            }
930        }
931        Expr::Not { arg } => lint_cond_expr(
932            arg,
933            where_,
934            step_outputs,
935            verdict_contracts,
936            referenced_values,
937            errors,
938        ),
939        _ => {}
940    }
941}
942
943/// Extract a `(Path, Lit value)` pair out of an `Eq`/`Ne`'s two operands,
944/// regardless of which side the `Path` is on. `None` when the pairing is
945/// not exactly one `Path` + one `Lit` (e.g. both are `Path`, or either is a
946/// compound expr) — those are not statically resolvable to a single
947/// literal token and are left for `EngineDispatcher`'s runtime eval.
948fn path_lit_operands<'a>(lhs: &'a Expr, rhs: &'a Expr) -> Option<(&'a Path, &'a Value)> {
949    match (lhs, rhs) {
950        (Expr::Path { at }, Expr::Lit { value }) => Some((at, value)),
951        (Expr::Lit { value }, Expr::Path { at }) => Some((at, value)),
952        _ => None,
953    }
954}
955
956/// Resolve `path` back to a producing step — either as the bare step
957/// output (`channel: Body`) or, via the literal `.parts.verdict` suffix
958/// (`channel: Part` — the "verdict" part name is a literal, per the
959/// "Returning verdicts to drive BP flow" guide's Pattern B), as that
960/// step's staged verdict part. A `path` that resolves to neither shape
961/// against any known step output is skipped silently (best-effort static
962/// lint only, same posture as [`collect_step_meta_refs`]).
963///
964/// When the resolved agent declares a [`VerdictContract`], validates the
965/// resolved channel against it first (a mismatch short-circuits — the
966/// value comparison is moot once the channel itself is wrong) and then
967/// every entry of `lits` against `contract.values`, pushing at most one
968/// `CompileError` per violation. When the resolved agent declares no
969/// contract, emits a `tracing::warn!` only (GH #50's opt-in requirement).
970fn resolve_and_check(
971    path: &Path,
972    lits: &[&Value],
973    where_: &str,
974    step_outputs: &HashMap<String, String>,
975    verdict_contracts: &HashMap<String, VerdictContract>,
976    referenced_values: &mut HashMap<String, std::collections::HashSet<String>>,
977    errors: &mut Vec<CompileError>,
978) {
979    let path_str = path.to_string();
980    let (agent, actual_shape) = if let Some(agent) = step_outputs.get(&path_str) {
981        (agent, "body")
982    } else if let Some(stripped) = path_str.strip_suffix(".parts.verdict") {
983        match step_outputs.get(stripped) {
984            Some(agent) => (agent, "part"),
985            None => return,
986        }
987    } else {
988        return;
989    };
990
991    let Some(contract) = verdict_contracts.get(agent) else {
992        tracing::warn!(
993            agent = %agent,
994            where_ = %where_,
995            "cond references agent output but no verdict contract declared"
996        );
997        return;
998    };
999
1000    let expected_channel = match contract.channel {
1001        VerdictChannel::Body => "body",
1002        VerdictChannel::Part => "part",
1003    };
1004    if expected_channel != actual_shape {
1005        errors.push(CompileError::VerdictChannelMismatch {
1006            where_: where_.to_string(),
1007            agent: agent.clone(),
1008            expected_channel: expected_channel.to_string(),
1009            actual_shape: actual_shape.to_string(),
1010        });
1011        return;
1012    }
1013
1014    for lit in lits {
1015        let value_str = lit
1016            .as_str()
1017            .map(str::to_string)
1018            .unwrap_or_else(|| lit.to_string());
1019        if !contract.values.iter().any(|v| v == &value_str) {
1020            errors.push(CompileError::VerdictValueNotInContract {
1021                where_: where_.to_string(),
1022                agent: agent.clone(),
1023                value: value_str.clone(),
1024                values: contract.values.clone(),
1025            });
1026        }
1027        // GH #50 follow-up (issue `33bc825b`): record the referenced value
1028        // regardless of contract membership. `VerdictValueNotInContract`
1029        // already caught the out-of-set case above; recording here still
1030        // helps future variants that widen the set later. The value string
1031        // is normalized identically to the membership check for symmetric
1032        // comparison in `check_unhandled_verdict_values`.
1033        referenced_values
1034            .entry(agent.clone())
1035            .or_default()
1036            .insert(value_str);
1037    }
1038}
1039
1040/// GH #50 follow-up (issue `33bc825b`): reverse-direction lint.
1041///
1042/// For every agent that declares a [`VerdictContract`], check that every
1043/// entry of `contract.values` was referenced by at least one downstream
1044/// `Branch`/`Loop` `cond` `Lit` (as collected into `referenced_values` by
1045/// [`resolve_and_check`] during the forward pass). Any declared value
1046/// that no cond references is a `verdict_value` the flow author declared
1047/// but forgot to write a handler for.
1048///
1049/// When `strict_verdict_handling` is `true` (opt-in via
1050/// [`BlueprintMetadata::strict_verdict_handling`]), every unhandled value
1051/// pushes a [`CompileError::VerdictValueUnhandled`] onto `errors` and
1052/// [`verify_verdict_conds`] surfaces the first one, rejecting the compile.
1053/// Under the default (`false`), unhandled values only surface via
1054/// `tracing::warn!` — existing Blueprints that intentionally leave some
1055/// verdict values as silent-pass informational tokens keep compiling
1056/// unchanged (back-compat with GH #50's opt-in posture).
1057fn check_unhandled_verdict_values(
1058    verdict_contracts: &HashMap<String, VerdictContract>,
1059    referenced_values: &HashMap<String, std::collections::HashSet<String>>,
1060    step_agents: &HashMap<String, String>,
1061    strict_verdict_handling: bool,
1062    errors: &mut Vec<CompileError>,
1063) {
1064    // Iterate in a stable order (BTreeMap-style sort by agent name, then
1065    // by declared value) so the first `VerdictValueUnhandled` error
1066    // surfaced under strict mode is deterministic across HashMap hash
1067    // seeds. This mirrors GH #50's other lint diagnostics, which are
1068    // stable because they walk the flow tree in source order.
1069    let mut agents: Vec<&String> = verdict_contracts.keys().collect();
1070    agents.sort();
1071    for agent in agents {
1072        let contract = &verdict_contracts[agent];
1073        let referenced = referenced_values.get(agent);
1074        let step_ref = step_agents
1075            .get(agent)
1076            .cloned()
1077            .unwrap_or_else(|| agent.clone());
1078        for value in &contract.values {
1079            let handled = referenced.map(|set| set.contains(value)).unwrap_or(false);
1080            if handled {
1081                continue;
1082            }
1083            if strict_verdict_handling {
1084                errors.push(CompileError::VerdictValueUnhandled {
1085                    agent: agent.clone(),
1086                    value: value.clone(),
1087                    declared_values: contract.values.clone(),
1088                    step_ref: step_ref.clone(),
1089                });
1090            } else {
1091                tracing::warn!(
1092                    agent = %agent,
1093                    value = %value,
1094                    step_ref = %step_ref,
1095                    "declared verdict value has no downstream cond handler; \
1096                     opt in to `metadata.strict_verdict_handling` to reject at compile"
1097                );
1098            }
1099        }
1100    }
1101}
1102
1103// ─── CompiledAgentTable ───────────────────────────────────────────────────────
1104
1105/// The compile result: an `agent name → SpawnerAdapter` lookup table.
1106///
1107/// Looks `routes` up by `ctx.agent` (the flow.ir `Step.ref`) and hands
1108/// the spawn to the matching `SpawnerAdapter`. If the name is not
1109/// registered and a `default` is configured, the default is used; if
1110/// there is no default, `SpawnError::NotRegistered` is returned.
1111///
1112/// Layer wrapping (`AuditMiddleware` / `MainAIMiddleware` and friends) is
1113/// not this type's concern — that is done separately in
1114/// `service::linker::link`.
1115pub struct CompiledAgentTable {
1116    pub(crate) routes: HashMap<String, Arc<dyn SpawnerAdapter>>,
1117    pub(crate) default: Option<Arc<dyn SpawnerAdapter>>,
1118    /// GH #50: `AgentDef.name` → declared `VerdictContract`, for every
1119    /// agent that declared one (built by `Compiler::compile`, alongside
1120    /// `routes`). Backs the submit-time enforcement point (a follow-up).
1121    pub(crate) verdict_contracts: HashMap<String, VerdictContract>,
1122}
1123
1124impl CompiledAgentTable {
1125    /// Whether the given agent name is registered in the table — i.e.,
1126    /// whether its spawner has been resolved.
1127    pub fn has_route(&self, agent: &str) -> bool {
1128        self.routes.contains_key(agent)
1129    }
1130    /// List every resolved agent name.
1131    pub fn routed_agents(&self) -> Vec<String> {
1132        self.routes.keys().cloned().collect()
1133    }
1134    /// GH #50: the declared [`VerdictContract`] for `agent`, if any —
1135    /// `None` both when `agent` is unresolved and when it resolved but
1136    /// declared no contract (opt-in; see `AgentDef::verdict`'s doc).
1137    pub fn verdict_contract_for(&self, agent: &str) -> Option<&VerdictContract> {
1138        self.verdict_contracts.get(agent)
1139    }
1140}
1141
1142#[async_trait]
1143impl SpawnerAdapter for CompiledAgentTable {
1144    async fn spawn(
1145        &self,
1146        engine: &Engine,
1147        ctx: &Ctx,
1148        task_id: StepId,
1149        attempt: u32,
1150        token: CapToken,
1151    ) -> Result<Box<dyn Worker>, SpawnError> {
1152        let sp = self
1153            .routes
1154            .get(&ctx.agent)
1155            .cloned()
1156            .or_else(|| self.default.clone())
1157            .ok_or_else(|| SpawnError::NotRegistered(ctx.agent.clone()))?;
1158        sp.spawn(engine, ctx, task_id, attempt, token).await
1159    }
1160}
1161
1162// ─── default factories (three variants) ───────────────────────────────────
1163
1164/// Factory for `AgentKind::Subprocess`. Turns the spec into a
1165/// [`ProcessSpawner`].
1166///
1167/// Naming convention: `<WorkerIMPL><AdapterType>SpawnerFactory`. Factory
1168/// names carry both the worker implementation and the host adapter so
1169/// they are not confused with each other; the old
1170/// `ShellSpawnerFactory` was renamed to this.
1171///
1172/// Spec shape:
1173/// ```jsonc
1174/// { "program": "agent-block", "args": ["-s","s.lua"],
1175///   "use_stdin": true,                       // optional, default = true
1176///   "stream_mode": "ndjson_lines" | "sse_events" | "length_prefixed" | null  // optional, default = null (plain)
1177/// }
1178/// ```
1179pub struct SubprocessProcessSpawnerFactory;
1180
1181impl SpawnerFactoryKind for SubprocessProcessSpawnerFactory {
1182    const KIND: AgentKind = AgentKind::Subprocess;
1183    type Worker = crate::worker::process_spawner::ProcessWorker;
1184}
1185
1186impl SpawnerFactory for SubprocessProcessSpawnerFactory {
1187    fn build(
1188        &self,
1189        agent_def: &AgentDef,
1190        _hint: Option<&Value>,
1191    ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1192        let agent_name = &agent_def.name;
1193        let spec = &agent_def.spec;
1194        let invalid = |msg: String| CompileError::InvalidSpec {
1195            name: agent_name.to_string(),
1196            msg,
1197        };
1198        let program = spec
1199            .get("program")
1200            .and_then(|v| v.as_str())
1201            .ok_or_else(|| invalid("shell spec: 'program' (string) required".into()))?
1202            .to_string();
1203        let args: Vec<String> = spec
1204            .get("args")
1205            .and_then(|v| v.as_array())
1206            .map(|a| {
1207                a.iter()
1208                    .filter_map(|x| x.as_str().map(|s| s.to_string()))
1209                    .collect()
1210            })
1211            .unwrap_or_default();
1212        let use_stdin = spec
1213            .get("use_stdin")
1214            .and_then(|v| v.as_bool())
1215            .unwrap_or(true);
1216        let stream_mode = match spec.get("stream_mode").and_then(|v| v.as_str()) {
1217            Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
1218            Some("sse_events") => Some(StreamMode::SseEvents),
1219            Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
1220            Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
1221            None => None,
1222        };
1223
1224        let mut sp = ProcessSpawner {
1225            program,
1226            args,
1227            use_stdin,
1228            stream_mode,
1229        };
1230        if let Some(mode) = sp.stream_mode.clone() {
1231            sp = sp.stream_mode(mode);
1232        }
1233        Ok(Arc::new(sp))
1234    }
1235}
1236
1237/// Factory for `AgentKind::Lua`. At `build` time it inspects the
1238/// `AgentDef.spec` and returns an [`InProcSpawner`] with the Lua-eval
1239/// `WorkerFn` registered under `agent_name` — one `InProcSpawner`
1240/// instance per agent.
1241///
1242/// Naming convention: `<WorkerIMPL><AdapterType>SpawnerFactory` (Lua
1243/// worker on InProcess adapter). One half of the old
1244/// `InProcSpawnerFactory`, split into Lua and RustFn variants.
1245///
1246/// Spec shape (choose one; `source` wins when both are present):
1247///
1248/// ```jsonc
1249/// // (a) Registry lookup — Lua source id pre-registered with the
1250/// //     factory via `register_lua` (used by the enhance flow's built-in
1251/// //     workers). Requires the factory to know the id at construction
1252/// //     time.
1253/// { "fn_id": "patch-spawner" }
1254///
1255/// // (b) Inline source — a Lua chunk carried by the Blueprint itself,
1256/// //     wrapped on the fly at `build` time. Combined with the loader's
1257/// //     `$file` ref expansion (`"source": {"$file": "gates/foo.lua"}`)
1258/// //     this lets a BP ship deterministic Lua gates without any
1259/// //     pre-registration. `label` is optional and defaults to
1260/// //     `"<agent_name>.lua"` for error messages.
1261/// { "source": "return { value = 42, ok = true }",
1262///   "label": "psim-gate.lua" }
1263/// ```
1264///
1265/// Host bridges registered on the factory (see [`Self::with_bridge`])
1266/// apply to both spec shapes.
1267pub struct LuaInProcessSpawnerFactory {
1268    registry: HashMap<String, WorkerFn>,
1269    bridges: HashMap<String, HostBridge>,
1270}
1271
1272/// Rust-side bridge function callable from Lua.
1273///
1274/// Inputs and outputs are both `serde_json::Value` (i.e. JSON). Lua
1275/// invokes it as `host.<name>(arg_table)`. If the implementation needs
1276/// to call async Rust, the caller does the sync-ification (typically
1277/// `tokio::runtime::Handle::current().block_on(...)`).
1278///
1279/// Design intent: keep Lua scripts focused on flow control and `ctx`
1280/// walking, while the heavy lifting (LLM calls, RFC 6902 apply,
1281/// verifiers, and so on) stays on the Rust side. Going "pure Lua" —
1282/// removing the bridge — is a carry.
1283#[derive(Clone)]
1284pub struct HostBridge(
1285    Arc<dyn Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync>,
1286);
1287
1288impl HostBridge {
1289    /// Wrap a Rust closure as a bridge callable from Lua.
1290    pub fn new<F>(f: F) -> Self
1291    where
1292        F: Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync + 'static,
1293    {
1294        Self(Arc::new(f))
1295    }
1296
1297    /// Invoke the bridge directly — a thin trampoline over the inner
1298    /// `Fn`. The production path goes through the Lua runtime, but this
1299    /// stays `pub` so unit tests can exercise the primitive directly.
1300    pub fn call(&self, arg: serde_json::Value) -> Result<serde_json::Value, String> {
1301        (self.0)(arg)
1302    }
1303}
1304
1305/// Carrier type for Lua script sources. Paths are not required — a
1306/// source string plus an identifying label is all it holds.
1307///
1308/// Callers bring in the source (via `include_str!` or similar) and
1309/// register it with the factory through
1310/// [`LuaInProcessSpawnerFactory::register_lua`].
1311#[derive(Clone)]
1312pub struct LuaScriptSource {
1313    /// The Lua chunk source.
1314    pub source: String,
1315    /// Label used in error messages — typically the script's logical id
1316    /// (for example `"patch_spawner.lua"`).
1317    pub label: String,
1318}
1319
1320impl LuaScriptSource {
1321    /// Wrap a Lua chunk source and its error-message label.
1322    pub fn new(source: impl Into<String>, label: impl Into<String>) -> Self {
1323        Self {
1324            source: source.into(),
1325            label: label.into(),
1326        }
1327    }
1328}
1329
1330impl LuaInProcessSpawnerFactory {
1331    /// Start with no registered scripts and no host bridges.
1332    pub fn new() -> Self {
1333        Self {
1334            registry: HashMap::new(),
1335            bridges: HashMap::new(),
1336        }
1337    }
1338
1339    /// Register a host bridge. Subsequent `register_lua` calls snapshot
1340    /// the current bridge set.
1341    ///
1342    /// Ordering rule: register bridges first, then call `register_lua`;
1343    /// bridges added after `register_lua` will not be visible to that
1344    /// script.
1345    pub fn with_bridge(mut self, name: impl Into<String>, bridge: HostBridge) -> Self {
1346        self.bridges.insert(name.into(), bridge);
1347        self
1348    }
1349
1350    /// Register a **Lua-eval Worker** under `fn_id`.
1351    ///
1352    /// Each dispatch spins up a fresh `mlua::Lua` VM, injects globals
1353    /// (`_PROMPT` / `_AGENT` / `_TASK_ID` / `_ATTEMPT` / `_CTX` — the last
1354    /// is `_PROMPT` parsed as JSON, or `nil` if that fails), evaluates
1355    /// the script, and marshals the returned table into a `WorkerResult`.
1356    ///
1357    /// Marshalling rules for the return value:
1358    /// - `{ value = ..., ok = bool }` → `WorkerResult.value` /
1359    ///   `WorkerResult.ok` verbatim.
1360    /// - Anything else → `value = <returned value>`, `ok = true`.
1361    ///
1362    /// Execution runs on `tokio::task::spawn_blocking` because `mlua::Lua`
1363    /// is `!Send` and needs to stay away from the tokio async context.
1364    /// Host bridges (the Lua-to-Rust callback path) previously registered
1365    /// with [`Self::with_bridge`] are snapshotted at call time and
1366    /// injected into every dispatch inside `run_lua_worker`.
1367    pub fn register_lua(mut self, fn_id: impl Into<String>, source: LuaScriptSource) -> Self {
1368        let source = Arc::new(source);
1369        let bridges = Arc::new(self.bridges.clone());
1370        let wrapped: WorkerFn = Arc::new(move |inv| {
1371            let source = source.clone();
1372            let bridges = bridges.clone();
1373            Box::pin(run_lua_worker(source, bridges, inv))
1374        });
1375        self.registry.insert(fn_id.into(), wrapped);
1376        self
1377    }
1378}
1379
1380/// Body of a single Lua-eval invocation (called from `register_lua`).
1381async fn run_lua_worker(
1382    source: Arc<LuaScriptSource>,
1383    bridges: Arc<HashMap<String, HostBridge>>,
1384    inv: crate::worker::adapter::WorkerInvocation,
1385) -> Result<crate::worker::adapter::WorkerResult, crate::worker::adapter::WorkerError> {
1386    use crate::worker::adapter::WorkerError;
1387    use mlua::LuaSerdeExt;
1388
1389    let label = source.label.clone();
1390    let outcome =
1391        tokio::task::spawn_blocking(move || -> Result<(serde_json::Value, bool), String> {
1392            let lua = mlua::Lua::new();
1393            let g = lua.globals();
1394
1395            // 1. Base globals.
1396            g.set("_PROMPT", inv.prompt.clone())
1397                .map_err(|e| format!("set _PROMPT: {e}"))?;
1398            g.set("_AGENT", inv.agent.clone())
1399                .map_err(|e| format!("set _AGENT: {e}"))?;
1400            g.set("_TASK_ID", inv.task_id.to_string())
1401                .map_err(|e| format!("set _TASK_ID: {e}"))?;
1402            g.set("_ATTEMPT", inv.attempt as i64)
1403                .map_err(|e| format!("set _ATTEMPT: {e}"))?;
1404
1405            // 2. _CTX = JSON parse(_PROMPT); nil on parse failure (co-exists with the plain-string prompt path).
1406            if let Ok(json_val) = serde_json::from_str::<serde_json::Value>(&inv.prompt) {
1407                let lua_val = lua
1408                    .to_value(&json_val)
1409                    .map_err(|e| format!("_CTX to_value: {e}"))?;
1410                g.set("_CTX", lua_val)
1411                    .map_err(|e| format!("set _CTX: {e}"))?;
1412            }
1413
1414            // 3. Inject the host bridge (Lua can call `host.<name>(arg)`).
1415            if !bridges.is_empty() {
1416                let host = lua
1417                    .create_table()
1418                    .map_err(|e| format!("create host table: {e}"))?;
1419                for (name, bridge) in bridges.iter() {
1420                    let bridge = bridge.clone();
1421                    let bname = name.clone();
1422                    let f = lua
1423                        .create_function(move |lua, arg: mlua::Value| {
1424                            let json_arg: serde_json::Value = lua.from_value(arg).map_err(|e| {
1425                                mlua::Error::external(format!("bridge {bname} arg → json: {e}"))
1426                            })?;
1427                            let result_json =
1428                                bridge.call(json_arg).map_err(mlua::Error::external)?;
1429                            lua.to_value(&result_json).map_err(|e| {
1430                                mlua::Error::external(format!("bridge {bname} ret → lua: {e}"))
1431                            })
1432                        })
1433                        .map_err(|e| format!("create_function {name}: {e}"))?;
1434                    host.set(name.as_str(), f)
1435                        .map_err(|e| format!("host.{name} set: {e}"))?;
1436                }
1437                g.set("host", host).map_err(|e| format!("set host: {e}"))?;
1438            }
1439
1440            // 4. eval
1441            let result: mlua::Value = lua
1442                .load(&source.source)
1443                .set_name(&source.label)
1444                .eval()
1445                .map_err(|e| format!("lua eval [{}]: {e}", source.label))?;
1446
1447            // 5. Marshal: shape `{ value=..., ok=true }` or raw value.
1448            let json_result: serde_json::Value = lua
1449                .from_value(result)
1450                .map_err(|e| format!("lua → json [{}]: {e}", source.label))?;
1451
1452            let (value, ok) = match &json_result {
1453                serde_json::Value::Object(map)
1454                    if map.contains_key("value") || map.contains_key("ok") =>
1455                {
1456                    let ok = map.get("ok").and_then(|v| v.as_bool()).unwrap_or(true);
1457                    let value = map.get("value").cloned().unwrap_or(json_result.clone());
1458                    (value, ok)
1459                }
1460                _ => (json_result, true),
1461            };
1462            Ok((value, ok))
1463        })
1464        .await
1465        .map_err(|e| WorkerError::Failed(format!("spawn_blocking join [{label}]: {e}")))?
1466        .map_err(WorkerError::Failed)?;
1467
1468    Ok(crate::worker::adapter::WorkerResult {
1469        value: outcome.0,
1470        ok: outcome.1,
1471    })
1472}
1473
1474impl Default for LuaInProcessSpawnerFactory {
1475    fn default() -> Self {
1476        Self::new()
1477    }
1478}
1479
1480impl SpawnerFactoryKind for LuaInProcessSpawnerFactory {
1481    const KIND: AgentKind = AgentKind::Lua;
1482    type Worker = LuaWorker;
1483}
1484
1485impl SpawnerFactory for LuaInProcessSpawnerFactory {
1486    fn build(
1487        &self,
1488        agent_def: &AgentDef,
1489        _hint: Option<&Value>,
1490    ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1491        // Inline `spec.source` (a Lua chunk carried by the BP itself) takes
1492        // precedence over `spec.fn_id`. This is the path a BP author uses to
1493        // ship a deterministic Lua gate without pre-registering it with the
1494        // factory — the plumbing (`run_lua_worker` / `LuaScriptSource`) is
1495        // the same, only the entry point differs.
1496        if let Some(source) = agent_def.spec.get("source").and_then(|v| v.as_str()) {
1497            let label = agent_def
1498                .spec
1499                .get("label")
1500                .and_then(|v| v.as_str())
1501                .map(str::to_string)
1502                .unwrap_or_else(|| format!("{}.lua", agent_def.name));
1503            let script = Arc::new(LuaScriptSource::new(source.to_string(), label));
1504            let bridges = Arc::new(self.bridges.clone());
1505            let wrapped: WorkerFn = Arc::new(move |inv| {
1506                let source = script.clone();
1507                let bridges = bridges.clone();
1508                Box::pin(run_lua_worker(source, bridges, inv))
1509            });
1510            let mut sp: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
1511            sp.registry.insert(agent_def.name.to_string(), wrapped);
1512            return Ok(Arc::new(sp));
1513        }
1514        build_inproc_from_registry::<LuaWorker>(&self.registry, agent_def, "lua")
1515    }
1516}
1517
1518/// Factory for `AgentKind::RustFn`. At `build` time it looks the `fn_id`
1519/// up in its internal registry and returns an [`InProcSpawner`] with the
1520/// Rust closure `WorkerFn` registered under `agent_name`.
1521///
1522/// Naming convention: `<WorkerIMPL><AdapterType>SpawnerFactory` (RustFn
1523/// worker on InProcess adapter). Sibling to
1524/// [`LuaInProcessSpawnerFactory`] — the Lua-worker half of the same
1525/// split.
1526///
1527/// Spec shape:
1528/// ```jsonc
1529/// { "fn_id": "echo" }     // Rust closure id pre-registered with the factory
1530/// ```
1531pub struct RustFnInProcessSpawnerFactory {
1532    registry: HashMap<String, WorkerFn>,
1533}
1534
1535impl RustFnInProcessSpawnerFactory {
1536    /// Start with no registered closures.
1537    pub fn new() -> Self {
1538        Self {
1539            registry: HashMap::new(),
1540        }
1541    }
1542
1543    /// Register a Rust closure `WorkerFn` under `fn_id`, wrapping it so
1544    /// it matches the `WorkerFn` signature (boxed, pinned future).
1545    pub fn register_fn<F, Fut>(mut self, fn_id: impl Into<String>, f: F) -> Self
1546    where
1547        F: Fn(crate::worker::adapter::WorkerInvocation) -> Fut + Send + Sync + 'static,
1548        Fut: std::future::Future<
1549                Output = Result<
1550                    crate::worker::adapter::WorkerResult,
1551                    crate::worker::adapter::WorkerError,
1552                >,
1553            > + Send
1554            + 'static,
1555    {
1556        let f = Arc::new(f);
1557        let wrapped: WorkerFn = Arc::new(move |inv| {
1558            let f = f.clone();
1559            Box::pin(f(inv))
1560        });
1561        self.registry.insert(fn_id.into(), wrapped);
1562        self
1563    }
1564}
1565
1566impl Default for RustFnInProcessSpawnerFactory {
1567    fn default() -> Self {
1568        Self::new()
1569    }
1570}
1571
1572impl SpawnerFactoryKind for RustFnInProcessSpawnerFactory {
1573    const KIND: AgentKind = AgentKind::RustFn;
1574    type Worker = RustFnWorker;
1575}
1576
1577impl SpawnerFactory for RustFnInProcessSpawnerFactory {
1578    fn build(
1579        &self,
1580        agent_def: &AgentDef,
1581        _hint: Option<&Value>,
1582    ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1583        build_inproc_from_registry::<RustFnWorker>(&self.registry, agent_def, "rust_fn")
1584    }
1585}
1586
1587/// Shared build helper used by both the Lua and the RustFn factories —
1588/// look `spec.fn_id` up in the registry and return an `InProcSpawner`.
1589/// The generic type parameter `W` fixes the per-kind Worker concrete
1590/// type at the type level (the build-site half of the trait's
1591/// associated-type binding across the four-layer cascade).
1592fn build_inproc_from_registry<W>(
1593    registry: &HashMap<String, WorkerFn>,
1594    agent_def: &AgentDef,
1595    kind_label: &str,
1596) -> Result<Arc<dyn SpawnerAdapter>, CompileError>
1597where
1598    W: crate::worker::Worker + From<crate::worker::WorkerJoinHandler> + Send + Sync + 'static,
1599{
1600    let agent_name = &agent_def.name;
1601    let spec = &agent_def.spec;
1602    let invalid = |msg: String| CompileError::InvalidSpec {
1603        name: agent_name.to_string(),
1604        msg,
1605    };
1606    let fn_id = spec
1607        .get("fn_id")
1608        .and_then(|v| v.as_str())
1609        .ok_or_else(|| invalid(format!("{kind_label} spec: 'fn_id' (string) required")))?;
1610    let f = registry
1611        .get(fn_id)
1612        .cloned()
1613        .ok_or_else(|| invalid(format!("fn_id '{fn_id}' not registered in factory")))?;
1614    let mut sp: InProcSpawner<W> = InProcSpawner::<W>::typed();
1615    // Register under `agent_name` (the flow's `Step.ref`). Both
1616    // `CompiledAgentTable` and the `InProcSpawner` look the function up
1617    // by name, so the same key is needed at both layers.
1618    sp.registry.insert(agent_name.to_string(), f);
1619    Ok(Arc::new(sp))
1620}
1621
1622/// Concrete Worker type for the Lua kind — a handle to a Lua-eval task
1623/// inside an mlua VM. Embeds a `WorkerJoinHandler`. Reserved as the home
1624/// for future Lua-specific extensions (an mlua VM cancellation
1625/// mechanism, Lua-side error type retention, and so on).
1626pub struct LuaWorker {
1627    /// The join handle / cancellation token for the underlying task.
1628    pub handler: crate::worker::WorkerJoinHandler,
1629}
1630
1631impl From<crate::worker::WorkerJoinHandler> for LuaWorker {
1632    fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
1633        Self { handler }
1634    }
1635}
1636
1637#[async_trait::async_trait]
1638impl crate::worker::Worker for LuaWorker {
1639    fn id(&self) -> &crate::types::WorkerId {
1640        &self.handler.worker_id
1641    }
1642    fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
1643        self.handler.cancel.clone()
1644    }
1645    async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
1646        self.handler.await_completion().await
1647    }
1648}
1649
1650/// Concrete Worker type for the RustFn kind — a handle to a task that
1651/// directly calls a Rust closure. Embeds a `WorkerJoinHandler`. Being a
1652/// pure function, there is minimal kind-specific extension surface here;
1653/// the primary purpose is to nail down the type binding.
1654pub struct RustFnWorker {
1655    /// The join handle / cancellation token for the underlying task.
1656    pub handler: crate::worker::WorkerJoinHandler,
1657}
1658
1659impl From<crate::worker::WorkerJoinHandler> for RustFnWorker {
1660    fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
1661        Self { handler }
1662    }
1663}
1664
1665#[async_trait::async_trait]
1666impl crate::worker::Worker for RustFnWorker {
1667    fn id(&self) -> &crate::types::WorkerId {
1668        &self.handler.worker_id
1669    }
1670    fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
1671        self.handler.cancel.clone()
1672    }
1673    async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
1674        self.handler.await_completion().await
1675    }
1676}
1677
1678/// Factory for `AgentKind::Operator`. Looks up the `Arc<dyn Operator>`
1679/// pre-registered under `spec.operator_ref` and wraps it in an
1680/// `OperatorSpawner`. Also resolves `AgentDef.profile.worker_binding` into
1681/// a `WorkerBinding` at compile time and fails loud (`CompileError::InvalidSpec`)
1682/// when the resolved operator's `Operator::requires_worker_binding` is `true`
1683/// and no binding was declared.
1684///
1685/// Spec shape:
1686/// ```jsonc
1687/// { "operator_ref": "main_ai" }     // Operator id pre-registered with the factory
1688/// ```
1689///
1690/// # Split of responsibilities with `OperatorDelegateMiddleware`
1691///
1692/// The two axes exist for different reasons:
1693///
1694/// - **This factory (`OperatorSpawnerFactory` → `OperatorSpawner`) — the
1695///   AgentSpec axis.** Bakes a separate Operator backend into each
1696///   `AgentDef`. A `kind = Operator` `AgentDef` names its backend through
1697///   `spec.operator_ref`; at `compile()` time the `Arc<dyn Operator>` is
1698///   baked into `routes[agent_name]`. Because the `agent.md` loader
1699///   (`agent_md_loader`) defaults `kind` to `Operator`, agents that flow
1700///   in through external agent.md files land here.
1701///
1702/// - **`OperatorDelegateMiddleware` — the Blueprint-global (session)
1703///   axis.** Delegates every agent to the same Operator backend. At
1704///   session-attach time you call `engine.register_operator(id, op)`
1705///   plus `attach_with_ids(.., operator_backend_id = Some(id))` to bind
1706///   it session-wide, and declare
1707///   `spawner_hints.layers = ["operator_delegate"]` to opt in. `ctx.agent`
1708///   is ignored; the operator handles every spawn in that session (a
1709///   MainAI-wide driver, a human-wide console, that sort of thing).
1710///
1711/// # Exclusivity (a double fire is structurally impossible)
1712///
1713/// When both are effective — the hint is declared, the session has an
1714/// operator backend, **and** the Blueprint has a `kind = Operator`
1715/// `AgentDef` — `OperatorDelegateMiddleware` sits at the outer end of
1716/// the stack and **completely bypasses** `inner.spawn`. The
1717/// `OperatorSpawner` is never reached, so under those conditions this
1718/// factory's routes entry is inert. This is not a double fire — the
1719/// session axis is overriding the agent axis. Consistent usage means
1720/// picking one axis per use case.
1721///
1722/// Interior mutability is provided by an `Arc<RwLock>`. Even after the
1723/// factory has been stored as `Arc<dyn SpawnerFactory>` in
1724/// `SpawnerRegistry`, a caller holding an `Arc` clone can still add
1725/// Operator backends dynamically via `register_operator(&self, id, op)`.
1726/// Typical uses: registering a `WSOperatorSession` under the session id
1727/// on WebSocket connect, binding agents that arrive via the `agent.md`
1728/// loader to arbitrary backends, and so on. `build()` performs a
1729/// `read()` lookup each time.
1730pub struct OperatorSpawnerFactory {
1731    operators: Arc<std::sync::RwLock<HashMap<String, Arc<dyn Operator>>>>,
1732}
1733
1734impl OperatorSpawnerFactory {
1735    /// Start with no registered Operator backends.
1736    pub fn new() -> Self {
1737        Self {
1738            operators: Arc::new(std::sync::RwLock::new(HashMap::new())),
1739        }
1740    }
1741
1742    /// Register an Operator backend dynamically through `&self`.
1743    /// Overwrites are allowed — later wins. Callers can still reach this
1744    /// after the factory has been stored as `Arc<dyn SpawnerFactory>` in
1745    /// `SpawnerRegistry`, as long as they hold an `Arc` clone; interior
1746    /// mutability is provided by the inner `RwLock`.
1747    pub fn register_operator(&self, id: impl Into<String>, op: Arc<dyn Operator>) -> &Self {
1748        self.operators
1749            .write()
1750            .expect("OperatorSpawnerFactory.operators RwLock poisoned")
1751            .insert(id.into(), op);
1752        self
1753    }
1754
1755    /// Dynamically unregister an id (used to clean up when a WebSocket
1756    /// disconnects, for example). A missing id is a no-op.
1757    pub fn unregister_operator(&self, id: &str) -> &Self {
1758        self.operators
1759            .write()
1760            .expect("OperatorSpawnerFactory.operators RwLock poisoned")
1761            .remove(id);
1762        self
1763    }
1764}
1765
1766impl Default for OperatorSpawnerFactory {
1767    fn default() -> Self {
1768        Self::new()
1769    }
1770}
1771
1772impl SpawnerFactoryKind for OperatorSpawnerFactory {
1773    const KIND: AgentKind = AgentKind::Operator;
1774    type Worker = crate::operator::OperatorWorker;
1775}
1776
1777impl SpawnerFactory for OperatorSpawnerFactory {
1778    fn build(
1779        &self,
1780        agent_def: &AgentDef,
1781        _hint: Option<&Value>,
1782    ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1783        let agent_name = &agent_def.name;
1784        let spec = &agent_def.spec;
1785        // Bake AgentDef.profile.system_prompt into the OperatorSpawner at compile time.
1786        // `Some` → adopted first at spawn time; `None` → falls back to fetch_prompt (initial_directive).
1787        // Fallback path. Sibling: AgentBlockInProcessSpawnerFactory
1788        // (agent_block/runtime.rs) does the same compile-time bake by stuffing
1789        // the profile into BlockConfig.context.
1790        let system_prompt = agent_def.profile.as_ref().map(|p| p.system_prompt.clone());
1791        let invalid = |msg: String| CompileError::InvalidSpec {
1792            name: agent_name.to_string(),
1793            msg,
1794        };
1795        let op_ref = spec
1796            .get("operator_ref")
1797            .and_then(|v| v.as_str())
1798            .ok_or_else(|| invalid("operator spec: 'operator_ref' (string) required".into()))?;
1799        let operators = self
1800            .operators
1801            .read()
1802            .expect("OperatorSpawnerFactory.operators RwLock poisoned");
1803        let op = operators.get(op_ref).cloned().ok_or_else(|| {
1804            let mut names: Vec<String> = operators.keys().cloned().collect();
1805            names.sort();
1806            let names_list = if names.is_empty() {
1807                "<none>".to_string()
1808            } else {
1809                names.join(", ")
1810            };
1811            invalid(format!(
1812                "operator_ref '{op_ref}' not registered in factory. \
1813                 Registered sids: [{names_list}]. \
1814                 Hint: call mse_operator_join(roles=[...]) to mint the sid first."
1815            ))
1816        })?;
1817        drop(operators);
1818
1819        // Resolve the Blueprint-baked worker binding from
1820        // `AgentDef.profile.worker_binding` — the SoT for the
1821        // declaration↔executor binding (see `WorkerBinding` doc). Fail
1822        // loud at compile time when the operator backend requires one
1823        // and the Blueprint didn't declare it; this is a compile-time
1824        // gate, not a runtime guess.
1825        let worker_binding = agent_def
1826            .profile
1827            .as_ref()
1828            .and_then(|p| p.worker_binding.as_ref())
1829            .map(|variant| WorkerBinding {
1830                variant: variant.clone(),
1831                tools: agent_def
1832                    .profile
1833                    .as_ref()
1834                    .map(|p| p.tools.clone())
1835                    .unwrap_or_default(),
1836                // Compile-time path: no immutable BoundAgent snapshot exists
1837                // here (the launch path resolves the digest). Self-check
1838                // inputs are supplied on the launch axis only.
1839                request_digest: None,
1840                requested_model: None,
1841            });
1842        if op.requires_worker_binding() && worker_binding.is_none() {
1843            // Issue #9: the two Blueprint authoring paths (direct JSON
1844            // and `$agent_md` file ref) both land here. Old message
1845            // pointed only at the `.md` frontmatter, which was
1846            // confusing for authors on the JSON-direct path.
1847            return Err(invalid(
1848                "profile.worker_binding is required for this operator backend. \
1849                 Fix by either: \
1850                 (a) if authoring the Blueprint JSON directly, add \
1851                 `agents[N].profile.worker_binding: \"<subagent-type>\"` \
1852                 to the JSON literal; or \
1853                 (b) if using an $agent_md file ref, add \
1854                 `worker_binding: <subagent-type>` to the agent .md frontmatter."
1855                    .into(),
1856            ));
1857        }
1858        Ok(Arc::new(OperatorSpawner::new(
1859            op,
1860            system_prompt,
1861            worker_binding,
1862        )))
1863    }
1864}
1865
1866#[cfg(test)]
1867mod operator_spawner_factory_worker_binding_tests {
1868    use super::*;
1869    use crate::blueprint::AgentProfile;
1870    use crate::core::ctx::Ctx;
1871    use crate::types::CapToken;
1872    use crate::worker::adapter::{WorkerError, WorkerResult};
1873
1874    /// Minimal `Operator` stub whose `requires_worker_binding` is
1875    /// configurable — enough to exercise the compile-time fail-loud gate
1876    /// without standing up a real backend (e.g. `WSOperatorSession`,
1877    /// which lives in a downstream crate).
1878    struct StubOperator {
1879        requires_binding: bool,
1880    }
1881
1882    #[async_trait]
1883    impl Operator for StubOperator {
1884        async fn execute(
1885            &self,
1886            _ctx: &Ctx,
1887            _system: Option<String>,
1888            _prompt: Value,
1889            _worker: Option<WorkerBinding>,
1890            _worker_token: CapToken,
1891        ) -> Result<WorkerResult, WorkerError> {
1892            Ok(WorkerResult {
1893                value: Value::Null,
1894                ok: true,
1895            })
1896        }
1897
1898        fn requires_worker_binding(&self) -> bool {
1899            self.requires_binding
1900        }
1901    }
1902
1903    fn agent_def_with(profile: Option<AgentProfile>) -> AgentDef {
1904        AgentDef {
1905            name: "test-agent".to_string(),
1906            kind: AgentKind::Operator,
1907            spec: serde_json::json!({ "operator_ref": "op1" }),
1908            profile,
1909            meta: None,
1910            runner: None,
1911            runner_ref: None,
1912            verdict: None,
1913        }
1914    }
1915
1916    #[test]
1917    fn build_fails_loud_when_binding_required_but_absent() {
1918        let factory = OperatorSpawnerFactory::new();
1919        factory.register_operator(
1920            "op1",
1921            Arc::new(StubOperator {
1922                requires_binding: true,
1923            }) as Arc<dyn Operator>,
1924        );
1925        let def = agent_def_with(Some(AgentProfile::default()));
1926        match factory.build(&def, None) {
1927            Err(CompileError::InvalidSpec { name, msg }) => {
1928                assert_eq!(name, "test-agent");
1929                assert!(
1930                    msg.contains("worker_binding is required"),
1931                    "unexpected message: {msg}"
1932                );
1933                // Issue #9: the message must be actionable for both
1934                // authoring paths — the JSON-direct hint and the
1935                // $agent_md hint both surface.
1936                assert!(
1937                    msg.contains("agents[N].profile.worker_binding"),
1938                    "message missing JSON-direct hint (issue #9): {msg}"
1939                );
1940                assert!(
1941                    msg.contains("agent .md frontmatter"),
1942                    "message missing $agent_md hint: {msg}"
1943                );
1944            }
1945            Err(other) => panic!("expected InvalidSpec, got: {other:?}"),
1946            Ok(_) => panic!("expected compile-time failure, got Ok"),
1947        }
1948    }
1949
1950    #[test]
1951    fn build_succeeds_when_binding_required_and_present() {
1952        let factory = OperatorSpawnerFactory::new();
1953        factory.register_operator(
1954            "op1",
1955            Arc::new(StubOperator {
1956                requires_binding: true,
1957            }) as Arc<dyn Operator>,
1958        );
1959        let profile = AgentProfile {
1960            worker_binding: Some("mse-worker-coder".to_string()),
1961            tools: vec!["Read".to_string(), "Edit".to_string()],
1962            ..Default::default()
1963        };
1964        let def = agent_def_with(Some(profile));
1965        assert!(
1966            factory.build(&def, None).is_ok(),
1967            "expected Ok when worker_binding is declared"
1968        );
1969    }
1970
1971    #[test]
1972    fn build_succeeds_when_binding_not_required_and_absent() {
1973        let factory = OperatorSpawnerFactory::new();
1974        factory.register_operator(
1975            "op1",
1976            Arc::new(StubOperator {
1977                requires_binding: false,
1978            }) as Arc<dyn Operator>,
1979        );
1980        let def = agent_def_with(Some(AgentProfile::default()));
1981        assert!(
1982            factory.build(&def, None).is_ok(),
1983            "backends that don't require a binding must not be gated by its absence"
1984        );
1985    }
1986}
1987
1988// ─── LuaInProcessSpawnerFactory: inline `spec.source` support ─────────────
1989//
1990// Issue `ab3d1145`: BPs served by `mse serve` couldn't declare `kind: lua`
1991// without pre-registering a `fn_id` on the factory. These tests cover the
1992// new inline path — `spec.source = "<lua chunk>"` (optionally with `label`)
1993// wraps a fresh `LuaScriptSource` at `build` time and runs it through the
1994// same `run_lua_worker` plumbing as the registry path.
1995#[cfg(test)]
1996mod lua_inline_source_tests {
1997    use super::*;
1998    use crate::types::{CapToken, Role, StepId};
1999
2000    fn agent(name: &str, spec: Value) -> AgentDef {
2001        AgentDef {
2002            name: name.to_string(),
2003            kind: AgentKind::Lua,
2004            spec,
2005            profile: None,
2006            meta: None,
2007            runner: None,
2008            runner_ref: None,
2009            verdict: None,
2010        }
2011    }
2012
2013    fn test_invocation(prompt: &str) -> crate::worker::adapter::WorkerInvocation {
2014        crate::worker::adapter::WorkerInvocation {
2015            token: CapToken {
2016                agent_id: "a".into(),
2017                role: Role::Worker,
2018                scopes: vec!["*".into()],
2019                issued_at: 0,
2020                expire_at: u64::MAX / 2,
2021                max_uses: None,
2022                nonce: "test-nonce".into(),
2023                sig_hex: "".into(),
2024            },
2025            task_id: StepId::parse("ST-test").expect("StepId parse"),
2026            attempt: 1,
2027            agent: "g".into(),
2028            prompt: prompt.into(),
2029            sink: None,
2030            cancel_token: None,
2031        }
2032    }
2033
2034    #[test]
2035    fn build_accepts_inline_source_without_pre_registration() {
2036        let factory = LuaInProcessSpawnerFactory::new();
2037        let def = agent(
2038            "g",
2039            serde_json::json!({ "source": "return { value = 42, ok = true }" }),
2040        );
2041        assert!(
2042            factory.build(&def, None).is_ok(),
2043            "inline spec.source must build without a pre-registered fn_id"
2044        );
2045    }
2046
2047    #[test]
2048    fn build_rejects_when_neither_source_nor_fn_id_is_present() {
2049        let factory = LuaInProcessSpawnerFactory::new();
2050        let def = agent("g", serde_json::json!({}));
2051        match factory.build(&def, None) {
2052            Err(CompileError::InvalidSpec { msg, .. }) => {
2053                assert!(
2054                    msg.contains("fn_id"),
2055                    "empty spec must still surface the fn_id-required message: {msg}"
2056                );
2057            }
2058            Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
2059            // `SpawnerAdapter` is not Debug, so we can't `unwrap_err()` /
2060            // pattern-print the Ok arm — describe the mismatch directly.
2061            Ok(_) => panic!("expected InvalidSpec, got Ok(SpawnerAdapter)"),
2062        }
2063    }
2064
2065    /// The inline path shares `run_lua_worker` with the registry path, so
2066    /// exercising the marshaller once through it is enough to prove the
2067    /// wrap is faithful.
2068    #[tokio::test]
2069    async fn inline_source_evaluates_and_marshals_result() {
2070        let source =
2071            LuaScriptSource::new("return { value = _PROMPT .. '!', ok = true }", "smoke.lua");
2072        let out = run_lua_worker(
2073            std::sync::Arc::new(source),
2074            std::sync::Arc::new(HashMap::new()),
2075            test_invocation("hello"),
2076        )
2077        .await
2078        .expect("lua worker ok");
2079        assert_eq!(out.value, serde_json::json!("hello!"));
2080        assert!(out.ok);
2081    }
2082
2083    #[tokio::test]
2084    async fn inline_source_can_signal_agent_level_failure() {
2085        // Deterministic gate pattern: return `ok = false` to flip the
2086        // dispatch outcome to `Blocked` (the flow.ir Try catch path).
2087        let source = LuaScriptSource::new("return { value = 'nope', ok = false }", "gate.lua");
2088        let out = run_lua_worker(
2089            std::sync::Arc::new(source),
2090            std::sync::Arc::new(HashMap::new()),
2091            test_invocation("input"),
2092        )
2093        .await
2094        .expect("lua worker ok");
2095        assert_eq!(out.value, serde_json::json!("nope"));
2096        assert!(!out.ok);
2097    }
2098}
2099
2100// ─── GH #21 Phase 2: `Blueprint.metas` / `AgentMeta.meta_ref` / static
2101// `$step_meta.ref` compile-time validation ─────────────────────────────────
2102#[cfg(test)]
2103mod meta_ref_validation_tests {
2104    use super::*;
2105    use crate::blueprint::{AgentMeta, MetaDef};
2106    use crate::worker::adapter::WorkerResult;
2107
2108    fn registry_with_echo() -> SpawnerRegistry {
2109        let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2110            Ok(WorkerResult {
2111                value: Value::String(inv.prompt),
2112                ok: true,
2113            })
2114        });
2115        let mut reg = SpawnerRegistry::new();
2116        reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2117        reg
2118    }
2119
2120    fn rustfn_agent(name: &str) -> AgentDef {
2121        AgentDef {
2122            name: name.to_string(),
2123            kind: AgentKind::RustFn,
2124            spec: serde_json::json!({ "fn_id": "echo" }),
2125            profile: None,
2126            meta: None,
2127            runner: None,
2128            runner_ref: None,
2129            verdict: None,
2130        }
2131    }
2132
2133    fn simple_flow(agent_ref: &str, in_: Expr) -> FlowNode {
2134        FlowNode::Step {
2135            ref_: agent_ref.to_string(),
2136            in_,
2137            out: Expr::Path {
2138                at: "$.output".parse().expect("literal test path: $.output"),
2139            },
2140        }
2141    }
2142
2143    fn minimal_bp(agents: Vec<AgentDef>, metas: Vec<MetaDef>, flow: FlowNode) -> Blueprint {
2144        Blueprint {
2145            schema_version: crate::blueprint::current_schema_version(),
2146            id: "meta-ref-ut".into(),
2147            flow,
2148            agents,
2149            operators: vec![],
2150            metas,
2151            hints: Default::default(),
2152            strategy: Default::default(),
2153            metadata: BlueprintMetadata::default(),
2154            spawner_hints: Default::default(),
2155            default_agent_kind: AgentKind::Operator,
2156            default_operator_kind: None,
2157            default_init_ctx: None,
2158            default_agent_ctx: None,
2159            default_context_policy: None,
2160            projection_placement: None,
2161            audits: vec![],
2162            degradation_policy: None,
2163            runners: vec![],
2164            default_runner: None,
2165            check_policy: None,
2166            blueprint_ref_includes: Vec::new(),
2167        }
2168    }
2169
2170    #[test]
2171    fn valid_meta_ref_compiles() {
2172        let mut agent = rustfn_agent("worker");
2173        agent.meta = Some(AgentMeta {
2174            meta_ref: Some("shared".to_string()),
2175            ..Default::default()
2176        });
2177        let bp = minimal_bp(
2178            vec![agent],
2179            vec![MetaDef {
2180                name: "shared".into(),
2181                ctx: serde_json::json!({ "k": "v" }),
2182            }],
2183            simple_flow(
2184                "worker",
2185                Expr::Path {
2186                    at: "$.input".parse().expect("literal test path: $.input"),
2187                },
2188            ),
2189        );
2190        let compiler = Compiler::new(registry_with_echo());
2191        assert!(
2192            compiler.compile(&bp).is_ok(),
2193            "a resolvable AgentMeta.meta_ref must compile"
2194        );
2195    }
2196
2197    #[test]
2198    fn unknown_agent_meta_ref_is_unresolved_meta_ref() {
2199        let mut agent = rustfn_agent("worker");
2200        agent.meta = Some(AgentMeta {
2201            meta_ref: Some("missing".to_string()),
2202            ..Default::default()
2203        });
2204        let bp = minimal_bp(
2205            vec![agent],
2206            vec![],
2207            simple_flow(
2208                "worker",
2209                Expr::Path {
2210                    at: "$.input".parse().expect("literal test path: $.input"),
2211                },
2212            ),
2213        );
2214        let compiler = Compiler::new(registry_with_echo());
2215        match compiler.compile(&bp) {
2216            Err(CompileError::UnresolvedMetaRef {
2217                where_,
2218                meta_ref,
2219                defined,
2220            }) => {
2221                assert!(
2222                    where_.contains("worker"),
2223                    "where_ must name the agent: {where_}"
2224                );
2225                assert_eq!(meta_ref, "missing");
2226                assert!(defined.is_empty());
2227            }
2228            Err(other) => {
2229                panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
2230            }
2231            Ok(_) => panic!("expected compile-time failure, got Ok"),
2232        }
2233    }
2234
2235    #[test]
2236    fn unknown_static_step_meta_ref_in_lit_is_unresolved_meta_ref() {
2237        let agent = rustfn_agent("worker");
2238        let in_ = Expr::Lit {
2239            value: serde_json::json!({ "$step_meta": { "ref": "missing" }, "$in": "go" }),
2240        };
2241        let bp = minimal_bp(vec![agent], vec![], simple_flow("worker", in_));
2242        let compiler = Compiler::new(registry_with_echo());
2243        match compiler.compile(&bp) {
2244            Err(CompileError::UnresolvedMetaRef {
2245                where_, meta_ref, ..
2246            }) => {
2247                assert!(
2248                    where_.contains("worker"),
2249                    "where_ must name the offending step: {where_}"
2250                );
2251                assert_eq!(meta_ref, "missing");
2252            }
2253            Err(other) => {
2254                panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
2255            }
2256            Ok(_) => panic!("expected compile-time failure, got Ok"),
2257        }
2258    }
2259
2260    #[test]
2261    fn path_op_input_with_no_static_envelope_compiles_fine() {
2262        let agent = rustfn_agent("worker");
2263        let bp = minimal_bp(
2264            vec![agent],
2265            vec![],
2266            simple_flow(
2267                "worker",
2268                Expr::Path {
2269                    at: "$.input".parse().expect("literal test path: $.input"),
2270                },
2271            ),
2272        );
2273        let compiler = Compiler::new(registry_with_echo());
2274        assert!(
2275            compiler.compile(&bp).is_ok(),
2276            "a non-Lit Step.in must not trigger the best-effort static $step_meta check"
2277        );
2278    }
2279}
2280
2281// ─── GH #34: `Blueprint.audits[].agent` compile-time validation ────────────
2282#[cfg(test)]
2283mod audit_agent_validation_tests {
2284    use super::*;
2285    use crate::worker::adapter::WorkerResult;
2286    use mlua_swarm_schema::{AuditDef, AuditMode};
2287
2288    fn registry_with_echo() -> SpawnerRegistry {
2289        let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2290            Ok(WorkerResult {
2291                value: Value::String(inv.prompt),
2292                ok: true,
2293            })
2294        });
2295        let mut reg = SpawnerRegistry::new();
2296        reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2297        reg
2298    }
2299
2300    fn rustfn_agent(name: &str) -> AgentDef {
2301        AgentDef {
2302            name: name.to_string(),
2303            kind: AgentKind::RustFn,
2304            spec: serde_json::json!({ "fn_id": "echo" }),
2305            profile: None,
2306            meta: None,
2307            runner: None,
2308            runner_ref: None,
2309            verdict: None,
2310        }
2311    }
2312
2313    fn minimal_bp(agents: Vec<AgentDef>, audits: Vec<AuditDef>) -> Blueprint {
2314        Blueprint {
2315            schema_version: crate::blueprint::current_schema_version(),
2316            id: "audit-ref-ut".into(),
2317            flow: FlowNode::Step {
2318                ref_: "worker".to_string(),
2319                in_: Expr::Path {
2320                    at: "$.input".parse().expect("literal test path: $.input"),
2321                },
2322                out: Expr::Path {
2323                    at: "$.output".parse().expect("literal test path: $.output"),
2324                },
2325            },
2326            agents,
2327            operators: vec![],
2328            metas: vec![],
2329            hints: Default::default(),
2330            strategy: Default::default(),
2331            metadata: BlueprintMetadata::default(),
2332            spawner_hints: Default::default(),
2333            default_agent_kind: AgentKind::Operator,
2334            default_operator_kind: None,
2335            default_init_ctx: None,
2336            default_agent_ctx: None,
2337            default_context_policy: None,
2338            projection_placement: None,
2339            audits,
2340            degradation_policy: None,
2341            runners: vec![],
2342            default_runner: None,
2343            check_policy: None,
2344            blueprint_ref_includes: Vec::new(),
2345        }
2346    }
2347
2348    #[test]
2349    fn unresolved_audit_agent_is_a_loud_compile_error() {
2350        let bp = minimal_bp(
2351            vec![rustfn_agent("worker")],
2352            vec![AuditDef {
2353                agent: "missing-auditor".to_string(),
2354                steps: None,
2355                mode: AuditMode::default(),
2356            }],
2357        );
2358        let compiler = Compiler::new(registry_with_echo());
2359        match compiler.compile(&bp) {
2360            Err(CompileError::UnresolvedAuditAgent { agent, defined }) => {
2361                assert_eq!(agent, "missing-auditor");
2362                assert_eq!(defined, vec!["worker".to_string()]);
2363            }
2364            Err(other) => {
2365                panic!("expected UnresolvedAuditAgent, got a different CompileError: {other}")
2366            }
2367            Ok(_) => panic!("expected compile-time failure, got Ok"),
2368        }
2369    }
2370
2371    #[test]
2372    fn resolved_audit_agent_compiles_fine() {
2373        let bp = minimal_bp(
2374            vec![rustfn_agent("worker"), rustfn_agent("auditor")],
2375            vec![AuditDef {
2376                agent: "auditor".to_string(),
2377                steps: None,
2378                mode: AuditMode::default(),
2379            }],
2380        );
2381        let compiler = Compiler::new(registry_with_echo());
2382        assert!(
2383            compiler.compile(&bp).is_ok(),
2384            "an audits[].agent that names a declared AgentDef must compile"
2385        );
2386    }
2387}
2388
2389// ─── GH #27 (follow-up to #23): `Blueprint.projection_placement` compile-time
2390// validation + `CompiledBlueprint.projection_placement` construction ────────
2391#[cfg(test)]
2392mod projection_placement_compile_tests {
2393    use super::*;
2394    use crate::core::projection_placement::{ProjectionPlacement, RootPreference};
2395    use crate::worker::adapter::WorkerResult;
2396    use mlua_swarm_schema::ProjectionPlacementSpec;
2397
2398    fn registry_with_echo() -> SpawnerRegistry {
2399        let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2400            Ok(WorkerResult {
2401                value: Value::String(inv.prompt),
2402                ok: true,
2403            })
2404        });
2405        let mut reg = SpawnerRegistry::new();
2406        reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2407        reg
2408    }
2409
2410    fn minimal_bp(projection_placement: Option<ProjectionPlacementSpec>) -> Blueprint {
2411        Blueprint {
2412            schema_version: crate::blueprint::current_schema_version(),
2413            id: "projection-placement-ut".into(),
2414            flow: FlowNode::Step {
2415                ref_: "worker".to_string(),
2416                in_: Expr::Path {
2417                    at: "$.input".parse().expect("literal test path: $.input"),
2418                },
2419                out: Expr::Path {
2420                    at: "$.output".parse().expect("literal test path: $.output"),
2421                },
2422            },
2423            agents: vec![AgentDef {
2424                name: "worker".to_string(),
2425                kind: AgentKind::RustFn,
2426                spec: serde_json::json!({ "fn_id": "echo" }),
2427                profile: None,
2428                meta: None,
2429                runner: None,
2430                runner_ref: None,
2431                verdict: None,
2432            }],
2433            operators: vec![],
2434            metas: vec![],
2435            hints: Default::default(),
2436            strategy: Default::default(),
2437            metadata: BlueprintMetadata::default(),
2438            spawner_hints: Default::default(),
2439            default_agent_kind: AgentKind::Operator,
2440            default_operator_kind: None,
2441            default_init_ctx: None,
2442            default_agent_ctx: None,
2443            default_context_policy: None,
2444            projection_placement,
2445            audits: vec![],
2446            degradation_policy: None,
2447            runners: vec![],
2448            default_runner: None,
2449            check_policy: None,
2450            blueprint_ref_includes: Vec::new(),
2451        }
2452    }
2453
2454    #[test]
2455    fn undeclared_projection_placement_compiles_to_byte_compat_default() {
2456        let bp = minimal_bp(None);
2457        let compiled = Compiler::new(registry_with_echo())
2458            .compile(&bp)
2459            .expect("undeclared projection_placement compiles");
2460        assert_eq!(
2461            *compiled.projection_placement,
2462            ProjectionPlacement::default()
2463        );
2464    }
2465
2466    #[test]
2467    fn declared_valid_projection_placement_compiles_to_matching_resolver() {
2468        let bp = minimal_bp(Some(ProjectionPlacementSpec {
2469            root: Some("project_root".to_string()),
2470            dir_template: Some("custom/{task_id}/out".to_string()),
2471        }));
2472        let compiled = Compiler::new(registry_with_echo())
2473            .compile(&bp)
2474            .expect("valid projection_placement compiles");
2475        assert_eq!(
2476            compiled.projection_placement.root_preference,
2477            RootPreference::ProjectRoot
2478        );
2479        assert_eq!(
2480            compiled.projection_placement.dir_template,
2481            "custom/{task_id}/out"
2482        );
2483    }
2484
2485    #[test]
2486    fn declared_invalid_dir_template_rejects_compile() {
2487        let bp = minimal_bp(Some(ProjectionPlacementSpec {
2488            root: None,
2489            dir_template: Some("workspace/tasks/ctx".to_string()), // missing {task_id}
2490        }));
2491        match Compiler::new(registry_with_echo()).compile(&bp) {
2492            Err(CompileError::InvalidProjectionPlacement(_)) => {}
2493            Err(other) => {
2494                panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
2495            }
2496            Ok(_) => {
2497                panic!("expected compile-time rejection for a missing {{task_id}} placeholder")
2498            }
2499        }
2500    }
2501
2502    #[test]
2503    fn declared_invalid_root_literal_rejects_compile() {
2504        let bp = minimal_bp(Some(ProjectionPlacementSpec {
2505            root: Some("nope".to_string()),
2506            dir_template: None,
2507        }));
2508        match Compiler::new(registry_with_echo()).compile(&bp) {
2509            Err(CompileError::InvalidProjectionPlacement(_)) => {}
2510            Err(other) => {
2511                panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
2512            }
2513            Ok(_) => panic!("expected compile-time rejection for an invalid root literal"),
2514        }
2515    }
2516}
2517
2518// ─── GH #50: `Blueprint.agents[].verdict` cond↔output-shape lint ──────────
2519#[cfg(test)]
2520mod verdict_contract_lint_tests {
2521    use super::*;
2522    use crate::worker::adapter::WorkerResult;
2523
2524    fn registry_with_echo() -> SpawnerRegistry {
2525        let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2526            Ok(WorkerResult {
2527                value: Value::String(inv.prompt),
2528                ok: true,
2529            })
2530        });
2531        let mut reg = SpawnerRegistry::new();
2532        reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2533        reg
2534    }
2535
2536    fn gate_agent(verdict: Option<VerdictContract>) -> AgentDef {
2537        AgentDef {
2538            name: "gate".to_string(),
2539            kind: AgentKind::RustFn,
2540            spec: serde_json::json!({ "fn_id": "echo" }),
2541            profile: None,
2542            meta: None,
2543            runner: None,
2544            runner_ref: None,
2545            verdict,
2546        }
2547    }
2548
2549    fn minimal_bp(agent: AgentDef, flow: FlowNode) -> Blueprint {
2550        Blueprint {
2551            schema_version: crate::blueprint::current_schema_version(),
2552            id: "verdict-contract-ut".into(),
2553            flow,
2554            agents: vec![agent],
2555            operators: vec![],
2556            metas: vec![],
2557            hints: Default::default(),
2558            strategy: Default::default(),
2559            metadata: BlueprintMetadata::default(),
2560            spawner_hints: Default::default(),
2561            default_agent_kind: AgentKind::Operator,
2562            default_operator_kind: None,
2563            default_init_ctx: None,
2564            default_agent_ctx: None,
2565            default_context_policy: None,
2566            projection_placement: None,
2567            audits: vec![],
2568            degradation_policy: None,
2569            runners: vec![],
2570            default_runner: None,
2571            check_policy: None,
2572            blueprint_ref_includes: Vec::new(),
2573        }
2574    }
2575
2576    fn step(ref_: &str, out_path: &str) -> FlowNode {
2577        FlowNode::Step {
2578            ref_: ref_.to_string(),
2579            in_: Expr::Lit { value: Value::Null },
2580            out: Expr::Path {
2581                at: out_path.parse().expect("literal test path"),
2582            },
2583        }
2584    }
2585
2586    fn noop() -> FlowNode {
2587        FlowNode::Seq { children: vec![] }
2588    }
2589
2590    fn eq_cond(path: &str, lit: &str) -> Expr {
2591        Expr::Eq {
2592            lhs: Box::new(Expr::Path {
2593                at: path.parse().expect("literal test path"),
2594            }),
2595            rhs: Box::new(Expr::Lit {
2596                value: Value::String(lit.to_string()),
2597            }),
2598        }
2599    }
2600
2601    fn branch(cond: Expr, then_: FlowNode, else_: FlowNode) -> FlowNode {
2602        FlowNode::Branch {
2603            cond,
2604            then_: Box::new(then_),
2605            else_: Box::new(else_),
2606        }
2607    }
2608
2609    fn body_contract(values: &[&str]) -> VerdictContract {
2610        VerdictContract {
2611            channel: VerdictChannel::Body,
2612            values: values.iter().map(|v| v.to_string()).collect(),
2613        }
2614    }
2615
2616    fn part_contract(values: &[&str]) -> VerdictContract {
2617        VerdictContract {
2618            channel: VerdictChannel::Part,
2619            values: values.iter().map(|v| v.to_string()).collect(),
2620        }
2621    }
2622
2623    #[test]
2624    fn contract_with_correct_body_channel_and_value_compiles() {
2625        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2626        let flow = FlowNode::Seq {
2627            children: vec![
2628                step("gate", "$.verdict"),
2629                branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
2630            ],
2631        };
2632        let bp = minimal_bp(agent, flow);
2633        assert!(
2634            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2635            "a cond addressing the bare step output must match a channel: \"body\" contract"
2636        );
2637    }
2638
2639    #[test]
2640    fn contract_with_correct_part_channel_and_value_compiles() {
2641        let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
2642        let flow = FlowNode::Seq {
2643            children: vec![
2644                step("gate", "$.gate"),
2645                branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
2646            ],
2647        };
2648        let bp = minimal_bp(agent, flow);
2649        assert!(
2650            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2651            "a cond addressing '<step>.parts.verdict' must match a channel: \"part\" contract"
2652        );
2653    }
2654
2655    #[test]
2656    fn body_channel_contract_rejects_cond_addressing_parts_verdict() {
2657        // Pattern A declared (channel: "body") but the cond addresses the
2658        // Pattern B shape ('$.gate.parts.verdict') instead of the bare
2659        // step output — GH #50 register-time enforcement point 1.
2660        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2661        let flow = FlowNode::Seq {
2662            children: vec![
2663                step("gate", "$.gate"),
2664                branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
2665            ],
2666        };
2667        let bp = minimal_bp(agent, flow);
2668        match Compiler::new(registry_with_echo()).compile(&bp) {
2669            Err(CompileError::VerdictChannelMismatch {
2670                where_,
2671                agent,
2672                expected_channel,
2673                actual_shape,
2674            }) => {
2675                assert_eq!(agent, "gate");
2676                assert_eq!(expected_channel, "body");
2677                assert_eq!(actual_shape, "part");
2678                assert!(where_.contains("Branch cond"), "where_: {where_}");
2679            }
2680            Err(other) => {
2681                panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
2682            }
2683            Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
2684        }
2685    }
2686
2687    #[test]
2688    fn part_channel_contract_rejects_cond_addressing_bare_output() {
2689        // Inverse of the previous case: channel: "part" declared, but the
2690        // cond addresses the bare step output.
2691        let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
2692        let flow = FlowNode::Seq {
2693            children: vec![
2694                step("gate", "$.verdict"),
2695                branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
2696            ],
2697        };
2698        let bp = minimal_bp(agent, flow);
2699        match Compiler::new(registry_with_echo()).compile(&bp) {
2700            Err(CompileError::VerdictChannelMismatch {
2701                agent,
2702                expected_channel,
2703                actual_shape,
2704                ..
2705            }) => {
2706                assert_eq!(agent, "gate");
2707                assert_eq!(expected_channel, "part");
2708                assert_eq!(actual_shape, "body");
2709            }
2710            Err(other) => {
2711                panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
2712            }
2713            Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
2714        }
2715    }
2716
2717    #[test]
2718    fn contract_rejects_lit_outside_declared_values() {
2719        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2720        let flow = FlowNode::Seq {
2721            children: vec![
2722                step("gate", "$.verdict"),
2723                branch(eq_cond("$.verdict", "UNKNOWN"), noop(), noop()),
2724            ],
2725        };
2726        let bp = minimal_bp(agent, flow);
2727        match Compiler::new(registry_with_echo()).compile(&bp) {
2728            Err(CompileError::VerdictValueNotInContract {
2729                agent,
2730                value,
2731                values,
2732                ..
2733            }) => {
2734                assert_eq!(agent, "gate");
2735                assert_eq!(value, "UNKNOWN");
2736                assert_eq!(values, vec!["PASS".to_string(), "BLOCKED".to_string()]);
2737            }
2738            Err(other) => {
2739                panic!("expected VerdictValueNotInContract, got a different CompileError: {other}")
2740            }
2741            Ok(_) => panic!("expected compile-time rejection for a Lit outside declared values"),
2742        }
2743    }
2744
2745    #[test]
2746    fn undeclared_agent_referenced_by_cond_compiles_with_warning_only() {
2747        let agent = gate_agent(None);
2748        let flow = FlowNode::Seq {
2749            children: vec![
2750                step("gate", "$.verdict"),
2751                branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
2752            ],
2753        };
2754        let bp = minimal_bp(agent, flow);
2755        assert!(
2756            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2757            "an undeclared verdict contract must never reject compile (opt-in, back-compat)"
2758        );
2759    }
2760
2761    #[test]
2762    fn in_expr_with_lit_haystack_members_compiles() {
2763        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2764        let cond = Expr::In {
2765            needle: Box::new(Expr::Path {
2766                at: "$.verdict".parse().expect("literal test path"),
2767            }),
2768            haystack: Box::new(Expr::Lit {
2769                value: serde_json::json!(["PASS", "BLOCKED"]),
2770            }),
2771        };
2772        let flow = FlowNode::Seq {
2773            children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
2774        };
2775        let bp = minimal_bp(agent, flow);
2776        assert!(
2777            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2778            "an `In` haystack whose every Lit is a declared value must compile"
2779        );
2780    }
2781
2782    /// GH #50 follow-up (issue `33bc825b`): opt-in strict mode rejects a
2783    /// Blueprint whose declared `verdict.values` set includes at least one
2784    /// entry that no downstream `Branch`/`Loop` `cond` references. The
2785    /// contract declares `["PASS", "BLOCKED"]` but only "BLOCKED" is
2786    /// referenced by the cond → "PASS" is unhandled → `CompileError::
2787    /// VerdictValueUnhandled` under `strict_verdict_handling: Some(true)`.
2788    #[test]
2789    fn strict_mode_rejects_unhandled_declared_value() {
2790        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2791        let flow = FlowNode::Seq {
2792            children: vec![
2793                step("gate", "$.verdict"),
2794                branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
2795            ],
2796        };
2797        let mut bp = minimal_bp(agent, flow);
2798        bp.metadata.strict_verdict_handling = Some(true);
2799        match Compiler::new(registry_with_echo()).compile(&bp) {
2800            Err(CompileError::VerdictValueUnhandled {
2801                agent,
2802                value,
2803                declared_values,
2804                step_ref,
2805            }) => {
2806                assert_eq!(agent, "gate");
2807                assert_eq!(value, "PASS");
2808                assert_eq!(
2809                    declared_values,
2810                    vec!["PASS".to_string(), "BLOCKED".to_string()]
2811                );
2812                assert_eq!(step_ref, "gate");
2813            }
2814            Err(other) => {
2815                panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
2816            }
2817            Ok(_) => panic!(
2818                "expected compile-time rejection for a declared verdict value with no \
2819                 downstream handler under strict_verdict_handling=Some(true)"
2820            ),
2821        }
2822    }
2823
2824    /// GH #50 follow-up (issue `33bc825b`): default mode (i.e.
2825    /// `strict_verdict_handling` absent or `Some(false)`) surfaces
2826    /// unhandled declared values via `tracing::warn!` only — the compile
2827    /// still succeeds. This preserves back-compat with GH #50's original
2828    /// test cases (many of which declare `values = ["PASS", "BLOCKED"]`
2829    /// and cond-reference only one).
2830    #[test]
2831    fn default_mode_permits_unhandled_declared_value() {
2832        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2833        let flow = FlowNode::Seq {
2834            children: vec![
2835                step("gate", "$.verdict"),
2836                branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
2837            ],
2838        };
2839        let bp = minimal_bp(agent, flow);
2840        // `strict_verdict_handling` left as `None` (default)
2841        assert!(
2842            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2843            "default mode must never reject a Blueprint for unhandled declared values \
2844             (opt-in, back-compat with GH #50)"
2845        );
2846    }
2847
2848    /// GH #50 follow-up (issue `33bc825b`): under strict mode, when every
2849    /// declared value is referenced by at least one downstream cond, the
2850    /// compile succeeds. This tests the positive path of the reverse-
2851    /// direction lint.
2852    #[test]
2853    fn strict_mode_accepts_all_declared_values_handled() {
2854        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2855        // Two branches, each cond referencing one declared value —
2856        // together they cover the full `values` set.
2857        let flow = FlowNode::Seq {
2858            children: vec![
2859                step("gate", "$.verdict"),
2860                branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
2861                branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
2862            ],
2863        };
2864        let mut bp = minimal_bp(agent, flow);
2865        bp.metadata.strict_verdict_handling = Some(true);
2866        assert!(
2867            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2868            "strict mode must accept a Blueprint that handles every declared value"
2869        );
2870    }
2871
2872    /// GH #50 follow-up (issue `33bc825b`): under strict mode, an `In`
2873    /// cond whose `Lit` haystack lists every declared value satisfies
2874    /// the handler-coverage check in one go.
2875    #[test]
2876    fn strict_mode_accepts_declared_values_covered_by_in_expr() {
2877        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2878        let cond = Expr::In {
2879            needle: Box::new(Expr::Path {
2880                at: "$.verdict".parse().expect("literal test path"),
2881            }),
2882            haystack: Box::new(Expr::Lit {
2883                value: serde_json::json!(["PASS", "BLOCKED"]),
2884            }),
2885        };
2886        let flow = FlowNode::Seq {
2887            children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
2888        };
2889        let mut bp = minimal_bp(agent, flow);
2890        bp.metadata.strict_verdict_handling = Some(true);
2891        assert!(
2892            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2893            "strict mode must accept an `In` haystack that covers every declared value"
2894        );
2895    }
2896
2897    /// GH #50 follow-up (issue `33bc825b`): under strict mode, a `part`
2898    /// channel contract with unhandled declared value is rejected the same
2899    /// way as the `body` channel case. Confirms channel-agnostic coverage.
2900    #[test]
2901    fn strict_mode_rejects_unhandled_part_channel_value() {
2902        let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
2903        let flow = FlowNode::Seq {
2904            children: vec![
2905                step("gate", "$.gate"),
2906                branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
2907            ],
2908        };
2909        let mut bp = minimal_bp(agent, flow);
2910        bp.metadata.strict_verdict_handling = Some(true);
2911        match Compiler::new(registry_with_echo()).compile(&bp) {
2912            Err(CompileError::VerdictValueUnhandled {
2913                agent,
2914                value,
2915                step_ref,
2916                ..
2917            }) => {
2918                assert_eq!(agent, "gate");
2919                assert_eq!(value, "PASS");
2920                assert_eq!(step_ref, "gate");
2921            }
2922            Err(other) => {
2923                panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
2924            }
2925            Ok(_) => panic!(
2926                "expected compile-time rejection for a declared verdict value with no \
2927                 downstream handler (part channel) under strict_verdict_handling=Some(true)"
2928            ),
2929        }
2930    }
2931
2932    /// Acceptance criterion #7 (5th case): a Blueprint shaped like the
2933    /// existing `02-verdict-loop.json` sample — a `Loop` retrying while
2934    /// `$.verdict == "BLOCKED"` plus a `Branch` on `$.verdict == "PASS"` —
2935    /// but with `verdict` omitted on every agent must compile unchanged
2936    /// (at most `tracing::warn!`) and leave `CompiledAgentTable.
2937    /// verdict_contracts` empty.
2938    #[test]
2939    fn verdict_omitted_blueprint_compiles_unchanged_with_empty_contracts() {
2940        let agent = gate_agent(None);
2941        let flow = FlowNode::Seq {
2942            children: vec![
2943                step("gate", "$.verdict"),
2944                FlowNode::Loop {
2945                    counter: Expr::Path {
2946                        at: "$.n".parse().expect("literal test path"),
2947                    },
2948                    cond: eq_cond("$.verdict", "BLOCKED"),
2949                    body: Box::new(step("gate", "$.verdict")),
2950                    max: 3,
2951                },
2952                branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
2953            ],
2954        };
2955        let bp = minimal_bp(agent, flow);
2956        let compiled = Compiler::new(registry_with_echo())
2957            .compile(&bp)
2958            .expect("a verdict-omitted Blueprint must compile unchanged");
2959        assert!(
2960            compiled.router.verdict_contracts.is_empty(),
2961            "no agent declared a verdict contract"
2962        );
2963    }
2964}