1use anyhow::Result;
2use serde_json::Value;
3
4#[async_trait::async_trait]
10pub trait AgentNodeHandler: Send + Sync {
11 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#[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 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 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 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 const SANITISED_ERROR_REPLY: &str = "Something went wrong. Please try again.";
116
117 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 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 pub struct RuntimeAgentNodeHandler {
173 runtime: Arc<AgentRuntime>,
174 audit_sink: Option<AuditSink>,
180 project_id: Option<String>,
189 }
190
191 impl RuntimeAgentNodeHandler {
192 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 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 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 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 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 pub struct HostConfigProvider {
318 agents: HashMap<String, AgentConfig>,
319 }
320
321 impl HostConfigProvider {
322 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 pub(crate) fn in_process_llm_backend() -> Arc<dyn greentic_aw_runtime::LlmBackend> {
726 in_process_llm_backend_with_key(None)
727 }
728
729 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 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 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 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 #[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 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 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 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 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 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 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 #[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 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 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; }
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 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 let ext_runtime = build_ext_runtime(Arc::new(EnvSecretsBackend))?;
1192
1193 let llm = build_llm_backend(&ext_runtime);
1199
1200 let agent_count = merged_agents.len();
1201 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 #[cfg(not(greentic_guardrail_ext))]
1252 {
1253 Arc::new(greentic_aw_runtime::guardrail::UnavailableGuardrailEvaluator)
1254 }
1255 },
1256 );
1257 let base = base.with_short_term_memory(Arc::new(
1261 greentic_aw_runtime::memory::InMemoryMemoryProvider::new(),
1262 ));
1263 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 #[cfg(feature = "long-term-chronicle")]
1278 let base = crate::runner::long_term_memory::attach(base).await;
1279 #[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 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 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 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 #[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 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 #[tokio::test]
1630 #[serial_test::serial]
1631 #[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 .respond_with(
1641 wiremock::ResponseTemplate::new(200)
1642 .set_body_json(serde_json::json!({"available": "0"})),
1643 )
1644 .mount(&server)
1645 .await;
1646
1647 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 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 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 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 #[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 #[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 #[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 #[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 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 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 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 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 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 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 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 unsafe {
2199 std::env::set_var("GREENTIC_AW_MCP", "1");
2200 }
2201 assert!(super::mcp_source_from_env().is_some());
2202
2203 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 #[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 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 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 None,
2275 )
2276 .await;
2277 assert!(
2278 handler.is_some(),
2279 "ephemeral builder must not require Redis"
2280 );
2281 }
2282
2283 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 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 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 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 std::fs::write(dir.path().join("broken.json"), b"{ not json").expect("write broken");
2365
2366 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"})), ("b".to_string(), serde_json::json!({"from": "sidecar"})), ]);
2395 super::merge_sidecar_into(&mut blobs, sidecar);
2396 assert_eq!(blobs["a"]["from"], "manifest"); assert_eq!(blobs["b"]["from"], "sidecar"); assert_eq!(blobs.len(), 2);
2399 }
2400 }
2401}
2402
2403#[allow(clippy::items_after_test_module)] #[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
2480pub 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#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2510#[allow(dead_code)] pub enum DwAgentDispatch {
2512 InProcess,
2514 Nats,
2516}
2517
2518#[must_use]
2523#[allow(dead_code)] pub 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#[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}