1use std::collections::{BTreeMap, HashMap, HashSet};
2use std::fs::File;
3use std::io::Read;
4use std::path::{Path, PathBuf};
5use std::str::FromStr;
6use std::sync::{Arc, Once};
7use std::time::Duration;
8
9use crate::cache::{ArtifactKey, CacheConfig, CacheManager, CpuPolicy, EngineProfile};
10use crate::component_api::{
11 self, node::ExecCtx as ComponentExecCtx, node::InvokeResult, node::NodeError,
12};
13use crate::oauth::{OAuthBrokerConfig, OAuthBrokerHost, OAuthHostContext};
14use crate::provider::{ProviderBinding, ProviderRegistry};
15use crate::provider_core::{
16 schema_core::SchemaCorePre as LegacySchemaCorePre,
17 schema_core_path::SchemaCorePre as PathSchemaCorePre,
18 schema_core_schema::SchemaCorePre as SchemaSchemaCorePre,
19};
20use crate::provider_core_only;
21use crate::runtime_wasmtime::{Component, Engine, InstancePre, Linker, ResourceTable};
22use anyhow::{Context, Result, anyhow, bail};
23use futures::executor::block_on;
24use greentic_distributor_client::dist::{
25 CachePolicy, DistClient, DistError, DistOptions, ResolvePolicy,
26};
27use greentic_interfaces_wasmtime::host_helpers::v1::{
28 self as host_v1, HostFns, add_all_v1_to_linker,
29 runner_host_http::RunnerHostHttp,
30 runner_host_kv::RunnerHostKv,
31 secrets_store::{SecretsError, SecretsErrorV1_1, SecretsStoreHost, SecretsStoreHostV1_1},
32 state_store::{
33 OpAck as StateOpAck, StateKey as HostStateKey, StateStoreError as StateError,
34 StateStoreHost, TenantCtx as StateTenantCtx,
35 },
36 telemetry_logger::{
37 OpAck as TelemetryAck, SpanContext as TelemetrySpanContext,
38 TelemetryLoggerError as TelemetryError, TelemetryLoggerHost,
39 TenantCtx as TelemetryTenantCtx,
40 },
41};
42use greentic_interfaces_wasmtime::http_client_client_v1_1::greentic::http::http_client as http_client_client_alias;
43use greentic_interfaces_wasmtime::{
44 http_client_client_v1_0::greentic::interfaces_types::types as http_types_v1_0,
45 http_client_client_v1_1::greentic::interfaces_types::types as http_types_v1_1,
46};
47use greentic_pack::builder as legacy_pack;
48use greentic_types::flow::FlowHasher;
49use greentic_types::{
50 ArtifactLocationV1, ComponentId, ComponentManifest, ComponentSourceRef, ComponentSourcesV1,
51 EXT_COMPONENT_SOURCES_V1, EnvId, ExtensionRef, Flow, FlowComponentRef, FlowId, FlowKind,
52 FlowMetadata, InputMapping, Node, NodeId, OutputMapping, Routing, StateKey as StoreStateKey,
53 TeamId, TelemetryHints, TenantCtx as TypesTenantCtx, TenantId, UserId, decode_pack_manifest,
54 pack_manifest::ExtensionInline,
55};
56use host_v1::http_client as host_http_client;
57use host_v1::http_client::{
58 HttpClientError, HttpClientErrorV1_1, HttpClientHost, HttpClientHostV1_1,
59 Request as HttpRequest, RequestOptionsV1_1 as HttpRequestOptionsV1_1,
60 RequestV1_1 as HttpRequestV1_1, Response as HttpResponse, ResponseV1_1 as HttpResponseV1_1,
61 TenantCtx as HttpTenantCtx, TenantCtxV1_1 as HttpTenantCtxV1_1,
62};
63use indexmap::IndexMap;
64use once_cell::sync::Lazy;
65use parking_lot::{Mutex, RwLock};
66use reqwest::blocking::Client as BlockingClient;
67use runner_core::normalize_under_root;
68use serde::{Deserialize, Serialize};
69use serde_cbor;
70use serde_json::{self, Value};
71use sha2::Digest;
72use tempfile::TempDir;
73use tokio::fs;
74use wasmparser::{Parser, Payload};
75use wasmtime::{Store, StoreContextMut};
76use wasmtime_wasi_http::WasiHttpCtx;
77use wasmtime_wasi_http::p2::{
78 WasiHttpCtxView, WasiHttpView, add_only_http_to_linker_sync as add_wasi_http_to_linker,
79};
80use wasmtime_wasi_tls::p2::{LinkOptions, add_to_linker as add_wasi_tls_to_linker};
81use wasmtime_wasi_tls::{WasiTlsCtx, WasiTlsCtxBuilder, WasiTlsCtxView, WasiTlsView};
82use zip::ZipArchive;
83
84use crate::runner::engine::{FlowContext, FlowEngine, FlowStatus};
85use crate::runner::flow_adapter::{FlowIR, flow_doc_to_ir, flow_ir_to_flow};
86use crate::runner::mocks::{HttpDecision, HttpMockRequest, HttpMockResponse, MockLayer};
87#[cfg(feature = "fault-injection")]
88use crate::testing::fault_injection::{FaultContext, FaultPoint, maybe_fail};
89
90use crate::config::HostConfig;
91use crate::fault;
92use crate::secrets::{
93 DynSecretsManager, canonicalize_secret_key, read_secret_blocking, write_secret_blocking,
94};
95use crate::storage::state::STATE_PREFIX;
96use crate::storage::{DynSessionStore, DynStateStore};
97use crate::verify;
98use crate::wasi::{PreopenSpec, RunnerWasiPolicy};
99use tracing::warn;
100use wasmtime_wasi::p2::add_to_linker_sync as add_wasi_to_linker;
101use wasmtime_wasi::{WasiCtx, WasiCtxView, WasiView};
102
103use greentic_flow::model::FlowDoc;
104
105static RUSTLS_CRYPTO_PROVIDER: Once = Once::new();
106
107fn ensure_rustls_crypto_provider() {
108 RUSTLS_CRYPTO_PROVIDER.call_once(|| {
109 let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
110 });
111}
112
113#[allow(dead_code)]
114pub struct PackRuntime {
115 path: PathBuf,
117 archive_path: Option<PathBuf>,
119 config: Arc<HostConfig>,
120 engine: Engine,
121 metadata: PackMetadata,
122 manifest: Option<greentic_types::PackManifest>,
123 legacy_manifest: Option<Box<legacy_pack::PackManifest>>,
124 component_manifests: HashMap<String, ComponentManifest>,
125 mocks: Option<Arc<MockLayer>>,
126 flows: Option<PackFlows>,
127 components: HashMap<String, PackComponent>,
128 http_client: Arc<BlockingClient>,
129 pre_cache: Mutex<HashMap<String, InstancePre<ComponentState>>>,
130 session_store: Option<DynSessionStore>,
131 state_store: Option<DynStateStore>,
132 wasi_policy: Arc<RunnerWasiPolicy>,
133 assets_tempdir: Option<TempDir>,
134 provider_registry: RwLock<Option<ProviderRegistry>>,
135 secrets: DynSecretsManager,
136 oauth_config: Option<OAuthBrokerConfig>,
137 cache: CacheManager,
138}
139
140struct PackComponent {
141 #[allow(dead_code)]
142 name: String,
143 #[allow(dead_code)]
144 version: String,
145 component: Arc<Component>,
146}
147
148fn run_on_wasi_thread<F, T>(task_name: &'static str, task: F) -> Result<T>
149where
150 F: FnOnce() -> Result<T> + Send + 'static,
151 T: Send + 'static,
152{
153 let builder = std::thread::Builder::new().name(format!("greentic-wasmtime-{task_name}"));
154 let handle = builder
155 .spawn(move || {
156 let pid = std::process::id();
157 let thread_id = std::thread::current().id();
158 let tokio_handle_present = tokio::runtime::Handle::try_current().is_ok();
159 tracing::info!(
160 event = "wasmtime.thread.start",
161 task = task_name,
162 pid,
163 thread_id = ?thread_id,
164 tokio_handle_present,
165 "starting Wasmtime thread"
166 );
167 task()
168 })
169 .context("failed to spawn Wasmtime thread")?;
170 handle
171 .join()
172 .map_err(|err| {
173 let reason = if let Some(msg) = err.downcast_ref::<&str>() {
174 msg.to_string()
175 } else if let Some(msg) = err.downcast_ref::<String>() {
176 msg.clone()
177 } else {
178 "unknown panic".to_string()
179 };
180 anyhow!("Wasmtime thread panicked: {reason}")
181 })
182 .and_then(|res| res)
183}
184
185#[derive(Debug, Default, Clone)]
186pub struct ComponentResolution {
187 pub materialized_root: Option<PathBuf>,
189 pub overrides: HashMap<String, PathBuf>,
191 pub dist_offline: bool,
193 pub dist_cache_dir: Option<PathBuf>,
195 pub allow_missing_hash: bool,
197}
198
199fn build_blocking_client() -> BlockingClient {
200 std::thread::spawn(|| {
201 BlockingClient::builder()
202 .no_proxy()
203 .build()
204 .expect("blocking client")
205 })
206 .join()
207 .expect("client build thread panicked")
208}
209
210fn normalize_pack_path(path: &Path) -> Result<(PathBuf, PathBuf)> {
211 let (root, candidate) = if path.is_absolute() {
212 let parent = path
213 .parent()
214 .ok_or_else(|| anyhow!("pack path {} has no parent", path.display()))?;
215 let root = parent
216 .canonicalize()
217 .with_context(|| format!("failed to canonicalize {}", parent.display()))?;
218 let file = path
219 .file_name()
220 .ok_or_else(|| anyhow!("pack path {} has no file name", path.display()))?;
221 (root, PathBuf::from(file))
222 } else {
223 let cwd = std::env::current_dir().context("failed to resolve current directory")?;
224 let base = if let Some(parent) = path.parent() {
225 cwd.join(parent)
226 } else {
227 cwd
228 };
229 let root = base
230 .canonicalize()
231 .with_context(|| format!("failed to canonicalize {}", base.display()))?;
232 let file = path
233 .file_name()
234 .ok_or_else(|| anyhow!("pack path {} has no file name", path.display()))?;
235 (root, PathBuf::from(file))
236 };
237 let safe = normalize_under_root(&root, &candidate)?;
238 Ok((root, safe))
239}
240
241static HTTP_CLIENT: Lazy<Arc<BlockingClient>> = Lazy::new(|| Arc::new(build_blocking_client()));
242
243#[derive(Debug, Clone, Serialize, Deserialize)]
244pub struct FlowDescriptor {
245 pub id: String,
246 #[serde(rename = "type")]
247 pub flow_type: String,
248 pub pack_id: String,
249 pub profile: String,
250 pub version: String,
251 #[serde(default)]
252 pub description: Option<String>,
253}
254
255pub struct HostState {
256 #[allow(dead_code)]
257 pack_id: String,
258 config: Arc<HostConfig>,
259 http_client: Arc<BlockingClient>,
260 default_env: String,
261 #[allow(dead_code)]
262 session_store: Option<DynSessionStore>,
263 state_store: Option<DynStateStore>,
264 mocks: Option<Arc<MockLayer>>,
265 secrets: DynSecretsManager,
266 oauth_config: Option<OAuthBrokerConfig>,
267 oauth_host: OAuthBrokerHost,
268 exec_ctx: Option<ComponentExecCtx>,
269 component_ref: Option<String>,
270 provider_core_component: bool,
271}
272
273impl HostState {
274 #[allow(clippy::default_constructed_unit_structs)]
275 #[allow(clippy::too_many_arguments)]
276 pub fn new(
277 pack_id: String,
278 config: Arc<HostConfig>,
279 http_client: Arc<BlockingClient>,
280 mocks: Option<Arc<MockLayer>>,
281 session_store: Option<DynSessionStore>,
282 state_store: Option<DynStateStore>,
283 secrets: DynSecretsManager,
284 oauth_config: Option<OAuthBrokerConfig>,
285 exec_ctx: Option<ComponentExecCtx>,
286 component_ref: Option<String>,
287 provider_core_component: bool,
288 ) -> Result<Self> {
289 let default_env = std::env::var("GREENTIC_ENV").unwrap_or_else(|_| "local".to_string());
290 Ok(Self {
291 pack_id,
292 config,
293 http_client,
294 default_env,
295 session_store,
296 state_store,
297 mocks,
298 secrets,
299 oauth_config,
300 oauth_host: OAuthBrokerHost::default(),
301 exec_ctx,
302 component_ref,
303 provider_core_component,
304 })
305 }
306
307 fn instantiate_component_result(
308 linker: &mut Linker<ComponentState>,
309 store: &mut Store<ComponentState>,
310 component: &Component,
311 ctx: &ComponentExecCtx,
312 component_ref: &str,
313 operation: &str,
314 input_json: &str,
315 ) -> Result<InvokeResult> {
316 let pre_instance = linker.instantiate_pre(component)?;
317 match component_api::v0_6::ComponentPre::new(pre_instance) {
318 Ok(pre) => {
319 let envelope = component_api::envelope_v0_6(ctx, component_ref, input_json)?;
320 let operation_owned = operation.to_string();
321 let result = block_on(async {
322 let bindings = pre.instantiate_async(&mut *store).await?;
323 let node = bindings.greentic_component_node();
324 node.call_invoke(&mut *store, &operation_owned, &envelope)
325 })?;
326 component_api::invoke_result_from_v0_6(result)
327 }
328 Err(err_v06) => {
329 if !is_missing_node_export(&err_v06, "0.6.0") {
330 return Err(err_v06.into());
331 }
332 let pre_instance = linker.instantiate_pre(component)?;
333 match component_api::v0_5::ComponentPre::new(pre_instance) {
334 Ok(pre) => {
335 let result = block_on(async {
336 let bindings = pre.instantiate_async(&mut *store).await?;
337 let node = bindings.greentic_component_node();
338 let ctx_v05 = component_api::exec_ctx_v0_5(ctx);
339 let operation_owned = operation.to_string();
340 let input_owned = input_json.to_string();
341 node.call_invoke(&mut *store, &ctx_v05, &operation_owned, &input_owned)
342 })?;
343 Ok(component_api::invoke_result_from_v0_5(result))
344 }
345 Err(err) => {
346 if !is_missing_node_export(&err, "0.5.0") {
347 return Err(err.into());
348 }
349 let pre_instance = linker.instantiate_pre(component)?;
350 match component_api::v0_4::ComponentPre::new(pre_instance) {
351 Ok(pre) => {
352 let result = block_on(async {
353 let bindings = pre.instantiate_async(&mut *store).await?;
354 let node = bindings.greentic_component_node();
355 let ctx_v04 = component_api::exec_ctx_v0_4(ctx);
356 let operation_owned = operation.to_string();
357 let input_owned = input_json.to_string();
358 node.call_invoke(
359 &mut *store,
360 &ctx_v04,
361 &operation_owned,
362 &input_owned,
363 )
364 })?;
365 Ok(component_api::invoke_result_from_v0_4(result))
366 }
367 Err(err_v04) => {
368 if is_missing_node_export(&err_v04, "0.4.0") {
369 Self::try_v06_runtime(linker, store, component, input_json)
370 } else {
371 Err(err_v04.into())
372 }
373 }
374 }
375 }
376 }
377 }
378 }
379 }
380
381 fn try_v06_runtime(
384 linker: &mut Linker<ComponentState>,
385 store: &mut Store<ComponentState>,
386 component: &Component,
387 input_json: &str,
388 ) -> Result<InvokeResult> {
389 let pre_instance = linker.instantiate_pre(component)?;
390 let pre = component_api::v0_6_runtime::ComponentV0V6RuntimePre::new(pre_instance).map_err(
391 |err| err.context("component exports neither node@0.5/0.4 nor component-runtime@0.6"),
392 )?;
393
394 let result = block_on(async {
395 let bindings = pre.instantiate_async(&mut *store).await?;
396 let runtime = bindings.greentic_component_component_runtime();
397
398 let input_value: Value = serde_json::from_str(input_json).unwrap_or(Value::Null);
400 let input_cbor =
401 serde_cbor::to_vec(&input_value).context("encode input as CBOR for v0.6")?;
402 let empty_state = serde_cbor::to_vec(&Value::Object(Default::default()))
403 .context("encode empty state")?;
404
405 let run_result = runtime
406 .call_run(&mut *store, &input_cbor, &empty_state)
407 .map_err(|err| err.context("v0.6 component-runtime::run call failed"))?;
408
409 let output_value: Value = serde_cbor::from_slice(&run_result.output)
411 .context("decode v0.6 run output CBOR")?;
412 let output_json = serde_json::to_string(&output_value)
413 .context("serialize v0.6 run output to JSON")?;
414
415 Ok::<_, anyhow::Error>(output_json)
416 })?;
417
418 Ok(InvokeResult::Ok(result))
419 }
420
421 fn convert_invoke_result(result: InvokeResult) -> Result<Value> {
422 match result {
423 InvokeResult::Ok(body) => {
424 if body.is_empty() {
425 return Ok(Value::Null);
426 }
427 serde_json::from_str(&body).or_else(|_| Ok(Value::String(body)))
428 }
429 InvokeResult::Err(NodeError {
430 code,
431 message,
432 retryable,
433 backoff_ms,
434 details,
435 }) => {
436 let mut obj = serde_json::Map::new();
437 obj.insert("ok".into(), Value::Bool(false));
438 let mut error = serde_json::Map::new();
439 error.insert("code".into(), Value::String(code));
440 error.insert("message".into(), Value::String(message));
441 error.insert("retryable".into(), Value::Bool(retryable));
442 if let Some(backoff) = backoff_ms {
443 error.insert("backoff_ms".into(), Value::Number(backoff.into()));
444 }
445 if let Some(details) = details {
446 error.insert(
447 "details".into(),
448 serde_json::from_str(&details).unwrap_or(Value::String(details)),
449 );
450 }
451 obj.insert("error".into(), Value::Object(error));
452 Ok(Value::Object(obj))
453 }
454 }
455 }
456
457 fn secrets_tenant_ctx(&self) -> TypesTenantCtx {
461 let mut ctx = self.config.tenant_ctx();
462 if let Some(exec_ctx) = self.exec_ctx.as_ref()
463 && let Some(team) = exec_ctx.tenant.team.as_ref()
464 && let Ok(team_id) = TeamId::from_str(team)
465 {
466 ctx = ctx.with_team(Some(team_id));
467 }
468 ctx
469 }
470
471 pub fn get_secret(&self, key: &str) -> Result<String> {
472 if provider_core_only::is_enabled() {
473 bail!(provider_core_only::blocked_message("secrets"))
474 }
475 if !self.config.secrets_policy.is_allowed(key) {
476 bail!("secret {key} is not permitted by bindings policy");
477 }
478 if let Some(mock) = &self.mocks
479 && let Some(value) = mock.secrets_lookup(key)
480 {
481 return Ok(value);
482 }
483 let ctx = self.secrets_tenant_ctx();
484 let canonical_key = canonicalize_secret_key(key);
485 let bytes = read_secret_blocking(&self.secrets, &ctx, &self.pack_id, &canonical_key)
486 .context("failed to read secret from manager")?;
487 let value = String::from_utf8(bytes).context("secret value is not valid UTF-8")?;
488 Ok(value)
489 }
490
491 fn allows_secret_write_in_provider_core_only(&self) -> bool {
492 self.provider_core_component || self.component_ref.is_none()
493 }
494
495 fn tenant_ctx_from_v1(&self, ctx: Option<StateTenantCtx>) -> Result<TypesTenantCtx> {
496 let tenant_raw = ctx
497 .as_ref()
498 .map(|ctx| ctx.tenant.clone())
499 .or_else(|| self.exec_ctx.as_ref().map(|ctx| ctx.tenant.tenant.clone()))
500 .unwrap_or_else(|| self.config.tenant.clone());
501 let env_raw = ctx
502 .as_ref()
503 .map(|ctx| ctx.env.clone())
504 .unwrap_or_else(|| self.default_env.clone());
505 let tenant_id = TenantId::from_str(&tenant_raw)
506 .with_context(|| format!("invalid tenant id `{tenant_raw}`"))?;
507 let env_id = EnvId::from_str(&env_raw)
508 .unwrap_or_else(|_| EnvId::from_str("local").expect("default env must be valid"));
509 let mut tenant_ctx = TypesTenantCtx::new(env_id, tenant_id);
510 if let Some(exec_ctx) = self.exec_ctx.as_ref() {
511 if let Some(team) = exec_ctx.tenant.team.as_ref() {
512 let team_id =
513 TeamId::from_str(team).with_context(|| format!("invalid team id `{team}`"))?;
514 tenant_ctx = tenant_ctx.with_team(Some(team_id));
515 }
516 if let Some(user) = exec_ctx.tenant.user.as_ref() {
517 let user_id =
518 UserId::from_str(user).with_context(|| format!("invalid user id `{user}`"))?;
519 tenant_ctx = tenant_ctx.with_user(Some(user_id));
520 }
521 tenant_ctx = tenant_ctx.with_flow(exec_ctx.flow_id.clone());
522 if let Some(node) = exec_ctx.node_id.as_ref() {
523 tenant_ctx = tenant_ctx.with_node(node.clone());
524 }
525 if let Some(session) = exec_ctx.tenant.correlation_id.as_ref() {
526 tenant_ctx = tenant_ctx.with_session(session.clone());
527 }
528 tenant_ctx.trace_id = exec_ctx.tenant.trace_id.clone();
529 }
530
531 if let Some(ctx) = ctx {
532 if let Some(team) = ctx.team.or(ctx.team_id) {
533 let team_id =
534 TeamId::from_str(&team).with_context(|| format!("invalid team id `{team}`"))?;
535 tenant_ctx = tenant_ctx.with_team(Some(team_id));
536 }
537 if let Some(user) = ctx.user.or(ctx.user_id) {
538 let user_id =
539 UserId::from_str(&user).with_context(|| format!("invalid user id `{user}`"))?;
540 tenant_ctx = tenant_ctx.with_user(Some(user_id));
541 }
542 if let Some(flow) = ctx.flow_id {
543 tenant_ctx = tenant_ctx.with_flow(flow);
544 }
545 if let Some(node) = ctx.node_id {
546 tenant_ctx = tenant_ctx.with_node(node);
547 }
548 if let Some(provider) = ctx.provider_id {
549 tenant_ctx = tenant_ctx.with_provider(provider);
550 }
551 if let Some(session) = ctx.session_id {
552 tenant_ctx = tenant_ctx.with_session(session);
553 }
554 tenant_ctx.trace_id = ctx.trace_id;
555 }
556 Ok(tenant_ctx)
557 }
558
559 fn send_http_request(
560 &mut self,
561 req: HttpRequest,
562 opts: Option<HttpRequestOptionsV1_1>,
563 _ctx: Option<HttpTenantCtx>,
564 ) -> Result<HttpResponse, HttpClientError> {
565 if !self.config.http_enabled {
566 return Err(HttpClientError {
567 code: "denied".into(),
568 message: "http client disabled by policy".into(),
569 });
570 }
571
572 let mut mock_state = None;
573 let raw_body = req.body.clone();
574 if let Some(mock) = &self.mocks
575 && let Ok(meta) = HttpMockRequest::new(&req.method, &req.url, raw_body.as_deref())
576 {
577 match mock.http_begin(&meta) {
578 HttpDecision::Mock(response) => {
579 let headers = response
580 .headers
581 .iter()
582 .map(|(k, v)| (k.clone(), v.clone()))
583 .collect();
584 return Ok(HttpResponse {
585 status: response.status,
586 headers,
587 body: response.body.clone().map(|b| b.into_bytes()),
588 });
589 }
590 HttpDecision::Deny(reason) => {
591 return Err(HttpClientError {
592 code: "denied".into(),
593 message: reason,
594 });
595 }
596 HttpDecision::Passthrough { record } => {
597 mock_state = Some((meta, record));
598 }
599 }
600 }
601
602 let method = req.method.parse().unwrap_or(reqwest::Method::GET);
603 let mut builder = self.http_client.request(method, &req.url);
604 for (key, value) in req.headers {
605 if let Ok(header) = reqwest::header::HeaderName::from_bytes(key.as_bytes())
606 && let Ok(header_value) = reqwest::header::HeaderValue::from_str(&value)
607 {
608 builder = builder.header(header, header_value);
609 }
610 }
611
612 if let Some(body) = raw_body.clone() {
613 builder = builder.body(body);
614 }
615
616 if let Some(opts) = opts {
617 if let Some(timeout_ms) = opts.timeout_ms {
618 builder = builder.timeout(Duration::from_millis(timeout_ms as u64));
619 }
620 if opts.allow_insecure == Some(true) {
621 warn!(url = %req.url, "allow-insecure not supported; using default TLS validation");
622 }
623 if let Some(follow_redirects) = opts.follow_redirects
624 && !follow_redirects
625 {
626 warn!(url = %req.url, "follow-redirects=false not supported; using default client behaviour");
627 }
628 }
629
630 let response = match builder.send() {
631 Ok(resp) => resp,
632 Err(err) => {
633 warn!(url = %req.url, error = %err, "http client request failed");
634 return Err(HttpClientError {
635 code: "unavailable".into(),
636 message: err.to_string(),
637 });
638 }
639 };
640
641 let status = response.status().as_u16();
642 let headers_vec = response
643 .headers()
644 .iter()
645 .map(|(k, v)| {
646 (
647 k.as_str().to_string(),
648 v.to_str().unwrap_or_default().to_string(),
649 )
650 })
651 .collect::<Vec<_>>();
652 let body_bytes = response.bytes().ok().map(|b| b.to_vec());
653
654 if let Some((meta, true)) = mock_state.take()
655 && let Some(mock) = &self.mocks
656 {
657 let recorded = HttpMockResponse::new(
658 status,
659 headers_vec.clone().into_iter().collect(),
660 body_bytes
661 .as_ref()
662 .map(|b| String::from_utf8_lossy(b).into_owned()),
663 );
664 mock.http_record(&meta, &recorded);
665 }
666
667 Ok(HttpResponse {
668 status,
669 headers: headers_vec,
670 body: body_bytes,
671 })
672 }
673}
674
675#[cfg(test)]
676mod canonicalize_tests {
677 use crate::secrets::canonicalize_secret_key;
678
679 #[test]
680 fn upper_snake_to_lower_snake() {
681 assert_eq!(
682 canonicalize_secret_key("TELEGRAM_BOT_TOKEN"),
683 "telegram_bot_token"
684 );
685 }
686
687 #[test]
688 fn trim_and_replace_non_alphanumeric() {
689 assert_eq!(
690 canonicalize_secret_key(" webex-bot-token "),
691 "webex_bot_token"
692 );
693 }
694
695 #[test]
696 fn preserve_existing_lower_snake_with_extra_underscores() {
697 assert_eq!(canonicalize_secret_key("MiXeD__Case"), "mixed__case");
698 }
699}
700
701impl SecretsStoreHost for HostState {
702 fn get(&mut self, key: String) -> Result<Option<Vec<u8>>, SecretsError> {
703 if provider_core_only::is_enabled() {
704 warn!(secret = %key, "provider-core only mode enabled; blocking secrets store");
705 return Err(SecretsError::Denied);
706 }
707 if !self.config.secrets_policy.is_allowed(&key) {
708 return Err(SecretsError::Denied);
709 }
710 if let Some(mock) = &self.mocks
711 && let Some(value) = mock.secrets_lookup(&key)
712 {
713 return Ok(Some(value.into_bytes()));
714 }
715 let ctx = self.secrets_tenant_ctx();
716 let canonical_key = canonicalize_secret_key(&key);
717 match read_secret_blocking(&self.secrets, &ctx, &self.pack_id, &canonical_key) {
718 Ok(bytes) => Ok(Some(bytes)),
719 Err(err) => {
720 warn!(secret = %key, canonical = %canonical_key, error = %err, "secret lookup failed");
721 Err(SecretsError::NotFound)
722 }
723 }
724 }
725}
726
727impl SecretsStoreHostV1_1 for HostState {
728 fn get(&mut self, key: String) -> Result<Option<Vec<u8>>, SecretsErrorV1_1> {
729 if provider_core_only::is_enabled() {
730 warn!(secret = %key, "provider-core only mode enabled; blocking secrets store");
731 return Err(SecretsErrorV1_1::Denied);
732 }
733 if !self.config.secrets_policy.is_allowed(&key) {
734 return Err(SecretsErrorV1_1::Denied);
735 }
736 if let Some(mock) = &self.mocks
737 && let Some(value) = mock.secrets_lookup(&key)
738 {
739 return Ok(Some(value.into_bytes()));
740 }
741 let ctx = self.secrets_tenant_ctx();
742 let canonical_key = canonicalize_secret_key(&key);
743 match read_secret_blocking(&self.secrets, &ctx, &self.pack_id, &canonical_key) {
744 Ok(bytes) => Ok(Some(bytes)),
745 Err(err) => {
746 warn!(secret = %key, canonical = %canonical_key, error = %err, "secret lookup failed");
747 Err(SecretsErrorV1_1::NotFound)
748 }
749 }
750 }
751
752 fn put(&mut self, key: String, value: Vec<u8>) {
753 if key.trim().is_empty() {
754 warn!(secret = %key, "secret write blocked: empty key");
755 panic!("secret write denied for key {key}: invalid key");
756 }
757 if provider_core_only::is_enabled() && !self.allows_secret_write_in_provider_core_only() {
758 warn!(
759 secret = %key,
760 component = self.component_ref.as_deref().unwrap_or("<pack>"),
761 "provider-core only mode enabled; blocking secrets store write"
762 );
763 panic!("secret write denied for key {key}: provider-core-only mode");
764 }
765 if !self.config.secrets_policy.is_allowed(&key) {
766 warn!(secret = %key, "secret write denied by bindings policy");
767 panic!("secret write denied for key {key}: policy");
768 }
769 let ctx = self.secrets_tenant_ctx();
770 let canonical_key = canonicalize_secret_key(&key);
771 if let Err(err) =
772 write_secret_blocking(&self.secrets, &ctx, &self.pack_id, &canonical_key, &value)
773 {
774 warn!(secret = %key, canonical = %canonical_key, error = %err, "secret write failed");
775 panic!("secret write failed for key {key}");
776 }
777 }
778}
779
780impl HttpClientHost for HostState {
781 fn send(
782 &mut self,
783 req: HttpRequest,
784 ctx: Option<HttpTenantCtx>,
785 ) -> Result<HttpResponse, HttpClientError> {
786 self.send_http_request(req, None, ctx)
787 }
788}
789
790impl HttpClientHostV1_1 for HostState {
791 fn send(
792 &mut self,
793 req: HttpRequestV1_1,
794 opts: Option<HttpRequestOptionsV1_1>,
795 ctx: Option<HttpTenantCtxV1_1>,
796 ) -> Result<HttpResponseV1_1, HttpClientErrorV1_1> {
797 let legacy_req = HttpRequest {
798 method: req.method,
799 url: req.url,
800 headers: req.headers,
801 body: req.body,
802 };
803 let legacy_ctx = ctx.map(|ctx| HttpTenantCtx {
804 env: ctx.env,
805 tenant: ctx.tenant,
806 tenant_id: ctx.tenant_id,
807 team: ctx.team,
808 team_id: ctx.team_id,
809 user: ctx.user,
810 user_id: ctx.user_id,
811 trace_id: ctx.trace_id,
812 correlation_id: ctx.correlation_id,
813 i18n_id: ctx.i18n_id,
814 attributes: ctx.attributes,
815 session_id: ctx.session_id,
816 flow_id: ctx.flow_id,
817 node_id: ctx.node_id,
818 provider_id: ctx.provider_id,
819 deadline_ms: ctx.deadline_ms,
820 attempt: ctx.attempt,
821 idempotency_key: ctx.idempotency_key,
822 impersonation: ctx.impersonation.map(|imp| http_types_v1_0::Impersonation {
823 actor_id: imp.actor_id,
824 reason: imp.reason,
825 }),
826 });
827
828 self.send_http_request(legacy_req, opts, legacy_ctx)
829 .map(|resp| HttpResponseV1_1 {
830 status: resp.status,
831 headers: resp.headers,
832 body: resp.body,
833 })
834 .map_err(|err| HttpClientErrorV1_1 {
835 code: err.code,
836 message: err.message,
837 })
838 }
839}
840
841impl StateStoreHost for HostState {
842 fn read(
843 &mut self,
844 key: HostStateKey,
845 ctx: Option<StateTenantCtx>,
846 ) -> Result<Vec<u8>, StateError> {
847 let store = match self.state_store.as_ref() {
848 Some(store) => store.clone(),
849 None => {
850 return Err(StateError {
851 code: "unavailable".into(),
852 message: "state store not configured".into(),
853 });
854 }
855 };
856 let tenant_ctx = match self.tenant_ctx_from_v1(ctx) {
857 Ok(ctx) => ctx,
858 Err(err) => {
859 return Err(StateError {
860 code: "invalid-ctx".into(),
861 message: err.to_string(),
862 });
863 }
864 };
865 #[cfg(feature = "fault-injection")]
866 {
867 let exec_ctx = self.exec_ctx.as_ref();
868 let flow_id = exec_ctx
869 .map(|ctx| ctx.flow_id.as_str())
870 .unwrap_or("unknown");
871 let node_id = exec_ctx.and_then(|ctx| ctx.node_id.as_deref());
872 let attempt = exec_ctx.map(|ctx| ctx.tenant.attempt).unwrap_or(1);
873 let fault_ctx = FaultContext {
874 pack_id: self.pack_id.as_str(),
875 flow_id,
876 node_id,
877 attempt,
878 };
879 if let Err(err) = maybe_fail(FaultPoint::StateRead, fault_ctx) {
880 return Err(StateError {
881 code: "internal".into(),
882 message: err.to_string(),
883 });
884 }
885 }
886 let key = StoreStateKey::from(key);
887 match store.get_json(&tenant_ctx, STATE_PREFIX, &key, None) {
888 Ok(Some(value)) => Ok(serde_json::to_vec(&value).unwrap_or_else(|_| Vec::new())),
889 Ok(None) => Err(StateError {
890 code: "not_found".into(),
891 message: "state key not found".into(),
892 }),
893 Err(err) => Err(StateError {
894 code: "internal".into(),
895 message: err.to_string(),
896 }),
897 }
898 }
899
900 fn write(
901 &mut self,
902 key: HostStateKey,
903 bytes: Vec<u8>,
904 ctx: Option<StateTenantCtx>,
905 ) -> Result<StateOpAck, StateError> {
906 let store = match self.state_store.as_ref() {
907 Some(store) => store.clone(),
908 None => {
909 return Err(StateError {
910 code: "unavailable".into(),
911 message: "state store not configured".into(),
912 });
913 }
914 };
915 let tenant_ctx = match self.tenant_ctx_from_v1(ctx) {
916 Ok(ctx) => ctx,
917 Err(err) => {
918 return Err(StateError {
919 code: "invalid-ctx".into(),
920 message: err.to_string(),
921 });
922 }
923 };
924 #[cfg(feature = "fault-injection")]
925 {
926 let exec_ctx = self.exec_ctx.as_ref();
927 let flow_id = exec_ctx
928 .map(|ctx| ctx.flow_id.as_str())
929 .unwrap_or("unknown");
930 let node_id = exec_ctx.and_then(|ctx| ctx.node_id.as_deref());
931 let attempt = exec_ctx.map(|ctx| ctx.tenant.attempt).unwrap_or(1);
932 let fault_ctx = FaultContext {
933 pack_id: self.pack_id.as_str(),
934 flow_id,
935 node_id,
936 attempt,
937 };
938 if let Err(err) = maybe_fail(FaultPoint::StateWrite, fault_ctx) {
939 return Err(StateError {
940 code: "internal".into(),
941 message: err.to_string(),
942 });
943 }
944 }
945 let key = StoreStateKey::from(key);
946 let value = serde_json::from_slice(&bytes)
947 .unwrap_or_else(|_| Value::String(String::from_utf8_lossy(&bytes).to_string()));
948 match store.set_json(&tenant_ctx, STATE_PREFIX, &key, None, &value, None) {
949 Ok(()) => Ok(StateOpAck::Ok),
950 Err(err) => Err(StateError {
951 code: "internal".into(),
952 message: err.to_string(),
953 }),
954 }
955 }
956
957 fn delete(
958 &mut self,
959 key: HostStateKey,
960 ctx: Option<StateTenantCtx>,
961 ) -> Result<StateOpAck, StateError> {
962 let store = match self.state_store.as_ref() {
963 Some(store) => store.clone(),
964 None => {
965 return Err(StateError {
966 code: "unavailable".into(),
967 message: "state store not configured".into(),
968 });
969 }
970 };
971 let tenant_ctx = match self.tenant_ctx_from_v1(ctx) {
972 Ok(ctx) => ctx,
973 Err(err) => {
974 return Err(StateError {
975 code: "invalid-ctx".into(),
976 message: err.to_string(),
977 });
978 }
979 };
980 let key = StoreStateKey::from(key);
981 match store.del(&tenant_ctx, STATE_PREFIX, &key) {
982 Ok(_) => Ok(StateOpAck::Ok),
983 Err(err) => Err(StateError {
984 code: "internal".into(),
985 message: err.to_string(),
986 }),
987 }
988 }
989}
990
991impl TelemetryLoggerHost for HostState {
992 fn log(
993 &mut self,
994 span: TelemetrySpanContext,
995 fields: Vec<(String, String)>,
996 _ctx: Option<TelemetryTenantCtx>,
997 ) -> Result<TelemetryAck, TelemetryError> {
998 if let Some(mock) = &self.mocks
999 && mock.telemetry_drain(&[("span_json", span.flow_id.as_str())])
1000 {
1001 return Ok(TelemetryAck::Ok);
1002 }
1003 let mut map = serde_json::Map::new();
1004 for (k, v) in fields {
1005 map.insert(k, Value::String(v));
1006 }
1007 tracing::info!(
1008 tenant = %span.tenant,
1009 flow_id = %span.flow_id,
1010 node = ?span.node_id,
1011 provider = %span.provider,
1012 fields = %serde_json::Value::Object(map.clone()),
1013 "telemetry log from pack"
1014 );
1015 Ok(TelemetryAck::Ok)
1016 }
1017}
1018
1019impl RunnerHostHttp for HostState {
1020 fn request(
1021 &mut self,
1022 method: String,
1023 url: String,
1024 headers: Vec<String>,
1025 body: Option<Vec<u8>>,
1026 ) -> Result<Vec<u8>, String> {
1027 let req = HttpRequest {
1028 method,
1029 url,
1030 headers: headers
1031 .chunks(2)
1032 .filter_map(|chunk| {
1033 if chunk.len() == 2 {
1034 Some((chunk[0].clone(), chunk[1].clone()))
1035 } else {
1036 None
1037 }
1038 })
1039 .collect(),
1040 body,
1041 };
1042 match HttpClientHost::send(self, req, None) {
1043 Ok(resp) => Ok(resp.body.unwrap_or_default()),
1044 Err(err) => Err(err.message),
1045 }
1046 }
1047}
1048
1049impl RunnerHostKv for HostState {
1050 fn get(&mut self, _ns: String, _key: String) -> Option<String> {
1051 None
1052 }
1053
1054 fn put(&mut self, _ns: String, _key: String, _val: String) {}
1055}
1056
1057enum ManifestLoad {
1058 New {
1059 manifest: Box<greentic_types::PackManifest>,
1060 flows: PackFlows,
1061 },
1062 Legacy {
1063 manifest: Box<legacy_pack::PackManifest>,
1064 flows: PackFlows,
1065 },
1066}
1067
1068fn load_manifest_and_flows(path: &Path) -> Result<ManifestLoad> {
1069 let mut archive = ZipArchive::new(File::open(path)?)
1070 .with_context(|| format!("{} is not a valid gtpack", path.display()))?;
1071 let bytes = read_entry(&mut archive, "manifest.cbor")
1072 .with_context(|| format!("missing manifest.cbor in {}", path.display()))?;
1073 match decode_pack_manifest(&bytes) {
1074 Ok(manifest) => {
1075 let cache = PackFlows::from_manifest(manifest.clone());
1076 Ok(ManifestLoad::New {
1077 manifest: Box::new(manifest),
1078 flows: cache,
1079 })
1080 }
1081 Err(err) => {
1082 tracing::debug!(
1083 error = %err,
1084 pack = %path.display(),
1085 "decode_pack_manifest failed for archive; falling back to legacy manifest"
1086 );
1087 let legacy: legacy_pack::PackManifest = serde_cbor::from_slice(&bytes)
1088 .context("failed to decode legacy pack manifest from manifest.cbor")?;
1089 let flows = load_legacy_flows_from_archive(&mut archive, &legacy)?;
1090 Ok(ManifestLoad::Legacy {
1091 manifest: Box::new(legacy),
1092 flows,
1093 })
1094 }
1095 }
1096}
1097
1098fn load_manifest_and_flows_from_dir(root: &Path) -> Result<ManifestLoad> {
1099 let manifest_path = root.join("manifest.cbor");
1100 let bytes = std::fs::read(&manifest_path)
1101 .with_context(|| format!("missing manifest.cbor in {}", root.display()))?;
1102 match decode_pack_manifest(&bytes) {
1103 Ok(manifest) => {
1104 let cache = PackFlows::from_manifest(manifest.clone());
1105 Ok(ManifestLoad::New {
1106 manifest: Box::new(manifest),
1107 flows: cache,
1108 })
1109 }
1110 Err(err) => {
1111 tracing::debug!(
1112 error = %err,
1113 pack = %root.display(),
1114 "decode_pack_manifest failed for materialized pack; trying legacy manifest"
1115 );
1116 let legacy: legacy_pack::PackManifest = serde_cbor::from_slice(&bytes)
1117 .context("failed to decode legacy pack manifest from manifest.cbor")?;
1118 let flows = load_legacy_flows_from_dir(root, &legacy)?;
1119 Ok(ManifestLoad::Legacy {
1120 manifest: Box::new(legacy),
1121 flows,
1122 })
1123 }
1124 }
1125}
1126
1127fn load_legacy_flows_from_dir(
1128 root: &Path,
1129 manifest: &legacy_pack::PackManifest,
1130) -> Result<PackFlows> {
1131 build_legacy_flows(manifest, |rel_path| {
1132 let path = root.join(rel_path);
1133 std::fs::read(&path).with_context(|| format!("missing flow json {}", path.display()))
1134 })
1135}
1136
1137fn load_legacy_flows_from_archive(
1138 archive: &mut ZipArchive<File>,
1139 manifest: &legacy_pack::PackManifest,
1140) -> Result<PackFlows> {
1141 build_legacy_flows(manifest, |rel_path| {
1142 read_entry(archive, rel_path).with_context(|| format!("missing flow json {}", rel_path))
1143 })
1144}
1145
1146fn build_legacy_flows(
1147 manifest: &legacy_pack::PackManifest,
1148 mut read_json: impl FnMut(&str) -> Result<Vec<u8>>,
1149) -> Result<PackFlows> {
1150 let mut flows = HashMap::new();
1151 let mut descriptors = Vec::new();
1152
1153 for entry in &manifest.flows {
1154 let bytes = read_json(&entry.file_json)
1155 .with_context(|| format!("missing flow json {}", entry.file_json))?;
1156 let doc = parse_flow_doc_with_legacy_aliases(&bytes)?;
1157 let normalized = normalize_flow_doc(doc);
1158 let flow_ir = flow_doc_to_ir(normalized)?;
1159 let flow = flow_ir_to_flow(flow_ir)?;
1160
1161 descriptors.push(FlowDescriptor {
1162 id: entry.id.clone(),
1163 flow_type: entry.kind.clone(),
1164 pack_id: manifest.meta.pack_id.clone(),
1165 profile: manifest.meta.pack_id.clone(),
1166 version: manifest.meta.version.to_string(),
1167 description: None,
1168 });
1169 flows.insert(entry.id.clone(), flow);
1170 }
1171
1172 let mut entry_flows = manifest.meta.entry_flows.clone();
1173 if entry_flows.is_empty() {
1174 entry_flows = manifest.flows.iter().map(|f| f.id.clone()).collect();
1175 }
1176 let metadata = PackMetadata {
1177 pack_id: manifest.meta.pack_id.clone(),
1178 version: manifest.meta.version.to_string(),
1179 entry_flows,
1180 secret_requirements: Vec::new(),
1181 };
1182
1183 Ok(PackFlows {
1184 descriptors,
1185 flows,
1186 metadata,
1187 })
1188}
1189
1190fn parse_flow_doc_with_legacy_aliases(bytes: &[u8]) -> Result<FlowDoc> {
1191 let mut value: Value =
1192 serde_json::from_slice(bytes).context("failed to decode flow doc JSON")?;
1193 if let Some(map) = value.as_object_mut()
1194 && !map.contains_key("type")
1195 && let Some(flow_type) = map.remove("flow_type")
1196 {
1197 map.insert("type".to_string(), flow_type);
1198 }
1199 serde_json::from_value(value).context("failed to decode flow doc structure")
1200}
1201
1202pub struct ComponentState {
1203 pub host: HostState,
1204 wasi_ctx: WasiCtx,
1205 wasi_tls_ctx: WasiTlsCtx,
1206 wasi_http_ctx: WasiHttpCtx,
1207 resource_table: ResourceTable,
1208}
1209
1210impl ComponentState {
1211 pub fn new(host: HostState, policy: Arc<RunnerWasiPolicy>) -> Result<Self> {
1212 ensure_rustls_crypto_provider();
1213
1214 let wasi_ctx = policy
1215 .instantiate()
1216 .context("failed to build WASI context")?;
1217 Ok(Self {
1218 host,
1219 wasi_ctx,
1220 wasi_tls_ctx: WasiTlsCtxBuilder::new().build(),
1221 wasi_http_ctx: WasiHttpCtx::new(),
1222 resource_table: ResourceTable::new(),
1223 })
1224 }
1225
1226 fn host_mut(&mut self) -> &mut HostState {
1227 &mut self.host
1228 }
1229
1230 fn should_cancel_host(&mut self) -> bool {
1231 false
1232 }
1233
1234 fn yield_now_host(&mut self) {
1235 }
1237}
1238
1239impl component_api::v0_4::greentic::component::control::Host for ComponentState {
1240 fn should_cancel(&mut self) -> bool {
1241 self.should_cancel_host()
1242 }
1243
1244 fn yield_now(&mut self) {
1245 self.yield_now_host();
1246 }
1247}
1248
1249impl component_api::v0_5::greentic::component::control::Host for ComponentState {
1250 fn should_cancel(&mut self) -> bool {
1251 self.should_cancel_host()
1252 }
1253
1254 fn yield_now(&mut self) {
1255 self.yield_now_host();
1256 }
1257}
1258
1259fn add_component_control_instance(
1260 linker: &mut Linker<ComponentState>,
1261 name: &str,
1262) -> wasmtime::Result<()> {
1263 let mut inst = linker.instance(name)?;
1264 inst.func_wrap(
1265 "should-cancel",
1266 |mut caller: StoreContextMut<'_, ComponentState>, (): ()| {
1267 let host = caller.data_mut();
1268 Ok((host.should_cancel_host(),))
1269 },
1270 )?;
1271 inst.func_wrap(
1272 "yield-now",
1273 |mut caller: StoreContextMut<'_, ComponentState>, (): ()| {
1274 let host = caller.data_mut();
1275 host.yield_now_host();
1276 Ok(())
1277 },
1278 )?;
1279 Ok(())
1280}
1281
1282fn add_component_control_to_linker(linker: &mut Linker<ComponentState>) -> wasmtime::Result<()> {
1283 add_component_control_instance(linker, "greentic:component/control@0.5.0")?;
1284 add_component_control_instance(linker, "greentic:component/control@0.4.0")?;
1285 Ok(())
1286}
1287
1288pub fn register_all(linker: &mut Linker<ComponentState>, allow_state_store: bool) -> Result<()> {
1289 add_wasi_to_linker(linker)?;
1290
1291 let mut opts = LinkOptions::default();
1293 opts.tls(true);
1294 add_wasi_tls_to_linker(linker, &opts)?;
1295
1296 add_wasi_http_to_linker(linker)?;
1298
1299 add_all_v1_to_linker(
1300 linker,
1301 HostFns {
1302 http_client_v1_1: Some(|state: &mut ComponentState| state.host_mut()),
1303 http_client: Some(|state: &mut ComponentState| state.host_mut()),
1304 oauth_broker: None,
1305 runner_host_http: Some(|state: &mut ComponentState| state.host_mut()),
1306 runner_host_kv: Some(|state: &mut ComponentState| state.host_mut()),
1307 telemetry_logger: Some(|state: &mut ComponentState| state.host_mut()),
1308 state_store: allow_state_store.then_some(|state: &mut ComponentState| state.host_mut()),
1309 secrets_store_v1_1: Some(|state: &mut ComponentState| state.host_mut()),
1310 secrets_store: None,
1311 },
1312 )?;
1313 add_http_client_client_world_aliases(linker)?;
1314 Ok(())
1315}
1316
1317fn add_http_client_client_world_aliases(linker: &mut Linker<ComponentState>) -> Result<()> {
1318 let mut inst_v1_1 = linker.instance("greentic:http/client@1.1.0")?;
1319 inst_v1_1.func_wrap(
1320 "send",
1321 move |mut caller: StoreContextMut<'_, ComponentState>,
1322 (req, opts, ctx): (
1323 http_client_client_alias::Request,
1324 Option<http_client_client_alias::RequestOptions>,
1325 Option<http_client_client_alias::TenantCtx>,
1326 )| {
1327 let host = caller.data_mut().host_mut();
1328 let result = HttpClientHostV1_1::send(
1329 host,
1330 alias_request_to_host(req),
1331 opts.map(alias_request_options_to_host),
1332 ctx.map(alias_tenant_ctx_to_host),
1333 );
1334 Ok((match result {
1335 Ok(resp) => Ok(alias_response_from_host(resp)),
1336 Err(err) => Err(alias_error_from_host(err)),
1337 },))
1338 },
1339 )?;
1340 let mut inst_v1_0 = linker.instance("greentic:http/client@1.0.0")?;
1341 inst_v1_0.func_wrap(
1342 "send",
1343 move |mut caller: StoreContextMut<'_, ComponentState>,
1344 (req, ctx): (
1345 host_http_client::Request,
1346 Option<host_http_client::TenantCtx>,
1347 )| {
1348 let host = caller.data_mut().host_mut();
1349 let result = HttpClientHost::send(host, req, ctx);
1350 Ok((result,))
1351 },
1352 )?;
1353 Ok(())
1354}
1355
1356fn alias_request_to_host(req: http_client_client_alias::Request) -> host_http_client::RequestV1_1 {
1357 host_http_client::RequestV1_1 {
1358 method: req.method,
1359 url: req.url,
1360 headers: req.headers,
1361 body: req.body,
1362 }
1363}
1364
1365fn alias_request_options_to_host(
1366 opts: http_client_client_alias::RequestOptions,
1367) -> host_http_client::RequestOptionsV1_1 {
1368 host_http_client::RequestOptionsV1_1 {
1369 timeout_ms: opts.timeout_ms,
1370 allow_insecure: opts.allow_insecure,
1371 follow_redirects: opts.follow_redirects,
1372 }
1373}
1374
1375fn alias_tenant_ctx_to_host(
1376 ctx: http_client_client_alias::TenantCtx,
1377) -> host_http_client::TenantCtxV1_1 {
1378 host_http_client::TenantCtxV1_1 {
1379 env: ctx.env,
1380 tenant: ctx.tenant,
1381 tenant_id: ctx.tenant_id,
1382 team: ctx.team,
1383 team_id: ctx.team_id,
1384 user: ctx.user,
1385 user_id: ctx.user_id,
1386 trace_id: ctx.trace_id,
1387 correlation_id: ctx.correlation_id,
1388 i18n_id: ctx.i18n_id,
1389 attributes: ctx.attributes,
1390 session_id: ctx.session_id,
1391 flow_id: ctx.flow_id,
1392 node_id: ctx.node_id,
1393 provider_id: ctx.provider_id,
1394 deadline_ms: ctx.deadline_ms,
1395 attempt: ctx.attempt,
1396 idempotency_key: ctx.idempotency_key,
1397 impersonation: ctx.impersonation.map(|imp| http_types_v1_1::Impersonation {
1398 actor_id: imp.actor_id,
1399 reason: imp.reason,
1400 }),
1401 }
1402}
1403
1404fn alias_response_from_host(
1405 resp: host_http_client::ResponseV1_1,
1406) -> http_client_client_alias::Response {
1407 http_client_client_alias::Response {
1408 status: resp.status,
1409 headers: resp.headers,
1410 body: resp.body,
1411 }
1412}
1413
1414fn alias_error_from_host(
1415 err: host_http_client::HttpClientErrorV1_1,
1416) -> http_client_client_alias::HostError {
1417 http_client_client_alias::HostError {
1418 code: err.code,
1419 message: err.message,
1420 }
1421}
1422
1423impl OAuthHostContext for ComponentState {
1424 fn tenant_id(&self) -> &str {
1425 &self.host.config.tenant
1426 }
1427
1428 fn env(&self) -> &str {
1429 &self.host.default_env
1430 }
1431
1432 fn oauth_broker_host(&mut self) -> &mut OAuthBrokerHost {
1433 &mut self.host.oauth_host
1434 }
1435
1436 fn oauth_config(&self) -> Option<&OAuthBrokerConfig> {
1437 self.host.oauth_config.as_ref()
1438 }
1439}
1440
1441impl WasiView for ComponentState {
1442 fn ctx(&mut self) -> WasiCtxView<'_> {
1443 WasiCtxView {
1444 ctx: &mut self.wasi_ctx,
1445 table: &mut self.resource_table,
1446 }
1447 }
1448}
1449
1450impl WasiTlsView for ComponentState {
1451 fn tls(&mut self) -> WasiTlsCtxView<'_> {
1452 WasiTlsCtxView {
1453 ctx: &mut self.wasi_tls_ctx,
1454 table: &mut self.resource_table,
1455 }
1456 }
1457}
1458
1459impl WasiHttpView for ComponentState {
1460 fn http(&mut self) -> WasiHttpCtxView<'_> {
1461 WasiHttpCtxView {
1462 ctx: &mut self.wasi_http_ctx,
1463 table: &mut self.resource_table,
1464 hooks: Default::default(),
1465 }
1466 }
1467}
1468
1469#[allow(unsafe_code)]
1470unsafe impl Send for ComponentState {}
1471#[allow(unsafe_code)]
1472unsafe impl Sync for ComponentState {}
1473
1474impl PackRuntime {
1475 fn allows_state_store(&self, component_ref: &str) -> bool {
1476 if self.state_store.is_none() {
1477 return false;
1478 }
1479 if !self.config.state_store_policy.allow {
1480 return false;
1481 }
1482 let Some(manifest) = self.component_manifests.get(component_ref) else {
1483 return true;
1485 };
1486 manifest
1491 .capabilities
1492 .host
1493 .state
1494 .as_ref()
1495 .map(|caps| caps.read || caps.write)
1496 .unwrap_or(true)
1497 }
1498
1499 pub fn contains_component(&self, component_ref: &str) -> bool {
1500 self.components.contains_key(component_ref)
1501 }
1502
1503 pub fn state_store_handle(&self) -> Option<crate::storage::DynStateStore> {
1508 self.state_store.clone()
1509 }
1510
1511 #[allow(clippy::too_many_arguments)]
1512 pub async fn load(
1513 path: impl AsRef<Path>,
1514 config: Arc<HostConfig>,
1515 mocks: Option<Arc<MockLayer>>,
1516 archive_source: Option<&Path>,
1517 session_store: Option<DynSessionStore>,
1518 state_store: Option<DynStateStore>,
1519 wasi_policy: Arc<RunnerWasiPolicy>,
1520 secrets: DynSecretsManager,
1521 oauth_config: Option<OAuthBrokerConfig>,
1522 verify_archive: bool,
1523 component_resolution: ComponentResolution,
1524 ) -> Result<Self> {
1525 let path = path.as_ref();
1526 let (_pack_root, safe_path) = normalize_pack_path(path)?;
1527 let path_meta = std::fs::metadata(&safe_path).ok();
1528 let is_dir = path_meta
1529 .as_ref()
1530 .map(|meta| meta.is_dir())
1531 .unwrap_or(false);
1532 let is_component = !is_dir
1533 && safe_path
1534 .extension()
1535 .and_then(|ext| ext.to_str())
1536 .map(|ext| ext.eq_ignore_ascii_case("wasm"))
1537 .unwrap_or(false);
1538 let archive_hint_path = if let Some(source) = archive_source {
1539 let (_, normalized) = normalize_pack_path(source)?;
1540 Some(normalized)
1541 } else if is_component || is_dir {
1542 None
1543 } else {
1544 Some(safe_path.clone())
1545 };
1546 let archive_hint = archive_hint_path.as_deref();
1547 if verify_archive {
1548 if let Some(verify_target) = archive_hint.and_then(|p| {
1549 std::fs::metadata(p)
1550 .ok()
1551 .filter(|meta| meta.is_file())
1552 .map(|_| p)
1553 }) {
1554 verify::verify_pack(verify_target).await?;
1555 tracing::info!(pack_path = %verify_target.display(), "pack verification complete");
1556 } else {
1557 tracing::debug!("skipping archive verification (no archive source)");
1558 }
1559 }
1560 let engine = Engine::default();
1561 let engine_profile =
1562 EngineProfile::from_engine(&engine, CpuPolicy::Native, "default".to_string());
1563 let cache = CacheManager::new(CacheConfig::default(), engine_profile);
1564 let mut metadata = PackMetadata::fallback(&safe_path);
1565 let mut manifest = None;
1566 let mut legacy_manifest: Option<Box<legacy_pack::PackManifest>> = None;
1567 let mut flows = None;
1568 let materialized_root = component_resolution.materialized_root.clone().or_else(|| {
1569 if is_dir {
1570 Some(safe_path.clone())
1571 } else {
1572 None
1573 }
1574 });
1575 let (pack_assets_dir, assets_tempdir) =
1576 locate_pack_assets(materialized_root.as_deref(), archive_hint)?;
1577 let setup_yaml_exists = pack_assets_dir
1578 .as_ref()
1579 .map(|dir| dir.join("setup.yaml").is_file())
1580 .unwrap_or(false);
1581 tracing::info!(
1582 pack_root = %safe_path.display(),
1583 assets_setup_yaml_exists = setup_yaml_exists,
1584 "pack unpack metadata"
1585 );
1586
1587 if let Some(root) = materialized_root.as_ref() {
1588 match load_manifest_and_flows_from_dir(root) {
1589 Ok(ManifestLoad::New {
1590 manifest: m,
1591 flows: cache,
1592 }) => {
1593 metadata = cache.metadata.clone();
1594 manifest = Some(*m);
1595 flows = Some(cache);
1596 }
1597 Ok(ManifestLoad::Legacy {
1598 manifest: m,
1599 flows: cache,
1600 }) => {
1601 metadata = cache.metadata.clone();
1602 legacy_manifest = Some(m);
1603 flows = Some(cache);
1604 }
1605 Err(err) => {
1606 warn!(error = %err, pack = %root.display(), "failed to parse materialized pack manifest");
1607 }
1608 }
1609 }
1610
1611 if manifest.is_none()
1612 && legacy_manifest.is_none()
1613 && let Some(archive_path) = archive_hint
1614 {
1615 let manifest_load = load_manifest_and_flows(archive_path).with_context(|| {
1616 format!(
1617 "failed to load manifest.cbor from {}",
1618 archive_path.display()
1619 )
1620 })?;
1621 match manifest_load {
1622 ManifestLoad::New {
1623 manifest: m,
1624 flows: cache,
1625 } => {
1626 metadata = cache.metadata.clone();
1627 manifest = Some(*m);
1628 flows = Some(cache);
1629 }
1630 ManifestLoad::Legacy {
1631 manifest: m,
1632 flows: cache,
1633 } => {
1634 metadata = cache.metadata.clone();
1635 legacy_manifest = Some(m);
1636 flows = Some(cache);
1637 }
1638 }
1639 }
1640 #[cfg(feature = "fault-injection")]
1641 {
1642 let fault_ctx = FaultContext {
1643 pack_id: metadata.pack_id.as_str(),
1644 flow_id: "unknown",
1645 node_id: None,
1646 attempt: 1,
1647 };
1648 maybe_fail(FaultPoint::PackResolve, fault_ctx)
1649 .map_err(|err| anyhow!(err.to_string()))?;
1650 }
1651 let mut pack_lock = None;
1652 for root in find_pack_lock_roots(&safe_path, is_dir, archive_hint) {
1653 pack_lock = load_pack_lock(&root)?;
1654 if pack_lock.is_some() {
1655 break;
1656 }
1657 }
1658 let component_sources_payload = if pack_lock.is_none() {
1659 if let Some(manifest) = manifest.as_ref() {
1660 manifest
1661 .get_component_sources_v1()
1662 .context("invalid component sources extension")?
1663 } else {
1664 None
1665 }
1666 } else {
1667 None
1668 };
1669 let component_sources = if let Some(lock) = pack_lock.as_ref() {
1670 Some(component_sources_table_from_pack_lock(
1671 lock,
1672 component_resolution.allow_missing_hash,
1673 )?)
1674 } else {
1675 component_sources_table(component_sources_payload.as_ref())?
1676 };
1677 let components = if is_component {
1678 let wasm_bytes = fs::read(&safe_path).await?;
1679 metadata = PackMetadata::from_wasm(&wasm_bytes)
1680 .unwrap_or_else(|| PackMetadata::fallback(&safe_path));
1681 let name = safe_path
1682 .file_stem()
1683 .map(|s| s.to_string_lossy().to_string())
1684 .unwrap_or_else(|| "component".to_string());
1685 let component = compile_component_with_cache(&cache, &engine, None, wasm_bytes).await?;
1686 let mut map = HashMap::new();
1687 map.insert(
1688 name.clone(),
1689 PackComponent {
1690 name,
1691 version: metadata.version.clone(),
1692 component,
1693 },
1694 );
1695 map
1696 } else {
1697 let specs = component_specs(
1698 manifest.as_ref(),
1699 legacy_manifest.as_deref(),
1700 component_sources_payload.as_ref(),
1701 pack_lock.as_ref(),
1702 );
1703 if specs.is_empty() {
1704 HashMap::new()
1705 } else {
1706 let mut loaded = HashMap::new();
1707 let mut missing: HashSet<String> =
1708 specs.iter().map(|spec| spec.id.clone()).collect();
1709 let mut searched = Vec::new();
1710
1711 if !component_resolution.overrides.is_empty() {
1712 load_components_from_overrides(
1713 &cache,
1714 &engine,
1715 &component_resolution.overrides,
1716 &specs,
1717 &mut missing,
1718 &mut loaded,
1719 )
1720 .await?;
1721 searched.push("override map".to_string());
1722 }
1723
1724 if let Some(component_sources) = component_sources.as_ref() {
1725 load_components_from_sources(
1726 &cache,
1727 &engine,
1728 component_sources,
1729 &component_resolution,
1730 &specs,
1731 &mut missing,
1732 &mut loaded,
1733 materialized_root.as_deref(),
1734 archive_hint,
1735 )
1736 .await?;
1737 searched.push(format!("extension {}", EXT_COMPONENT_SOURCES_V1));
1738 }
1739
1740 if let Some(root) = materialized_root.as_ref() {
1741 load_components_from_dir(
1742 &cache,
1743 &engine,
1744 root,
1745 &specs,
1746 &mut missing,
1747 &mut loaded,
1748 )
1749 .await?;
1750 searched.push(format!("components dir {}", root.display()));
1751 }
1752
1753 if let Some(archive_path) = archive_hint {
1754 load_components_from_archive(
1755 &cache,
1756 &engine,
1757 archive_path,
1758 &specs,
1759 &mut missing,
1760 &mut loaded,
1761 )
1762 .await?;
1763 searched.push(format!("archive {}", archive_path.display()));
1764 }
1765
1766 if !missing.is_empty() {
1767 let missing_list = missing.into_iter().collect::<Vec<_>>().join(", ");
1768 let sources = if searched.is_empty() {
1769 "no component sources".to_string()
1770 } else {
1771 searched.join(", ")
1772 };
1773 bail!(
1774 "components missing: {}; looked in {}",
1775 missing_list,
1776 sources
1777 );
1778 }
1779
1780 loaded
1781 }
1782 };
1783 let http_client = Arc::clone(&HTTP_CLIENT);
1784 let mut component_manifests = HashMap::new();
1785 if let Some(manifest) = manifest.as_ref() {
1786 for component in &manifest.components {
1787 component_manifests.insert(component.id.as_str().to_string(), component.clone());
1788 }
1789 }
1790 let mut pack_policy = (*wasi_policy).clone();
1791 if let Some(dir) = pack_assets_dir {
1792 tracing::debug!(path = %dir.display(), "preopening pack assets directory for WASI /assets");
1793 pack_policy =
1794 pack_policy.with_preopen(PreopenSpec::new(dir, "/assets").read_only(true));
1795 }
1796 let wasi_policy = Arc::new(pack_policy);
1797 Ok(Self {
1798 path: safe_path,
1799 archive_path: archive_hint.map(Path::to_path_buf),
1800 config,
1801 engine,
1802 metadata,
1803 manifest,
1804 legacy_manifest,
1805 component_manifests,
1806 mocks,
1807 flows,
1808 components,
1809 http_client,
1810 pre_cache: Mutex::new(HashMap::new()),
1811 session_store,
1812 state_store,
1813 wasi_policy,
1814 assets_tempdir,
1815 provider_registry: RwLock::new(None),
1816 secrets,
1817 oauth_config,
1818 cache,
1819 })
1820 }
1821
1822 pub async fn list_flows(&self) -> Result<Vec<FlowDescriptor>> {
1823 if let Some(cache) = &self.flows {
1824 return Ok(cache.descriptors.clone());
1825 }
1826 if let Some(manifest) = &self.manifest {
1827 let descriptors = manifest
1828 .flows
1829 .iter()
1830 .map(|flow| FlowDescriptor {
1831 id: flow.id.as_str().to_string(),
1832 flow_type: flow_kind_to_str(flow.kind).to_string(),
1833 pack_id: manifest.pack_id.as_str().to_string(),
1834 profile: manifest.pack_id.as_str().to_string(),
1835 version: manifest.version.to_string(),
1836 description: None,
1837 })
1838 .collect();
1839 return Ok(descriptors);
1840 }
1841 Ok(Vec::new())
1842 }
1843
1844 #[allow(dead_code)]
1845 pub async fn run_flow(
1846 &self,
1847 flow_id: &str,
1848 input: serde_json::Value,
1849 ) -> Result<serde_json::Value> {
1850 let pack = Arc::new(
1851 PackRuntime::load(
1852 &self.path,
1853 Arc::clone(&self.config),
1854 self.mocks.clone(),
1855 self.archive_path.as_deref(),
1856 self.session_store.clone(),
1857 self.state_store.clone(),
1858 Arc::clone(&self.wasi_policy),
1859 self.secrets.clone(),
1860 self.oauth_config.clone(),
1861 false,
1862 ComponentResolution::default(),
1863 )
1864 .await?,
1865 );
1866
1867 let engine = FlowEngine::new(vec![Arc::clone(&pack)], Arc::clone(&self.config)).await?;
1868 let retry_config = self.config.retry_config().into();
1869 let mocks = pack.mocks.as_deref();
1870 let tenant = self.config.tenant.as_str();
1871
1872 let ctx = FlowContext {
1873 tenant,
1874 pack_id: pack.metadata().pack_id.as_str(),
1875 flow_id,
1876 node_id: None,
1877 tool: None,
1878 action: None,
1879 session_id: None,
1880 provider_id: None,
1881 retry_config,
1882 attempt: 1,
1883 observer: None,
1884 mocks,
1885 };
1886
1887 let execution = engine.execute(ctx, input).await?;
1888 match execution.status {
1889 FlowStatus::Completed => Ok(execution.output),
1890 FlowStatus::Waiting(wait) => Ok(serde_json::json!({
1891 "status": "pending",
1892 "reason": wait.reason,
1893 "resume": wait.snapshot,
1894 "response": execution.output,
1895 })),
1896 }
1897 }
1898
1899 pub async fn invoke_component(
1900 &self,
1901 component_ref: &str,
1902 ctx: ComponentExecCtx,
1903 operation: &str,
1904 config_json: Option<String>,
1905 input_json: String,
1906 ) -> Result<Value> {
1907 let pack_component = self
1908 .components
1909 .get(component_ref)
1910 .with_context(|| format!("component '{component_ref}' not found in pack"))?;
1911 let engine = self.engine.clone();
1912 let config = Arc::clone(&self.config);
1913 let http_client = Arc::clone(&self.http_client);
1914 let mocks = self.mocks.clone();
1915 let session_store = self.session_store.clone();
1916 let state_store = self.state_store.clone();
1917 let secrets = Arc::clone(&self.secrets);
1918 let oauth_config = self.oauth_config.clone();
1919 let wasi_policy = Arc::clone(&self.wasi_policy);
1920 let pack_id = self.metadata().pack_id.clone();
1921 let allow_state_store = self.allows_state_store(component_ref);
1922 let component = pack_component.component.clone();
1923 let component_ref_owned = component_ref.to_string();
1924 let operation_owned = operation.to_string();
1925 let input_owned =
1926 Self::merge_component_config_into_input_json(config_json.as_deref(), &input_json)
1927 .context("merge component config into invocation payload")?;
1928 let ctx_owned = ctx;
1929
1930 run_on_wasi_thread("component.invoke", move || {
1931 let mut linker = Linker::new(&engine);
1932 register_all(&mut linker, allow_state_store)?;
1933 add_component_control_to_linker(&mut linker)?;
1934
1935 let host_state = HostState::new(
1936 pack_id.clone(),
1937 config,
1938 http_client,
1939 mocks,
1940 session_store,
1941 state_store,
1942 secrets,
1943 oauth_config,
1944 Some(ctx_owned.clone()),
1945 Some(component_ref_owned.clone()),
1946 false,
1947 )?;
1948 let store_state = ComponentState::new(host_state, wasi_policy)?;
1949 let mut store = wasmtime::Store::new(&engine, store_state);
1950
1951 let invoke_result = HostState::instantiate_component_result(
1952 &mut linker,
1953 &mut store,
1954 &component,
1955 &ctx_owned,
1956 &component_ref_owned,
1957 &operation_owned,
1958 &input_owned,
1959 )?;
1960 HostState::convert_invoke_result(invoke_result)
1961 })
1962 }
1963
1964 fn merge_component_config_into_input_json(
1965 config_json: Option<&str>,
1966 input_json: &str,
1967 ) -> Result<String> {
1968 let Some(config_json) = config_json else {
1969 return Ok(input_json.to_string());
1970 };
1971
1972 let config_value: Value =
1973 serde_json::from_str(config_json).context("parse component config JSON")?;
1974
1975 if let Ok(mut invocation) =
1976 serde_json::from_str::<greentic_types::InvocationEnvelope>(input_json)
1977 {
1978 let payload_value = serde_json::from_slice(&invocation.payload).unwrap_or_else(|_| {
1979 Value::String(String::from_utf8_lossy(&invocation.payload).into_owned())
1980 });
1981 invocation.payload = serde_json::to_vec(&serde_json::json!({
1982 "config": config_value,
1983 "input": payload_value,
1984 }))
1985 .context("serialize merged invocation payload")?;
1986 return serde_json::to_string(&invocation)
1987 .context("serialize merged invocation envelope");
1988 }
1989
1990 let input_value = serde_json::from_str(input_json)
1991 .unwrap_or_else(|_| Value::String(input_json.to_string()));
1992 serde_json::to_string(&serde_json::json!({
1993 "config": config_value,
1994 "input": input_value,
1995 }))
1996 .context("serialize merged component input")
1997 }
1998
1999 pub fn resolve_provider(
2000 &self,
2001 provider_id: Option<&str>,
2002 provider_type: Option<&str>,
2003 ) -> Result<ProviderBinding> {
2004 let registry = self.provider_registry()?;
2005 registry.resolve(provider_id, provider_type)
2006 }
2007
2008 pub async fn invoke_provider(
2009 &self,
2010 binding: &ProviderBinding,
2011 ctx: ComponentExecCtx,
2012 op: &str,
2013 input_json: Vec<u8>,
2014 ) -> Result<Value> {
2015 let component_ref_owned = binding.component_ref.clone();
2016 let pack_component = self.components.get(&component_ref_owned).with_context(|| {
2017 format!("provider component '{component_ref_owned}' not found in pack")
2018 })?;
2019 let component = pack_component.component.clone();
2020
2021 let engine = self.engine.clone();
2022 let config = Arc::clone(&self.config);
2023 let http_client = Arc::clone(&self.http_client);
2024 let mocks = self.mocks.clone();
2025 let session_store = self.session_store.clone();
2026 let state_store = self.state_store.clone();
2027 let secrets = Arc::clone(&self.secrets);
2028 let oauth_config = self.oauth_config.clone();
2029 let wasi_policy = Arc::clone(&self.wasi_policy);
2030 let pack_id = self.metadata().pack_id.clone();
2031 let allow_state_store = self.allows_state_store(&component_ref_owned);
2032 let input_owned = input_json;
2033 let op_owned = op.to_string();
2034 let ctx_owned = ctx;
2035 let world = binding.world.clone();
2036
2037 run_on_wasi_thread("provider.invoke", move || {
2038 let mut linker = Linker::new(&engine);
2039 register_all(&mut linker, allow_state_store)?;
2040 add_component_control_to_linker(&mut linker)?;
2041 let host_state = HostState::new(
2042 pack_id.clone(),
2043 config,
2044 http_client,
2045 mocks,
2046 session_store,
2047 state_store,
2048 secrets,
2049 oauth_config,
2050 Some(ctx_owned.clone()),
2051 Some(component_ref_owned.clone()),
2052 true,
2053 )?;
2054 let store_state = ComponentState::new(host_state, wasi_policy)?;
2055 let mut store = wasmtime::Store::new(&engine, store_state);
2056 let use_schema_core_schema = world.contains("provider-schema-core");
2057 let use_schema_core_path = world.contains("provider/schema-core");
2058 let result = if use_schema_core_schema {
2059 let pre_instance = linker.instantiate_pre(component.as_ref())?;
2060 let pre: SchemaSchemaCorePre<ComponentState> =
2061 SchemaSchemaCorePre::new(pre_instance)?;
2062 let bindings = block_on(async { pre.instantiate_async(&mut store).await })?;
2063 let provider = bindings.greentic_provider_schema_core_schema_core_api();
2064 provider.call_invoke(&mut store, &op_owned, &input_owned)?
2065 } else if use_schema_core_path {
2066 let pre_instance = linker.instantiate_pre(component.as_ref())?;
2067 let path_attempt = (|| -> Result<Vec<u8>> {
2068 let pre: PathSchemaCorePre<ComponentState> =
2069 PathSchemaCorePre::new(pre_instance)?;
2070 let bindings = block_on(async { pre.instantiate_async(&mut store).await })?;
2071 let provider = bindings.greentic_provider_schema_core_api();
2072 Ok(provider.call_invoke(&mut store, &op_owned, &input_owned)?)
2073 })();
2074 match path_attempt {
2075 Ok(value) => value,
2076 Err(path_err)
2077 if path_err.to_string().contains("no exported instance named") =>
2078 {
2079 let pre_instance = linker.instantiate_pre(component.as_ref())?;
2080 let pre: SchemaSchemaCorePre<ComponentState> =
2081 SchemaSchemaCorePre::new(pre_instance)?;
2082 let bindings = block_on(async { pre.instantiate_async(&mut store).await })?;
2083 let provider = bindings.greentic_provider_schema_core_schema_core_api();
2084 provider.call_invoke(&mut store, &op_owned, &input_owned)?
2085 }
2086 Err(path_err) => return Err(path_err),
2087 }
2088 } else {
2089 let pre_instance = linker.instantiate_pre(component.as_ref())?;
2090 let pre: LegacySchemaCorePre<ComponentState> =
2091 LegacySchemaCorePre::new(pre_instance)?;
2092 let bindings = block_on(async { pre.instantiate_async(&mut store).await })?;
2093 let provider = bindings.greentic_provider_core_schema_core_api();
2094 provider.call_invoke(&mut store, &op_owned, &input_owned)?
2095 };
2096 deserialize_json_bytes(result)
2097 })
2098 }
2099
2100 pub(crate) fn provider_registry(&self) -> Result<ProviderRegistry> {
2101 if let Some(registry) = self.provider_registry.read().clone() {
2102 return Ok(registry);
2103 }
2104 let manifest = self
2105 .manifest
2106 .as_ref()
2107 .context("pack manifest required for provider resolution")?;
2108 let env = std::env::var("GREENTIC_ENV").unwrap_or_else(|_| "local".to_string());
2109 let registry = ProviderRegistry::new(
2110 manifest,
2111 self.state_store.clone(),
2112 &self.config.tenant,
2113 &env,
2114 )?;
2115 *self.provider_registry.write() = Some(registry.clone());
2116 Ok(registry)
2117 }
2118
2119 pub(crate) fn provider_registry_optional(&self) -> Result<Option<ProviderRegistry>> {
2120 if self.manifest.is_none() {
2121 return Ok(None);
2122 }
2123 Ok(Some(self.provider_registry()?))
2124 }
2125
2126 pub fn load_flow(&self, flow_id: &str) -> Result<Flow> {
2127 if let Some(cache) = &self.flows {
2128 return cache
2129 .flows
2130 .get(flow_id)
2131 .cloned()
2132 .ok_or_else(|| anyhow!("flow '{flow_id}' not found in pack"));
2133 }
2134 if let Some(manifest) = &self.manifest {
2135 let entry = manifest
2136 .flows
2137 .iter()
2138 .find(|f| f.id.as_str() == flow_id)
2139 .ok_or_else(|| anyhow!("flow '{flow_id}' not found in manifest"))?;
2140 return Ok(entry.flow.clone());
2141 }
2142 bail!("flow '{flow_id}' not available (pack exports disabled)")
2143 }
2144
2145 pub fn metadata(&self) -> &PackMetadata {
2146 &self.metadata
2147 }
2148
2149 pub fn read_asset(&self, asset_path: &str) -> Result<Vec<u8>> {
2154 let normalized = asset_path
2155 .trim_start_matches("assets/")
2156 .trim_start_matches("/assets/");
2157 if let Some(tempdir) = &self.assets_tempdir {
2159 let full = tempdir.path().join("assets").join(normalized);
2160 if full.exists() {
2161 return std::fs::read(&full)
2162 .with_context(|| format!("read asset {}", full.display()));
2163 }
2164 }
2165 let full = self.path.join("assets").join(normalized);
2167 if full.exists() {
2168 return std::fs::read(&full).with_context(|| format!("read asset {}", full.display()));
2169 }
2170 bail!("asset not found: {}", asset_path)
2171 }
2172
2173 pub fn component_manifest(&self, component_ref: &str) -> Option<&ComponentManifest> {
2174 self.component_manifests.get(component_ref)
2175 }
2176
2177 pub fn describe_component_contract_v0_6(&self, component_ref: &str) -> Result<Option<Value>> {
2178 let pack_component = self
2179 .components
2180 .get(component_ref)
2181 .with_context(|| format!("component '{component_ref}' not found in pack"))?;
2182 let engine = self.engine.clone();
2183 let config = Arc::clone(&self.config);
2184 let http_client = Arc::clone(&self.http_client);
2185 let mocks = self.mocks.clone();
2186 let session_store = self.session_store.clone();
2187 let state_store = self.state_store.clone();
2188 let secrets = Arc::clone(&self.secrets);
2189 let oauth_config = self.oauth_config.clone();
2190 let wasi_policy = Arc::clone(&self.wasi_policy);
2191 let pack_id = self.metadata().pack_id.clone();
2192 let allow_state_store = self.allows_state_store(component_ref);
2193 let component = pack_component.component.clone();
2194 let component_ref_owned = component_ref.to_string();
2195
2196 run_on_wasi_thread("component.describe", move || {
2197 let mut linker = Linker::new(&engine);
2198 register_all(&mut linker, allow_state_store)?;
2199 add_component_control_to_linker(&mut linker)?;
2200
2201 let host_state = HostState::new(
2202 pack_id.clone(),
2203 config,
2204 http_client,
2205 mocks,
2206 session_store,
2207 state_store,
2208 secrets,
2209 oauth_config,
2210 None,
2211 Some(component_ref_owned),
2212 false,
2213 )?;
2214 let store_state = ComponentState::new(host_state, wasi_policy)?;
2215 let mut store = wasmtime::Store::new(&engine, store_state);
2216 let pre_instance = linker.instantiate_pre(&component)?;
2217 let pre = match component_api::v0_6_descriptor::ComponentV0V6V0Pre::new(pre_instance) {
2218 Ok(pre) => pre,
2219 Err(_) => return Ok(None),
2220 };
2221 let bytes = block_on(async {
2222 let bindings = pre.instantiate_async(&mut store).await?;
2223 let descriptor = bindings.greentic_component_component_descriptor();
2224 descriptor.call_describe(&mut store)
2225 })?;
2226
2227 if bytes.is_empty() {
2228 return Ok(Some(Value::Null));
2229 }
2230 if let Ok(value) = serde_cbor::from_slice::<Value>(&bytes) {
2231 return Ok(Some(value));
2232 }
2233 if let Ok(value) = serde_json::from_slice::<Value>(&bytes) {
2234 return Ok(Some(value));
2235 }
2236 if let Ok(text) = String::from_utf8(bytes) {
2237 if let Ok(value) = serde_json::from_str::<Value>(&text) {
2238 return Ok(Some(value));
2239 }
2240 return Ok(Some(Value::String(text)));
2241 }
2242 Ok(Some(Value::Null))
2243 })
2244 }
2245
2246 pub fn load_schema_json(&self, schema_ref: &str) -> Result<Option<Value>> {
2247 let rel = normalize_schema_ref(schema_ref)?;
2248 if self.path.is_dir() {
2249 let candidate = self.path.join(&rel);
2250 if candidate.exists() {
2251 let bytes = std::fs::read(&candidate).with_context(|| {
2252 format!("failed to read schema file {}", candidate.display())
2253 })?;
2254 let value = serde_json::from_slice::<Value>(&bytes)
2255 .with_context(|| format!("invalid schema JSON in {}", candidate.display()))?;
2256 return Ok(Some(value));
2257 }
2258 }
2259
2260 if let Some(archive_path) = self
2261 .archive_path
2262 .as_ref()
2263 .or_else(|| path_is_gtpack(&self.path).then_some(&self.path))
2264 {
2265 let file = File::open(archive_path)
2266 .with_context(|| format!("failed to open {}", archive_path.display()))?;
2267 let mut archive = ZipArchive::new(file)
2268 .with_context(|| format!("failed to read pack {}", archive_path.display()))?;
2269 match archive.by_name(&rel) {
2270 Ok(mut entry) => {
2271 let mut bytes = Vec::new();
2272 entry.read_to_end(&mut bytes)?;
2273 let value = serde_json::from_slice::<Value>(&bytes).with_context(|| {
2274 format!("invalid schema JSON in {}:{}", archive_path.display(), rel)
2275 })?;
2276 Ok(Some(value))
2277 }
2278 Err(zip::result::ZipError::FileNotFound) => Ok(None),
2279 Err(err) => Err(anyhow!(err)).with_context(|| {
2280 format!(
2281 "failed to read schema `{}` from {}",
2282 rel,
2283 archive_path.display()
2284 )
2285 }),
2286 }
2287 } else {
2288 Ok(None)
2289 }
2290 }
2291
2292 pub fn required_secrets(&self) -> &[greentic_types::SecretRequirement] {
2293 &self.metadata.secret_requirements
2294 }
2295
2296 pub fn missing_secrets(
2297 &self,
2298 tenant_ctx: &TypesTenantCtx,
2299 ) -> Vec<greentic_types::SecretRequirement> {
2300 let env = tenant_ctx.env.as_str().to_string();
2301 let tenant = tenant_ctx.tenant.as_str().to_string();
2302 let team = tenant_ctx.team.as_ref().map(|t| t.as_str().to_string());
2303 self.required_secrets()
2304 .iter()
2305 .filter(|req| {
2306 if let Some(scope) = &req.scope {
2308 if scope.env != env {
2309 return false;
2310 }
2311 if scope.tenant != tenant {
2312 return false;
2313 }
2314 if let Some(ref team_req) = scope.team
2315 && team.as_ref() != Some(team_req)
2316 {
2317 return false;
2318 }
2319 }
2320 let ctx = self.config.tenant_ctx();
2321 read_secret_blocking(
2322 &self.secrets,
2323 &ctx,
2324 &self.metadata.pack_id,
2325 canonicalize_secret_key(req.key.as_str()).as_str(),
2326 )
2327 .is_err()
2328 })
2329 .cloned()
2330 .collect()
2331 }
2332
2333 pub fn for_component_test(
2334 components: Vec<(String, PathBuf)>,
2335 flows: HashMap<String, FlowIR>,
2336 pack_id: &str,
2337 config: Arc<HostConfig>,
2338 ) -> Result<Self> {
2339 let engine = Engine::default();
2340 let engine_profile =
2341 EngineProfile::from_engine(&engine, CpuPolicy::Native, "default".to_string());
2342 let cache = CacheManager::new(CacheConfig::default(), engine_profile);
2343 let mut component_map = HashMap::new();
2344 for (name, path) in components {
2345 if !path.exists() {
2346 bail!("component artifact missing: {}", path.display());
2347 }
2348 let wasm_bytes = std::fs::read(&path)?;
2349 let component =
2350 Arc::new(Component::from_binary(&engine, &wasm_bytes).map_err(|err| {
2351 anyhow!("failed to compile component {}: {err}", path.display())
2352 })?);
2353 component_map.insert(
2354 name.clone(),
2355 PackComponent {
2356 name,
2357 version: "0.0.0".into(),
2358 component,
2359 },
2360 );
2361 }
2362
2363 let mut flow_map = HashMap::new();
2364 let mut descriptors = Vec::new();
2365 for (id, ir) in flows {
2366 let flow_type = ir.flow_type.clone();
2367 let flow = flow_ir_to_flow(ir)?;
2368 flow_map.insert(id.clone(), flow);
2369 descriptors.push(FlowDescriptor {
2370 id: id.clone(),
2371 flow_type,
2372 pack_id: pack_id.to_string(),
2373 profile: "test".into(),
2374 version: "0.0.0".into(),
2375 description: None,
2376 });
2377 }
2378 let entry_flows = descriptors.iter().map(|flow| flow.id.clone()).collect();
2379 let metadata = PackMetadata {
2380 pack_id: pack_id.to_string(),
2381 version: "0.0.0".into(),
2382 entry_flows,
2383 secret_requirements: Vec::new(),
2384 };
2385 let flows_cache = PackFlows {
2386 descriptors: descriptors.clone(),
2387 flows: flow_map,
2388 metadata: metadata.clone(),
2389 };
2390
2391 Ok(Self {
2392 path: PathBuf::new(),
2393 archive_path: None,
2394 config,
2395 engine,
2396 metadata,
2397 manifest: None,
2398 legacy_manifest: None,
2399 component_manifests: HashMap::new(),
2400 mocks: None,
2401 flows: Some(flows_cache),
2402 components: component_map,
2403 http_client: Arc::clone(&HTTP_CLIENT),
2404 pre_cache: Mutex::new(HashMap::new()),
2405 session_store: None,
2406 state_store: None,
2407 wasi_policy: Arc::new(RunnerWasiPolicy::new()),
2408 assets_tempdir: None,
2409 provider_registry: RwLock::new(None),
2410 secrets: crate::secrets::default_manager()?,
2411 oauth_config: None,
2412 cache,
2413 })
2414 }
2415}
2416
2417fn normalize_schema_ref(schema_ref: &str) -> Result<String> {
2418 let candidate = schema_ref.trim();
2419 if candidate.is_empty() {
2420 bail!("schema ref cannot be empty");
2421 }
2422 let path = Path::new(candidate);
2423 if path.is_absolute() {
2424 bail!("schema ref must be relative: {}", schema_ref);
2425 }
2426 let mut normalized = PathBuf::new();
2427 for component in path.components() {
2428 match component {
2429 std::path::Component::Normal(part) => normalized.push(part),
2430 std::path::Component::CurDir => {}
2431 _ => bail!("schema ref must not contain traversal: {}", schema_ref),
2432 }
2433 }
2434 let normalized = normalized
2435 .to_str()
2436 .map(ToString::to_string)
2437 .ok_or_else(|| anyhow!("schema ref must be valid UTF-8"))?;
2438 if normalized.is_empty() {
2439 bail!("schema ref cannot normalize to empty path");
2440 }
2441 Ok(normalized)
2442}
2443
2444fn path_is_gtpack(path: &Path) -> bool {
2445 path.extension()
2446 .and_then(|ext| ext.to_str())
2447 .map(|ext| ext.eq_ignore_ascii_case("gtpack"))
2448 .unwrap_or(false)
2449}
2450
2451fn is_missing_node_export(err: &wasmtime::Error, version: &str) -> bool {
2452 let message = err.to_string();
2453 message.contains("no exported instance named")
2454 && message.contains(&format!("greentic:component/node@{version}"))
2455}
2456
2457struct PackFlows {
2458 descriptors: Vec<FlowDescriptor>,
2459 flows: HashMap<String, Flow>,
2460 metadata: PackMetadata,
2461}
2462
2463const RUNTIME_FLOW_EXTENSION_IDS: [&str; 3] = [
2464 "greentic.pack.runtime_flow",
2465 "greentic.pack.flow_runtime",
2466 "greentic.pack.runtime_flows",
2467];
2468
2469#[derive(Debug, Deserialize)]
2470struct RuntimeFlowBundle {
2471 flows: Vec<RuntimeFlow>,
2472}
2473
2474#[derive(Debug, Deserialize)]
2475struct RuntimeFlow {
2476 id: String,
2477 #[serde(alias = "flow_type")]
2478 kind: FlowKind,
2479 #[serde(default)]
2480 schema_version: Option<String>,
2481 #[serde(default)]
2482 start: Option<String>,
2483 #[serde(default)]
2484 entrypoints: BTreeMap<String, Value>,
2485 nodes: BTreeMap<String, RuntimeNode>,
2486 #[serde(default)]
2487 metadata: Option<FlowMetadata>,
2488}
2489
2490#[derive(Debug, Deserialize)]
2491struct RuntimeNode {
2492 #[serde(alias = "component")]
2493 component_id: String,
2494 #[serde(default, alias = "operation")]
2495 operation_name: Option<String>,
2496 #[serde(default, alias = "payload", alias = "input")]
2497 operation_payload: Value,
2498 #[serde(default)]
2499 config: Value,
2500 #[serde(default)]
2501 routing: Option<Routing>,
2502 #[serde(default)]
2503 telemetry: Option<TelemetryHints>,
2504}
2505
2506fn deserialize_json_bytes(bytes: Vec<u8>) -> Result<Value> {
2507 if bytes.is_empty() {
2508 return Ok(Value::Null);
2509 }
2510 serde_json::from_slice(&bytes).or_else(|_| {
2511 String::from_utf8(bytes)
2512 .map(Value::String)
2513 .map_err(|err| anyhow!(err))
2514 })
2515}
2516
2517impl PackFlows {
2518 fn from_manifest(manifest: greentic_types::PackManifest) -> Self {
2519 if let Some(flows) = flows_from_runtime_extension(&manifest) {
2520 return flows;
2521 }
2522 let descriptors = manifest
2523 .flows
2524 .iter()
2525 .map(|entry| FlowDescriptor {
2526 id: entry.id.as_str().to_string(),
2527 flow_type: flow_kind_to_str(entry.kind).to_string(),
2528 pack_id: manifest.pack_id.as_str().to_string(),
2529 profile: manifest.pack_id.as_str().to_string(),
2530 version: manifest.version.to_string(),
2531 description: None,
2532 })
2533 .collect();
2534 let mut flows = HashMap::new();
2535 for entry in &manifest.flows {
2536 flows.insert(entry.id.as_str().to_string(), entry.flow.clone());
2537 }
2538 Self {
2539 metadata: PackMetadata::from_manifest(&manifest),
2540 descriptors,
2541 flows,
2542 }
2543 }
2544}
2545
2546fn flows_from_runtime_extension(manifest: &greentic_types::PackManifest) -> Option<PackFlows> {
2547 let extensions = manifest.extensions.as_ref()?;
2548 let extension = extensions.iter().find_map(|(key, ext)| {
2549 if RUNTIME_FLOW_EXTENSION_IDS
2550 .iter()
2551 .any(|candidate| candidate == key)
2552 {
2553 Some(ext)
2554 } else {
2555 None
2556 }
2557 })?;
2558 let runtime_flows = match decode_runtime_flow_extension(extension) {
2559 Some(flows) if !flows.is_empty() => flows,
2560 _ => return None,
2561 };
2562
2563 let descriptors = runtime_flows
2564 .iter()
2565 .map(|flow| FlowDescriptor {
2566 id: flow.id.as_str().to_string(),
2567 flow_type: flow_kind_to_str(flow.kind).to_string(),
2568 pack_id: manifest.pack_id.as_str().to_string(),
2569 profile: manifest.pack_id.as_str().to_string(),
2570 version: manifest.version.to_string(),
2571 description: None,
2572 })
2573 .collect::<Vec<_>>();
2574 let flows = runtime_flows
2575 .into_iter()
2576 .map(|flow| (flow.id.as_str().to_string(), flow))
2577 .collect();
2578
2579 Some(PackFlows {
2580 metadata: PackMetadata::from_manifest(manifest),
2581 descriptors,
2582 flows,
2583 })
2584}
2585
2586fn decode_runtime_flow_extension(extension: &ExtensionRef) -> Option<Vec<Flow>> {
2587 let value = match extension.inline.as_ref()? {
2588 ExtensionInline::Other(value) => value.clone(),
2589 _ => return None,
2590 };
2591
2592 if let Ok(bundle) = serde_json::from_value::<RuntimeFlowBundle>(value.clone()) {
2593 return Some(collect_runtime_flows(bundle.flows));
2594 }
2595
2596 if let Ok(flows) = serde_json::from_value::<Vec<RuntimeFlow>>(value.clone()) {
2597 return Some(collect_runtime_flows(flows));
2598 }
2599
2600 if let Ok(flows) = serde_json::from_value::<Vec<Flow>>(value) {
2601 return Some(flows);
2602 }
2603
2604 warn!(
2605 extension = %extension.kind,
2606 version = %extension.version,
2607 "runtime flow extension present but could not be decoded"
2608 );
2609 None
2610}
2611
2612fn collect_runtime_flows(flows: Vec<RuntimeFlow>) -> Vec<Flow> {
2613 flows
2614 .into_iter()
2615 .filter_map(|flow| match runtime_flow_to_flow(flow) {
2616 Ok(flow) => Some(flow),
2617 Err(err) => {
2618 warn!(error = %err, "failed to decode runtime flow");
2619 None
2620 }
2621 })
2622 .collect()
2623}
2624
2625fn runtime_flow_to_flow(runtime: RuntimeFlow) -> Result<Flow> {
2626 let flow_id = FlowId::from_str(&runtime.id)
2627 .with_context(|| format!("invalid flow id `{}`", runtime.id))?;
2628 let mut entrypoints = runtime.entrypoints;
2629 if entrypoints.is_empty()
2630 && let Some(start) = &runtime.start
2631 {
2632 entrypoints.insert("default".into(), Value::String(start.clone()));
2633 }
2634
2635 let mut nodes: IndexMap<NodeId, Node, FlowHasher> = IndexMap::default();
2636 for (id, node) in runtime.nodes {
2637 let node_id = NodeId::from_str(&id).with_context(|| format!("invalid node id `{id}`"))?;
2638 let component_id = ComponentId::from_str(&node.component_id)
2639 .with_context(|| format!("invalid component id `{}`", node.component_id))?;
2640 let operation_payload = if node.config.is_null() {
2641 node.operation_payload
2642 } else {
2643 serde_json::json!({
2644 "input": node.operation_payload,
2645 "config": node.config,
2646 })
2647 };
2648 let component = FlowComponentRef {
2649 id: component_id,
2650 pack_alias: None,
2651 operation: node.operation_name,
2652 };
2653 let routing = node.routing.unwrap_or(Routing::End);
2654 let telemetry = node.telemetry.unwrap_or_default();
2655 nodes.insert(
2656 node_id.clone(),
2657 Node {
2658 id: node_id,
2659 component,
2660 input: InputMapping {
2661 mapping: operation_payload,
2662 },
2663 output: OutputMapping {
2664 mapping: Value::Null,
2665 },
2666 err_map: None,
2667 routing,
2668 telemetry,
2669 },
2670 );
2671 }
2672
2673 Ok(Flow {
2674 schema_version: runtime.schema_version.unwrap_or_else(|| "1.0".to_string()),
2675 id: flow_id,
2676 kind: runtime.kind,
2677 entrypoints,
2678 nodes,
2679 metadata: runtime.metadata.unwrap_or_default(),
2680 })
2681}
2682
2683fn flow_kind_to_str(kind: greentic_types::FlowKind) -> &'static str {
2684 match kind {
2685 greentic_types::FlowKind::Messaging => "messaging",
2686 greentic_types::FlowKind::Event => "event",
2687 greentic_types::FlowKind::ComponentConfig => "component-config",
2688 greentic_types::FlowKind::Job => "job",
2689 greentic_types::FlowKind::Http => "http",
2690 }
2691}
2692
2693fn read_entry(archive: &mut ZipArchive<File>, name: &str) -> Result<Vec<u8>> {
2694 let mut file = archive
2695 .by_name(name)
2696 .with_context(|| format!("entry {name} missing from archive"))?;
2697 let mut buf = Vec::new();
2698 file.read_to_end(&mut buf)?;
2699 Ok(buf)
2700}
2701
2702fn normalize_flow_doc(mut doc: FlowDoc) -> FlowDoc {
2703 for node in doc.nodes.values_mut() {
2704 let Some((component_ref, payload)) = node
2705 .raw
2706 .iter()
2707 .next()
2708 .map(|(key, value)| (key.clone(), value.clone()))
2709 else {
2710 continue;
2711 };
2712 if component_ref.starts_with("emit.") {
2713 node.operation = Some(component_ref);
2714 node.payload = payload;
2715 node.raw.clear();
2716 continue;
2717 }
2718 let (target_component, operation, input, config) =
2719 infer_component_exec(&payload, &component_ref);
2720 let mut payload_obj = serde_json::Map::new();
2721 payload_obj.insert("component".into(), Value::String(target_component));
2723 payload_obj.insert("operation".into(), Value::String(operation));
2724 payload_obj.insert("input".into(), input);
2725 if let Some(cfg) = config {
2726 payload_obj.insert("config".into(), cfg);
2727 }
2728 node.operation = Some("component.exec".to_string());
2729 node.payload = Value::Object(payload_obj);
2730 node.raw.clear();
2731 }
2732 doc
2733}
2734
2735fn infer_component_exec(
2736 payload: &Value,
2737 component_ref: &str,
2738) -> (String, String, Value, Option<Value>) {
2739 let default_op = if component_ref.starts_with("templating.") {
2740 "render"
2741 } else {
2742 "invoke"
2743 }
2744 .to_string();
2745
2746 if let Value::Object(map) = payload {
2747 let has_embedded_component =
2748 map.get("component").is_some() || map.get("component_ref").is_some();
2749 let op = map
2750 .get("op")
2751 .or_else(|| map.get("operation"))
2752 .and_then(Value::as_str)
2753 .map(|s| s.to_string())
2754 .unwrap_or_else(|| {
2755 if has_embedded_component {
2756 component_ref.to_string()
2757 } else {
2758 default_op.clone()
2759 }
2760 });
2761
2762 let mut input = map.clone();
2763 let config = input.remove("config");
2764 let canonical_input = if has_embedded_component {
2765 input.get("input").cloned()
2766 } else {
2767 None
2768 };
2769 let component = input
2770 .get("component")
2771 .or_else(|| input.get("component_ref"))
2772 .and_then(Value::as_str)
2773 .map(|s| s.to_string())
2774 .unwrap_or_else(|| component_ref.to_string());
2775 input.remove("component");
2776 input.remove("component_ref");
2777 input.remove("op");
2778 input.remove("operation");
2779 let input = canonical_input.unwrap_or(Value::Object(input));
2780 return (component, op, input, config);
2781 }
2782
2783 (component_ref.to_string(), default_op, payload.clone(), None)
2784}
2785
2786#[derive(Clone, Debug)]
2787struct ComponentSpec {
2788 id: String,
2789 version: String,
2790 legacy_path: Option<String>,
2791}
2792
2793#[derive(Clone, Debug)]
2794struct ComponentSourceInfo {
2795 digest: Option<String>,
2796 source: ComponentSourceRef,
2797 artifact: ComponentArtifactLocation,
2798 expected_wasm_sha256: Option<String>,
2799 skip_digest_verification: bool,
2800}
2801
2802#[derive(Clone, Debug)]
2803enum ComponentArtifactLocation {
2804 Inline { wasm_path: String },
2805 Remote,
2806}
2807
2808#[derive(Clone, Debug, Deserialize)]
2809struct PackLockV1 {
2810 schema_version: u32,
2811 components: Vec<PackLockComponent>,
2812}
2813
2814#[derive(Clone, Debug, Deserialize)]
2815struct PackLockComponent {
2816 name: String,
2817 #[serde(default, rename = "source_ref")]
2818 source_ref: Option<String>,
2819 #[serde(default, rename = "ref")]
2820 legacy_ref: Option<String>,
2821 #[serde(default)]
2822 component_id: Option<ComponentId>,
2823 #[serde(default)]
2824 bundled: Option<bool>,
2825 #[serde(default, rename = "bundled_path")]
2826 bundled_path: Option<String>,
2827 #[serde(default, rename = "path")]
2828 legacy_path: Option<String>,
2829 #[serde(default)]
2830 wasm_sha256: Option<String>,
2831 #[serde(default, rename = "sha256")]
2832 legacy_sha256: Option<String>,
2833 #[serde(default)]
2834 resolved_digest: Option<String>,
2835 #[serde(default)]
2836 digest: Option<String>,
2837}
2838
2839fn component_specs(
2840 manifest: Option<&greentic_types::PackManifest>,
2841 legacy_manifest: Option<&legacy_pack::PackManifest>,
2842 component_sources: Option<&ComponentSourcesV1>,
2843 pack_lock: Option<&PackLockV1>,
2844) -> Vec<ComponentSpec> {
2845 if let Some(manifest) = manifest {
2846 if !manifest.components.is_empty() {
2847 return manifest
2848 .components
2849 .iter()
2850 .map(|entry| ComponentSpec {
2851 id: entry.id.as_str().to_string(),
2852 version: entry.version.to_string(),
2853 legacy_path: None,
2854 })
2855 .collect();
2856 }
2857 if let Some(lock) = pack_lock {
2858 let mut seen = HashSet::new();
2859 let mut specs = Vec::new();
2860 for entry in &lock.components {
2861 let id = entry
2862 .component_id
2863 .as_ref()
2864 .map(|id| id.as_str())
2865 .unwrap_or(entry.name.as_str());
2866 if seen.insert(id.to_string()) {
2867 specs.push(ComponentSpec {
2868 id: id.to_string(),
2869 version: "0.0.0".to_string(),
2870 legacy_path: None,
2871 });
2872 }
2873 }
2874 return specs;
2875 }
2876 if let Some(sources) = component_sources {
2877 let mut seen = HashSet::new();
2878 let mut specs = Vec::new();
2879 for entry in &sources.components {
2880 let id = entry
2881 .component_id
2882 .as_ref()
2883 .map(|id| id.as_str())
2884 .unwrap_or(entry.name.as_str());
2885 if seen.insert(id.to_string()) {
2886 specs.push(ComponentSpec {
2887 id: id.to_string(),
2888 version: "0.0.0".to_string(),
2889 legacy_path: None,
2890 });
2891 }
2892 }
2893 return specs;
2894 }
2895 }
2896 if let Some(legacy_manifest) = legacy_manifest {
2897 return legacy_manifest
2898 .components
2899 .iter()
2900 .map(|entry| ComponentSpec {
2901 id: entry.name.clone(),
2902 version: entry.version.to_string(),
2903 legacy_path: Some(entry.file_wasm.clone()),
2904 })
2905 .collect();
2906 }
2907 Vec::new()
2908}
2909
2910fn component_sources_table(
2911 sources: Option<&ComponentSourcesV1>,
2912) -> Result<Option<HashMap<String, ComponentSourceInfo>>> {
2913 let Some(sources) = sources else {
2914 return Ok(None);
2915 };
2916 let mut table = HashMap::new();
2917 for entry in &sources.components {
2918 let artifact = match &entry.artifact {
2919 ArtifactLocationV1::Inline { wasm_path, .. } => ComponentArtifactLocation::Inline {
2920 wasm_path: wasm_path.clone(),
2921 },
2922 ArtifactLocationV1::Remote => ComponentArtifactLocation::Remote,
2923 };
2924 let info = ComponentSourceInfo {
2925 digest: Some(entry.resolved.digest.clone()),
2926 source: entry.source.clone(),
2927 artifact,
2928 expected_wasm_sha256: None,
2929 skip_digest_verification: false,
2930 };
2931 if let Some(component_id) = entry.component_id.as_ref() {
2932 table.insert(component_id.as_str().to_string(), info.clone());
2933 }
2934 table.insert(entry.name.clone(), info);
2935 }
2936 Ok(Some(table))
2937}
2938
2939fn load_pack_lock(path: &Path) -> Result<Option<PackLockV1>> {
2940 let lock_path = if path.is_dir() {
2941 let candidate = path.join("pack.lock");
2942 if candidate.exists() {
2943 Some(candidate)
2944 } else {
2945 let candidate = path.join("pack.lock.json");
2946 candidate.exists().then_some(candidate)
2947 }
2948 } else {
2949 None
2950 };
2951 let Some(lock_path) = lock_path else {
2952 return Ok(None);
2953 };
2954 let raw = std::fs::read_to_string(&lock_path)
2955 .with_context(|| format!("failed to read {}", lock_path.display()))?;
2956 let lock: PackLockV1 = serde_json::from_str(&raw).context("failed to parse pack.lock")?;
2957 if lock.schema_version != 1 {
2958 bail!("pack.lock schema_version must be 1");
2959 }
2960 Ok(Some(lock))
2961}
2962
2963fn find_pack_lock_roots(
2964 pack_path: &Path,
2965 is_dir: bool,
2966 archive_hint: Option<&Path>,
2967) -> Vec<PathBuf> {
2968 if is_dir {
2969 return vec![pack_path.to_path_buf()];
2970 }
2971 let mut roots = Vec::new();
2972 if let Some(archive_path) = archive_hint {
2973 if let Some(parent) = archive_path.parent() {
2974 roots.push(parent.to_path_buf());
2975 if let Some(grandparent) = parent.parent() {
2976 roots.push(grandparent.to_path_buf());
2977 }
2978 }
2979 } else if let Some(parent) = pack_path.parent() {
2980 roots.push(parent.to_path_buf());
2981 if let Some(grandparent) = parent.parent() {
2982 roots.push(grandparent.to_path_buf());
2983 }
2984 }
2985 roots
2986}
2987
2988fn normalize_sha256(digest: &str) -> Result<String> {
2989 let trimmed = digest.trim();
2990 if trimmed.is_empty() {
2991 bail!("sha256 digest cannot be empty");
2992 }
2993 if let Some(stripped) = trimmed.strip_prefix("sha256:") {
2994 if stripped.is_empty() {
2995 bail!("sha256 digest must include hex bytes after sha256:");
2996 }
2997 return Ok(trimmed.to_string());
2998 }
2999 if trimmed.chars().all(|c| c.is_ascii_hexdigit()) {
3000 return Ok(format!("sha256:{trimmed}"));
3001 }
3002 bail!("sha256 digest must be hex or sha256:<hex>");
3003}
3004
3005fn component_sources_table_from_pack_lock(
3006 lock: &PackLockV1,
3007 allow_missing_hash: bool,
3008) -> Result<HashMap<String, ComponentSourceInfo>> {
3009 let mut table = HashMap::new();
3010 let mut names = HashSet::new();
3011 for entry in &lock.components {
3012 if !names.insert(entry.name.clone()) {
3013 bail!(
3014 "pack.lock contains duplicate component name `{}`",
3015 entry.name
3016 );
3017 }
3018 let source_ref = match (&entry.source_ref, &entry.legacy_ref) {
3019 (Some(primary), Some(legacy)) => {
3020 if primary != legacy {
3021 bail!(
3022 "pack.lock component {} has conflicting refs: {} vs {}",
3023 entry.name,
3024 primary,
3025 legacy
3026 );
3027 }
3028 primary.as_str()
3029 }
3030 (Some(primary), None) => primary.as_str(),
3031 (None, Some(legacy)) => legacy.as_str(),
3032 (None, None) => {
3033 bail!("pack.lock component {} missing source_ref", entry.name);
3034 }
3035 };
3036 let source: ComponentSourceRef = source_ref
3037 .parse()
3038 .with_context(|| format!("invalid component ref `{}`", source_ref))?;
3039 let bundled_path = match (&entry.bundled_path, &entry.legacy_path) {
3040 (Some(primary), Some(legacy)) => {
3041 if primary != legacy {
3042 bail!(
3043 "pack.lock component {} has conflicting bundled paths: {} vs {}",
3044 entry.name,
3045 primary,
3046 legacy
3047 );
3048 }
3049 Some(primary.clone())
3050 }
3051 (Some(primary), None) => Some(primary.clone()),
3052 (None, Some(legacy)) => Some(legacy.clone()),
3053 (None, None) => None,
3054 };
3055 let bundled = entry.bundled.unwrap_or(false) || bundled_path.is_some();
3056 let (artifact, digest, expected_wasm_sha256, skip_digest_verification) = if bundled {
3057 let wasm_path = bundled_path.ok_or_else(|| {
3058 anyhow!(
3059 "pack.lock component {} marked bundled but bundled_path is missing",
3060 entry.name
3061 )
3062 })?;
3063 let expected_raw = match (&entry.wasm_sha256, &entry.legacy_sha256) {
3064 (Some(primary), Some(legacy)) => {
3065 if primary != legacy {
3066 bail!(
3067 "pack.lock component {} has conflicting wasm_sha256 values: {} vs {}",
3068 entry.name,
3069 primary,
3070 legacy
3071 );
3072 }
3073 Some(primary.as_str())
3074 }
3075 (Some(primary), None) => Some(primary.as_str()),
3076 (None, Some(legacy)) => Some(legacy.as_str()),
3077 (None, None) => None,
3078 };
3079 let expected = match expected_raw {
3080 Some(value) => Some(normalize_sha256(value)?),
3081 None => None,
3082 };
3083 if expected.is_none() && !allow_missing_hash {
3084 bail!(
3085 "pack.lock component {} missing wasm_sha256 for bundled component",
3086 entry.name
3087 );
3088 }
3089 (
3090 ComponentArtifactLocation::Inline { wasm_path },
3091 expected.clone(),
3092 expected,
3093 allow_missing_hash && expected_raw.is_none(),
3094 )
3095 } else {
3096 if source.is_tag() {
3097 bail!(
3098 "component {} uses tag ref {} but is not bundled; rebuild the pack",
3099 entry.name,
3100 source
3101 );
3102 }
3103 let expected = entry
3104 .resolved_digest
3105 .as_deref()
3106 .or(entry.digest.as_deref())
3107 .ok_or_else(|| {
3108 anyhow!(
3109 "pack.lock component {} missing resolved_digest for remote component",
3110 entry.name
3111 )
3112 })?;
3113 (
3114 ComponentArtifactLocation::Remote,
3115 Some(normalize_digest(expected)),
3116 None,
3117 false,
3118 )
3119 };
3120 let info = ComponentSourceInfo {
3121 digest,
3122 source,
3123 artifact,
3124 expected_wasm_sha256,
3125 skip_digest_verification,
3126 };
3127 if let Some(component_id) = entry.component_id.as_ref() {
3128 let key = component_id.as_str().to_string();
3129 if table.contains_key(&key) {
3130 bail!(
3131 "pack.lock contains duplicate component id `{}`",
3132 component_id.as_str()
3133 );
3134 }
3135 table.insert(key, info.clone());
3136 }
3137 if entry.name
3138 != entry
3139 .component_id
3140 .as_ref()
3141 .map(|id| id.as_str())
3142 .unwrap_or("")
3143 {
3144 table.insert(entry.name.clone(), info);
3145 }
3146 }
3147 Ok(table)
3148}
3149
3150fn component_path_for_spec(root: &Path, spec: &ComponentSpec) -> PathBuf {
3151 if let Some(path) = &spec.legacy_path {
3152 return root.join(path);
3153 }
3154 root.join("components").join(format!("{}.wasm", spec.id))
3155}
3156
3157fn normalize_digest(digest: &str) -> String {
3158 if digest.starts_with("sha256:") || digest.starts_with("blake3:") {
3159 digest.to_string()
3160 } else {
3161 format!("sha256:{digest}")
3162 }
3163}
3164
3165fn compute_digest_for(bytes: &[u8], digest: &str) -> Result<String> {
3166 if digest.starts_with("blake3:") {
3167 let hash = blake3::hash(bytes);
3168 return Ok(format!("blake3:{}", hash.to_hex()));
3169 }
3170 let mut hasher = sha2::Sha256::new();
3171 hasher.update(bytes);
3172 Ok(format!("sha256:{}", to_hex(&hasher.finalize())))
3173}
3174
3175fn compute_sha256_digest_for(bytes: &[u8]) -> String {
3176 let mut hasher = sha2::Sha256::new();
3177 hasher.update(bytes);
3178 format!("sha256:{}", to_hex(&hasher.finalize()))
3179}
3180
3181fn build_artifact_key(cache: &CacheManager, digest: Option<&str>, bytes: &[u8]) -> ArtifactKey {
3182 let wasm_digest = digest
3183 .map(normalize_digest)
3184 .unwrap_or_else(|| compute_sha256_digest_for(bytes));
3185 ArtifactKey::new(cache.engine_profile_id().to_string(), wasm_digest)
3186}
3187
3188async fn compile_component_with_cache(
3189 cache: &CacheManager,
3190 engine: &Engine,
3191 digest: Option<&str>,
3192 bytes: Vec<u8>,
3193) -> Result<Arc<Component>> {
3194 let key = build_artifact_key(cache, digest, &bytes);
3195 cache.get_component(engine, &key, || Ok(bytes)).await
3196}
3197
3198fn verify_component_digest(component_id: &str, expected: &str, bytes: &[u8]) -> Result<()> {
3199 let normalized_expected = normalize_digest(expected);
3200 let actual = compute_digest_for(bytes, &normalized_expected)?;
3201 if normalize_digest(&actual) != normalized_expected {
3202 bail!(
3203 "component {component_id} digest mismatch: expected {normalized_expected}, got {actual}"
3204 );
3205 }
3206 Ok(())
3207}
3208
3209fn verify_wasm_sha256(component_id: &str, expected: &str, bytes: &[u8]) -> Result<()> {
3210 let normalized_expected = normalize_sha256(expected)?;
3211 let actual = compute_sha256_digest_for(bytes);
3212 if actual != normalized_expected {
3213 bail!(
3214 "component {component_id} bundled digest mismatch: expected {normalized_expected}, got {actual}"
3215 );
3216 }
3217 Ok(())
3218}
3219
3220fn to_hex(digest: &[u8]) -> String {
3221 digest.iter().map(|byte| format!("{byte:02x}")).collect()
3222}
3223
3224#[cfg(test)]
3225mod pack_lock_tests {
3226 use super::*;
3227 use tempfile::TempDir;
3228
3229 #[test]
3230 fn pack_lock_tag_ref_requires_bundle() {
3231 let lock = PackLockV1 {
3232 schema_version: 1,
3233 components: vec![PackLockComponent {
3234 name: "templates".to_string(),
3235 source_ref: Some("oci://registry.test/templates:latest".to_string()),
3236 legacy_ref: None,
3237 component_id: None,
3238 bundled: Some(false),
3239 bundled_path: None,
3240 legacy_path: None,
3241 wasm_sha256: None,
3242 legacy_sha256: None,
3243 resolved_digest: None,
3244 digest: None,
3245 }],
3246 };
3247 let err = component_sources_table_from_pack_lock(&lock, false).unwrap_err();
3248 assert!(
3249 err.to_string().contains("tag ref") && err.to_string().contains("rebuild the pack"),
3250 "unexpected error: {err}"
3251 );
3252 }
3253
3254 #[test]
3255 fn bundled_hash_mismatch_errors() {
3256 let rt = tokio::runtime::Runtime::new().expect("runtime");
3257 let temp = TempDir::new().expect("temp dir");
3258 let engine = Engine::default();
3259 let engine_profile =
3260 EngineProfile::from_engine(&engine, CpuPolicy::Native, "default".to_string());
3261 let cache_config = CacheConfig {
3262 root: temp.path().join("cache"),
3263 ..CacheConfig::default()
3264 };
3265 let cache = CacheManager::new(cache_config, engine_profile);
3266 let wasm_path = temp.path().join("component.wasm");
3267 let fixture_wasm = Path::new(env!("CARGO_MANIFEST_DIR"))
3268 .join("../../tests/fixtures/packs/secrets_store_smoke/components/echo_secret.wasm");
3269 let bytes = std::fs::read(&fixture_wasm).expect("read fixture wasm");
3270 std::fs::write(&wasm_path, &bytes).expect("write temp wasm");
3271
3272 let spec = ComponentSpec {
3273 id: "qa.process".to_string(),
3274 version: "0.0.0".to_string(),
3275 legacy_path: None,
3276 };
3277 let mut missing = HashSet::new();
3278 missing.insert(spec.id.clone());
3279
3280 let mut sources = HashMap::new();
3281 sources.insert(
3282 spec.id.clone(),
3283 ComponentSourceInfo {
3284 digest: Some("sha256:deadbeef".to_string()),
3285 source: ComponentSourceRef::Oci("registry.test/qa.process@sha256:deadbeef".into()),
3286 artifact: ComponentArtifactLocation::Inline {
3287 wasm_path: "component.wasm".to_string(),
3288 },
3289 expected_wasm_sha256: Some("sha256:deadbeef".to_string()),
3290 skip_digest_verification: false,
3291 },
3292 );
3293
3294 let mut loaded = HashMap::new();
3295 let result = rt.block_on(load_components_from_sources(
3296 &cache,
3297 &engine,
3298 &sources,
3299 &ComponentResolution::default(),
3300 &[spec],
3301 &mut missing,
3302 &mut loaded,
3303 Some(temp.path()),
3304 None,
3305 ));
3306 let err = result.unwrap_err();
3307 assert!(
3308 err.to_string().contains("bundled digest mismatch"),
3309 "unexpected error: {err}"
3310 );
3311 }
3312}
3313
3314#[cfg(test)]
3315mod pack_resolution_prop_tests {
3316 use super::*;
3317 use greentic_types::{ArtifactLocationV1, ComponentSourceEntryV1, ResolvedComponentV1};
3318 use proptest::prelude::*;
3319 use proptest::test_runner::{Config as ProptestConfig, RngAlgorithm, TestRng, TestRunner};
3320 use std::collections::BTreeSet;
3321 use std::path::Path;
3322 use std::str::FromStr;
3323
3324 #[derive(Clone, Debug)]
3325 enum ResolveRequest {
3326 ById(String),
3327 ByName(String),
3328 }
3329
3330 #[derive(Clone, Debug, PartialEq, Eq)]
3331 struct ResolvedComponent {
3332 key: String,
3333 source: String,
3334 artifact: String,
3335 digest: Option<String>,
3336 expected_wasm_sha256: Option<String>,
3337 skip_digest_verification: bool,
3338 }
3339
3340 #[derive(Clone, Debug, PartialEq, Eq)]
3341 struct ResolveError {
3342 code: String,
3343 message: String,
3344 context_key: String,
3345 }
3346
3347 #[derive(Clone, Debug)]
3348 struct Scenario {
3349 pack_lock: Option<PackLockV1>,
3350 component_sources: Option<ComponentSourcesV1>,
3351 request: ResolveRequest,
3352 expected_sha256: Option<String>,
3353 bytes: Vec<u8>,
3354 }
3355
3356 fn resolve_component_test(
3357 sources: Option<&ComponentSourcesV1>,
3358 lock: Option<&PackLockV1>,
3359 request: &ResolveRequest,
3360 ) -> Result<ResolvedComponent, ResolveError> {
3361 let table = if let Some(lock) = lock {
3362 component_sources_table_from_pack_lock(lock, false).map_err(|err| ResolveError {
3363 code: classify_pack_lock_error(err.to_string().as_str()).to_string(),
3364 message: err.to_string(),
3365 context_key: request_key(request).to_string(),
3366 })?
3367 } else {
3368 let sources = component_sources_table(sources).map_err(|err| ResolveError {
3369 code: "component_sources_error".to_string(),
3370 message: err.to_string(),
3371 context_key: request_key(request).to_string(),
3372 })?;
3373 sources.ok_or_else(|| ResolveError {
3374 code: "missing_component_sources".to_string(),
3375 message: "component sources not provided".to_string(),
3376 context_key: request_key(request).to_string(),
3377 })?
3378 };
3379
3380 let key = request_key(request);
3381 let source = table.get(key).ok_or_else(|| ResolveError {
3382 code: "component_not_found".to_string(),
3383 message: format!("component {key} not found"),
3384 context_key: key.to_string(),
3385 })?;
3386
3387 Ok(ResolvedComponent {
3388 key: key.to_string(),
3389 source: source.source.to_string(),
3390 artifact: match source.artifact {
3391 ComponentArtifactLocation::Inline { .. } => "inline".to_string(),
3392 ComponentArtifactLocation::Remote => "remote".to_string(),
3393 },
3394 digest: source.digest.clone(),
3395 expected_wasm_sha256: source.expected_wasm_sha256.clone(),
3396 skip_digest_verification: source.skip_digest_verification,
3397 })
3398 }
3399
3400 fn request_key(request: &ResolveRequest) -> &str {
3401 match request {
3402 ResolveRequest::ById(value) => value.as_str(),
3403 ResolveRequest::ByName(value) => value.as_str(),
3404 }
3405 }
3406
3407 fn classify_pack_lock_error(message: &str) -> &'static str {
3408 if message.contains("duplicate component name") {
3409 "duplicate_name"
3410 } else if message.contains("duplicate component id") {
3411 "duplicate_id"
3412 } else if message.contains("conflicting refs") {
3413 "conflicting_ref"
3414 } else if message.contains("conflicting bundled paths") {
3415 "conflicting_bundled_path"
3416 } else if message.contains("conflicting wasm_sha256") {
3417 "conflicting_wasm_sha256"
3418 } else if message.contains("missing source_ref") {
3419 "missing_source_ref"
3420 } else if message.contains("marked bundled but bundled_path is missing") {
3421 "missing_bundled_path"
3422 } else if message.contains("missing wasm_sha256") {
3423 "missing_wasm_sha256"
3424 } else if message.contains("tag ref") && message.contains("not bundled") {
3425 "tag_ref_requires_bundle"
3426 } else if message.contains("missing resolved_digest") {
3427 "missing_resolved_digest"
3428 } else if message.contains("invalid component ref") {
3429 "invalid_component_ref"
3430 } else if message.contains("sha256 digest") {
3431 "invalid_sha256"
3432 } else {
3433 "unknown_error"
3434 }
3435 }
3436
3437 fn known_error_codes() -> BTreeSet<&'static str> {
3438 [
3439 "component_sources_error",
3440 "missing_component_sources",
3441 "component_not_found",
3442 "duplicate_name",
3443 "duplicate_id",
3444 "conflicting_ref",
3445 "conflicting_bundled_path",
3446 "conflicting_wasm_sha256",
3447 "missing_source_ref",
3448 "missing_bundled_path",
3449 "missing_wasm_sha256",
3450 "tag_ref_requires_bundle",
3451 "missing_resolved_digest",
3452 "invalid_component_ref",
3453 "invalid_sha256",
3454 "unknown_error",
3455 ]
3456 .into_iter()
3457 .collect()
3458 }
3459
3460 fn proptest_config() -> ProptestConfig {
3461 let cases = std::env::var("PROPTEST_CASES")
3462 .ok()
3463 .and_then(|value| value.parse::<u32>().ok())
3464 .unwrap_or(128);
3465 ProptestConfig {
3466 cases,
3467 failure_persistence: None,
3468 ..ProptestConfig::default()
3469 }
3470 }
3471
3472 fn proptest_seed() -> Option<[u8; 32]> {
3473 let seed = std::env::var("PROPTEST_SEED")
3474 .ok()
3475 .and_then(|value| value.parse::<u64>().ok())?;
3476 let mut bytes = [0u8; 32];
3477 bytes[..8].copy_from_slice(&seed.to_le_bytes());
3478 Some(bytes)
3479 }
3480
3481 fn run_cases(strategy: impl Strategy<Value = Scenario>, cases: u32, seed: Option<[u8; 32]>) {
3482 let config = ProptestConfig {
3483 cases,
3484 failure_persistence: None,
3485 ..ProptestConfig::default()
3486 };
3487 let mut runner = match seed {
3488 Some(bytes) => {
3489 TestRunner::new_with_rng(config, TestRng::from_seed(RngAlgorithm::ChaCha, &bytes))
3490 }
3491 None => TestRunner::new(config),
3492 };
3493 runner
3494 .run(&strategy, |scenario| {
3495 run_scenario(&scenario);
3496 Ok(())
3497 })
3498 .unwrap();
3499 }
3500
3501 fn run_scenario(scenario: &Scenario) {
3502 let known_codes = known_error_codes();
3503 let first = resolve_component_test(
3504 scenario.component_sources.as_ref(),
3505 scenario.pack_lock.as_ref(),
3506 &scenario.request,
3507 );
3508 let second = resolve_component_test(
3509 scenario.component_sources.as_ref(),
3510 scenario.pack_lock.as_ref(),
3511 &scenario.request,
3512 );
3513 assert_eq!(normalize_result(&first), normalize_result(&second));
3514
3515 if let Some(lock) = scenario.pack_lock.as_ref() {
3516 let lock_only = resolve_component_test(None, Some(lock), &scenario.request);
3517 assert_eq!(normalize_result(&first), normalize_result(&lock_only));
3518 }
3519
3520 if let Err(err) = first.as_ref() {
3521 assert!(
3522 known_codes.contains(err.code.as_str()),
3523 "unexpected error code {}: {}",
3524 err.code,
3525 err.message
3526 );
3527 }
3528
3529 if let Some(expected) = scenario.expected_sha256.as_deref() {
3530 let expected_ok =
3531 verify_wasm_sha256("test.component", expected, &scenario.bytes).is_ok();
3532 let actual = compute_sha256_digest_for(&scenario.bytes);
3533 if actual == normalize_sha256(expected).unwrap_or_default() {
3534 assert!(expected_ok, "expected sha256 match to succeed");
3535 } else {
3536 assert!(!expected_ok, "expected sha256 mismatch to fail");
3537 }
3538 }
3539 }
3540
3541 fn normalize_result(
3542 result: &Result<ResolvedComponent, ResolveError>,
3543 ) -> Result<ResolvedComponent, ResolveError> {
3544 match result {
3545 Ok(value) => Ok(value.clone()),
3546 Err(err) => Err(err.clone()),
3547 }
3548 }
3549
3550 fn scenario_strategy() -> impl Strategy<Value = Scenario> {
3551 let name = any::<u8>().prop_map(|n| format!("component{n}.core"));
3552 let alt_name = any::<u8>().prop_map(|n| format!("component_alt{n}.core"));
3553 let tag_ref = any::<bool>();
3554 let bundled = any::<bool>();
3555 let include_sha = any::<bool>();
3556 let include_component_id = any::<bool>();
3557 let request_by_id = any::<bool>();
3558 let use_lock = any::<bool>();
3559 let use_sources = any::<bool>();
3560 let bytes = prop::collection::vec(any::<u8>(), 1..64);
3561
3562 (
3563 name,
3564 alt_name,
3565 tag_ref,
3566 bundled,
3567 include_sha,
3568 include_component_id,
3569 request_by_id,
3570 use_lock,
3571 use_sources,
3572 bytes,
3573 )
3574 .prop_map(
3575 |(
3576 name,
3577 alt_name,
3578 tag_ref,
3579 bundled,
3580 include_sha,
3581 include_component_id,
3582 request_by_id,
3583 use_lock,
3584 use_sources,
3585 bytes,
3586 )| {
3587 let component_id_str = if include_component_id {
3588 alt_name.clone()
3589 } else {
3590 name.clone()
3591 };
3592 let component_id = ComponentId::from_str(&component_id_str).ok();
3593 let source_ref = if tag_ref {
3594 format!("oci://registry.test/{name}:v1")
3595 } else {
3596 format!(
3597 "oci://registry.test/{name}@sha256:{}",
3598 hex::encode([0x11u8; 32])
3599 )
3600 };
3601 let expected_sha256 = if bundled && include_sha {
3602 Some(compute_sha256_digest_for(&bytes))
3603 } else {
3604 None
3605 };
3606
3607 let lock_component = PackLockComponent {
3608 name: name.clone(),
3609 source_ref: Some(source_ref),
3610 legacy_ref: None,
3611 component_id,
3612 bundled: Some(bundled),
3613 bundled_path: if bundled {
3614 Some(format!("components/{name}.wasm"))
3615 } else {
3616 None
3617 },
3618 legacy_path: None,
3619 wasm_sha256: expected_sha256.clone(),
3620 legacy_sha256: None,
3621 resolved_digest: if bundled {
3622 None
3623 } else {
3624 Some("sha256:deadbeef".to_string())
3625 },
3626 digest: None,
3627 };
3628
3629 let pack_lock = if use_lock {
3630 Some(PackLockV1 {
3631 schema_version: 1,
3632 components: vec![lock_component],
3633 })
3634 } else {
3635 None
3636 };
3637
3638 let component_sources = if use_sources {
3639 Some(ComponentSourcesV1::new(vec![ComponentSourceEntryV1 {
3640 name: name.clone(),
3641 component_id: ComponentId::from_str(&name).ok(),
3642 source: ComponentSourceRef::from_str(
3643 "oci://registry.test/component@sha256:deadbeef",
3644 )
3645 .expect("component ref"),
3646 resolved: ResolvedComponentV1 {
3647 digest: "sha256:deadbeef".to_string(),
3648 signature: None,
3649 signed_by: None,
3650 },
3651 artifact: if bundled {
3652 ArtifactLocationV1::Inline {
3653 wasm_path: format!("components/{name}.wasm"),
3654 manifest_path: None,
3655 }
3656 } else {
3657 ArtifactLocationV1::Remote
3658 },
3659 licensing_hint: None,
3660 metering_hint: None,
3661 }]))
3662 } else {
3663 None
3664 };
3665
3666 let request = if request_by_id {
3667 ResolveRequest::ById(component_id_str.clone())
3668 } else {
3669 ResolveRequest::ByName(name.clone())
3670 };
3671
3672 Scenario {
3673 pack_lock,
3674 component_sources,
3675 request,
3676 expected_sha256,
3677 bytes,
3678 }
3679 },
3680 )
3681 }
3682
3683 #[test]
3684 fn pack_resolution_proptest() {
3685 let seed = proptest_seed();
3686 run_cases(scenario_strategy(), proptest_config().cases, seed);
3687 }
3688
3689 #[test]
3690 fn pack_resolution_regression_seeds() {
3691 let seeds_path =
3692 Path::new(env!("CARGO_MANIFEST_DIR")).join("../../tests/fixtures/proptest-seeds.txt");
3693 let raw = std::fs::read_to_string(&seeds_path).expect("read proptest seeds");
3694 for line in raw.lines() {
3695 let line = line.trim();
3696 if line.is_empty() || line.starts_with('#') {
3697 continue;
3698 }
3699 let seed = line.parse::<u64>().expect("seed must be an integer");
3700 let mut bytes = [0u8; 32];
3701 bytes[..8].copy_from_slice(&seed.to_le_bytes());
3702 run_cases(scenario_strategy(), 1, Some(bytes));
3703 }
3704 }
3705}
3706
3707fn locate_pack_assets(
3708 materialized_root: Option<&Path>,
3709 archive_hint: Option<&Path>,
3710) -> Result<(Option<PathBuf>, Option<TempDir>)> {
3711 if let Some(root) = materialized_root {
3712 let assets = root.join("assets");
3713 if assets.is_dir() {
3714 return Ok((Some(assets), None));
3715 }
3716 }
3717 if let Some(path) = archive_hint
3718 && let Some((tempdir, assets)) = extract_assets_from_archive(path)?
3719 {
3720 return Ok((Some(assets), Some(tempdir)));
3721 }
3722 Ok((None, None))
3723}
3724
3725fn extract_assets_from_archive(path: &Path) -> Result<Option<(TempDir, PathBuf)>> {
3726 let file =
3727 File::open(path).with_context(|| format!("failed to open pack {}", path.display()))?;
3728 let mut archive =
3729 ZipArchive::new(file).with_context(|| format!("failed to read pack {}", path.display()))?;
3730 let temp = TempDir::new().context("failed to create temporary assets directory")?;
3731 let mut found = false;
3732 for idx in 0..archive.len() {
3733 let mut entry = archive.by_index(idx)?;
3734 let name = entry.name();
3735 if !name.starts_with("assets/") {
3736 continue;
3737 }
3738 let dest = temp.path().join(name);
3739 if name.ends_with('/') {
3740 std::fs::create_dir_all(&dest)?;
3741 found = true;
3742 continue;
3743 }
3744 if let Some(parent) = dest.parent() {
3745 std::fs::create_dir_all(parent)?;
3746 }
3747 let mut outfile = std::fs::File::create(&dest)?;
3748 std::io::copy(&mut entry, &mut outfile)?;
3749 found = true;
3750 }
3751 if found {
3752 let assets_path = temp.path().join("assets");
3753 Ok(Some((temp, assets_path)))
3754 } else {
3755 Ok(None)
3756 }
3757}
3758
3759fn dist_options_from(component_resolution: &ComponentResolution) -> DistOptions {
3760 let mut opts = DistOptions {
3761 allow_tags: true,
3762 ..DistOptions::default()
3763 };
3764 if let Some(cache_dir) = component_resolution.dist_cache_dir.clone() {
3765 opts.cache_dir = cache_dir;
3766 }
3767 if component_resolution.dist_offline {
3768 opts.offline = true;
3769 }
3770 opts
3771}
3772
3773#[allow(clippy::too_many_arguments)]
3774async fn load_components_from_sources(
3775 cache: &CacheManager,
3776 engine: &Engine,
3777 component_sources: &HashMap<String, ComponentSourceInfo>,
3778 component_resolution: &ComponentResolution,
3779 specs: &[ComponentSpec],
3780 missing: &mut HashSet<String>,
3781 into: &mut HashMap<String, PackComponent>,
3782 materialized_root: Option<&Path>,
3783 archive_hint: Option<&Path>,
3784) -> Result<()> {
3785 let mut archive = if let Some(path) = archive_hint {
3786 Some(
3787 ZipArchive::new(File::open(path)?)
3788 .with_context(|| format!("{} is not a valid gtpack", path.display()))?,
3789 )
3790 } else {
3791 None
3792 };
3793 let mut dist_client: Option<DistClient> = None;
3794
3795 for spec in specs {
3796 if !missing.contains(&spec.id) {
3797 continue;
3798 }
3799 let Some(source) = component_sources.get(&spec.id) else {
3800 continue;
3801 };
3802
3803 let bytes = match &source.artifact {
3804 ComponentArtifactLocation::Inline { wasm_path } => {
3805 if let Some(root) = materialized_root {
3806 let path = root.join(wasm_path);
3807 if path.exists() {
3808 std::fs::read(&path).with_context(|| {
3809 format!(
3810 "failed to read inline component {} from {}",
3811 spec.id,
3812 path.display()
3813 )
3814 })?
3815 } else if archive.is_none() {
3816 bail!("inline component {} missing at {}", spec.id, path.display());
3817 } else {
3818 read_entry(
3819 archive.as_mut().expect("archive present when needed"),
3820 wasm_path,
3821 )
3822 .with_context(|| {
3823 format!(
3824 "inline component {} missing at {} in pack archive",
3825 spec.id, wasm_path
3826 )
3827 })?
3828 }
3829 } else if let Some(archive) = archive.as_mut() {
3830 read_entry(archive, wasm_path).with_context(|| {
3831 format!(
3832 "inline component {} missing at {} in pack archive",
3833 spec.id, wasm_path
3834 )
3835 })?
3836 } else {
3837 bail!(
3838 "inline component {} missing and no pack source available",
3839 spec.id
3840 );
3841 }
3842 }
3843 ComponentArtifactLocation::Remote => {
3844 if source.source.is_tag() {
3845 bail!(
3846 "component {} uses tag ref {} but is not bundled; rebuild the pack",
3847 spec.id,
3848 source.source
3849 );
3850 }
3851 let client = dist_client.get_or_insert_with(|| {
3852 DistClient::new(dist_options_from(component_resolution))
3853 });
3854 let reference = source.source.to_string();
3855 fault::maybe_fail_asset(&reference)
3856 .await
3857 .with_context(|| format!("fault injection blocked asset {reference}"))?;
3858 let digest = source.digest.as_deref().ok_or_else(|| {
3859 anyhow!(
3860 "component {} missing expected digest for remote component",
3861 spec.id
3862 )
3863 })?;
3864 let cache_path = if let Ok(cache_path) = client.fetch_digest(digest).await {
3865 cache_path
3866 } else if component_resolution.dist_offline {
3867 client
3868 .fetch_digest(digest)
3869 .await
3870 .map_err(|err| dist_error_for_component(err, &spec.id, &reference))?
3871 } else {
3872 let source = client
3873 .parse_source(&reference)
3874 .map_err(|err| dist_error_for_component(err, &spec.id, &reference))?;
3875 let descriptor = client
3876 .resolve(source, ResolvePolicy)
3877 .await
3878 .map_err(|err| dist_error_for_component(err, &spec.id, &reference))?;
3879 let resolved = client
3880 .fetch(&descriptor, CachePolicy)
3881 .await
3882 .map_err(|err| dist_error_for_component(err, &spec.id, &reference))?;
3883 let expected = normalize_digest(digest);
3884 let actual = normalize_digest(&resolved.digest);
3885 if expected != actual {
3886 bail!(
3887 "component {} digest mismatch after fetch: expected {}, got {}",
3888 spec.id,
3889 expected,
3890 actual
3891 );
3892 }
3893 resolved.cache_path.ok_or_else(|| {
3894 anyhow!(
3895 "component {} resolved from {} but cache path is missing",
3896 spec.id,
3897 reference
3898 )
3899 })?
3900 };
3901 std::fs::read(&cache_path).with_context(|| {
3902 format!(
3903 "failed to read cached component {} from {}",
3904 spec.id,
3905 cache_path.display()
3906 )
3907 })?
3908 }
3909 };
3910
3911 if let Some(expected) = source.expected_wasm_sha256.as_deref() {
3912 verify_wasm_sha256(&spec.id, expected, &bytes)?;
3913 } else if source.skip_digest_verification {
3914 let actual = compute_sha256_digest_for(&bytes);
3915 warn!(
3916 component_id = %spec.id,
3917 digest = %actual,
3918 "bundled component missing wasm_sha256; allowing due to flag"
3919 );
3920 } else {
3921 let expected = source.digest.as_deref().ok_or_else(|| {
3922 anyhow!(
3923 "component {} missing expected digest for verification",
3924 spec.id
3925 )
3926 })?;
3927 verify_component_digest(&spec.id, expected, &bytes)?;
3928 }
3929 let component =
3930 compile_component_with_cache(cache, engine, source.digest.as_deref(), bytes)
3931 .await
3932 .with_context(|| format!("failed to compile component {}", spec.id))?;
3933 into.insert(
3934 spec.id.clone(),
3935 PackComponent {
3936 name: spec.id.clone(),
3937 version: spec.version.clone(),
3938 component,
3939 },
3940 );
3941 missing.remove(&spec.id);
3942 }
3943
3944 Ok(())
3945}
3946
3947fn dist_error_for_component(err: DistError, component_id: &str, reference: &str) -> anyhow::Error {
3948 match err {
3949 DistError::NotFound { reference: missing } => anyhow!(
3950 "remote component {} is not cached for {}. Run `greentic-dist pull --lock <pack.lock>` or `greentic-dist pull {}`",
3951 component_id,
3952 missing,
3953 reference
3954 ),
3955 DistError::Offline { reference: blocked } => anyhow!(
3956 "offline mode blocked fetching component {} from {}; run `greentic-dist pull --lock <pack.lock>` or `greentic-dist pull {}`",
3957 component_id,
3958 blocked,
3959 reference
3960 ),
3961 DistError::Unauthorized { target } => anyhow!(
3962 "component {} requires authenticated source {}; run `greentic-dist pull --lock <pack.lock>` or `greentic-dist pull {}`",
3963 component_id,
3964 target,
3965 reference
3966 ),
3967 other => anyhow!(
3968 "failed to resolve component {} from {}: {}",
3969 component_id,
3970 reference,
3971 other
3972 ),
3973 }
3974}
3975
3976async fn load_components_from_overrides(
3977 cache: &CacheManager,
3978 engine: &Engine,
3979 overrides: &HashMap<String, PathBuf>,
3980 specs: &[ComponentSpec],
3981 missing: &mut HashSet<String>,
3982 into: &mut HashMap<String, PackComponent>,
3983) -> Result<()> {
3984 for spec in specs {
3985 if !missing.contains(&spec.id) {
3986 continue;
3987 }
3988 let Some(path) = overrides.get(&spec.id) else {
3989 continue;
3990 };
3991 let bytes = std::fs::read(path)
3992 .with_context(|| format!("failed to read override component {}", path.display()))?;
3993 let component = compile_component_with_cache(cache, engine, None, bytes)
3994 .await
3995 .with_context(|| {
3996 format!(
3997 "failed to compile component {} from override {}",
3998 spec.id,
3999 path.display()
4000 )
4001 })?;
4002 into.insert(
4003 spec.id.clone(),
4004 PackComponent {
4005 name: spec.id.clone(),
4006 version: spec.version.clone(),
4007 component,
4008 },
4009 );
4010 missing.remove(&spec.id);
4011 }
4012 Ok(())
4013}
4014
4015async fn load_components_from_dir(
4016 cache: &CacheManager,
4017 engine: &Engine,
4018 root: &Path,
4019 specs: &[ComponentSpec],
4020 missing: &mut HashSet<String>,
4021 into: &mut HashMap<String, PackComponent>,
4022) -> Result<()> {
4023 for spec in specs {
4024 if !missing.contains(&spec.id) {
4025 continue;
4026 }
4027 let path = component_path_for_spec(root, spec);
4028 if !path.exists() {
4029 tracing::debug!(component = %spec.id, path = %path.display(), "materialized component missing; will try other sources");
4030 continue;
4031 }
4032 let bytes = std::fs::read(&path)
4033 .with_context(|| format!("failed to read component {}", path.display()))?;
4034 let component = compile_component_with_cache(cache, engine, None, bytes)
4035 .await
4036 .with_context(|| {
4037 format!(
4038 "failed to compile component {} from {}",
4039 spec.id,
4040 path.display()
4041 )
4042 })?;
4043 into.insert(
4044 spec.id.clone(),
4045 PackComponent {
4046 name: spec.id.clone(),
4047 version: spec.version.clone(),
4048 component,
4049 },
4050 );
4051 missing.remove(&spec.id);
4052 }
4053 Ok(())
4054}
4055
4056async fn load_components_from_archive(
4057 cache: &CacheManager,
4058 engine: &Engine,
4059 path: &Path,
4060 specs: &[ComponentSpec],
4061 missing: &mut HashSet<String>,
4062 into: &mut HashMap<String, PackComponent>,
4063) -> Result<()> {
4064 let mut archive = ZipArchive::new(File::open(path)?)
4065 .with_context(|| format!("{} is not a valid gtpack", path.display()))?;
4066 for spec in specs {
4067 if !missing.contains(&spec.id) {
4068 continue;
4069 }
4070 let file_name = spec
4071 .legacy_path
4072 .clone()
4073 .unwrap_or_else(|| format!("components/{}.wasm", spec.id));
4074 let bytes = match read_entry(&mut archive, &file_name) {
4075 Ok(bytes) => bytes,
4076 Err(err) => {
4077 warn!(component = %spec.id, pack = %path.display(), error = %err, "component entry missing in pack archive");
4078 continue;
4079 }
4080 };
4081 let component = compile_component_with_cache(cache, engine, None, bytes)
4082 .await
4083 .with_context(|| format!("failed to compile component {}", spec.id))?;
4084 into.insert(
4085 spec.id.clone(),
4086 PackComponent {
4087 name: spec.id.clone(),
4088 version: spec.version.clone(),
4089 component,
4090 },
4091 );
4092 missing.remove(&spec.id);
4093 }
4094 Ok(())
4095}
4096
4097#[cfg(test)]
4098mod tests {
4099 use super::*;
4100 use greentic_flow::model::{FlowDoc, NodeDoc};
4101 use indexmap::IndexMap;
4102 use serde_json::json;
4103
4104 #[test]
4105 fn normalizes_raw_component_to_component_exec() {
4106 let mut nodes = IndexMap::new();
4107 let mut raw = IndexMap::new();
4108 raw.insert(
4109 "templating.handlebars".into(),
4110 json!({ "template": "Hi {{name}}" }),
4111 );
4112 nodes.insert(
4113 "start".into(),
4114 NodeDoc {
4115 raw,
4116 routing: json!([{"out": true}]),
4117 ..Default::default()
4118 },
4119 );
4120 let doc = FlowDoc {
4121 id: "welcome".into(),
4122 title: None,
4123 description: None,
4124 flow_type: "messaging".into(),
4125 start: Some("start".into()),
4126 parameters: json!({}),
4127 tags: Vec::new(),
4128 schema_version: None,
4129 entrypoints: IndexMap::new(),
4130 meta: None,
4131 nodes,
4132 };
4133
4134 let normalized = normalize_flow_doc(doc);
4135 let node = normalized.nodes.get("start").expect("node exists");
4136 assert_eq!(node.operation.as_deref(), Some("component.exec"));
4137 assert!(node.raw.is_empty());
4138 let payload = node.payload.as_object().expect("payload object");
4139 assert_eq!(
4140 payload.get("component"),
4141 Some(&Value::String("templating.handlebars".into()))
4142 );
4143 assert_eq!(
4144 payload.get("operation"),
4145 Some(&Value::String("render".into()))
4146 );
4147 let input = payload.get("input").unwrap();
4148 assert_eq!(input, &json!({ "template": "Hi {{name}}" }));
4149 }
4150
4151 #[test]
4152 fn normalizes_canonical_operation_node_to_component_exec_with_config() {
4153 let mut nodes = IndexMap::new();
4154 let mut raw = IndexMap::new();
4155 raw.insert(
4156 "handle_message".into(),
4157 json!({
4158 "component": "oci://ghcr.io/greenticai/component/component-llm-openai:stable",
4159 "config": {
4160 "provider": "ollama",
4161 "base_url": "http://127.0.0.1:11434/v1",
4162 "default_model": "llama3.2"
4163 },
4164 "input": {
4165 "messages": [{
4166 "role": "user",
4167 "content": "Say hello from Ollama."
4168 }]
4169 }
4170 }),
4171 );
4172 nodes.insert(
4173 "llm".into(),
4174 NodeDoc {
4175 raw,
4176 routing: json!([{"out": true}]),
4177 ..Default::default()
4178 },
4179 );
4180 let doc = FlowDoc {
4181 id: "ollama-repro".into(),
4182 title: None,
4183 description: None,
4184 flow_type: "messaging".into(),
4185 start: Some("llm".into()),
4186 parameters: json!({}),
4187 tags: Vec::new(),
4188 schema_version: None,
4189 entrypoints: IndexMap::new(),
4190 meta: None,
4191 nodes,
4192 };
4193
4194 let normalized = normalize_flow_doc(doc);
4195 let node = normalized.nodes.get("llm").expect("node exists");
4196 assert_eq!(node.operation.as_deref(), Some("component.exec"));
4197 assert!(node.raw.is_empty());
4198 let payload = node.payload.as_object().expect("payload object");
4199 assert_eq!(
4200 payload.get("component"),
4201 Some(&Value::String(
4202 "oci://ghcr.io/greenticai/component/component-llm-openai:stable".into()
4203 ))
4204 );
4205 assert_eq!(
4206 payload.get("operation"),
4207 Some(&Value::String("handle_message".into()))
4208 );
4209 assert_eq!(
4210 payload.get("config"),
4211 Some(&json!({
4212 "provider": "ollama",
4213 "base_url": "http://127.0.0.1:11434/v1",
4214 "default_model": "llama3.2"
4215 }))
4216 );
4217 assert_eq!(
4218 payload.get("input"),
4219 Some(&json!({
4220 "messages": [{
4221 "role": "user",
4222 "content": "Say hello from Ollama."
4223 }]
4224 }))
4225 );
4226 }
4227}
4228
4229#[derive(Clone, Debug, Default, Serialize, Deserialize)]
4230pub struct PackMetadata {
4231 pub pack_id: String,
4232 pub version: String,
4233 #[serde(default)]
4234 pub entry_flows: Vec<String>,
4235 #[serde(default)]
4236 pub secret_requirements: Vec<greentic_types::SecretRequirement>,
4237}
4238
4239impl PackMetadata {
4240 fn from_wasm(bytes: &[u8]) -> Option<Self> {
4241 let parser = Parser::new(0);
4242 for payload in parser.parse_all(bytes) {
4243 let payload = payload.ok()?;
4244 match payload {
4245 Payload::CustomSection(section) => {
4246 if section.name() == "greentic.manifest"
4247 && let Ok(meta) = Self::from_bytes(section.data())
4248 {
4249 return Some(meta);
4250 }
4251 }
4252 Payload::DataSection(reader) => {
4253 for segment in reader.into_iter().flatten() {
4254 if let Ok(meta) = Self::from_bytes(segment.data) {
4255 return Some(meta);
4256 }
4257 }
4258 }
4259 _ => {}
4260 }
4261 }
4262 None
4263 }
4264
4265 fn from_bytes(bytes: &[u8]) -> Result<Self, serde_cbor::Error> {
4266 #[derive(Deserialize)]
4267 struct RawManifest {
4268 pack_id: String,
4269 version: String,
4270 #[serde(default)]
4271 entry_flows: Vec<String>,
4272 #[serde(default)]
4273 flows: Vec<RawFlow>,
4274 #[serde(default)]
4275 secret_requirements: Vec<greentic_types::SecretRequirement>,
4276 }
4277
4278 #[derive(Deserialize)]
4279 struct RawFlow {
4280 id: String,
4281 }
4282
4283 let manifest: RawManifest = serde_cbor::from_slice(bytes)?;
4284 let mut entry_flows = if manifest.entry_flows.is_empty() {
4285 manifest.flows.iter().map(|f| f.id.clone()).collect()
4286 } else {
4287 manifest.entry_flows.clone()
4288 };
4289 entry_flows.retain(|id| !id.is_empty());
4290 Ok(Self {
4291 pack_id: manifest.pack_id,
4292 version: manifest.version,
4293 entry_flows,
4294 secret_requirements: manifest.secret_requirements,
4295 })
4296 }
4297
4298 pub fn fallback(path: &Path) -> Self {
4299 let pack_id = path
4300 .file_stem()
4301 .map(|s| s.to_string_lossy().into_owned())
4302 .unwrap_or_else(|| "unknown-pack".to_string());
4303 Self {
4304 pack_id,
4305 version: "0.0.0".to_string(),
4306 entry_flows: Vec::new(),
4307 secret_requirements: Vec::new(),
4308 }
4309 }
4310
4311 pub fn from_manifest(manifest: &greentic_types::PackManifest) -> Self {
4312 let entry_flows = manifest
4313 .flows
4314 .iter()
4315 .map(|flow| flow.id.as_str().to_string())
4316 .collect::<Vec<_>>();
4317 Self {
4318 pack_id: manifest.pack_id.as_str().to_string(),
4319 version: manifest.version.to_string(),
4320 entry_flows,
4321 secret_requirements: manifest.secret_requirements.clone(),
4322 }
4323 }
4324}