Skip to main content

greentic_runner_host/runner/
agent_node.rs

1use anyhow::Result;
2use serde_json::Value;
3
4/// Bridges a `DwAgent` flow node into the agentic-worker runtime.
5///
6/// The concrete impl (constructed in the runner binary, Task 4.3) wraps
7/// `greentic_aw_runtime::AgentRuntime`. The engine holds it as a trait
8/// object so `engine.rs` stays free of AW-runtime construction details.
9#[async_trait::async_trait]
10pub trait AgentNodeHandler: Send + Sync {
11    /// Execute one agentic step. `flow_input` is the upstream node's
12    /// JSON payload (expects at least `{"user_text": "..."}`); returns
13    /// the node output JSON (`{"reply", "trail", "terminated_by"}`).
14    async fn execute(
15        &self,
16        tenant_id: &str,
17        env_id: &str,
18        agent_id: &str,
19        session_id: &str,
20        flow_input: &Value,
21    ) -> Result<Value>;
22}
23
24// ---------------------------------------------------------------------------
25// agentic-worker feature: full DwAgent / AgentRuntime integration
26// ---------------------------------------------------------------------------
27
28#[cfg(feature = "agentic-worker")]
29mod aw {
30    use std::collections::HashMap;
31    use std::future::Future;
32    use std::path::PathBuf;
33    use std::pin::Pin;
34    use std::str::FromStr;
35    use std::sync::Arc;
36
37    use anyhow::Result;
38    use greentic_aw_runtime::config::AgentConfig;
39    use greentic_aw_runtime::config_provider::ConfigProvider;
40    use greentic_aw_runtime::error::{AgentError, ConfigError};
41    use greentic_aw_runtime::guardrail::GuardrailDirection;
42    use greentic_aw_runtime::{AgentInput, AgentRuntime, AgentStep, StepObserver, TenantContext};
43    use serde_json::{Value, json};
44
45    use crate::trace::agent_audit::AgentAuditObserver;
46    use crate::trace::audit_sink::AuditSink;
47
48    use super::AgentNodeHandler;
49
50    // -----------------------------------------------------------------------
51    // Pack-manifest agent helpers
52    // -----------------------------------------------------------------------
53
54    /// Deserialize raw agent blobs from a pack manifest into typed
55    /// [`AgentConfig`] structs.
56    ///
57    /// Malformed blobs are skipped with a [`tracing::warn!`] so a single
58    /// bad entry never prevents other agents (or pack-level operators) from
59    /// loading. The `pack_id` argument is used only in log messages.
60    pub fn agent_configs_from_manifest(
61        pack_id: &str,
62        blobs: &std::collections::BTreeMap<String, Value>,
63    ) -> HashMap<String, AgentConfig> {
64        blobs
65            .iter()
66            .filter_map(|(agent_id, blob)| {
67                match serde_json::from_value::<AgentConfig>(blob.clone()) {
68                    Ok(config) => Some((agent_id.clone(), config)),
69                    Err(deserialize_error) => {
70                        tracing::warn!(
71                            pack_id,
72                            agent_id,
73                            error = %deserialize_error,
74                            "skipping malformed agent blob in pack manifest"
75                        );
76                        None
77                    }
78                }
79            })
80            .collect()
81    }
82
83    /// Merge pack-provided agent configs with operator-declared ones.
84    ///
85    /// Pack agents form the base layer; operator entries take precedence on
86    /// `agent_id` collision (operator always wins). This ensures operators can
87    /// override or refine any pack-embedded agent without touching the pack
88    /// itself.
89    pub fn merge_agent_sources(
90        pack_agents: HashMap<String, AgentConfig>,
91        operator_agents: HashMap<String, AgentConfig>,
92    ) -> HashMap<String, AgentConfig> {
93        let mut merged = pack_agents;
94        for (agent_id, operator_config) in operator_agents {
95            merged.insert(agent_id, operator_config);
96        }
97        merged
98    }
99
100    /// Fill `blobs` from a `dw-agents.json` `sidecar` map, inserting each entry only
101    /// when its agent_id is absent — so `manifest.agents` stays authoritative and the
102    /// sidecar only bridges packs whose manifest could not carry agents.
103    pub fn merge_sidecar_into(
104        blobs: &mut std::collections::BTreeMap<String, serde_json::Value>,
105        sidecar: std::collections::BTreeMap<String, serde_json::Value>,
106    ) {
107        for (agent_id, blob) in sidecar {
108            blobs.entry(agent_id).or_insert(blob);
109        }
110    }
111
112    /// Fixed, user-safe reply returned when an agentic step fails. The detailed
113    /// [`greentic_aw_runtime::AgentError`] is logged but never surfaced to the
114    /// flow output, so internal failure modes do not leak to end users.
115    const SANITISED_ERROR_REPLY: &str = "Something went wrong. Please try again.";
116
117    /// Build the structured JSON output emitted by a `DwAgent` node when a
118    /// guardrail blocks the step.
119    ///
120    /// The returned value gives the downstream flow a machine-readable signal
121    /// it can branch on (`guardrail.blocked == true`, `terminated_by ==
122    /// "guardrail_denied"`) without leaking raw internal error details.
123    ///
124    /// - `direction` — inbound (user→agent) or outbound (agent→user)
125    /// - `code` — guardrail-defined denial code (e.g. `"permission_denied"`)
126    /// - `message` — human-readable denial reason surfaced to the user
127    /// - `details` — optional JSON string with additional context; parsed into a
128    ///   structured value when present, omitted (null) otherwise
129    pub(super) fn guardrail_denied_json(
130        direction: GuardrailDirection,
131        code: &str,
132        message: &str,
133        details: Option<&str>,
134    ) -> Value {
135        let details_val =
136            details.and_then(|detail_str| serde_json::from_str::<Value>(detail_str).ok());
137        json!({
138            "guardrail": {
139                "blocked": true,
140                "direction": direction.as_str(),
141                "code": code,
142                "message": message,
143                "details": details_val,
144            },
145            "reply": message,
146            "trail": Vec::<AgentStep>::new(),
147            "terminated_by": "guardrail_denied",
148        })
149    }
150
151    /// Build an [`HttpGuardrailPolicy`] from `GREENTIC_AW_ADMIN_ENDPOINT` +
152    /// `GREENTIC_AW_ADMIN_TOKEN` (the same pair the agent registry uses).
153    /// Returns `None` when either is unset/empty, so a non-admin deploy
154    /// enforces no mandatory policy (today's behavior).
155    fn guardrail_policy_from_env()
156    -> Option<greentic_aw_runtime::guardrail_provider::HttpGuardrailPolicy> {
157        let endpoint = std::env::var("GREENTIC_AW_ADMIN_ENDPOINT")
158            .ok()
159            .filter(|s| !s.is_empty())?;
160        let token = std::env::var("GREENTIC_AW_ADMIN_TOKEN")
161            .ok()
162            .filter(|s| !s.is_empty())?;
163        Some(greentic_aw_runtime::guardrail_provider::HttpGuardrailPolicy::new(endpoint, token))
164    }
165
166    /// Production [`AgentNodeHandler`] wrapping the agentic-worker runtime.
167    ///
168    /// Holds a shared [`AgentRuntime`] and translates a `DwAgent` flow node's
169    /// JSON payload into an [`AgentInput`], invoking one Plan-Act-Observe step
170    /// per call. Construction (Task 4.3b) lives in the runner binary; the engine
171    /// only ever sees this through the [`AgentNodeHandler`] trait object.
172    pub struct RuntimeAgentNodeHandler {
173        runtime: Arc<AgentRuntime>,
174        /// Best-effort agent-step audit sink (EPIC-B B-3). `None` — the
175        /// default when no NATS audit client is configured
176        /// (`GREENTIC_EVENTS_NATS_URL` unset/unreachable) — keeps `execute`
177        /// on the plain [`AgentRuntime::step`] path, byte-identical to the
178        /// behaviour before this observer existed.
179        audit_sink: Option<AuditSink>,
180        /// Identity of the deployed unit these agents were loaded from — the
181        /// revision's `bundle_id`, stamped onto every step's
182        /// [`TenantContext`] so billing can attribute spend to a project.
183        ///
184        /// `None` on the tenant-only (legacy, non-revision) pack path, where no
185        /// bundle is pinned. Billing then OMITS the `project_id` dimension
186        /// rather than substituting the agent id, which is not unique across
187        /// packs — see [`greentic_aw_runtime::TenantContext::project_id`].
188        project_id: Option<String>,
189    }
190
191    impl RuntimeAgentNodeHandler {
192        /// Wrap a shared [`AgentRuntime`] in a flow-node handler. `audit_sink`
193        /// is `Some` only when a NATS audit client was configured; `execute`
194        /// then drives the step via [`AgentRuntime::step_with_observer`] with
195        /// an [`AgentAuditObserver`] so tool calls/results are published to
196        /// `audit.<tenant>.agent.<event>`. When `None`, `execute` uses
197        /// [`AgentRuntime::step`] directly (no observer, no behaviour change).
198        ///
199        /// `project_id` is the deployed unit's `bundle_id` (`None` on the
200        /// legacy tenant-only path); it becomes the `project_id` billing
201        /// dimension of every step this handler runs.
202        pub fn new(
203            runtime: Arc<AgentRuntime>,
204            audit_sink: Option<AuditSink>,
205            project_id: Option<String>,
206        ) -> Self {
207            Self {
208                runtime,
209                audit_sink,
210                project_id,
211            }
212        }
213    }
214
215    /// Build a [`greentic_types::TenantCtx`] for the agent-audit observer from
216    /// the flow node's plain `tenant_id`/`env_id` strings. Mirrors
217    /// `HostConfig::tenant_ctx`'s fallback-to-"local" pattern: an id that fails
218    /// the newtype's validation (which should not happen once a flow has been
219    /// routed to a tenant) still yields a well-formed `TenantCtx` rather than
220    /// panicking — the audit event is best-effort, never load-bearing.
221    fn tenant_ctx_for_audit(tenant_id: &str, env_id: &str) -> greentic_types::TenantCtx {
222        let env = greentic_types::EnvId::from_str(env_id)
223            .unwrap_or_else(|_| greentic_types::EnvId::new("local").expect("local env id"));
224        let tenant = greentic_types::TenantId::from_str(tenant_id)
225            .unwrap_or_else(|_| greentic_types::TenantId::new("local").expect("local tenant id"));
226        greentic_types::TenantCtx::new(env, tenant)
227    }
228
229    #[async_trait::async_trait]
230    impl AgentNodeHandler for RuntimeAgentNodeHandler {
231        async fn execute(
232            &self,
233            tenant_id: &str,
234            env_id: &str,
235            agent_id: &str,
236            session_id: &str,
237            flow_input: &Value,
238        ) -> Result<Value> {
239            let user_text = flow_input
240                .get("user_text")
241                .and_then(Value::as_str)
242                .unwrap_or("")
243                .to_string();
244            let tenant =
245                TenantContext::new(tenant_id, env_id).with_project_id(self.project_id.clone());
246            let input = AgentInput { text: user_text };
247
248            // Off by default: with no audit sink configured, this is exactly
249            // the pre-existing `self.runtime.step(...)` call — no observer is
250            // constructed and behaviour is byte-identical to before EPIC-B B-3.
251            let step_result = match &self.audit_sink {
252                Some(sink) => {
253                    let observer: Arc<dyn StepObserver> = Arc::new(AgentAuditObserver::new(
254                        sink.clone(),
255                        tenant_ctx_for_audit(tenant_id, env_id),
256                        agent_id.to_string(),
257                        session_id.to_string(),
258                    ));
259                    self.runtime
260                        .step_with_observer(tenant, session_id, agent_id, input, observer)
261                        .await
262                }
263                None => self.runtime.step(tenant, session_id, agent_id, input).await,
264            };
265
266            match step_result {
267                Ok(output) => Ok(json!({
268                    "reply": output.reply,
269                    "trail": output.trail,
270                    "terminated_by": output.terminated_by,
271                })),
272                Err(AgentError::GuardrailDenied {
273                    direction,
274                    code,
275                    message,
276                    details,
277                }) => {
278                    // Surface guardrail denials as a structured, machine-readable
279                    // node output so flows can branch on `guardrail.blocked`.
280                    // The denial code and message are intentionally visible to
281                    // the flow (they are governance messages, not internal detail).
282                    tracing::info!(
283                        agent_id,
284                        session_id,
285                        %code,
286                        "DwAgent blocked by guardrail"
287                    );
288                    Ok(guardrail_denied_json(
289                        direction,
290                        &code,
291                        &message,
292                        details.as_deref(),
293                    ))
294                }
295                Err(error) => {
296                    // Never leak the internal AgentError to the flow output. Log
297                    // the detail for operators; return a sanitised reply only.
298                    tracing::warn!(error = %error, agent_id, session_id, "DwAgent step failed");
299                    Ok(json!({
300                        "reply": SANITISED_ERROR_REPLY,
301                        "trail": Vec::<AgentStep>::new(),
302                        "terminated_by": "error",
303                    }))
304                }
305            }
306        }
307    }
308
309    /// [`ConfigProvider`] backed by the operator's [`HostConfig::agents`] map.
310    ///
311    /// Agents are operator-global for the MVP: lookup is keyed purely by
312    /// `agent_id`; the `tenant`/`env` arguments are accepted (to satisfy the
313    /// trait contract) but not used for keying. This avoids a tenant/env
314    /// key-matching footgun against the dispatch path, which derives those
315    /// values independently. A future per-tenant config source can replace
316    /// this implementation without touching callers.
317    pub struct HostConfigProvider {
318        agents: HashMap<String, AgentConfig>,
319    }
320
321    impl HostConfigProvider {
322        /// Wrap the operator-declared agents map in a [`ConfigProvider`].
323        pub fn new(agents: HashMap<String, AgentConfig>) -> Self {
324            Self { agents }
325        }
326    }
327
328    impl ConfigProvider for HostConfigProvider {
329        fn agent_config<'a>(
330            &'a self,
331            _tenant: &'a TenantContext,
332            agent_id: &'a str,
333        ) -> Pin<Box<dyn Future<Output = Result<AgentConfig, ConfigError>> + Send + 'a>> {
334            let found = self.agents.get(agent_id).cloned();
335            let agent_id_owned = agent_id.to_string();
336            Box::pin(async move { found.ok_or(ConfigError::AgentNotFound(agent_id_owned)) })
337        }
338    }
339
340    /// Resolve the extension discovery directory for tool dispatch.
341    ///
342    /// Honours `GREENTIC_EXTENSIONS_DIR`; otherwise falls back to
343    /// `~/.greentic/extensions` (the platform convention), and finally to a
344    /// temp-dir path when no home directory can be resolved. A missing or
345    /// empty directory is harmless — `list_tools`/`invoke_tool` simply return
346    /// empty/NotFound, which is correct for tool-less agents.
347    fn extension_discovery_dir() -> PathBuf {
348        if let Ok(dir) = std::env::var("GREENTIC_EXTENSIONS_DIR")
349            && !dir.is_empty()
350        {
351            return PathBuf::from(dir);
352        }
353        if let Some(home) = std::env::var_os("HOME") {
354            return PathBuf::from(home).join(".greentic").join("extensions");
355        }
356        std::env::temp_dir().join("greentic").join("extensions")
357    }
358
359    /// Resolve the directory scanned for `<agent_id>.json` Digital Worker manifests.
360    ///
361    /// Honours `GREENTIC_AGENT_MANIFESTS_DIR`; otherwise `~/.greentic/agents`, and
362    /// finally a temp-dir path when no home is resolvable (keeps the fn total). A
363    /// missing dir is harmless — the overlay provider simply finds no manifest and
364    /// returns the YAML base unchanged.
365    fn manifests_discovery_dir() -> PathBuf {
366        if let Ok(dir) = std::env::var("GREENTIC_AGENT_MANIFESTS_DIR")
367            && !dir.is_empty()
368        {
369            return PathBuf::from(dir);
370        }
371        if let Some(home) = std::env::var_os("HOME") {
372            return PathBuf::from(home).join(".greentic").join("agents");
373        }
374        std::env::temp_dir().join("greentic").join("agents")
375    }
376
377    /// Build an [`HttpConfigProvider`] from `GREENTIC_AW_ADMIN_ENDPOINT` +
378    /// `GREENTIC_AW_ADMIN_TOKEN`. Returns `None` when either is unset/empty, so
379    /// the runtime keeps using the local overlay alone.
380    fn registry_from_env() -> Option<greentic_aw_runtime::HttpConfigProvider> {
381        let endpoint = std::env::var("GREENTIC_AW_ADMIN_ENDPOINT")
382            .ok()
383            .filter(|s| !s.is_empty())?;
384        let token = std::env::var("GREENTIC_AW_ADMIN_TOKEN")
385            .ok()
386            .filter(|s| !s.is_empty())?;
387        Some(greentic_aw_runtime::HttpConfigProvider::new(
388            endpoint, token,
389        ))
390    }
391
392    /// Build an [`McpToolSource`] from the same admin endpoint/token the agent
393    /// registry uses (`GREENTIC_AW_ADMIN_ENDPOINT` + `GREENTIC_AW_ADMIN_TOKEN`).
394    ///
395    /// MCP tools are ON by default whenever the admin credentials are present:
396    /// exposure is already authorized twice upstream (the tenant registers the
397    /// server with the `agentic_worker` role in admin, and the agent's
398    /// allowlist must explicitly reference `mcp:<server_id>`), so a configured
399    /// runner participates without extra ceremony. `GREENTIC_AW_MCP=0` is the
400    /// operator opt-out escape hatch for environments where outbound calls to
401    /// tenant-registered MCP servers must stay disabled. Returns `None` on
402    /// opt-out or when either credential is missing/empty.
403    ///
404    /// [`McpToolSource`]: greentic_aw_runtime::McpToolSource
405    fn mcp_source_from_env() -> Option<Arc<greentic_aw_runtime::McpToolSource>> {
406        if std::env::var("GREENTIC_AW_MCP").ok().as_deref() == Some("0") {
407            tracing::info!("GREENTIC_AW_MCP=0; MCP tool source disabled");
408            return None;
409        }
410        let endpoint = std::env::var("GREENTIC_AW_ADMIN_ENDPOINT")
411            .ok()
412            .filter(|s| !s.is_empty())?;
413        let token = std::env::var("GREENTIC_AW_ADMIN_TOKEN")
414            .ok()
415            .filter(|s| !s.is_empty())?;
416        tracing::info!(endpoint = %endpoint, "MCP tool source constructed");
417        Some(Arc::new(greentic_aw_runtime::McpToolSource::new(
418            endpoint, token,
419        )))
420    }
421
422    /// Build the component tool source from the operator's loaded packs, gated
423    /// by `GREENTIC_AW_COMPONENT_TOOLS` (set to "0" to disable). Returns `None`
424    /// when disabled or when no packs are loaded, so `component:` tool refs then
425    /// resolve to nothing. Mirrors [`mcp_source_from_env`] but discovers tools
426    /// from in-pack components rather than a remote admin.
427    fn component_source_from_packs(
428        packs: &[Arc<crate::pack::PackRuntime>],
429        tenant: &str,
430    ) -> Option<Arc<greentic_aw_runtime::ComponentToolSource>> {
431        if std::env::var("GREENTIC_AW_COMPONENT_TOOLS").ok().as_deref() == Some("0") {
432            tracing::info!("GREENTIC_AW_COMPONENT_TOOLS=0; component tool source disabled");
433            return None;
434        }
435        if packs.is_empty() {
436            return None;
437        }
438        let invoker = Arc::new(
439            crate::runner::component_invoker::PackRuntimeComponentInvoker::new(
440                packs.to_vec(),
441                tenant.to_string(),
442            ),
443        );
444        tracing::info!(tenant = %tenant, packs = packs.len(), "component tool source constructed");
445        Some(Arc::new(greentic_aw_runtime::ComponentToolSource::new(
446            invoker,
447        )))
448    }
449
450    /// Build the production [`greentic_ext_runtime::ExtensionRuntime`] used for
451    /// tool dispatch, wrapped in an [`Arc`] for sharing with [`AgentRuntime`].
452    ///
453    /// On construction failure (e.g. wasmtime engine init), logs the error and
454    /// returns `None`; the caller then disables `DwAgent` nodes rather than
455    /// panicking.
456    ///
457    /// Unlike the designer (which installs extensions through an explicit flow),
458    /// the runner has no install step — so it performs an initial scan of the
459    /// `design/` kind directory under the discovery root and registers each
460    /// on-disk extension here. Without this the agentic worker would boot with
461    /// an empty tool runtime and every extension tool would be silently dropped.
462    /// Per-extension failures (bad signature, malformed describe) are logged and
463    /// skipped so one broken extension never aborts boot; the watcher still
464    /// hot-reloads later changes.
465    /// Resolves an agentic-worker tool extension's `secret://…` reference to an
466    /// environment variable: strip the `secret://` scheme and upper-case every
467    /// run of non-alphanumeric chars to a single `_` — e.g.
468    /// `secret://tavily/api_key` → `TAVILY_API_KEY`. Lets local/desktop runs
469    /// supply tool secrets via the env (the in-process AW path has no broker).
470    pub(crate) struct EnvSecretsBackend;
471
472    impl greentic_ext_runtime::SecretsBackend for EnvSecretsBackend {
473        fn get(&self, uri: &str) -> Result<String, greentic_ext_runtime::SecretsError> {
474            let name = env_var_name_for_secret(uri);
475            std::env::var(&name)
476                .map_err(|_| greentic_ext_runtime::SecretsError::NotFound(uri.to_string()))
477        }
478    }
479
480    /// Tool-secret backend that resolves an extension's `secret://<provider>/<key>`
481    /// reference from the per-tenant secrets store first, then falls back to the
482    /// process env. This is what makes `gtc start` zero-env: `gtc setup` persists
483    /// the value in the dev store, and the injected secrets manager's read-side
484    /// candidate fallback bridges the canonical `secrets://{env}/{tenant}/_/{provider}/{key}`
485    /// scope to the pack-namespaced scope setup actually wrote. The env fallback
486    /// preserves existing `TAVILY_API_KEY`-style runs.
487    struct StoreToolSecretsBackend {
488        secrets: crate::secrets::DynSecretsManager,
489        tenant: String,
490        env: String,
491    }
492
493    impl StoreToolSecretsBackend {
494        fn new(secrets: crate::secrets::DynSecretsManager, tenant: String) -> Self {
495            let env = std::env::var("GREENTIC_ENV")
496                .ok()
497                .filter(|value| !value.trim().is_empty())
498                .unwrap_or_else(|| "dev".to_string());
499            Self {
500                secrets,
501                tenant,
502                env,
503            }
504        }
505
506        /// Map `secret://<provider>/<key>` to the canonical store URI
507        /// `secrets://{env}/{tenant}/_/{provider}/{key}`. The injected manager's
508        /// candidate fallback handles the env/team/pack-namespace bridging.
509        fn canonical_store_uri(&self, uri: &str) -> Option<String> {
510            let body = uri.strip_prefix("secret://").unwrap_or(uri);
511            let (provider, key) = body.split_once('/')?;
512            if provider.is_empty() || key.is_empty() {
513                return None;
514            }
515            Some(format!(
516                "secrets://{}/{}/_/{}/{}",
517                self.env, self.tenant, provider, key
518            ))
519        }
520    }
521
522    impl greentic_ext_runtime::SecretsBackend for StoreToolSecretsBackend {
523        fn get(&self, uri: &str) -> Result<String, greentic_ext_runtime::SecretsError> {
524            if let Some(store_uri) = self.canonical_store_uri(uri) {
525                // Read off a dedicated thread with its own current-thread runtime:
526                // the extension runtime may invoke this from within the async
527                // runner, where a nested `block_on` would panic.
528                let secrets = self.secrets.clone();
529                let resolved = std::thread::spawn(move || {
530                    let runtime = tokio::runtime::Builder::new_current_thread()
531                        .enable_all()
532                        .build()
533                        .ok()?;
534                    runtime.block_on(async move { secrets.read(&store_uri).await.ok() })
535                })
536                .join()
537                .ok()
538                .flatten();
539                if let Some(bytes) = resolved
540                    && let Ok(value) = String::from_utf8(bytes)
541                {
542                    return Ok(value);
543                }
544            }
545            // Fallback: env var (preserves pre-store behaviour).
546            let name = env_var_name_for_secret(uri);
547            std::env::var(&name)
548                .map_err(|_| greentic_ext_runtime::SecretsError::NotFound(uri.to_string()))
549        }
550    }
551
552    /// A process-shared blocking HTTP client for tool extensions.
553    ///
554    /// `reqwest::blocking::Client` owns an internal tokio runtime; dropping it
555    /// from within an async context panics ("cannot drop a runtime …"). The
556    /// in-process AW path creates and drops short-lived `ExtensionRuntime`s
557    /// inside the async runner, so we keep ONE client alive for the whole
558    /// process (built off the async runtime) and hand out cheap clones — a
559    /// clone dropped in async context never drops the underlying runtime, which
560    /// is released only at process exit (outside any runtime).
561    fn shared_blocking_http_client() -> Option<greentic_ext_runtime::reqwest::blocking::Client> {
562        use std::sync::OnceLock;
563        static CLIENT: OnceLock<Option<greentic_ext_runtime::reqwest::blocking::Client>> =
564            OnceLock::new();
565        CLIENT
566            .get_or_init(|| {
567                // Build on a plain OS thread so reqwest's internal runtime is
568                // not constructed inside a tokio context.
569                std::thread::spawn(|| {
570                    greentic_ext_runtime::reqwest::blocking::Client::builder()
571                        .timeout(std::time::Duration::from_secs(30))
572                        .build()
573                        .ok()
574                })
575                .join()
576                .ok()
577                .flatten()
578            })
579            .clone()
580    }
581
582    pub(crate) fn env_var_name_for_secret(uri: &str) -> String {
583        let body = uri.strip_prefix("secret://").unwrap_or(uri);
584        let mut out = String::with_capacity(body.len());
585        let mut prev_underscore = false;
586        for ch in body.chars() {
587            if ch.is_ascii_alphanumeric() {
588                out.push(ch.to_ascii_uppercase());
589                prev_underscore = false;
590            } else if !prev_underscore {
591                out.push('_');
592                prev_underscore = true;
593            }
594        }
595        out.trim_matches('_').to_string()
596    }
597
598    pub(crate) fn build_ext_runtime(
599        secrets_backend: Arc<dyn greentic_ext_runtime::SecretsBackend>,
600    ) -> Option<Arc<greentic_ext_runtime::ExtensionRuntime>> {
601        use greentic_ext_runtime::{
602            DiscoveryPaths, ExtensionRuntime, HostOverrides, RuntimeConfig, discovery,
603        };
604
605        let root = extension_discovery_dir();
606        let paths = DiscoveryPaths::new(root.clone());
607        // Wire the provided secrets backend and an HTTP client so tool
608        // extensions can read their declared `secret://…` references and reach
609        // their upstreams. `HostOverrides::default()` leaves both empty, which
610        // silently breaks any AW tool that needs either (e.g. tavily_search).
611        // The per-tenant path passes a store-backed backend (zero-env); the
612        // process-level serve paths pass the env-only backend.
613        // This lane's HostOverrides has no `Default`, and `with_host_overrides`
614        // is a builder on the runtime rather than on the config — so spell
615        // every field out and apply the overrides after construction. The
616        // non-obvious ones mirror the crate's own `defaults_for_tests`:
617        // `runtime_weak` stays unset until the cross-extension dispatch
618        // cascade lands, and `call_depth_start` is the recursion guard's floor.
619        let overrides = HostOverrides {
620            translator: std::sync::Arc::new(greentic_ext_runtime::host_ports::KeyTranslator),
621            secrets_backend,
622            http_client: shared_blocking_http_client(),
623            url_matcher: greentic_ext_runtime::url_matcher::UrlMatcher::default(),
624            runtime_weak: std::sync::Weak::new(),
625            call_depth_start: 0,
626        };
627        let config = RuntimeConfig::from_paths(paths);
628        let mut runtime = match ExtensionRuntime::new(config) {
629            Ok(runtime) => runtime.with_host_overrides(overrides),
630            Err(error) => {
631                tracing::warn!(error = %error, "extension runtime init failed; DwAgent nodes disabled");
632                return None;
633            }
634        };
635
636        // Initial load of on-disk design extensions (agentic-worker tools live
637        // in `<root>/design/<ext>/`).
638        let design_dir = root.join("design");
639        match discovery::scan_kind_dir(&design_dir) {
640            Ok(ext_dirs) => {
641                let mut loaded = 0usize;
642                for ext_dir in ext_dirs {
643                    match runtime.register_loaded_from_dir(&ext_dir) {
644                        Ok(()) => loaded += 1,
645                        Err(error) => tracing::warn!(
646                            error = %error, dir = %ext_dir.display(),
647                            "skipping extension that failed to load"
648                        ),
649                    }
650                }
651                tracing::info!(loaded, dir = %design_dir.display(), "loaded design extensions");
652            }
653            Err(error) => {
654                tracing::warn!(error = %error, dir = %design_dir.display(), "scanning design extensions failed")
655            }
656        }
657
658        Some(Arc::new(runtime))
659    }
660
661    /// Resolve the [`LlmBackend`] from the environment.
662    ///
663    /// Prefers the LLM bridge extension when `GREENTIC_AW_LLM_EXTENSION` is set
664    /// (LLM-as-extension); otherwise falls back to the env-keyed in-process
665    /// OpenAI client. Shared by the single-agent (`build_agent_node_handler`)
666    /// and graph (`graph_node::build_graph_node_handler`) construction paths so
667    /// both resolve the backend identically.
668    pub(crate) fn build_llm_backend(
669        ext_runtime: &Arc<greentic_ext_runtime::ExtensionRuntime>,
670    ) -> Arc<dyn greentic_aw_runtime::LlmBackend> {
671        use std::time::Duration;
672
673        use greentic_aw_runtime::{ExtensionLlmBackend, OpenAiLlmBackend, RetryingLlmBackend};
674
675        match std::env::var("GREENTIC_AW_LLM_EXTENSION")
676            .ok()
677            .filter(|s| !s.trim().is_empty())
678        {
679            Some(ext_id) => {
680                let api_key = std::env::var("GREENTIC_LLM_API_KEY")
681                    .or_else(|_| std::env::var("OPENAI_API_KEY"))
682                    .unwrap_or_default();
683                match bridge_credential(
684                    std::env::var("GREENTIC_LLM_PROVIDER").ok(),
685                    std::env::var("GREENTIC_LLM_MODEL").ok(),
686                    api_key,
687                    std::env::var("GREENTIC_LLM_BASE_URL").ok(),
688                ) {
689                    Some(cred) => {
690                        tracing::info!(
691                            extension = %ext_id, provider = %cred.provider, model = %cred.model,
692                            "AW LLM via bridge extension"
693                        );
694                        Arc::new(RetryingLlmBackend::new(
695                            ExtensionLlmBackend::new(ext_runtime.clone(), ext_id, cred),
696                            3,
697                            Duration::from_millis(250),
698                        ))
699                    }
700                    None => {
701                        tracing::warn!(
702                            "GREENTIC_AW_LLM_EXTENSION set but no LLM API key; \
703                             falling back to in-process OpenAI client"
704                        );
705                        Arc::new(RetryingLlmBackend::new(
706                            OpenAiLlmBackend::new(String::new()),
707                            3,
708                            Duration::from_millis(250),
709                        ))
710                    }
711                }
712            }
713            None => in_process_llm_backend(),
714        }
715    }
716
717    /// In-process LLM backend when no bridge extension is configured.
718    ///
719    /// With the `greentic-llm-backend` feature and an LLM key present, routes the
720    /// worker's LLM call through greentic-llm so a `dw.agent` can use any provider
721    /// its `AgentConfig.llm` declares (DeepSeek, Anthropic, Gemini, …) — the
722    /// provider + model ride on each request, the key + optional base URL come
723    /// from the env. Otherwise falls back to the legacy env-keyed OpenAI client.
724    /// Shared by every non-bridge construction path so they never drift.
725    pub(crate) fn in_process_llm_backend() -> Arc<dyn greentic_aw_runtime::LlmBackend> {
726        in_process_llm_backend_with_key(None)
727    }
728
729    /// In-process LLM backend, optionally given a store-resolved API key.
730    ///
731    /// `override_key` (resolved from an agent's `credential_ref` via the
732    /// per-tenant secrets store) takes precedence over the env key when present —
733    /// this is what makes the in-process desktop LLM path zero-env. When `None`,
734    /// the key comes from `GREENTIC_LLM_API_KEY`/`OPENAI_API_KEY` exactly as
735    /// before, so env-based and bridge-less runs are unaffected.
736    pub(crate) fn in_process_llm_backend_with_key(
737        override_key: Option<String>,
738    ) -> Arc<dyn greentic_aw_runtime::LlmBackend> {
739        use greentic_aw_runtime::{OpenAiLlmBackend, RetryingLlmBackend};
740        use std::time::Duration;
741
742        // Only read under `greentic-llm-backend` (below); prefixed so the
743        // binding doesn't trip `unused_variables` when that feature is off
744        // (it isn't a default feature — see Cargo.toml).
745        let _store_resolved = override_key
746            .as_ref()
747            .map(|key| !key.trim().is_empty())
748            .unwrap_or(false);
749
750        #[cfg(feature = "greentic-llm-backend")]
751        {
752            let api_key = override_key
753                .clone()
754                .filter(|key| !key.trim().is_empty())
755                .or_else(|| std::env::var("GREENTIC_LLM_API_KEY").ok())
756                .or_else(|| std::env::var("OPENAI_API_KEY").ok())
757                .unwrap_or_default();
758            if !api_key.trim().is_empty() {
759                let base_url = std::env::var("GREENTIC_LLM_BASE_URL").ok();
760                tracing::info!(
761                    store_resolved = _store_resolved,
762                    "AW LLM via in-process greentic-llm (multi-provider)"
763                );
764                return Arc::new(RetryingLlmBackend::new(
765                    greentic_aw_runtime::GreenticLlmBackend::new(api_key, base_url),
766                    3,
767                    Duration::from_millis(250),
768                ));
769            }
770        }
771        let openai_key = override_key
772            .filter(|key| !key.trim().is_empty())
773            .or_else(|| std::env::var("OPENAI_API_KEY").ok())
774            .unwrap_or_default();
775        Arc::new(RetryingLlmBackend::new(
776            OpenAiLlmBackend::new(openai_key),
777            3,
778            Duration::from_millis(250),
779        ))
780    }
781
782    /// Resolve an LLM API key from the per-tenant secrets store via the first
783    /// agent that declares `llm.credential_ref`, mirroring the credential URI
784    /// `secrets://default/{tenant}/_/llm/{credential_ref}` that
785    /// [`greentic_aw_runtime::llm_credential::SecretsBackedCredentialResolver`]
786    /// reads. This lets the in-process LLM backend be zero-env (key from store)
787    /// when no `GREENTIC_LLM_API_KEY` env is set.
788    ///
789    /// Returns `None` when no agent declares a credential_ref or the read misses.
790    /// The in-process backend carries a single key, so when agents declare
791    /// different credential_refs only the first is used — matching the existing
792    /// one-key in-process model; the bridge-extension path resolves per-request.
793    async fn resolve_in_process_llm_key(
794        secrets: &crate::secrets::DynSecretsManager,
795        tenant: &str,
796        merged_agents: &HashMap<String, AgentConfig>,
797    ) -> Option<String> {
798        let credential_ref = merged_agents
799            .values()
800            .find_map(|agent| agent.llm.credential_ref.clone())?;
801        let uri = format!("secrets://default/{tenant}/_/llm/{credential_ref}");
802        let bytes = secrets.read(&uri).await.ok()?;
803        let key = String::from_utf8(bytes).ok()?.trim().to_string();
804        if key.is_empty() { None } else { Some(key) }
805    }
806
807    /// Build a vault-style `BridgeCredential` from resolved parts. `None` when no
808    /// API key is present. Defaults: provider "openai", model "gpt-4o". Pure (no
809    /// env) so it is unit-testable without global state.
810    pub(super) fn bridge_credential(
811        provider: Option<String>,
812        model: Option<String>,
813        api_key: String,
814        base_url: Option<String>,
815    ) -> Option<greentic_aw_runtime::BridgeCredential> {
816        if api_key.trim().is_empty() {
817            return None;
818        }
819        Some(greentic_aw_runtime::BridgeCredential {
820            provider: provider
821                .filter(|s| !s.trim().is_empty())
822                .unwrap_or_else(|| "openai".into()),
823            model: model
824                .filter(|s| !s.trim().is_empty())
825                .unwrap_or_else(|| "gpt-4o".into()),
826            api_key,
827            base_url: base_url.filter(|s| !s.trim().is_empty()),
828        })
829    }
830
831    /// Shared store-agnostic tail: builds the extension runtime, LLM backend,
832    /// config providers, and [`AgentRuntime`] given the three already-constructed
833    /// store trait objects.
834    ///
835    /// Extracted so both the Redis path ([`build_agent_node_handler`]) and the
836    /// ephemeral desktop path ([`build_agent_node_handler_ephemeral`]) share
837    /// identical post-store construction logic; store differences are the only
838    /// divergence between the two callers.
839    ///
840    /// Returns `None` when the extension runtime fails to initialise (the only
841    /// failure mode at this layer — store errors are handled by callers).
842    ///
843    /// `audit_sink` (EPIC-B B-3) is forwarded verbatim to the constructed
844    /// [`RuntimeAgentNodeHandler`] — `None` keeps `dw.agent` execution on the
845    /// plain [`AgentRuntime::step`] path.
846    ///
847    /// `project_id` is the deployed unit's `bundle_id`, forwarded verbatim so
848    /// billing can attribute this runtime's spend to a project. `None` on the
849    /// legacy tenant-only path, where the dimension is omitted entirely.
850    #[allow(clippy::too_many_arguments)]
851    async fn build_runtime_handler_with_stores(
852        merged_agents: HashMap<String, AgentConfig>,
853        tenant: String,
854        secrets: crate::secrets::DynSecretsManager,
855        packs: Vec<Arc<crate::pack::PackRuntime>>,
856        state_store: Arc<dyn greentic_aw_runtime::state::AgentStateStore>,
857        token_meter: Arc<dyn greentic_aw_runtime::cost::TokenMeter>,
858        ledger: Arc<dyn greentic_aw_runtime::tools::ToolLedger>,
859        audit_sink: Option<AuditSink>,
860        project_id: Option<String>,
861    ) -> Option<Arc<dyn AgentNodeHandler>> {
862        use std::time::Duration;
863
864        use greentic_aw_runtime::LayeredConfigProvider;
865        use greentic_aw_runtime::ManifestToolOverlayProvider;
866        use greentic_aw_runtime::config_provider::CachingConfigProvider;
867        use greentic_aw_runtime::{
868            ExtensionLlmBackend, LlmBackend, OtelTelemetry, RetryingLlmBackend,
869        };
870
871        // Per-tenant path: tool secrets resolve from the store first (zero-env),
872        // env as fallback. `secrets` is the injected per-tenant manager whose
873        // candidate fallback bridges the gtc-setup dev-store scope.
874        let secrets_backend: Arc<dyn greentic_ext_runtime::SecretsBackend> = Arc::new(
875            StoreToolSecretsBackend::new(secrets.clone(), tenant.clone()),
876        );
877        let ext_runtime = build_ext_runtime(secrets_backend)?;
878
879        // When the LLM bridge extension is configured, resolve credentials
880        // per-tenant from the secrets broker rather than from global env vars.
881        // The env-keyed OpenAI fallback is preserved for both branches.
882        let llm: Arc<dyn LlmBackend> = match std::env::var("GREENTIC_AW_LLM_EXTENSION")
883            .ok()
884            .filter(|s| !s.trim().is_empty())
885        {
886            Some(ext_id) => {
887                use greentic_aw_runtime::llm_credential::SecretsBackedCredentialResolver;
888                let resolver = Arc::new(SecretsBackedCredentialResolver::new(
889                    secrets.clone(),
890                    tenant.clone(),
891                ));
892                tracing::info!(
893                    extension = %ext_id,
894                    tenant = %tenant,
895                    "AW LLM via bridge (per-tenant creds)"
896                );
897                Arc::new(RetryingLlmBackend::new(
898                    ExtensionLlmBackend::with_resolver_runtime(
899                        ext_runtime.clone(),
900                        ext_id,
901                        resolver,
902                    ),
903                    3,
904                    Duration::from_millis(250),
905                ))
906            }
907            None => {
908                // Zero-env LLM: with no bridge extension and no env key, resolve
909                // the agent's `credential_ref` from the per-tenant store (the same
910                // manager whose candidate fallback bridges the gtc-setup scope).
911                // An env key still wins (legacy single-provider runs unaffected).
912                let env_key_present = std::env::var("GREENTIC_LLM_API_KEY")
913                    .ok()
914                    .filter(|value| !value.trim().is_empty())
915                    .or_else(|| {
916                        std::env::var("OPENAI_API_KEY")
917                            .ok()
918                            .filter(|value| !value.trim().is_empty())
919                    })
920                    .is_some();
921                let store_key = if env_key_present {
922                    None
923                } else {
924                    resolve_in_process_llm_key(&secrets, &tenant, &merged_agents).await
925                };
926                if store_key.is_some() {
927                    tracing::info!(
928                        tenant = %tenant,
929                        "AW LLM key resolved from store via credential_ref (zero-env)"
930                    );
931                }
932                in_process_llm_backend_with_key(store_key)
933            }
934        };
935
936        let agent_count = merged_agents.len();
937        let overlay = ManifestToolOverlayProvider::new(
938            HostConfigProvider::new(merged_agents),
939            manifests_discovery_dir(),
940        );
941        let config_provider: Arc<dyn ConfigProvider> = match registry_from_env() {
942            Some(http) => Arc::new(CachingConfigProvider::new(LayeredConfigProvider::new(
943                http, overlay,
944            ))),
945            None => Arc::new(CachingConfigProvider::new(overlay)),
946        };
947        let telemetry = Arc::new(OtelTelemetry);
948
949        let base = AgentRuntime::new(
950            config_provider,
951            state_store,
952            ext_runtime,
953            llm,
954            telemetry,
955            token_meter,
956            ledger,
957            mcp_source_from_env(),
958        )
959        .with_component_source(component_source_from_packs(&packs, &tenant));
960
961        // Billing metering, identical to the `build_agent_runtime` serve path.
962        // Without this the in-process `dw.agent` node ran on the default
963        // `NoopBillingMeter`: its LLM spend was never metered AND the credit
964        // gate never fired, so an out-of-credit tenant kept running for free
965        // through this node while the out-of-process path stopped them.
966        //
967        // Ship-dark, same as the serve path: a no-op until an operator sets
968        // GREENTIC_BILLING_BASE_URL + GREENTIC_BILLING_SERVICE_SECRET.
969        let billing_enabled;
970        let base = match greentic_aw_runtime::billing::HttpBillingMeter::from_env() {
971            Some(http_meter) => {
972                billing_enabled = true;
973                base.with_billing_meter(Arc::new(http_meter))
974            }
975            None => {
976                billing_enabled = false;
977                base
978            }
979        };
980        let runtime = Arc::new(base);
981
982        // `billing_enabled` is logged rather than left implicit: this path
983        // silently metered nothing for its whole existence, and a boot line
984        // stating it either way is what makes that visible to an operator.
985        tracing::info!(
986            agent_count,
987            tenant = %tenant,
988            billing_enabled,
989            "AW runtime constructed"
990        );
991        Some(Arc::new(RuntimeAgentNodeHandler::new(
992            runtime, audit_sink, project_id,
993        )))
994    }
995
996    /// Build the production `DwAgent` handler if the environment is configured.
997    ///
998    /// Returns `None` (so `DwAgent` flow dispatch errors clearly) under any of
999    /// these graceful-degradation conditions:
1000    /// - `merged_agents` is empty (no agents from packs or operator config);
1001    /// - `GREENTIC_AW_REDIS_URL` is unset/empty;
1002    /// - the AW Redis connection fails;
1003    /// - the extension runtime fails to initialise.
1004    ///
1005    /// `merged_agents` is the result of merging pack-embedded agents (base)
1006    /// with operator-declared [`HostConfig::agents`] (operator wins on
1007    /// collision). This merged map replaces the former direct read of
1008    /// `config.agents` so pack-provided agents are included in the runtime.
1009    ///
1010    /// Redis is sourced from the environment because the runner uses an
1011    /// in-memory flow-state store by default and carries no Redis URL in
1012    /// [`HostConfig`]; this mirrors the existing env-config convention.
1013    ///
1014    /// The `tenant` and `secrets` arguments wire in per-tenant LLM credential
1015    /// resolution when `GREENTIC_AW_LLM_EXTENSION` is set: requests resolve
1016    /// credentials from the secrets broker for the calling tenant rather than
1017    /// reading global env vars. Callers without per-tenant context (e.g.
1018    /// `serve_agentic`) should use [`build_agent_runtime`] directly, which uses
1019    /// the env-keyed backend and accepts no secrets context.
1020    ///
1021    /// `audit_sink` (EPIC-B B-3) is threaded straight through to the
1022    /// constructed [`RuntimeAgentNodeHandler`]; `None` (no NATS audit client
1023    /// configured) keeps `dw.agent` execution on the plain
1024    /// [`greentic_aw_runtime::AgentRuntime::step`] path — zero behaviour
1025    /// change from before this parameter existed.
1026    ///
1027    /// `project_id` is the deployed unit's `bundle_id` — the value
1028    /// greentic-designer records as `pack_name` — used as the `project_id`
1029    /// billing dimension. Pass `None` when no bundle is pinned (the legacy
1030    /// tenant-only pack path); billing then omits the dimension instead of
1031    /// attributing spend to a non-unique agent id.
1032    pub async fn build_agent_node_handler(
1033        merged_agents: HashMap<String, AgentConfig>,
1034        tenant: String,
1035        secrets: crate::secrets::DynSecretsManager,
1036        packs: Vec<Arc<crate::pack::PackRuntime>>,
1037        audit_sink: Option<AuditSink>,
1038        project_id: Option<String>,
1039    ) -> Option<Arc<dyn AgentNodeHandler>> {
1040        use greentic_aw_runtime::RedisAgentStateStore;
1041        use greentic_aw_runtime::cost::RedisTokenMeter;
1042        use greentic_aw_runtime::tools::RedisToolLedger;
1043
1044        if merged_agents.is_empty() {
1045            return None;
1046        }
1047
1048        let redis_url = match std::env::var("GREENTIC_AW_REDIS_URL") {
1049            Ok(url) if !url.is_empty() => url,
1050            _ => {
1051                tracing::info!("GREENTIC_AW_REDIS_URL unset; DwAgent nodes disabled");
1052                return None;
1053            }
1054        };
1055
1056        let state_store = match RedisAgentStateStore::connect(&redis_url).await {
1057            Ok(store) => Arc::new(store),
1058            Err(error) => {
1059                tracing::warn!(error = %error, "AW Redis connect failed; DwAgent nodes disabled");
1060                return None;
1061            }
1062        };
1063
1064        let manager = state_store.manager();
1065        let token_meter = Arc::new(RedisTokenMeter::new(manager.clone()));
1066        let ledger = Arc::new(RedisToolLedger::new(manager));
1067
1068        build_runtime_handler_with_stores(
1069            merged_agents,
1070            tenant,
1071            secrets,
1072            packs,
1073            state_store,
1074            token_meter,
1075            ledger,
1076            audit_sink,
1077            project_id,
1078        )
1079        .await
1080    }
1081
1082    /// Desktop/local builder: in-memory state, token meter, and ledger so
1083    /// `gtc start` runs agents with NO external infra. State is ephemeral
1084    /// (lost on process exit) — never used by the server path.
1085    ///
1086    /// Returns `None` only when `merged_agents` is empty or the extension
1087    /// runtime fails to initialise. Unlike [`build_agent_node_handler`] this
1088    /// function never returns `None` due to a missing Redis URL, making it safe
1089    /// for desktop environments where no Redis is available.
1090    ///
1091    /// `project_id` mirrors [`build_agent_node_handler`]: the deployed unit's
1092    /// `bundle_id`, or `None` when none is pinned (billing then omits the
1093    /// dimension).
1094    #[cfg(feature = "desktop-agent-ephemeral")]
1095    pub async fn build_agent_node_handler_ephemeral(
1096        merged_agents: HashMap<String, AgentConfig>,
1097        tenant: String,
1098        secrets: crate::secrets::DynSecretsManager,
1099        packs: Vec<Arc<crate::pack::PackRuntime>>,
1100        audit_sink: Option<AuditSink>,
1101        project_id: Option<String>,
1102    ) -> Option<Arc<dyn AgentNodeHandler>> {
1103        use greentic_aw_runtime::cost::MockTokenMeter;
1104        use greentic_aw_runtime::mock::{MockAgentStateStore, NoopToolLedger};
1105        use std::sync::OnceLock;
1106
1107        if merged_agents.is_empty() {
1108            return None;
1109        }
1110        // Process-global in-memory state store, shared across every flow
1111        // invocation in this process. The desktop runner rebuilds the agent
1112        // handler on each `dw.agent` node call, so a per-call store would erase
1113        // conversation memory between turns — a multi-turn agent could never
1114        // act on a user's "yes" to a proposal made on the previous turn. A
1115        // single shared store keeps memory alive for the lifetime of the
1116        // process WITHOUT any external infrastructure (Redis remains optional,
1117        // selected only when GREENTIC_AW_REDIS_URL is set). State is still lost
1118        // on process exit — that is the documented desktop trade-off.
1119        static EPHEMERAL_STATE_STORE: OnceLock<Arc<MockAgentStateStore>> = OnceLock::new();
1120        tracing::warn!(
1121            tenant = %tenant,
1122            "AW desktop ephemeral state store (in-memory, process-global; persists across \
1123             turns for this process, lost on exit)"
1124        );
1125        let state_store =
1126            Arc::clone(EPHEMERAL_STATE_STORE.get_or_init(|| Arc::new(MockAgentStateStore::new())));
1127        let token_meter = Arc::new(MockTokenMeter::new(0));
1128        let ledger = Arc::new(NoopToolLedger);
1129        build_runtime_handler_with_stores(
1130            merged_agents,
1131            tenant,
1132            secrets,
1133            packs,
1134            state_store,
1135            token_meter,
1136            ledger,
1137            audit_sink,
1138            project_id,
1139        )
1140        .await
1141    }
1142
1143    /// Construct the shared [`AgentRuntime`] from the environment.
1144    ///
1145    /// Factored out of [`build_agent_node_handler`] so both the in-process
1146    /// `dw.agent`/`agentic.call` flow node and the out-of-process NATS serve
1147    /// mode ([`serve_agentic`]) build an identical runtime (Redis state, env-
1148    /// resolved LLM backend, design extensions, agent config providers, MCP).
1149    ///
1150    /// Returns `None` under the same graceful-degradation conditions as the node
1151    /// handler: empty agent map, missing/unreachable `GREENTIC_AW_REDIS_URL`, or
1152    /// extension-runtime init failure.
1153    pub async fn build_agent_runtime(
1154        merged_agents: HashMap<String, AgentConfig>,
1155    ) -> Option<Arc<AgentRuntime>> {
1156        use greentic_aw_runtime::LayeredConfigProvider;
1157        use greentic_aw_runtime::ManifestToolOverlayProvider;
1158        use greentic_aw_runtime::config_provider::CachingConfigProvider;
1159        use greentic_aw_runtime::cost::RedisTokenMeter;
1160        use greentic_aw_runtime::tools::RedisToolLedger;
1161        use greentic_aw_runtime::{OtelTelemetry, RedisAgentStateStore};
1162
1163        if merged_agents.is_empty() {
1164            return None; // nothing to serve
1165        }
1166
1167        let redis_url = match std::env::var("GREENTIC_AW_REDIS_URL") {
1168            Ok(url) if !url.is_empty() => url,
1169            _ => {
1170                tracing::info!("GREENTIC_AW_REDIS_URL unset; DwAgent nodes disabled");
1171                return None;
1172            }
1173        };
1174
1175        let state_store = match RedisAgentStateStore::connect(&redis_url).await {
1176            Ok(store) => Arc::new(store),
1177            Err(error) => {
1178                tracing::warn!(error = %error, "AW Redis connect failed; DwAgent nodes disabled");
1179                return None;
1180            }
1181        };
1182
1183        // The connection manager is cheap to clone (multiplexed, ref-counted);
1184        // share it with the token meter and idempotency ledger.
1185        let manager = state_store.manager();
1186        let token_meter = Arc::new(RedisTokenMeter::new(manager.clone()));
1187        let ledger = Arc::new(RedisToolLedger::new(manager));
1188
1189        // Process-level serve path has no per-tenant secrets context, so tool
1190        // secrets resolve from the env only.
1191        let ext_runtime = build_ext_runtime(Arc::new(EnvSecretsBackend))?;
1192
1193        // Prefer the LLM bridge extension when configured (LLM-as-extension);
1194        // fall back to the env-keyed in-process OpenAI client otherwise.
1195        // NOTE: this path has no per-tenant secrets context (it is used by
1196        // `serve_agentic` and process-level in-proc serve). Per-tenant
1197        // credential resolution is only available via `build_agent_node_handler`.
1198        let llm = build_llm_backend(&ext_runtime);
1199
1200        let agent_count = merged_agents.len();
1201        // Base config source = the merged agents (pack-embedded ⊕ operator,
1202        // operator wins). Wrap in the manifest-tool overlay, then layer the
1203        // admin agent registry on top when configured (registry first, overlay
1204        // fallback); cache the result either way.
1205        let overlay = ManifestToolOverlayProvider::new(
1206            HostConfigProvider::new(merged_agents),
1207            manifests_discovery_dir(),
1208        );
1209        let config_provider: Arc<dyn ConfigProvider> = match registry_from_env() {
1210            Some(http) => Arc::new(CachingConfigProvider::new(LayeredConfigProvider::new(
1211                http, overlay,
1212            ))),
1213            None => Arc::new(CachingConfigProvider::new(overlay)),
1214        };
1215        let telemetry = Arc::new(OtelTelemetry);
1216
1217        let base = AgentRuntime::new(
1218            config_provider,
1219            state_store,
1220            ext_runtime.clone(),
1221            llm,
1222            telemetry,
1223            token_meter,
1224            ledger,
1225            mcp_source_from_env(),
1226        )
1227        .with_guardrails(
1228            {
1229                let policy: Arc<dyn greentic_aw_runtime::guardrail::GuardrailPolicy> =
1230                    match guardrail_policy_from_env() {
1231                        Some(http) => Arc::new(http),
1232                        None => Arc::new(greentic_aw_runtime::guardrail::StaticGuardrailPolicy(
1233                            Vec::new(),
1234                        )),
1235                    };
1236                policy
1237            },
1238            {
1239                #[cfg(greentic_guardrail_ext)]
1240                {
1241                    Arc::new(
1242                        greentic_aw_runtime::guardrail::ExtRuntimeGuardrailEvaluator {
1243                            ext_runtime: ext_runtime.clone(),
1244                        },
1245                    )
1246                }
1247                // This lane's greentic-ext-runtime has no guardrail interface
1248                // (extension-design is at 0.2.0). Fail closed rather than
1249                // silently accepting: an agent that configured a guardrail
1250                // stops loudly instead of running unprotected.
1251                #[cfg(not(greentic_guardrail_ext))]
1252                {
1253                    Arc::new(greentic_aw_runtime::guardrail::UnavailableGuardrailEvaluator)
1254                }
1255            },
1256        );
1257        // Short-term ("working") memory: in-memory provider is always available
1258        // (no external deps); the remember/recall tools stay gated by
1259        // config.memory.short_term in the loop.
1260        let base = base.with_short_term_memory(Arc::new(
1261            greentic_aw_runtime::memory::InMemoryMemoryProvider::new(),
1262        ));
1263        // Billing metering: install the HTTP sink when both env vars are set;
1264        // fall back to the built-in no-op (billing disabled) otherwise.
1265        // Ship-dark: no-op until GREENTIC_BILLING_BASE_URL +
1266        // GREENTIC_BILLING_SERVICE_SECRET are configured by the operator.
1267        let base =
1268            if let Some(http_meter) = greentic_aw_runtime::billing::HttpBillingMeter::from_env() {
1269                tracing::info!("billing metering enabled for digital-worker LLM usage");
1270                base.with_billing_meter(std::sync::Arc::new(http_meter))
1271            } else {
1272                base
1273            };
1274        // Optionally attach an operator-configured native long-term memory
1275        // backend. With the `long-term-chronicle` feature off (default) this is
1276        // a no-op and `base` is wrapped unchanged.
1277        #[cfg(feature = "long-term-chronicle")]
1278        let base = crate::runner::long_term_memory::attach(base).await;
1279        // Optionally attach an operator-configured Chronicle knowledge (document
1280        // RAG) backend for auto pre-retrieval. No-op with the `knowledge-chronicle`
1281        // feature off (default).
1282        #[cfg(feature = "knowledge-chronicle")]
1283        let base = crate::runner::knowledge_mount::attach(base).await;
1284        let runtime = Arc::new(base);
1285
1286        tracing::info!(agent_count, "AW runtime constructed");
1287        Some(runtime)
1288    }
1289
1290    /// Run the agentic-worker runtime as a NATS-consuming service.
1291    ///
1292    /// Builds the production [`AgentRuntime`] via [`build_agent_runtime`] and,
1293    /// when it could be constructed, serves `greentic.agentic.request.v1`
1294    /// forever via the shared `aw-event-bridge`. This is the out-of-process
1295    /// (`agentic.call`) counterpart to the in-process `dw.agent` node.
1296    ///
1297    /// When `GREENTIC_AW_REDIS_URL` is set and reachable the serve path wires
1298    /// a [`greentic_aw_runtime::RedisDispatchLedger`] so JetStream at-least-once
1299    /// redeliveries are short-circuited without re-running the LLM step. If the
1300    /// Redis connect fails the ledger falls back to [`greentic_aw_runtime::NoopDispatchLedger`]
1301    /// (idempotency disabled, warning logged) so serving is never blocked.
1302    ///
1303    /// Returns `Ok(())` immediately (a no-op) when the runtime cannot be built
1304    /// (e.g. no agents, no Redis) so the host can call this unconditionally.
1305    pub async fn serve_agentic(
1306        nats_url: &str,
1307        merged_agents: HashMap<String, AgentConfig>,
1308    ) -> anyhow::Result<()> {
1309        use greentic_aw_runtime::dispatch_ledger::RedisDispatchLedger;
1310        use greentic_aw_runtime::{DispatchLedger, NoopDispatchLedger, RedisAgentStateStore};
1311
1312        match build_agent_runtime(merged_agents).await {
1313            Some(runtime) => {
1314                // Activate dispatch idempotency when Redis is reachable for the
1315                // ledger. Best-effort: a connect failure disables idempotency
1316                // but never blocks serving.
1317                let (ledger, ledger_active): (Arc<dyn DispatchLedger>, bool) =
1318                    match std::env::var("GREENTIC_AW_REDIS_URL") {
1319                        Ok(url) if !url.is_empty() => {
1320                            match RedisAgentStateStore::connect(&url).await {
1321                                Ok(store) => {
1322                                    (Arc::new(RedisDispatchLedger::new(store.manager())), true)
1323                                }
1324                                Err(error) => {
1325                                    tracing::warn!(
1326                                        %error,
1327                                        "dispatch ledger Redis connect failed; \
1328                                         idempotency disabled"
1329                                    );
1330                                    (Arc::new(NoopDispatchLedger), false)
1331                                }
1332                            }
1333                        }
1334                        _ => (Arc::new(NoopDispatchLedger), false),
1335                    };
1336
1337                tracing::info!(
1338                    nats_url,
1339                    dispatch_ledger_active = ledger_active,
1340                    "agentic serve mode starting"
1341                );
1342                greentic_aw_runtime::serve::serve_with_ledger(nats_url, runtime, ledger).await
1343            }
1344            None => {
1345                tracing::info!(
1346                    "agentic serve mode skipped: no agentic runtime could be constructed"
1347                );
1348                Ok(())
1349            }
1350        }
1351    }
1352
1353    /// Load process-level base agent configs from the manifests directory.
1354    ///
1355    /// Reads every `<agent_id>.json` file in [`manifests_discovery_dir`] as a
1356    /// full [`AgentConfig`] (NOT the tool-only Digital Worker manifest consumed
1357    /// by [`ManifestToolOverlayProvider`]). This is the ONLY process-level base
1358    /// agent source: pack-embedded agents and `HostConfig::agents` are both
1359    /// per-tenant and only materialise inside `TenantRuntime::from_packs`, so an
1360    /// in-process serve started at process startup cannot see them.
1361    ///
1362    /// Returns an empty map when the directory is absent or unreadable. Files
1363    /// that fail to decode into an [`AgentConfig`], or whose `agent_id` does not
1364    /// match the file stem, are logged and skipped so one malformed file never
1365    /// aborts loading. The file stem is the authoritative key (the in-map id is
1366    /// taken from the stem), mirroring the `<agent_id>.json` convention.
1367    pub fn load_process_agent_configs() -> HashMap<String, AgentConfig> {
1368        let dir = manifests_discovery_dir();
1369        let entries = match std::fs::read_dir(&dir) {
1370            Ok(entries) => entries,
1371            Err(error) => {
1372                tracing::debug!(
1373                    dir = %dir.display(),
1374                    error = %error,
1375                    "agent manifests dir not readable; no process-level agents loaded"
1376                );
1377                return HashMap::new();
1378            }
1379        };
1380
1381        let mut agents: HashMap<String, AgentConfig> = HashMap::new();
1382        for entry in entries.flatten() {
1383            let path = entry.path();
1384            let is_json = path
1385                .extension()
1386                .and_then(|ext| ext.to_str())
1387                .is_some_and(|ext| ext.eq_ignore_ascii_case("json"));
1388            if !is_json {
1389                continue;
1390            }
1391            let Some(stem) = path.file_stem().and_then(|stem| stem.to_str()) else {
1392                continue;
1393            };
1394            let bytes = match std::fs::read(&path) {
1395                Ok(bytes) => bytes,
1396                Err(error) => {
1397                    tracing::warn!(path = %path.display(), error = %error, "agent config read failed; skipping");
1398                    continue;
1399                }
1400            };
1401            match serde_json::from_slice::<AgentConfig>(&bytes) {
1402                Ok(config) => {
1403                    if config.agent_id != stem {
1404                        tracing::warn!(
1405                            file_stem = stem,
1406                            agent_id = config.agent_id.as_str(),
1407                            "agent config id does not match filename; keying by filename"
1408                        );
1409                    }
1410                    agents.insert(stem.to_string(), config);
1411                }
1412                Err(error) => {
1413                    tracing::warn!(path = %path.display(), error = %error, "agent config decode failed; skipping");
1414                }
1415            }
1416        }
1417        agents
1418    }
1419
1420    #[cfg(test)]
1421    mod tests {
1422        use std::collections::HashMap;
1423        use std::sync::Arc;
1424
1425        use greentic_aw_runtime::cost::MockTokenMeter;
1426        use greentic_aw_runtime::llm::LlmResponse;
1427        use greentic_aw_runtime::mock::{
1428            MockAgentStateStore, MockConfigProvider, MockLlmBackend, MockTelemetry, NoopToolLedger,
1429        };
1430        use greentic_aw_runtime::{AgentConfig, AgentLimits, LlmProviderRef};
1431        use greentic_aw_runtime::{AgentRuntime, TenantContext};
1432        use serde_json::json;
1433
1434        use super::*;
1435
1436        fn sample_agent_config(agent_id: &str) -> AgentConfig {
1437            AgentConfig {
1438                agent_id: agent_id.into(),
1439                system_prompt: "sys".into(),
1440                tools: vec![],
1441                guardrails: vec![],
1442                llm: LlmProviderRef {
1443                    provider: "openai".into(),
1444                    model: "gpt-4o-mini".into(),
1445                    credential_ref: None,
1446                },
1447                limits: AgentLimits::default(),
1448                memory: None,
1449                knowledge: None,
1450            }
1451        }
1452
1453        #[tokio::test]
1454        async fn execute_returns_reply_json() {
1455            let llm = Arc::new(MockLlmBackend::new(vec![Ok(LlmResponse {
1456                content: Some("pong".into()),
1457                tool_calls: vec![],
1458                tokens_in: 1,
1459                tokens_out: 1,
1460            })]));
1461            let store = Arc::new(MockAgentStateStore::new());
1462            let telemetry = Arc::new(MockTelemetry::new());
1463
1464            let config_provider = MockConfigProvider::new();
1465            let tenant = TenantContext::new("t", "e");
1466            config_provider.insert(
1467                &tenant,
1468                "greeter",
1469                AgentConfig {
1470                    agent_id: "greeter".into(),
1471                    system_prompt: "sys".into(),
1472                    tools: vec![],
1473                    guardrails: vec![],
1474                    llm: LlmProviderRef {
1475                        provider: "mock".into(),
1476                        model: "m".into(),
1477                        credential_ref: None,
1478                    },
1479                    limits: AgentLimits::default(),
1480                    memory: None,
1481                    knowledge: None,
1482                },
1483            );
1484            let config_provider = Arc::new(config_provider);
1485
1486            let token_meter = Arc::new(MockTokenMeter::new(0));
1487            let ledger = Arc::new(NoopToolLedger);
1488            let ext_runtime = Arc::new(crate::runner::agent_node::test_extension_runtime());
1489
1490            let runtime = Arc::new(AgentRuntime::new(
1491                config_provider,
1492                store,
1493                ext_runtime,
1494                llm,
1495                telemetry,
1496                token_meter,
1497                ledger,
1498                None,
1499            ));
1500            let handler = RuntimeAgentNodeHandler::new(runtime, None, None);
1501
1502            let output = handler
1503                .execute("t", "e", "greeter", "sess-1", &json!({"user_text": "ping"}))
1504                .await
1505                .expect("execute should succeed");
1506
1507            assert_eq!(output["reply"].as_str(), Some("pong"));
1508        }
1509
1510        // -------------------------------------------------------------------
1511        // `project_id` billing dimension = the deployed unit's bundle id
1512        // -------------------------------------------------------------------
1513
1514        /// Captures the `TenantContext.project_id` of every billing emit the
1515        /// agent loop makes, so a test can assert what `execute` actually
1516        /// handed the sink rather than what it looks like it should have.
1517        #[derive(Default)]
1518        struct RecordingBillingMeter {
1519            project_ids: std::sync::Mutex<Vec<Option<String>>>,
1520        }
1521
1522        impl greentic_aw_runtime::billing::BillingMeter for RecordingBillingMeter {
1523            fn emit<'a>(
1524                &'a self,
1525                tenant: &'a TenantContext,
1526                _input_tokens: u64,
1527                _output_tokens: u64,
1528                _agent_id: &'a str,
1529                _model: &'a str,
1530            ) -> std::pin::Pin<
1531                Box<
1532                    dyn std::future::Future<
1533                            Output = Result<(), greentic_aw_runtime::billing::BillingError>,
1534                        > + Send
1535                        + 'a,
1536                >,
1537            > {
1538                self.project_ids
1539                    .lock()
1540                    .expect("recording meter lock poisoned")
1541                    .push(tenant.project_id.clone());
1542                Box::pin(async { Ok(()) })
1543            }
1544
1545            fn over_budget<'a>(
1546                &'a self,
1547                _tenant: &'a TenantContext,
1548            ) -> std::pin::Pin<Box<dyn std::future::Future<Output = bool> + Send + 'a>>
1549            {
1550                Box::pin(std::future::ready(false))
1551            }
1552        }
1553
1554        /// Run one `execute` through a handler built with `project_id` and
1555        /// return the project ids the billing sink observed.
1556        async fn project_ids_seen_by_billing(project_id: Option<String>) -> Vec<Option<String>> {
1557            let llm = Arc::new(MockLlmBackend::new(vec![Ok(LlmResponse {
1558                content: Some("pong".into()),
1559                tool_calls: vec![],
1560                tokens_in: 1,
1561                tokens_out: 1,
1562            })]));
1563            let config_provider = MockConfigProvider::new();
1564            config_provider.insert(
1565                &TenantContext::new("t", "e"),
1566                "greeter",
1567                sample_agent_config("greeter"),
1568            );
1569            let meter = Arc::new(RecordingBillingMeter::default());
1570            let runtime = Arc::new(
1571                AgentRuntime::new(
1572                    Arc::new(config_provider),
1573                    Arc::new(MockAgentStateStore::new()),
1574                    Arc::new(crate::runner::agent_node::test_extension_runtime()),
1575                    llm,
1576                    Arc::new(MockTelemetry::new()),
1577                    Arc::new(MockTokenMeter::new(0)),
1578                    Arc::new(NoopToolLedger),
1579                    None,
1580                )
1581                .with_billing_meter(meter.clone()),
1582            );
1583
1584            RuntimeAgentNodeHandler::new(runtime, None, project_id)
1585                .execute("t", "e", "greeter", "sess-1", &json!({"user_text": "ping"}))
1586                .await
1587                .expect("execute should succeed");
1588
1589            let seen = meter
1590                .project_ids
1591                .lock()
1592                .expect("recording meter lock poisoned");
1593            seen.clone()
1594        }
1595
1596        #[tokio::test]
1597        async fn execute_bills_the_pack_identity_as_project_id() {
1598            let seen = project_ids_seen_by_billing(Some("customer.support".into())).await;
1599            assert_eq!(
1600                seen,
1601                vec![Some("customer.support".to_string())],
1602                "the deployed bundle id must reach the billing sink, not the in-pack agent id"
1603            );
1604        }
1605
1606        #[tokio::test]
1607        async fn execute_bills_no_project_id_when_the_pack_identity_is_unknown() {
1608            let seen = project_ids_seen_by_billing(None).await;
1609            assert_eq!(
1610                seen,
1611                vec![None],
1612                "an unknown pack identity must stay absent — never fall back to the agent id"
1613            );
1614        }
1615
1616        // -------------------------------------------------------------------
1617        // Billing wiring on the in-process `dw.agent` path
1618        // -------------------------------------------------------------------
1619
1620        /// The in-process `dw.agent` runtime must talk to the configured
1621        /// billing service, exactly as the out-of-process serve path does.
1622        ///
1623        /// Asserted through observable behaviour rather than by inspecting the
1624        /// runtime: the agent loop consults `BillingMeter::over_budget` before
1625        /// it touches the LLM or the state store, and the HTTP sink implements
1626        /// that as `GET /v1/tenants/{tenant}/wallet`. So a request arriving at
1627        /// the billing server proves a real meter is installed; with the
1628        /// default `NoopBillingMeter` nothing is ever sent.
1629        #[tokio::test]
1630        #[serial_test::serial]
1631        // `set_var`/`remove_var` are process-global; `serial` keeps them from
1632        // racing other env-reading tests.
1633        #[allow(unsafe_code)]
1634        async fn in_process_handler_consults_the_configured_billing_service() {
1635            let server = wiremock::MockServer::start().await;
1636            wiremock::Mock::given(wiremock::matchers::method("GET"))
1637                .and(wiremock::matchers::path("/v1/tenants/acme/wallet"))
1638                // `available: 0` short-circuits the step at the credit gate, so
1639                // the test never reaches the LLM backend or acquires a lock.
1640                .respond_with(
1641                    wiremock::ResponseTemplate::new(200)
1642                        .set_body_json(serde_json::json!({"available": "0"})),
1643                )
1644                .mount(&server)
1645                .await;
1646
1647            // Pin extension discovery at an empty dir: without this the runtime
1648            // scans the developer's real `~/.greentic/extensions`, which makes
1649            // the test depend on machine state and cost ~45s locally.
1650            let empty_extensions = tempfile::tempdir().expect("tempdir");
1651            unsafe {
1652                std::env::set_var("GREENTIC_BILLING_BASE_URL", server.uri());
1653                std::env::set_var("GREENTIC_BILLING_SERVICE_SECRET", "secret");
1654                std::env::set_var("GREENTIC_EXTENSIONS_DIR", empty_extensions.path());
1655            }
1656
1657            let mut agents = HashMap::new();
1658            agents.insert("greeter".to_string(), sample_agent_config("greeter"));
1659            let handler = build_runtime_handler_with_stores(
1660                agents,
1661                "acme".to_string(),
1662                crate::secrets::default_manager().expect("env secrets manager"),
1663                vec![],
1664                Arc::new(MockAgentStateStore::new()),
1665                Arc::new(MockTokenMeter::new(0)),
1666                Arc::new(NoopToolLedger),
1667                None,
1668                None,
1669            )
1670            .await
1671            .expect("handler should build from mock stores");
1672
1673            let _ = handler
1674                .execute("acme", "prod", "greeter", "s", &json!({"user_text": "hi"}))
1675                .await;
1676
1677            unsafe {
1678                std::env::remove_var("GREENTIC_BILLING_BASE_URL");
1679                std::env::remove_var("GREENTIC_BILLING_SERVICE_SECRET");
1680                std::env::remove_var("GREENTIC_EXTENSIONS_DIR");
1681            }
1682
1683            let wallet_calls = server
1684                .received_requests()
1685                .await
1686                .unwrap_or_default()
1687                .iter()
1688                .filter(|r| r.url.path() == "/v1/tenants/acme/wallet")
1689                .count();
1690            assert!(
1691                wallet_calls >= 1,
1692                "in-process dw.agent must consult the billing service; got {wallet_calls} \
1693                 wallet requests, which means it is still running on NoopBillingMeter \
1694                 and its LLM spend is never metered"
1695            );
1696        }
1697
1698        // -----------------------------------------------------------------------
1699        // audit_sink enable/disable branch (EPIC-B B-3)
1700        // -----------------------------------------------------------------------
1701
1702        /// Build an [`AgentRuntime`] scripted to make one `remember` (host
1703        /// built-in short-term-memory) tool call before replying "done". The
1704        /// host built-in path fires `StepObserver::on_tool_call`/`on_tool_result`
1705        /// without needing a real WASM extension dispatch, so it is the
1706        /// cheapest way to drive a genuine tool call through `execute`.
1707        fn runtime_with_scripted_remember_call(tenant_id: &str, env_id: &str) -> Arc<AgentRuntime> {
1708            use greentic_aw_runtime::state::ToolCallRecord;
1709            use greentic_aw_runtime::{InMemoryMemoryProvider, MemoryProviderRef, MemorySettings};
1710
1711            let llm = Arc::new(MockLlmBackend::new(vec![
1712                Ok(LlmResponse {
1713                    content: None,
1714                    tool_calls: vec![ToolCallRecord {
1715                        call_id: "c1".into(),
1716                        extension_id: "host".into(),
1717                        tool_name: "remember".into(),
1718                        args: json!({"key": "k", "value": "v"}),
1719                    }],
1720                    tokens_in: 1,
1721                    tokens_out: 1,
1722                }),
1723                Ok(LlmResponse {
1724                    content: Some("done".into()),
1725                    tool_calls: vec![],
1726                    tokens_in: 1,
1727                    tokens_out: 1,
1728                }),
1729            ]));
1730            let store = Arc::new(MockAgentStateStore::new());
1731            let telemetry = Arc::new(MockTelemetry::new());
1732
1733            let config_provider = MockConfigProvider::new();
1734            let tenant = TenantContext::new(tenant_id, env_id);
1735            let mut cfg = sample_agent_config("greeter");
1736            cfg.memory = Some(MemorySettings {
1737                short_term: Some(MemoryProviderRef {
1738                    provider: "inmemory".into(),
1739                    capability: "cap://memory/short-term".into(),
1740                    params: Default::default(),
1741                    credential_ref: None,
1742                }),
1743                long_term: None,
1744            });
1745            config_provider.insert(&tenant, "greeter", cfg);
1746            let config_provider = Arc::new(config_provider);
1747
1748            let token_meter = Arc::new(MockTokenMeter::new(0));
1749            let ledger = Arc::new(NoopToolLedger);
1750            let ext_runtime = Arc::new(crate::runner::agent_node::test_extension_runtime());
1751
1752            Arc::new(
1753                AgentRuntime::new(
1754                    config_provider,
1755                    store,
1756                    ext_runtime,
1757                    llm,
1758                    telemetry,
1759                    token_meter,
1760                    ledger,
1761                    None,
1762                )
1763                .with_short_term_memory(Arc::new(InMemoryMemoryProvider::new())),
1764            )
1765        }
1766
1767        #[tokio::test]
1768        async fn execute_with_audit_sink_routes_through_step_with_observer_and_enqueues_events() {
1769            let runtime = runtime_with_scripted_remember_call("t1", "e1");
1770
1771            let (tx, mut rx) = tokio::sync::mpsc::channel(16);
1772            let sink = AuditSink::from_sender(tx);
1773            let handler = RuntimeAgentNodeHandler::new(runtime, Some(sink), None);
1774
1775            let output = handler
1776                .execute(
1777                    "t1",
1778                    "e1",
1779                    "greeter",
1780                    "sess-1",
1781                    &json!({"user_text": "remember this"}),
1782                )
1783                .await
1784                .expect("execute should succeed");
1785            assert_eq!(output["reply"].as_str(), Some("done"));
1786
1787            let (subject, bytes) = rx.try_recv().expect("tool_call event enqueued");
1788            assert_eq!(subject, "audit.t1.agent.tool_call");
1789            let value: Value = serde_json::from_slice(&bytes).expect("valid JSON");
1790            assert_eq!(value["payload"]["tool"], json!("remember"));
1791            assert_eq!(value["payload"]["agent_id"], json!("greeter"));
1792
1793            let (subject, bytes) = rx.try_recv().expect("tool_result event enqueued");
1794            assert_eq!(subject, "audit.t1.agent.tool_result");
1795            let value: Value = serde_json::from_slice(&bytes).expect("valid JSON");
1796            assert_eq!(value["payload"]["tool"], json!("remember"));
1797
1798            assert!(
1799                rx.try_recv().is_err(),
1800                "exactly two audit events enqueued (one tool_call, one tool_result)"
1801            );
1802        }
1803
1804        #[tokio::test]
1805        async fn execute_without_audit_sink_uses_plain_step_path_unchanged() {
1806            // Same scripted tool call as the audited test above, but the
1807            // handler carries no audit sink at all — proves the "off" branch
1808            // (self.runtime.step, no observer constructed) still dispatches
1809            // the tool call and returns the same reply, exactly as it did
1810            // before AgentAuditObserver existed.
1811            let runtime = runtime_with_scripted_remember_call("t1", "e1");
1812            let handler = RuntimeAgentNodeHandler::new(runtime, None, None);
1813
1814            let output = handler
1815                .execute(
1816                    "t1",
1817                    "e1",
1818                    "greeter",
1819                    "sess-1",
1820                    &json!({"user_text": "remember this"}),
1821                )
1822                .await
1823                .expect("execute should succeed");
1824            assert_eq!(output["reply"].as_str(), Some("done"));
1825        }
1826
1827        #[test]
1828        fn tenant_ctx_for_audit_uses_real_ids_when_valid() {
1829            let ctx = super::tenant_ctx_for_audit("acme", "prod");
1830            assert_eq!(ctx.tenant.as_str(), "acme");
1831            assert_eq!(ctx.env.as_str(), "prod");
1832        }
1833
1834        #[test]
1835        fn tenant_ctx_for_audit_falls_back_to_local_on_invalid_ids() {
1836            // Empty strings fail the newtype validation; the helper must not
1837            // panic and should fall back to "local" for both fields.
1838            let ctx = super::tenant_ctx_for_audit("", "");
1839            assert_eq!(ctx.tenant.as_str(), "local");
1840            assert_eq!(ctx.env.as_str(), "local");
1841        }
1842
1843        #[tokio::test]
1844        async fn host_config_provider_returns_config_for_known_agent() {
1845            let mut agents = HashMap::new();
1846            agents.insert("greeter".to_string(), sample_agent_config("greeter"));
1847            let provider = HostConfigProvider::new(agents);
1848
1849            let tenant = TenantContext::new("acme", "prod");
1850            let resolved = provider
1851                .agent_config(&tenant, "greeter")
1852                .await
1853                .expect("known agent resolves");
1854
1855            assert_eq!(resolved.agent_id, "greeter");
1856        }
1857
1858        /// Requires `greentic_dw_manifest_tools`: without it this lane's
1859        /// `DigitalWorkerManifest` has no `extension_tools` to parse, so
1860        /// `manifest_to_tool_refs` yields an empty overlay and there is
1861        /// nothing to replace. `a_tool_declaring_manifest_does_not_reach_the_agent`
1862        /// below pins what happens here instead.
1863        #[cfg(greentic_dw_manifest_tools)]
1864        #[tokio::test]
1865        async fn overlay_provider_replaces_tools_from_manifest() {
1866            use greentic_aw_runtime::ManifestToolOverlayProvider;
1867            use greentic_aw_runtime::config::ToolRef;
1868            use greentic_aw_runtime::config_provider::ConfigProvider;
1869
1870            let tmp = tempfile::tempdir().unwrap();
1871            std::fs::write(
1872                tmp.path().join("greeter.json"),
1873                r#"{"id":"greeter","display_name":"G",
1874                "tenancy":{"tenant":"t","team_policy":"disabled"},
1875                "locale":{"worker_default_locale":"en-US","policy":"worker_default",
1876                          "propagation":"current_task_only","output":"worker_default"},
1877                "extension_tools":[{"extension_id":"greentic.tavily","extension_version":"1.0.0",
1878                  "tool_name":"web_search","description":"d","input_schema_json":"{\"type\":\"object\"}",
1879                  "capabilities":["agentic_worker"],"agentic_worker_metadata":{}}]}"#,
1880            )
1881            .unwrap();
1882
1883            let mut agents = HashMap::new();
1884            agents.insert("greeter".to_string(), sample_agent_config("greeter"));
1885            let provider = ManifestToolOverlayProvider::new(
1886                HostConfigProvider::new(agents),
1887                tmp.path().to_path_buf(),
1888            );
1889
1890            let tenant = TenantContext::new("acme", "prod");
1891            let cfg = provider.agent_config(&tenant, "greeter").await.unwrap();
1892            assert_eq!(
1893                cfg.tools,
1894                vec![ToolRef {
1895                    extension_id: "greentic.tavily".into(),
1896                    tool_name: "web_search".into()
1897                }]
1898            );
1899        }
1900
1901        /// The agent-side mirror of
1902        /// `manifest_provider::tests::a_tool_declaring_manifest_is_ignored_on_this_lane`:
1903        /// a Digital Worker manifest may declare agentic-worker tools, but on
1904        /// this lane none of them reach the agent's config. The overlay is
1905        /// fail-soft, so this is silent — pin it so a future port has to
1906        /// delete this test rather than discover the behaviour.
1907        #[cfg(not(greentic_dw_manifest_tools))]
1908        #[tokio::test]
1909        async fn a_tool_declaring_manifest_does_not_reach_the_agent() {
1910            use greentic_aw_runtime::ManifestToolOverlayProvider;
1911            use greentic_aw_runtime::config_provider::ConfigProvider;
1912
1913            let tmp = tempfile::tempdir().unwrap();
1914            std::fs::write(
1915                tmp.path().join("greeter.json"),
1916                r#"{"id":"greeter","display_name":"G",
1917                "tenancy":{"tenant":"t","team_policy":"disabled"},
1918                "locale":{"worker_default_locale":"en-US","policy":"worker_default",
1919                          "propagation":"current_task_only","output":"worker_default"},
1920                "extension_tools":[{"extension_id":"greentic.tavily","extension_version":"1.0.0",
1921                  "tool_name":"web_search","description":"d","input_schema_json":"{\"type\":\"object\"}",
1922                  "capabilities":["agentic_worker"],"agentic_worker_metadata":{}}]}"#,
1923            )
1924            .unwrap();
1925
1926            let mut agents = HashMap::new();
1927            agents.insert("greeter".to_string(), sample_agent_config("greeter"));
1928            let base_tools = sample_agent_config("greeter").tools;
1929            let provider = ManifestToolOverlayProvider::new(
1930                HostConfigProvider::new(agents),
1931                tmp.path().to_path_buf(),
1932            );
1933
1934            let tenant = TenantContext::new("acme", "prod");
1935            let cfg = provider.agent_config(&tenant, "greeter").await.unwrap();
1936            assert_eq!(
1937                cfg.tools, base_tools,
1938                "the manifest's greentic.tavily/web_search must not reach the agent on this lane"
1939            );
1940        }
1941
1942        #[test]
1943        fn bridge_credential_defaults_provider_and_model() {
1944            let c = super::bridge_credential(None, None, "sk-x".into(), None).unwrap();
1945            assert_eq!(c.provider, "openai");
1946            assert_eq!(c.model, "gpt-4o");
1947            assert_eq!(c.api_key, "sk-x");
1948            assert!(c.base_url.is_none());
1949        }
1950
1951        #[test]
1952        fn bridge_credential_honors_explicit_parts() {
1953            let c = super::bridge_credential(
1954                Some("anthropic".into()),
1955                Some("claude-x".into()),
1956                "sk-ant".into(),
1957                Some("https://proxy".into()),
1958            )
1959            .unwrap();
1960            assert_eq!(c.provider, "anthropic");
1961            assert_eq!(c.model, "claude-x");
1962            assert_eq!(c.base_url.as_deref(), Some("https://proxy"));
1963        }
1964
1965        #[test]
1966        fn bridge_credential_none_without_key() {
1967            assert!(
1968                super::bridge_credential(Some("openai".into()), None, "  ".into(), None).is_none()
1969            );
1970        }
1971
1972        #[tokio::test]
1973        async fn host_config_provider_returns_not_found_for_unknown_agent() {
1974            use greentic_aw_runtime::error::ConfigError;
1975
1976            let provider = HostConfigProvider::new(HashMap::new());
1977
1978            let tenant = TenantContext::new("acme", "prod");
1979            let result = provider.agent_config(&tenant, "missing").await;
1980
1981            assert!(matches!(result, Err(ConfigError::AgentNotFound(_))));
1982        }
1983
1984        // -----------------------------------------------------------------------
1985        // merge_agent_sources tests
1986        // -----------------------------------------------------------------------
1987
1988        #[test]
1989        fn merge_pack_only_agent_resolves() {
1990            let mut pack_agents = HashMap::new();
1991            pack_agents.insert("pack-bot".to_string(), sample_agent_config("pack-bot"));
1992
1993            let merged = super::merge_agent_sources(pack_agents, HashMap::new());
1994
1995            assert!(merged.contains_key("pack-bot"));
1996            assert_eq!(merged["pack-bot"].agent_id, "pack-bot");
1997        }
1998
1999        #[test]
2000        fn merge_operator_only_agent_resolves() {
2001            let mut operator_agents = HashMap::new();
2002            operator_agents.insert("op-bot".to_string(), sample_agent_config("op-bot"));
2003
2004            let merged = super::merge_agent_sources(HashMap::new(), operator_agents);
2005
2006            assert!(merged.contains_key("op-bot"));
2007            assert_eq!(merged["op-bot"].agent_id, "op-bot");
2008        }
2009
2010        #[test]
2011        fn merge_operator_wins_on_collision() {
2012            let mut pack_agents = HashMap::new();
2013            let mut pack_config = sample_agent_config("shared-bot");
2014            pack_config.system_prompt = "pack prompt".to_string();
2015            pack_agents.insert("shared-bot".to_string(), pack_config);
2016
2017            let mut operator_agents = HashMap::new();
2018            let mut operator_config = sample_agent_config("shared-bot");
2019            operator_config.system_prompt = "operator prompt".to_string();
2020            operator_agents.insert("shared-bot".to_string(), operator_config);
2021
2022            let merged = super::merge_agent_sources(pack_agents, operator_agents);
2023
2024            assert_eq!(merged.len(), 1);
2025            assert_eq!(
2026                merged["shared-bot"].system_prompt, "operator prompt",
2027                "operator config must override pack config on agent_id collision"
2028            );
2029        }
2030
2031        // -----------------------------------------------------------------------
2032        // agent_configs_from_manifest tests
2033        // -----------------------------------------------------------------------
2034
2035        #[test]
2036        fn deserialize_agent_blob_produces_correct_config() {
2037            let blob = serde_json::json!({
2038                "agent_id": "demo-agent",
2039                "system_prompt": "You are helpful.",
2040                "tools": [],
2041                "llm": {
2042                    "provider": "openai",
2043                    "model": "gpt-4o-mini"
2044                },
2045                "limits": {
2046                    "max_iter": 5,
2047                    "timeout": 30,
2048                    "max_history_turns": 10,
2049                    "llm_retry_attempts": 2,
2050                    "llm_retry_backoff": 500,
2051                    "provider_failure_message": null,
2052                    "daily_token_cap_per_tenant": null
2053                }
2054            });
2055
2056            let config: AgentConfig =
2057                serde_json::from_value(blob).expect("valid blob must deserialize");
2058
2059            assert_eq!(config.agent_id, "demo-agent");
2060            assert_eq!(config.system_prompt, "You are helpful.");
2061            assert_eq!(config.limits.max_iter, 5);
2062            assert_eq!(config.limits.timeout, std::time::Duration::from_secs(30));
2063        }
2064
2065        #[test]
2066        fn agent_configs_from_manifest_skips_malformed_blobs() {
2067            use std::collections::BTreeMap;
2068
2069            let mut blobs: BTreeMap<String, serde_json::Value> = BTreeMap::new();
2070
2071            // Valid agent blob
2072            blobs.insert(
2073                "good-agent".to_string(),
2074                serde_json::json!({
2075                    "agent_id": "good-agent",
2076                    "system_prompt": "Valid.",
2077                    "tools": [],
2078                    "llm": { "provider": "openai", "model": "gpt-4o-mini" },
2079                    "limits": {
2080                        "max_iter": 8,
2081                        "timeout": 60,
2082                        "max_history_turns": 20,
2083                        "llm_retry_attempts": 3,
2084                        "llm_retry_backoff": 250,
2085                        "provider_failure_message": null,
2086                        "daily_token_cap_per_tenant": null
2087                    }
2088                }),
2089            );
2090
2091            // Malformed blob (missing required fields)
2092            blobs.insert(
2093                "bad-agent".to_string(),
2094                serde_json::json!({ "broken": true }),
2095            );
2096
2097            let configs = super::agent_configs_from_manifest("test-pack", &blobs);
2098
2099            assert_eq!(configs.len(), 1, "malformed blob must be skipped");
2100            assert!(configs.contains_key("good-agent"));
2101            assert!(!configs.contains_key("bad-agent"));
2102        }
2103
2104        #[test]
2105        #[serial_test::serial]
2106        #[allow(unsafe_code)]
2107        fn registry_from_env_requires_both_vars() {
2108            // SAFETY: #[serial] serializes env-mutating tests (crate convention),
2109            // so no concurrent test observes a torn env; vars cleaned up at the end.
2110            unsafe {
2111                std::env::remove_var("GREENTIC_AW_ADMIN_ENDPOINT");
2112                std::env::remove_var("GREENTIC_AW_ADMIN_TOKEN");
2113            }
2114            assert!(super::registry_from_env().is_none());
2115
2116            unsafe {
2117                std::env::set_var("GREENTIC_AW_ADMIN_ENDPOINT", "http://localhost:9999");
2118            }
2119            assert!(
2120                super::registry_from_env().is_none(),
2121                "endpoint alone is not enough"
2122            );
2123
2124            unsafe {
2125                std::env::set_var("GREENTIC_AW_ADMIN_TOKEN", "gtc_live_x");
2126            }
2127            assert!(super::registry_from_env().is_some());
2128
2129            unsafe {
2130                std::env::remove_var("GREENTIC_AW_ADMIN_ENDPOINT");
2131                std::env::remove_var("GREENTIC_AW_ADMIN_TOKEN");
2132            }
2133        }
2134
2135        #[test]
2136        #[serial_test::serial]
2137        #[allow(unsafe_code)]
2138        fn guardrail_policy_from_env_requires_both_vars() {
2139            // SAFETY: #[serial] serializes env-mutating tests (crate convention),
2140            // so no concurrent test observes a torn env; vars cleaned up at the end.
2141            unsafe {
2142                std::env::remove_var("GREENTIC_AW_ADMIN_ENDPOINT");
2143                std::env::remove_var("GREENTIC_AW_ADMIN_TOKEN");
2144            }
2145            assert!(super::guardrail_policy_from_env().is_none());
2146
2147            unsafe {
2148                std::env::set_var("GREENTIC_AW_ADMIN_ENDPOINT", "http://localhost:9999");
2149            }
2150            assert!(
2151                super::guardrail_policy_from_env().is_none(),
2152                "endpoint alone is not enough"
2153            );
2154
2155            unsafe {
2156                std::env::set_var("GREENTIC_AW_ADMIN_TOKEN", "gtc_live_x");
2157            }
2158            assert!(super::guardrail_policy_from_env().is_some());
2159
2160            unsafe {
2161                std::env::remove_var("GREENTIC_AW_ADMIN_ENDPOINT");
2162                std::env::remove_var("GREENTIC_AW_ADMIN_TOKEN");
2163            }
2164        }
2165
2166        #[test]
2167        #[serial_test::serial]
2168        #[allow(unsafe_code)]
2169        fn mcp_source_from_env_default_on_with_opt_out() {
2170            // SAFETY: #[serial] serializes env-mutating tests (crate convention),
2171            // so no concurrent test observes a torn env; vars cleaned up at the end.
2172            unsafe {
2173                std::env::remove_var("GREENTIC_AW_MCP");
2174                std::env::remove_var("GREENTIC_AW_ADMIN_ENDPOINT");
2175                std::env::remove_var("GREENTIC_AW_ADMIN_TOKEN");
2176            }
2177
2178            // (a) Default-on: endpoint + token present, gate unset → Some.
2179            unsafe {
2180                std::env::set_var("GREENTIC_AW_ADMIN_ENDPOINT", "http://localhost:9999");
2181                std::env::set_var("GREENTIC_AW_ADMIN_TOKEN", "gtc_live_x");
2182            }
2183            assert!(
2184                super::mcp_source_from_env().is_some(),
2185                "MCP is on by default when admin credentials are configured"
2186            );
2187
2188            // (b) Explicit opt-out wins even with full credentials.
2189            unsafe {
2190                std::env::set_var("GREENTIC_AW_MCP", "0");
2191            }
2192            assert!(
2193                super::mcp_source_from_env().is_none(),
2194                "GREENTIC_AW_MCP=0 disables MCP regardless of credentials"
2195            );
2196
2197            // (b') Legacy opt-in value still enables (any non-"0" value does).
2198            unsafe {
2199                std::env::set_var("GREENTIC_AW_MCP", "1");
2200            }
2201            assert!(super::mcp_source_from_env().is_some());
2202
2203            // (c) Missing credential → None even without an opt-out.
2204            unsafe {
2205                std::env::remove_var("GREENTIC_AW_MCP");
2206                std::env::remove_var("GREENTIC_AW_ADMIN_ENDPOINT");
2207            }
2208            assert!(
2209                super::mcp_source_from_env().is_none(),
2210                "no endpoint → no MCP source"
2211            );
2212
2213            unsafe {
2214                std::env::set_var("GREENTIC_AW_ADMIN_ENDPOINT", "http://localhost:9999");
2215                std::env::remove_var("GREENTIC_AW_ADMIN_TOKEN");
2216            }
2217            assert!(
2218                super::mcp_source_from_env().is_none(),
2219                "no token → no MCP source"
2220            );
2221
2222            unsafe {
2223                std::env::remove_var("GREENTIC_AW_MCP");
2224                std::env::remove_var("GREENTIC_AW_ADMIN_ENDPOINT");
2225                std::env::remove_var("GREENTIC_AW_ADMIN_TOKEN");
2226            }
2227        }
2228
2229        // -----------------------------------------------------------------------
2230        // guardrail_denied_json tests
2231        // -----------------------------------------------------------------------
2232
2233        #[test]
2234        fn guardrail_denied_maps_to_structured_output() {
2235            use greentic_aw_runtime::guardrail::GuardrailDirection;
2236            let v = super::guardrail_denied_json(
2237                GuardrailDirection::Inbound,
2238                "permission_denied",
2239                "blocked",
2240                None,
2241            );
2242            assert_eq!(v["guardrail"]["blocked"], serde_json::json!(true));
2243            assert_eq!(v["guardrail"]["direction"], serde_json::json!("inbound"));
2244            assert_eq!(
2245                v["guardrail"]["code"],
2246                serde_json::json!("permission_denied")
2247            );
2248            assert_eq!(v["terminated_by"], serde_json::json!("guardrail_denied"));
2249            assert_eq!(v["reply"], serde_json::json!("blocked"));
2250        }
2251
2252        #[cfg(feature = "desktop-agent-ephemeral")]
2253        #[tokio::test]
2254        #[allow(unsafe_code)]
2255        async fn ephemeral_builder_yields_handler_without_redis() {
2256            // Remove Redis URL so we prove the ephemeral builder does not need it.
2257            // SAFETY: single-threaded test; no concurrent env mutation.
2258            unsafe {
2259                std::env::remove_var("GREENTIC_AW_REDIS_URL");
2260            }
2261            let mut agents = HashMap::new();
2262            agents.insert("greeter".to_string(), sample_agent_config("greeter"));
2263            // Build a minimal env-backed secrets manager (no broker configured in tests).
2264            let secrets: crate::secrets::DynSecretsManager =
2265                Arc::new(greentic_secrets_lib::env::EnvSecretsManager);
2266            let handler = super::build_agent_node_handler_ephemeral(
2267                agents,
2268                "t1".to_string(),
2269                secrets,
2270                Vec::new(),
2271                None,
2272                // project_id: no pack identity to attribute billing to, matching
2273                // the desktop ephemeral call site in greentic-runner-desktop.
2274                None,
2275            )
2276            .await;
2277            assert!(
2278                handler.is_some(),
2279                "ephemeral builder must not require Redis"
2280            );
2281        }
2282
2283        /// In-memory `SecretsManager` for backend tests: returns seeded values,
2284        /// `NotFound` otherwise.
2285        struct MapSecrets(std::collections::HashMap<String, Vec<u8>>);
2286
2287        #[async_trait::async_trait]
2288        impl greentic_secrets_lib::SecretsManager for MapSecrets {
2289            async fn read(&self, path: &str) -> greentic_secrets_lib::Result<Vec<u8>> {
2290                self.0
2291                    .get(path)
2292                    .cloned()
2293                    .ok_or_else(|| greentic_secrets_lib::SecretError::NotFound(path.to_string()))
2294            }
2295            async fn write(&self, _: &str, _: &[u8]) -> greentic_secrets_lib::Result<()> {
2296                Ok(())
2297            }
2298            async fn delete(&self, _: &str) -> greentic_secrets_lib::Result<()> {
2299                Ok(())
2300            }
2301        }
2302
2303        #[test]
2304        fn store_tool_secret_backend_maps_secret_uri_to_store_scope() {
2305            use greentic_ext_runtime::SecretsBackend as _;
2306            let mut map = std::collections::HashMap::new();
2307            // The injected manager resolves the canonical store scope (in real
2308            // runs its candidate fallback bridges to the pack scope; here we seed
2309            // the canonical path the backend constructs directly).
2310            map.insert(
2311                "secrets://dev/acme/_/tavily/api_key".to_string(),
2312                b"tvly-xyz".to_vec(),
2313            );
2314            let secrets: crate::secrets::DynSecretsManager = Arc::new(MapSecrets(map));
2315            let backend = super::StoreToolSecretsBackend {
2316                secrets,
2317                tenant: "acme".to_string(),
2318                env: "dev".to_string(),
2319            };
2320            let got = backend
2321                .get("secret://tavily/api_key")
2322                .expect("resolve tavily key from store");
2323            assert_eq!(got, "tvly-xyz");
2324        }
2325
2326        #[test]
2327        #[allow(unsafe_code)]
2328        fn store_tool_secret_backend_falls_back_to_env_on_store_miss() {
2329            use greentic_ext_runtime::SecretsBackend as _;
2330            let secrets: crate::secrets::DynSecretsManager =
2331                Arc::new(MapSecrets(std::collections::HashMap::new()));
2332            let backend = super::StoreToolSecretsBackend {
2333                secrets,
2334                tenant: "acme".to_string(),
2335                env: "dev".to_string(),
2336            };
2337            // SAFETY: single-threaded test; no concurrent env mutation.
2338            unsafe {
2339                std::env::set_var("ZEROENV_MYPROVIDER_MYKEY", "from-env");
2340            }
2341            let got = backend
2342                .get("secret://zeroenv_myprovider/mykey")
2343                .expect("env fallback resolves");
2344            assert_eq!(got, "from-env");
2345            unsafe {
2346                std::env::remove_var("ZEROENV_MYPROVIDER_MYKEY");
2347            }
2348        }
2349
2350        #[test]
2351        #[allow(unsafe_code)]
2352        fn load_process_agent_configs_reads_full_configs_and_skips_bad_files() {
2353            let dir = tempfile::tempdir().expect("tempdir");
2354
2355            // Valid full AgentConfig keyed by file stem.
2356            let good = sample_agent_config("greeter");
2357            std::fs::write(
2358                dir.path().join("greeter.json"),
2359                serde_json::to_vec(&good).expect("serialize"),
2360            )
2361            .expect("write good");
2362
2363            // Malformed JSON — skipped, must not abort the load.
2364            std::fs::write(dir.path().join("broken.json"), b"{ not json").expect("write broken");
2365
2366            // Non-JSON file — ignored.
2367            std::fs::write(dir.path().join("README.md"), b"ignore me").expect("write md");
2368
2369            let previous = std::env::var("GREENTIC_AGENT_MANIFESTS_DIR").ok();
2370            unsafe {
2371                std::env::set_var("GREENTIC_AGENT_MANIFESTS_DIR", dir.path());
2372            }
2373            let loaded = super::load_process_agent_configs();
2374            unsafe {
2375                match &previous {
2376                    Some(value) => std::env::set_var("GREENTIC_AGENT_MANIFESTS_DIR", value),
2377                    None => std::env::remove_var("GREENTIC_AGENT_MANIFESTS_DIR"),
2378                }
2379            }
2380
2381            assert_eq!(loaded.len(), 1, "only the valid config should load");
2382            assert!(loaded.contains_key("greeter"));
2383            assert_eq!(loaded["greeter"].agent_id, "greeter");
2384        }
2385
2386        #[test]
2387        fn merge_sidecar_fills_only_missing_keys() {
2388            use std::collections::BTreeMap;
2389            let mut blobs: BTreeMap<String, serde_json::Value> =
2390                BTreeMap::from([("a".to_string(), serde_json::json!({"from": "manifest"}))]);
2391            let sidecar: BTreeMap<String, serde_json::Value> = BTreeMap::from([
2392                ("a".to_string(), serde_json::json!({"from": "sidecar"})), // must NOT override
2393                ("b".to_string(), serde_json::json!({"from": "sidecar"})), // must be added
2394            ]);
2395            super::merge_sidecar_into(&mut blobs, sidecar);
2396            assert_eq!(blobs["a"]["from"], "manifest"); // manifest wins
2397            assert_eq!(blobs["b"]["from"], "sidecar"); // gap filled
2398            assert_eq!(blobs.len(), 2);
2399        }
2400    }
2401}
2402
2403#[allow(clippy::items_after_test_module)] // helper fn + re-exports follow
2404#[cfg(test)]
2405mod gating_tests {
2406    use super::{DwAgentDispatch, dw_agent_dispatch_mode, should_serve_agentic_inproc};
2407    use std::collections::HashMap;
2408
2409    fn env_from(pairs: &[(&str, &str)]) -> impl Fn(&str) -> Option<String> + use<> {
2410        let map: HashMap<String, String> = pairs
2411            .iter()
2412            .map(|(key, value)| (key.to_string(), value.to_string()))
2413            .collect();
2414        move |key: &str| map.get(key).cloned()
2415    }
2416
2417    #[test]
2418    fn skips_when_opt_in_unset() {
2419        let env = env_from(&[("GREENTIC_EVENTS_NATS_URL", "nats://127.0.0.1:4222")]);
2420        assert!(!should_serve_agentic_inproc(env));
2421    }
2422
2423    #[test]
2424    fn skips_when_nats_url_unset() {
2425        let env = env_from(&[("GREENTIC_AGENTIC_SERVE_INPROC", "1")]);
2426        assert!(!should_serve_agentic_inproc(env));
2427    }
2428
2429    #[test]
2430    fn skips_when_nats_url_blank() {
2431        let env = env_from(&[
2432            ("GREENTIC_AGENTIC_SERVE_INPROC", "1"),
2433            ("GREENTIC_EVENTS_NATS_URL", "   "),
2434        ]);
2435        assert!(!should_serve_agentic_inproc(env));
2436    }
2437
2438    #[test]
2439    fn serves_when_both_set() {
2440        for truthy in ["1", "true", "TRUE", "yes", "on"] {
2441            let env = env_from(&[
2442                ("GREENTIC_AGENTIC_SERVE_INPROC", truthy),
2443                ("GREENTIC_EVENTS_NATS_URL", "nats://127.0.0.1:4222"),
2444            ]);
2445            assert!(should_serve_agentic_inproc(env), "{truthy} should enable");
2446        }
2447    }
2448
2449    #[test]
2450    fn skips_on_falsey_opt_in() {
2451        for falsey in ["0", "false", "no", "off", "maybe"] {
2452            let env = env_from(&[
2453                ("GREENTIC_AGENTIC_SERVE_INPROC", falsey),
2454                ("GREENTIC_EVENTS_NATS_URL", "nats://127.0.0.1:4222"),
2455            ]);
2456            assert!(!should_serve_agentic_inproc(env), "{falsey} should skip");
2457        }
2458    }
2459
2460    #[test]
2461    fn dw_agent_dispatch_mode_defaults_inproc_and_parses_nats() {
2462        assert_eq!(dw_agent_dispatch_mode(|_| None), DwAgentDispatch::InProcess);
2463        assert_eq!(
2464            dw_agent_dispatch_mode(|k| (k == "GREENTIC_AW_DISPATCH").then(|| "nats".to_string())),
2465            DwAgentDispatch::Nats
2466        );
2467        assert_eq!(
2468            dw_agent_dispatch_mode(|k| {
2469                (k == "GREENTIC_AW_DISPATCH").then(|| "inproc".to_string())
2470            }),
2471            DwAgentDispatch::InProcess
2472        );
2473        assert_eq!(
2474            dw_agent_dispatch_mode(|k| (k == "GREENTIC_AW_DISPATCH").then(|| "NATS".to_string())),
2475            DwAgentDispatch::Nats
2476        );
2477    }
2478}
2479
2480/// Decide whether the runner process should host the agentic-worker NATS
2481/// service in-process (the opt-in co-host path).
2482///
2483/// Returns `true` only when BOTH gates are satisfied:
2484/// - `GREENTIC_AGENTIC_SERVE_INPROC` is truthy (`1`/`true`/`yes`/`on`,
2485///   case-insensitive) — opt-in, default OFF; and
2486/// - `GREENTIC_EVENTS_NATS_URL` is set to a non-empty value (no NATS bus means
2487///   nothing to serve on).
2488///
2489/// Pure over its `get_env` closure so it is unit-testable without touching the
2490/// real process environment. Feature-independent (no `agentic-worker` gate) so
2491/// the gating logic stays trivially testable; the actual spawn is gated at the
2492/// call site.
2493pub fn should_serve_agentic_inproc(get_env: impl Fn(&str) -> Option<String>) -> bool {
2494    let opt_in = get_env("GREENTIC_AGENTIC_SERVE_INPROC")
2495        .map(|value| {
2496            matches!(
2497                value.trim().to_ascii_lowercase().as_str(),
2498                "1" | "true" | "yes" | "on"
2499            )
2500        })
2501        .unwrap_or(false);
2502    let nats_set = get_env("GREENTIC_EVENTS_NATS_URL")
2503        .map(|value| !value.trim().is_empty())
2504        .unwrap_or(false);
2505    opt_in && nats_set
2506}
2507
2508/// How a `dw.agent` flow node executes.
2509#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2510#[allow(dead_code)] // consumed by Task 2.2 (engine.rs)
2511pub enum DwAgentDispatch {
2512    /// Run the agentic step in-process (default; today's behaviour).
2513    InProcess,
2514    /// Publish to the durable agentic NATS path (scale-to-zero compute).
2515    Nats,
2516}
2517
2518/// Resolve how `dw.agent` nodes execute. `GREENTIC_AW_DISPATCH=nats` routes them
2519/// over the out-of-process agentic NATS path; anything else (incl. unset) keeps
2520/// the in-process path — zero regression by default. Pure over `get_env` for
2521/// testability (mirrors `should_serve_agentic_inproc`).
2522#[must_use]
2523#[allow(dead_code)] // consumed by Task 2.2 (engine.rs)
2524pub fn dw_agent_dispatch_mode(get_env: impl Fn(&str) -> Option<String>) -> DwAgentDispatch {
2525    match get_env("GREENTIC_AW_DISPATCH") {
2526        Some(v) if v.trim().eq_ignore_ascii_case("nats") => DwAgentDispatch::Nats,
2527        _ => DwAgentDispatch::InProcess,
2528    }
2529}
2530
2531#[cfg(feature = "agentic-worker")]
2532pub use aw::{
2533    HostConfigProvider, RuntimeAgentNodeHandler, agent_configs_from_manifest,
2534    build_agent_node_handler, build_agent_runtime, load_process_agent_configs, merge_agent_sources,
2535    merge_sidecar_into, serve_agentic,
2536};
2537
2538#[cfg(feature = "desktop-agent-ephemeral")]
2539pub use aw::build_agent_node_handler_ephemeral;
2540
2541#[cfg(feature = "agentic-worker")]
2542pub(crate) use aw::{EnvSecretsBackend, build_ext_runtime, build_llm_backend};
2543
2544/// Test-only stand-in for `ExtensionRuntime::for_test()`, which this lane's
2545/// greentic-ext-runtime does not expose. Builds the equivalent: a runtime
2546/// rooted at a discovery path holding no extensions, plus the crate's own test
2547/// host overrides. Every lookup therefore returns an empty tool catalog, which
2548/// is what the agent/graph node tests want — they drive canned LLM replies and
2549/// never dispatch to a real extension.
2550#[cfg(all(test, feature = "agentic-worker"))]
2551pub(crate) fn test_extension_runtime() -> greentic_ext_runtime::ExtensionRuntime {
2552    greentic_ext_runtime::ExtensionRuntime::new(greentic_ext_runtime::RuntimeConfig::from_paths(
2553        greentic_ext_runtime::DiscoveryPaths::new(std::path::PathBuf::from(
2554            "/nonexistent/greentic-runner-host-test-extensions",
2555        )),
2556    ))
2557    .expect("wasmtime engine init for the runner-host test extension runtime")
2558    .with_host_overrides(greentic_ext_runtime::HostOverrides::defaults_for_tests())
2559}