Skip to main content

greentic_runner_host/
pack.rs

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    /// Component artifact path (wasm file).
116    path: PathBuf,
117    /// Optional archive (.gtpack) used to load flows/manifests.
118    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    /// Root of a materialized pack directory containing `manifest.cbor` and `components/`.
188    pub materialized_root: Option<PathBuf>,
189    /// Explicit overrides mapping component id -> wasm path.
190    pub overrides: HashMap<String, PathBuf>,
191    /// If true, do not fetch remote components; require cached artifacts.
192    pub dist_offline: bool,
193    /// Optional cache directory for resolved remote components.
194    pub dist_cache_dir: Option<PathBuf>,
195    /// Allow bundled components without wasm_sha256 (dev-only escape hatch).
196    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    /// Fallback for v0.6 components that export `component-runtime::run(input, state)`
382    /// instead of the legacy `node::invoke(ctx, op, input)`.
383    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            // Encode input as CBOR — the component's run() expects CBOR bytes.
399            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            // Decode output CBOR to JSON.
410            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    /// Build a `TenantCtx` for secrets lookups that includes the team from the
458    /// execution context. `config.tenant_ctx()` only populates env + tenant;
459    /// without this, secrets scoped to a specific team are unreachable.
460    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        // no-op cooperative yield
1236    }
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    // Add wasi-tls types and turn on the feature in linker
1292    let mut opts = LinkOptions::default();
1293    opts.tls(true);
1294    add_wasi_tls_to_linker(linker, &opts)?;
1295
1296    // Add wasi-http types and turn on the feature in linker
1297    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            // No manifest entry — allow state-store; Wasmtime rejects if not imported.
1484            return true;
1485        };
1486        // If manifest declares host.state capabilities, honour them.
1487        // If host.state is None (not declared in manifest), default to true so
1488        // components whose CBOR manifest omits the field still get state-store
1489        // linked — Wasmtime will reject at instantiation if not actually imported.
1490        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    /// Returns a clonable handle to the pack's state store, when one is
1504    /// configured. Used by the flow engine's built-in `state.get`/`state.set`
1505    /// operators which call into the same store that WASM components read
1506    /// through their `state.read`/`state.write` host imports.
1507    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    /// Read an asset file from the pack's assets directory.
2150    ///
2151    /// Accepts paths like `assets/cards/card-a.json` or `cards/card-a.json`
2152    /// (the `assets/` prefix is stripped automatically).
2153    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        // Try assets tempdir first (extracted from archive).
2158        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        // Try materialized directory.
2166        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                // scope must match current context if provided
2307                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        // component.exec is meta; ensure the payload carries the actual target component.
2722        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}