From 48decb0e47618061b25de6740bccf242483b8013 Mon Sep 17 00:00:00 2001 From: Natanael Mojica Date: Mon, 3 Aug 2026 08:45:39 -0600 Subject: [PATCH 1/2] fix(runtime): compile hydrated invoke target fields --- .../plasm-runtime/src/compile_stub_value.rs | 120 +++++++++++ .../src/execution/compile_preflight.rs | 4 +- crates/plasm-runtime/src/lib.rs | 1 + crates/plasm-runtime/src/preflight.rs | 183 ++++++++++++++-- crates/plasm-runtime/src/view_stub_rows.rs | 18 ++ .../tests/hydrate_invoke_target.rs | 204 ++++++++++++++++++ .../schemas/hydrate_invoke_target/domain.yaml | 133 ++++++++++++ .../hydrate_invoke_target/mappings.yaml | 68 ++++++ 8 files changed, 707 insertions(+), 24 deletions(-) create mode 100644 crates/plasm-runtime/src/compile_stub_value.rs create mode 100644 crates/plasm-runtime/tests/hydrate_invoke_target.rs create mode 100644 fixtures/schemas/hydrate_invoke_target/domain.yaml create mode 100644 fixtures/schemas/hydrate_invoke_target/mappings.yaml diff --git a/crates/plasm-runtime/src/compile_stub_value.rs b/crates/plasm-runtime/src/compile_stub_value.rs new file mode 100644 index 00000000..5b519636 --- /dev/null +++ b/crates/plasm-runtime/src/compile_stub_value.rs @@ -0,0 +1,120 @@ +//! Schema-derived placeholder values used by compile-only runtime paths. +//! +//! These values must satisfy the declared domain shape closely enough for CML to parse, format, +//! or descend into them. They are never transport values. + +use indexmap::IndexMap; +use plasm_core::{FieldType, NamedValueSchema, TemporalWireFormat, Value, ValueWireFormat, CGS}; + +pub(crate) const ZERO_UUID: &str = "00000000-0000-0000-0000-000000000000"; +const STUB_RFC3339: &str = "1970-01-01T00:00:00Z"; +const STUB_ISO_DATE: &str = "1970-01-01"; +pub(crate) const STUB_STRING: &str = "preflight-stub"; + +/// Produce a valid, deterministic placeholder for a named value during static compilation. +pub(crate) fn preflight_compile_stub_value(named_value: &NamedValueSchema, cgs: &CGS) -> Value { + stub_value(named_value, cgs, 0) +} + +fn stub_value(named_value: &NamedValueSchema, cgs: &CGS, depth: usize) -> Value { + match &named_value.field_type { + FieldType::Boolean => Value::Bool(false), + FieldType::Number => Value::Float(0.0), + FieldType::Integer => Value::Integer(0), + FieldType::MultiSelect | FieldType::Array => Value::Array(Vec::new()), + FieldType::Json => Value::Object(IndexMap::new()), + FieldType::Uuid => Value::String(ZERO_UUID.to_string()), + FieldType::Date => match named_value.value_format { + Some(ValueWireFormat::Temporal( + TemporalWireFormat::UnixMs | TemporalWireFormat::UnixSec, + )) => Value::Integer(0), + Some(ValueWireFormat::Temporal(TemporalWireFormat::Iso8601Date)) => { + Value::String(STUB_ISO_DATE.to_string()) + } + Some(ValueWireFormat::Temporal(TemporalWireFormat::Rfc3339)) | None => { + Value::String(STUB_RFC3339.to_string()) + } + }, + FieldType::EntityRef { target } => entity_ref_stub_value(cgs, target, depth), + FieldType::String | FieldType::Select | FieldType::Blob => { + Value::String(STUB_STRING.to_string()) + } + } +} + +fn entity_ref_stub_value(cgs: &CGS, target_name: &str, depth: usize) -> Value { + let Some(target) = cgs.get_entity(target_name) else { + return Value::String(STUB_STRING.to_string()); + }; + + let key_stub = |key_name: &str| { + if depth < 2 { + target + .fields + .get(key_name) + .and_then(|field| field.named_value(cgs).ok()) + .map_or_else( + || Value::String(STUB_STRING.to_string()), + |key_type| stub_value(key_type, cgs, depth + 1), + ) + } else { + Value::String(STUB_STRING.to_string()) + } + }; + + // Entity refs use an object only for compound identities. Unary key_vars and the ordinary + // id_field identity are normalized to their scalar atom by live CML environment handling. + if target.key_vars.len() >= 2 { + return Value::Object( + target + .key_vars + .iter() + .map(|key| (key.to_string(), key_stub(key.as_str()))) + .collect(), + ); + } + + let key_name = target + .key_vars + .first() + .map(|key| key.as_str()) + .unwrap_or_else(|| target.id_field.as_str()); + key_stub(key_name) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn hydrate_fixture() -> CGS { + let dir = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("../../fixtures/schemas/hydrate_invoke_target"); + plasm_core::load_schema(&dir).expect("load hydrate_invoke_target fixture") + } + + #[test] + fn emits_semantically_valid_scalar_stubs() { + let cgs = hydrate_fixture(); + assert_eq!( + preflight_compile_stub_value(&cgs.values["nv_request_id"], &cgs), + Value::String(ZERO_UUID.to_string()) + ); + assert_eq!( + preflight_compile_stub_value(&cgs.values["nv_observed_at"], &cgs), + Value::String(STUB_RFC3339.to_string()) + ); + assert_eq!( + preflight_compile_stub_value(&cgs.values["nv_observed_at_ms"], &cgs), + Value::Integer(0) + ); + } + + #[test] + fn emits_scalar_entity_ref_stub_for_single_identity() { + let cgs = hydrate_fixture(); + assert_eq!( + preflight_compile_stub_value(&cgs.values["nv_team_ref"], &cgs), + Value::String(STUB_STRING.to_string()) + ); + } +} diff --git a/crates/plasm-runtime/src/execution/compile_preflight.rs b/crates/plasm-runtime/src/execution/compile_preflight.rs index 41f4eb9f..0aecce63 100644 --- a/crates/plasm-runtime/src/execution/compile_preflight.rs +++ b/crates/plasm-runtime/src/execution/compile_preflight.rs @@ -138,7 +138,7 @@ fn preflight_compile_create( message: e.to_string(), } })?; - apply_preflight_compile_stubs(&mut env, capability); + apply_preflight_compile_stubs(&mut env, capability, cgs)?; merge_plasm_execute_session_env(&mut env); compile_operation_dispatch(&capability_template, &env).map(|_| ()) } @@ -237,7 +237,7 @@ fn preflight_compile_invoke(invoke: &InvokeExpr, cgs: &CGS) -> Result<(), Runtim } })?; merge_entity_id_from_into_input_env(&mut env, target_ent, capability); - apply_preflight_compile_stubs(&mut env, capability); + apply_preflight_compile_stubs(&mut env, capability, cgs)?; merge_plasm_execute_session_env(&mut env); compile_operation_dispatch(&capability_template, &env).map(|_| ()) } diff --git a/crates/plasm-runtime/src/lib.rs b/crates/plasm-runtime/src/lib.rs index 860e26f9..8c38bfc1 100644 --- a/crates/plasm-runtime/src/lib.rs +++ b/crates/plasm-runtime/src/lib.rs @@ -105,6 +105,7 @@ pub mod auth_resolution; pub mod binding_kv; pub mod branch_commit; pub mod cache; +mod compile_stub_value; pub mod error; pub mod evm; pub mod execution; diff --git a/crates/plasm-runtime/src/preflight.rs b/crates/plasm-runtime/src/preflight.rs index 5a48cc72..43344a18 100644 --- a/crates/plasm-runtime/src/preflight.rs +++ b/crates/plasm-runtime/src/preflight.rs @@ -11,17 +11,20 @@ use plasm_core::preflight::{ }; use plasm_core::TypedFieldValue; use plasm_core::{ - CapabilitySchema, EntityDef, GetExpr, InvokeExpr, Predicate, QueryExpr, Ref, Value, CGS, + CapabilityKind, CapabilitySchema, EntityDef, GetExpr, InvokeExpr, Predicate, QueryExpr, Ref, + Value, CGS, }; use std::collections::HashSet; +use crate::compile_stub_value::{preflight_compile_stub_value, STUB_STRING, ZERO_UUID}; + pub(crate) fn merge_preflight_fields_into_env( env: &mut CmlEnv, prefix: &str, fields: &IndexMap, ) { - for (k, v) in fields { - env.insert(format!("{prefix}_{k}"), v.to_value()); + for (field_name, value) in fields { + env.insert(format!("{prefix}_{field_name}"), value.to_value()); } } @@ -53,7 +56,10 @@ pub(crate) async fn apply_preflight_steps( let Some(PreflightInvoke { invoke }) = invoke else { continue; }; - hydrate_invoke_target(engine, cgs, cache, mode, env, invoke, get, prefix).await?; + hydrate_invoke_target( + engine, capability, cgs, cache, mode, env, invoke, get, prefix, + ) + .await?; } PreflightStep::HydrateEntityRefParam { param, get, merge } => { if !env_param_present(env, param) { @@ -121,9 +127,18 @@ pub(crate) async fn apply_preflight_steps( } /// Compile-only preflight: inject stub merge keys so CML templates compile without HTTP hydration. -pub(crate) fn apply_preflight_compile_stubs(env: &mut CmlEnv, capability: &CapabilitySchema) { +/// +/// Invoke-target stubs cover the referenced Get entity's declared fields, preserving the catalog +/// contract that `provides` does not strip decoded fields. This compile environment is necessarily +/// a schema-derived superset: live hydration merges only fields actually present in the decoded +/// row. Undeclared CML variables such as `ds_typo` are still rejected. +pub(crate) fn apply_preflight_compile_stubs( + env: &mut CmlEnv, + capability: &CapabilitySchema, + cgs: &CGS, +) -> Result<(), RuntimeError> { let Some(PreflightPlan(steps)) = capability.preflight.as_ref() else { - return; + return Ok(()); }; for step in steps { match step { @@ -133,7 +148,10 @@ pub(crate) fn apply_preflight_compile_stubs(env: &mut CmlEnv, capability: &Capab } for wire_key in merge.keys() { if env.get(wire_key).is_none() { - env.insert(wire_key.clone(), preflight_compile_stub_value(wire_key)); + env.insert( + wire_key.clone(), + preflight_wire_key_compile_stub_value(wire_key), + ); } } } @@ -145,7 +163,10 @@ pub(crate) fn apply_preflight_compile_stubs(env: &mut CmlEnv, capability: &Capab } for wire_key in merge.keys() { if env.get(wire_key).is_none() { - env.insert(wire_key.clone(), preflight_compile_stub_value(wire_key)); + env.insert( + wire_key.clone(), + preflight_wire_key_compile_stub_value(wire_key), + ); } } } @@ -161,21 +182,117 @@ pub(crate) fn apply_preflight_compile_stubs(env: &mut CmlEnv, capability: &Capab env.insert(merge.clone(), Value::Array(Vec::new())); } } - PreflightStep::HydrateInvokeTarget { .. } => {} + PreflightStep::HydrateInvokeTarget { get, prefix } => { + // Schema validation rejects this step on Create. Match live preflight's defensive + // behavior if an unvalidated CGS is passed directly to the runtime. + if capability.kind != CapabilityKind::Create { + add_hydrate_invoke_target_compile_stubs(env, capability, cgs, get, prefix)?; + } + } PreflightStep::ExistenceCheck { .. } => {} } } + Ok(()) +} + +struct HydrateInvokeTargetContract<'a> { + prefix: String, + get: &'a CapabilitySchema, + entity: &'a EntityDef, } -fn preflight_compile_stub_value(wire_key: &str) -> Value { +/// Resolve the schema contract shared by compile-only stubbing and live hydration. +/// +/// `CGS::validate` owns these invariants for loaded catalogs. The checks here are a defensive +/// boundary for callers that construct a CGS programmatically and invoke runtime APIs directly. +fn resolve_hydrate_invoke_target_contract<'a>( + capability: &CapabilitySchema, + cgs: &'a CGS, + get_name: &str, + prefix: &str, +) -> Result, RuntimeError> { + let prefix = prefix.trim(); + if prefix.is_empty() { + return Err(RuntimeError::ConfigurationError { + message: format!( + "preflight hydrate_invoke_target on capability '{}': prefix must not be empty", + capability.name + ), + }); + } + let get = cgs.get_capability(get_name).ok_or_else(|| { + RuntimeError::ConfigurationError { + message: format!( + "preflight hydrate_invoke_target on capability '{}': Get capability '{}' was not found", + capability.name, get_name + ), + } + })?; + if get.kind != CapabilityKind::Get { + return Err(RuntimeError::ConfigurationError { + message: format!( + "preflight hydrate_invoke_target on capability '{}': capability '{}' must be kind get", + capability.name, get_name + ), + }); + } + if get.domain != capability.domain { + return Err(RuntimeError::ConfigurationError { + message: format!( + "preflight hydrate_invoke_target on capability '{}': Get '{}' is for entity {}, expected {}", + capability.name, get_name, get.domain, capability.domain + ), + }); + } + let entity = cgs.get_entity(get.domain.as_str()).ok_or_else(|| { + RuntimeError::ConfigurationError { + message: format!( + "preflight hydrate_invoke_target on capability '{}': entity '{}' for Get '{}' was not found", + capability.name, get.domain, get_name + ), + } + })?; + Ok(HydrateInvokeTargetContract { + prefix: prefix.to_string(), + get, + entity, + }) +} + +fn add_hydrate_invoke_target_compile_stubs( + env: &mut CmlEnv, + capability: &CapabilitySchema, + cgs: &CGS, + get_name: &str, + prefix: &str, +) -> Result<(), RuntimeError> { + let contract = resolve_hydrate_invoke_target_contract(capability, cgs, get_name, prefix)?; + for (field_name, field) in &contract.entity.fields { + let env_key = format!("{}_{field_name}", contract.prefix); + let named_value = cgs.named_value_for_slot(field).map_err(|error| { + RuntimeError::ConfigurationError { + message: format!( + "preflight hydrate_invoke_target on capability '{}': cannot resolve type of field '{}': {error}", + capability.name, field_name + ), + } + })?; + // Live hydration treats this prefix as authoritative and overwrites colliding input keys. + // Compile-only hydration must do the same so both paths compile against the same namespace. + env.insert(env_key, preflight_compile_stub_value(named_value, cgs)); + } + Ok(()) +} + +fn preflight_wire_key_compile_stub_value(wire_key: &str) -> Value { if wire_key.ends_with("Id") || wire_key.ends_with("_id") || wire_key == "id" || wire_key.ends_with("Ids") { - Value::String("00000000-0000-0000-0000-000000000000".to_string()) + Value::String(ZERO_UUID.to_string()) } else { - Value::String("preflight-stub".to_string()) + Value::String(STUB_STRING.to_string()) } } @@ -186,24 +303,20 @@ fn env_param_present(env: &CmlEnv, name: &str) -> bool { #[allow(clippy::too_many_arguments)] async fn hydrate_invoke_target( engine: &ExecutionEngine, + capability: &CapabilitySchema, cgs: &CGS, cache: &mut SessionMaterialization, mode: ExecutionMode, env: &mut CmlEnv, invoke: &InvokeExpr, - get_cap: &str, + get_name: &str, prefix: &str, ) -> Result<(), RuntimeError> { - let prefix = prefix.trim(); - if prefix.is_empty() { - return Err(RuntimeError::ConfigurationError { - message: "preflight hydrate_invoke_target: prefix must not be empty".to_string(), - }); - } + let contract = resolve_hydrate_invoke_target_contract(capability, cgs, get_name, prefix)?; if let Some(entity) = cache.get(&invoke.target) { if entity.completeness == EntityCompleteness::Complete { - merge_preflight_fields_into_env(env, prefix, &entity.fields); + merge_preflight_fields_into_env(env, &contract.prefix, &entity.fields); return Ok(()); } } @@ -219,14 +332,14 @@ async fn hydrate_invoke_target( &get, cgs, mode, - Some(get_cap), + Some(contract.get.name.as_str()), false, Some(cache), &ViewAmbientContext::default(), ) .await?; cache.insert(cached.clone())?; - merge_preflight_fields_into_env(env, prefix, &cached.fields); + merge_preflight_fields_into_env(env, &contract.prefix, &cached.fields); Ok(()) } @@ -671,6 +784,32 @@ mod tests { use super::*; use plasm_core::Predicate; + fn hydrate_fixture() -> CGS { + let dir = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("../../fixtures/schemas/hydrate_invoke_target"); + plasm_core::load_schema(&dir).expect("load hydrate_invoke_target fixture") + } + + #[test] + fn hydrate_invoke_target_compile_stub_overwrites_existing_env_like_live_hydration() { + let cgs = hydrate_fixture(); + let capability = cgs + .get_capability("datasource_run") + .expect("datasource_run capability"); + let mut env = CmlEnv::new(); + env.insert( + "ds_type".to_string(), + Value::String("already-present".to_string()), + ); + + apply_preflight_compile_stubs(&mut env, capability, &cgs).expect("compile stubs"); + + assert_eq!( + env.get("ds_type"), + Some(&Value::String(STUB_STRING.to_string())) + ); + } + #[test] fn scope_bind_from_param_builds_predicate() { let mut env = CmlEnv::new(); diff --git a/crates/plasm-runtime/src/view_stub_rows.rs b/crates/plasm-runtime/src/view_stub_rows.rs index 0f24f7c7..ea0198e9 100644 --- a/crates/plasm-runtime/src/view_stub_rows.rs +++ b/crates/plasm-runtime/src/view_stub_rows.rs @@ -10,6 +10,9 @@ use crate::cache::{CachedEntity, EntityCompleteness}; use crate::execution::{current_timestamp, ExecutionResult, ExecutionSource, ExecutionStats}; use crate::RuntimeError; +/// View preflight rows are rendered into entity references and downstream node bindings. Keep +/// their placeholders scalar and stable rather than sharing CML compile-environment stubs, whose +/// normalized entity-ref and wire-format shapes serve a different contract. fn placeholder_value(field_type: &FieldType) -> Value { match field_type { FieldType::Boolean => Value::Bool(false), @@ -173,6 +176,21 @@ mod tests { use super::*; use crate::view_test_support::matrix_views_cgs; + #[test] + fn view_placeholders_remain_scalar_and_stable() { + assert_eq!(placeholder_value(&FieldType::Number), Value::Integer(0)); + assert_eq!( + placeholder_value(&FieldType::String), + Value::String(String::new()) + ); + assert_eq!( + placeholder_value(&FieldType::EntityRef { + target: "File".into() + }), + Value::String("stub-File".into()) + ); + } + #[test] fn stub_query_uses_provides_fields() { let cgs = matrix_views_cgs(); diff --git a/crates/plasm-runtime/tests/hydrate_invoke_target.rs b/crates/plasm-runtime/tests/hydrate_invoke_target.rs new file mode 100644 index 00000000..534ca003 --- /dev/null +++ b/crates/plasm-runtime/tests/hydrate_invoke_target.rs @@ -0,0 +1,204 @@ +use async_trait::async_trait; +use indexmap::IndexMap; +use plasm_compile::CompiledRequest; +use plasm_core::{Expr, InvokeExpr, Ref, Value, CGS}; +use plasm_runtime::auth::ResolvedAuth; +use plasm_runtime::http_transport::HttpTransport; +use plasm_runtime::{ + preflight_compile_expr, CachedEntity, EntityCompleteness, ExecuteOptions, ExecutionConfig, + ExecutionEngine, ExecutionMode, RuntimeError, SessionMaterialization, StreamConsumeOpts, + ViewAmbientContext, +}; +use std::sync::{Arc, Mutex}; + +fn fixture() -> CGS { + let dir = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("../../fixtures/schemas/hydrate_invoke_target"); + plasm_core::load_schema(&dir).expect("load hydrate_invoke_target fixture") +} + +fn run_expr(capability: &str) -> Expr { + let mut input = indexmap::indexmap! { + "expr".to_string() => Value::String("up".to_string()), + }; + if capability == "datasource_run" { + input.insert( + "from".to_string(), + Value::String("2026-08-03T03:50:00Z".to_string()), + ); + input.insert( + "to".to_string(), + Value::String("2026-08-03T04:00:00Z".to_string()), + ); + } + Expr::Invoke(InvokeExpr::new( + capability, + "Datasource", + "prometheus", + Some(Value::Object(input)), + )) +} + +#[test] +fn static_compile_hydrates_declared_entity_fields_and_honors_prefix() { + let cgs = fixture(); + let ambient = ViewAmbientContext::default(); + + preflight_compile_expr(&run_expr("datasource_run"), &cgs, &ambient) + .expect("ds_type comes from Datasource fields even though provides omits it"); + preflight_compile_expr(&run_expr("datasource_run_source_prefix"), &cgs, &ambient) + .expect("source_type should honor the configured prefix"); + + let error = preflight_compile_expr(&run_expr("datasource_run_typo"), &cgs, &ambient) + .expect_err("ds_typo must remain an unknown CML variable"); + assert!(error.to_string().contains("ds_typo"), "{error}"); +} + +#[test] +fn grafana_datasource_actions_compile_with_unchanged_ds_type_mapping() { + let dir = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../apis/grafana"); + let cgs = plasm_core::load_schema(&dir).expect("load Grafana catalog"); + let ambient = ViewAmbientContext::default(); + let cases = [ + ( + "datasource_query_run", + indexmap::indexmap! { + "expr".to_string() => Value::String("up".to_string()), + }, + ), + ( + "datasource_clickhouse_query_run", + indexmap::indexmap! { + "query".to_string() => Value::String("SELECT 1".to_string()), + }, + ), + ( + "panel_query_run", + indexmap::indexmap! { + "expr".to_string() => Value::String("up".to_string()), + }, + ), + ]; + + for (capability, input) in cases { + let expr = Expr::Invoke(InvokeExpr::new( + capability, + "Datasource", + "prometheus", + Some(Value::Object(input)), + )); + preflight_compile_expr(&expr, &cgs, &ambient) + .unwrap_or_else(|error| panic!("{capability} failed static compile: {error}")); + } +} + +#[derive(Clone, Default)] +struct RecordingTransport { + requests: Arc>>, +} + +#[async_trait] +impl HttpTransport for RecordingTransport { + async fn send_compiled_http( + &self, + _base_url: &str, + request: &CompiledRequest, + _auth: Option, + ) -> Result<(serde_json::Value, Option), RuntimeError> { + self.requests.lock().unwrap().push(request.clone()); + if request.path.ends_with("/api/datasources/uid/prometheus") { + return Ok(( + serde_json::json!({"uid": "prometheus", "type": "prometheus"}), + None, + )); + } + if request.path.ends_with("/api/ds/query") { + return Ok(( + serde_json::json!({ + "uid": "prometheus", + "query_results": {"status": "success"} + }), + None, + )); + } + panic!("unexpected request: {request:?}"); + } + + async fn get_json_absolute( + &self, + url: &str, + _auth: Option, + ) -> Result<(serde_json::Value, Option), RuntimeError> { + panic!("unexpected absolute GET: {url}"); + } +} + +async fn execute(cgs: &CGS, cached: Option) -> Vec { + let transport = RecordingTransport::default(); + let requests = transport.requests.clone(); + let engine = + ExecutionEngine::new_with_transport(ExecutionConfig::default(), Arc::new(transport), None); + let mut cache = SessionMaterialization::new(); + if let Some(completeness) = cached { + cache + .insert(CachedEntity::from_decoded( + Ref::new("Datasource", "prometheus"), + indexmap::indexmap! { + "uid".to_string() => Value::String("prometheus".to_string()), + "type".to_string() => Value::String("prometheus".to_string()), + }, + IndexMap::new(), + 1, + completeness, + )) + .expect("seed cache"); + } + + engine + .execute( + &run_expr("datasource_run"), + cgs, + &mut cache, + Some(ExecutionMode::Live), + StreamConsumeOpts::default(), + ExecuteOptions::default(), + ) + .await + .expect("execute hydrated action"); + let captured = requests.lock().unwrap().clone(); + captured +} + +fn action_body(requests: &[CompiledRequest]) -> serde_json::Value { + let request = requests + .iter() + .find(|request| request.path.ends_with("/api/ds/query")) + .expect("outbound action request"); + serde_json::to_value(request.body.as_ref().expect("action body")).expect("serialize body") +} + +#[tokio::test] +async fn live_hydration_dispatches_real_value_for_fresh_complete_and_incomplete_cache_paths() { + let cgs = fixture(); + let fresh = execute(&cgs, None).await; + let complete = execute(&cgs, Some(EntityCompleteness::Complete)).await; + let incomplete = execute(&cgs, Some(EntityCompleteness::Summary)).await; + + assert_eq!(fresh.len(), 2, "fresh path must GET then POST"); + assert_eq!(complete.len(), 1, "complete cache must skip GET"); + assert_eq!(incomplete.len(), 2, "incomplete cache must GET then POST"); + + let expected = serde_json::json!({ + "queries": [{ + "datasource": {"type": "prometheus", "uid": "prometheus"}, + "expr": "up" + }], + "from": "2026-08-03T03:50:00Z", + "to": "2026-08-03T04:00:00Z" + }); + for requests in [&fresh, &complete, &incomplete] { + let body = action_body(requests); + assert_eq!(body, expected); + assert!(!body.to_string().contains("preflight-stub")); + } +} diff --git a/fixtures/schemas/hydrate_invoke_target/domain.yaml b/fixtures/schemas/hydrate_invoke_target/domain.yaml new file mode 100644 index 00000000..dc0a8c24 --- /dev/null +++ b/fixtures/schemas/hydrate_invoke_target/domain.yaml @@ -0,0 +1,133 @@ +version: 1 +http_backend: http://localhost:1080 + +values: + nv_datasource_uid: + type: string + string_semantics: short + nv_datasource_type: + type: string + string_semantics: short + nv_query_expr: + type: string + string_semantics: short + nv_query_time: + type: string + string_semantics: short + nv_query_results: + type: json + nv_request_id: + type: uuid + nv_observed_at: + type: date + value_format: rfc3339 + nv_observed_at_ms: + type: date + value_format: unix_ms + nv_team_key: + type: string + string_semantics: short + nv_team_ref: + type: entity_ref + target: Team + +entities: + Team: + id_field: key + fields: + key: + required: true + value_ref: nv_team_key + Datasource: + id_field: uid + description: Datasource used to test invoke-target preflight hydration. + fields: + uid: + required: true + value_ref: nv_datasource_uid + type: + required: true + value_ref: nv_datasource_type + query_results: + required: false + value_ref: nv_query_results + request_id: + required: false + value_ref: nv_request_id + observed_at: + required: false + value_ref: nv_observed_at + observed_at_ms: + required: false + value_ref: nv_observed_at_ms + team: + required: false + value_ref: nv_team_ref + +capabilities: + team_get: + kind: get + entity: Team + + datasource_get: + kind: get + entity: Datasource + provides: + - uid + # `type` is deliberately omitted: provides must not restrict decoded-row hydration. + # Unknown provides entries also degrade gracefully. + - unavailable_legacy_field + + datasource_run: + kind: action + entity: Datasource + preflight: + - kind: hydrate_invoke_target + get: datasource_get + prefix: ds + parameters: + - name: expr + value_ref: nv_query_expr + required: true + - name: from + value_ref: nv_query_time + required: true + - name: to + value_ref: nv_query_time + required: true + provides: + - uid + - query_results + + datasource_run_typo: + kind: action + entity: Datasource + preflight: + - kind: hydrate_invoke_target + get: datasource_get + prefix: ds + parameters: + - name: expr + value_ref: nv_query_expr + required: true + provides: + - uid + - query_results + + datasource_run_source_prefix: + kind: action + entity: Datasource + preflight: + - kind: hydrate_invoke_target + get: datasource_get + prefix: source + parameters: + - name: expr + value_ref: nv_query_expr + required: true + provides: + - uid + - query_results + +auth: + scheme: none diff --git a/fixtures/schemas/hydrate_invoke_target/mappings.yaml b/fixtures/schemas/hydrate_invoke_target/mappings.yaml new file mode 100644 index 00000000..ad27ab1d --- /dev/null +++ b/fixtures/schemas/hydrate_invoke_target/mappings.yaml @@ -0,0 +1,68 @@ +team_get: + method: GET + path: + - { type: literal, value: teams } + - { type: var, name: id } + response: + single: true + +datasource_get: + method: GET + path: + - { type: literal, value: api } + - { type: literal, value: datasources } + - { type: literal, value: uid } + - { type: var, name: id } + response: + single: true + +datasource_run: + method: POST + path: + - { type: literal, value: api } + - { type: literal, value: ds } + - { type: literal, value: query } + body: + type: object + fields: + - [queries, { type: array, elements: [ + { type: object, fields: [ + [datasource, { type: object, fields: [ + [type, { type: var, name: ds_type }], + [uid, { type: var, name: id }] + ]}], + [expr, { type: var, name: expr }] + ]} + ]}] + - [from, { type: var, name: from }] + - [to, { type: var, name: to }] + response: + single: true + +datasource_run_typo: + method: POST + path: + - { type: literal, value: api } + - { type: literal, value: ds } + - { type: literal, value: query } + body: + type: object + fields: + - [datasourceType, { type: var, name: ds_typo }] + - [expr, { type: var, name: expr }] + response: + single: true + +datasource_run_source_prefix: + method: POST + path: + - { type: literal, value: api } + - { type: literal, value: ds } + - { type: literal, value: query } + body: + type: object + fields: + - [datasourceType, { type: var, name: source_type }] + - [expr, { type: var, name: expr }] + response: + single: true From 666bccf987fc0251c9b9adcda65a0b5462a1e87e Mon Sep 17 00:00:00 2001 From: Natanael Mojica Date: Fri, 14 Aug 2026 08:07:42 -0600 Subject: [PATCH 2/2] fix(runtime): keep invoke hydration compile stubs narrow --- .../plasm-runtime/src/compile_stub_value.rs | 120 ------------ .../src/execution/compile_preflight.rs | 4 +- crates/plasm-runtime/src/lib.rs | 1 - crates/plasm-runtime/src/preflight.rs | 172 +++--------------- crates/plasm-runtime/src/view_stub_rows.rs | 18 -- .../tests/hydrate_invoke_target.rs | 163 +---------------- .../schemas/hydrate_invoke_target/domain.yaml | 38 ---- .../hydrate_invoke_target/mappings.yaml | 8 - 8 files changed, 31 insertions(+), 493 deletions(-) delete mode 100644 crates/plasm-runtime/src/compile_stub_value.rs diff --git a/crates/plasm-runtime/src/compile_stub_value.rs b/crates/plasm-runtime/src/compile_stub_value.rs deleted file mode 100644 index 5b519636..00000000 --- a/crates/plasm-runtime/src/compile_stub_value.rs +++ /dev/null @@ -1,120 +0,0 @@ -//! Schema-derived placeholder values used by compile-only runtime paths. -//! -//! These values must satisfy the declared domain shape closely enough for CML to parse, format, -//! or descend into them. They are never transport values. - -use indexmap::IndexMap; -use plasm_core::{FieldType, NamedValueSchema, TemporalWireFormat, Value, ValueWireFormat, CGS}; - -pub(crate) const ZERO_UUID: &str = "00000000-0000-0000-0000-000000000000"; -const STUB_RFC3339: &str = "1970-01-01T00:00:00Z"; -const STUB_ISO_DATE: &str = "1970-01-01"; -pub(crate) const STUB_STRING: &str = "preflight-stub"; - -/// Produce a valid, deterministic placeholder for a named value during static compilation. -pub(crate) fn preflight_compile_stub_value(named_value: &NamedValueSchema, cgs: &CGS) -> Value { - stub_value(named_value, cgs, 0) -} - -fn stub_value(named_value: &NamedValueSchema, cgs: &CGS, depth: usize) -> Value { - match &named_value.field_type { - FieldType::Boolean => Value::Bool(false), - FieldType::Number => Value::Float(0.0), - FieldType::Integer => Value::Integer(0), - FieldType::MultiSelect | FieldType::Array => Value::Array(Vec::new()), - FieldType::Json => Value::Object(IndexMap::new()), - FieldType::Uuid => Value::String(ZERO_UUID.to_string()), - FieldType::Date => match named_value.value_format { - Some(ValueWireFormat::Temporal( - TemporalWireFormat::UnixMs | TemporalWireFormat::UnixSec, - )) => Value::Integer(0), - Some(ValueWireFormat::Temporal(TemporalWireFormat::Iso8601Date)) => { - Value::String(STUB_ISO_DATE.to_string()) - } - Some(ValueWireFormat::Temporal(TemporalWireFormat::Rfc3339)) | None => { - Value::String(STUB_RFC3339.to_string()) - } - }, - FieldType::EntityRef { target } => entity_ref_stub_value(cgs, target, depth), - FieldType::String | FieldType::Select | FieldType::Blob => { - Value::String(STUB_STRING.to_string()) - } - } -} - -fn entity_ref_stub_value(cgs: &CGS, target_name: &str, depth: usize) -> Value { - let Some(target) = cgs.get_entity(target_name) else { - return Value::String(STUB_STRING.to_string()); - }; - - let key_stub = |key_name: &str| { - if depth < 2 { - target - .fields - .get(key_name) - .and_then(|field| field.named_value(cgs).ok()) - .map_or_else( - || Value::String(STUB_STRING.to_string()), - |key_type| stub_value(key_type, cgs, depth + 1), - ) - } else { - Value::String(STUB_STRING.to_string()) - } - }; - - // Entity refs use an object only for compound identities. Unary key_vars and the ordinary - // id_field identity are normalized to their scalar atom by live CML environment handling. - if target.key_vars.len() >= 2 { - return Value::Object( - target - .key_vars - .iter() - .map(|key| (key.to_string(), key_stub(key.as_str()))) - .collect(), - ); - } - - let key_name = target - .key_vars - .first() - .map(|key| key.as_str()) - .unwrap_or_else(|| target.id_field.as_str()); - key_stub(key_name) -} - -#[cfg(test)] -mod tests { - use super::*; - - fn hydrate_fixture() -> CGS { - let dir = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) - .join("../../fixtures/schemas/hydrate_invoke_target"); - plasm_core::load_schema(&dir).expect("load hydrate_invoke_target fixture") - } - - #[test] - fn emits_semantically_valid_scalar_stubs() { - let cgs = hydrate_fixture(); - assert_eq!( - preflight_compile_stub_value(&cgs.values["nv_request_id"], &cgs), - Value::String(ZERO_UUID.to_string()) - ); - assert_eq!( - preflight_compile_stub_value(&cgs.values["nv_observed_at"], &cgs), - Value::String(STUB_RFC3339.to_string()) - ); - assert_eq!( - preflight_compile_stub_value(&cgs.values["nv_observed_at_ms"], &cgs), - Value::Integer(0) - ); - } - - #[test] - fn emits_scalar_entity_ref_stub_for_single_identity() { - let cgs = hydrate_fixture(); - assert_eq!( - preflight_compile_stub_value(&cgs.values["nv_team_ref"], &cgs), - Value::String(STUB_STRING.to_string()) - ); - } -} diff --git a/crates/plasm-runtime/src/execution/compile_preflight.rs b/crates/plasm-runtime/src/execution/compile_preflight.rs index 0aecce63..c8c82dcb 100644 --- a/crates/plasm-runtime/src/execution/compile_preflight.rs +++ b/crates/plasm-runtime/src/execution/compile_preflight.rs @@ -138,7 +138,7 @@ fn preflight_compile_create( message: e.to_string(), } })?; - apply_preflight_compile_stubs(&mut env, capability, cgs)?; + apply_preflight_compile_stubs(&mut env, capability, cgs); merge_plasm_execute_session_env(&mut env); compile_operation_dispatch(&capability_template, &env).map(|_| ()) } @@ -237,7 +237,7 @@ fn preflight_compile_invoke(invoke: &InvokeExpr, cgs: &CGS) -> Result<(), Runtim } })?; merge_entity_id_from_into_input_env(&mut env, target_ent, capability); - apply_preflight_compile_stubs(&mut env, capability, cgs)?; + apply_preflight_compile_stubs(&mut env, capability, cgs); merge_plasm_execute_session_env(&mut env); compile_operation_dispatch(&capability_template, &env).map(|_| ()) } diff --git a/crates/plasm-runtime/src/lib.rs b/crates/plasm-runtime/src/lib.rs index 8c38bfc1..860e26f9 100644 --- a/crates/plasm-runtime/src/lib.rs +++ b/crates/plasm-runtime/src/lib.rs @@ -105,7 +105,6 @@ pub mod auth_resolution; pub mod binding_kv; pub mod branch_commit; pub mod cache; -mod compile_stub_value; pub mod error; pub mod evm; pub mod execution; diff --git a/crates/plasm-runtime/src/preflight.rs b/crates/plasm-runtime/src/preflight.rs index 43344a18..df825e63 100644 --- a/crates/plasm-runtime/src/preflight.rs +++ b/crates/plasm-runtime/src/preflight.rs @@ -11,20 +11,17 @@ use plasm_core::preflight::{ }; use plasm_core::TypedFieldValue; use plasm_core::{ - CapabilityKind, CapabilitySchema, EntityDef, GetExpr, InvokeExpr, Predicate, QueryExpr, Ref, - Value, CGS, + CapabilitySchema, EntityDef, GetExpr, InvokeExpr, Predicate, QueryExpr, Ref, Value, CGS, }; use std::collections::HashSet; -use crate::compile_stub_value::{preflight_compile_stub_value, STUB_STRING, ZERO_UUID}; - pub(crate) fn merge_preflight_fields_into_env( env: &mut CmlEnv, prefix: &str, fields: &IndexMap, ) { - for (field_name, value) in fields { - env.insert(format!("{prefix}_{field_name}"), value.to_value()); + for (k, v) in fields { + env.insert(format!("{prefix}_{k}"), v.to_value()); } } @@ -56,10 +53,7 @@ pub(crate) async fn apply_preflight_steps( let Some(PreflightInvoke { invoke }) = invoke else { continue; }; - hydrate_invoke_target( - engine, capability, cgs, cache, mode, env, invoke, get, prefix, - ) - .await?; + hydrate_invoke_target(engine, cgs, cache, mode, env, invoke, get, prefix).await?; } PreflightStep::HydrateEntityRefParam { param, get, merge } => { if !env_param_present(env, param) { @@ -127,18 +121,13 @@ pub(crate) async fn apply_preflight_steps( } /// Compile-only preflight: inject stub merge keys so CML templates compile without HTTP hydration. -/// -/// Invoke-target stubs cover the referenced Get entity's declared fields, preserving the catalog -/// contract that `provides` does not strip decoded fields. This compile environment is necessarily -/// a schema-derived superset: live hydration merges only fields actually present in the decoded -/// row. Undeclared CML variables such as `ds_typo` are still rejected. pub(crate) fn apply_preflight_compile_stubs( env: &mut CmlEnv, capability: &CapabilitySchema, cgs: &CGS, -) -> Result<(), RuntimeError> { +) { let Some(PreflightPlan(steps)) = capability.preflight.as_ref() else { - return Ok(()); + return; }; for step in steps { match step { @@ -183,105 +172,20 @@ pub(crate) fn apply_preflight_compile_stubs( } } PreflightStep::HydrateInvokeTarget { get, prefix } => { - // Schema validation rejects this step on Create. Match live preflight's defensive - // behavior if an unvalidated CGS is passed directly to the runtime. - if capability.kind != CapabilityKind::Create { - add_hydrate_invoke_target_compile_stubs(env, capability, cgs, get, prefix)?; + let Some(get_cap) = cgs.get_capability(get) else { + continue; + }; + let Some(entity) = cgs.get_entity(get_cap.domain.as_str()) else { + continue; + }; + for field_name in entity.fields.keys() { + let key = format!("{prefix}_{field_name}"); + env.insert(key.clone(), preflight_wire_key_compile_stub_value(&key)); } } PreflightStep::ExistenceCheck { .. } => {} } } - Ok(()) -} - -struct HydrateInvokeTargetContract<'a> { - prefix: String, - get: &'a CapabilitySchema, - entity: &'a EntityDef, -} - -/// Resolve the schema contract shared by compile-only stubbing and live hydration. -/// -/// `CGS::validate` owns these invariants for loaded catalogs. The checks here are a defensive -/// boundary for callers that construct a CGS programmatically and invoke runtime APIs directly. -fn resolve_hydrate_invoke_target_contract<'a>( - capability: &CapabilitySchema, - cgs: &'a CGS, - get_name: &str, - prefix: &str, -) -> Result, RuntimeError> { - let prefix = prefix.trim(); - if prefix.is_empty() { - return Err(RuntimeError::ConfigurationError { - message: format!( - "preflight hydrate_invoke_target on capability '{}': prefix must not be empty", - capability.name - ), - }); - } - let get = cgs.get_capability(get_name).ok_or_else(|| { - RuntimeError::ConfigurationError { - message: format!( - "preflight hydrate_invoke_target on capability '{}': Get capability '{}' was not found", - capability.name, get_name - ), - } - })?; - if get.kind != CapabilityKind::Get { - return Err(RuntimeError::ConfigurationError { - message: format!( - "preflight hydrate_invoke_target on capability '{}': capability '{}' must be kind get", - capability.name, get_name - ), - }); - } - if get.domain != capability.domain { - return Err(RuntimeError::ConfigurationError { - message: format!( - "preflight hydrate_invoke_target on capability '{}': Get '{}' is for entity {}, expected {}", - capability.name, get_name, get.domain, capability.domain - ), - }); - } - let entity = cgs.get_entity(get.domain.as_str()).ok_or_else(|| { - RuntimeError::ConfigurationError { - message: format!( - "preflight hydrate_invoke_target on capability '{}': entity '{}' for Get '{}' was not found", - capability.name, get.domain, get_name - ), - } - })?; - Ok(HydrateInvokeTargetContract { - prefix: prefix.to_string(), - get, - entity, - }) -} - -fn add_hydrate_invoke_target_compile_stubs( - env: &mut CmlEnv, - capability: &CapabilitySchema, - cgs: &CGS, - get_name: &str, - prefix: &str, -) -> Result<(), RuntimeError> { - let contract = resolve_hydrate_invoke_target_contract(capability, cgs, get_name, prefix)?; - for (field_name, field) in &contract.entity.fields { - let env_key = format!("{}_{field_name}", contract.prefix); - let named_value = cgs.named_value_for_slot(field).map_err(|error| { - RuntimeError::ConfigurationError { - message: format!( - "preflight hydrate_invoke_target on capability '{}': cannot resolve type of field '{}': {error}", - capability.name, field_name - ), - } - })?; - // Live hydration treats this prefix as authoritative and overwrites colliding input keys. - // Compile-only hydration must do the same so both paths compile against the same namespace. - env.insert(env_key, preflight_compile_stub_value(named_value, cgs)); - } - Ok(()) } fn preflight_wire_key_compile_stub_value(wire_key: &str) -> Value { @@ -290,9 +194,9 @@ fn preflight_wire_key_compile_stub_value(wire_key: &str) -> Value { || wire_key == "id" || wire_key.ends_with("Ids") { - Value::String(ZERO_UUID.to_string()) + Value::String("00000000-0000-0000-0000-000000000000".to_string()) } else { - Value::String(STUB_STRING.to_string()) + Value::String("preflight-stub".to_string()) } } @@ -303,20 +207,24 @@ fn env_param_present(env: &CmlEnv, name: &str) -> bool { #[allow(clippy::too_many_arguments)] async fn hydrate_invoke_target( engine: &ExecutionEngine, - capability: &CapabilitySchema, cgs: &CGS, cache: &mut SessionMaterialization, mode: ExecutionMode, env: &mut CmlEnv, invoke: &InvokeExpr, - get_name: &str, + get_cap: &str, prefix: &str, ) -> Result<(), RuntimeError> { - let contract = resolve_hydrate_invoke_target_contract(capability, cgs, get_name, prefix)?; + let prefix = prefix.trim(); + if prefix.is_empty() { + return Err(RuntimeError::ConfigurationError { + message: "preflight hydrate_invoke_target: prefix must not be empty".to_string(), + }); + } if let Some(entity) = cache.get(&invoke.target) { if entity.completeness == EntityCompleteness::Complete { - merge_preflight_fields_into_env(env, &contract.prefix, &entity.fields); + merge_preflight_fields_into_env(env, prefix, &entity.fields); return Ok(()); } } @@ -332,14 +240,14 @@ async fn hydrate_invoke_target( &get, cgs, mode, - Some(contract.get.name.as_str()), + Some(get_cap), false, Some(cache), &ViewAmbientContext::default(), ) .await?; cache.insert(cached.clone())?; - merge_preflight_fields_into_env(env, &contract.prefix, &cached.fields); + merge_preflight_fields_into_env(env, prefix, &cached.fields); Ok(()) } @@ -784,32 +692,6 @@ mod tests { use super::*; use plasm_core::Predicate; - fn hydrate_fixture() -> CGS { - let dir = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) - .join("../../fixtures/schemas/hydrate_invoke_target"); - plasm_core::load_schema(&dir).expect("load hydrate_invoke_target fixture") - } - - #[test] - fn hydrate_invoke_target_compile_stub_overwrites_existing_env_like_live_hydration() { - let cgs = hydrate_fixture(); - let capability = cgs - .get_capability("datasource_run") - .expect("datasource_run capability"); - let mut env = CmlEnv::new(); - env.insert( - "ds_type".to_string(), - Value::String("already-present".to_string()), - ); - - apply_preflight_compile_stubs(&mut env, capability, &cgs).expect("compile stubs"); - - assert_eq!( - env.get("ds_type"), - Some(&Value::String(STUB_STRING.to_string())) - ); - } - #[test] fn scope_bind_from_param_builds_predicate() { let mut env = CmlEnv::new(); diff --git a/crates/plasm-runtime/src/view_stub_rows.rs b/crates/plasm-runtime/src/view_stub_rows.rs index ea0198e9..0f24f7c7 100644 --- a/crates/plasm-runtime/src/view_stub_rows.rs +++ b/crates/plasm-runtime/src/view_stub_rows.rs @@ -10,9 +10,6 @@ use crate::cache::{CachedEntity, EntityCompleteness}; use crate::execution::{current_timestamp, ExecutionResult, ExecutionSource, ExecutionStats}; use crate::RuntimeError; -/// View preflight rows are rendered into entity references and downstream node bindings. Keep -/// their placeholders scalar and stable rather than sharing CML compile-environment stubs, whose -/// normalized entity-ref and wire-format shapes serve a different contract. fn placeholder_value(field_type: &FieldType) -> Value { match field_type { FieldType::Boolean => Value::Bool(false), @@ -176,21 +173,6 @@ mod tests { use super::*; use crate::view_test_support::matrix_views_cgs; - #[test] - fn view_placeholders_remain_scalar_and_stable() { - assert_eq!(placeholder_value(&FieldType::Number), Value::Integer(0)); - assert_eq!( - placeholder_value(&FieldType::String), - Value::String(String::new()) - ); - assert_eq!( - placeholder_value(&FieldType::EntityRef { - target: "File".into() - }), - Value::String("stub-File".into()) - ); - } - #[test] fn stub_query_uses_provides_fields() { let cgs = matrix_views_cgs(); diff --git a/crates/plasm-runtime/tests/hydrate_invoke_target.rs b/crates/plasm-runtime/tests/hydrate_invoke_target.rs index 534ca003..ce12bad0 100644 --- a/crates/plasm-runtime/tests/hydrate_invoke_target.rs +++ b/crates/plasm-runtime/tests/hydrate_invoke_target.rs @@ -1,15 +1,5 @@ -use async_trait::async_trait; -use indexmap::IndexMap; -use plasm_compile::CompiledRequest; -use plasm_core::{Expr, InvokeExpr, Ref, Value, CGS}; -use plasm_runtime::auth::ResolvedAuth; -use plasm_runtime::http_transport::HttpTransport; -use plasm_runtime::{ - preflight_compile_expr, CachedEntity, EntityCompleteness, ExecuteOptions, ExecutionConfig, - ExecutionEngine, ExecutionMode, RuntimeError, SessionMaterialization, StreamConsumeOpts, - ViewAmbientContext, -}; -use std::sync::{Arc, Mutex}; +use plasm_core::{Expr, InvokeExpr, Value, CGS}; +use plasm_runtime::{preflight_compile_expr, ViewAmbientContext}; fn fixture() -> CGS { let dir = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) @@ -53,152 +43,3 @@ fn static_compile_hydrates_declared_entity_fields_and_honors_prefix() { .expect_err("ds_typo must remain an unknown CML variable"); assert!(error.to_string().contains("ds_typo"), "{error}"); } - -#[test] -fn grafana_datasource_actions_compile_with_unchanged_ds_type_mapping() { - let dir = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../apis/grafana"); - let cgs = plasm_core::load_schema(&dir).expect("load Grafana catalog"); - let ambient = ViewAmbientContext::default(); - let cases = [ - ( - "datasource_query_run", - indexmap::indexmap! { - "expr".to_string() => Value::String("up".to_string()), - }, - ), - ( - "datasource_clickhouse_query_run", - indexmap::indexmap! { - "query".to_string() => Value::String("SELECT 1".to_string()), - }, - ), - ( - "panel_query_run", - indexmap::indexmap! { - "expr".to_string() => Value::String("up".to_string()), - }, - ), - ]; - - for (capability, input) in cases { - let expr = Expr::Invoke(InvokeExpr::new( - capability, - "Datasource", - "prometheus", - Some(Value::Object(input)), - )); - preflight_compile_expr(&expr, &cgs, &ambient) - .unwrap_or_else(|error| panic!("{capability} failed static compile: {error}")); - } -} - -#[derive(Clone, Default)] -struct RecordingTransport { - requests: Arc>>, -} - -#[async_trait] -impl HttpTransport for RecordingTransport { - async fn send_compiled_http( - &self, - _base_url: &str, - request: &CompiledRequest, - _auth: Option, - ) -> Result<(serde_json::Value, Option), RuntimeError> { - self.requests.lock().unwrap().push(request.clone()); - if request.path.ends_with("/api/datasources/uid/prometheus") { - return Ok(( - serde_json::json!({"uid": "prometheus", "type": "prometheus"}), - None, - )); - } - if request.path.ends_with("/api/ds/query") { - return Ok(( - serde_json::json!({ - "uid": "prometheus", - "query_results": {"status": "success"} - }), - None, - )); - } - panic!("unexpected request: {request:?}"); - } - - async fn get_json_absolute( - &self, - url: &str, - _auth: Option, - ) -> Result<(serde_json::Value, Option), RuntimeError> { - panic!("unexpected absolute GET: {url}"); - } -} - -async fn execute(cgs: &CGS, cached: Option) -> Vec { - let transport = RecordingTransport::default(); - let requests = transport.requests.clone(); - let engine = - ExecutionEngine::new_with_transport(ExecutionConfig::default(), Arc::new(transport), None); - let mut cache = SessionMaterialization::new(); - if let Some(completeness) = cached { - cache - .insert(CachedEntity::from_decoded( - Ref::new("Datasource", "prometheus"), - indexmap::indexmap! { - "uid".to_string() => Value::String("prometheus".to_string()), - "type".to_string() => Value::String("prometheus".to_string()), - }, - IndexMap::new(), - 1, - completeness, - )) - .expect("seed cache"); - } - - engine - .execute( - &run_expr("datasource_run"), - cgs, - &mut cache, - Some(ExecutionMode::Live), - StreamConsumeOpts::default(), - ExecuteOptions::default(), - ) - .await - .expect("execute hydrated action"); - let captured = requests.lock().unwrap().clone(); - captured -} - -fn action_body(requests: &[CompiledRequest]) -> serde_json::Value { - let request = requests - .iter() - .find(|request| request.path.ends_with("/api/ds/query")) - .expect("outbound action request"); - serde_json::to_value(request.body.as_ref().expect("action body")).expect("serialize body") -} - -#[tokio::test] -async fn live_hydration_dispatches_real_value_for_fresh_complete_and_incomplete_cache_paths() { - let cgs = fixture(); - let fresh = execute(&cgs, None).await; - let complete = execute(&cgs, Some(EntityCompleteness::Complete)).await; - let incomplete = execute(&cgs, Some(EntityCompleteness::Summary)).await; - - assert_eq!(fresh.len(), 2, "fresh path must GET then POST"); - assert_eq!(complete.len(), 1, "complete cache must skip GET"); - assert_eq!(incomplete.len(), 2, "incomplete cache must GET then POST"); - - let expected = serde_json::json!({ - "queries": [{ - "datasource": {"type": "prometheus", "uid": "prometheus"}, - "expr": "up" - }], - "from": "2026-08-03T03:50:00Z", - "to": "2026-08-03T04:00:00Z" - }); - for requests in [&fresh, &complete, &incomplete] { - let body = action_body(requests); - assert_eq!(body, expected); - assert!(!body.to_string().contains("preflight-stub")); - } -} diff --git a/fixtures/schemas/hydrate_invoke_target/domain.yaml b/fixtures/schemas/hydrate_invoke_target/domain.yaml index dc0a8c24..c514f7fe 100644 --- a/fixtures/schemas/hydrate_invoke_target/domain.yaml +++ b/fixtures/schemas/hydrate_invoke_target/domain.yaml @@ -16,28 +16,8 @@ values: string_semantics: short nv_query_results: type: json - nv_request_id: - type: uuid - nv_observed_at: - type: date - value_format: rfc3339 - nv_observed_at_ms: - type: date - value_format: unix_ms - nv_team_key: - type: string - string_semantics: short - nv_team_ref: - type: entity_ref - target: Team entities: - Team: - id_field: key - fields: - key: - required: true - value_ref: nv_team_key Datasource: id_field: uid description: Datasource used to test invoke-target preflight hydration. @@ -51,32 +31,14 @@ entities: query_results: required: false value_ref: nv_query_results - request_id: - required: false - value_ref: nv_request_id - observed_at: - required: false - value_ref: nv_observed_at - observed_at_ms: - required: false - value_ref: nv_observed_at_ms - team: - required: false - value_ref: nv_team_ref capabilities: - team_get: - kind: get - entity: Team - datasource_get: kind: get entity: Datasource provides: - uid # `type` is deliberately omitted: provides must not restrict decoded-row hydration. - # Unknown provides entries also degrade gracefully. - - unavailable_legacy_field datasource_run: kind: action diff --git a/fixtures/schemas/hydrate_invoke_target/mappings.yaml b/fixtures/schemas/hydrate_invoke_target/mappings.yaml index ad27ab1d..79778d1a 100644 --- a/fixtures/schemas/hydrate_invoke_target/mappings.yaml +++ b/fixtures/schemas/hydrate_invoke_target/mappings.yaml @@ -1,11 +1,3 @@ -team_get: - method: GET - path: - - { type: literal, value: teams } - - { type: var, name: id } - response: - single: true - datasource_get: method: GET path: