From 026910b6f435c47b1549ee39e0879111213eab61 Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Tue, 18 Aug 2026 22:04:20 -0700 Subject: [PATCH 1/7] test: add calculated channels v2 gRPC service mock --- .../src/mock/calculated_channels/mod.rs | 1 + .../src/mock/calculated_channels/v2.rs | 93 +++++++++++++++++++ rust/crates/sift_test_util/src/mock/mod.rs | 1 + 3 files changed, 95 insertions(+) create mode 100644 rust/crates/sift_test_util/src/mock/calculated_channels/mod.rs create mode 100644 rust/crates/sift_test_util/src/mock/calculated_channels/v2.rs diff --git a/rust/crates/sift_test_util/src/mock/calculated_channels/mod.rs b/rust/crates/sift_test_util/src/mock/calculated_channels/mod.rs new file mode 100644 index 0000000000..7083bd82d0 --- /dev/null +++ b/rust/crates/sift_test_util/src/mock/calculated_channels/mod.rs @@ -0,0 +1 @@ +pub mod v2; diff --git a/rust/crates/sift_test_util/src/mock/calculated_channels/v2.rs b/rust/crates/sift_test_util/src/mock/calculated_channels/v2.rs new file mode 100644 index 0000000000..c573075038 --- /dev/null +++ b/rust/crates/sift_test_util/src/mock/calculated_channels/v2.rs @@ -0,0 +1,93 @@ +use async_trait::async_trait; +use mockall::mock; +use sift_rs::calculated_channels::v2::{ + BatchResolveCalculatedChannelsRequest, BatchResolveCalculatedChannelsResponse, + CreateCalculatedChannelRequest, CreateCalculatedChannelResponse, + GetCalculatedChannelDependentsRequest, GetCalculatedChannelDependentsResponse, + GetCalculatedChannelRequest, GetCalculatedChannelResponse, GetCalculatedChannelVersionsRequest, + GetCalculatedChannelVersionsResponse, ListCalculatedChannelVersionsRequest, + ListCalculatedChannelVersionsResponse, ListCalculatedChannelsRequest, + ListCalculatedChannelsResponse, ListResolvedCalculatedChannelsRequest, + ListResolvedCalculatedChannelsResponse, ResolveCalculatedChannelRequest, + ResolveCalculatedChannelResponse, UpdateCalculatedChannelRequest, + UpdateCalculatedChannelResponse, calculated_channel_service_server::CalculatedChannelService, +}; +use tonic::{Request, Response, Status}; + +mock! { + pub CalculatedChannelServiceImpl {} + + #[async_trait] + impl CalculatedChannelService for CalculatedChannelServiceImpl { + async fn get_calculated_channel( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn create_calculated_channel( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn list_calculated_channels( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn update_calculated_channel( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn list_calculated_channel_versions( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn resolve_calculated_channel( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn batch_resolve_calculated_channels( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn list_resolved_calculated_channels( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn get_calculated_channel_versions( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn get_calculated_channel_dependents( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + } +} diff --git a/rust/crates/sift_test_util/src/mock/mod.rs b/rust/crates/sift_test_util/src/mock/mod.rs index 114d1917ac..a7a7c812c8 100644 --- a/rust/crates/sift_test_util/src/mock/mod.rs +++ b/rust/crates/sift_test_util/src/mock/mod.rs @@ -1,5 +1,6 @@ pub mod annotations; pub mod assets; +pub mod calculated_channels; pub mod channels; pub mod data; pub mod docs; From 54f88744360810cbab6b0e7090990bc617601451 Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Tue, 18 Aug 2026 22:10:53 -0700 Subject: [PATCH 2/7] feat: add calculated channel service with list, versions, create, update, and archive --- .../src/service/calculated_channels/mod.rs | 528 +++++++++ .../src/service/calculated_channels/test.rs | 1052 +++++++++++++++++ rust/crates/sift_mcp/src/service/mod.rs | 1 + 3 files changed, 1581 insertions(+) create mode 100644 rust/crates/sift_mcp/src/service/calculated_channels/mod.rs create mode 100644 rust/crates/sift_mcp/src/service/calculated_channels/test.rs diff --git a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs new file mode 100644 index 0000000000..28431a358f --- /dev/null +++ b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs @@ -0,0 +1,528 @@ +use std::time::{SystemTime, UNIX_EPOCH}; + +use crate::policy::{RetryPolicy, with_retry}; +use crate::service::common; +use anyhow::{Context, Result, anyhow}; +use pbjson_types::{FieldMask, Timestamp}; +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, + }, + metadata::v1::MetadataValue, +}; + +#[cfg(test)] +mod test; + +/// A full calculated channel definition to create. The caller guarantees the +/// asset scope is unambiguous: either `all_assets` is set, or `asset_ids` / +/// `tag_ids` name a selection. +#[derive(Debug, Default)] +pub struct NewCalculatedChannel { + pub name: String, + pub description: Option, + pub user_notes: Option, + pub units: Option, + pub client_key: Option, + pub metadata: Vec, + pub expression: String, + pub expression_channel_references: Vec, + pub all_assets: bool, + pub asset_ids: Vec, + pub tag_ids: Vec, +} + +/// A partial set of changes to apply to an existing calculated channel. Every +/// field is optional; `None` means "leave unchanged". Only the fields set here +/// land in the update mask, so everything else on the channel is preserved. +#[derive(Debug, Default)] +pub struct CalculatedChannelUpdate { + pub name: Option, + pub description: Option, + pub units: Option, + pub metadata: Option>, + pub expression: Option, + pub expression_channel_references: Option>, + pub all_assets: Option, + pub asset_ids: Option>, + pub tag_ids: Option>, + pub user_notes: Option, +} + +impl CalculatedChannelUpdate { + /// Whether any updatable field is set. An update with nothing set would + /// produce an empty mask, which the API treats as a no-op. + pub fn is_empty(&self) -> bool { + self.name.is_none() + && self.description.is_none() + && self.units.is_none() + && self.metadata.is_none() + && self.expression.is_none() + && self.expression_channel_references.is_none() + && self.all_assets.is_none() + && self.asset_ids.is_none() + && self.tag_ids.is_none() + && self.user_notes.is_none() + } +} + +/// The result of a create or update write: the stored calculated channel plus +/// the assets the API reported as inapplicable (they lack a channel the +/// expression references), which the caller should surface to the user. +#[derive(Debug)] +pub struct CalculatedChannelWrite { + pub calculated_channel: CalculatedChannel, + pub inapplicable_assets: Vec, +} + +#[derive(Clone)] +pub struct CalculatedChannelService { + channel: SiftChannel, + policy: RetryPolicy, +} + +impl CalculatedChannelService { + pub fn new(channel: SiftChannel, policy: RetryPolicy) -> Self { + Self { channel, policy } + } + + pub async fn list_calculated_channels( + &self, + filter: String, + order_by: Option, + limit: Option, + ) -> Result> { + let (page_size, record_limit) = common::paging(limit); + + let mut page_token = String::new(); + let mut results = Vec::new(); + + let order_by = order_by.unwrap_or_default(); + + loop { + let grpc_channel = self.channel.clone(); + let filter = filter.clone(); + let order_by = order_by.clone(); + let token = page_token.clone(); + + let resp = with_retry(&self.policy, move || { + let grpc_channel = grpc_channel.clone(); + let filter = filter.clone(); + let order_by = order_by.clone(); + let token = token.clone(); + async move { + let mut client = CalculatedChannelServiceClient::new(grpc_channel); + client + .list_calculated_channels(ListCalculatedChannelsRequest { + page_size, + page_token: token, + filter, + organization_id: String::new(), + order_by, + }) + .await + .map(|resp| resp.into_inner()) + } + }) + .await + .context("failed to query calculated channels")?; + + let ListCalculatedChannelsResponse { + calculated_channels, + next_page_token, + } = resp; + if calculated_channels.is_empty() { + break; + } + results.extend(calculated_channels); + + if results.len() >= record_limit || next_page_token.is_empty() { + break; + } + page_token = next_page_token; + } + + results.truncate(record_limit); + + Ok(results) + } + + /// Lists the version history of a single calculated channel. Each version is + /// a full [`CalculatedChannel`] snapshot, not a reduced version record. + pub async fn list_calculated_channel_versions( + &self, + calculated_channel_id: String, + filter: String, + order_by: Option, + limit: Option, + ) -> Result> { + let (page_size, record_limit) = common::paging(limit); + + let mut page_token = String::new(); + let mut results = Vec::new(); + + let order_by = order_by.unwrap_or_default(); + + loop { + let grpc_channel = self.channel.clone(); + let calculated_channel_id = calculated_channel_id.clone(); + let filter = filter.clone(); + let order_by = order_by.clone(); + let token = page_token.clone(); + + let resp = with_retry(&self.policy, move || { + let grpc_channel = grpc_channel.clone(); + let calculated_channel_id = calculated_channel_id.clone(); + let filter = filter.clone(); + let order_by = order_by.clone(); + let token = token.clone(); + async move { + let mut client = CalculatedChannelServiceClient::new(grpc_channel); + client + .list_calculated_channel_versions(ListCalculatedChannelVersionsRequest { + calculated_channel_id, + client_key: String::new(), + page_size, + page_token: token, + filter, + organization_id: String::new(), + order_by, + }) + .await + .map(|resp| resp.into_inner()) + } + }) + .await + .context("failed to query calculated channel versions")?; + + let ListCalculatedChannelVersionsResponse { + calculated_channel_versions, + next_page_token, + } = resp; + if calculated_channel_versions.is_empty() { + break; + } + results.extend(calculated_channel_versions); + + if results.len() >= record_limit || next_page_token.is_empty() { + break; + } + page_token = next_page_token; + } + + results.truncate(record_limit); + + Ok(results) + } + + pub async fn create_calculated_channel( + &self, + new: NewCalculatedChannel, + ) -> Result { + let NewCalculatedChannel { + name, + description, + user_notes, + units, + client_key, + metadata, + expression, + expression_channel_references, + all_assets, + asset_ids, + tag_ids, + } = new; + + let configuration = CalculatedChannelConfiguration { + asset_configuration: Some(CalculatedChannelAssetConfiguration { + asset_scope: Some(asset_scope(all_assets, asset_ids, tag_ids)), + }), + query_configuration: Some(CalculatedChannelQueryConfiguration { + query: Some(Query::Sel(Sel { + expression, + expression_channel_references, + })), + }), + }; + + let request = CreateCalculatedChannelRequest { + name, + description: description.unwrap_or_default(), + user_notes: user_notes.unwrap_or_default(), + units, + client_key, + calculated_channel_configuration: Some(configuration), + metadata, + }; + + let grpc_channel = self.channel.clone(); + let resp = with_retry(&self.policy, move || { + let grpc_channel = grpc_channel.clone(); + let request = request.clone(); + async move { + let mut client = CalculatedChannelServiceClient::new(grpc_channel); + client + .create_calculated_channel(request) + .await + .map(|resp| resp.into_inner()) + } + }) + .await + .context("failed to create calculated channel")?; + + let calculated_channel = resp.calculated_channel.ok_or_else(|| { + anyhow!("create_calculated_channel response missing calculated channel") + })?; + + Ok(CalculatedChannelWrite { + calculated_channel, + inapplicable_assets: resp.inapplicable_assets, + }) + } + + /// Updates an existing calculated channel, creating a new version. The + /// current channel is fetched first so partially-specified nested + /// configuration (an expression without new references, asset ids without + /// tag ids) overlays the stored value instead of clearing it. Only the + /// fields the caller set are named in the update mask. + pub async fn update_calculated_channel( + &self, + calculated_channel_id: String, + changes: CalculatedChannelUpdate, + ) -> Result { + let mut channel = self + .get_calculated_channel(calculated_channel_id.clone()) + .await? + .ok_or_else(|| anyhow!("calculated channel '{calculated_channel_id}' not found"))?; + channel.calculated_channel_id = calculated_channel_id; + + let CalculatedChannelUpdate { + name, + description, + units, + metadata, + expression, + expression_channel_references, + all_assets, + asset_ids, + tag_ids, + user_notes, + } = changes; + + let mut paths = Vec::new(); + + if let Some(v) = name { + channel.name = v; + paths.push("name".to_string()); + } + if let Some(v) = description { + channel.description = v; + paths.push("description".to_string()); + } + if let Some(v) = units { + channel.units = Some(v); + paths.push("units".to_string()); + } + if let Some(v) = metadata { + channel.metadata = v; + paths.push("metadata".to_string()); + } + + let mut configuration = channel.calculated_channel_configuration.unwrap_or_default(); + + if expression.is_some() || expression_channel_references.is_some() { + let mut sel = current_sel(&configuration); + if let Some(v) = expression { + sel.expression = v; + } + if let Some(v) = expression_channel_references { + sel.expression_channel_references = v; + } + configuration.query_configuration = Some(CalculatedChannelQueryConfiguration { + query: Some(Query::Sel(sel)), + }); + paths.push("query_configuration".to_string()); + } + + if all_assets.is_some() || asset_ids.is_some() || tag_ids.is_some() { + let scope = match all_assets { + Some(true) => AssetScope::AllAssets(true), + _ => { + let mut selection = current_selection(&configuration); + if let Some(v) = asset_ids { + selection.asset_ids = v; + } + if let Some(v) = tag_ids { + selection.tag_ids = v; + } + AssetScope::Selection(selection) + } + }; + configuration.asset_configuration = Some(CalculatedChannelAssetConfiguration { + asset_scope: Some(scope), + }); + paths.push("asset_configuration".to_string()); + } + + channel.calculated_channel_configuration = Some(configuration); + + self.send_update(channel, paths, user_notes).await + } + + /// Archives a calculated channel by stamping `archived_date`. There is no + /// dedicated archive RPC; the API archives through the update mask. + pub async fn archive_calculated_channel( + &self, + calculated_channel_id: String, + ) -> Result { + let channel = CalculatedChannel { + calculated_channel_id, + archived_date: Some(now_timestamp()), + ..Default::default() + }; + + self.send_update(channel, vec!["archived_date".to_string()], None) + .await + .context("failed to archive calculated channel") + } + + /// Unarchives a calculated channel by clearing `archived_date` through the + /// update mask. A masked field left at its default is cleared. + pub async fn unarchive_calculated_channel( + &self, + calculated_channel_id: String, + ) -> Result { + let channel = CalculatedChannel { + calculated_channel_id, + archived_date: None, + ..Default::default() + }; + + self.send_update(channel, vec!["archived_date".to_string()], None) + .await + .context("failed to unarchive calculated channel") + } + + /// Retrieves the latest version of a calculated channel by id, or `None` if + /// it does not exist. Not exposed as a tool: `list_calculated_channels` + /// filtered by id covers the read case. + async fn get_calculated_channel( + &self, + calculated_channel_id: String, + ) -> Result> { + let grpc_channel = self.channel.clone(); + + let resp = with_retry(&self.policy, move || { + let grpc_channel = grpc_channel.clone(); + let calculated_channel_id = calculated_channel_id.clone(); + async move { + let mut client = CalculatedChannelServiceClient::new(grpc_channel); + client + .get_calculated_channel(GetCalculatedChannelRequest { + calculated_channel_id, + client_key: String::new(), + organization_id: String::new(), + calculated_channel_version_id: String::new(), + }) + .await + .map(|resp| resp.into_inner()) + } + }) + .await + .context("failed to fetch calculated channel")?; + + Ok(resp.calculated_channel) + } + + async fn send_update( + &self, + channel: CalculatedChannel, + paths: Vec, + user_notes: Option, + ) -> Result { + let grpc_channel = self.channel.clone(); + let resp = with_retry(&self.policy, move || { + let grpc_channel = grpc_channel.clone(); + let channel = channel.clone(); + let paths = paths.clone(); + let user_notes = user_notes.clone(); + async move { + let mut client = CalculatedChannelServiceClient::new(grpc_channel); + client + .update_calculated_channel(UpdateCalculatedChannelRequest { + calculated_channel: Some(channel), + update_mask: Some(FieldMask { paths }), + user_notes, + }) + .await + .map(|resp| resp.into_inner()) + } + }) + .await + .context("failed to update calculated channel")?; + + let calculated_channel = resp.calculated_channel.ok_or_else(|| { + anyhow!("update_calculated_channel response missing calculated channel") + })?; + + Ok(CalculatedChannelWrite { + calculated_channel, + inapplicable_assets: resp.inapplicable_assets, + }) + } +} + +/// Build the asset scope oneof from the flat inputs the tool validated. +fn asset_scope(all_assets: bool, asset_ids: Vec, tag_ids: Vec) -> AssetScope { + if all_assets { + AssetScope::AllAssets(true) + } else { + AssetScope::Selection(AssetSelection { asset_ids, tag_ids }) + } +} + +/// The stored SEL query, or an empty one when the channel carries no query +/// configuration yet. +fn current_sel(configuration: &CalculatedChannelConfiguration) -> Sel { + match configuration + .query_configuration + .as_ref() + .and_then(|q| q.query.as_ref()) + { + Some(Query::Sel(sel)) => sel.clone(), + None => Sel::default(), + } +} + +/// The stored asset selection, or an empty one when the channel is scoped to +/// all assets or carries no asset configuration yet. +fn current_selection(configuration: &CalculatedChannelConfiguration) -> AssetSelection { + match configuration + .asset_configuration + .as_ref() + .and_then(|a| a.asset_scope.as_ref()) + { + Some(AssetScope::Selection(selection)) => selection.clone(), + _ => AssetSelection::default(), + } +} + +fn now_timestamp() -> Timestamp { + let elapsed = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default(); + Timestamp { + seconds: elapsed.as_secs() as i64, + nanos: elapsed.subsec_nanos() as i32, + } +} diff --git a/rust/crates/sift_mcp/src/service/calculated_channels/test.rs b/rust/crates/sift_mcp/src/service/calculated_channels/test.rs new file mode 100644 index 0000000000..c37a38da0a --- /dev/null +++ b/rust/crates/sift_mcp/src/service/calculated_channels/test.rs @@ -0,0 +1,1052 @@ +use pbjson_types::Timestamp; +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, + }, + metadata::v1::{ + MetadataKey, MetadataKeyType, MetadataValue, metadata_value::Value as MetadataValueInner, + }, +}; +use sift_test_util::{ + grpc::memory_sift_channel, mock::calculated_channels::v2::MockCalculatedChannelServiceImpl, +}; +use tokio::task::JoinHandle; +use tonic::{Response, Status, transport::Server}; + +use super::{CalculatedChannelService, CalculatedChannelUpdate, NewCalculatedChannel}; +use crate::policy::RetryPolicy; +use crate::service::common::{DEFAULT_LIMIT, PAGE_SIZE}; + +async fn service_with_mock( + mock: MockCalculatedChannelServiceImpl, +) -> (CalculatedChannelService, 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(CalculatedChannelServiceServer::new(mock)) + .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server))) + .await + .unwrap(); + }); + + ( + CalculatedChannelService::new(channel, RetryPolicy::default()), + handle, + ) +} + +fn channel_ref(reference: &str, identifier: &str) -> CalculatedChannelAbstractChannelReference { + CalculatedChannelAbstractChannelReference { + channel_reference: reference.into(), + channel_identifier: identifier.into(), + calculated_channel_reference: None, + } +} + +fn string_metadata(name: &str, value: &str) -> MetadataValue { + MetadataValue { + key: Some(MetadataKey { + name: name.into(), + r#type: MetadataKeyType::String.into(), + ..Default::default() + }), + value: Some(MetadataValueInner::StringValue(value.into())), + ..Default::default() + } +} + +/// A stored calculated channel with both halves of its configuration populated, +/// used as the read side of the update read-modify-write path. +fn existing_channel(id: &str) -> CalculatedChannel { + CalculatedChannel { + calculated_channel_id: id.into(), + name: "thrust_margin".into(), + description: "margin".into(), + units: Some("N".into()), + calculated_channel_configuration: Some(CalculatedChannelConfiguration { + asset_configuration: Some(CalculatedChannelAssetConfiguration { + asset_scope: Some(AssetScope::Selection(AssetSelection { + asset_ids: vec!["asset-1".into()], + tag_ids: vec!["tag-1".into()], + })), + }), + query_configuration: Some(CalculatedChannelQueryConfiguration { + query: Some(Query::Sel(Sel { + expression: "$1 - $2".into(), + expression_channel_references: vec![ + channel_ref("$1", "thrust"), + channel_ref("$2", "thrust_limit"), + ], + })), + }), + }), + ..Default::default() + } +} + +fn new_channel() -> NewCalculatedChannel { + NewCalculatedChannel { + name: "thrust_margin".into(), + description: Some("headroom".into()), + user_notes: Some("initial".into()), + units: Some("N".into()), + client_key: Some("ck-1".into()), + metadata: vec![string_metadata("owner", "propulsion")], + expression: "$1 - $2".into(), + expression_channel_references: vec![ + channel_ref("$1", "thrust"), + channel_ref("$2", "thrust_limit"), + ], + all_assets: false, + asset_ids: vec!["asset-1".into()], + tag_ids: vec!["tag-1".into()], + } +} + +/// Extract the `Sel` query out of a calculated channel's configuration. +fn sel_of(channel: &CalculatedChannel) -> &Sel { + match channel + .calculated_channel_configuration + .as_ref() + .and_then(|c| c.query_configuration.as_ref()) + .and_then(|q| q.query.as_ref()) + { + Some(Query::Sel(sel)) => sel, + None => panic!("expected a sel query configuration"), + } +} + +/// Extract the asset scope out of a calculated channel's configuration. +fn asset_scope_of(channel: &CalculatedChannel) -> &AssetScope { + channel + .calculated_channel_configuration + .as_ref() + .and_then(|c| c.asset_configuration.as_ref()) + .and_then(|a| a.asset_scope.as_ref()) + .expect("expected an asset scope") +} + +#[tokio::test] +async fn list_calculated_channels_returns_single_page() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels() + .withf(|req| { + let req = req.get_ref(); + req.filter == "name == \"thrust_margin\"" && req.order_by == "name desc" + }) + .returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![ + CalculatedChannel { + calculated_channel_id: "cc1".into(), + name: "thrust_margin".into(), + ..Default::default() + }, + CalculatedChannel { + calculated_channel_id: "cc2".into(), + name: "thrust_margin".into(), + ..Default::default() + }, + ], + next_page_token: String::new(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let channels = service + .list_calculated_channels( + "name == \"thrust_margin\"".to_string(), + Some("name desc".to_string()), + None, + ) + .await + .expect("list_calculated_channels failed"); + + assert_eq!(channels.len(), 2); + assert_eq!(channels[0].calculated_channel_id, "cc1"); + assert_eq!(channels[1].calculated_channel_id, "cc2"); +} + +#[tokio::test] +async fn list_calculated_channels_paginates_until_token_empty() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(|req| { + let req = req.into_inner(); + assert_eq!(req.page_size, DEFAULT_LIMIT); + let (calculated_channels, next) = match req.page_token.as_str() { + "" => ( + vec![CalculatedChannel { + calculated_channel_id: "cc1".into(), + ..Default::default() + }], + "page-2".to_string(), + ), + "page-2" => ( + vec![CalculatedChannel { + calculated_channel_id: "cc2".into(), + ..Default::default() + }], + "page-3".to_string(), + ), + "page-3" => ( + vec![CalculatedChannel { + calculated_channel_id: "cc3".into(), + ..Default::default() + }], + String::new(), + ), + other => return Err(Status::invalid_argument(format!("bad token: {other}"))), + }; + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels, + next_page_token: next, + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let channels = service + .list_calculated_channels(String::new(), None, None) + .await + .expect("list_calculated_channels failed"); + + let ids: Vec<&str> = channels + .iter() + .map(|c| c.calculated_channel_id.as_str()) + .collect(); + assert_eq!(ids, vec!["cc1", "cc2", "cc3"]); +} + +#[tokio::test] +async fn list_calculated_channels_respects_limit() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels() + .times(1) + .returning(|req| { + let req = req.into_inner(); + assert_eq!(req.page_size, 2); + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![ + CalculatedChannel { + calculated_channel_id: "cc1".into(), + ..Default::default() + }, + CalculatedChannel { + calculated_channel_id: "cc2".into(), + ..Default::default() + }, + ], + next_page_token: "page-2".into(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let channels = service + .list_calculated_channels(String::new(), None, Some(2)) + .await + .expect("list_calculated_channels failed"); + + assert_eq!(channels.len(), 2); +} + +#[tokio::test] +async fn list_calculated_channels_clamps_limit_to_page_size() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels() + .times(1) + .returning(|req| { + let req = req.into_inner(); + assert_eq!(req.page_size, PAGE_SIZE); + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![CalculatedChannel { + calculated_channel_id: "cc1".into(), + ..Default::default() + }], + next_page_token: String::new(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let channels = service + .list_calculated_channels(String::new(), None, Some(5_000)) + .await + .expect("list_calculated_channels failed"); + + assert_eq!(channels.len(), 1); +} + +#[tokio::test] +async fn list_calculated_channels_breaks_on_empty_page() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels() + .times(1) + .returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![], + next_page_token: "ignored".into(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let channels = service + .list_calculated_channels(String::new(), None, None) + .await + .expect("list_calculated_channels failed"); + + assert!(channels.is_empty()); +} + +#[tokio::test] +async fn list_calculated_channels_propagates_grpc_error() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels() + .returning(|_| Err(Status::invalid_argument("bad filter"))); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .list_calculated_channels("nope".to_string(), None, None) + .await + .expect_err("expected error"); + + assert!( + err.to_string() + .contains("failed to query calculated channels") + ); +} + +#[tokio::test] +async fn list_calculated_channel_versions_builds_request() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channel_versions() + .times(1) + .withf(|req| { + let req = req.get_ref(); + req.calculated_channel_id == "cc1" + && req.filter == "version == 2" + && req.order_by == "version desc" + && req.page_size == 10 + && req.page_token.is_empty() + }) + .returning(|_| { + Ok(Response::new(ListCalculatedChannelVersionsResponse { + calculated_channel_versions: vec![CalculatedChannel { + calculated_channel_id: "cc1".into(), + version: 2, + ..Default::default() + }], + next_page_token: String::new(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let versions = service + .list_calculated_channel_versions( + "cc1".to_string(), + "version == 2".to_string(), + Some("version desc".to_string()), + Some(10), + ) + .await + .expect("list_calculated_channel_versions failed"); + + assert_eq!(versions.len(), 1); + assert_eq!(versions[0].version, 2); +} + +#[tokio::test] +async fn list_calculated_channel_versions_paginates_until_token_empty() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channel_versions() + .returning(|req| { + let req = req.into_inner(); + let (calculated_channel_versions, next) = match req.page_token.as_str() { + "" => ( + vec![CalculatedChannel { + version: 1, + ..Default::default() + }], + "page-2".to_string(), + ), + "page-2" => ( + vec![CalculatedChannel { + version: 2, + ..Default::default() + }], + String::new(), + ), + other => return Err(Status::invalid_argument(format!("bad token: {other}"))), + }; + Ok(Response::new(ListCalculatedChannelVersionsResponse { + calculated_channel_versions, + next_page_token: next, + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let versions = service + .list_calculated_channel_versions("cc1".to_string(), String::new(), None, None) + .await + .expect("list_calculated_channel_versions failed"); + + let numbers: Vec = versions.iter().map(|v| v.version).collect(); + assert_eq!(numbers, vec![1, 2]); +} + +#[tokio::test] +async fn list_calculated_channel_versions_propagates_grpc_error() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channel_versions() + .returning(|_| Err(Status::not_found("no such calculated channel"))); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .list_calculated_channel_versions("missing".to_string(), String::new(), None, None) + .await + .expect_err("expected error"); + + assert!( + err.to_string() + .contains("failed to query calculated channel versions") + ); +} + +#[tokio::test] +async fn create_calculated_channel_builds_request_with_asset_selection() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_create_calculated_channel() + .times(1) + .withf(|req| { + let req = req.get_ref(); + let config = req + .calculated_channel_configuration + .as_ref() + .expect("configuration present"); + let sel = match config + .query_configuration + .as_ref() + .and_then(|q| q.query.as_ref()) + { + Some(Query::Sel(sel)) => sel, + None => return false, + }; + let scope = config + .asset_configuration + .as_ref() + .and_then(|a| a.asset_scope.as_ref()) + .expect("asset scope present"); + + req.name == "thrust_margin" + && req.description == "headroom" + && req.user_notes == "initial" + && req.units.as_deref() == Some("N") + && req.client_key.as_deref() == Some("ck-1") + && req.metadata.len() == 1 + && sel.expression == "$1 - $2" + && sel.expression_channel_references.len() == 2 + && sel.expression_channel_references[0].channel_reference == "$1" + && sel.expression_channel_references[0].channel_identifier == "thrust" + && *scope + == AssetScope::Selection(AssetSelection { + asset_ids: vec!["asset-1".to_string()], + tag_ids: vec!["tag-1".to_string()], + }) + }) + .returning(|_| { + Ok(Response::new(CreateCalculatedChannelResponse { + calculated_channel: Some(CalculatedChannel { + calculated_channel_id: "cc1".into(), + name: "thrust_margin".into(), + ..Default::default() + }), + inapplicable_assets: vec![], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let written = service + .create_calculated_channel(new_channel()) + .await + .expect("create_calculated_channel failed"); + + assert_eq!(written.calculated_channel.calculated_channel_id, "cc1"); + assert!(written.inapplicable_assets.is_empty()); +} + +#[tokio::test] +async fn create_calculated_channel_builds_all_assets_scope() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_create_calculated_channel() + .times(1) + .withf(|req| { + let scope = req + .get_ref() + .calculated_channel_configuration + .as_ref() + .and_then(|c| c.asset_configuration.as_ref()) + .and_then(|a| a.asset_scope.as_ref()) + .expect("asset scope present"); + *scope == AssetScope::AllAssets(true) + }) + .returning(|_| { + Ok(Response::new(CreateCalculatedChannelResponse { + calculated_channel: Some(CalculatedChannel { + calculated_channel_id: "cc2".into(), + ..Default::default() + }), + inapplicable_assets: vec![], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let mut new = new_channel(); + new.all_assets = true; + new.asset_ids = vec![]; + new.tag_ids = vec![]; + + let written = service + .create_calculated_channel(new) + .await + .expect("create_calculated_channel failed"); + + assert_eq!(written.calculated_channel.calculated_channel_id, "cc2"); +} + +#[tokio::test] +async fn create_calculated_channel_surfaces_inapplicable_assets() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_create_calculated_channel().returning(|_| { + Ok(Response::new(CreateCalculatedChannelResponse { + calculated_channel: Some(CalculatedChannel { + calculated_channel_id: "cc3".into(), + ..Default::default() + }), + inapplicable_assets: vec![ + sift_rs::calculated_channels::v2::CalculatedChannelValidationResult { + asset_id: "asset-9".into(), + asset_name: Some("rover-09".into()), + tag_names: vec![], + missing_channels: vec!["thrust".into()], + }, + ], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let written = service + .create_calculated_channel(new_channel()) + .await + .expect("create_calculated_channel failed"); + + assert_eq!(written.inapplicable_assets.len(), 1); + assert_eq!(written.inapplicable_assets[0].asset_id, "asset-9"); +} + +#[tokio::test] +async fn create_calculated_channel_propagates_grpc_error() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_create_calculated_channel() + .returning(|_| Err(Status::invalid_argument("bad expression"))); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .create_calculated_channel(new_channel()) + .await + .expect_err("expected error"); + + assert!( + err.to_string() + .contains("failed to create calculated channel") + ); +} + +#[tokio::test] +async fn create_calculated_channel_errors_when_response_missing_channel() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_create_calculated_channel().returning(|_| { + Ok(Response::new(CreateCalculatedChannelResponse { + calculated_channel: None, + inapplicable_assets: vec![], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .create_calculated_channel(new_channel()) + .await + .expect_err("expected error"); + + assert!( + err.to_string() + .contains("create_calculated_channel response missing calculated channel") + ); +} + +#[tokio::test] +async fn update_calculated_channel_name_only_sets_name_mask() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_get_calculated_channel() + .times(1) + .withf(|req| req.get_ref().calculated_channel_id == "cc1") + .returning(|_| { + Ok(Response::new(GetCalculatedChannelResponse { + calculated_channel: Some(existing_channel("cc1")), + })) + }); + mock.expect_update_calculated_channel() + .times(1) + .withf(|req| { + let req = req.get_ref(); + let channel = req.calculated_channel.as_ref().expect("channel present"); + let mask = req.update_mask.as_ref().expect("mask present"); + + // Only `name` is in the mask, and the untouched configuration + // survives the read-modify-write round trip. + mask.paths == vec!["name".to_string()] + && channel.calculated_channel_id == "cc1" + && channel.name == "renamed" + && sel_of(channel).expression == "$1 - $2" + && *asset_scope_of(channel) + == AssetScope::Selection(AssetSelection { + asset_ids: vec!["asset-1".to_string()], + tag_ids: vec!["tag-1".to_string()], + }) + }) + .returning(|req| { + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: req.into_inner().calculated_channel, + inapplicable_assets: vec![], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let written = service + .update_calculated_channel( + "cc1".to_string(), + CalculatedChannelUpdate { + name: Some("renamed".into()), + ..Default::default() + }, + ) + .await + .expect("update_calculated_channel failed"); + + assert_eq!(written.calculated_channel.name, "renamed"); +} + +#[tokio::test] +async fn update_calculated_channel_sets_every_provided_field_in_mask() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_get_calculated_channel().returning(|_| { + Ok(Response::new(GetCalculatedChannelResponse { + calculated_channel: Some(existing_channel("cc1")), + })) + }); + mock.expect_update_calculated_channel() + .times(1) + .withf(|req| { + let req = req.get_ref(); + let mask = req.update_mask.as_ref().expect("mask present"); + let channel = req.calculated_channel.as_ref().expect("channel present"); + + mask.paths + == vec![ + "name".to_string(), + "description".to_string(), + "units".to_string(), + "metadata".to_string(), + "query_configuration".to_string(), + "asset_configuration".to_string(), + ] + && channel.description == "new description" + && channel.units.as_deref() == Some("kN") + && channel.metadata.len() == 1 + && req.user_notes.as_deref() == Some("bumped scaling") + }) + .returning(|req| { + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: req.into_inner().calculated_channel, + inapplicable_assets: vec![], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + service + .update_calculated_channel( + "cc1".to_string(), + CalculatedChannelUpdate { + name: Some("renamed".into()), + description: Some("new description".into()), + units: Some("kN".into()), + metadata: Some(vec![string_metadata("owner", "avionics")]), + expression: Some("$1 * 2".into()), + expression_channel_references: Some(vec![channel_ref("$1", "thrust")]), + all_assets: Some(true), + user_notes: Some("bumped scaling".into()), + ..Default::default() + }, + ) + .await + .expect("update_calculated_channel failed"); +} + +#[tokio::test] +async fn update_calculated_channel_expression_replaces_query_configuration() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_get_calculated_channel().returning(|_| { + Ok(Response::new(GetCalculatedChannelResponse { + calculated_channel: Some(existing_channel("cc1")), + })) + }); + mock.expect_update_calculated_channel() + .times(1) + .withf(|req| { + let req = req.get_ref(); + let channel = req.calculated_channel.as_ref().expect("channel present"); + let mask = req.update_mask.as_ref().expect("mask present"); + let sel = sel_of(channel); + + mask.paths == vec!["query_configuration".to_string()] + && sel.expression == "$1 * 2" + && sel.expression_channel_references.len() == 1 + && sel.expression_channel_references[0].channel_identifier == "thrust" + }) + .returning(|req| { + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: req.into_inner().calculated_channel, + inapplicable_assets: vec![], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + service + .update_calculated_channel( + "cc1".to_string(), + CalculatedChannelUpdate { + expression: Some("$1 * 2".into()), + expression_channel_references: Some(vec![channel_ref("$1", "thrust")]), + ..Default::default() + }, + ) + .await + .expect("update_calculated_channel failed"); +} + +#[tokio::test] +async fn update_calculated_channel_asset_ids_preserve_existing_tag_ids() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_get_calculated_channel().returning(|_| { + Ok(Response::new(GetCalculatedChannelResponse { + calculated_channel: Some(existing_channel("cc1")), + })) + }); + mock.expect_update_calculated_channel() + .times(1) + .withf(|req| { + let req = req.get_ref(); + let channel = req.calculated_channel.as_ref().expect("channel present"); + let mask = req.update_mask.as_ref().expect("mask present"); + + mask.paths == vec!["asset_configuration".to_string()] + && *asset_scope_of(channel) + == AssetScope::Selection(AssetSelection { + asset_ids: vec!["asset-2".to_string(), "asset-3".to_string()], + tag_ids: vec!["tag-1".to_string()], + }) + }) + .returning(|req| { + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: req.into_inner().calculated_channel, + inapplicable_assets: vec![], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + service + .update_calculated_channel( + "cc1".to_string(), + CalculatedChannelUpdate { + asset_ids: Some(vec!["asset-2".into(), "asset-3".into()]), + ..Default::default() + }, + ) + .await + .expect("update_calculated_channel failed"); +} + +#[tokio::test] +async fn update_calculated_channel_all_assets_replaces_selection_scope() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_get_calculated_channel().returning(|_| { + Ok(Response::new(GetCalculatedChannelResponse { + calculated_channel: Some(existing_channel("cc1")), + })) + }); + mock.expect_update_calculated_channel() + .times(1) + .withf(|req| { + let channel = req + .get_ref() + .calculated_channel + .as_ref() + .expect("channel present"); + *asset_scope_of(channel) == AssetScope::AllAssets(true) + }) + .returning(|req| { + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: req.into_inner().calculated_channel, + inapplicable_assets: vec![], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + service + .update_calculated_channel( + "cc1".to_string(), + CalculatedChannelUpdate { + all_assets: Some(true), + ..Default::default() + }, + ) + .await + .expect("update_calculated_channel failed"); +} + +#[tokio::test] +async fn update_calculated_channel_errors_when_channel_missing() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_get_calculated_channel().returning(|_| { + Ok(Response::new(GetCalculatedChannelResponse { + calculated_channel: None, + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .update_calculated_channel( + "missing".to_string(), + CalculatedChannelUpdate { + name: Some("renamed".into()), + ..Default::default() + }, + ) + .await + .expect_err("expected error"); + + assert!(err.to_string().contains("calculated channel")); + assert!(err.to_string().contains("missing")); +} + +#[tokio::test] +async fn update_calculated_channel_propagates_grpc_error() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_get_calculated_channel().returning(|_| { + Ok(Response::new(GetCalculatedChannelResponse { + calculated_channel: Some(existing_channel("cc1")), + })) + }); + mock.expect_update_calculated_channel() + .returning(|_| Err(Status::permission_denied("no write access"))); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .update_calculated_channel( + "cc1".to_string(), + CalculatedChannelUpdate { + name: Some("renamed".into()), + ..Default::default() + }, + ) + .await + .expect_err("expected error"); + + assert!( + err.to_string() + .contains("failed to update calculated channel") + ); +} + +#[tokio::test] +async fn archive_calculated_channel_masks_archived_date_only() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_update_calculated_channel() + .times(1) + .withf(|req| { + let req = req.get_ref(); + let channel = req.calculated_channel.as_ref().expect("channel present"); + let mask = req.update_mask.as_ref().expect("mask present"); + + mask.paths == vec!["archived_date".to_string()] + && channel.calculated_channel_id == "cc1" + && channel.archived_date.is_some() + }) + .returning(|req| { + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: req.into_inner().calculated_channel, + inapplicable_assets: vec![], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let written = service + .archive_calculated_channel("cc1".to_string()) + .await + .expect("archive_calculated_channel failed"); + + assert_eq!(written.calculated_channel.calculated_channel_id, "cc1"); +} + +#[tokio::test] +async fn archive_calculated_channel_does_not_read_first() { + // Archive is a masked field write; it must not need the read-modify-write + // round trip that `update_calculated_channel` performs. + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_get_calculated_channel().times(0); + mock.expect_update_calculated_channel().returning(|req| { + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: req.into_inner().calculated_channel, + inapplicable_assets: vec![], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + service + .archive_calculated_channel("cc1".to_string()) + .await + .expect("archive_calculated_channel failed"); +} + +#[tokio::test] +async fn unarchive_calculated_channel_clears_archived_date_with_mask_set() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_update_calculated_channel() + .times(1) + .withf(|req| { + let req = req.get_ref(); + let channel = req.calculated_channel.as_ref().expect("channel present"); + let mask = req.update_mask.as_ref().expect("mask present"); + + mask.paths == vec!["archived_date".to_string()] + && channel.calculated_channel_id == "cc1" + && channel.archived_date.is_none() + }) + .returning(|req| { + let mut channel = req + .into_inner() + .calculated_channel + .expect("channel present"); + channel.archived_date = None; + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: Some(channel), + inapplicable_assets: vec![], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let written = service + .unarchive_calculated_channel("cc1".to_string()) + .await + .expect("unarchive_calculated_channel failed"); + + assert!(written.calculated_channel.archived_date.is_none()); +} + +#[tokio::test] +async fn unarchive_calculated_channel_propagates_grpc_error() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_update_calculated_channel() + .returning(|_| Err(Status::not_found("no such calculated channel"))); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .unarchive_calculated_channel("missing".to_string()) + .await + .expect_err("expected error"); + + assert!( + err.to_string() + .contains("failed to unarchive calculated channel") + ); +} + +#[tokio::test] +async fn archive_calculated_channel_errors_when_response_missing_channel() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_update_calculated_channel().returning(|_| { + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: None, + inapplicable_assets: vec![], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .archive_calculated_channel("cc1".to_string()) + .await + .expect_err("expected error"); + + // `{:#}` renders the whole context chain; archive wraps the shared update path. + assert!( + format!("{err:#}") + .contains("update_calculated_channel response missing calculated channel") + ); +} + +#[tokio::test] +async fn archive_calculated_channel_stamps_a_current_timestamp() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_update_calculated_channel().returning(|req| { + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: req.into_inner().calculated_channel, + inapplicable_assets: vec![], + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let written = service + .archive_calculated_channel("cc1".to_string()) + .await + .expect("archive_calculated_channel failed"); + + let Timestamp { seconds, .. } = written + .calculated_channel + .archived_date + .expect("archived_date stamped"); + // Sanity floor: 2020-01-01T00:00:00Z. Guards against a zero-valued default + // being sent as the archive timestamp. + assert!(seconds > 1_577_836_800); +} diff --git a/rust/crates/sift_mcp/src/service/mod.rs b/rust/crates/sift_mcp/src/service/mod.rs index 2ab197f716..fe778f3e8a 100644 --- a/rust/crates/sift_mcp/src/service/mod.rs +++ b/rust/crates/sift_mcp/src/service/mod.rs @@ -1,5 +1,6 @@ pub mod annotations; pub mod assets; +pub mod calculated_channels; pub mod channels; pub mod data; pub mod docs; From 261793eeeebcb934337e2827e2698ef0f7444e91 Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Tue, 18 Aug 2026 22:19:53 -0700 Subject: [PATCH 3/7] feat: add calculated channel MCP tools for listing, versions, create, update, and archive --- rust/crates/sift_mcp/src/server/mod.rs | 10 +- .../src/tool/calculated_channels/mod.rs | 683 ++++++++++++++++++ .../src/tool/calculated_channels/test.rs | 550 ++++++++++++++ rust/crates/sift_mcp/src/tool/mod.rs | 1 + rust/crates/sift_mcp/src/tool_events.json | 6 + 5 files changed, 1248 insertions(+), 2 deletions(-) create mode 100644 rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs create mode 100644 rust/crates/sift_mcp/src/tool/calculated_channels/test.rs diff --git a/rust/crates/sift_mcp/src/server/mod.rs b/rust/crates/sift_mcp/src/server/mod.rs index a1c0a859c7..81f26bff7e 100644 --- a/rust/crates/sift_mcp/src/server/mod.rs +++ b/rust/crates/sift_mcp/src/server/mod.rs @@ -39,8 +39,9 @@ pub(crate) const BASE_INSTRUCTIONS: &str = concat!( #[cfg(feature = "test-reports")] use crate::service::test_reports::TestReportService; use crate::service::{ - annotations::AnnotationService, assets::AssetService, channels::ChannelService, - data::DataService, docs::DocsService, ingest::IngestService, ping::PingService, + annotations::AnnotationService, assets::AssetService, + calculated_channels::CalculatedChannelService, channels::ChannelService, data::DataService, + docs::DocsService, ingest::IngestService, ping::PingService, report_templates::ReportTemplateService, reports::ReportService, rules::RuleService, runs::RunService, url::UrlService, users::UserService, }; @@ -52,6 +53,7 @@ pub struct SiftMcpServer { pub annotation_service: AnnotationService, pub asset_service: AssetService, + pub calculated_channel_service: CalculatedChannelService, pub channel_service: ChannelService, pub data_service: DataService, pub url_service: UrlService, @@ -179,6 +181,7 @@ impl SiftMcpServer { let mut tool_router = Self::assets_router(); tool_router.merge(Self::runs_router()); tool_router.merge(Self::channels_router()); + tool_router.merge(Self::calculated_channels_router()); tool_router.merge(Self::reports_router()); tool_router.merge(Self::report_templates_router()); tool_router.merge(Self::data_router()); @@ -200,6 +203,8 @@ impl SiftMcpServer { let annotation_service = AnnotationService::new(channel.clone(), retry_policy.clone()); let asset_service = AssetService::new(channel.clone(), retry_policy.clone()); + let calculated_channel_service = + CalculatedChannelService::new(channel.clone(), retry_policy.clone()); let data_service = DataService::new(channel.clone(), retry_policy.clone()); let channel_service = ChannelService::new(channel.clone(), retry_policy.clone()); let url_service = UrlService::new(app_uri); @@ -218,6 +223,7 @@ impl SiftMcpServer { Self { annotation_service, asset_service, + calculated_channel_service, channel_service, data_service, url_service, diff --git a/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs b/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs new file mode 100644 index 0000000000..70cd594877 --- /dev/null +++ b/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs @@ -0,0 +1,683 @@ +use rmcp::{ + ErrorData, + handler::server::wrapper::Parameters, + model::{CallToolResult, ContentBlock}, + schemars::{self, JsonSchema}, + tool, tool_router, +}; +use serde::Deserialize; +use sift_rs::{ + calculated_channels::v2::CalculatedChannelAbstractChannelReference, metadata::v1::MetadataValue, +}; + +use crate::{ + error::{self, from_anyhow}, + server::SiftMcpServer, + service::calculated_channels::{ + CalculatedChannelUpdate, CalculatedChannelWrite, NewCalculatedChannel, + }, + tool::common::{ListParams, MetadataEntry}, +}; + +#[cfg(test)] +mod test; + +#[derive(Debug, Deserialize, JsonSchema)] +pub struct CalculatedChannelVersionListParams { + calculated_channel_id: String, + filter: Option, + order_by: Option, + limit: Option, +} + +#[derive(Debug, Deserialize, JsonSchema)] +pub struct CreateCalculatedChannelParams { + name: String, + expression: String, + expression_channel_references_json: String, + description: Option, + user_notes: Option, + units: Option, + client_key: Option, + all_assets: Option, + asset_ids: Option>, + tag_ids: Option>, + metadata: Option>, +} + +#[derive(Debug, Deserialize, JsonSchema)] +pub struct UpdateCalculatedChannelParams { + calculated_channel_id: String, + name: Option, + description: Option, + units: Option, + expression: Option, + expression_channel_references_json: Option, + all_assets: Option, + asset_ids: Option>, + tag_ids: Option>, + metadata: Option>, + user_notes: Option, +} + +#[derive(Debug, Deserialize, JsonSchema)] +pub struct CalculatedChannelArchiveParams { + calculated_channel_id: String, +} + +#[tool_router(router = calculated_channels_router, vis = "pub(crate)")] +impl SiftMcpServer { + #[tool( + name = "list_calculated_channels", + description = " + List calculated channels in Sift, optionally filtered by a CEL expression and ordered by one or + more fields. A calculated channel is a derived channel: a SEL expression over other channels, + scoped to a set of assets. + + Output: + - `{ \"calculated_channels\": [CalculatedChannel, ...] }`. Each item is the full Sift + `CalculatedChannel` shape including `calculated_channel_id`, `version_id`, `version`, `name`, + `description`, `units`, `client_key`, `calculated_channel_configuration` (the asset scope plus + the SEL expression and its channel references), `metadata`, `folder_ids`, `is_archived`, + `archived_date`, and timestamps. + - Fields at their proto3 default are OMITTED from the JSON: a missing `is_archived` key means + `false`, not \"unknown\". + + Parameters: + - `filter`: CEL expression. Pass an empty string to list everything. Filterable fields: + `calculated_channel_id`, `organization_id`, `client_key`, `name`, `description`, `asset_id`, + `asset_name`, `tag_id`, `tag_name`, `units`, `calculated_channel_version_id`, `created_date`, + `modified_date`, `created_by_user_id`, `modified_by_user_id`, `is_archived`, `archived_date`. + Folder membership is filterable via `folders` and `activeFolders` (folder-id lists; + `activeFolders` excludes archived folders): `\"\" in folders` returns calculated + channels in a folder, `size(activeFolders) == 0` returns uncategorized ones. + When filtering or searching, use `name.matches(\"(?i)thrust\")`, not `==`. Use `==` only for an + exact value from a prior result. `contains`/`startsWith`/`endsWith` are case-SENSITIVE: + `contains(\"Thrust\")` silently misses `thrust_margin`. Calculated channel names can embed `.`, + a regex wildcard, so match a full literal name with `contains`, not `matches`. + - `order_by`: optional comma-separated `FIELD_NAME[ desc]` list. Orderable fields: `created_date`, + `modified_date`, `name`, `description`, `units`, `archived_date`. Default sort is + `created_date desc` (newest first). Example: `\"created_date desc,modified_date\"`. + - `limit`: max items to return. Start at 50 and only raise it if the result is capped + and you still need more. Values are clamped to `1..=200`; omitting it defaults to 50. + + Errors: + - `INVALID_PARAMS` if `filter` is not a valid CEL expression or `order_by` references an unknown field. + - `INTERNAL_ERROR` for upstream gRPC failures. + + Guidance: + - Default add `is_archived == false` to the filter. Include archived calculated channels only when + the user explicitly asks for them. + - Scope with `asset_id == \"...\"` when the asset is known — it is the most selective field. + - Calculated channels do not carry data of their own; they are evaluated per asset. Use this tool to + discover what derived channels exist before reaching for `get_data`. + ", + annotations( + title = "calculated_channels/list_calculated_channels", + read_only_hint = true + ) + )] + pub async fn list_calculated_channels( + &self, + params: Parameters, + ) -> error::McpResult { + let Parameters(ListParams { + filter, + order_by, + limit, + }) = params; + + let out = self + .calculated_channel_service + .list_calculated_channels(filter, order_by, limit) + .await + .map(|channels| serde_json::json!({ "calculated_channels": channels })) + .map_err(from_anyhow)?; + + Ok(CallToolResult::structured(out)) + } + + #[tool( + name = "list_calculated_channel_versions", + description = " + List the version history of a single calculated channel. Every update creates a new version, so this + is how you see what changed and when. + + Output: + - `{ \"calculated_channel_versions\": [CalculatedChannel, ...], \"next_step\": string }`. Each item + is a full `CalculatedChannel` snapshot of that version, including `version`, `version_id`, + `change_message`, `user_notes`, `calculated_channel_configuration`, and + `modified_by_user_id` — not a reduced version record. + + Parameters: + - `calculated_channel_id`: required. The calculated channel whose versions to list. Resolve it with + `list_calculated_channels` first if you only have the name. + - `filter`: optional CEL expression. Filterable fields: `calculated_channel_id`, `organization_id`, + `client_key`, `name`, `description`, `asset_id`, `asset_name`, `tag_id`, `tag_name`, `version`, + `units`, `calculated_channel_version_id`, `created_date`, `modified_date`, `created_by_user_id`, + `modified_by_user_id`, `is_archived`, `archived_date`. Omit or pass an empty string to list all + versions. When filtering or searching text, use `name.matches(\"(?i)thrust\")`, not `==`. Use `==` + only for an exact value from a prior result. `contains`/`startsWith`/`endsWith` are + case-SENSITIVE: `contains(\"Thrust\")` silently misses `thrust_margin`. + - `order_by`: optional comma-separated `FIELD_NAME[ desc]` list. Orderable fields: `version`, + `created_date`, `modified_date`, `name`, `description`, `units`, `archived_date`. Default sort is + `created_date` ascending (oldest first) — note this differs from `list_calculated_channels`. + - `limit`: max items to return. Start at 50 and only raise it if the result is capped + and you still need more. Values are clamped to `1..=200`; omitting it defaults to 50. + + Errors: + - `INVALID_PARAMS` if `calculated_channel_id` is empty or `filter` is not a valid CEL expression. + - `RESOURCE_NOT_FOUND` if no calculated channel matches `calculated_channel_id`. + - `INTERNAL_ERROR` for upstream gRPC failures. + + Guidance: + - Use `order_by: \"version desc\"` with `limit: 1` to fetch just the most recent version. + - Read the prior version's `calculated_channel_configuration` before calling + `update_calculated_channel`, so you know what the expression and asset scope currently are. + ", + annotations( + title = "calculated_channels/list_calculated_channel_versions", + read_only_hint = true + ) + )] + pub async fn list_calculated_channel_versions( + &self, + params: Parameters, + ) -> error::McpResult { + let Parameters(CalculatedChannelVersionListParams { + calculated_channel_id, + filter, + order_by, + limit, + }) = params; + + require_id(&calculated_channel_id)?; + + let versions = self + .calculated_channel_service + .list_calculated_channel_versions( + calculated_channel_id, + filter.unwrap_or_default(), + order_by, + limit, + ) + .await + .map_err(from_anyhow)?; + + let next_step = format!( + "Listed {} calculated channel versions. Surface the version history to the user, \ + highlighting what changed between versions.", + versions.len(), + ); + + let mut result = CallToolResult::structured(serde_json::json!({ + "calculated_channel_versions": versions, + "next_step": next_step, + })); + result.content = vec![ContentBlock::text(next_step)]; + Ok(result) + } + + #[tool( + name = "create_calculated_channel", + description = " + Create a calculated channel: a derived channel defined by a SEL expression over other channels, + scoped to a set of assets. This is a WRITE. + + Output: + - `{ \"calculated_channel\": CalculatedChannel, \"inapplicable_assets\": [...], + \"next_step\": string }`. `inapplicable_assets` lists in-scope assets that do NOT have every + channel the expression references; each entry carries `asset_id`, `asset_name`, `tag_names`, and + `missing_channels`. A non-empty list means the channel was created but will not evaluate on those + assets. + + Parameters: + - `name`: required. The calculated channel's name. + - `expression`: required. A SEL expression whose channel operands are placeholders (`$1`, `$2`, …) + resolved by `expression_channel_references_json`. Example: `\"$1 - $2\"`. + - `expression_channel_references_json`: required. A JSON array string mapping each placeholder to a + channel. Each entry is + `{ \"channel_reference\": \"$1\", \"channel_identifier\": \"\" }`. To reference + another calculated channel instead of a raw channel, replace `channel_identifier` with + `\"calculated_channel_version_id\": \"\"`. Pass `[]` only for an expression with no + channel operands. This parameter is a JSON STRING, not an object. + - `description`: optional. Free-text description. + - `user_notes`: optional. Notes recorded against this version. + - `units`: optional. Units of the computed output. + - `client_key`: optional. A caller-defined identifier. Immutable after creation. + - `all_assets`: optional. `true` scopes the channel to every asset in the organization. + - `asset_ids` / `tag_ids`: optional. Scope the channel to specific assets and/or tagged assets. + - Exactly one scope form is allowed: either `all_assets: true`, or a non-empty `asset_ids` / + `tag_ids` selection. Setting both is rejected. + - `metadata`: optional. Array of `{ \"name\": \"\", \"value\": }` entries. + + Errors: + - `INVALID_PARAMS` if `name` or `expression` is empty, `expression_channel_references_json` is not a + valid reference array, no asset scope is given, or both scope forms are set. + - `INVALID_REQUEST` if the server was launched without `--allow-create`. + - `INTERNAL_ERROR` for upstream gRPC failures (e.g. a referenced channel does not exist). + + Guidance: + - Resolve the exact channel names with `list_channels` before writing the references — a typo makes + the channel inapplicable on every asset rather than failing outright. + - This creates a live resource. Confirm the name, the expression, and the asset scope with the user + before calling. + - Report any `inapplicable_assets` back to the user; that list is the API telling you the scope and + the expression disagree. + ", + annotations( + title = "calculated_channels/create_calculated_channel", + read_only_hint = false, + destructive_hint = false, + idempotent_hint = false, + ) + )] + pub async fn create_calculated_channel( + &self, + params: Parameters, + ) -> error::McpResult { + self.require_create()?; + + let Parameters(CreateCalculatedChannelParams { + name, + expression, + expression_channel_references_json, + description, + user_notes, + units, + client_key, + all_assets, + asset_ids, + tag_ids, + metadata, + }) = params; + + if name.is_empty() { + return Err(ErrorData::invalid_params("`name` must not be empty", None)); + } + if expression.is_empty() { + return Err(ErrorData::invalid_params( + "`expression` must not be empty", + None, + )); + } + + let expression_channel_references = + parse_channel_references(&expression_channel_references_json)?; + check_scope_exclusive(all_assets, &asset_ids, &tag_ids)?; + + let asset_ids = asset_ids.unwrap_or_default(); + let tag_ids = tag_ids.unwrap_or_default(); + let all_assets = all_assets.unwrap_or_default(); + if !all_assets && asset_ids.is_empty() && tag_ids.is_empty() { + return Err(ErrorData::invalid_params( + "set `all_assets` to true, or name at least one `asset_ids` or `tag_ids` entry", + None, + )); + } + + let written = self + .calculated_channel_service + .create_calculated_channel(NewCalculatedChannel { + name, + description, + user_notes, + units, + client_key, + metadata: metadata_values(metadata), + expression, + expression_channel_references, + all_assets, + asset_ids, + tag_ids, + }) + .await + .map_err(from_anyhow)?; + + let next_step = format!( + "Created calculated channel `{}` (`{}`).{} Tell the user the new id. If they haven't \ + indicated a next step, offer to confirm it with `list_calculated_channels` \ + (filter `calculated_channel_id == \"{}\"`).", + written.calculated_channel.name, + written.calculated_channel.calculated_channel_id, + inapplicable_clause(&written), + written.calculated_channel.calculated_channel_id, + ); + + Ok(write_result(written, next_step, None)) + } + + #[tool( + name = "update_calculated_channel", + description = " + Update an existing calculated channel, creating a new version. This is a WRITE. Only the fields you + set are changed; the tool reads the current channel, overlays your changes, and saves the result, so + unspecified fields are preserved. + + Output: + - `{ \"calculated_channel\": CalculatedChannel, \"inapplicable_assets\": [...], + \"next_step\": string }`. The returned channel is the new version's post-update state. + `inapplicable_assets` lists in-scope assets missing a channel the expression references. + + Parameters: + - `calculated_channel_id`: required. The calculated channel to update. + - `name`: optional. New name. + - `description`: optional. New description. + - `units`: optional. New units. + - `expression`: optional. New SEL expression. + - `expression_channel_references_json`: optional. New reference array, same shape as in + `create_calculated_channel`. `expression` and `expression_channel_references_json` must be + supplied together — a new expression with stale references silently misbinds its operands. + - `all_assets`: optional. `true` rescopes the channel to every asset. + - `asset_ids` / `tag_ids`: optional. REPLACE the corresponding list on the channel's selection + scope; the list you omit is preserved. Setting either alongside `all_assets: true` is rejected. + - `metadata`: optional. REPLACES the full metadata list. Pass `[]` to clear. + - `user_notes`: optional. Notes recorded against this new version. + - At least one field besides `calculated_channel_id` must be set. + + Errors: + - `INVALID_PARAMS` if no updatable field is set, `expression` and + `expression_channel_references_json` are not supplied together, the references JSON is invalid, or + both asset-scope forms are set. + - `INVALID_REQUEST` if the server was launched without `--allow-destructive`. + - `RESOURCE_NOT_FOUND` if no calculated channel matches `calculated_channel_id`. + - `INTERNAL_ERROR` for upstream gRPC failures. + + Guidance: + - Read the current definition with `list_calculated_channel_versions` before rewriting the + expression or the asset scope, and confirm the change with the user. + - This does not archive. Use `archive_calculated_channel` to retire a channel. + - Every update adds a version; the previous version stays readable through + `list_calculated_channel_versions`. + ", + annotations( + title = "calculated_channels/update_calculated_channel", + read_only_hint = false, + destructive_hint = true, + idempotent_hint = true, + ) + )] + pub async fn update_calculated_channel( + &self, + params: Parameters, + ) -> error::McpResult { + self.require_destructive()?; + + let Parameters(UpdateCalculatedChannelParams { + calculated_channel_id, + name, + description, + units, + expression, + expression_channel_references_json, + all_assets, + asset_ids, + tag_ids, + metadata, + user_notes, + }) = params; + + require_id(&calculated_channel_id)?; + + if expression.is_some() != expression_channel_references_json.is_some() { + return Err(ErrorData::invalid_params( + "`expression` and `expression_channel_references_json` must be provided together", + None, + )); + } + check_scope_exclusive(all_assets, &asset_ids, &tag_ids)?; + if all_assets == Some(false) && asset_ids.is_none() && tag_ids.is_none() { + return Err(ErrorData::invalid_params( + "`all_assets: false` needs `asset_ids` or `tag_ids` to define the new scope", + None, + )); + } + + let expression_channel_references = expression_channel_references_json + .as_deref() + .map(parse_channel_references) + .transpose()?; + + let changes = CalculatedChannelUpdate { + name, + description, + units, + metadata: metadata.map(|m| m.into_iter().map(MetadataValue::from).collect()), + expression, + expression_channel_references, + all_assets, + asset_ids, + tag_ids, + user_notes, + }; + + if changes.is_empty() { + return Err(ErrorData::invalid_params( + "at least one field besides `calculated_channel_id` must be set", + None, + )); + } + + let written = self + .calculated_channel_service + .update_calculated_channel(calculated_channel_id, changes) + .await + .map_err(from_anyhow)?; + + let next_step = format!( + "Updated calculated channel `{}` to version {}.{} Surface the new definition to the user \ + and confirm it matches their intent.", + written.calculated_channel.calculated_channel_id, + written.calculated_channel.version, + inapplicable_clause(&written), + ); + + Ok(write_result(written, next_step, None)) + } + + #[tool( + name = "archive_calculated_channel", + description = " + Archive a calculated channel so it stops being offered for plotting and querying. This is a WRITE. + Reversible with `unarchive_calculated_channel`. + + Output: + - `{ \"archived\": true, \"calculated_channel\": CalculatedChannel, \"next_step\": string }`. The + returned channel carries the `archived_date` the server recorded. + + Parameters: + - `calculated_channel_id`: required. The calculated channel to archive. + + Errors: + - `INVALID_PARAMS` if `calculated_channel_id` is empty. + - `INVALID_REQUEST` if the server was launched without `--allow-destructive`. + - `RESOURCE_NOT_FOUND` if no calculated channel matches `calculated_channel_id`. + - `INTERNAL_ERROR` for upstream gRPC failures. + + Guidance: + - Archiving does not delete the definition or its version history, and + `unarchive_calculated_channel` restores it. Confirm the target with the user before calling. + - Other calculated channels and rules may reference this one. Check with + `list_calculated_channels` before archiving something that looks like a shared building block. + ", + annotations( + title = "calculated_channels/archive_calculated_channel", + read_only_hint = false, + destructive_hint = true, + idempotent_hint = true, + ) + )] + pub async fn archive_calculated_channel( + &self, + params: Parameters, + ) -> error::McpResult { + self.require_destructive()?; + + let Parameters(CalculatedChannelArchiveParams { + calculated_channel_id, + }) = params; + + require_id(&calculated_channel_id)?; + + let written = self + .calculated_channel_service + .archive_calculated_channel(calculated_channel_id) + .await + .map_err(from_anyhow)?; + + let next_step = format!( + "Archived calculated channel `{}`. Tell the user it is archived and no longer offered \ + for plotting or querying, and that `unarchive_calculated_channel` restores it.", + written.calculated_channel.calculated_channel_id, + ); + + Ok(write_result(written, next_step, Some(("archived", true)))) + } + + #[tool( + name = "unarchive_calculated_channel", + description = " + Restore a previously archived calculated channel so it is offered again. This is a WRITE. + + Output: + - `{ \"unarchived\": true, \"calculated_channel\": CalculatedChannel, \"next_step\": string }`. The + returned channel has no `archived_date`. + + Parameters: + - `calculated_channel_id`: required. The calculated channel to restore. + + Errors: + - `INVALID_PARAMS` if `calculated_channel_id` is empty. + - `INVALID_REQUEST` if the server was launched without `--allow-destructive`. + - `RESOURCE_NOT_FOUND` if no calculated channel matches `calculated_channel_id`. + - `INTERNAL_ERROR` for upstream gRPC failures. + + Guidance: + - Find archived channels with `list_calculated_channels` filtered by `is_archived == true`. + - Confirm the target with the user before calling. + ", + annotations( + title = "calculated_channels/unarchive_calculated_channel", + read_only_hint = false, + destructive_hint = true, + idempotent_hint = true, + ) + )] + pub async fn unarchive_calculated_channel( + &self, + params: Parameters, + ) -> error::McpResult { + self.require_destructive()?; + + let Parameters(CalculatedChannelArchiveParams { + calculated_channel_id, + }) = params; + + require_id(&calculated_channel_id)?; + + let written = self + .calculated_channel_service + .unarchive_calculated_channel(calculated_channel_id) + .await + .map_err(from_anyhow)?; + + let next_step = format!( + "Unarchived calculated channel `{}`. Tell the user it is restored and available again.", + written.calculated_channel.calculated_channel_id, + ); + + Ok(write_result(written, next_step, Some(("unarchived", true)))) + } +} + +/// Reject an empty `calculated_channel_id` before any RPC. +fn require_id(calculated_channel_id: &str) -> Result<(), ErrorData> { + if calculated_channel_id.is_empty() { + return Err(ErrorData::invalid_params( + "`calculated_channel_id` must not be empty", + None, + )); + } + Ok(()) +} + +/// Deserialize the JSON-string channel reference array, mapping any parse error +/// to `INVALID_PARAMS` so the agent can correct it. Nested channel references +/// are the one irreducibly nested input on these tools. +fn parse_channel_references( + references_json: &str, +) -> Result, ErrorData> { + serde_json::from_str::>(references_json).map_err( + |e| { + ErrorData::invalid_params( + format!( + "`expression_channel_references_json` is not a valid channel reference array: {e}" + ), + None, + ) + }, + ) +} + +/// `all_assets: true` and an explicit asset/tag selection are two spellings of +/// the same oneof, so only one may be set. +fn check_scope_exclusive( + all_assets: Option, + asset_ids: &Option>, + tag_ids: &Option>, +) -> Result<(), ErrorData> { + if all_assets == Some(true) && (asset_ids.is_some() || tag_ids.is_some()) { + return Err(ErrorData::invalid_params( + "`all_assets` and `asset_ids`/`tag_ids` are mutually exclusive; set one scope form", + None, + )); + } + Ok(()) +} + +fn metadata_values(metadata: Option>) -> Vec { + metadata + .unwrap_or_default() + .into_iter() + .map(MetadataValue::from) + .collect() +} + +/// A trailing `next_step` clause naming the assets the API reported the channel +/// cannot evaluate on. Empty when every in-scope asset applies. +fn inapplicable_clause(written: &CalculatedChannelWrite) -> String { + if written.inapplicable_assets.is_empty() { + return String::new(); + } + format!( + " {} in-scope asset(s) are missing a referenced channel and will not evaluate this \ + calculated channel; see `inapplicable_assets`.", + written.inapplicable_assets.len() + ) +} + +/// Shape a write result: the stored channel, the inapplicable assets when the +/// API reported any, `next_step` on both the structured body and the content +/// block, plus an optional state flag (`archived` / `unarchived`). +fn write_result( + written: CalculatedChannelWrite, + next_step: String, + flag: Option<(&str, bool)>, +) -> CallToolResult { + let mut body = serde_json::json!({ + "calculated_channel": written.calculated_channel, + "next_step": next_step.clone(), + }); + match flag { + Some((key, value)) => { + body[key] = serde_json::json!(value); + } + None => { + body["inapplicable_assets"] = serde_json::json!(written.inapplicable_assets); + } + } + + let mut result = CallToolResult::structured(body); + result.content = vec![ContentBlock::text(next_step)]; + result +} diff --git a/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs b/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs new file mode 100644 index 0000000000..c2ec99a1df --- /dev/null +++ b/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs @@ -0,0 +1,550 @@ +use rmcp::{handler::server::wrapper::Parameters, model::ErrorCode}; +use sift_rs::calculated_channels::v2::{ + CalculatedChannel, CalculatedChannelValidationResult, CreateCalculatedChannelResponse, + GetCalculatedChannelResponse, ListCalculatedChannelVersionsResponse, + ListCalculatedChannelsResponse, UpdateCalculatedChannelResponse, + calculated_channel_service_server::CalculatedChannelServiceServer, +}; +use sift_test_util::{ + grpc::memory_sift_channel, mock::calculated_channels::v2::MockCalculatedChannelServiceImpl, +}; +use tokio::task::JoinHandle; +use tonic::{Response, Status, transport::Server}; + +use super::{ + CalculatedChannelArchiveParams, CalculatedChannelVersionListParams, + CreateCalculatedChannelParams, UpdateCalculatedChannelParams, +}; +use crate::{ + server::SiftMcpServer, + tool::common::test_support::{list_params, structured, structured_field}, +}; + +const REFERENCES_JSON: &str = r#"[ + { "channel_reference": "$1", "channel_identifier": "thrust" }, + { "channel_reference": "$2", "channel_identifier": "thrust_limit" } +]"#; + +async fn server_with_mock( + mock: MockCalculatedChannelServiceImpl, +) -> (SiftMcpServer, JoinHandle<()>) { + server_with_mock_and_flags(mock, true, true).await +} + +async fn server_with_mock_and_flags( + mock: MockCalculatedChannelServiceImpl, + allow_create: bool, + allow_destructive: bool, +) -> (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(CalculatedChannelServiceServer::new(mock)) + .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server))) + .await + .unwrap(); + }); + + ( + SiftMcpServer::new( + channel, + String::from("https://app.test.local"), + allow_create, + allow_destructive, + ), + handle, + ) +} + +fn create_params() -> CreateCalculatedChannelParams { + CreateCalculatedChannelParams { + name: "thrust_margin".into(), + expression: "$1 - $2".into(), + expression_channel_references_json: REFERENCES_JSON.into(), + description: None, + user_notes: None, + units: None, + client_key: None, + all_assets: Some(true), + asset_ids: None, + tag_ids: None, + metadata: None, + } +} + +fn update_params() -> UpdateCalculatedChannelParams { + UpdateCalculatedChannelParams { + calculated_channel_id: "cc1".into(), + name: None, + description: None, + units: None, + expression: None, + expression_channel_references_json: None, + all_assets: None, + asset_ids: None, + tag_ids: None, + metadata: None, + user_notes: None, + } +} + +fn stored_channel() -> CalculatedChannel { + CalculatedChannel { + calculated_channel_id: "cc1".into(), + name: "thrust_margin".into(), + version: 3, + ..Default::default() + } +} + +#[tokio::test] +async fn list_calculated_channels_returns_single_page() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels() + .withf(|req| req.get_ref().filter == "is_archived == false") + .returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![ + CalculatedChannel { + calculated_channel_id: "cc1".into(), + name: "thrust_margin".into(), + ..Default::default() + }, + CalculatedChannel { + calculated_channel_id: "cc2".into(), + name: "chamber_dp".into(), + ..Default::default() + }, + ], + next_page_token: String::new(), + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let resp = server + .list_calculated_channels(list_params("is_archived == false", None)) + .await + .expect("list_calculated_channels failed"); + + let channels = structured_field(resp, "calculated_channels"); + let channels = channels.as_array().expect("expected an array"); + assert_eq!(channels.len(), 2); + assert_eq!(channels[0]["calculatedChannelId"], "cc1"); + assert_eq!(channels[1]["name"], "chamber_dp"); +} + +#[tokio::test] +async fn list_calculated_channels_propagates_grpc_error() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels() + .returning(|_| Err(Status::invalid_argument("bad filter"))); + + let (server, _h) = server_with_mock(mock).await; + + let err = server + .list_calculated_channels(list_params("nope", None)) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); + assert!(err.message.contains("bad filter")); +} + +#[tokio::test] +async fn list_calculated_channel_versions_returns_versions() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channel_versions() + .withf(|req| req.get_ref().calculated_channel_id == "cc1") + .returning(|_| { + Ok(Response::new(ListCalculatedChannelVersionsResponse { + calculated_channel_versions: vec![CalculatedChannel { + calculated_channel_id: "cc1".into(), + version: 2, + ..Default::default() + }], + next_page_token: String::new(), + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let resp = server + .list_calculated_channel_versions(Parameters(CalculatedChannelVersionListParams { + calculated_channel_id: "cc1".into(), + filter: None, + order_by: None, + limit: None, + })) + .await + .expect("list_calculated_channel_versions failed"); + + let body = structured(resp); + let versions = body["calculated_channel_versions"] + .as_array() + .expect("expected an array"); + assert_eq!(versions.len(), 1); + assert_eq!(versions[0]["version"], 2); + assert!(body["next_step"].is_string()); +} + +#[tokio::test] +async fn list_calculated_channel_versions_rejects_empty_id() { + let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; + + let err = server + .list_calculated_channel_versions(Parameters(CalculatedChannelVersionListParams { + calculated_channel_id: String::new(), + filter: None, + order_by: None, + limit: None, + })) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +#[tokio::test] +async fn create_calculated_channel_blocked_without_allow_create() { + // No expectations on the mock: the gate must fire before any RPC. + let mock = MockCalculatedChannelServiceImpl::new(); + let (server, _h) = server_with_mock_and_flags(mock, false, false).await; + + let err = server + .create_calculated_channel(Parameters(create_params())) + .await + .expect_err("expected create gate to reject the call"); + + assert_eq!(err.code, ErrorCode::INVALID_REQUEST); + assert!(err.message.contains("--allow-create")); + assert!(err.message.contains("sift-cli agent update --allow-create")); +} + +#[tokio::test] +async fn create_calculated_channel_returns_structured_result() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_create_calculated_channel() + .times(1) + .returning(|_| { + Ok(Response::new(CreateCalculatedChannelResponse { + calculated_channel: Some(stored_channel()), + inapplicable_assets: vec![CalculatedChannelValidationResult { + asset_id: "asset-9".into(), + asset_name: Some("rover-09".into()), + tag_names: vec![], + missing_channels: vec!["thrust".into()], + }], + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let resp = server + .create_calculated_channel(Parameters(create_params())) + .await + .expect("create_calculated_channel failed"); + + let body = structured(resp); + assert_eq!(body["calculated_channel"]["calculatedChannelId"], "cc1"); + assert_eq!( + body["inapplicable_assets"] + .as_array() + .expect("expected an array") + .len(), + 1 + ); + assert!(body["next_step"].is_string()); +} + +#[tokio::test] +async fn create_calculated_channel_rejects_malformed_references_json() { + let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; + + let mut params = create_params(); + params.expression_channel_references_json = "{not json".into(); + + let err = server + .create_calculated_channel(Parameters(params)) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); + assert!(err.message.contains("expression_channel_references_json")); +} + +#[tokio::test] +async fn create_calculated_channel_rejects_all_assets_with_asset_ids() { + let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; + + let mut params = create_params(); + params.all_assets = Some(true); + params.asset_ids = Some(vec!["asset-1".into()]); + + let err = server + .create_calculated_channel(Parameters(params)) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +#[tokio::test] +async fn create_calculated_channel_requires_an_asset_scope() { + let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; + + let mut params = create_params(); + params.all_assets = None; + + let err = server + .create_calculated_channel(Parameters(params)) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +#[tokio::test] +async fn create_calculated_channel_rejects_empty_name() { + let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; + + let mut params = create_params(); + params.name = String::new(); + + let err = server + .create_calculated_channel(Parameters(params)) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +#[tokio::test] +async fn create_calculated_channel_rejects_empty_expression() { + let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; + + let mut params = create_params(); + params.expression = String::new(); + + let err = server + .create_calculated_channel(Parameters(params)) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +#[tokio::test] +async fn update_calculated_channel_blocked_without_allow_destructive() { + let mock = MockCalculatedChannelServiceImpl::new(); + let (server, _h) = server_with_mock_and_flags(mock, false, false).await; + + let mut params = update_params(); + params.name = Some("renamed".into()); + + let err = server + .update_calculated_channel(Parameters(params)) + .await + .expect_err("expected destructive gate to reject the call"); + + assert_eq!(err.code, ErrorCode::INVALID_REQUEST); + assert!(err.message.contains("--allow-destructive")); +} + +#[tokio::test] +async fn update_calculated_channel_rejects_empty_update() { + let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; + + let err = server + .update_calculated_channel(Parameters(update_params())) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +#[tokio::test] +async fn update_calculated_channel_rejects_expression_without_references() { + let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; + + let mut params = update_params(); + params.expression = Some("$1 * 2".into()); + + let err = server + .update_calculated_channel(Parameters(params)) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +#[tokio::test] +async fn update_calculated_channel_rejects_malformed_references_json() { + let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; + + let mut params = update_params(); + params.expression = Some("$1 * 2".into()); + params.expression_channel_references_json = Some("{not json".into()); + + let err = server + .update_calculated_channel(Parameters(params)) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +#[tokio::test] +async fn update_calculated_channel_rejects_all_assets_with_asset_ids() { + let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; + + let mut params = update_params(); + params.all_assets = Some(true); + params.asset_ids = Some(vec!["asset-1".into()]); + + let err = server + .update_calculated_channel(Parameters(params)) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +#[tokio::test] +async fn update_calculated_channel_returns_structured_result() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_get_calculated_channel().returning(|_| { + Ok(Response::new(GetCalculatedChannelResponse { + calculated_channel: Some(stored_channel()), + })) + }); + mock.expect_update_calculated_channel() + .times(1) + .returning(|req| { + let mut channel = req + .into_inner() + .calculated_channel + .expect("channel present"); + channel.version = 4; + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: Some(channel), + inapplicable_assets: vec![], + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let mut params = update_params(); + params.name = Some("renamed".into()); + + let resp = server + .update_calculated_channel(Parameters(params)) + .await + .expect("update_calculated_channel failed"); + + let body = structured(resp); + assert_eq!(body["calculated_channel"]["name"], "renamed"); + assert_eq!(body["calculated_channel"]["version"], 4); + assert!(body["next_step"].is_string()); +} + +#[tokio::test] +async fn archive_calculated_channel_blocked_without_allow_destructive() { + let mock = MockCalculatedChannelServiceImpl::new(); + let (server, _h) = server_with_mock_and_flags(mock, false, false).await; + + let err = server + .archive_calculated_channel(Parameters(CalculatedChannelArchiveParams { + calculated_channel_id: "cc1".into(), + })) + .await + .expect_err("expected destructive gate to reject the call"); + + assert_eq!(err.code, ErrorCode::INVALID_REQUEST); + assert!(err.message.contains("--allow-destructive")); +} + +#[tokio::test] +async fn unarchive_calculated_channel_blocked_without_allow_destructive() { + let mock = MockCalculatedChannelServiceImpl::new(); + let (server, _h) = server_with_mock_and_flags(mock, false, false).await; + + let err = server + .unarchive_calculated_channel(Parameters(CalculatedChannelArchiveParams { + calculated_channel_id: "cc1".into(), + })) + .await + .expect_err("expected destructive gate to reject the call"); + + assert_eq!(err.code, ErrorCode::INVALID_REQUEST); + assert!(err.message.contains("--allow-destructive")); +} + +#[tokio::test] +async fn archive_calculated_channel_returns_structured_result() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_update_calculated_channel() + .times(1) + .returning(|req| { + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: req.into_inner().calculated_channel, + inapplicable_assets: vec![], + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let resp = server + .archive_calculated_channel(Parameters(CalculatedChannelArchiveParams { + calculated_channel_id: "cc1".into(), + })) + .await + .expect("archive_calculated_channel failed"); + + let body = structured(resp); + assert_eq!(body["archived"], true); + assert_eq!(body["calculated_channel"]["calculatedChannelId"], "cc1"); + assert!(body["next_step"].is_string()); +} + +#[tokio::test] +async fn unarchive_calculated_channel_returns_structured_result() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_update_calculated_channel() + .times(1) + .returning(|req| { + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: req.into_inner().calculated_channel, + inapplicable_assets: vec![], + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let resp = server + .unarchive_calculated_channel(Parameters(CalculatedChannelArchiveParams { + calculated_channel_id: "cc1".into(), + })) + .await + .expect("unarchive_calculated_channel failed"); + + let body = structured(resp); + assert_eq!(body["unarchived"], true); + assert!(body["next_step"].is_string()); +} + +#[tokio::test] +async fn archive_calculated_channel_rejects_empty_id() { + let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; + + let err = server + .archive_calculated_channel(Parameters(CalculatedChannelArchiveParams { + calculated_channel_id: String::new(), + })) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} diff --git a/rust/crates/sift_mcp/src/tool/mod.rs b/rust/crates/sift_mcp/src/tool/mod.rs index c5f5d0c9e2..331d92b16f 100644 --- a/rust/crates/sift_mcp/src/tool/mod.rs +++ b/rust/crates/sift_mcp/src/tool/mod.rs @@ -1,5 +1,6 @@ pub mod annotations; pub mod assets; +pub mod calculated_channels; pub mod channels; pub mod common; pub mod data; diff --git a/rust/crates/sift_mcp/src/tool_events.json b/rust/crates/sift_mcp/src/tool_events.json index 24220f9712..c6b3fb8ca0 100644 --- a/rust/crates/sift_mcp/src/tool_events.json +++ b/rust/crates/sift_mcp/src/tool_events.json @@ -1,10 +1,12 @@ { "append_test_measurements": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_APPEND_TEST_MEASUREMENTS", + "archive_calculated_channel": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_ARCHIVE_CALCULATED_CHANNEL", "archive_rule": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_ARCHIVE_RULE", "check_for_updates": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_CHECK_FOR_UPDATES", "count_test_measurements": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_COUNT_TEST_MEASUREMENTS", "count_test_steps": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_COUNT_TEST_STEPS", "create_annotation": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_CREATE_ANNOTATION", + "create_calculated_channel": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_CREATE_CALCULATED_CHANNEL", "create_report": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_CREATE_REPORT", "create_report_template": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_CREATE_REPORT_TEMPLATE", "create_rule": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_CREATE_RULE", @@ -13,6 +15,8 @@ "get_data": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_GET_DATA", "list_annotations": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_LIST_ANNOTATIONS", "list_assets": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_LIST_ASSETS", + "list_calculated_channel_versions": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_LIST_CALCULATED_CHANNEL_VERSIONS", + "list_calculated_channels": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_LIST_CALCULATED_CHANNELS", "list_channels": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_LIST_CHANNELS", "list_report_rule_summaries": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_LIST_REPORT_RULE_SUMMARIES", "list_report_templates": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_LIST_REPORT_TEMPLATES", @@ -27,9 +31,11 @@ "ping": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_PING", "search_docs": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_SEARCH_DOCS", "sql": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_SQL", + "unarchive_calculated_channel": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_UNARCHIVE_CALCULATED_CHANNEL", "unarchive_rule": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_UNARCHIVE_RULE", "update_annotation": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_UPDATE_ANNOTATION", "update_asset": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_UPDATE_ASSET", + "update_calculated_channel": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_UPDATE_CALCULATED_CHANNEL", "update_report": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_UPDATE_REPORT", "update_report_template": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_UPDATE_REPORT_TEMPLATE", "update_rule": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_UPDATE_RULE", From 2597184143ac86a83fabd1826a0f9b00c574cd33 Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Tue, 18 Aug 2026 22:42:58 -0700 Subject: [PATCH 4/7] fix: require filter on calculated channel version listing and reject user-notes-only updates --- .../src/service/calculated_channels/mod.rs | 7 +++-- .../src/tool/calculated_channels/mod.rs | 31 ++++++++++--------- .../src/tool/calculated_channels/test.rs | 27 ++++++++++++++-- 3 files changed, 45 insertions(+), 20 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 28431a358f..50e014e524 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs @@ -60,8 +60,10 @@ pub struct CalculatedChannelUpdate { } impl CalculatedChannelUpdate { - /// Whether any updatable field is set. An update with nothing set would - /// produce an empty mask, which the API treats as a no-op. + /// Whether any *maskable* field is set. `user_notes` is deliberately not + /// counted: it rides on the request rather than the update mask, so on its + /// own it produces an empty mask — a no-op the caller would mistake for a + /// new version. pub fn is_empty(&self) -> bool { self.name.is_none() && self.description.is_none() @@ -72,7 +74,6 @@ impl CalculatedChannelUpdate { && self.all_assets.is_none() && self.asset_ids.is_none() && self.tag_ids.is_none() - && self.user_notes.is_none() } } diff --git a/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs b/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs index 70cd594877..1367f9c7ad 100644 --- a/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs +++ b/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs @@ -25,7 +25,7 @@ mod test; #[derive(Debug, Deserialize, JsonSchema)] pub struct CalculatedChannelVersionListParams { calculated_channel_id: String, - filter: Option, + filter: String, order_by: Option, limit: Option, } @@ -109,8 +109,9 @@ impl SiftMcpServer { - Default add `is_archived == false` to the filter. Include archived calculated channels only when the user explicitly asks for them. - Scope with `asset_id == \"...\"` when the asset is known — it is the most selective field. - - Calculated channels do not carry data of their own; they are evaluated per asset. Use this tool to - discover what derived channels exist before reaching for `get_data`. + - Calculated channels store a definition, not stored samples: each one is a SEL expression evaluated + per asset. Use this tool to answer what derived channels exist, what each one computes (read + `calculated_channel_configuration`), and which assets it is scoped to. ", annotations( title = "calculated_channels/list_calculated_channels", @@ -152,10 +153,10 @@ impl SiftMcpServer { Parameters: - `calculated_channel_id`: required. The calculated channel whose versions to list. Resolve it with `list_calculated_channels` first if you only have the name. - - `filter`: optional CEL expression. Filterable fields: `calculated_channel_id`, `organization_id`, + - `filter`: CEL expression. Filterable fields: `calculated_channel_id`, `organization_id`, `client_key`, `name`, `description`, `asset_id`, `asset_name`, `tag_id`, `tag_name`, `version`, `units`, `calculated_channel_version_id`, `created_date`, `modified_date`, `created_by_user_id`, - `modified_by_user_id`, `is_archived`, `archived_date`. Omit or pass an empty string to list all + `modified_by_user_id`, `is_archived`, `archived_date`. Pass an empty string to list all versions. When filtering or searching text, use `name.matches(\"(?i)thrust\")`, not `==`. Use `==` only for an exact value from a prior result. `contains`/`startsWith`/`endsWith` are case-SENSITIVE: `contains(\"Thrust\")` silently misses `thrust_margin`. @@ -195,12 +196,7 @@ impl SiftMcpServer { let versions = self .calculated_channel_service - .list_calculated_channel_versions( - calculated_channel_id, - filter.unwrap_or_default(), - order_by, - limit, - ) + .list_calculated_channel_versions(calculated_channel_id, filter, order_by, limit) .await .map_err(from_anyhow)?; @@ -372,11 +368,13 @@ impl SiftMcpServer { - `asset_ids` / `tag_ids`: optional. REPLACE the corresponding list on the channel's selection scope; the list you omit is preserved. Setting either alongside `all_assets: true` is rejected. - `metadata`: optional. REPLACES the full metadata list. Pass `[]` to clear. - - `user_notes`: optional. Notes recorded against this new version. - - At least one field besides `calculated_channel_id` must be set. + - `user_notes`: optional. Notes recorded against this new version. This ANNOTATES a change; it + cannot be the change. Sending only `user_notes` is rejected, because it would write nothing. + - At least one of `name`, `description`, `units`, `metadata`, `expression`, `all_assets`, + `asset_ids`, or `tag_ids` must be set. Errors: - - `INVALID_PARAMS` if no updatable field is set, `expression` and + - `INVALID_PARAMS` if no updatable field is set (including a `user_notes`-only call), `expression` and `expression_channel_references_json` are not supplied together, the references JSON is invalid, or both asset-scope forms are set. - `INVALID_REQUEST` if the server was launched without `--allow-destructive`. @@ -453,7 +451,10 @@ impl SiftMcpServer { if changes.is_empty() { return Err(ErrorData::invalid_params( - "at least one field besides `calculated_channel_id` must be set", + "at least one changed field besides `calculated_channel_id` must be set; \ + `user_notes` only annotates a change, so it must accompany at least one of \ + `name`, `description`, `units`, `metadata`, `expression`, `all_assets`, \ + `asset_ids`, or `tag_ids`", None, )); } diff --git a/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs b/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs index c2ec99a1df..69e83e7aea 100644 --- a/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs +++ b/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs @@ -174,7 +174,7 @@ async fn list_calculated_channel_versions_returns_versions() { let resp = server .list_calculated_channel_versions(Parameters(CalculatedChannelVersionListParams { calculated_channel_id: "cc1".into(), - filter: None, + filter: String::new(), order_by: None, limit: None, })) @@ -197,7 +197,7 @@ async fn list_calculated_channel_versions_rejects_empty_id() { let err = server .list_calculated_channel_versions(Parameters(CalculatedChannelVersionListParams { calculated_channel_id: String::new(), - filter: None, + filter: String::new(), order_by: None, limit: None, })) @@ -365,6 +365,29 @@ async fn update_calculated_channel_rejects_empty_update() { assert_eq!(err.code, ErrorCode::INVALID_PARAMS); } +#[tokio::test] +async fn update_calculated_channel_rejects_user_notes_only_update() { + // `user_notes` is a request field, not a mask path. On its own it would send + // an empty mask — a no-op the tool would then report as a new version. No + // mock expectations: nothing may reach the wire. + let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; + + let mut params = update_params(); + params.user_notes = Some("just a note".into()); + + let err = server + .update_calculated_channel(Parameters(params)) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); + assert!( + err.message.contains("user_notes"), + "error should name `user_notes` as needing a maskable field: {}", + err.message + ); +} + #[tokio::test] async fn update_calculated_channel_rejects_expression_without_references() { let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; From 277bdc6a26e82023dc7e47d68f3ed676063ae1d1 Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Wed, 26 Aug 2026 00:39:53 -0700 Subject: [PATCH 5/7] feat: add fields projection and item count to calculated channel list tools --- .../src/service/calculated_channels/mod.rs | 32 ++++++-- .../src/service/calculated_channels/test.rs | 21 +++-- .../src/tool/calculated_channels/mod.rs | 63 ++++++++++++--- .../src/tool/calculated_channels/test.rs | 79 ++++++++++++++++++- 4 files changed, 169 insertions(+), 26 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 50e014e524..652fd450b8 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs @@ -102,11 +102,12 @@ impl CalculatedChannelService { filter: String, order_by: Option, limit: Option, - ) -> Result> { + ) -> Result> { let (page_size, record_limit) = common::paging(limit); let mut page_token = String::new(); let mut results = Vec::new(); + let mut has_more = false; let order_by = order_by.unwrap_or_default(); @@ -147,7 +148,13 @@ impl CalculatedChannelService { } results.extend(calculated_channels); - if results.len() >= record_limit || next_page_token.is_empty() { + if results.len() >= record_limit { + // The cap, not the end of the data: report that more exist so the + // caller does not read this page's size as the match total. + has_more = results.len() > record_limit || !next_page_token.is_empty(); + break; + } + if next_page_token.is_empty() { break; } page_token = next_page_token; @@ -155,7 +162,10 @@ impl CalculatedChannelService { results.truncate(record_limit); - Ok(results) + Ok(common::Page { + items: results, + has_more, + }) } /// Lists the version history of a single calculated channel. Each version is @@ -166,11 +176,12 @@ impl CalculatedChannelService { filter: String, order_by: Option, limit: Option, - ) -> Result> { + ) -> Result> { let (page_size, record_limit) = common::paging(limit); let mut page_token = String::new(); let mut results = Vec::new(); + let mut has_more = false; let order_by = order_by.unwrap_or_default(); @@ -215,7 +226,13 @@ impl CalculatedChannelService { } results.extend(calculated_channel_versions); - if results.len() >= record_limit || next_page_token.is_empty() { + if results.len() >= record_limit { + // The cap, not the end of the data: report that more exist so the + // caller does not read this page's size as the match total. + has_more = results.len() > record_limit || !next_page_token.is_empty(); + break; + } + if next_page_token.is_empty() { break; } page_token = next_page_token; @@ -223,7 +240,10 @@ impl CalculatedChannelService { results.truncate(record_limit); - Ok(results) + Ok(common::Page { + items: results, + has_more, + }) } pub async fn create_calculated_channel( 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 c37a38da0a..4f7112d034 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/test.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/test.rs @@ -170,7 +170,8 @@ async fn list_calculated_channels_returns_single_page() { None, ) .await - .expect("list_calculated_channels failed"); + .expect("list_calculated_channels failed") + .items; assert_eq!(channels.len(), 2); assert_eq!(channels[0].calculated_channel_id, "cc1"); @@ -218,7 +219,8 @@ async fn list_calculated_channels_paginates_until_token_empty() { let channels = service .list_calculated_channels(String::new(), None, None) .await - .expect("list_calculated_channels failed"); + .expect("list_calculated_channels failed") + .items; let ids: Vec<&str> = channels .iter() @@ -255,7 +257,8 @@ async fn list_calculated_channels_respects_limit() { let channels = service .list_calculated_channels(String::new(), None, Some(2)) .await - .expect("list_calculated_channels failed"); + .expect("list_calculated_channels failed") + .items; assert_eq!(channels.len(), 2); } @@ -282,7 +285,8 @@ async fn list_calculated_channels_clamps_limit_to_page_size() { let channels = service .list_calculated_channels(String::new(), None, Some(5_000)) .await - .expect("list_calculated_channels failed"); + .expect("list_calculated_channels failed") + .items; assert_eq!(channels.len(), 1); } @@ -304,7 +308,8 @@ async fn list_calculated_channels_breaks_on_empty_page() { let channels = service .list_calculated_channels(String::new(), None, None) .await - .expect("list_calculated_channels failed"); + .expect("list_calculated_channels failed") + .items; assert!(channels.is_empty()); } @@ -362,7 +367,8 @@ async fn list_calculated_channel_versions_builds_request() { Some(10), ) .await - .expect("list_calculated_channel_versions failed"); + .expect("list_calculated_channel_versions failed") + .items; assert_eq!(versions.len(), 1); assert_eq!(versions[0].version, 2); @@ -402,7 +408,8 @@ async fn list_calculated_channel_versions_paginates_until_token_empty() { let versions = service .list_calculated_channel_versions("cc1".to_string(), String::new(), None, None) .await - .expect("list_calculated_channel_versions failed"); + .expect("list_calculated_channel_versions failed") + .items; let numbers: Vec = versions.iter().map(|v| v.version).collect(); assert_eq!(numbers, vec![1, 2]); diff --git a/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs b/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs index 1367f9c7ad..ef495b6e42 100644 --- a/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs +++ b/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs @@ -16,7 +16,7 @@ use crate::{ service::calculated_channels::{ CalculatedChannelUpdate, CalculatedChannelWrite, NewCalculatedChannel, }, - tool::common::{ListParams, MetadataEntry}, + tool::common::{ListParams, MetadataEntry, list_body, to_values}, }; #[cfg(test)] @@ -28,6 +28,7 @@ pub struct CalculatedChannelVersionListParams { filter: String, order_by: Option, limit: Option, + fields: Option>, } #[derive(Debug, Deserialize, JsonSchema)] @@ -82,6 +83,12 @@ impl SiftMcpServer { `archived_date`, and timestamps. - Fields at their proto3 default are OMITTED from the JSON: a missing `is_archived` key means `false`, not \"unknown\". + - `count`: how many items THIS response carries — read it instead of + counting the array yourself. It is the size of the page you got back, not + how many items match `filter`. + - `has_more`: `true` when the service hit `limit` with matches left over, so + this page is not the whole set. Never report `count` as a total while + `has_more` is `true` — narrow `filter` or raise `limit` and ask again. Parameters: - `filter`: CEL expression. Pass an empty string to list everything. Filterable fields: @@ -100,6 +107,14 @@ impl SiftMcpServer { `created_date desc` (newest first). Example: `\"created_date desc,modified_date\"`. - `limit`: max items to return. Start at 50 and only raise it if the result is capped and you still need more. Values are clamped to `1..=200`; omitting it defaults to 50. + - `fields`: optional array of field names to keep on each item, e.g. + `[\"name\"]`. Omit it for the full object. Names match case-insensitively + and ignore underscores and hyphens, so `asset_id`, `assetId` and + `asset-id` all work. Any name that matched nothing on any returned item + is listed in `unmatched_fields`; an empty page reports none, since it + says nothing about whether a name was spelled right. + Reach for this whenever you need only a few fields: full objects are wide, + and a large listing can exceed the response size limit without it. Errors: - `INVALID_PARAMS` if `filter` is not a valid CEL expression or `order_by` references an unknown field. @@ -126,16 +141,23 @@ impl SiftMcpServer { filter, order_by, limit, + fields, }) = params; - let out = self + let page = self .calculated_channel_service .list_calculated_channels(filter, order_by, limit) .await - .map(|channels| serde_json::json!({ "calculated_channels": channels })) .map_err(from_anyhow)?; - Ok(CallToolResult::structured(out)) + let channels = to_values(&page.items)?; + + Ok(CallToolResult::structured(list_body( + "calculated_channels", + channels, + fields, + page.has_more, + ))) } #[tool( @@ -149,6 +171,12 @@ impl SiftMcpServer { is a full `CalculatedChannel` snapshot of that version, including `version`, `version_id`, `change_message`, `user_notes`, `calculated_channel_configuration`, and `modified_by_user_id` — not a reduced version record. + - `count`: how many items THIS response carries — read it instead of + counting the array yourself. It is the size of the page you got back, not + how many items match `filter`. + - `has_more`: `true` when the service hit `limit` with matches left over, so + this page is not the whole set. Never report `count` as a total while + `has_more` is `true` — narrow `filter` or raise `limit` and ask again. Parameters: - `calculated_channel_id`: required. The calculated channel whose versions to list. Resolve it with @@ -165,6 +193,14 @@ impl SiftMcpServer { `created_date` ascending (oldest first) — note this differs from `list_calculated_channels`. - `limit`: max items to return. Start at 50 and only raise it if the result is capped and you still need more. Values are clamped to `1..=200`; omitting it defaults to 50. + - `fields`: optional array of field names to keep on each item, e.g. + `[\"name\"]`. Omit it for the full object. Names match case-insensitively + and ignore underscores and hyphens, so `asset_id`, `assetId` and + `asset-id` all work. Any name that matched nothing on any returned item + is listed in `unmatched_fields`; an empty page reports none, since it + says nothing about whether a name was spelled right. + Reach for this whenever you need only a few fields: full objects are wide, + and a large listing can exceed the response size limit without it. Errors: - `INVALID_PARAMS` if `calculated_channel_id` is empty or `filter` is not a valid CEL expression. @@ -190,11 +226,12 @@ impl SiftMcpServer { filter, order_by, limit, + fields, }) = params; require_id(&calculated_channel_id)?; - let versions = self + let page = self .calculated_channel_service .list_calculated_channel_versions(calculated_channel_id, filter, order_by, limit) .await @@ -203,13 +240,19 @@ impl SiftMcpServer { let next_step = format!( "Listed {} calculated channel versions. Surface the version history to the user, \ highlighting what changed between versions.", - versions.len(), + page.items.len(), + ); + + let versions = to_values(&page.items)?; + let mut body = list_body( + "calculated_channel_versions", + versions, + fields, + page.has_more, ); + body["next_step"] = serde_json::Value::String(next_step.clone()); - let mut result = CallToolResult::structured(serde_json::json!({ - "calculated_channel_versions": versions, - "next_step": next_step, - })); + let mut result = CallToolResult::structured(body); result.content = vec![ContentBlock::text(next_step)]; Ok(result) } diff --git a/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs b/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs index 69e83e7aea..bdc58d4b52 100644 --- a/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs +++ b/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs @@ -17,7 +17,7 @@ use super::{ }; use crate::{ server::SiftMcpServer, - tool::common::test_support::{list_params, structured, structured_field}, + tool::common::test_support::{list_params, list_params_with_fields, structured}, }; const REFERENCES_JSON: &str = r#"[ @@ -129,11 +129,14 @@ async fn list_calculated_channels_returns_single_page() { .await .expect("list_calculated_channels failed"); - let channels = structured_field(resp, "calculated_channels"); - let channels = channels.as_array().expect("expected an array"); + let body = structured(resp); + let channels = body["calculated_channels"] + .as_array() + .expect("expected an array"); assert_eq!(channels.len(), 2); assert_eq!(channels[0]["calculatedChannelId"], "cc1"); assert_eq!(channels[1]["name"], "chamber_dp"); + assert_eq!(body["count"], 2); } #[tokio::test] @@ -177,6 +180,7 @@ async fn list_calculated_channel_versions_returns_versions() { filter: String::new(), order_by: None, limit: None, + fields: None, })) .await .expect("list_calculated_channel_versions failed"); @@ -187,6 +191,7 @@ async fn list_calculated_channel_versions_returns_versions() { .expect("expected an array"); assert_eq!(versions.len(), 1); assert_eq!(versions[0]["version"], 2); + assert_eq!(body["count"], 1); assert!(body["next_step"].is_string()); } @@ -200,6 +205,7 @@ async fn list_calculated_channel_versions_rejects_empty_id() { filter: String::new(), order_by: None, limit: None, + fields: None, })) .await .expect_err("expected error"); @@ -207,6 +213,73 @@ async fn list_calculated_channel_versions_rejects_empty_id() { assert_eq!(err.code, ErrorCode::INVALID_PARAMS); } +#[tokio::test] +async fn list_calculated_channels_projects_requested_fields() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channels().returning(|_| { + Ok(Response::new(ListCalculatedChannelsResponse { + calculated_channels: vec![CalculatedChannel { + calculated_channel_id: "cc1".into(), + name: "thrust_margin".into(), + description: "engine headroom".into(), + ..Default::default() + }], + next_page_token: String::new(), + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let resp = server + .list_calculated_channels(list_params_with_fields("", &["name"])) + .await + .expect("list_calculated_channels failed"); + + let body = structured(resp); + assert_eq!( + body["calculated_channels"], + serde_json::json!([{ "name": "thrust_margin" }]) + ); + assert_eq!(body["count"], 1); +} + +#[tokio::test] +async fn list_calculated_channel_versions_projects_requested_fields() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_list_calculated_channel_versions() + .returning(|_| { + Ok(Response::new(ListCalculatedChannelVersionsResponse { + calculated_channel_versions: vec![CalculatedChannel { + calculated_channel_id: "cc1".into(), + version: 2, + name: "thrust_margin".into(), + ..Default::default() + }], + next_page_token: String::new(), + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let resp = server + .list_calculated_channel_versions(Parameters(CalculatedChannelVersionListParams { + calculated_channel_id: "cc1".into(), + filter: String::new(), + order_by: None, + limit: None, + fields: Some(vec!["version".into()]), + })) + .await + .expect("list_calculated_channel_versions failed"); + + let body = structured(resp); + assert_eq!( + body["calculated_channel_versions"], + serde_json::json!([{ "version": 2 }]) + ); + assert_eq!(body["count"], 1); +} + #[tokio::test] async fn create_calculated_channel_blocked_without_allow_create() { // No expectations on the mock: the gate must fire before any RPC. From 7d91bad07ad62c50896190699a95435f1b6ab00d Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Wed, 26 Aug 2026 01:16:29 -0700 Subject: [PATCH 6/7] fix: archive calculated channels through the is_archived mask --- .../src/service/calculated_channels/mod.rs | 28 ++----- .../src/service/calculated_channels/test.rs | 32 +++----- .../src/tool/calculated_channels/mod.rs | 56 ++++++++++---- .../src/tool/calculated_channels/test.rs | 76 +++++++++++++++++++ 4 files changed, 134 insertions(+), 58 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 652fd450b8..6d2a82c41f 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/mod.rs @@ -1,9 +1,7 @@ -use std::time::{SystemTime, UNIX_EPOCH}; - use crate::policy::{RetryPolicy, with_retry}; use crate::service::common; use anyhow::{Context, Result, anyhow}; -use pbjson_types::{FieldMask, Timestamp}; +use pbjson_types::FieldMask; use sift_rs::{ SiftChannel, calculated_channels::v2::{ @@ -400,36 +398,34 @@ impl CalculatedChannelService { self.send_update(channel, paths, user_notes).await } - /// Archives a calculated channel by stamping `archived_date`. There is no - /// dedicated archive RPC; the API archives through the update mask. + /// Archives a calculated channel through the `is_archived` update mask. pub async fn archive_calculated_channel( &self, calculated_channel_id: String, ) -> Result { let channel = CalculatedChannel { calculated_channel_id, - archived_date: Some(now_timestamp()), + is_archived: true, ..Default::default() }; - self.send_update(channel, vec!["archived_date".to_string()], None) + self.send_update(channel, vec!["is_archived".to_string()], None) .await .context("failed to archive calculated channel") } - /// Unarchives a calculated channel by clearing `archived_date` through the - /// update mask. A masked field left at its default is cleared. + /// Unarchives a calculated channel through the `is_archived` update mask. pub async fn unarchive_calculated_channel( &self, calculated_channel_id: String, ) -> Result { let channel = CalculatedChannel { calculated_channel_id, - archived_date: None, + is_archived: false, ..Default::default() }; - self.send_update(channel, vec!["archived_date".to_string()], None) + self.send_update(channel, vec!["is_archived".to_string()], None) .await .context("failed to unarchive calculated channel") } @@ -537,13 +533,3 @@ fn current_selection(configuration: &CalculatedChannelConfiguration) -> AssetSel _ => AssetSelection::default(), } } - -fn now_timestamp() -> Timestamp { - let elapsed = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default(); - Timestamp { - seconds: elapsed.as_secs() as i64, - nanos: elapsed.subsec_nanos() as i32, - } -} 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 4f7112d034..3a9787f05e 100644 --- a/rust/crates/sift_mcp/src/service/calculated_channels/test.rs +++ b/rust/crates/sift_mcp/src/service/calculated_channels/test.rs @@ -1,4 +1,3 @@ -use pbjson_types::Timestamp; use sift_rs::{ calculated_channels::v2::{ CalculatedChannel, CalculatedChannelAbstractChannelReference, @@ -902,7 +901,7 @@ async fn update_calculated_channel_propagates_grpc_error() { } #[tokio::test] -async fn archive_calculated_channel_masks_archived_date_only() { +async fn archive_calculated_channel_masks_is_archived_only() { let mut mock = MockCalculatedChannelServiceImpl::new(); mock.expect_update_calculated_channel() .times(1) @@ -911,9 +910,9 @@ async fn archive_calculated_channel_masks_archived_date_only() { let channel = req.calculated_channel.as_ref().expect("channel present"); let mask = req.update_mask.as_ref().expect("mask present"); - mask.paths == vec!["archived_date".to_string()] + mask.paths == vec!["is_archived".to_string()] && channel.calculated_channel_id == "cc1" - && channel.archived_date.is_some() + && channel.is_archived }) .returning(|req| { Ok(Response::new(UpdateCalculatedChannelResponse { @@ -954,7 +953,7 @@ async fn archive_calculated_channel_does_not_read_first() { } #[tokio::test] -async fn unarchive_calculated_channel_clears_archived_date_with_mask_set() { +async fn unarchive_calculated_channel_masks_is_archived_false() { let mut mock = MockCalculatedChannelServiceImpl::new(); mock.expect_update_calculated_channel() .times(1) @@ -963,18 +962,13 @@ async fn unarchive_calculated_channel_clears_archived_date_with_mask_set() { let channel = req.calculated_channel.as_ref().expect("channel present"); let mask = req.update_mask.as_ref().expect("mask present"); - mask.paths == vec!["archived_date".to_string()] + mask.paths == vec!["is_archived".to_string()] && channel.calculated_channel_id == "cc1" - && channel.archived_date.is_none() + && !channel.is_archived }) .returning(|req| { - let mut channel = req - .into_inner() - .calculated_channel - .expect("channel present"); - channel.archived_date = None; Ok(Response::new(UpdateCalculatedChannelResponse { - calculated_channel: Some(channel), + calculated_channel: req.into_inner().calculated_channel, inapplicable_assets: vec![], })) }); @@ -986,7 +980,7 @@ async fn unarchive_calculated_channel_clears_archived_date_with_mask_set() { .await .expect("unarchive_calculated_channel failed"); - assert!(written.calculated_channel.archived_date.is_none()); + assert!(!written.calculated_channel.is_archived); } #[tokio::test] @@ -1033,7 +1027,7 @@ async fn archive_calculated_channel_errors_when_response_missing_channel() { } #[tokio::test] -async fn archive_calculated_channel_stamps_a_current_timestamp() { +async fn archive_calculated_channel_sets_is_archived_true() { let mut mock = MockCalculatedChannelServiceImpl::new(); mock.expect_update_calculated_channel().returning(|req| { Ok(Response::new(UpdateCalculatedChannelResponse { @@ -1049,11 +1043,5 @@ async fn archive_calculated_channel_stamps_a_current_timestamp() { .await .expect("archive_calculated_channel failed"); - let Timestamp { seconds, .. } = written - .calculated_channel - .archived_date - .expect("archived_date stamped"); - // Sanity floor: 2020-01-01T00:00:00Z. Guards against a zero-valued default - // being sent as the archive timestamp. - assert!(seconds > 1_577_836_800); + assert!(written.calculated_channel.is_archived); } diff --git a/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs b/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs index ef495b6e42..e033c7cb79 100644 --- a/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs +++ b/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs @@ -526,8 +526,8 @@ impl SiftMcpServer { Reversible with `unarchive_calculated_channel`. Output: - - `{ \"archived\": true, \"calculated_channel\": CalculatedChannel, \"next_step\": string }`. The - returned channel carries the `archived_date` the server recorded. + - `{ \"archived\": boolean, \"calculated_channel\": CalculatedChannel, \"next_step\": string }`. + `archived` reflects the returned channel state. Parameters: - `calculated_channel_id`: required. The calculated channel to archive. @@ -569,13 +569,26 @@ impl SiftMcpServer { .await .map_err(from_anyhow)?; - let next_step = format!( - "Archived calculated channel `{}`. Tell the user it is archived and no longer offered \ - for plotting or querying, and that `unarchive_calculated_channel` restores it.", - written.calculated_channel.calculated_channel_id, - ); + let archived = written.calculated_channel.is_archived; + let next_step = if archived { + format!( + "Archived calculated channel `{}`. Tell the user it is archived and no longer offered \ + for plotting or querying, and that `unarchive_calculated_channel` restores it.", + written.calculated_channel.calculated_channel_id, + ) + } else { + format!( + "The server returned calculated channel `{}` as unarchived after the archive request. \ + Do not report it as archived; verify its state before continuing.", + written.calculated_channel.calculated_channel_id, + ) + }; - Ok(write_result(written, next_step, Some(("archived", true)))) + Ok(write_result( + written, + next_step, + Some(("archived", archived)), + )) } #[tool( @@ -584,8 +597,8 @@ impl SiftMcpServer { Restore a previously archived calculated channel so it is offered again. This is a WRITE. Output: - - `{ \"unarchived\": true, \"calculated_channel\": CalculatedChannel, \"next_step\": string }`. The - returned channel has no `archived_date`. + - `{ \"unarchived\": boolean, \"calculated_channel\": CalculatedChannel, \"next_step\": string }`. + `unarchived` reflects the returned channel state. Parameters: - `calculated_channel_id`: required. The calculated channel to restore. @@ -625,12 +638,25 @@ impl SiftMcpServer { .await .map_err(from_anyhow)?; - let next_step = format!( - "Unarchived calculated channel `{}`. Tell the user it is restored and available again.", - written.calculated_channel.calculated_channel_id, - ); + let unarchived = !written.calculated_channel.is_archived; + let next_step = if unarchived { + format!( + "Unarchived calculated channel `{}`. Tell the user it is restored and available again.", + written.calculated_channel.calculated_channel_id, + ) + } else { + format!( + "The server returned calculated channel `{}` as archived after the unarchive request. \ + Do not report it as restored; verify its state before continuing.", + written.calculated_channel.calculated_channel_id, + ) + }; - Ok(write_result(written, next_step, Some(("unarchived", true)))) + Ok(write_result( + written, + next_step, + Some(("unarchived", unarchived)), + )) } } diff --git a/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs b/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs index bdc58d4b52..ec893578a8 100644 --- a/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs +++ b/rust/crates/sift_mcp/src/tool/calculated_channels/test.rs @@ -583,6 +583,15 @@ async fn archive_calculated_channel_returns_structured_result() { let mut mock = MockCalculatedChannelServiceImpl::new(); mock.expect_update_calculated_channel() .times(1) + .withf(|req| { + let req = req.get_ref(); + let channel = req.calculated_channel.as_ref().expect("channel present"); + let mask = req.update_mask.as_ref().expect("mask present"); + + mask.paths == vec!["is_archived".to_string()] + && channel.calculated_channel_id == "cc1" + && channel.is_archived + }) .returning(|req| { Ok(Response::new(UpdateCalculatedChannelResponse { calculated_channel: req.into_inner().calculated_channel, @@ -610,6 +619,15 @@ async fn unarchive_calculated_channel_returns_structured_result() { let mut mock = MockCalculatedChannelServiceImpl::new(); mock.expect_update_calculated_channel() .times(1) + .withf(|req| { + let req = req.get_ref(); + let channel = req.calculated_channel.as_ref().expect("channel present"); + let mask = req.update_mask.as_ref().expect("mask present"); + + mask.paths == vec!["is_archived".to_string()] + && channel.calculated_channel_id == "cc1" + && !channel.is_archived + }) .returning(|req| { Ok(Response::new(UpdateCalculatedChannelResponse { calculated_channel: req.into_inner().calculated_channel, @@ -631,6 +649,64 @@ async fn unarchive_calculated_channel_returns_structured_result() { assert!(body["next_step"].is_string()); } +#[tokio::test] +async fn archive_calculated_channel_surfaces_a_returned_unarchived_state() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_update_calculated_channel().returning(|req| { + let mut channel = req + .into_inner() + .calculated_channel + .expect("channel present"); + channel.is_archived = false; + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: Some(channel), + inapplicable_assets: vec![], + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let resp = server + .archive_calculated_channel(Parameters(CalculatedChannelArchiveParams { + calculated_channel_id: "cc1".into(), + })) + .await + .expect("archive_calculated_channel failed"); + + let body = structured(resp); + assert_eq!(body["archived"], false); + assert!(body["next_step"].as_str().unwrap().contains("unarchived")); +} + +#[tokio::test] +async fn unarchive_calculated_channel_surfaces_a_returned_archived_state() { + let mut mock = MockCalculatedChannelServiceImpl::new(); + mock.expect_update_calculated_channel().returning(|req| { + let mut channel = req + .into_inner() + .calculated_channel + .expect("channel present"); + channel.is_archived = true; + Ok(Response::new(UpdateCalculatedChannelResponse { + calculated_channel: Some(channel), + inapplicable_assets: vec![], + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let resp = server + .unarchive_calculated_channel(Parameters(CalculatedChannelArchiveParams { + calculated_channel_id: "cc1".into(), + })) + .await + .expect("unarchive_calculated_channel failed"); + + let body = structured(resp); + assert_eq!(body["unarchived"], false); + assert!(body["next_step"].as_str().unwrap().contains("archived")); +} + #[tokio::test] async fn archive_calculated_channel_rejects_empty_id() { let (server, _h) = server_with_mock(MockCalculatedChannelServiceImpl::new()).await; From 0bbb0e991cfa2947c6deb100c17484c17a2fff6f Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Wed, 26 Aug 2026 21:36:41 -0700 Subject: [PATCH 7/7] docs: match order_by docs to the backend's orderable fields and defaults --- rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs b/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs index e033c7cb79..8b01db110c 100644 --- a/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs +++ b/rust/crates/sift_mcp/src/tool/calculated_channels/mod.rs @@ -103,7 +103,7 @@ impl SiftMcpServer { `contains(\"Thrust\")` silently misses `thrust_margin`. Calculated channel names can embed `.`, a regex wildcard, so match a full literal name with `contains`, not `matches`. - `order_by`: optional comma-separated `FIELD_NAME[ desc]` list. Orderable fields: `created_date`, - `modified_date`, `name`, `description`, `units`, `archived_date`. Default sort is + `modified_date`. Default sort is `created_date desc` (newest first). Example: `\"created_date desc,modified_date\"`. - `limit`: max items to return. Start at 50 and only raise it if the result is capped and you still need more. Values are clamped to `1..=200`; omitting it defaults to 50. @@ -188,9 +188,9 @@ impl SiftMcpServer { versions. When filtering or searching text, use `name.matches(\"(?i)thrust\")`, not `==`. Use `==` only for an exact value from a prior result. `contains`/`startsWith`/`endsWith` are case-SENSITIVE: `contains(\"Thrust\")` silently misses `thrust_margin`. - - `order_by`: optional comma-separated `FIELD_NAME[ desc]` list. Orderable fields: `version`, - `created_date`, `modified_date`, `name`, `description`, `units`, `archived_date`. Default sort is - `created_date` ascending (oldest first) — note this differs from `list_calculated_channels`. + - `order_by`: optional comma-separated `FIELD_NAME[ desc]` list. Orderable fields: `created_date`, + `modified_date`, `version`. Default sort is `version` ascending (oldest first) — note this + differs from `list_calculated_channels`. Example: `\"version desc,created_date\"`. - `limit`: max items to return. Start at 50 and only raise it if the result is capped and you still need more. Values are clamped to `1..=200`; omitting it defaults to 50. - `fields`: optional array of field names to keep on each item, e.g.