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