From 0949bda832c4b4c323cd5290daf923c22a0dabb4 Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Tue, 18 Aug 2026 22:42:03 -0700 Subject: [PATCH 1/7] feat: resolve saved calculated channels by name for one asset Adds a resolution step to the calculated channel service: names are looked up among active saved channels, then resolved through the API for a single asset and optional run. Names that do not resolve come back reported instead of dropped so a caller can name them. --- .../src/service/calculated_channels/mod.rs | 205 ++++++++- .../src/service/calculated_channels/test.rs | 399 +++++++++++++++++- .../crates/sift_mcp/src/service/common/mod.rs | 5 + 3 files changed, 588 insertions(+), 21 deletions(-) diff --git a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs index 6d2a82c41..3b5a48548 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs @@ -1,20 +1,29 @@ use crate::policy::{RetryPolicy, with_retry}; use crate::service::common; -use anyhow::{Context, Result, anyhow}; +use anyhow::{Context, Result, anyhow, bail}; use pbjson_types::FieldMask; use sift_rs::{ SiftChannel, - calculated_channels::v2::{ - CalculatedChannel, CalculatedChannelAbstractChannelReference, - CalculatedChannelAssetConfiguration, CalculatedChannelConfiguration, - CalculatedChannelQueryConfiguration, CalculatedChannelValidationResult, - CreateCalculatedChannelRequest, GetCalculatedChannelRequest, - ListCalculatedChannelVersionsRequest, ListCalculatedChannelVersionsResponse, - ListCalculatedChannelsRequest, ListCalculatedChannelsResponse, - UpdateCalculatedChannelRequest, - calculated_channel_asset_configuration::{AssetScope, AssetSelection}, - calculated_channel_query_configuration::{Query, Sel}, - calculated_channel_service_client::CalculatedChannelServiceClient, + calculated_channels::{ + v1::ExpressionRequest, + v2::{ + BatchResolveCalculatedChannelsRequest, BatchResolveCalculatedChannelsResponse, + CalculatedChannel, CalculatedChannelAbstractChannelReference, + CalculatedChannelAssetConfiguration, CalculatedChannelConfiguration, + CalculatedChannelQueryConfiguration, CalculatedChannelValidationResult, + CreateCalculatedChannelRequest, GetCalculatedChannelRequest, + ListCalculatedChannelVersionsRequest, ListCalculatedChannelVersionsResponse, + ListCalculatedChannelsRequest, ListCalculatedChannelsResponse, + ResolveCalculatedChannelRequest, UpdateCalculatedChannelRequest, + calculated_channel_asset_configuration::{AssetScope, AssetSelection}, + calculated_channel_query_configuration::{Query, Sel}, + calculated_channel_service_client::CalculatedChannelServiceClient, + resolve_calculated_channel_request::CalculatedChannel as ResolveTarget, + }, + }, + common::r#type::v1::{ + Ids, NamedResources, ResourceIdentifier, named_resources::Resources, + resource_identifier::Identifier, }, metadata::v1::MetadataValue, }; @@ -84,6 +93,39 @@ pub struct CalculatedChannelWrite { pub inapplicable_assets: Vec, } +/// A saved calculated channel resolved against one asset: the expression and +/// its channel references already point at that asset's channels, so it can be +/// queried for data as-is. +#[derive(Debug)] +pub struct ResolvedCalculation { + /// The name the caller asked for; also the data query's channel key. + pub name: String, + pub asset_name: String, + pub expression_request: ExpressionRequest, +} + +/// A requested calculated channel that yielded no query for the asset, with the +/// reason to report to the caller. +#[derive(Debug)] +pub struct UnresolvedCalculation { + pub name: String, + pub reason: String, +} + +/// The outcome of resolving a set of names: what can be queried, and what +/// cannot. Both halves are returned so a caller never silently drops a name it +/// was asked for. +#[derive(Debug)] +pub struct CalculationResolution { + pub resolved: Vec, + pub unresolved: Vec, +} + +const UNKNOWN_NAME_REASON: &str = "no active saved calculated channel has this name"; +const INAPPLICABLE_REASON: &str = + "does not apply to this asset: the asset is outside its asset scope or lacks a channel its \ + expression references"; + #[derive(Clone)] pub struct CalculatedChannelService { channel: SiftChannel, @@ -244,6 +286,145 @@ impl CalculatedChannelService { }) } + /// Resolves saved calculated channels, named by the caller, into concrete + /// expressions for a single asset, optionally narrowed to a run. Each name + /// is looked up among the active saved channels and then resolved by the + /// API, which is what decides whether the channel applies to the asset. + /// + /// Names with no saved channel, and channels the API cannot resolve for the + /// asset, are returned in `unresolved` rather than dropped, so the caller + /// can name them instead of returning a partial result silently. + pub async fn resolve_calculated_channels( + &self, + names: Vec, + asset_id: String, + run_id: Option, + ) -> Result { + let mut resolved = Vec::new(); + let mut unresolved = Vec::new(); + + if names.is_empty() { + return Ok(CalculationResolution { + resolved, + unresolved, + }); + } + + let quoted = names + .iter() + .map(|name| format!("\"{}\"", common::cel_escape(name))) + .collect::>() + .join(", "); + let stored = self + .list_calculated_channels( + format!("is_archived == false && name in [{quoted}]"), + None, + Some(common::PAGE_SIZE), + ) + .await?; + + // Keep the caller's order so the batch responses map back by index. + let mut targets = Vec::new(); + for name in names { + match stored.iter().find(|channel| channel.name == name) { + Some(channel) => targets.push((name, channel.calculated_channel_id.clone())), + None => unresolved.push(UnresolvedCalculation { + name, + reason: UNKNOWN_NAME_REASON.to_string(), + }), + } + } + + if targets.is_empty() { + return Ok(CalculationResolution { + resolved, + unresolved, + }); + } + + let requests = targets + .iter() + .map(|(_, id)| ResolveCalculatedChannelRequest { + calculated_channel: Some(ResolveTarget::Identifier(ResourceIdentifier { + identifier: Some(Identifier::Id(id.clone())), + })), + organization_id: String::new(), + assets: Some(NamedResources { + resources: Some(Resources::Ids(Ids { + ids: vec![asset_id.clone()], + })), + }), + run: run_id.clone().map(|id| ResourceIdentifier { + identifier: Some(Identifier::Id(id)), + }), + }) + .collect::>(); + + let grpc_channel = self.channel.clone(); + let resp = with_retry(&self.policy, move || { + let grpc_channel = grpc_channel.clone(); + let requests = requests.clone(); + async move { + let mut client = CalculatedChannelServiceClient::new(grpc_channel); + client + .batch_resolve_calculated_channels(BatchResolveCalculatedChannelsRequest { + requests, + }) + .await + .map(|resp| resp.into_inner()) + } + }) + .await + .context("failed to resolve calculated channels")?; + + let BatchResolveCalculatedChannelsResponse { responses } = resp; + if responses.len() != targets.len() { + bail!( + "resolve returned {} response(s) for {} requested calculated channel(s)", + responses.len(), + targets.len(), + ); + } + + for ((name, _), response) in targets.into_iter().zip(responses) { + // Only an entry for the requested asset is safe to query; an entry + // for another asset would pull that asset's channels instead. + let mut candidates = response.resolved; + let picked = candidates + .iter() + .position(|entry| entry.asset_id == asset_id) + .map(|index| candidates.swap_remove(index)); + + match picked { + Some(entry) => { + let Some(expression_request) = entry.expression_request else { + bail!("resolved calculated channel '{name}' is missing its expression"); + }; + resolved.push(ResolvedCalculation { + name, + asset_name: entry.asset_name, + expression_request, + }); + } + None => { + let reason = response + .unresolved + .into_iter() + .next() + .map(|entry| entry.error_message) + .filter(|message| !message.trim().is_empty()) + .unwrap_or_else(|| INAPPLICABLE_REASON.to_string()); + unresolved.push(UnresolvedCalculation { name, reason }); + } + } + } + + Ok(CalculationResolution { + resolved, + unresolved, + }) + } + pub async fn create_calculated_channel( &self, new: NewCalculatedChannel, diff --git a/rust/crates/sift_mcp/src/service/calculated_channels/test.rs b/rust/crates/sift_mcp/src/service/calculated_channels/test.rs index 3a9787f05..4cf62c555 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/test.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/test.rs @@ -1,13 +1,22 @@ use sift_rs::{ - calculated_channels::v2::{ - CalculatedChannel, CalculatedChannelAbstractChannelReference, - CalculatedChannelAssetConfiguration, CalculatedChannelConfiguration, - CalculatedChannelQueryConfiguration, CreateCalculatedChannelResponse, - GetCalculatedChannelResponse, ListCalculatedChannelVersionsResponse, - ListCalculatedChannelsResponse, UpdateCalculatedChannelResponse, - calculated_channel_asset_configuration::{AssetScope, AssetSelection}, - calculated_channel_query_configuration::{Query, Sel}, - calculated_channel_service_server::CalculatedChannelServiceServer, + calculated_channels::{ + v1::{ExpressionChannelReference, ExpressionRequest}, + v2::{ + BatchResolveCalculatedChannelsResponse, CalculatedChannel, + CalculatedChannelAbstractChannelReference, CalculatedChannelAssetConfiguration, + CalculatedChannelConfiguration, CalculatedChannelQueryConfiguration, + CreateCalculatedChannelResponse, GetCalculatedChannelResponse, + ListCalculatedChannelVersionsResponse, ListCalculatedChannelsResponse, + ResolveCalculatedChannelResponse, ResolvedCalculatedChannel, + UnresolvedCalculatedChannel, UpdateCalculatedChannelResponse, + calculated_channel_asset_configuration::{AssetScope, AssetSelection}, + calculated_channel_query_configuration::{Query, Sel}, + calculated_channel_service_server::CalculatedChannelServiceServer, + resolve_calculated_channel_request::CalculatedChannel as ResolveTarget, + }, + }, + common::r#type::v1::{ + ResourceIdentifier, named_resources::Resources, resource_identifier::Identifier, }, metadata::v1::{ MetadataKey, MetadataKeyType, MetadataValue, metadata_value::Value as MetadataValueInner, @@ -1045,3 +1054,375 @@ async fn archive_calculated_channel_sets_is_archived_true() { assert!(written.calculated_channel.is_archived); } + +/// Helper: a stored calculated channel carrying only the fields the resolution +/// path reads. +fn stored_channel(id: &str, name: &str) -> CalculatedChannel { + CalculatedChannel { + calculated_channel_id: id.into(), + name: name.into(), + ..Default::default() + } +} + +fn expression_request(expression: &str, channel_id: &str) -> ExpressionRequest { + ExpressionRequest { + expression: expression.into(), + expression_channel_references: vec![ExpressionChannelReference { + channel_reference: "$1".into(), + channel_id: channel_id.into(), + calculated_channel_reference: None, + }], + ..Default::default() + } +} + +#[tokio::test] +async fn resolve_calculated_channels_builds_lookup_and_resolve_requests() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels() + .times(1) + .withf(|req| { + let filter = &req.get_ref().filter; + filter.contains("is_archived == false") + && filter.contains("name in [\"thrust_margin\", \"chamber_delta\"]") + }) + .returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![ + stored_channel("cc2", "chamber_delta"), + stored_channel("cc1", "thrust_margin"), + ], + next_page_token: String::new(), + })) + }); + mock.expect_batch_resolve_calculated_channels() + .times(1) + .withf(|req| { + let requests = &req.get_ref().requests; + if requests.len() != 2 { + return false; + } + // Requests keep the caller's name order so responses can be mapped + // back by index. + let ids = requests + .iter() + .map(|r| match r.calculated_channel.as_ref() { + Some(ResolveTarget::Identifier(ResourceIdentifier { + identifier: Some(Identifier::Id(id)), + })) => id.clone(), + _ => String::new(), + }) + .collect::>(); + let assets = match requests[0].assets.as_ref().and_then(|a| a.resources.as_ref()) { + Some(Resources::Ids(ids)) => ids.ids.clone(), + _ => Vec::new(), + }; + let run = match requests[0].run.as_ref().and_then(|r| r.identifier.as_ref()) { + Some(Identifier::Id(id)) => id.clone(), + _ => String::new(), + }; + ids == vec!["cc1".to_string(), "cc2".to_string()] + && assets == vec!["asset-1".to_string()] + && run == "run-1" + }) + .returning(|req| { + let responses = req + .into_inner() + .requests + .into_iter() + .enumerate() + .map(|(i, _)| ResolveCalculatedChannelResponse { + calculated_channel_id: None, + resolved: vec![ResolvedCalculatedChannel { + asset_name: "bench".into(), + asset_id: "asset-1".into(), + expression_request: Some(expression_request( + "$1 * 2", + &format!("ch-{i}"), + )), + output_data_type: 0, + }], + unresolved: vec![], + }) + .collect(); + Ok(Response::new(BatchResolveCalculatedChannelsResponse { responses })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let resolution = service + .resolve_calculated_channels( + vec!["thrust_margin".to_string(), "chamber_delta".to_string()], + "asset-1".to_string(), + Some("run-1".to_string()), + ) + .await + .expect("resolve_calculated_channels failed"); + + assert!(resolution.unresolved.is_empty()); + assert_eq!( + resolution + .resolved + .iter() + .map(|r| r.name.as_str()) + .collect::>(), + vec!["thrust_margin", "chamber_delta"], + ); + assert_eq!(resolution.resolved[0].asset_name, "bench"); + assert_eq!( + resolution.resolved[0] + .expression_request + .expression_channel_references[0] + .channel_id, + "ch-0", + ); +} + +#[tokio::test] +async fn resolve_calculated_channels_omits_run_when_absent() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![stored_channel("cc1", "thrust_margin")], + next_page_token: String::new(), + })) + }); + mock.expect_batch_resolve_calculated_channels() + .times(1) + .withf(|req| req.get_ref().requests[0].run.is_none()) + .returning(|_| { + Ok(Response::new(BatchResolveCalculatedChannelsResponse { + responses: vec![ResolveCalculatedChannelResponse { + calculated_channel_id: None, + resolved: vec![ResolvedCalculatedChannel { + asset_name: "bench".into(), + asset_id: "asset-1".into(), + expression_request: Some(expression_request("$1", "ch-0")), + output_data_type: 0, + }], + unresolved: vec![], + }], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let resolution = service + .resolve_calculated_channels( + vec!["thrust_margin".to_string()], + "asset-1".to_string(), + None, + ) + .await + .expect("resolve_calculated_channels failed"); + + assert_eq!(resolution.resolved.len(), 1); +} + +/// One requested channel applies to the asset and one does not. The +/// inapplicable one must come back named, with the API's reason, instead of +/// being dropped so the caller only sees the applicable channel. +#[tokio::test] +async fn resolve_calculated_channels_reports_inapplicable_channel() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![ + stored_channel("cc1", "thrust_margin"), + stored_channel("cc2", "chamber_delta"), + ], + next_page_token: String::new(), + })) + }); + mock.expect_batch_resolve_calculated_channels() + .returning(|_| { + Ok(Response::new(BatchResolveCalculatedChannelsResponse { + responses: vec![ + ResolveCalculatedChannelResponse { + calculated_channel_id: None, + resolved: vec![ResolvedCalculatedChannel { + asset_name: "bench".into(), + asset_id: "asset-1".into(), + expression_request: Some(expression_request("$1", "ch-0")), + output_data_type: 0, + }], + unresolved: vec![], + }, + ResolveCalculatedChannelResponse { + calculated_channel_id: None, + resolved: vec![], + unresolved: vec![UnresolvedCalculatedChannel { + asset_name: "bench".into(), + error_message: "asset is missing channel chamber_pressure".into(), + }], + }, + ], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let resolution = service + .resolve_calculated_channels( + vec!["thrust_margin".to_string(), "chamber_delta".to_string()], + "asset-1".to_string(), + Some("run-1".to_string()), + ) + .await + .expect("resolve_calculated_channels failed"); + + assert_eq!(resolution.resolved.len(), 1); + assert_eq!(resolution.resolved[0].name, "thrust_margin"); + assert_eq!(resolution.unresolved.len(), 1); + assert_eq!(resolution.unresolved[0].name, "chamber_delta"); + assert!( + resolution.unresolved[0] + .reason + .contains("missing channel chamber_pressure"), + "reason should carry the API message: {}", + resolution.unresolved[0].reason, + ); +} + +/// A resolution that returns only other assets' entries does not apply to the +/// requested asset, so it must be reported instead of queried. +#[tokio::test] +async fn resolve_calculated_channels_rejects_other_asset_resolution() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![stored_channel("cc1", "thrust_margin")], + next_page_token: String::new(), + })) + }); + mock.expect_batch_resolve_calculated_channels() + .returning(|_| { + Ok(Response::new(BatchResolveCalculatedChannelsResponse { + responses: vec![ResolveCalculatedChannelResponse { + calculated_channel_id: None, + resolved: vec![ResolvedCalculatedChannel { + asset_name: "other-bench".into(), + asset_id: "asset-9".into(), + expression_request: Some(expression_request("$1", "ch-9")), + output_data_type: 0, + }], + unresolved: vec![], + }], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let resolution = service + .resolve_calculated_channels( + vec!["thrust_margin".to_string()], + "asset-1".to_string(), + None, + ) + .await + .expect("resolve_calculated_channels failed"); + + assert!(resolution.resolved.is_empty()); + assert_eq!(resolution.unresolved.len(), 1); + assert_eq!(resolution.unresolved[0].name, "thrust_margin"); +} + +/// A name with no stored calculated channel never reaches the resolve RPC and +/// comes back as unresolved with a reason the caller can act on. +#[tokio::test] +async fn resolve_calculated_channels_flags_unknown_name() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![], + next_page_token: String::new(), + })) + }); + mock.expect_batch_resolve_calculated_channels().times(0); + + let (service, _h) = service_with_mock(mock).await; + + let resolution = service + .resolve_calculated_channels( + vec!["not_a_channel".to_string()], + "asset-1".to_string(), + None, + ) + .await + .expect("resolve_calculated_channels failed"); + + assert!(resolution.resolved.is_empty()); + assert_eq!(resolution.unresolved.len(), 1); + assert_eq!(resolution.unresolved[0].name, "not_a_channel"); + assert!( + resolution.unresolved[0].reason.contains("no active saved"), + "reason should say the name is unknown: {}", + resolution.unresolved[0].reason, + ); +} + +#[tokio::test] +async fn resolve_calculated_channels_propagates_grpc_error() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![stored_channel("cc1", "thrust_margin")], + next_page_token: String::new(), + })) + }); + mock.expect_batch_resolve_calculated_channels() + .returning(|_| Err(Status::permission_denied("no access"))); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .resolve_calculated_channels( + vec!["thrust_margin".to_string()], + "asset-1".to_string(), + None, + ) + .await + .expect_err("expected the gRPC error to propagate"); + + assert!(err.to_string().contains("failed to resolve calculated channels")); +} + +/// A response count that does not line up with the requests would silently +/// mis-assign expressions to names, so it must fail loudly. +#[tokio::test] +async fn resolve_calculated_channels_errors_on_response_count_mismatch() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![ + stored_channel("cc1", "thrust_margin"), + stored_channel("cc2", "chamber_delta"), + ], + next_page_token: String::new(), + })) + }); + mock.expect_batch_resolve_calculated_channels() + .returning(|_| { + Ok(Response::new(BatchResolveCalculatedChannelsResponse { + responses: vec![ResolveCalculatedChannelResponse { + calculated_channel_id: None, + resolved: vec![], + unresolved: vec![], + }], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .resolve_calculated_channels( + vec!["thrust_margin".to_string(), "chamber_delta".to_string()], + "asset-1".to_string(), + None, + ) + .await + .expect_err("expected a mismatched response count to error"); + + assert!(err.to_string().contains("resolve")); +} diff --git a/rust/crates/sift_mcp/src/service/common/mod.rs b/rust/crates/sift_mcp/src/service/common/mod.rs index 1d0a888c4..07f82f3bb 100644 --- a/rust/crates/sift_mcp/src/service/common/mod.rs +++ b/rust/crates/sift_mcp/src/service/common/mod.rs @@ -57,6 +57,11 @@ pub fn paging(limit: Option) -> (u32, usize) { (limit, limit as usize) } +/// Escapes a value for interpolation into a double-quoted CEL string literal. +pub fn cel_escape(s: &str) -> String { + s.replace('\\', "\\\\").replace('"', "\\\"") +} + pub fn unix_nanos_to_secs_and_subsec_nanos(nanos: i64) -> (i64, i32) { let secs = nanos.div_euclid(NANOS_PER_SEC); let subsec_nanos = nanos.rem_euclid(NANOS_PER_SEC) as i32; From c67a9cf6879533d92b1ef6809d4085a7fa3233ca Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Tue, 18 Aug 2026 23:01:12 -0700 Subject: [PATCH 2/7] feat: serve saved calculated channels from get_data A channel name with no raw channel on the asset is resolved as an active saved calculated channel and queried with its resolved expression. Raw channels keep precedence on a shared name, so the existing path is unchanged. Channels that do not resolve are named in the result and in next_step, or in a RESOURCE_NOT_FOUND when nothing requested can be served, instead of returning a silently partial file. A calculated channel page carries no channel name, so its column falls back to the query key. --- .../src/service/calculated_channels/mod.rs | 5 +- .../src/service/calculated_channels/test.rs | 21 +- rust/crates/sift_mcp/src/service/data/mod.rs | 127 +++-- rust/crates/sift_mcp/src/service/data/test.rs | 75 ++- rust/crates/sift_mcp/src/tool/data/mod.rs | 155 ++++-- rust/crates/sift_mcp/src/tool/data/test.rs | 458 +++++++++++++++++- 6 files changed, 734 insertions(+), 107 deletions(-) diff --git a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs index 3b5a48548..0b1cfa86a 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs @@ -100,7 +100,6 @@ pub struct CalculatedChannelWrite { pub struct ResolvedCalculation { /// The name the caller asked for; also the data query's channel key. pub name: String, - pub asset_name: String, pub expression_request: ExpressionRequest, } @@ -122,8 +121,7 @@ pub struct CalculationResolution { } const UNKNOWN_NAME_REASON: &str = "no active saved calculated channel has this name"; -const INAPPLICABLE_REASON: &str = - "does not apply to this asset: the asset is outside its asset scope or lacks a channel its \ +const INAPPLICABLE_REASON: &str = "does not apply to this asset: the asset is outside its asset scope or lacks a channel its \ expression references"; #[derive(Clone)] @@ -402,7 +400,6 @@ impl CalculatedChannelService { }; resolved.push(ResolvedCalculation { name, - asset_name: entry.asset_name, expression_request, }); } diff --git a/rust/crates/sift_mcp/src/service/calculated_channels/test.rs b/rust/crates/sift_mcp/src/service/calculated_channels/test.rs index 4cf62c555..cf9c9e558 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/test.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/test.rs @@ -1114,7 +1114,11 @@ async fn resolve_calculated_channels_builds_lookup_and_resolve_requests() { _ => String::new(), }) .collect::>(); - let assets = match requests[0].assets.as_ref().and_then(|a| a.resources.as_ref()) { + let assets = match requests[0] + .assets + .as_ref() + .and_then(|a| a.resources.as_ref()) + { Some(Resources::Ids(ids)) => ids.ids.clone(), _ => Vec::new(), }; @@ -1137,16 +1141,15 @@ async fn resolve_calculated_channels_builds_lookup_and_resolve_requests() { resolved: vec![ResolvedCalculatedChannel { asset_name: "bench".into(), asset_id: "asset-1".into(), - expression_request: Some(expression_request( - "$1 * 2", - &format!("ch-{i}"), - )), + expression_request: Some(expression_request("$1 * 2", &format!("ch-{i}"))), output_data_type: 0, }], unresolved: vec![], }) .collect(); - Ok(Response::new(BatchResolveCalculatedChannelsResponse { responses })) + Ok(Response::new(BatchResolveCalculatedChannelsResponse { + responses, + })) }); let (service, _h) = service_with_mock(mock).await; @@ -1169,7 +1172,6 @@ async fn resolve_calculated_channels_builds_lookup_and_resolve_requests() { .collect::>(), vec!["thrust_margin", "chamber_delta"], ); - assert_eq!(resolution.resolved[0].asset_name, "bench"); assert_eq!( resolution.resolved[0] .expression_request @@ -1385,7 +1387,10 @@ async fn resolve_calculated_channels_propagates_grpc_error() { .await .expect_err("expected the gRPC error to propagate"); - assert!(err.to_string().contains("failed to resolve calculated channels")); + assert!( + err.to_string() + .contains("failed to resolve calculated channels") + ); } /// A response count that does not line up with the requests would silently diff --git a/rust/crates/sift_mcp/src/service/data/mod.rs b/rust/crates/sift_mcp/src/service/data/mod.rs index 86b6eecd3..f9ad27cae 100644 --- a/rust/crates/sift_mcp/src/service/data/mod.rs +++ b/rust/crates/sift_mcp/src/service/data/mod.rs @@ -32,7 +32,7 @@ use sift_rs::{ BytesValues, CalculatedChannelQuery, ChannelQuery, DoubleValue, DoubleValues, EnumValue, EnumValues, FloatValue, FloatValues, GetDataRequest, GetDataResponse, Int32Value, Int32Values, Int64Value, Int64Values, Query, StringValue, StringValues, Uint32Value, - Uint32Values, Uint64Value, Uint64Values, data_service_client::DataServiceClient, + Uint32Values, Uint64Value, Uint64Values, data_service_client::DataServiceClient, metadata, query::Query as QueryKind, }, runs::v2::Run, @@ -50,6 +50,17 @@ mod test; const ROW_FLUSH_THRESHOLD: usize = 1_000_000; const SIZE_FLUSH_THRESHOLD: usize = 64 << 20; +/// The column label for a returned channel page. A calculated channel page +/// carries no channel name and puts the query's channel key in `channel_id`, so +/// fall back to that key rather than emitting an unlabelled column. +fn column_label(channel: &metadata::Channel) -> &str { + if channel.name.is_empty() { + &channel.channel_id + } else { + &channel.name + } +} + #[derive(Clone)] pub struct DataService { channel: SiftChannel, @@ -106,6 +117,13 @@ pub enum ChannelInput { input_channels: Vec<(String, Channel)>, expression: String, }, + /// A saved calculated channel already resolved against the target asset. + /// The expression and its references come from the API's resolution, so + /// references to other calculated channels survive intact. + SavedCalculation { + channel_key: String, + expression_request: Box, + }, } pub enum TimeRange { @@ -295,6 +313,18 @@ impl DataService { })), } } + ChannelInput::SavedCalculation { + channel_key, + expression_request, + } => Query { + query: Some(QueryKind::CalculatedChannel(CalculatedChannelQuery { + channel_key: channel_key.clone(), + expression: Some((**expression_request).clone()), + run_id: run_id.clone(), + mode: Some(ExpressionMode::CalculatedChannels.into()), + combine_run_data: Some(false), + })), + }, }) .collect::>(); @@ -352,10 +382,11 @@ impl DataService { bail!("unexpected missing channel from metadata"); }; - let column_name = ColumnName::builder(&channel.name, &channel.channel_id) - .run(run_id.as_deref()) - .units(channel.unit.as_ref().map(|u| u.name.as_str())) - .build(); + let column_name = + ColumnName::builder(column_label(&channel), &channel.channel_id) + .run(run_id.as_deref()) + .units(channel.unit.as_ref().map(|u| u.name.as_str())) + .build(); let values = values .into_iter() @@ -408,10 +439,11 @@ impl DataService { let enum_config_md = serde_json::to_string(&channel.enum_types) .context("failed to serialize enum config to JSON")?; - let column_name = ColumnName::builder(&channel.name, &channel.channel_id) - .run(run_id.as_deref()) - .units(channel.unit.as_ref().map(|u| u.name.as_str())) - .build(); + let column_name = + ColumnName::builder(column_label(&channel), &channel.channel_id) + .run(run_id.as_deref()) + .units(channel.unit.as_ref().map(|u| u.name.as_str())) + .build(); let values = values .into_iter() @@ -468,7 +500,7 @@ impl DataService { for BitFieldElementValues { name, values } in values { let column_name = - ColumnName::builder(&channel.name, &channel.channel_id) + ColumnName::builder(column_label(&channel), &channel.channel_id) .bit_field_element(Some(&name)) .run(run_id.as_deref()) .units(channel.unit.as_ref().map(|u| u.name.as_str())) @@ -516,10 +548,11 @@ impl DataService { bail!("unexpected missing channel from metadata"); }; - let column_name = ColumnName::builder(&channel.name, &channel.channel_id) - .run(run_id.as_deref()) - .units(channel.unit.as_ref().map(|u| u.name.as_str())) - .build(); + let column_name = + ColumnName::builder(column_label(&channel), &channel.channel_id) + .run(run_id.as_deref()) + .units(channel.unit.as_ref().map(|u| u.name.as_str())) + .build(); let values = values .into_iter() @@ -562,10 +595,11 @@ impl DataService { bail!("unexpected missing channel from metadata"); }; - let column_name = ColumnName::builder(&channel.name, &channel.channel_id) - .run(run_id.as_deref()) - .units(channel.unit.as_ref().map(|u| u.name.as_str())) - .build(); + let column_name = + ColumnName::builder(column_label(&channel), &channel.channel_id) + .run(run_id.as_deref()) + .units(channel.unit.as_ref().map(|u| u.name.as_str())) + .build(); let values = values .into_iter() @@ -608,10 +642,11 @@ impl DataService { bail!("unexpected missing channel from metadata"); }; - let column_name = ColumnName::builder(&channel.name, &channel.channel_id) - .run(run_id.as_deref()) - .units(channel.unit.as_ref().map(|u| u.name.as_str())) - .build(); + let column_name = + ColumnName::builder(column_label(&channel), &channel.channel_id) + .run(run_id.as_deref()) + .units(channel.unit.as_ref().map(|u| u.name.as_str())) + .build(); let values = values .into_iter() @@ -654,10 +689,11 @@ impl DataService { bail!("unexpected missing channel from metadata"); }; - let column_name = ColumnName::builder(&channel.name, &channel.channel_id) - .run(run_id.as_deref()) - .units(channel.unit.as_ref().map(|u| u.name.as_str())) - .build(); + let column_name = + ColumnName::builder(column_label(&channel), &channel.channel_id) + .run(run_id.as_deref()) + .units(channel.unit.as_ref().map(|u| u.name.as_str())) + .build(); let values = values .into_iter() @@ -700,10 +736,11 @@ impl DataService { bail!("unexpected missing channel from metadata"); }; - let column_name = ColumnName::builder(&channel.name, &channel.channel_id) - .run(run_id.as_deref()) - .units(channel.unit.as_ref().map(|u| u.name.as_str())) - .build(); + let column_name = + ColumnName::builder(column_label(&channel), &channel.channel_id) + .run(run_id.as_deref()) + .units(channel.unit.as_ref().map(|u| u.name.as_str())) + .build(); let values = values .into_iter() @@ -746,10 +783,11 @@ impl DataService { bail!("unexpected missing channel from metadata"); }; - let column_name = ColumnName::builder(&channel.name, &channel.channel_id) - .run(run_id.as_deref()) - .units(channel.unit.as_ref().map(|u| u.name.as_str())) - .build(); + let column_name = + ColumnName::builder(column_label(&channel), &channel.channel_id) + .run(run_id.as_deref()) + .units(channel.unit.as_ref().map(|u| u.name.as_str())) + .build(); let values = values .into_iter() @@ -792,10 +830,11 @@ impl DataService { bail!("unexpected missing channel from metadata"); }; - let column_name = ColumnName::builder(&channel.name, &channel.channel_id) - .run(run_id.as_deref()) - .units(channel.unit.as_ref().map(|u| u.name.as_str())) - .build(); + let column_name = + ColumnName::builder(column_label(&channel), &channel.channel_id) + .run(run_id.as_deref()) + .units(channel.unit.as_ref().map(|u| u.name.as_str())) + .build(); let values = values .into_iter() @@ -838,10 +877,11 @@ impl DataService { bail!("unexpected missing channel from metadata"); }; - let column_name = ColumnName::builder(&channel.name, &channel.channel_id) - .run(run_id.as_deref()) - .units(channel.unit.as_ref().map(|u| u.name.as_str())) - .build(); + let column_name = + ColumnName::builder(column_label(&channel), &channel.channel_id) + .run(run_id.as_deref()) + .units(channel.unit.as_ref().map(|u| u.name.as_str())) + .build(); let values = values .into_iter() @@ -899,6 +939,9 @@ impl DataService { let (key, reported_as) = match input { ChannelInput::Raw(channel) => (&channel.channel_id, &channel.name), ChannelInput::Calculation { name, .. } => (name, name), + ChannelInput::SavedCalculation { channel_key, .. } => { + (channel_key, channel_key) + } }; (!produced.contains(key.as_str())).then(|| reported_as.clone()) }) diff --git a/rust/crates/sift_mcp/src/service/data/test.rs b/rust/crates/sift_mcp/src/service/data/test.rs index 21c2605ae..820d9c673 100644 --- a/rust/crates/sift_mcp/src/service/data/test.rs +++ b/rust/crates/sift_mcp/src/service/data/test.rs @@ -7,10 +7,11 @@ use parquet::arrow::{ArrowWriter, arrow_reader::ParquetRecordBatchReaderBuilder} use pbjson_types::{Any, Timestamp}; use prost::Message; use sift_rs::{ + calculated_channels::v1::{ExpressionChannelReference, ExpressionRequest}, channels::v3::Channel, data::v2::{ DoubleValue, DoubleValues, GetDataResponse, Metadata, - data_service_server::DataServiceServer, metadata, + data_service_server::DataServiceServer, metadata, query::Query as QueryKind, }, runs::v2::Run, }; @@ -55,6 +56,21 @@ fn named_raw_channel(channel_id: &str, name: &str) -> ChannelInput { })) } +fn saved_calculation(name: &str, expression: &str, input_channel_id: &str) -> ChannelInput { + ChannelInput::SavedCalculation { + channel_key: name.into(), + expression_request: Box::new(ExpressionRequest { + expression: expression.into(), + expression_channel_references: vec![ExpressionChannelReference { + channel_reference: "$1".into(), + channel_id: input_channel_id.into(), + calculated_channel_reference: None, + }], + ..Default::default() + }), + } +} + fn asset_range(start_nanos: i64, end_nanos: i64) -> TimeRange { TimeRange::Asset { start_time_unix_nanos: start_nanos, @@ -616,3 +632,60 @@ async fn sql_missing_input_file_errors() { "unexpected error: {msg}" ); } + +/// A saved calculated channel query sends the resolved expression and keys the +/// result on the channel name. +#[tokio::test] +async fn get_data_sends_saved_calculation_query() { + let mut mock = MockDataServiceImpl::new(); + mock.expect_get_data() + .times(1) + .withf(|req| { + let queries = &req.get_ref().queries; + queries.len() == 1 + && match queries[0].query.as_ref() { + Some(QueryKind::CalculatedChannel(query)) => { + query.channel_key == "thrust_margin" + && query.combine_run_data == Some(false) + && query.expression.as_ref().is_some_and(|expression| { + expression.expression == "$1 * 2" + && expression.expression_channel_references[0].channel_id + == "c1" + }) + } + _ => false, + } + }) + .returning(|_| { + Ok(Response::new(GetDataResponse { + // A calculated channel page carries the query's channel key in + // `channel_id` and no channel name. + data: vec![double_page( + "thrust_margin", + "", + vec![(1_000_000_000, 42.0)], + )], + next_page_token: String::new(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + let mut buffer = Vec::new(); + service + .get_data( + &[saved_calculation("thrust_margin", "$1 * 2", "c1")], + asset_range(0, 3_000_000_000), + 0, + &mut buffer, + ) + .await + .expect("get_data failed"); + + let batches = read_parquet(buffer); + let schema = batches[0].schema(); + assert!( + schema.field(1).name().starts_with("thrust_margin {"), + "a nameless calculated channel column must fall back to its key: {}", + schema.field(1).name(), + ); +} diff --git a/rust/crates/sift_mcp/src/tool/data/mod.rs b/rust/crates/sift_mcp/src/tool/data/mod.rs index 769f97c6b..12eaec8e8 100644 --- a/rust/crates/sift_mcp/src/tool/data/mod.rs +++ b/rust/crates/sift_mcp/src/tool/data/mod.rs @@ -17,7 +17,8 @@ use crate::{ error::{self, from_anyhow}, server::SiftMcpServer, service::{ - common, + calculated_channels::UnresolvedCalculation, + common::{self, cel_escape}, data::{ChannelInput, DataService, NoChannelData, TimeRange}, ingest::RunForm, }, @@ -96,21 +97,26 @@ impl SiftMcpServer { name = "get_data", description = " Retrieve time-series data for one or more channels of a single asset and write the result to a Parquet file. + Serves both raw channels and saved calculated channels. Output schema: - Column 0 is `timestamp_unix_nanos` (Int64, non-null) holding the merged ascending timestamps across all requested channels. - One column per matched channel, named ` {channel_id=\"...\", run=\"...\", units=\"...\"}`. - Cells are null where that channel has no sample at the row's timestamp. + Cells are null where that channel has no sample at the row's timestamp. A saved calculated channel + has no channel id, so its column carries the calculated channel's name in both places. - Enum and BitField channels carry their decode config in field metadata under the `enum_config` and `bit_field_elements` keys respectively. - A requested channel that produced no samples has NO column at all, not an all-null one. The tool result reports these so they never have to be inferred from the schema: - `unmatched_channel_names` lists requested names that matched no channel on the asset, and - `empty_channels` lists channels that matched but returned no samples in the window. Both keys are - ALWAYS present; two empty arrays mean every requested channel is in the file. When either is - non-empty, name those channels to the user before presenting any analysis — the file is a partial - answer. + `unmatched_channel_names` lists requested names served by neither a raw channel nor a saved + calculated channel, and `empty_channels` lists channels that matched but returned no samples in + the window. Both keys are ALWAYS present; two empty arrays mean every requested channel is in the + file. When either is non-empty, name those channels to the user before presenting any analysis — + the file is a partial answer. + - `unresolved_calculated_channels` (`[{ \"name\", \"reason\" }]`) is present when a requested name + reached calculated-channel resolution and could not be served. It carries the reason for every + name in `unmatched_channel_names`. Parameters: - `asset_name`: optional, exact asset name (not a pattern). Mutually exclusive with `asset_id`; @@ -124,8 +130,12 @@ impl SiftMcpServer { - `sample_ms`: decimation interval in milliseconds. Use `0` for raw samples; larger values reduce volume. - `channel_names`: optional array of exact channel names. Mutually exclusive with `channel_regex`; exactly one of the two MUST be set. Prefer this form when the set is known — it's more predictable. + A name with no raw channel on the asset is resolved as an active saved calculated channel and + evaluated for the requested asset and run. A raw channel wins a name it shares with a calculated + channel, so a calculated channel named after an existing raw channel is not served here. - `channel_regex`: optional RE2 pattern matched against the channel name. Mutually exclusive with - `channel_names`; exactly one of the two MUST be set. + `channel_names`; exactly one of the two MUST be set. Matches raw channels only; name saved + calculated channels explicitly in `channel_names`. - `output`: filesystem path for the Parquet file. The file is opened in truncate mode; existing contents are overwritten. @@ -136,6 +146,10 @@ impl SiftMcpServer { The error's `data` carries `empty_channels`, and `unmatched_channel_names` when the request also held a name that matched nothing — a failed call still reports both, so a retry does not repeat a typo the first call already detected. + - `RESOURCE_NOT_FOUND` naming every unresolved name when nothing requested can be served: no raw + channel matched and no named calculated channel exists or applies to the asset. A calculated + channel does not apply when the asset is outside its scope or lacks a channel its expression + references. Verify the name with `list_calculated_channels` filtered by `asset_id`. - `INVALID_PARAMS` if neither `asset_name` nor `asset_id` is set, or if both are set. - `INVALID_PARAMS` if `run_name` is absent and the full time range is not supplied, if neither `channel_names` nor `channel_regex` is set, if both are set, or if `channel_names` is empty. @@ -155,6 +169,9 @@ impl SiftMcpServer { - A successful call does NOT mean every requested channel is in the file. Check `unmatched_channel_names` and `empty_channels` before reporting the result or aggregating over it, and check the same two keys on the error's `data` when a call fails. + - A partial result is possible: when some calculated channels resolve and others do not, the file is + written from what resolved and `unresolved_calculated_channels` names the rest. Never report on the + data without telling the user what is missing. - After a successful call, if the user hasn't already indicated a next step, offer to run a SQL query against the resulting Parquet file using the `sql` tool. ", @@ -285,7 +302,24 @@ impl SiftMcpServer { .map_err(from_anyhow)?; let channels = page.items; - if channels.is_empty() { + // A raw channel wins a name it shares with a saved calculated channel; + // only names with no raw channel go on to calculated-channel resolution. + // A regex selection carries no per-name expectation, so it contributes none. + let unmatched_names = requested_names + .map(|names| { + let matched = channels + .iter() + .map(|c| c.name.as_str()) + .collect::>(); + + names + .into_iter() + .filter(|name| !matched.contains(name.as_str())) + .collect::>() + }) + .unwrap_or_default(); + + if channels.is_empty() && unmatched_names.is_empty() { return Err(ErrorData::resource_not_found( format!( "no channels matched the search criteria for asset '{}'", @@ -311,27 +345,52 @@ impl SiftMcpServer { )); } - // A regex selection has no per-name expectation to check against, so this - // only applies to an explicit `channel_names` list. - let unmatched_channel_names = requested_names - .map(|names| { - let matched = channels - .iter() - .map(|c| c.name.as_str()) - .collect::>(); - - names - .into_iter() - .filter(|name| !matched.contains(name.as_str())) - .collect::>() - }) - .unwrap_or_default(); - - let channel_inputs = channels + let mut channel_inputs = channels .into_iter() .map(|c| ChannelInput::Raw(Box::new(c))) .collect::>(); + let mut unresolved = Vec::new(); + if !unmatched_names.is_empty() { + let resolution = self + .calculated_channel_service + .resolve_calculated_channels( + unmatched_names, + asset.asset_id.clone(), + run.as_ref().map(|r| r.run_id.clone()), + ) + .await + .map_err(from_anyhow)?; + + channel_inputs.extend(resolution.resolved.into_iter().map(|calculation| { + ChannelInput::SavedCalculation { + channel_key: calculation.name, + expression_request: Box::new(calculation.expression_request), + } + })); + unresolved = resolution.unresolved; + } + + let unresolved_report = (!unresolved.is_empty()) + .then(|| unresolved_message(&unresolved, &asset.name, run_name.as_deref())); + + if channel_inputs.is_empty() { + return Err(ErrorData::resource_not_found( + unresolved_report.unwrap_or_else(|| { + format!("no channels matched the search criteria for asset '{}'", asset.name) + }), + None, + )); + } + + // Everything the request asked for and neither path could serve. A name + // that resolved to a calculated channel is in the file, so it is not + // unmatched; what is left is exactly the unresolved set. + let unmatched_channel_names = unresolved + .iter() + .map(|entry| entry.name.clone()) + .collect::>(); + let time_range = match run { Some(run) => TimeRange::Run { run: Box::new(run), @@ -376,12 +435,8 @@ impl SiftMcpServer { let output_str = output.to_string_lossy().into_owned(); let mut gaps = Vec::new(); - if !unmatched_channel_names.is_empty() { - gaps.push(format!( - "{} matched no channel on this asset: {}.", - unmatched_channel_names.len(), - common::name_list(&unmatched_channel_names), - )); + if let Some(report) = &unresolved_report { + gaps.push(format!("{report}.")); } if !data_output.empty_channels.is_empty() { gaps.push(format!( @@ -427,6 +482,21 @@ impl SiftMcpServer { string_array(data_output.empty_channels), ); + if !unresolved.is_empty() { + body.insert( + "unresolved_calculated_channels".to_string(), + serde_json::json!( + unresolved + .iter() + .map(|entry| serde_json::json!({ + "name": entry.name, + "reason": entry.reason, + })) + .collect::>() + ), + ); + } + let mut result = CallToolResult::structured(Value::Object(body)); result.content = vec![ContentBlock::text(next_step)]; Ok(result) @@ -639,6 +709,23 @@ impl SiftMcpServer { } } -fn cel_escape(s: &str) -> String { - s.replace('\\', "\\\\").replace('"', "\\\"") +/// One phrasing for calculated channels that yielded no data, shared by the +/// partial-result report and the all-unresolved error so the caller reads the +/// same wording either way. +fn unresolved_message( + unresolved: &[UnresolvedCalculation], + asset_name: &str, + run_name: Option<&str>, +) -> String { + let scope = match run_name { + Some(run) => format!("asset '{asset_name}' (run '{run}')"), + None => format!("asset '{asset_name}'"), + }; + let items = unresolved + .iter() + .map(|entry| format!("'{}' ({})", entry.name, entry.reason)) + .collect::>() + .join("; "); + + format!("calculated channel data was not resolved for {scope}: {items}") } diff --git a/rust/crates/sift_mcp/src/tool/data/test.rs b/rust/crates/sift_mcp/src/tool/data/test.rs index 27751a1de..bbcb246b9 100644 --- a/rust/crates/sift_mcp/src/tool/data/test.rs +++ b/rust/crates/sift_mcp/src/tool/data/test.rs @@ -1,20 +1,34 @@ +use std::path::PathBuf; + use bytes::Bytes; use pbjson_types::{Any, Timestamp}; use prost::Message; use rmcp::{handler::server::wrapper::Parameters, model::ErrorCode}; use sift_rs::{ assets::v1::{Asset, ListAssetsResponse, asset_service_server::AssetServiceServer}, + calculated_channels::{ + v1::{ExpressionChannelReference, ExpressionRequest}, + v2::{ + BatchResolveCalculatedChannelsResponse, CalculatedChannel, + ListCalculatedChannelsResponse, ResolveCalculatedChannelResponse, + ResolvedCalculatedChannel, UnresolvedCalculatedChannel, + calculated_channel_service_server::CalculatedChannelServiceServer, + }, + }, channels::v3::{Channel, ListChannelsResponse, channel_service_server::ChannelServiceServer}, data::v2::{ DoubleValue, DoubleValues, GetDataResponse, Metadata, - data_service_server::DataServiceServer, metadata, + data_service_server::DataServiceServer, metadata, query::Query as QueryKind, }, + runs::v2::{ListRunsResponse, Run, run_service_server::RunServiceServer}, }; use sift_test_util::{ grpc::memory_sift_channel, mock::{ - assets::v1::MockAssetServiceImpl, channels::v3::MockChannelServiceImpl, - data::v2::MockDataServiceImpl, + assets::v1::MockAssetServiceImpl, + calculated_channels::v2::MockCalculatedChannelServiceImpl, + channels::v3::MockChannelServiceImpl, data::v2::MockDataServiceImpl, + runs::v2::MockRunServiceImpl, }, }; use tempdir::TempDir; @@ -282,7 +296,144 @@ async fn get_data_accepts_a_full_page_with_nothing_left() { ); } -fn double_page(channel_id: &str, channel_name: &str, ts_nanos: i64, value: f64) -> Any { +/// Server wired to every service `get_data` touches, including the calculated +/// channel and data services the saved-calculation path needs. +async fn server_with_calculation_mocks( + assets: MockAssetServiceImpl, + channels: MockChannelServiceImpl, + runs: MockRunServiceImpl, + calculated: MockCalculatedChannelServiceImpl, + data: MockDataServiceImpl, +) -> (SiftMcpServer, JoinHandle<()>) { + let (client, server) = tokio::io::duplex(1024); + let channel = memory_sift_channel(client).await; + + let handle = tokio::spawn(async move { + Server::builder() + .add_service(AssetServiceServer::new(assets)) + .add_service(ChannelServiceServer::new(channels)) + .add_service(RunServiceServer::new(runs)) + .add_service(CalculatedChannelServiceServer::new(calculated)) + .add_service(DataServiceServer::new(data)) + .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server))) + .await + .unwrap(); + }); + + ( + SiftMcpServer::new( + channel, + String::from("https://api.test.local"), + false, + false, + ), + handle, + ) +} + +fn named_params( + names: &[&str], + run_name: Option<&str>, + output: PathBuf, +) -> Parameters { + Parameters(GetDataParams { + asset_name: Some("bench".into()), + asset_id: None, + run_name: run_name.map(String::from), + start_time_unix_nanos: Some(0), + end_time_unix_nanos: Some(3_000_000_000), + sample_ms: 0, + channel_names: Some(names.iter().map(|n| (*n).to_string()).collect()), + channel_regex: None, + output, + }) +} + +fn raw_channel_mock(names: &[&str]) -> MockChannelServiceImpl { + let channels = names + .iter() + .enumerate() + .map(|(i, name)| Channel { + channel_id: format!("ch-{i}"), + name: (*name).to_string(), + ..Default::default() + }) + .collect::>(); + + let mut mock = MockChannelServiceImpl::new(); + mock.expect_list_channels().returning(move |_| { + Ok(Response::new(ListChannelsResponse { + channels: channels.clone(), + next_page_token: String::new(), + })) + }); + mock +} + +fn one_run_mock() -> MockRunServiceImpl { + let mut runs = MockRunServiceImpl::new(); + runs.expect_list_runs().returning(|_| { + Ok(Response::new(ListRunsResponse { + runs: vec![Run { + run_id: "run-1".into(), + name: "hotfire-3".into(), + start_time: Some(Timestamp { + seconds: 0, + nanos: 0, + }), + stop_time: Some(Timestamp { + seconds: 3, + nanos: 0, + }), + ..Default::default() + }], + next_page_token: String::new(), + })) + }); + runs +} + +/// A calculated channel service that finds `name` and resolves it for the +/// asset. +fn resolving_calculation_mock(name: &'static str) -> MockCalculatedChannelServiceImpl { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(move |_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![CalculatedChannel { + calculated_channel_id: "cc1".into(), + name: name.into(), + ..Default::default() + }], + next_page_token: String::new(), + })) + }); + mock.expect_batch_resolve_calculated_channels() + .returning(|_| { + Ok(Response::new(BatchResolveCalculatedChannelsResponse { + responses: vec![ResolveCalculatedChannelResponse { + calculated_channel_id: None, + resolved: vec![ResolvedCalculatedChannel { + asset_name: "bench".into(), + asset_id: "asset-1".into(), + expression_request: Some(ExpressionRequest { + expression: "$1 * 2".into(), + expression_channel_references: vec![ExpressionChannelReference { + channel_reference: "$1".into(), + channel_id: "ch-9".into(), + calculated_channel_reference: None, + }], + ..Default::default() + }), + output_data_type: 0, + }], + unresolved: vec![], + }], + })) + }); + mock +} + +fn double_page(channel_id: &str, channel_name: &str, samples: Vec<(i64, f64)>) -> Any { let payload = DoubleValues { metadata: Some(Metadata { channel: Some(metadata::Channel { @@ -292,14 +443,17 @@ fn double_page(channel_id: &str, channel_name: &str, ts_nanos: i64, value: f64) }), ..Default::default() }), - values: vec![DoubleValue { - timestamp: Some(Timestamp { - seconds: ts_nanos / 1_000_000_000, - nanos: (ts_nanos % 1_000_000_000) as i32, - }), - value, - }], - extras: vec![], + values: samples + .into_iter() + .map(|(ts_nanos, value)| DoubleValue { + timestamp: Some(Timestamp { + seconds: ts_nanos / 1_000_000_000, + nanos: (ts_nanos % 1_000_000_000) as i32, + }), + value, + }) + .collect(), + ..Default::default() }; Any { @@ -308,6 +462,260 @@ fn double_page(channel_id: &str, channel_name: &str, ts_nanos: i64, value: f64) } } +/// A channel name with no raw channel on the asset is served as a saved +/// calculated channel: the query carries the resolved expression, not a +/// channel id. +#[tokio::test] +async fn get_data_queries_saved_calculated_channel() { + let mut data = MockDataServiceImpl::new(); + data.expect_get_data() + .times(1) + .withf(|req| { + let queries = &req.get_ref().queries; + queries.len() == 1 + && match queries[0].query.as_ref() { + Some(QueryKind::CalculatedChannel(query)) => { + let expression = query.expression.as_ref(); + query.channel_key == "thrust_margin" + && expression.is_some_and(|e| { + e.expression == "$1 * 2" + && e.expression_channel_references + .first() + .is_some_and(|r| r.channel_id == "ch-9") + }) + } + _ => false, + } + }) + .returning(|_| { + Ok(Response::new(GetDataResponse { + data: vec![double_page("", "thrust_margin", vec![(1_000_000_000, 4.0)])], + next_page_token: String::new(), + })) + }); + + let dir = TempDir::new("sift-mcp-cc-data").expect("temp dir"); + let output = dir.path().join("out.parquet"); + + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + raw_channel_mock(&[]), + MockRunServiceImpl::new(), + resolving_calculation_mock("thrust_margin"), + data, + ) + .await; + + let result = server + .get_data(named_params(&["thrust_margin"], None, output.clone())) + .await + .expect("get_data should serve a saved calculated channel"); + + let body = structured(result); + assert_eq!( + body.get("output").and_then(|v| v.as_str()), + Some(output.to_string_lossy().as_ref()), + ); + assert!( + body.get("unresolved_calculated_channels").is_none(), + "nothing should be reported as unresolved: {body}", + ); + assert!(output.exists(), "parquet file should be written"); +} + +/// One requested name resolves and one does not. The file is still written, but +/// the response must name the channel that did not resolve along with the asset +/// and run it was requested for. +#[tokio::test] +async fn get_data_reports_unresolved_calculated_channel() { + let mut calculated = MockCalculatedChannelServiceImpl::new(); + calculated.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![CalculatedChannel { + calculated_channel_id: "cc1".into(), + name: "chamber_delta".into(), + ..Default::default() + }], + next_page_token: String::new(), + })) + }); + calculated + .expect_batch_resolve_calculated_channels() + .returning(|_| { + Ok(Response::new(BatchResolveCalculatedChannelsResponse { + responses: vec![ResolveCalculatedChannelResponse { + calculated_channel_id: None, + resolved: vec![], + unresolved: vec![UnresolvedCalculatedChannel { + asset_name: "bench".into(), + error_message: "asset is missing channel chamber_pressure".into(), + }], + }], + })) + }); + + let mut data = MockDataServiceImpl::new(); + data.expect_get_data().times(1).returning(|_| { + Ok(Response::new(GetDataResponse { + data: vec![double_page( + "ch-0", + "temperature_c", + vec![(1_000_000_000, 20.0)], + )], + next_page_token: String::new(), + })) + }); + + let dir = TempDir::new("sift-mcp-cc-partial").expect("temp dir"); + let output = dir.path().join("out.parquet"); + + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + raw_channel_mock(&["temperature_c"]), + one_run_mock(), + calculated, + data, + ) + .await; + + let result = server + .get_data(named_params( + &["temperature_c", "chamber_delta"], + Some("hotfire-3"), + output.clone(), + )) + .await + .expect("a partial resolution should still return the resolved data"); + + let body = structured(result); + let reported = body + .get("unresolved_calculated_channels") + .expect("partial resolution must be reported") + .to_string(); + assert!( + reported.contains("chamber_delta") && reported.contains("chamber_pressure"), + "report should name the channel and the reason: {reported}", + ); + + let next_step = body + .get("next_step") + .and_then(|v| v.as_str()) + .expect("next_step") + .to_string(); + assert!( + next_step.contains("calculated channel data was not resolved for asset 'bench'") + && next_step.contains("hotfire-3") + && next_step.contains("chamber_delta"), + "next_step must name what did not resolve, and for which asset and run: {next_step}", + ); +} + +/// When nothing the caller named can be served, the call fails and the error +/// names every channel that did not resolve. +#[tokio::test] +async fn get_data_errors_when_no_channel_resolves() { + let mut calculated = MockCalculatedChannelServiceImpl::new(); + calculated.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![], + next_page_token: String::new(), + })) + }); + + let mut data = MockDataServiceImpl::new(); + data.expect_get_data().times(0); + + let dir = TempDir::new("sift-mcp-cc-none").expect("temp dir"); + let output = dir.path().join("out.parquet"); + + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + raw_channel_mock(&[]), + MockRunServiceImpl::new(), + calculated, + data, + ) + .await; + + let err = server + .get_data(named_params(&["chamber_delta"], None, output)) + .await + .expect_err("a request where nothing resolves must fail"); + + assert_eq!(err.code, ErrorCode::RESOURCE_NOT_FOUND); + assert!( + err.message + .contains("calculated channel data was not resolved for asset 'bench'") + && err.message.contains("chamber_delta"), + "error must name what did not resolve: {}", + err.message, + ); +} + +/// A name that exists as both a raw channel and a saved calculated channel is +/// served as the raw channel; the calculated channel service is never consulted. +#[tokio::test] +async fn get_data_prefers_raw_channel_over_calculated_channel() { + let mut calculated = MockCalculatedChannelServiceImpl::new(); + calculated.expect_list_calculated_channels().times(0); + calculated + .expect_batch_resolve_calculated_channels() + .times(0); + + let mut data = MockDataServiceImpl::new(); + data.expect_get_data() + .times(1) + .withf(|req| { + let queries = &req.get_ref().queries; + queries.len() == 1 + && matches!( + queries[0].query.as_ref(), + Some(QueryKind::Channel(query)) if query.channel_id == "ch-0" + ) + }) + .returning(|_| { + Ok(Response::new(GetDataResponse { + data: vec![double_page( + "ch-0", + "thrust_margin", + vec![(1_000_000_000, 1.0)], + )], + next_page_token: String::new(), + })) + }); + + let dir = TempDir::new("sift-mcp-cc-precedence").expect("temp dir"); + let output = dir.path().join("out.parquet"); + + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + raw_channel_mock(&["thrust_margin"]), + MockRunServiceImpl::new(), + calculated, + data, + ) + .await; + + server + .get_data(named_params(&["thrust_margin"], None, output)) + .await + .expect("the raw channel should be served without any calculated lookup"); +} + +/// A calculated channel service that knows no calculated channels, so a name +/// with no raw channel comes back unresolved rather than served. +fn no_calculation_mock() -> MockCalculatedChannelServiceImpl { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![], + next_page_token: String::new(), + })) + }); + mock.expect_batch_resolve_calculated_channels().times(0); + mock +} + /// `name in [...]` matches what it can and says nothing about the rest, so a /// misspelled channel used to come back as a narrower table reported as a /// complete fetch. @@ -337,15 +745,22 @@ async fn get_data_reports_channel_names_that_matched_nothing() { data.expect_get_data().returning(|_| { Ok(Response::new(GetDataResponse { data: vec![ - double_page("ch-1", "pressure", 1_000_000_000, 1.0), - double_page("ch-2", "temperature", 1_000_000_000, 2.0), + double_page("ch-1", "pressure", vec![(1_000_000_000, 1.0)]), + double_page("ch-2", "temperature", vec![(1_000_000_000, 2.0)]), ], next_page_token: String::new(), })) }); let dir = TempDir::new("sift-mcp-get-data").expect("failed to create temp dir"); - let (server, _h) = server_with_all_mocks(one_asset_mock(), channels, data).await; + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + channels, + MockRunServiceImpl::new(), + no_calculation_mock(), + data, + ) + .await; let resp = server .get_data(Parameters(GetDataParams { @@ -406,7 +821,7 @@ async fn get_data_reports_no_unmatched_names_for_a_regex_selection() { let mut data = MockDataServiceImpl::new(); data.expect_get_data().returning(|_| { Ok(Response::new(GetDataResponse { - data: vec![double_page("ch-1", "pressure", 1_000_000_000, 1.0)], + data: vec![double_page("ch-1", "pressure", vec![(1_000_000_000, 1.0)])], next_page_token: String::new(), })) }); @@ -467,7 +882,7 @@ async fn get_data_reports_matched_channels_that_returned_no_samples() { data.expect_get_data().returning(|_| { // Only one of the two matched channels has samples in this window. Ok(Response::new(GetDataResponse { - data: vec![double_page("ch-1", "pressure", 1_000_000_000, 1.0)], + data: vec![double_page("ch-1", "pressure", vec![(1_000_000_000, 1.0)])], next_page_token: String::new(), })) }); @@ -531,7 +946,14 @@ async fn no_data_error_reports_both_the_empty_and_the_unmatched_channels() { }); let dir = TempDir::new("sift-mcp-get-data").expect("failed to create temp dir"); - let (server, _h) = server_with_all_mocks(one_asset_mock(), channels, data).await; + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + channels, + MockRunServiceImpl::new(), + no_calculation_mock(), + data, + ) + .await; let err = server .get_data(Parameters(GetDataParams { From 24fe754d1a27328d9dcbd7181f2a06865afefd3a Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Tue, 18 Aug 2026 23:17:37 -0700 Subject: [PATCH 3/7] fix: verify resolve response ids, dedupe names, and keep the unresolved report on failure Batch resolve responses are mapped by position, so compare the echoed calculated channel id against the requested one and fail when they disagree; an absent echo still falls back to position. Drop repeated names before resolution: two queries sharing a channel key merge into one column and repeat timestamps. Carry the unresolved-channel report into a failed data query so an empty window does not read as "the asset has no data" when part of the request never resolved. --- .../src/service/calculated_channels/mod.rs | 14 +- .../src/service/calculated_channels/test.rs | 125 +++++++++++++++ rust/crates/sift_mcp/src/tool/data/mod.rs | 25 ++- rust/crates/sift_mcp/src/tool/data/test.rs | 145 ++++++++++++++++++ 4 files changed, 303 insertions(+), 6 deletions(-) diff --git a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs index 0b1cfa86a..19f8ec3b0 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs @@ -384,7 +384,19 @@ impl CalculatedChannelService { ); } - for ((name, _), response) in targets.into_iter().zip(responses) { + for ((name, id), response) in targets.into_iter().zip(responses) { + // Responses are matched to requests by position, so verify the id + // the API echoes back before trusting one: a shifted response would + // hand this name another channel's expression. The echo is + // optional, and an empty one leaves position as the only mapping. + let echoed = response.calculated_channel_id.unwrap_or_default(); + if !echoed.is_empty() && echoed != id { + bail!( + "resolve answered for calculated channel '{echoed}' where '{id}' was \ + requested; responses are out of order" + ); + } + // Only an entry for the requested asset is safe to query; an entry // for another asset would pull that asset's channels instead. let mut candidates = response.resolved; diff --git a/rust/crates/sift_mcp/src/service/calculated_channels/test.rs b/rust/crates/sift_mcp/src/service/calculated_channels/test.rs index cf9c9e558..98aad54db 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/test.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/test.rs @@ -1431,3 +1431,128 @@ async fn resolve_calculated_channels_errors_on_response_count_mismatch() { assert!(err.to_string().contains("resolve")); } + +/// The API echoes the resolved channel's id. Responses that do not line up with +/// their requests would assign one channel's expression to another channel's +/// name, so a mismatch must fail loudly rather than resolve to the wrong data. +#[tokio::test] +async fn resolve_calculated_channels_errors_when_responses_are_reordered() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![ + stored_channel("cc1", "thrust_margin"), + stored_channel("cc2", "chamber_delta"), + ], + next_page_token: String::new(), + })) + }); + mock.expect_batch_resolve_calculated_channels() + .returning(|_| { + // Requested cc1 then cc2; answered cc2 then cc1. + Ok(Response::new(BatchResolveCalculatedChannelsResponse { + responses: vec![ + ResolveCalculatedChannelResponse { + calculated_channel_id: Some("cc2".into()), + resolved: vec![ResolvedCalculatedChannel { + asset_name: "bench".into(), + asset_id: "asset-1".into(), + expression_request: Some(expression_request("$1", "ch-2")), + output_data_type: 0, + }], + unresolved: vec![], + }, + ResolveCalculatedChannelResponse { + calculated_channel_id: Some("cc1".into()), + resolved: vec![ResolvedCalculatedChannel { + asset_name: "bench".into(), + asset_id: "asset-1".into(), + expression_request: Some(expression_request("$1", "ch-1")), + output_data_type: 0, + }], + unresolved: vec![], + }, + ], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .resolve_calculated_channels( + vec!["thrust_margin".to_string(), "chamber_delta".to_string()], + "asset-1".to_string(), + None, + ) + .await + .expect_err("a reordered response must not be mapped by position"); + + let message = err.to_string(); + assert!( + message.contains("cc2") && message.contains("cc1"), + "error should name both the answered and the requested channel: {message}", + ); +} + +/// A response that echoes the requested id maps cleanly; an empty echo is +/// accepted, since the field is optional. +#[tokio::test] +async fn resolve_calculated_channels_accepts_matching_and_absent_echo() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![ + stored_channel("cc1", "thrust_margin"), + stored_channel("cc2", "chamber_delta"), + ], + next_page_token: String::new(), + })) + }); + mock.expect_batch_resolve_calculated_channels() + .returning(|_| { + Ok(Response::new(BatchResolveCalculatedChannelsResponse { + responses: vec![ + ResolveCalculatedChannelResponse { + calculated_channel_id: Some("cc1".into()), + resolved: vec![ResolvedCalculatedChannel { + asset_name: "bench".into(), + asset_id: "asset-1".into(), + expression_request: Some(expression_request("$1", "ch-1")), + output_data_type: 0, + }], + unresolved: vec![], + }, + ResolveCalculatedChannelResponse { + calculated_channel_id: None, + resolved: vec![ResolvedCalculatedChannel { + asset_name: "bench".into(), + asset_id: "asset-1".into(), + expression_request: Some(expression_request("$1", "ch-2")), + output_data_type: 0, + }], + unresolved: vec![], + }, + ], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let resolution = service + .resolve_calculated_channels( + vec!["thrust_margin".to_string(), "chamber_delta".to_string()], + "asset-1".to_string(), + None, + ) + .await + .expect("resolve_calculated_channels failed"); + + assert_eq!( + resolution + .resolved + .iter() + .map(|r| r.name.as_str()) + .collect::>(), + vec!["thrust_margin", "chamber_delta"], + ); +} diff --git a/rust/crates/sift_mcp/src/tool/data/mod.rs b/rust/crates/sift_mcp/src/tool/data/mod.rs index 12eaec8e8..fbfb1bfb5 100644 --- a/rust/crates/sift_mcp/src/tool/data/mod.rs +++ b/rust/crates/sift_mcp/src/tool/data/mod.rs @@ -305,6 +305,8 @@ impl SiftMcpServer { // A raw channel wins a name it shares with a saved calculated channel; // only names with no raw channel go on to calculated-channel resolution. // A regex selection carries no per-name expectation, so it contributes none. + // Repeats are dropped: two queries sharing a channel key merge into one + // column, which duplicates timestamps in the output. let unmatched_names = requested_names .map(|names| { let matched = channels @@ -312,13 +314,17 @@ impl SiftMcpServer { .map(|c| c.name.as_str()) .collect::>(); - names - .into_iter() - .filter(|name| !matched.contains(name.as_str())) - .collect::>() + let mut unmatched = Vec::::new(); + for name in names { + if !matched.contains(name.as_str()) && !unmatched.contains(&name) { + unmatched.push(name); + } + } + unmatched }) .unwrap_or_default(); + if channels.is_empty() && unmatched_names.is_empty() { return Err(ErrorData::resource_not_found( format!( @@ -426,7 +432,16 @@ impl SiftMcpServer { let empty_channels = err .downcast_ref::() .map(|no_data| no_data.empty_channels.clone()); - let mut error = from_anyhow(err.context("get data call failure - data_router")); + let err = err.context("get data call failure - data_router"); + // A failure here says nothing about channels that were never + // queried, so carry the report into it. Otherwise an empty window + // reads as "the asset has no data" when part of the request never + // resolved. + let err = match unresolved_report.as_ref() { + Some(report) => err.context(report.clone()), + None => err, + }; + let mut error = from_anyhow(err); error.data = gap_report(empty_channels, &unmatched_channel_names); return Err(error); } diff --git a/rust/crates/sift_mcp/src/tool/data/test.rs b/rust/crates/sift_mcp/src/tool/data/test.rs index bbcb246b9..99abff605 100644 --- a/rust/crates/sift_mcp/src/tool/data/test.rs +++ b/rust/crates/sift_mcp/src/tool/data/test.rs @@ -702,6 +702,151 @@ async fn get_data_prefers_raw_channel_over_calculated_channel() { .expect("the raw channel should be served without any calculated lookup"); } +/// A window with no samples still has to tell the caller which calculated +/// channels never resolved; otherwise the failure reads as "the asset has no +/// data" when part of the request was never queried at all. +#[tokio::test] +async fn get_data_reports_unresolved_calculated_channel_when_no_samples() { + let mut calculated = MockCalculatedChannelServiceImpl::new(); + calculated.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![CalculatedChannel { + calculated_channel_id: "cc1".into(), + name: "chamber_delta".into(), + ..Default::default() + }], + next_page_token: String::new(), + })) + }); + calculated + .expect_batch_resolve_calculated_channels() + .returning(|_| { + Ok(Response::new(BatchResolveCalculatedChannelsResponse { + responses: vec![ResolveCalculatedChannelResponse { + calculated_channel_id: Some("cc1".into()), + resolved: vec![], + unresolved: vec![UnresolvedCalculatedChannel { + asset_name: "bench".into(), + error_message: "asset is missing channel chamber_pressure".into(), + }], + }], + })) + }); + + let mut data = MockDataServiceImpl::new(); + data.expect_get_data().times(1).returning(|_| { + Ok(Response::new(GetDataResponse { + data: vec![], + next_page_token: String::new(), + })) + }); + + let dir = TempDir::new("sift-mcp-cc-nodata").expect("temp dir"); + let output = dir.path().join("out.parquet"); + + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + raw_channel_mock(&["temperature_c"]), + MockRunServiceImpl::new(), + calculated, + data, + ) + .await; + + let err = server + .get_data(named_params( + &["temperature_c", "chamber_delta"], + None, + output, + )) + .await + .expect_err("an empty window is still an error"); + + assert!( + err.message + .contains("calculated channel data was not resolved for asset 'bench'") + && err.message.contains("chamber_delta"), + "the empty-window error must still name what did not resolve: {}", + err.message, + ); +} + +/// The same calculated channel named twice must be queried once. Two identical +/// queries share a channel key, so their pages merge into one column and repeat +/// timestamps in the output file. +#[tokio::test] +async fn get_data_deduplicates_repeated_calculated_channel_name() { + let mut calculated = MockCalculatedChannelServiceImpl::new(); + calculated.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![CalculatedChannel { + calculated_channel_id: "cc1".into(), + name: "thrust_margin".into(), + ..Default::default() + }], + next_page_token: String::new(), + })) + }); + calculated + .expect_batch_resolve_calculated_channels() + .times(1) + .withf(|req| req.get_ref().requests.len() == 1) + .returning(|_| { + Ok(Response::new(BatchResolveCalculatedChannelsResponse { + responses: vec![ResolveCalculatedChannelResponse { + calculated_channel_id: Some("cc1".into()), + resolved: vec![ResolvedCalculatedChannel { + asset_name: "bench".into(), + asset_id: "asset-1".into(), + expression_request: Some(ExpressionRequest { + expression: "$1 * 2".into(), + expression_channel_references: vec![ExpressionChannelReference { + channel_reference: "$1".into(), + channel_id: "ch-9".into(), + calculated_channel_reference: None, + }], + ..Default::default() + }), + output_data_type: 0, + }], + unresolved: vec![], + }], + })) + }); + + let mut data = MockDataServiceImpl::new(); + data.expect_get_data() + .times(1) + .withf(|req| req.get_ref().queries.len() == 1) + .returning(|_| { + Ok(Response::new(GetDataResponse { + data: vec![double_page("thrust_margin", "", vec![(1_000_000_000, 4.0)])], + next_page_token: String::new(), + })) + }); + + let dir = TempDir::new("sift-mcp-cc-dedupe").expect("temp dir"); + let output = dir.path().join("out.parquet"); + + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + raw_channel_mock(&[]), + MockRunServiceImpl::new(), + calculated, + data, + ) + .await; + + server + .get_data(named_params( + &["thrust_margin", "thrust_margin"], + None, + output, + )) + .await + .expect("a repeated name must resolve and query once"); +} + /// A calculated channel service that knows no calculated channels, so a name /// with no raw channel comes back unresolved rather than served. fn no_calculation_mock() -> MockCalculatedChannelServiceImpl { From 525baab34c12636a28b68c1026e5c6045264f892 Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Wed, 26 Aug 2026 00:42:30 -0700 Subject: [PATCH 4/7] fix: adapt calculated channel name resolution to paged list results --- rust/crates/sift_mcp/src/service/calculated_channels/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs index 19f8ec3b0..3e126004e 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs @@ -324,7 +324,7 @@ impl CalculatedChannelService { // Keep the caller's order so the batch responses map back by index. let mut targets = Vec::new(); for name in names { - match stored.iter().find(|channel| channel.name == name) { + match stored.items.iter().find(|channel| channel.name == name) { Some(channel) => targets.push((name, channel.calculated_channel_id.clone())), None => unresolved.push(UnresolvedCalculation { name, From 9b21ce780102333c50934f812ca77dea61162c15 Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Wed, 26 Aug 2026 00:55:15 -0700 Subject: [PATCH 5/7] style: wrap get_data unresolved-report message construction --- rust/crates/sift_mcp/src/tool/data/mod.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/rust/crates/sift_mcp/src/tool/data/mod.rs b/rust/crates/sift_mcp/src/tool/data/mod.rs index fbfb1bfb5..f569c4f3f 100644 --- a/rust/crates/sift_mcp/src/tool/data/mod.rs +++ b/rust/crates/sift_mcp/src/tool/data/mod.rs @@ -383,7 +383,10 @@ impl SiftMcpServer { if channel_inputs.is_empty() { return Err(ErrorData::resource_not_found( unresolved_report.unwrap_or_else(|| { - format!("no channels matched the search criteria for asset '{}'", asset.name) + format!( + "no channels matched the search criteria for asset '{}'", + asset.name + ) }), None, )); From c2303e1218e3a1643c260ba21fde76593faa0bc4 Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Wed, 26 Aug 2026 01:12:01 -0700 Subject: [PATCH 6/7] fix: scope calculated channel name resolution to the asset and report ambiguity --- .../src/service/calculated_channels/mod.rs | 35 +++++++-- .../src/service/calculated_channels/test.rs | 71 ++++++++++++++++++- rust/crates/sift_mcp/src/tool/data/mod.rs | 1 - 3 files changed, 101 insertions(+), 6 deletions(-) diff --git a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs index 3e126004e..1bc36c67f 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs @@ -315,21 +315,48 @@ impl CalculatedChannelService { .join(", "); let stored = self .list_calculated_channels( - format!("is_archived == false && name in [{quoted}]"), + format!( + "is_archived == false && asset_id == \"{}\" && name in [{quoted}]", + common::cel_escape(&asset_id), + ), None, Some(common::PAGE_SIZE), ) .await?; + if stored.has_more { + bail!( + "saved calculated channel lookup matched more than {} channels and is incomplete; \ + narrow the channel names or split the request", + common::PAGE_SIZE, + ); + } + // Keep the caller's order so the batch responses map back by index. let mut targets = Vec::new(); for name in names { - match stored.items.iter().find(|channel| channel.name == name) { - Some(channel) => targets.push((name, channel.calculated_channel_id.clone())), - None => unresolved.push(UnresolvedCalculation { + let candidates = stored + .items + .iter() + .filter(|channel| channel.name == name) + .collect::>(); + match candidates.len() { + 0 => unresolved.push(UnresolvedCalculation { name, reason: UNKNOWN_NAME_REASON.to_string(), }), + 1 => targets.push((name, candidates[0].calculated_channel_id.clone())), + _ => unresolved.push(UnresolvedCalculation { + name, + reason: format!( + "ambiguous saved calculated channel name; matching calculated channel ids: {}", + candidates + .iter() + .map(|channel| channel.calculated_channel_id.as_str()) + .collect::>() + .join(", "), + ), + }), } } diff --git a/rust/crates/sift_mcp/src/service/calculated_channels/test.rs b/rust/crates/sift_mcp/src/service/calculated_channels/test.rs index 98aad54db..0fadad1c4 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/test.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/test.rs @@ -1078,13 +1078,14 @@ fn expression_request(expression: &str, channel_id: &str) -> ExpressionRequest { } #[tokio::test] -async fn resolve_calculated_channels_builds_lookup_and_resolve_requests() { +async fn resolve_calculated_channels_scopes_lookup_to_asset_and_builds_resolve_requests() { let mut mock = MockCalculatedChannelServiceImpl::new(); mock.expect_list_calculated_channels() .times(1) .withf(|req| { let filter = &req.get_ref().filter; filter.contains("is_archived == false") + && filter.contains(r#"asset_id == "asset-1""#) && filter.contains("name in [\"thrust_margin\", \"chamber_delta\"]") }) .returning(|_| { @@ -1364,6 +1365,74 @@ async fn resolve_calculated_channels_flags_unknown_name() { ); } +/// Duplicate saved-calculated-channel names are unsafe to resolve because the +/// caller could otherwise receive an arbitrary expression for the name. +#[tokio::test] +async fn resolve_calculated_channels_reports_ambiguous_name() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![ + stored_channel("cc-asset", "thrust_margin"), + stored_channel("cc-all-assets", "thrust_margin"), + ], + next_page_token: String::new(), + })) + }); + mock.expect_batch_resolve_calculated_channels().times(0); + + let (service, _h) = service_with_mock(mock).await; + + let resolution = service + .resolve_calculated_channels( + vec!["thrust_margin".to_string()], + "asset-1".to_string(), + None, + ) + .await + .expect("ambiguous names should be reported, not resolved"); + + assert!(resolution.resolved.is_empty()); + assert_eq!(resolution.unresolved.len(), 1); + assert_eq!(resolution.unresolved[0].name, "thrust_margin"); + assert!( + resolution.unresolved[0].reason.contains("ambiguous") + && resolution.unresolved[0].reason.contains("cc-asset") + && resolution.unresolved[0].reason.contains("cc-all-assets"), + "reason should name each colliding calculated channel: {}", + resolution.unresolved[0].reason, + ); +} + +/// A capped lookup cannot safely classify names missing from the returned page +/// as unknown, so resolution must fail before it builds any resolve request. +#[tokio::test] +async fn resolve_calculated_channels_rejects_truncated_lookup() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(|req| { + let page_size = req.get_ref().page_size as usize; + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: (0..page_size) + .map(|i| stored_channel(&format!("cc-{i}"), &format!("channel.{i}"))) + .collect(), + next_page_token: "more-calculated-channels".into(), + })) + }); + mock.expect_batch_resolve_calculated_channels().times(0); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .resolve_calculated_channels(vec!["channel.0".to_string()], "asset-1".to_string(), None) + .await + .expect_err("a truncated lookup must fail before resolving a partial result"); + + assert!( + err.to_string().contains(&PAGE_SIZE.to_string()) && err.to_string().contains("incomplete"), + "error should describe the capped lookup: {err}", + ); +} + #[tokio::test] async fn resolve_calculated_channels_propagates_grpc_error() { let mut mock = MockCalculatedChannelServiceImpl::new(); diff --git a/rust/crates/sift_mcp/src/tool/data/mod.rs b/rust/crates/sift_mcp/src/tool/data/mod.rs index f569c4f3f..e32bfa579 100644 --- a/rust/crates/sift_mcp/src/tool/data/mod.rs +++ b/rust/crates/sift_mcp/src/tool/data/mod.rs @@ -324,7 +324,6 @@ impl SiftMcpServer { }) .unwrap_or_default(); - if channels.is_empty() && unmatched_names.is_empty() { return Err(ErrorData::resource_not_found( format!( From 350af7a423e3db351ef49a1315a53df87b2ea97f Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang <159062208+evan-sift@users.noreply.github.com> Date: Thu, 27 Aug 2026 12:39:49 -0700 Subject: [PATCH 7/7] rust(feat): preview_rule MCP tool (#738) Co-authored-by: Liam Neville --- rust/crates/sift_cli/CHANGELOG.md | 26 + rust/crates/sift_cli/Cargo.toml | 2 +- .../sift_cli/assets/skills/sift/SKILL.md | 93 +- rust/crates/sift_mcp/Cargo.toml | 1 + rust/crates/sift_mcp/src/server/mod.rs | 15 +- .../sift_mcp/src/service/annotations/mod.rs | 267 +++++- .../sift_mcp/src/service/annotations/test.rs | 185 +++- rust/crates/sift_mcp/src/service/mod.rs | 2 + .../src/service/rule_evaluation/mod.rs | 91 ++ .../src/service/rule_evaluation/test.rs | 146 +++ .../src/service/user_defined_functions/mod.rs | 377 ++++++++ .../service/user_defined_functions/test.rs | 860 ++++++++++++++++++ .../sift_mcp/src/tool/annotations/mod.rs | 179 +++- .../sift_mcp/src/tool/annotations/test.rs | 480 +++++++++- rust/crates/sift_mcp/src/tool/mod.rs | 2 + .../sift_mcp/src/tool/rule_evaluation/mod.rs | 194 ++++ .../sift_mcp/src/tool/rule_evaluation/test.rs | 290 ++++++ .../src/tool/user_defined_functions/mod.rs | 719 +++++++++++++++ .../src/tool/user_defined_functions/test.rs | 799 ++++++++++++++++ rust/crates/sift_mcp/src/tool_events.json | 7 + rust/crates/sift_test_util/src/mock/mod.rs | 1 + .../src/mock/user_defined_functions/mod.rs | 1 + .../src/mock/user_defined_functions/v1.rs | 94 ++ 23 files changed, 4755 insertions(+), 76 deletions(-) create mode 100644 rust/crates/sift_mcp/src/service/rule_evaluation/mod.rs create mode 100644 rust/crates/sift_mcp/src/service/rule_evaluation/test.rs create mode 100644 rust/crates/sift_mcp/src/service/user_defined_functions/mod.rs create mode 100644 rust/crates/sift_mcp/src/service/user_defined_functions/test.rs create mode 100644 rust/crates/sift_mcp/src/tool/rule_evaluation/mod.rs create mode 100644 rust/crates/sift_mcp/src/tool/rule_evaluation/test.rs create mode 100644 rust/crates/sift_mcp/src/tool/user_defined_functions/mod.rs create mode 100644 rust/crates/sift_mcp/src/tool/user_defined_functions/test.rs create mode 100644 rust/crates/sift_test_util/src/mock/user_defined_functions/mod.rs create mode 100644 rust/crates/sift_test_util/src/mock/user_defined_functions/v1.rs diff --git a/rust/crates/sift_cli/CHANGELOG.md b/rust/crates/sift_cli/CHANGELOG.md index d8d0acbce..090d08c0f 100644 --- a/rust/crates/sift_cli/CHANGELOG.md +++ b/rust/crates/sift_cli/CHANGELOG.md @@ -7,6 +7,32 @@ This project adheres to [Semantic Versioning](http://semver.org/). ### What's New +## [v0.5.0] - August 26, 2026 + +### What's New + +- Added MCP tools for managing calculated channels: `list_calculated_channels`, + `list_calculated_channel_versions`, `create_calculated_channel`, + `update_calculated_channel`, `archive_calculated_channel`, and + `unarchive_calculated_channel`. +- `get_data` now serves saved calculated channels. A name in `channel_names` + with no raw-channel match resolves as an active saved calculated channel for + the asset and run; unresolvable names are reported explicitly. +- `get_data` now accepts `asset_id` as an alternative to `asset_name`; exactly + one must be set. +- Added `preview_rule`, which dry-runs a saved rule or an ad-hoc draft rule + config against a run without persisting anything. +- Added MCP tools for managing user-defined functions: + `list_user_defined_functions`, `list_user_defined_function_versions`, + `create_user_defined_function`, `update_user_defined_function`, + `archive_user_defined_function`, and `unarchive_user_defined_function`. +- `update_annotation` now requires `annotation_ids` instead of `annotation_id`, + a breaking change for existing callers; pass a one-element list for one + annotation. It updates 1 to 1000 annotations per call with per-ID failure + reporting, and its new `is_archived` parameter archives or unarchives + annotations. +- Refreshed the bundled Sift agent skill to cover the expanded MCP tool surface. + ## [v0.4.4] - August 24, 2026 ### What's New diff --git a/rust/crates/sift_cli/Cargo.toml b/rust/crates/sift_cli/Cargo.toml index 933525bd9..f1b096a04 100644 --- a/rust/crates/sift_cli/Cargo.toml +++ b/rust/crates/sift_cli/Cargo.toml @@ -3,7 +3,7 @@ test-reports = ["sift_mcp/test-reports"] [package] name = "sift_cli" -version = "0.4.4" +version = "0.5.0" authors.workspace = true edition.workspace = true categories.workspace = true diff --git a/rust/crates/sift_cli/assets/skills/sift/SKILL.md b/rust/crates/sift_cli/assets/skills/sift/SKILL.md index de9eab083..47f56b3cb 100644 --- a/rust/crates/sift_cli/assets/skills/sift/SKILL.md +++ b/rust/crates/sift_cli/assets/skills/sift/SKILL.md @@ -1,19 +1,21 @@ --- name: sift description: >- - Use when working with Sift: ingesting or importing time-series data, - querying assets/runs/channels/users, exporting data, decimating or running - SQL over data, opening a view in the Sift Explore web app, writing code that - integrates with Sift, installing, updating, or diagnosing the Sift agent - integration, or looking up how Sift works in its product and API - documentation. Covers the Sift MCP server (started by `sift-cli mcp`), the - `sift-cli` itself, the Sift REST API over cURL, the Sift Python library - (`sift_client`), and the Sift Rust streaming library (`sift_stream`). + Use for Sift tasks: ingesting or importing time-series data, querying + assets/runs/channels/users, managing calculated channels, rules, and + user-defined functions, exporting or decimating data, running SQL over data, + opening a view in Sift Explore, writing code that integrates with Sift, + installing, updating, or diagnosing the Sift agent integration, or looking + up how Sift works in its product and API documentation. Covers the Sift MCP + server (started by `sift-cli mcp`), `sift-cli`, the Sift REST API over cURL, + the Sift Python library (`sift_client`), and the Sift Rust streaming library + (`sift_stream`). Triggers include phrases like "import this file into Sift", "stream data to Sift", "list assets/runs/channels", "runs I created", "runs a teammate - created", "export a run", "query Sift", "graph", "plot", "visualize", "open - in Explore", "write code to integrate with Sift", "how does X work in Sift", - "what does this endpoint do", or "look up the Sift API reference". + created", "export a run", "query Sift", "graph", "plot", "visualize", "open in + Explore", "write code to integrate with Sift", "how does X work in Sift", + "what does this endpoint do", "list calculated channels", "preview a rule", + or "look up the Sift API reference". ---