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.map(|set| set.contains(value)).unwrap_or(false);
1018            if handled {
1019                continue;
1020            }
1021            if strict_verdict_handling {
1022                errors.push(CompileError::VerdictValueUnhandled {
1023                    agent: agent.clone(),
1024                    value: value.clone(),
1025                    declared_values: contract.values.clone(),
1026                    step_ref: step_ref.clone(),
1027                });
1028            } else {
1029                tracing::warn!(
1030                    agent = %agent,
1031                    value = %value,
1032                    step_ref = %step_ref,
1033                    "declared verdict value has no downstream cond handler; \
1034                     opt in to `metadata.strict_verdict_handling` to reject at compile"
1035                );
1036            }
1037        }
1038    }
1039}
1040
1041// ─── CompiledAgentTable ───────────────────────────────────────────────────────
1042
1043/// The compile result: an `agent name → SpawnerAdapter` lookup table.
1044///
1045/// Looks `routes` up by `ctx.agent` (the flow.ir `Step.ref`) and hands
1046/// the spawn to the matching `SpawnerAdapter`. If the name is not
1047/// registered and a `default` is configured, the default is used; if
1048/// there is no default, `SpawnError::NotRegistered` is returned.
1049///
1050/// Layer wrapping (`AuditMiddleware` / `MainAIMiddleware` and friends) is
1051/// not this type's concern — that is done separately in
1052/// `service::linker::link`.
1053pub struct CompiledAgentTable {
1054    pub(crate) routes: HashMap<String, Arc<dyn SpawnerAdapter>>,
1055    pub(crate) default: Option<Arc<dyn SpawnerAdapter>>,
1056    /// GH #50: `AgentDef.name` → declared `VerdictContract`, for every
1057    /// agent that declared one (built by `Compiler::compile`, alongside
1058    /// `routes`). Backs the submit-time enforcement point (a follow-up).
1059    pub(crate) verdict_contracts: HashMap<String, VerdictContract>,
1060}
1061
1062impl CompiledAgentTable {
1063    /// Whether the given agent name is registered in the table — i.e.,
1064    /// whether its spawner has been resolved.
1065    pub fn has_route(&self, agent: &str) -> bool {
1066        self.routes.contains_key(agent)
1067    }
1068    /// List every resolved agent name.
1069    pub fn routed_agents(&self) -> Vec<String> {
1070        self.routes.keys().cloned().collect()
1071    }
1072    /// GH #50: the declared [`VerdictContract`] for `agent`, if any —
1073    /// `None` both when `agent` is unresolved and when it resolved but
1074    /// declared no contract (opt-in; see `AgentDef::verdict`'s doc).
1075    pub fn verdict_contract_for(&self, agent: &str) -> Option<&VerdictContract> {
1076        self.verdict_contracts.get(agent)
1077    }
1078}
1079
1080#[async_trait]
1081impl SpawnerAdapter for CompiledAgentTable {
1082    async fn spawn(
1083        &self,
1084        engine: &Engine,
1085        ctx: &Ctx,
1086        task_id: StepId,
1087        attempt: u32,
1088        token: CapToken,
1089    ) -> Result<Box<dyn Worker>, SpawnError> {
1090        let sp = self
1091            .routes
1092            .get(&ctx.agent)
1093            .cloned()
1094            .or_else(|| self.default.clone())
1095            .ok_or_else(|| SpawnError::NotRegistered(ctx.agent.clone()))?;
1096        sp.spawn(engine, ctx, task_id, attempt, token).await
1097    }
1098}
1099
1100// ─── default factories (three variants) ───────────────────────────────────
1101
1102/// Factory for `AgentKind::Subprocess`. Turns the spec into a
1103/// [`ProcessSpawner`].
1104///
1105/// Naming convention: `<WorkerIMPL><AdapterType>SpawnerFactory`. Factory
1106/// names carry both the worker implementation and the host adapter so
1107/// they are not confused with each other; the old
1108/// `ShellSpawnerFactory` was renamed to this.
1109///
1110/// Spec shape:
1111/// ```jsonc
1112/// { "program": "agent-block", "args": ["-s","s.lua"],
1113///   "use_stdin": true,                       // optional, default = true
1114///   "stream_mode": "ndjson_lines" | "sse_events" | "length_prefixed" | null  // optional, default = null (plain)
1115/// }
1116/// ```
1117pub struct SubprocessProcessSpawnerFactory;
1118
1119impl SpawnerFactoryKind for SubprocessProcessSpawnerFactory {
1120    const KIND: AgentKind = AgentKind::Subprocess;
1121    type Worker = crate::worker::process_spawner::ProcessWorker;
1122}
1123
1124impl SpawnerFactory for SubprocessProcessSpawnerFactory {
1125    fn build(
1126        &self,
1127        agent_def: &AgentDef,
1128        _hint: Option<&Value>,
1129    ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1130        let agent_name = &agent_def.name;
1131        let spec = &agent_def.spec;
1132        let invalid = |msg: String| CompileError::InvalidSpec {
1133            name: agent_name.to_string(),
1134            msg,
1135        };
1136        let program = spec
1137            .get("program")
1138            .and_then(|v| v.as_str())
1139            .ok_or_else(|| invalid("shell spec: 'program' (string) required".into()))?
1140            .to_string();
1141        let args: Vec<String> = spec
1142            .get("args")
1143            .and_then(|v| v.as_array())
1144            .map(|a| {
1145                a.iter()
1146                    .filter_map(|x| x.as_str().map(|s| s.to_string()))
1147                    .collect()
1148            })
1149            .unwrap_or_default();
1150        let use_stdin = spec
1151            .get("use_stdin")
1152            .and_then(|v| v.as_bool())
1153            .unwrap_or(true);
1154        let stream_mode = match spec.get("stream_mode").and_then(|v| v.as_str()) {
1155            Some("ndjson_lines") => Some(StreamMode::NdjsonLines),
1156            Some("sse_events") => Some(StreamMode::SseEvents),
1157            Some("length_prefixed") => Some(StreamMode::LengthPrefixed),
1158            Some(other) => return Err(invalid(format!("unknown stream_mode: {other}"))),
1159            None => None,
1160        };
1161
1162        let mut sp = ProcessSpawner {
1163            program,
1164            args,
1165            use_stdin,
1166            stream_mode,
1167        };
1168        if let Some(mode) = sp.stream_mode.clone() {
1169            sp = sp.stream_mode(mode);
1170        }
1171        Ok(Arc::new(sp))
1172    }
1173}
1174
1175/// Factory for `AgentKind::Lua`. At `build` time it inspects the
1176/// `AgentDef.spec` and returns an [`InProcSpawner`] with the Lua-eval
1177/// `WorkerFn` registered under `agent_name` — one `InProcSpawner`
1178/// instance per agent.
1179///
1180/// Naming convention: `<WorkerIMPL><AdapterType>SpawnerFactory` (Lua
1181/// worker on InProcess adapter). One half of the old
1182/// `InProcSpawnerFactory`, split into Lua and RustFn variants.
1183///
1184/// Spec shape (choose one; `source` wins when both are present):
1185///
1186/// ```jsonc
1187/// // (a) Registry lookup — Lua source id pre-registered with the
1188/// //     factory via `register_lua` (used by the enhance flow's built-in
1189/// //     workers). Requires the factory to know the id at construction
1190/// //     time.
1191/// { "fn_id": "patch-spawner" }
1192///
1193/// // (b) Inline source — a Lua chunk carried by the Blueprint itself,
1194/// //     wrapped on the fly at `build` time. Combined with the loader's
1195/// //     `$file` ref expansion (`"source": {"$file": "gates/foo.lua"}`)
1196/// //     this lets a BP ship deterministic Lua gates without any
1197/// //     pre-registration. `label` is optional and defaults to
1198/// //     `"<agent_name>.lua"` for error messages.
1199/// { "source": "return { value = 42, ok = true }",
1200///   "label": "psim-gate.lua" }
1201/// ```
1202///
1203/// Host bridges registered on the factory (see [`Self::with_bridge`])
1204/// apply to both spec shapes.
1205pub struct LuaInProcessSpawnerFactory {
1206    registry: HashMap<String, WorkerFn>,
1207    bridges: HashMap<String, HostBridge>,
1208}
1209
1210/// Rust-side bridge function callable from Lua.
1211///
1212/// Inputs and outputs are both `serde_json::Value` (i.e. JSON). Lua
1213/// invokes it as `host.<name>(arg_table)`. If the implementation needs
1214/// to call async Rust, the caller does the sync-ification (typically
1215/// `tokio::runtime::Handle::current().block_on(...)`).
1216///
1217/// Design intent: keep Lua scripts focused on flow control and `ctx`
1218/// walking, while the heavy lifting (LLM calls, RFC 6902 apply,
1219/// verifiers, and so on) stays on the Rust side. Going "pure Lua" —
1220/// removing the bridge — is a carry.
1221#[derive(Clone)]
1222pub struct HostBridge(
1223    Arc<dyn Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync>,
1224);
1225
1226impl HostBridge {
1227    /// Wrap a Rust closure as a bridge callable from Lua.
1228    pub fn new<F>(f: F) -> Self
1229    where
1230        F: Fn(serde_json::Value) -> Result<serde_json::Value, String> + Send + Sync + 'static,
1231    {
1232        Self(Arc::new(f))
1233    }
1234
1235    /// Invoke the bridge directly — a thin trampoline over the inner
1236    /// `Fn`. The production path goes through the Lua runtime, but this
1237    /// stays `pub` so unit tests can exercise the primitive directly.
1238    pub fn call(&self, arg: serde_json::Value) -> Result<serde_json::Value, String> {
1239        (self.0)(arg)
1240    }
1241}
1242
1243/// Carrier type for Lua script sources. Paths are not required — a
1244/// source string plus an identifying label is all it holds.
1245///
1246/// Callers bring in the source (via `include_str!` or similar) and
1247/// register it with the factory through
1248/// [`LuaInProcessSpawnerFactory::register_lua`].
1249#[derive(Clone)]
1250pub struct LuaScriptSource {
1251    /// The Lua chunk source.
1252    pub source: String,
1253    /// Label used in error messages — typically the script's logical id
1254    /// (for example `"patch_spawner.lua"`).
1255    pub label: String,
1256}
1257
1258impl LuaScriptSource {
1259    /// Wrap a Lua chunk source and its error-message label.
1260    pub fn new(source: impl Into<String>, label: impl Into<String>) -> Self {
1261        Self {
1262            source: source.into(),
1263            label: label.into(),
1264        }
1265    }
1266}
1267
1268impl LuaInProcessSpawnerFactory {
1269    /// Start with no registered scripts and no host bridges.
1270    pub fn new() -> Self {
1271        Self {
1272            registry: HashMap::new(),
1273            bridges: HashMap::new(),
1274        }
1275    }
1276
1277    /// Register a host bridge. Subsequent `register_lua` calls snapshot
1278    /// the current bridge set.
1279    ///
1280    /// Ordering rule: register bridges first, then call `register_lua`;
1281    /// bridges added after `register_lua` will not be visible to that
1282    /// script.
1283    pub fn with_bridge(mut self, name: impl Into<String>, bridge: HostBridge) -> Self {
1284        self.bridges.insert(name.into(), bridge);
1285        self
1286    }
1287
1288    /// Register a **Lua-eval Worker** under `fn_id`.
1289    ///
1290    /// Each dispatch spins up a fresh `mlua::Lua` VM, injects globals
1291    /// (`_PROMPT` / `_AGENT` / `_TASK_ID` / `_ATTEMPT` / `_CTX` — the last
1292    /// is `_PROMPT` parsed as JSON, or `nil` if that fails), evaluates
1293    /// the script, and marshals the returned table into a `WorkerResult`.
1294    ///
1295    /// Marshalling rules for the return value:
1296    /// - `{ value = ..., ok = bool }` → `WorkerResult.value` /
1297    ///   `WorkerResult.ok` verbatim.
1298    /// - Anything else → `value = <returned value>`, `ok = true`.
1299    ///
1300    /// Execution runs on `tokio::task::spawn_blocking` because `mlua::Lua`
1301    /// is `!Send` and needs to stay away from the tokio async context.
1302    /// Host bridges (the Lua-to-Rust callback path) previously registered
1303    /// with [`Self::with_bridge`] are snapshotted at call time and
1304    /// injected into every dispatch inside `run_lua_worker`.
1305    pub fn register_lua(mut self, fn_id: impl Into<String>, source: LuaScriptSource) -> Self {
1306        let source = Arc::new(source);
1307        let bridges = Arc::new(self.bridges.clone());
1308        let wrapped: WorkerFn = Arc::new(move |inv| {
1309            let source = source.clone();
1310            let bridges = bridges.clone();
1311            Box::pin(run_lua_worker(source, bridges, inv))
1312        });
1313        self.registry.insert(fn_id.into(), wrapped);
1314        self
1315    }
1316}
1317
1318/// Body of a single Lua-eval invocation (called from `register_lua`).
1319async fn run_lua_worker(
1320    source: Arc<LuaScriptSource>,
1321    bridges: Arc<HashMap<String, HostBridge>>,
1322    inv: crate::worker::adapter::WorkerInvocation,
1323) -> Result<crate::worker::adapter::WorkerResult, crate::worker::adapter::WorkerError> {
1324    use crate::worker::adapter::WorkerError;
1325    use mlua::LuaSerdeExt;
1326
1327    let label = source.label.clone();
1328    let outcome =
1329        tokio::task::spawn_blocking(move || -> Result<(serde_json::Value, bool), String> {
1330            let lua = mlua::Lua::new();
1331            let g = lua.globals();
1332
1333            // 1. Base globals.
1334            g.set("_PROMPT", inv.prompt.clone())
1335                .map_err(|e| format!("set _PROMPT: {e}"))?;
1336            g.set("_AGENT", inv.agent.clone())
1337                .map_err(|e| format!("set _AGENT: {e}"))?;
1338            g.set("_TASK_ID", inv.task_id.to_string())
1339                .map_err(|e| format!("set _TASK_ID: {e}"))?;
1340            g.set("_ATTEMPT", inv.attempt as i64)
1341                .map_err(|e| format!("set _ATTEMPT: {e}"))?;
1342
1343            // 2. _CTX = JSON parse(_PROMPT); nil on parse failure (co-exists with the plain-string prompt path).
1344            if let Ok(json_val) = serde_json::from_str::<serde_json::Value>(&inv.prompt) {
1345                let lua_val = lua
1346                    .to_value(&json_val)
1347                    .map_err(|e| format!("_CTX to_value: {e}"))?;
1348                g.set("_CTX", lua_val)
1349                    .map_err(|e| format!("set _CTX: {e}"))?;
1350            }
1351
1352            // 3. Inject the host bridge (Lua can call `host.<name>(arg)`).
1353            if !bridges.is_empty() {
1354                let host = lua
1355                    .create_table()
1356                    .map_err(|e| format!("create host table: {e}"))?;
1357                for (name, bridge) in bridges.iter() {
1358                    let bridge = bridge.clone();
1359                    let bname = name.clone();
1360                    let f = lua
1361                        .create_function(move |lua, arg: mlua::Value| {
1362                            let json_arg: serde_json::Value = lua.from_value(arg).map_err(|e| {
1363                                mlua::Error::external(format!("bridge {bname} arg → json: {e}"))
1364                            })?;
1365                            let result_json =
1366                                bridge.call(json_arg).map_err(mlua::Error::external)?;
1367                            lua.to_value(&result_json).map_err(|e| {
1368                                mlua::Error::external(format!("bridge {bname} ret → lua: {e}"))
1369                            })
1370                        })
1371                        .map_err(|e| format!("create_function {name}: {e}"))?;
1372                    host.set(name.as_str(), f)
1373                        .map_err(|e| format!("host.{name} set: {e}"))?;
1374                }
1375                g.set("host", host).map_err(|e| format!("set host: {e}"))?;
1376            }
1377
1378            // 4. eval
1379            let result: mlua::Value = lua
1380                .load(&source.source)
1381                .set_name(&source.label)
1382                .eval()
1383                .map_err(|e| format!("lua eval [{}]: {e}", source.label))?;
1384
1385            // 5. Marshal: shape `{ value=..., ok=true }` or raw value.
1386            let json_result: serde_json::Value = lua
1387                .from_value(result)
1388                .map_err(|e| format!("lua → json [{}]: {e}", source.label))?;
1389
1390            let (value, ok) = match &json_result {
1391                serde_json::Value::Object(map)
1392                    if map.contains_key("value") || map.contains_key("ok") =>
1393                {
1394                    let ok = map.get("ok").and_then(|v| v.as_bool()).unwrap_or(true);
1395                    let value = map.get("value").cloned().unwrap_or(json_result.clone());
1396                    (value, ok)
1397                }
1398                _ => (json_result, true),
1399            };
1400            Ok((value, ok))
1401        })
1402        .await
1403        .map_err(|e| WorkerError::Failed(format!("spawn_blocking join [{label}]: {e}")))?
1404        .map_err(WorkerError::Failed)?;
1405
1406    Ok(crate::worker::adapter::WorkerResult {
1407        value: outcome.0,
1408        ok: outcome.1,
1409    })
1410}
1411
1412impl Default for LuaInProcessSpawnerFactory {
1413    fn default() -> Self {
1414        Self::new()
1415    }
1416}
1417
1418impl SpawnerFactoryKind for LuaInProcessSpawnerFactory {
1419    const KIND: AgentKind = AgentKind::Lua;
1420    type Worker = LuaWorker;
1421}
1422
1423impl SpawnerFactory for LuaInProcessSpawnerFactory {
1424    fn build(
1425        &self,
1426        agent_def: &AgentDef,
1427        _hint: Option<&Value>,
1428    ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1429        // Inline `spec.source` (a Lua chunk carried by the BP itself) takes
1430        // precedence over `spec.fn_id`. This is the path a BP author uses to
1431        // ship a deterministic Lua gate without pre-registering it with the
1432        // factory — the plumbing (`run_lua_worker` / `LuaScriptSource`) is
1433        // the same, only the entry point differs.
1434        if let Some(source) = agent_def.spec.get("source").and_then(|v| v.as_str()) {
1435            let label = agent_def
1436                .spec
1437                .get("label")
1438                .and_then(|v| v.as_str())
1439                .map(str::to_string)
1440                .unwrap_or_else(|| format!("{}.lua", agent_def.name));
1441            let script = Arc::new(LuaScriptSource::new(source.to_string(), label));
1442            let bridges = Arc::new(self.bridges.clone());
1443            let wrapped: WorkerFn = Arc::new(move |inv| {
1444                let source = script.clone();
1445                let bridges = bridges.clone();
1446                Box::pin(run_lua_worker(source, bridges, inv))
1447            });
1448            let mut sp: InProcSpawner<LuaWorker> = InProcSpawner::<LuaWorker>::typed();
1449            sp.registry.insert(agent_def.name.to_string(), wrapped);
1450            return Ok(Arc::new(sp));
1451        }
1452        build_inproc_from_registry::<LuaWorker>(&self.registry, agent_def, "lua")
1453    }
1454}
1455
1456/// Factory for `AgentKind::RustFn`. At `build` time it looks the `fn_id`
1457/// up in its internal registry and returns an [`InProcSpawner`] with the
1458/// Rust closure `WorkerFn` registered under `agent_name`.
1459///
1460/// Naming convention: `<WorkerIMPL><AdapterType>SpawnerFactory` (RustFn
1461/// worker on InProcess adapter). Sibling to
1462/// [`LuaInProcessSpawnerFactory`] — the Lua-worker half of the same
1463/// split.
1464///
1465/// Spec shape:
1466/// ```jsonc
1467/// { "fn_id": "echo" }     // Rust closure id pre-registered with the factory
1468/// ```
1469pub struct RustFnInProcessSpawnerFactory {
1470    registry: HashMap<String, WorkerFn>,
1471}
1472
1473impl RustFnInProcessSpawnerFactory {
1474    /// Start with no registered closures.
1475    pub fn new() -> Self {
1476        Self {
1477            registry: HashMap::new(),
1478        }
1479    }
1480
1481    /// Register a Rust closure `WorkerFn` under `fn_id`, wrapping it so
1482    /// it matches the `WorkerFn` signature (boxed, pinned future).
1483    pub fn register_fn<F, Fut>(mut self, fn_id: impl Into<String>, f: F) -> Self
1484    where
1485        F: Fn(crate::worker::adapter::WorkerInvocation) -> Fut + Send + Sync + 'static,
1486        Fut: std::future::Future<
1487                Output = Result<
1488                    crate::worker::adapter::WorkerResult,
1489                    crate::worker::adapter::WorkerError,
1490                >,
1491            > + Send
1492            + 'static,
1493    {
1494        let f = Arc::new(f);
1495        let wrapped: WorkerFn = Arc::new(move |inv| {
1496            let f = f.clone();
1497            Box::pin(f(inv))
1498        });
1499        self.registry.insert(fn_id.into(), wrapped);
1500        self
1501    }
1502}
1503
1504impl Default for RustFnInProcessSpawnerFactory {
1505    fn default() -> Self {
1506        Self::new()
1507    }
1508}
1509
1510impl SpawnerFactoryKind for RustFnInProcessSpawnerFactory {
1511    const KIND: AgentKind = AgentKind::RustFn;
1512    type Worker = RustFnWorker;
1513}
1514
1515impl SpawnerFactory for RustFnInProcessSpawnerFactory {
1516    fn build(
1517        &self,
1518        agent_def: &AgentDef,
1519        _hint: Option<&Value>,
1520    ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1521        build_inproc_from_registry::<RustFnWorker>(&self.registry, agent_def, "rust_fn")
1522    }
1523}
1524
1525/// Shared build helper used by both the Lua and the RustFn factories —
1526/// look `spec.fn_id` up in the registry and return an `InProcSpawner`.
1527/// The generic type parameter `W` fixes the per-kind Worker concrete
1528/// type at the type level (the build-site half of the trait's
1529/// associated-type binding across the four-layer cascade).
1530fn build_inproc_from_registry<W>(
1531    registry: &HashMap<String, WorkerFn>,
1532    agent_def: &AgentDef,
1533    kind_label: &str,
1534) -> Result<Arc<dyn SpawnerAdapter>, CompileError>
1535where
1536    W: crate::worker::Worker + From<crate::worker::WorkerJoinHandler> + Send + Sync + 'static,
1537{
1538    let agent_name = &agent_def.name;
1539    let spec = &agent_def.spec;
1540    let invalid = |msg: String| CompileError::InvalidSpec {
1541        name: agent_name.to_string(),
1542        msg,
1543    };
1544    let fn_id = spec
1545        .get("fn_id")
1546        .and_then(|v| v.as_str())
1547        .ok_or_else(|| invalid(format!("{kind_label} spec: 'fn_id' (string) required")))?;
1548    let f = registry
1549        .get(fn_id)
1550        .cloned()
1551        .ok_or_else(|| invalid(format!("fn_id '{fn_id}' not registered in factory")))?;
1552    let mut sp: InProcSpawner<W> = InProcSpawner::<W>::typed();
1553    // Register under `agent_name` (the flow's `Step.ref`). Both
1554    // `CompiledAgentTable` and the `InProcSpawner` look the function up
1555    // by name, so the same key is needed at both layers.
1556    sp.registry.insert(agent_name.to_string(), f);
1557    Ok(Arc::new(sp))
1558}
1559
1560/// Concrete Worker type for the Lua kind — a handle to a Lua-eval task
1561/// inside an mlua VM. Embeds a `WorkerJoinHandler`. Reserved as the home
1562/// for future Lua-specific extensions (an mlua VM cancellation
1563/// mechanism, Lua-side error type retention, and so on).
1564pub struct LuaWorker {
1565    /// The join handle / cancellation token for the underlying task.
1566    pub handler: crate::worker::WorkerJoinHandler,
1567}
1568
1569impl From<crate::worker::WorkerJoinHandler> for LuaWorker {
1570    fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
1571        Self { handler }
1572    }
1573}
1574
1575#[async_trait::async_trait]
1576impl crate::worker::Worker for LuaWorker {
1577    fn id(&self) -> &crate::types::WorkerId {
1578        &self.handler.worker_id
1579    }
1580    fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
1581        self.handler.cancel.clone()
1582    }
1583    async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
1584        self.handler.await_completion().await
1585    }
1586}
1587
1588/// Concrete Worker type for the RustFn kind — a handle to a task that
1589/// directly calls a Rust closure. Embeds a `WorkerJoinHandler`. Being a
1590/// pure function, there is minimal kind-specific extension surface here;
1591/// the primary purpose is to nail down the type binding.
1592pub struct RustFnWorker {
1593    /// The join handle / cancellation token for the underlying task.
1594    pub handler: crate::worker::WorkerJoinHandler,
1595}
1596
1597impl From<crate::worker::WorkerJoinHandler> for RustFnWorker {
1598    fn from(handler: crate::worker::WorkerJoinHandler) -> Self {
1599        Self { handler }
1600    }
1601}
1602
1603#[async_trait::async_trait]
1604impl crate::worker::Worker for RustFnWorker {
1605    fn id(&self) -> &crate::types::WorkerId {
1606        &self.handler.worker_id
1607    }
1608    fn cancel_token(&self) -> tokio_util::sync::CancellationToken {
1609        self.handler.cancel.clone()
1610    }
1611    async fn join(self: Box<Self>) -> Result<(), crate::worker::adapter::WorkerError> {
1612        self.handler.await_completion().await
1613    }
1614}
1615
1616/// Factory for `AgentKind::Operator`. Looks up the `Arc<dyn Operator>`
1617/// pre-registered under `spec.operator_ref` and wraps it in an
1618/// `OperatorSpawner`. Also resolves `AgentDef.profile.worker_binding` into
1619/// a `WorkerBinding` at compile time and fails loud (`CompileError::InvalidSpec`)
1620/// when the resolved operator's `Operator::requires_worker_binding` is `true`
1621/// and no binding was declared.
1622///
1623/// Spec shape:
1624/// ```jsonc
1625/// { "operator_ref": "main_ai" }     // Operator id pre-registered with the factory
1626/// ```
1627///
1628/// # Split of responsibilities with `OperatorDelegateMiddleware`
1629///
1630/// The two axes exist for different reasons:
1631///
1632/// - **This factory (`OperatorSpawnerFactory` → `OperatorSpawner`) — the
1633///   AgentSpec axis.** Bakes a separate Operator backend into each
1634///   `AgentDef`. A `kind = Operator` `AgentDef` names its backend through
1635///   `spec.operator_ref`; at `compile()` time the `Arc<dyn Operator>` is
1636///   baked into `routes[agent_name]`. Because the `agent.md` loader
1637///   (`agent_md_loader`) defaults `kind` to `Operator`, agents that flow
1638///   in through agent-profiles land here.
1639///
1640/// - **`OperatorDelegateMiddleware` — the Blueprint-global (session)
1641///   axis.** Delegates every agent to the same Operator backend. At
1642///   session-attach time you call `engine.register_operator(id, op)`
1643///   plus `attach_with_ids(.., operator_backend_id = Some(id))` to bind
1644///   it session-wide, and declare
1645///   `spawner_hints.layers = ["operator_delegate"]` to opt in. `ctx.agent`
1646///   is ignored; the operator handles every spawn in that session (a
1647///   MainAI-wide driver, a human-wide console, that sort of thing).
1648///
1649/// # Exclusivity (a double fire is structurally impossible)
1650///
1651/// When both are effective — the hint is declared, the session has an
1652/// operator backend, **and** the Blueprint has a `kind = Operator`
1653/// `AgentDef` — `OperatorDelegateMiddleware` sits at the outer end of
1654/// the stack and **completely bypasses** `inner.spawn`. The
1655/// `OperatorSpawner` is never reached, so under those conditions this
1656/// factory's routes entry is inert. This is not a double fire — the
1657/// session axis is overriding the agent axis. Consistent usage means
1658/// picking one axis per use case.
1659///
1660/// Interior mutability is provided by an `Arc<RwLock>`. Even after the
1661/// factory has been stored as `Arc<dyn SpawnerFactory>` in
1662/// `SpawnerRegistry`, a caller holding an `Arc` clone can still add
1663/// Operator backends dynamically via `register_operator(&self, id, op)`.
1664/// Typical uses: registering a `WSOperatorSession` under the session id
1665/// on WebSocket connect, binding agents that arrive via the `agent.md`
1666/// loader to arbitrary backends, and so on. `build()` performs a
1667/// `read()` lookup each time.
1668pub struct OperatorSpawnerFactory {
1669    operators: Arc<std::sync::RwLock<HashMap<String, Arc<dyn Operator>>>>,
1670}
1671
1672impl OperatorSpawnerFactory {
1673    /// Start with no registered Operator backends.
1674    pub fn new() -> Self {
1675        Self {
1676            operators: Arc::new(std::sync::RwLock::new(HashMap::new())),
1677        }
1678    }
1679
1680    /// Register an Operator backend dynamically through `&self`.
1681    /// Overwrites are allowed — later wins. Callers can still reach this
1682    /// after the factory has been stored as `Arc<dyn SpawnerFactory>` in
1683    /// `SpawnerRegistry`, as long as they hold an `Arc` clone; interior
1684    /// mutability is provided by the inner `RwLock`.
1685    pub fn register_operator(&self, id: impl Into<String>, op: Arc<dyn Operator>) -> &Self {
1686        self.operators
1687            .write()
1688            .expect("OperatorSpawnerFactory.operators RwLock poisoned")
1689            .insert(id.into(), op);
1690        self
1691    }
1692
1693    /// Dynamically unregister an id (used to clean up when a WebSocket
1694    /// disconnects, for example). A missing id is a no-op.
1695    pub fn unregister_operator(&self, id: &str) -> &Self {
1696        self.operators
1697            .write()
1698            .expect("OperatorSpawnerFactory.operators RwLock poisoned")
1699            .remove(id);
1700        self
1701    }
1702}
1703
1704impl Default for OperatorSpawnerFactory {
1705    fn default() -> Self {
1706        Self::new()
1707    }
1708}
1709
1710impl SpawnerFactoryKind for OperatorSpawnerFactory {
1711    const KIND: AgentKind = AgentKind::Operator;
1712    type Worker = crate::operator::OperatorWorker;
1713}
1714
1715impl SpawnerFactory for OperatorSpawnerFactory {
1716    fn build(
1717        &self,
1718        agent_def: &AgentDef,
1719        _hint: Option<&Value>,
1720    ) -> Result<Arc<dyn SpawnerAdapter>, CompileError> {
1721        let agent_name = &agent_def.name;
1722        let spec = &agent_def.spec;
1723        // Bake AgentDef.profile.system_prompt into the OperatorSpawner at compile time.
1724        // `Some` → adopted first at spawn time; `None` → falls back to fetch_prompt (initial_directive).
1725        // Fallback path. Sibling: AgentBlockInProcessSpawnerFactory
1726        // (agent_block/runtime.rs) does the same compile-time bake by stuffing
1727        // the profile into BlockConfig.context.
1728        let system_prompt = agent_def.profile.as_ref().map(|p| p.system_prompt.clone());
1729        let invalid = |msg: String| CompileError::InvalidSpec {
1730            name: agent_name.to_string(),
1731            msg,
1732        };
1733        let op_ref = spec
1734            .get("operator_ref")
1735            .and_then(|v| v.as_str())
1736            .ok_or_else(|| invalid("operator spec: 'operator_ref' (string) required".into()))?;
1737        let operators = self
1738            .operators
1739            .read()
1740            .expect("OperatorSpawnerFactory.operators RwLock poisoned");
1741        let op = operators.get(op_ref).cloned().ok_or_else(|| {
1742            let mut names: Vec<String> = operators.keys().cloned().collect();
1743            names.sort();
1744            let names_list = if names.is_empty() {
1745                "<none>".to_string()
1746            } else {
1747                names.join(", ")
1748            };
1749            invalid(format!(
1750                "operator_ref '{op_ref}' not registered in factory. \
1751                 Registered sids: [{names_list}]. \
1752                 Hint: call mse_operator_join(roles=[...]) to mint the sid first."
1753            ))
1754        })?;
1755        drop(operators);
1756
1757        // Resolve the Blueprint-baked worker binding from
1758        // `AgentDef.profile.worker_binding` — the SoT for the
1759        // declaration↔executor binding (see `WorkerBinding` doc). Fail
1760        // loud at compile time when the operator backend requires one
1761        // and the Blueprint didn't declare it; this is a compile-time
1762        // gate, not a runtime guess.
1763        let worker_binding = agent_def
1764            .profile
1765            .as_ref()
1766            .and_then(|p| p.worker_binding.as_ref())
1767            .map(|variant| WorkerBinding {
1768                variant: variant.clone(),
1769                tools: agent_def
1770                    .profile
1771                    .as_ref()
1772                    .map(|p| p.tools.clone())
1773                    .unwrap_or_default(),
1774            });
1775        if op.requires_worker_binding() && worker_binding.is_none() {
1776            // Issue #9: the two Blueprint authoring paths (direct JSON
1777            // and `$agent_md` file ref) both land here. Old message
1778            // pointed only at the `.md` frontmatter, which was
1779            // confusing for authors on the JSON-direct path.
1780            return Err(invalid(
1781                "profile.worker_binding is required for this operator backend. \
1782                 Fix by either: \
1783                 (a) if authoring the Blueprint JSON directly, add \
1784                 `agents[N].profile.worker_binding: \"<subagent-type>\"` \
1785                 to the JSON literal; or \
1786                 (b) if using an $agent_md file ref, add \
1787                 `worker_binding: <subagent-type>` to the agent .md frontmatter."
1788                    .into(),
1789            ));
1790        }
1791        Ok(Arc::new(OperatorSpawner::new(
1792            op,
1793            system_prompt,
1794            worker_binding,
1795        )))
1796    }
1797}
1798
1799#[cfg(test)]
1800mod operator_spawner_factory_worker_binding_tests {
1801    use super::*;
1802    use crate::blueprint::AgentProfile;
1803    use crate::core::ctx::Ctx;
1804    use crate::types::CapToken;
1805    use crate::worker::adapter::{WorkerError, WorkerResult};
1806
1807    /// Minimal `Operator` stub whose `requires_worker_binding` is
1808    /// configurable — enough to exercise the compile-time fail-loud gate
1809    /// without standing up a real backend (e.g. `WSOperatorSession`,
1810    /// which lives in a downstream crate).
1811    struct StubOperator {
1812        requires_binding: bool,
1813    }
1814
1815    #[async_trait]
1816    impl Operator for StubOperator {
1817        async fn execute(
1818            &self,
1819            _ctx: &Ctx,
1820            _system: Option<String>,
1821            _prompt: Value,
1822            _worker: Option<WorkerBinding>,
1823            _worker_token: CapToken,
1824        ) -> Result<WorkerResult, WorkerError> {
1825            Ok(WorkerResult {
1826                value: Value::Null,
1827                ok: true,
1828            })
1829        }
1830
1831        fn requires_worker_binding(&self) -> bool {
1832            self.requires_binding
1833        }
1834    }
1835
1836    fn agent_def_with(profile: Option<AgentProfile>) -> AgentDef {
1837        AgentDef {
1838            name: "test-agent".to_string(),
1839            kind: AgentKind::Operator,
1840            spec: serde_json::json!({ "operator_ref": "op1" }),
1841            profile,
1842            meta: None,
1843            runner: None,
1844            runner_ref: None,
1845            verdict: None,
1846        }
1847    }
1848
1849    #[test]
1850    fn build_fails_loud_when_binding_required_but_absent() {
1851        let factory = OperatorSpawnerFactory::new();
1852        factory.register_operator(
1853            "op1",
1854            Arc::new(StubOperator {
1855                requires_binding: true,
1856            }) as Arc<dyn Operator>,
1857        );
1858        let def = agent_def_with(Some(AgentProfile::default()));
1859        match factory.build(&def, None) {
1860            Err(CompileError::InvalidSpec { name, msg }) => {
1861                assert_eq!(name, "test-agent");
1862                assert!(
1863                    msg.contains("worker_binding is required"),
1864                    "unexpected message: {msg}"
1865                );
1866                // Issue #9: the message must be actionable for both
1867                // authoring paths — the JSON-direct hint and the
1868                // $agent_md hint both surface.
1869                assert!(
1870                    msg.contains("agents[N].profile.worker_binding"),
1871                    "message missing JSON-direct hint (issue #9): {msg}"
1872                );
1873                assert!(
1874                    msg.contains("agent .md frontmatter"),
1875                    "message missing $agent_md hint: {msg}"
1876                );
1877            }
1878            Err(other) => panic!("expected InvalidSpec, got: {other:?}"),
1879            Ok(_) => panic!("expected compile-time failure, got Ok"),
1880        }
1881    }
1882
1883    #[test]
1884    fn build_succeeds_when_binding_required_and_present() {
1885        let factory = OperatorSpawnerFactory::new();
1886        factory.register_operator(
1887            "op1",
1888            Arc::new(StubOperator {
1889                requires_binding: true,
1890            }) as Arc<dyn Operator>,
1891        );
1892        let profile = AgentProfile {
1893            worker_binding: Some("mse-worker-coder".to_string()),
1894            tools: vec!["Read".to_string(), "Edit".to_string()],
1895            ..Default::default()
1896        };
1897        let def = agent_def_with(Some(profile));
1898        assert!(
1899            factory.build(&def, None).is_ok(),
1900            "expected Ok when worker_binding is declared"
1901        );
1902    }
1903
1904    #[test]
1905    fn build_succeeds_when_binding_not_required_and_absent() {
1906        let factory = OperatorSpawnerFactory::new();
1907        factory.register_operator(
1908            "op1",
1909            Arc::new(StubOperator {
1910                requires_binding: false,
1911            }) as Arc<dyn Operator>,
1912        );
1913        let def = agent_def_with(Some(AgentProfile::default()));
1914        assert!(
1915            factory.build(&def, None).is_ok(),
1916            "backends that don't require a binding must not be gated by its absence"
1917        );
1918    }
1919}
1920
1921// ─── LuaInProcessSpawnerFactory: inline `spec.source` support ─────────────
1922//
1923// Issue `ab3d1145`: BPs served by `mse serve` couldn't declare `kind: lua`
1924// without pre-registering a `fn_id` on the factory. These tests cover the
1925// new inline path — `spec.source = "<lua chunk>"` (optionally with `label`)
1926// wraps a fresh `LuaScriptSource` at `build` time and runs it through the
1927// same `run_lua_worker` plumbing as the registry path.
1928#[cfg(test)]
1929mod lua_inline_source_tests {
1930    use super::*;
1931    use crate::types::{CapToken, Role, StepId};
1932
1933    fn agent(name: &str, spec: Value) -> AgentDef {
1934        AgentDef {
1935            name: name.to_string(),
1936            kind: AgentKind::Lua,
1937            spec,
1938            profile: None,
1939            meta: None,
1940            runner: None,
1941            runner_ref: None,
1942            verdict: None,
1943        }
1944    }
1945
1946    fn test_invocation(prompt: &str) -> crate::worker::adapter::WorkerInvocation {
1947        crate::worker::adapter::WorkerInvocation {
1948            token: CapToken {
1949                agent_id: "a".into(),
1950                role: Role::Worker,
1951                scopes: vec!["*".into()],
1952                issued_at: 0,
1953                expire_at: u64::MAX / 2,
1954                max_uses: None,
1955                nonce: "test-nonce".into(),
1956                sig_hex: "".into(),
1957            },
1958            task_id: StepId::parse("ST-test").expect("StepId parse"),
1959            attempt: 1,
1960            agent: "g".into(),
1961            prompt: prompt.into(),
1962            sink: None,
1963            cancel_token: None,
1964        }
1965    }
1966
1967    #[test]
1968    fn build_accepts_inline_source_without_pre_registration() {
1969        let factory = LuaInProcessSpawnerFactory::new();
1970        let def = agent(
1971            "g",
1972            serde_json::json!({ "source": "return { value = 42, ok = true }" }),
1973        );
1974        assert!(
1975            factory.build(&def, None).is_ok(),
1976            "inline spec.source must build without a pre-registered fn_id"
1977        );
1978    }
1979
1980    #[test]
1981    fn build_rejects_when_neither_source_nor_fn_id_is_present() {
1982        let factory = LuaInProcessSpawnerFactory::new();
1983        let def = agent("g", serde_json::json!({}));
1984        match factory.build(&def, None) {
1985            Err(CompileError::InvalidSpec { msg, .. }) => {
1986                assert!(
1987                    msg.contains("fn_id"),
1988                    "empty spec must still surface the fn_id-required message: {msg}"
1989                );
1990            }
1991            Err(other) => panic!("expected InvalidSpec, got a different CompileError: {other}"),
1992            // `SpawnerAdapter` is not Debug, so we can't `unwrap_err()` /
1993            // pattern-print the Ok arm — describe the mismatch directly.
1994            Ok(_) => panic!("expected InvalidSpec, got Ok(SpawnerAdapter)"),
1995        }
1996    }
1997
1998    /// The inline path shares `run_lua_worker` with the registry path, so
1999    /// exercising the marshaller once through it is enough to prove the
2000    /// wrap is faithful.
2001    #[tokio::test]
2002    async fn inline_source_evaluates_and_marshals_result() {
2003        let source =
2004            LuaScriptSource::new("return { value = _PROMPT .. '!', ok = true }", "smoke.lua");
2005        let out = run_lua_worker(
2006            std::sync::Arc::new(source),
2007            std::sync::Arc::new(HashMap::new()),
2008            test_invocation("hello"),
2009        )
2010        .await
2011        .expect("lua worker ok");
2012        assert_eq!(out.value, serde_json::json!("hello!"));
2013        assert!(out.ok);
2014    }
2015
2016    #[tokio::test]
2017    async fn inline_source_can_signal_agent_level_failure() {
2018        // Deterministic gate pattern: return `ok = false` to flip the
2019        // dispatch outcome to `Blocked` (the flow.ir Try catch path).
2020        let source = LuaScriptSource::new("return { value = 'nope', ok = false }", "gate.lua");
2021        let out = run_lua_worker(
2022            std::sync::Arc::new(source),
2023            std::sync::Arc::new(HashMap::new()),
2024            test_invocation("input"),
2025        )
2026        .await
2027        .expect("lua worker ok");
2028        assert_eq!(out.value, serde_json::json!("nope"));
2029        assert!(!out.ok);
2030    }
2031}
2032
2033// ─── GH #21 Phase 2: `Blueprint.metas` / `AgentMeta.meta_ref` / static
2034// `$step_meta.ref` compile-time validation ─────────────────────────────────
2035#[cfg(test)]
2036mod meta_ref_validation_tests {
2037    use super::*;
2038    use crate::blueprint::{AgentMeta, MetaDef};
2039    use crate::worker::adapter::WorkerResult;
2040
2041    fn registry_with_echo() -> SpawnerRegistry {
2042        let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2043            Ok(WorkerResult {
2044                value: Value::String(inv.prompt),
2045                ok: true,
2046            })
2047        });
2048        let mut reg = SpawnerRegistry::new();
2049        reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2050        reg
2051    }
2052
2053    fn rustfn_agent(name: &str) -> AgentDef {
2054        AgentDef {
2055            name: name.to_string(),
2056            kind: AgentKind::RustFn,
2057            spec: serde_json::json!({ "fn_id": "echo" }),
2058            profile: None,
2059            meta: None,
2060            runner: None,
2061            runner_ref: None,
2062            verdict: None,
2063        }
2064    }
2065
2066    fn simple_flow(agent_ref: &str, in_: Expr) -> FlowNode {
2067        FlowNode::Step {
2068            ref_: agent_ref.to_string(),
2069            in_,
2070            out: Expr::Path {
2071                at: "$.output".parse().expect("literal test path: $.output"),
2072            },
2073        }
2074    }
2075
2076    fn minimal_bp(agents: Vec<AgentDef>, metas: Vec<MetaDef>, flow: FlowNode) -> Blueprint {
2077        Blueprint {
2078            schema_version: crate::blueprint::current_schema_version(),
2079            id: "meta-ref-ut".into(),
2080            flow,
2081            agents,
2082            operators: vec![],
2083            metas,
2084            hints: Default::default(),
2085            strategy: Default::default(),
2086            metadata: BlueprintMetadata::default(),
2087            spawner_hints: Default::default(),
2088            default_agent_kind: AgentKind::Operator,
2089            default_operator_kind: None,
2090            default_init_ctx: None,
2091            default_agent_ctx: None,
2092            default_context_policy: None,
2093            projection_placement: None,
2094            audits: vec![],
2095            degradation_policy: None,
2096            runners: vec![],
2097            default_runner: None,
2098            check_policy: None,
2099        }
2100    }
2101
2102    #[test]
2103    fn valid_meta_ref_compiles() {
2104        let mut agent = rustfn_agent("worker");
2105        agent.meta = Some(AgentMeta {
2106            meta_ref: Some("shared".to_string()),
2107            ..Default::default()
2108        });
2109        let bp = minimal_bp(
2110            vec![agent],
2111            vec![MetaDef {
2112                name: "shared".into(),
2113                ctx: serde_json::json!({ "k": "v" }),
2114            }],
2115            simple_flow(
2116                "worker",
2117                Expr::Path {
2118                    at: "$.input".parse().expect("literal test path: $.input"),
2119                },
2120            ),
2121        );
2122        let compiler = Compiler::new(registry_with_echo());
2123        assert!(
2124            compiler.compile(&bp).is_ok(),
2125            "a resolvable AgentMeta.meta_ref must compile"
2126        );
2127    }
2128
2129    #[test]
2130    fn unknown_agent_meta_ref_is_unresolved_meta_ref() {
2131        let mut agent = rustfn_agent("worker");
2132        agent.meta = Some(AgentMeta {
2133            meta_ref: Some("missing".to_string()),
2134            ..Default::default()
2135        });
2136        let bp = minimal_bp(
2137            vec![agent],
2138            vec![],
2139            simple_flow(
2140                "worker",
2141                Expr::Path {
2142                    at: "$.input".parse().expect("literal test path: $.input"),
2143                },
2144            ),
2145        );
2146        let compiler = Compiler::new(registry_with_echo());
2147        match compiler.compile(&bp) {
2148            Err(CompileError::UnresolvedMetaRef {
2149                where_,
2150                meta_ref,
2151                defined,
2152            }) => {
2153                assert!(
2154                    where_.contains("worker"),
2155                    "where_ must name the agent: {where_}"
2156                );
2157                assert_eq!(meta_ref, "missing");
2158                assert!(defined.is_empty());
2159            }
2160            Err(other) => {
2161                panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
2162            }
2163            Ok(_) => panic!("expected compile-time failure, got Ok"),
2164        }
2165    }
2166
2167    #[test]
2168    fn unknown_static_step_meta_ref_in_lit_is_unresolved_meta_ref() {
2169        let agent = rustfn_agent("worker");
2170        let in_ = Expr::Lit {
2171            value: serde_json::json!({ "$step_meta": { "ref": "missing" }, "$in": "go" }),
2172        };
2173        let bp = minimal_bp(vec![agent], vec![], simple_flow("worker", in_));
2174        let compiler = Compiler::new(registry_with_echo());
2175        match compiler.compile(&bp) {
2176            Err(CompileError::UnresolvedMetaRef {
2177                where_, meta_ref, ..
2178            }) => {
2179                assert!(
2180                    where_.contains("worker"),
2181                    "where_ must name the offending step: {where_}"
2182                );
2183                assert_eq!(meta_ref, "missing");
2184            }
2185            Err(other) => {
2186                panic!("expected UnresolvedMetaRef, got a different CompileError: {other}")
2187            }
2188            Ok(_) => panic!("expected compile-time failure, got Ok"),
2189        }
2190    }
2191
2192    #[test]
2193    fn path_op_input_with_no_static_envelope_compiles_fine() {
2194        let agent = rustfn_agent("worker");
2195        let bp = minimal_bp(
2196            vec![agent],
2197            vec![],
2198            simple_flow(
2199                "worker",
2200                Expr::Path {
2201                    at: "$.input".parse().expect("literal test path: $.input"),
2202                },
2203            ),
2204        );
2205        let compiler = Compiler::new(registry_with_echo());
2206        assert!(
2207            compiler.compile(&bp).is_ok(),
2208            "a non-Lit Step.in must not trigger the best-effort static $step_meta check"
2209        );
2210    }
2211}
2212
2213// ─── GH #34: `Blueprint.audits[].agent` compile-time validation ────────────
2214#[cfg(test)]
2215mod audit_agent_validation_tests {
2216    use super::*;
2217    use crate::worker::adapter::WorkerResult;
2218    use mlua_swarm_schema::{AuditDef, AuditMode};
2219
2220    fn registry_with_echo() -> SpawnerRegistry {
2221        let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2222            Ok(WorkerResult {
2223                value: Value::String(inv.prompt),
2224                ok: true,
2225            })
2226        });
2227        let mut reg = SpawnerRegistry::new();
2228        reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2229        reg
2230    }
2231
2232    fn rustfn_agent(name: &str) -> AgentDef {
2233        AgentDef {
2234            name: name.to_string(),
2235            kind: AgentKind::RustFn,
2236            spec: serde_json::json!({ "fn_id": "echo" }),
2237            profile: None,
2238            meta: None,
2239            runner: None,
2240            runner_ref: None,
2241            verdict: None,
2242        }
2243    }
2244
2245    fn minimal_bp(agents: Vec<AgentDef>, audits: Vec<AuditDef>) -> Blueprint {
2246        Blueprint {
2247            schema_version: crate::blueprint::current_schema_version(),
2248            id: "audit-ref-ut".into(),
2249            flow: FlowNode::Step {
2250                ref_: "worker".to_string(),
2251                in_: Expr::Path {
2252                    at: "$.input".parse().expect("literal test path: $.input"),
2253                },
2254                out: Expr::Path {
2255                    at: "$.output".parse().expect("literal test path: $.output"),
2256                },
2257            },
2258            agents,
2259            operators: vec![],
2260            metas: vec![],
2261            hints: Default::default(),
2262            strategy: Default::default(),
2263            metadata: BlueprintMetadata::default(),
2264            spawner_hints: Default::default(),
2265            default_agent_kind: AgentKind::Operator,
2266            default_operator_kind: None,
2267            default_init_ctx: None,
2268            default_agent_ctx: None,
2269            default_context_policy: None,
2270            projection_placement: None,
2271            audits,
2272            degradation_policy: None,
2273            runners: vec![],
2274            default_runner: None,
2275            check_policy: None,
2276        }
2277    }
2278
2279    #[test]
2280    fn unresolved_audit_agent_is_a_loud_compile_error() {
2281        let bp = minimal_bp(
2282            vec![rustfn_agent("worker")],
2283            vec![AuditDef {
2284                agent: "missing-auditor".to_string(),
2285                steps: None,
2286                mode: AuditMode::default(),
2287            }],
2288        );
2289        let compiler = Compiler::new(registry_with_echo());
2290        match compiler.compile(&bp) {
2291            Err(CompileError::UnresolvedAuditAgent { agent, defined }) => {
2292                assert_eq!(agent, "missing-auditor");
2293                assert_eq!(defined, vec!["worker".to_string()]);
2294            }
2295            Err(other) => {
2296                panic!("expected UnresolvedAuditAgent, got a different CompileError: {other}")
2297            }
2298            Ok(_) => panic!("expected compile-time failure, got Ok"),
2299        }
2300    }
2301
2302    #[test]
2303    fn resolved_audit_agent_compiles_fine() {
2304        let bp = minimal_bp(
2305            vec![rustfn_agent("worker"), rustfn_agent("auditor")],
2306            vec![AuditDef {
2307                agent: "auditor".to_string(),
2308                steps: None,
2309                mode: AuditMode::default(),
2310            }],
2311        );
2312        let compiler = Compiler::new(registry_with_echo());
2313        assert!(
2314            compiler.compile(&bp).is_ok(),
2315            "an audits[].agent that names a declared AgentDef must compile"
2316        );
2317    }
2318}
2319
2320// ─── GH #27 (follow-up to #23): `Blueprint.projection_placement` compile-time
2321// validation + `CompiledBlueprint.projection_placement` construction ────────
2322#[cfg(test)]
2323mod projection_placement_compile_tests {
2324    use super::*;
2325    use crate::core::projection_placement::{ProjectionPlacement, RootPreference};
2326    use crate::worker::adapter::WorkerResult;
2327    use mlua_swarm_schema::ProjectionPlacementSpec;
2328
2329    fn registry_with_echo() -> SpawnerRegistry {
2330        let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2331            Ok(WorkerResult {
2332                value: Value::String(inv.prompt),
2333                ok: true,
2334            })
2335        });
2336        let mut reg = SpawnerRegistry::new();
2337        reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2338        reg
2339    }
2340
2341    fn minimal_bp(projection_placement: Option<ProjectionPlacementSpec>) -> Blueprint {
2342        Blueprint {
2343            schema_version: crate::blueprint::current_schema_version(),
2344            id: "projection-placement-ut".into(),
2345            flow: FlowNode::Step {
2346                ref_: "worker".to_string(),
2347                in_: Expr::Path {
2348                    at: "$.input".parse().expect("literal test path: $.input"),
2349                },
2350                out: Expr::Path {
2351                    at: "$.output".parse().expect("literal test path: $.output"),
2352                },
2353            },
2354            agents: vec![AgentDef {
2355                name: "worker".to_string(),
2356                kind: AgentKind::RustFn,
2357                spec: serde_json::json!({ "fn_id": "echo" }),
2358                profile: None,
2359                meta: None,
2360                runner: None,
2361                runner_ref: None,
2362                verdict: None,
2363            }],
2364            operators: vec![],
2365            metas: vec![],
2366            hints: Default::default(),
2367            strategy: Default::default(),
2368            metadata: BlueprintMetadata::default(),
2369            spawner_hints: Default::default(),
2370            default_agent_kind: AgentKind::Operator,
2371            default_operator_kind: None,
2372            default_init_ctx: None,
2373            default_agent_ctx: None,
2374            default_context_policy: None,
2375            projection_placement,
2376            audits: vec![],
2377            degradation_policy: None,
2378            runners: vec![],
2379            default_runner: None,
2380            check_policy: None,
2381        }
2382    }
2383
2384    #[test]
2385    fn undeclared_projection_placement_compiles_to_byte_compat_default() {
2386        let bp = minimal_bp(None);
2387        let compiled = Compiler::new(registry_with_echo())
2388            .compile(&bp)
2389            .expect("undeclared projection_placement compiles");
2390        assert_eq!(
2391            *compiled.projection_placement,
2392            ProjectionPlacement::default()
2393        );
2394    }
2395
2396    #[test]
2397    fn declared_valid_projection_placement_compiles_to_matching_resolver() {
2398        let bp = minimal_bp(Some(ProjectionPlacementSpec {
2399            root: Some("project_root".to_string()),
2400            dir_template: Some("custom/{task_id}/out".to_string()),
2401        }));
2402        let compiled = Compiler::new(registry_with_echo())
2403            .compile(&bp)
2404            .expect("valid projection_placement compiles");
2405        assert_eq!(
2406            compiled.projection_placement.root_preference,
2407            RootPreference::ProjectRoot
2408        );
2409        assert_eq!(
2410            compiled.projection_placement.dir_template,
2411            "custom/{task_id}/out"
2412        );
2413    }
2414
2415    #[test]
2416    fn declared_invalid_dir_template_rejects_compile() {
2417        let bp = minimal_bp(Some(ProjectionPlacementSpec {
2418            root: None,
2419            dir_template: Some("workspace/tasks/ctx".to_string()), // missing {task_id}
2420        }));
2421        match Compiler::new(registry_with_echo()).compile(&bp) {
2422            Err(CompileError::InvalidProjectionPlacement(_)) => {}
2423            Err(other) => {
2424                panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
2425            }
2426            Ok(_) => {
2427                panic!("expected compile-time rejection for a missing {{task_id}} placeholder")
2428            }
2429        }
2430    }
2431
2432    #[test]
2433    fn declared_invalid_root_literal_rejects_compile() {
2434        let bp = minimal_bp(Some(ProjectionPlacementSpec {
2435            root: Some("nope".to_string()),
2436            dir_template: None,
2437        }));
2438        match Compiler::new(registry_with_echo()).compile(&bp) {
2439            Err(CompileError::InvalidProjectionPlacement(_)) => {}
2440            Err(other) => {
2441                panic!("expected InvalidProjectionPlacement, got a different CompileError: {other}")
2442            }
2443            Ok(_) => panic!("expected compile-time rejection for an invalid root literal"),
2444        }
2445    }
2446}
2447
2448// ─── GH #50: `Blueprint.agents[].verdict` cond↔output-shape lint ──────────
2449#[cfg(test)]
2450mod verdict_contract_lint_tests {
2451    use super::*;
2452    use crate::worker::adapter::WorkerResult;
2453
2454    fn registry_with_echo() -> SpawnerRegistry {
2455        let factory = RustFnInProcessSpawnerFactory::new().register_fn("echo", |inv| async move {
2456            Ok(WorkerResult {
2457                value: Value::String(inv.prompt),
2458                ok: true,
2459            })
2460        });
2461        let mut reg = SpawnerRegistry::new();
2462        reg.register::<RustFnInProcessSpawnerFactory>(Arc::new(factory));
2463        reg
2464    }
2465
2466    fn gate_agent(verdict: Option<VerdictContract>) -> AgentDef {
2467        AgentDef {
2468            name: "gate".to_string(),
2469            kind: AgentKind::RustFn,
2470            spec: serde_json::json!({ "fn_id": "echo" }),
2471            profile: None,
2472            meta: None,
2473            runner: None,
2474            runner_ref: None,
2475            verdict,
2476        }
2477    }
2478
2479    fn minimal_bp(agent: AgentDef, flow: FlowNode) -> Blueprint {
2480        Blueprint {
2481            schema_version: crate::blueprint::current_schema_version(),
2482            id: "verdict-contract-ut".into(),
2483            flow,
2484            agents: vec![agent],
2485            operators: vec![],
2486            metas: vec![],
2487            hints: Default::default(),
2488            strategy: Default::default(),
2489            metadata: BlueprintMetadata::default(),
2490            spawner_hints: Default::default(),
2491            default_agent_kind: AgentKind::Operator,
2492            default_operator_kind: None,
2493            default_init_ctx: None,
2494            default_agent_ctx: None,
2495            default_context_policy: None,
2496            projection_placement: None,
2497            audits: vec![],
2498            degradation_policy: None,
2499            runners: vec![],
2500            default_runner: None,
2501            check_policy: None,
2502        }
2503    }
2504
2505    fn step(ref_: &str, out_path: &str) -> FlowNode {
2506        FlowNode::Step {
2507            ref_: ref_.to_string(),
2508            in_: Expr::Lit { value: Value::Null },
2509            out: Expr::Path {
2510                at: out_path.parse().expect("literal test path"),
2511            },
2512        }
2513    }
2514
2515    fn noop() -> FlowNode {
2516        FlowNode::Seq { children: vec![] }
2517    }
2518
2519    fn eq_cond(path: &str, lit: &str) -> Expr {
2520        Expr::Eq {
2521            lhs: Box::new(Expr::Path {
2522                at: path.parse().expect("literal test path"),
2523            }),
2524            rhs: Box::new(Expr::Lit {
2525                value: Value::String(lit.to_string()),
2526            }),
2527        }
2528    }
2529
2530    fn branch(cond: Expr, then_: FlowNode, else_: FlowNode) -> FlowNode {
2531        FlowNode::Branch {
2532            cond,
2533            then_: Box::new(then_),
2534            else_: Box::new(else_),
2535        }
2536    }
2537
2538    fn body_contract(values: &[&str]) -> VerdictContract {
2539        VerdictContract {
2540            channel: VerdictChannel::Body,
2541            values: values.iter().map(|v| v.to_string()).collect(),
2542        }
2543    }
2544
2545    fn part_contract(values: &[&str]) -> VerdictContract {
2546        VerdictContract {
2547            channel: VerdictChannel::Part,
2548            values: values.iter().map(|v| v.to_string()).collect(),
2549        }
2550    }
2551
2552    #[test]
2553    fn contract_with_correct_body_channel_and_value_compiles() {
2554        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2555        let flow = FlowNode::Seq {
2556            children: vec![
2557                step("gate", "$.verdict"),
2558                branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
2559            ],
2560        };
2561        let bp = minimal_bp(agent, flow);
2562        assert!(
2563            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2564            "a cond addressing the bare step output must match a channel: \"body\" contract"
2565        );
2566    }
2567
2568    #[test]
2569    fn contract_with_correct_part_channel_and_value_compiles() {
2570        let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
2571        let flow = FlowNode::Seq {
2572            children: vec![
2573                step("gate", "$.gate"),
2574                branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
2575            ],
2576        };
2577        let bp = minimal_bp(agent, flow);
2578        assert!(
2579            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2580            "a cond addressing '<step>.parts.verdict' must match a channel: \"part\" contract"
2581        );
2582    }
2583
2584    #[test]
2585    fn body_channel_contract_rejects_cond_addressing_parts_verdict() {
2586        // Pattern A declared (channel: "body") but the cond addresses the
2587        // Pattern B shape ('$.gate.parts.verdict') instead of the bare
2588        // step output — GH #50 register-time enforcement point 1.
2589        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2590        let flow = FlowNode::Seq {
2591            children: vec![
2592                step("gate", "$.gate"),
2593                branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
2594            ],
2595        };
2596        let bp = minimal_bp(agent, flow);
2597        match Compiler::new(registry_with_echo()).compile(&bp) {
2598            Err(CompileError::VerdictChannelMismatch {
2599                where_,
2600                agent,
2601                expected_channel,
2602                actual_shape,
2603            }) => {
2604                assert_eq!(agent, "gate");
2605                assert_eq!(expected_channel, "body");
2606                assert_eq!(actual_shape, "part");
2607                assert!(where_.contains("Branch cond"), "where_: {where_}");
2608            }
2609            Err(other) => {
2610                panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
2611            }
2612            Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
2613        }
2614    }
2615
2616    #[test]
2617    fn part_channel_contract_rejects_cond_addressing_bare_output() {
2618        // Inverse of the previous case: channel: "part" declared, but the
2619        // cond addresses the bare step output.
2620        let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
2621        let flow = FlowNode::Seq {
2622            children: vec![
2623                step("gate", "$.verdict"),
2624                branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
2625            ],
2626        };
2627        let bp = minimal_bp(agent, flow);
2628        match Compiler::new(registry_with_echo()).compile(&bp) {
2629            Err(CompileError::VerdictChannelMismatch {
2630                agent,
2631                expected_channel,
2632                actual_shape,
2633                ..
2634            }) => {
2635                assert_eq!(agent, "gate");
2636                assert_eq!(expected_channel, "part");
2637                assert_eq!(actual_shape, "body");
2638            }
2639            Err(other) => {
2640                panic!("expected VerdictChannelMismatch, got a different CompileError: {other}")
2641            }
2642            Ok(_) => panic!("expected compile-time rejection for the wrong channel shape"),
2643        }
2644    }
2645
2646    #[test]
2647    fn contract_rejects_lit_outside_declared_values() {
2648        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2649        let flow = FlowNode::Seq {
2650            children: vec![
2651                step("gate", "$.verdict"),
2652                branch(eq_cond("$.verdict", "UNKNOWN"), noop(), noop()),
2653            ],
2654        };
2655        let bp = minimal_bp(agent, flow);
2656        match Compiler::new(registry_with_echo()).compile(&bp) {
2657            Err(CompileError::VerdictValueNotInContract {
2658                agent,
2659                value,
2660                values,
2661                ..
2662            }) => {
2663                assert_eq!(agent, "gate");
2664                assert_eq!(value, "UNKNOWN");
2665                assert_eq!(values, vec!["PASS".to_string(), "BLOCKED".to_string()]);
2666            }
2667            Err(other) => {
2668                panic!("expected VerdictValueNotInContract, got a different CompileError: {other}")
2669            }
2670            Ok(_) => panic!("expected compile-time rejection for a Lit outside declared values"),
2671        }
2672    }
2673
2674    #[test]
2675    fn undeclared_agent_referenced_by_cond_compiles_with_warning_only() {
2676        let agent = gate_agent(None);
2677        let flow = FlowNode::Seq {
2678            children: vec![
2679                step("gate", "$.verdict"),
2680                branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
2681            ],
2682        };
2683        let bp = minimal_bp(agent, flow);
2684        assert!(
2685            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2686            "an undeclared verdict contract must never reject compile (opt-in, back-compat)"
2687        );
2688    }
2689
2690    #[test]
2691    fn in_expr_with_lit_haystack_members_compiles() {
2692        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2693        let cond = Expr::In {
2694            needle: Box::new(Expr::Path {
2695                at: "$.verdict".parse().expect("literal test path"),
2696            }),
2697            haystack: Box::new(Expr::Lit {
2698                value: serde_json::json!(["PASS", "BLOCKED"]),
2699            }),
2700        };
2701        let flow = FlowNode::Seq {
2702            children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
2703        };
2704        let bp = minimal_bp(agent, flow);
2705        assert!(
2706            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2707            "an `In` haystack whose every Lit is a declared value must compile"
2708        );
2709    }
2710
2711    /// GH #50 follow-up (issue `33bc825b`): opt-in strict mode rejects a
2712    /// Blueprint whose declared `verdict.values` set includes at least one
2713    /// entry that no downstream `Branch`/`Loop` `cond` references. The
2714    /// contract declares `["PASS", "BLOCKED"]` but only "BLOCKED" is
2715    /// referenced by the cond → "PASS" is unhandled → `CompileError::
2716    /// VerdictValueUnhandled` under `strict_verdict_handling: Some(true)`.
2717    #[test]
2718    fn strict_mode_rejects_unhandled_declared_value() {
2719        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2720        let flow = FlowNode::Seq {
2721            children: vec![
2722                step("gate", "$.verdict"),
2723                branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
2724            ],
2725        };
2726        let mut bp = minimal_bp(agent, flow);
2727        bp.metadata.strict_verdict_handling = Some(true);
2728        match Compiler::new(registry_with_echo()).compile(&bp) {
2729            Err(CompileError::VerdictValueUnhandled {
2730                agent,
2731                value,
2732                declared_values,
2733                step_ref,
2734            }) => {
2735                assert_eq!(agent, "gate");
2736                assert_eq!(value, "PASS");
2737                assert_eq!(
2738                    declared_values,
2739                    vec!["PASS".to_string(), "BLOCKED".to_string()]
2740                );
2741                assert_eq!(step_ref, "gate");
2742            }
2743            Err(other) => {
2744                panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
2745            }
2746            Ok(_) => panic!(
2747                "expected compile-time rejection for a declared verdict value with no \
2748                 downstream handler under strict_verdict_handling=Some(true)"
2749            ),
2750        }
2751    }
2752
2753    /// GH #50 follow-up (issue `33bc825b`): default mode (i.e.
2754    /// `strict_verdict_handling` absent or `Some(false)`) surfaces
2755    /// unhandled declared values via `tracing::warn!` only — the compile
2756    /// still succeeds. This preserves back-compat with GH #50's original
2757    /// test cases (many of which declare `values = ["PASS", "BLOCKED"]`
2758    /// and cond-reference only one).
2759    #[test]
2760    fn default_mode_permits_unhandled_declared_value() {
2761        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2762        let flow = FlowNode::Seq {
2763            children: vec![
2764                step("gate", "$.verdict"),
2765                branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
2766            ],
2767        };
2768        let bp = minimal_bp(agent, flow);
2769        // `strict_verdict_handling` left as `None` (default)
2770        assert!(
2771            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2772            "default mode must never reject a Blueprint for unhandled declared values \
2773             (opt-in, back-compat with GH #50)"
2774        );
2775    }
2776
2777    /// GH #50 follow-up (issue `33bc825b`): under strict mode, when every
2778    /// declared value is referenced by at least one downstream cond, the
2779    /// compile succeeds. This tests the positive path of the reverse-
2780    /// direction lint.
2781    #[test]
2782    fn strict_mode_accepts_all_declared_values_handled() {
2783        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2784        // Two branches, each cond referencing one declared value —
2785        // together they cover the full `values` set.
2786        let flow = FlowNode::Seq {
2787            children: vec![
2788                step("gate", "$.verdict"),
2789                branch(eq_cond("$.verdict", "BLOCKED"), noop(), noop()),
2790                branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
2791            ],
2792        };
2793        let mut bp = minimal_bp(agent, flow);
2794        bp.metadata.strict_verdict_handling = Some(true);
2795        assert!(
2796            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2797            "strict mode must accept a Blueprint that handles every declared value"
2798        );
2799    }
2800
2801    /// GH #50 follow-up (issue `33bc825b`): under strict mode, an `In`
2802    /// cond whose `Lit` haystack lists every declared value satisfies
2803    /// the handler-coverage check in one go.
2804    #[test]
2805    fn strict_mode_accepts_declared_values_covered_by_in_expr() {
2806        let agent = gate_agent(Some(body_contract(&["PASS", "BLOCKED"])));
2807        let cond = Expr::In {
2808            needle: Box::new(Expr::Path {
2809                at: "$.verdict".parse().expect("literal test path"),
2810            }),
2811            haystack: Box::new(Expr::Lit {
2812                value: serde_json::json!(["PASS", "BLOCKED"]),
2813            }),
2814        };
2815        let flow = FlowNode::Seq {
2816            children: vec![step("gate", "$.verdict"), branch(cond, noop(), noop())],
2817        };
2818        let mut bp = minimal_bp(agent, flow);
2819        bp.metadata.strict_verdict_handling = Some(true);
2820        assert!(
2821            Compiler::new(registry_with_echo()).compile(&bp).is_ok(),
2822            "strict mode must accept an `In` haystack that covers every declared value"
2823        );
2824    }
2825
2826    /// GH #50 follow-up (issue `33bc825b`): under strict mode, a `part`
2827    /// channel contract with unhandled declared value is rejected the same
2828    /// way as the `body` channel case. Confirms channel-agnostic coverage.
2829    #[test]
2830    fn strict_mode_rejects_unhandled_part_channel_value() {
2831        let agent = gate_agent(Some(part_contract(&["PASS", "BLOCKED"])));
2832        let flow = FlowNode::Seq {
2833            children: vec![
2834                step("gate", "$.gate"),
2835                branch(eq_cond("$.gate.parts.verdict", "BLOCKED"), noop(), noop()),
2836            ],
2837        };
2838        let mut bp = minimal_bp(agent, flow);
2839        bp.metadata.strict_verdict_handling = Some(true);
2840        match Compiler::new(registry_with_echo()).compile(&bp) {
2841            Err(CompileError::VerdictValueUnhandled {
2842                agent,
2843                value,
2844                step_ref,
2845                ..
2846            }) => {
2847                assert_eq!(agent, "gate");
2848                assert_eq!(value, "PASS");
2849                assert_eq!(step_ref, "gate");
2850            }
2851            Err(other) => {
2852                panic!("expected VerdictValueUnhandled, got a different CompileError: {other}")
2853            }
2854            Ok(_) => panic!(
2855                "expected compile-time rejection for a declared verdict value with no \
2856                 downstream handler (part channel) under strict_verdict_handling=Some(true)"
2857            ),
2858        }
2859    }
2860
2861    /// Acceptance criterion #7 (5th case): a Blueprint shaped like the
2862    /// existing `02-verdict-loop.json` sample — a `Loop` retrying while
2863    /// `$.verdict == "BLOCKED"` plus a `Branch` on `$.verdict == "PASS"` —
2864    /// but with `verdict` omitted on every agent must compile unchanged
2865    /// (at most `tracing::warn!`) and leave `CompiledAgentTable.
2866    /// verdict_contracts` empty.
2867    #[test]
2868    fn verdict_omitted_blueprint_compiles_unchanged_with_empty_contracts() {
2869        let agent = gate_agent(None);
2870        let flow = FlowNode::Seq {
2871            children: vec![
2872                step("gate", "$.verdict"),
2873                FlowNode::Loop {
2874                    counter: Expr::Path {
2875                        at: "$.n".parse().expect("literal test path"),
2876                    },
2877                    cond: eq_cond("$.verdict", "BLOCKED"),
2878                    body: Box::new(step("gate", "$.verdict")),
2879                    max: 3,
2880                },
2881                branch(eq_cond("$.verdict", "PASS"), noop(), noop()),
2882            ],
2883        };
2884        let bp = minimal_bp(agent, flow);
2885        let compiled = Compiler::new(registry_with_echo())
2886            .compile(&bp)
2887            .expect("a verdict-omitted Blueprint must compile unchanged");
2888        assert!(
2889            compiled.router.verdict_contracts.is_empty(),
2890            "no agent declared a verdict contract"
2891        );
2892    }
2893}