From 8b120a5c2579353e4b316a6ef37b9a390f24dfee Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Thu, 20 Aug 2026 15:28:23 -0700 Subject: [PATCH 1/7] feat: add bulk annotation updates --- .../sift_mcp/src/service/annotations/mod.rs | 33 ++- .../sift_mcp/src/service/annotations/test.rs | 2 + .../sift_mcp/src/tool/annotations/mod.rs | 123 ++++++++---- .../sift_mcp/src/tool/annotations/test.rs | 189 +++++++++++++++++- 4 files changed, 298 insertions(+), 49 deletions(-) diff --git a/rust/crates/sift_mcp/src/service/annotations/mod.rs b/rust/crates/sift_mcp/src/service/annotations/mod.rs index 8790a3810..aebc76a64 100644 --- a/rust/crates/sift_mcp/src/service/annotations/mod.rs +++ b/rust/crates/sift_mcp/src/service/annotations/mod.rs @@ -6,8 +6,8 @@ use sift_rs::{ SiftChannel, annotations::v1::{ Annotation, AnnotationLinkedChannel, AnnotationState, AnnotationType, - CreateAnnotationRequest, ListAnnotationsRequest, ListAnnotationsResponse, - UpdateAnnotationRequest, annotation_linked_channel, + BatchArchiveAnnotationsRequest, CreateAnnotationRequest, ListAnnotationsRequest, + ListAnnotationsResponse, UpdateAnnotationRequest, annotation_linked_channel, annotation_service_client::AnnotationServiceClient, }, metadata::v1::MetadataValue, @@ -176,11 +176,33 @@ impl AnnotationService { .ok_or_else(|| anyhow!("create_annotation response missing annotation")) } + pub async fn batch_archive_annotations( + &self, + annotation_ids: Vec, + ) -> Result> { + let channel = self.channel.clone(); + let resp = with_retry(&self.policy, move || { + let channel = channel.clone(); + let annotation_ids = annotation_ids.clone(); + async move { + let mut client = AnnotationServiceClient::new(channel); + client + .batch_archive_annotations(BatchArchiveAnnotationsRequest { annotation_ids }) + .await + .map(|resp| resp.into_inner()) + } + }) + .await + .context("failed to batch archive annotations")?; + + Ok(resp.annotations) + } + /// Update a subset of an existing annotation's fields. Per /// `protos/sift/annotations/v1/annotations.proto::UpdateAnnotationRequest` the /// updatable fields are `name`, `description`, `start_time`, `end_time`, /// `assigned_to_user_id`, `state`, `tags`, `legend_config`, `linked_channels`, - /// and `metadata`. This service exposes all but `legend_config`. + /// `metadata`, and `is_archived`. This service exposes all but `legend_config`. /// /// `tags`, `linked_channels`, and `metadata` use REPLACE semantics — passing /// `Some(vec![])` clears the field. @@ -197,6 +219,7 @@ impl AnnotationService { tags: Option>, linked_channel_ids: Option>, metadata: Option>, + is_archived: Option, ) -> Result { let mut annotation = Annotation { annotation_id, @@ -240,6 +263,10 @@ impl AnnotationService { annotation.metadata = v; paths.push("metadata".to_string()); } + if let Some(v) = is_archived { + annotation.is_archived = v; + paths.push("is_archived".to_string()); + } let channel = self.channel.clone(); let resp = with_retry(&self.policy, move || { diff --git a/rust/crates/sift_mcp/src/service/annotations/test.rs b/rust/crates/sift_mcp/src/service/annotations/test.rs index 1cda4956a..f71bdb3d6 100644 --- a/rust/crates/sift_mcp/src/service/annotations/test.rs +++ b/rust/crates/sift_mcp/src/service/annotations/test.rs @@ -311,6 +311,7 @@ async fn update_annotation_builds_mask_from_provided_fields() { Some(vec!["important".to_string()]), None, None, + None, ) .await .expect("update_annotation failed"); @@ -338,6 +339,7 @@ async fn update_annotation_propagates_grpc_error() { None, None, None, + None, ) .await .expect_err("expected error"); diff --git a/rust/crates/sift_mcp/src/tool/annotations/mod.rs b/rust/crates/sift_mcp/src/tool/annotations/mod.rs index 528cf390a..ef8283e4f 100644 --- a/rust/crates/sift_mcp/src/tool/annotations/mod.rs +++ b/rust/crates/sift_mcp/src/tool/annotations/mod.rs @@ -20,6 +20,8 @@ use crate::{ #[cfg(test)] mod test; +const MAX_UPDATE_ANNOTATIONS: usize = 1_000; + #[derive(Debug, Deserialize, JsonSchema)] pub struct AnnotationListParams { pub(crate) filter: String, @@ -48,7 +50,7 @@ pub struct CreateAnnotationParams { #[derive(Debug, Deserialize, JsonSchema)] pub struct UpdateAnnotationParams { - annotation_id: String, + annotation_ids: Vec, name: Option, description: Option, start_time_unix_nanos: Option, @@ -58,6 +60,7 @@ pub struct UpdateAnnotationParams { tags: Option>, linked_channel_ids: Option>, metadata: Option>, + is_archived: Option, } fn parse_annotation_type(s: &str) -> Result { @@ -305,15 +308,16 @@ impl SiftMcpServer { #[tool( name = "update_annotation", description = " - Update an existing annotation. Wraps `annotations/v1 UpdateAnnotation`. + Update one or more existing annotations. Uses `annotations/v1 BatchArchiveAnnotations` when archiving. + Other changes, including unarchiving, use one `UpdateAnnotation` request per annotation. Output: - - `{ \"annotation\": Annotation, \"annotation_url\": string|null, \"next_step\": string }`. The - returned `Annotation` is the post-update state from the server; `annotation_url` is its Sift web - link, or null when the host can't be derived. + - `{ \"annotations\": [Annotation, ...], \"next_step\": string }`. Each item is the post-update + state from the server and includes a `url` field with its Sift web link when the host can be derived. Parameters: - - `annotation_id`: required; the id of the annotation to update. + - `annotation_ids`: required list of 1 to 1000 annotation ids. The same changes are applied to every + annotation. - `name`: optional new name. - `description`: optional new description. - `start_time_unix_nanos` / `end_time_unix_nanos`: optional new time bounds in Unix nanoseconds. @@ -324,17 +328,23 @@ impl SiftMcpServer { Pass `[]` to clear. Bit-field and calculated-channel links are not exposed here. - `metadata`: optional; REPLACES the full metadata list. Each entry is `{ \"name\": \"\", \"value\": }`. Pass `[]` to clear. + - `is_archived`: optional archive state. `true` uses one batch-archive request. `false` is included in + each annotation's individual update request. When `true` is combined with other fields, annotations + are updated before archiving. At least one updatable field must be set; otherwise the tool returns `INVALID_PARAMS`. Errors: - - `INVALID_PARAMS` if `annotation_id` is empty, `state` is unrecognized, or no updatable field is set. - - `RESOURCE_NOT_FOUND` if no annotation matches `annotation_id`. + - `INVALID_PARAMS` if `annotation_ids` is empty, contains an empty id, exceeds 1000 ids, `state` is + unrecognized, or no updatable field is set. + - `RESOURCE_NOT_FOUND` if no annotation matches one of the ids. - `INTERNAL_ERROR` for upstream gRPC failures. Guidance: - - This is a write. CONFIRM the target and the full proposed values with the user before invoking — + - This is a write. CONFIRM every target and the full proposed values with the user before invoking — `tags`, `linked_channel_ids`, and `metadata` are REPLACE operations, not merges. + - General field updates and unarchive operations issue one API request per annotation and are not + atomic. A failure can leave earlier annotations updated. Archive uses a single batch API request. - For appends, read the current annotation via `list_annotations` filtered by `annotation_id == \"\"`, then send the union. ", @@ -352,7 +362,7 @@ impl SiftMcpServer { self.require_destructive()?; let Parameters(UpdateAnnotationParams { - annotation_id, + annotation_ids, name, description, start_time_unix_nanos, @@ -362,16 +372,31 @@ impl SiftMcpServer { tags, linked_channel_ids, metadata, + is_archived, }) = params; - if annotation_id.is_empty() { + if annotation_ids.is_empty() { + return Err(ErrorData::invalid_params( + "`annotation_ids` must contain at least one id", + None, + )); + } + + if annotation_ids.len() > MAX_UPDATE_ANNOTATIONS { return Err(ErrorData::invalid_params( - "`annotation_id` must not be empty", + format!("`annotation_ids` must contain at most {MAX_UPDATE_ANNOTATIONS} ids"), None, )); } - let has_update = name.is_some() + if annotation_ids.iter().any(String::is_empty) { + return Err(ErrorData::invalid_params( + "`annotation_ids` must not contain empty ids", + None, + )); + } + + let has_field_update = name.is_some() || description.is_some() || start_time_unix_nanos.is_some() || end_time_unix_nanos.is_some() @@ -380,7 +405,7 @@ impl SiftMcpServer { || tags.is_some() || linked_channel_ids.is_some() || metadata.is_some(); - if !has_update { + if !has_field_update && is_archived.is_none() { return Err(ErrorData::invalid_params( "at least one updatable field must be provided", None, @@ -389,39 +414,55 @@ impl SiftMcpServer { let state = state.map(|s| parse_annotation_state(&s)).transpose()?; let metadata = metadata.map(|m| m.into_iter().map(MetadataValue::from).collect::>()); + let unarchive = (is_archived == Some(false)).then_some(false); + let has_individual_update = has_field_update || unarchive.is_some(); + + let mut annotations = Vec::new(); + if has_individual_update { + annotations.reserve(annotation_ids.len()); + for annotation_id in &annotation_ids { + let annotation = self + .annotation_service + .update_annotation( + annotation_id.clone(), + name.clone(), + description.clone(), + start_time_unix_nanos, + end_time_unix_nanos, + assigned_to_user_id.clone(), + state, + tags.clone(), + linked_channel_ids.clone(), + metadata.clone(), + unarchive, + ) + .await + .map_err(from_anyhow)?; + annotations.push(annotation); + } + } - let annotation = self - .annotation_service - .update_annotation( - annotation_id, - name, - description, - start_time_unix_nanos, - end_time_unix_nanos, - assigned_to_user_id, - state, - tags, - linked_channel_ids, - metadata, - ) - .await - .map_err(from_anyhow)?; + if is_archived == Some(true) { + annotations = self + .annotation_service + .batch_archive_annotations(annotation_ids) + .await + .map_err(from_anyhow)?; + } - let annotation_url = self - .url_service - .build_annotation_url(&annotation.annotation_id) - .ok(); + let updated_count = annotations.len(); + let annotations = with_urls(&annotations, |annotation| { + self.url_service + .build_annotation_url(&annotation.annotation_id) + .ok() + })?; let next_step = format!( - "Updated annotation `{}` ({}).{} Surface the new state to the user and confirm the change \ - matches their intent. Remember: tags, linked channels, and metadata are REPLACE operations.", - annotation.name, - annotation.annotation_id, - url_clause(annotation_url.as_deref()), + "Updated {updated_count} annotation(s). Surface the new states and links to the user and confirm the \ + changes match their intent. Remember: tags, linked channels, and metadata are REPLACE operations.", ); let mut result = CallToolResult::structured(serde_json::json!({ - "annotation": annotation, - "annotation_url": annotation_url, + "annotations": annotations, "next_step": next_step, })); result.content = vec![ContentBlock::text(next_step)]; diff --git a/rust/crates/sift_mcp/src/tool/annotations/test.rs b/rust/crates/sift_mcp/src/tool/annotations/test.rs index 614f10641..cd0a7ff95 100644 --- a/rust/crates/sift_mcp/src/tool/annotations/test.rs +++ b/rust/crates/sift_mcp/src/tool/annotations/test.rs @@ -1,7 +1,12 @@ +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, +}; + use rmcp::{handler::server::wrapper::Parameters, model::ErrorCode}; use sift_rs::annotations::v1::{ - Annotation, CreateAnnotationResponse, ListAnnotationsResponse, UpdateAnnotationResponse, - annotation_service_server::AnnotationServiceServer, + Annotation, BatchArchiveAnnotationsResponse, CreateAnnotationResponse, ListAnnotationsResponse, + UpdateAnnotationResponse, annotation_service_server::AnnotationServiceServer, }; use sift_test_util::{grpc::memory_sift_channel, mock::annotations::v1::MockAnnotationServiceImpl}; use tokio::task::JoinHandle; @@ -48,7 +53,7 @@ fn create_params() -> CreateAnnotationParams { fn update_params(annotation_id: &str) -> UpdateAnnotationParams { UpdateAnnotationParams { - annotation_id: annotation_id.into(), + annotation_ids: vec![annotation_id.into()], name: None, description: None, start_time_unix_nanos: None, @@ -58,9 +63,20 @@ fn update_params(annotation_id: &str) -> UpdateAnnotationParams { tags: None, linked_channel_ids: None, metadata: None, + is_archived: None, } } +#[test] +fn update_annotation_params_accept_bulk_ids() { + let params = serde_json::from_value::(serde_json::json!({ + "annotation_ids": ["ann1", "ann2"], + "name": "renamed", + })); + + assert!(params.is_ok()); +} + #[tokio::test] async fn list_annotations_returns_single_page() { let mut mock = MockAnnotationServiceImpl::new(); @@ -208,8 +224,171 @@ async fn update_annotation_happy_path() { .await .expect("update_annotation failed"); - let annotation = structured_field(resp, "annotation"); - assert_eq!(annotation["name"], "renamed"); + let annotations = structured_field(resp, "annotations"); + assert_eq!(annotations[0]["name"], "renamed"); + assert_eq!( + annotations[0]["url"], + "https://app.test.local/annotation/ann1" + ); +} + +#[tokio::test] +async fn update_annotation_updates_each_annotation() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_update_annotation().times(2).returning(|req| { + let annotation = req.into_inner().annotation.unwrap(); + Ok(Response::new(UpdateAnnotationResponse { + annotation: Some(annotation), + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let mut params = update_params("ann1"); + params.annotation_ids.push("ann2".into()); + params.name = Some("renamed".into()); + + let resp = server + .update_annotation(Parameters(params)) + .await + .expect("update_annotation failed"); + + let annotations = structured_field(resp, "annotations"); + assert_eq!(annotations[0]["annotationId"], "ann1"); + assert_eq!(annotations[1]["annotationId"], "ann2"); +} + +#[tokio::test] +async fn update_annotation_batch_archives_annotations() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_batch_archive_annotations() + .withf(|req| req.get_ref().annotation_ids == ["ann1", "ann2"]) + .returning(|_| { + Ok(Response::new(BatchArchiveAnnotationsResponse { + annotations: vec![ + Annotation { + annotation_id: "ann1".into(), + is_archived: true, + ..Default::default() + }, + Annotation { + annotation_id: "ann2".into(), + is_archived: true, + ..Default::default() + }, + ], + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let mut params = update_params("ann1"); + params.annotation_ids.push("ann2".into()); + params.is_archived = Some(true); + + let resp = server + .update_annotation(Parameters(params)) + .await + .expect("update_annotation failed"); + + let annotations = structured_field(resp, "annotations"); + assert_eq!(annotations.as_array().unwrap().len(), 2); + assert_eq!(annotations[0]["isArchived"], true); +} + +#[tokio::test] +async fn update_annotation_updates_fields_before_archiving() { + let mut mock = MockAnnotationServiceImpl::new(); + let updated_count = Arc::new(AtomicUsize::new(0)); + + let update_count = Arc::clone(&updated_count); + mock.expect_update_annotation() + .times(2) + .returning(move |req| { + update_count.fetch_add(1, Ordering::SeqCst); + let annotation = req.into_inner().annotation.unwrap(); + Ok(Response::new(UpdateAnnotationResponse { + annotation: Some(annotation), + })) + }); + + let archive_count = Arc::clone(&updated_count); + mock.expect_batch_archive_annotations() + .withf(move |_| archive_count.load(Ordering::SeqCst) == 2) + .returning(|req| { + let annotations = req + .into_inner() + .annotation_ids + .into_iter() + .map(|annotation_id| Annotation { + annotation_id, + is_archived: true, + ..Default::default() + }) + .collect(); + Ok(Response::new(BatchArchiveAnnotationsResponse { + annotations, + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let mut params = update_params("ann1"); + params.annotation_ids.push("ann2".into()); + params.name = Some("renamed".into()); + params.is_archived = Some(true); + + server + .update_annotation(Parameters(params)) + .await + .expect("update_annotation failed"); +} + +#[tokio::test] +async fn update_annotation_unarchives_each_annotation() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_update_annotation() + .times(2) + .withf(|req| { + let req = req.get_ref(); + !req.annotation.as_ref().unwrap().is_archived + && req.update_mask.as_ref().unwrap().paths == ["is_archived"] + }) + .returning(|req| { + let annotation = req.into_inner().annotation.unwrap(); + Ok(Response::new(UpdateAnnotationResponse { + annotation: Some(annotation), + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let mut params = update_params("ann1"); + params.annotation_ids.push("ann2".into()); + params.is_archived = Some(false); + + let resp = server + .update_annotation(Parameters(params)) + .await + .expect("update_annotation failed"); + + let annotations = structured_field(resp, "annotations"); + assert_eq!(annotations.as_array().unwrap().len(), 2); +} + +#[tokio::test] +async fn update_annotation_rejects_empty_ids() { + let (server, _h) = server_with_mock(MockAnnotationServiceImpl::new()).await; + + let mut params = update_params("ann1"); + params.annotation_ids.clear(); + + let err = server + .update_annotation(Parameters(params)) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); } #[tokio::test] From bacfcac8eab31b329509c61b8d17d7279a75b19e Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Fri, 21 Aug 2026 17:21:17 -0700 Subject: [PATCH 2/7] fix: handle partial bulk annotation updates --- rust/crates/sift_mcp/Cargo.toml | 1 + .../sift_mcp/src/service/annotations/mod.rs | 134 ++++++++++++++- .../sift_mcp/src/service/annotations/test.rs | 152 +++++++++++++++++- .../sift_mcp/src/tool/annotations/mod.rs | 134 +++++++++------ .../sift_mcp/src/tool/annotations/test.rs | 108 ++++++++++++- 5 files changed, 474 insertions(+), 55 deletions(-) diff --git a/rust/crates/sift_mcp/Cargo.toml b/rust/crates/sift_mcp/Cargo.toml index be3185cef..172bfc460 100644 --- a/rust/crates/sift_mcp/Cargo.toml +++ b/rust/crates/sift_mcp/Cargo.toml @@ -23,6 +23,7 @@ tonic.workspace = true anyhow.workspace = true chrono.workspace = true clap = { workspace = true, features = ["cargo"] } +futures.workspace = true pbjson-types.workspace = true percent-encoding.workspace = true prost.workspace = true diff --git a/rust/crates/sift_mcp/src/service/annotations/mod.rs b/rust/crates/sift_mcp/src/service/annotations/mod.rs index aebc76a64..ff54f0fc0 100644 --- a/rust/crates/sift_mcp/src/service/annotations/mod.rs +++ b/rust/crates/sift_mcp/src/service/annotations/mod.rs @@ -1,6 +1,7 @@ use crate::policy::{RetryPolicy, with_retry}; use crate::service::common; -use anyhow::{Context, Result, anyhow}; +use anyhow::{Context, Error, Result, anyhow}; +use futures::{StreamExt, stream}; use pbjson_types::{FieldMask, Timestamp}; use sift_rs::{ SiftChannel, @@ -16,6 +17,39 @@ use sift_rs::{ #[cfg(test)] mod test; +const MAX_CONCURRENT_ANNOTATION_UPDATES: usize = 50; + +#[derive(Debug)] +pub struct AnnotationUpdateFailure { + pub annotation_id: String, + pub error: Error, +} + +#[derive(Debug)] +pub struct UpdateAnnotationsResult { + pub annotations: Vec, + pub failures: Vec, + pub batch_archive_error: Option, + pub archive_skipped: bool, +} + +async fn fan_out_bounded(items: Vec, concurrency: usize, mut op: F) -> Vec +where + F: FnMut(T) -> Fut, + Fut: Future, +{ + let mut results = stream::iter(items.into_iter().enumerate()) + .map(|(index, item)| { + let future = op(item); + async move { (index, future.await) } + }) + .buffer_unordered(concurrency) + .collect::>() + .await; + results.sort_unstable_by_key(|(index, _)| *index); + results.into_iter().map(|(_, result)| result).collect() +} + /// Build a protobuf `Timestamp` from Unix nanoseconds via the shared helper. fn timestamp_from_unix_nanos(nanos: i64) -> Timestamp { let (seconds, nanos) = common::unix_nanos_to_secs_and_subsec_nanos(nanos); @@ -198,6 +232,104 @@ impl AnnotationService { Ok(resp.annotations) } + #[allow(clippy::too_many_arguments)] + pub async fn update_annotations( + &self, + annotation_ids: Vec, + name: Option, + description: Option, + start_time_unix_nanos: Option, + end_time_unix_nanos: Option, + assigned_to_user_id: Option, + state: Option, + tags: Option>, + linked_channel_ids: Option>, + metadata: Option>, + is_archived: Option, + ) -> Result { + let has_field_update = name.is_some() + || description.is_some() + || start_time_unix_nanos.is_some() + || end_time_unix_nanos.is_some() + || assigned_to_user_id.is_some() + || state.is_some() + || tags.is_some() + || linked_channel_ids.is_some() + || metadata.is_some(); + let unarchive = (is_archived == Some(false)).then_some(false); + let has_individual_update = has_field_update || unarchive.is_some(); + + let mut annotations = Vec::new(); + let mut failures = Vec::new(); + + if has_individual_update { + let service = self.clone(); + let updates = fan_out_bounded( + annotation_ids.clone(), + MAX_CONCURRENT_ANNOTATION_UPDATES, + move |annotation_id| { + let service = service.clone(); + let name = name.clone(); + let description = description.clone(); + let assigned_to_user_id = assigned_to_user_id.clone(); + let tags = tags.clone(); + let linked_channel_ids = linked_channel_ids.clone(); + let metadata = metadata.clone(); + async move { + let result = service + .update_annotation( + annotation_id.clone(), + name, + description, + start_time_unix_nanos, + end_time_unix_nanos, + assigned_to_user_id, + state, + tags, + linked_channel_ids, + metadata, + unarchive, + ) + .await; + (annotation_id, result) + } + }, + ) + .await; + + annotations.reserve(updates.len()); + for (annotation_id, result) in updates { + match result { + Ok(annotation) => annotations.push(annotation), + Err(error) => failures.push(AnnotationUpdateFailure { + annotation_id, + error, + }), + } + } + } + + let mut batch_archive_error = None; + let mut archive_skipped = false; + if is_archived == Some(true) { + if failures.is_empty() { + match self.batch_archive_annotations(annotation_ids).await { + Ok(archived) => annotations = archived, + Err(error) => batch_archive_error = Some(error), + } + } else { + archive_skipped = true; + } + } + + Ok(UpdateAnnotationsResult { + annotations, + failures, + batch_archive_error, + archive_skipped, + }) + } + /// Update a subset of an existing annotation's fields. Per /// `protos/sift/annotations/v1/annotations.proto::UpdateAnnotationRequest` the /// updatable fields are `name`, `description`, `start_time`, `end_time`, diff --git a/rust/crates/sift_mcp/src/service/annotations/test.rs b/rust/crates/sift_mcp/src/service/annotations/test.rs index f71bdb3d6..294599617 100644 --- a/rust/crates/sift_mcp/src/service/annotations/test.rs +++ b/rust/crates/sift_mcp/src/service/annotations/test.rs @@ -1,12 +1,21 @@ +use std::{ + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, +}; + use sift_rs::annotations::v1::{ - Annotation, AnnotationState, AnnotationType, CreateAnnotationResponse, ListAnnotationsResponse, - UpdateAnnotationResponse, annotation_service_server::AnnotationServiceServer, + Annotation, AnnotationState, AnnotationType, BatchArchiveAnnotationsResponse, + CreateAnnotationResponse, ListAnnotationsResponse, UpdateAnnotationResponse, + annotation_service_server::AnnotationServiceServer, }; use sift_test_util::{grpc::memory_sift_channel, mock::annotations::v1::MockAnnotationServiceImpl}; -use tokio::task::JoinHandle; +use tokio::{sync::Semaphore, task::JoinHandle}; use tonic::{Response, Status, transport::Server}; -use super::AnnotationService; +use super::{AnnotationService, fan_out_bounded}; use crate::service::common::DEFAULT_LIMIT; async fn service_with_mock(mock: MockAnnotationServiceImpl) -> (AnnotationService, JoinHandle<()>) { @@ -27,6 +36,47 @@ async fn service_with_mock(mock: MockAnnotationServiceImpl) -> (AnnotationServic ) } +#[tokio::test] +async fn bounded_fan_out_limits_concurrency_and_preserves_order() { + let active = Arc::new(AtomicUsize::new(0)); + let max_active = Arc::new(AtomicUsize::new(0)); + let started = Arc::new(AtomicUsize::new(0)); + let gate = Arc::new(Semaphore::new(0)); + + let active_for_task = Arc::clone(&active); + let max_for_task = Arc::clone(&max_active); + let started_for_task = Arc::clone(&started); + let gate_for_task = Arc::clone(&gate); + let task = tokio::spawn(fan_out_bounded((0..120).collect(), 50, move |item| { + let active = Arc::clone(&active_for_task); + let max_active = Arc::clone(&max_for_task); + let started = Arc::clone(&started_for_task); + let gate = Arc::clone(&gate_for_task); + async move { + let current = active.fetch_add(1, Ordering::SeqCst) + 1; + max_active.fetch_max(current, Ordering::SeqCst); + started.fetch_add(1, Ordering::SeqCst); + gate.acquire().await.unwrap().forget(); + active.fetch_sub(1, Ordering::SeqCst); + item + } + })); + + tokio::time::timeout(Duration::from_secs(1), async { + while started.load(Ordering::SeqCst) < 50 { + tokio::task::yield_now().await; + } + }) + .await + .expect("50 updates should start concurrently"); + assert_eq!(started.load(Ordering::SeqCst), 50); + gate.add_permits(120); + let results = task.await.unwrap(); + + assert_eq!(results, (0..120).collect::>()); + assert_eq!(max_active.load(Ordering::SeqCst), 50); +} + #[tokio::test] async fn list_annotations_returns_single_page() { let mut mock = MockAnnotationServiceImpl::new(); @@ -275,6 +325,100 @@ async fn create_annotation_propagates_grpc_error() { assert!(err.to_string().contains("failed to create annotation")); } +#[tokio::test] +async fn batch_archive_annotations_forwards_ids() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_batch_archive_annotations() + .withf(|req| req.get_ref().annotation_ids == ["ann1", "ann2"]) + .returning(|req| { + let annotations = req + .into_inner() + .annotation_ids + .into_iter() + .map(|annotation_id| Annotation { + annotation_id, + is_archived: true, + ..Default::default() + }) + .collect(); + Ok(Response::new(BatchArchiveAnnotationsResponse { + annotations, + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let annotations = service + .batch_archive_annotations(vec!["ann1".into(), "ann2".into()]) + .await + .expect("batch archive failed"); + + assert_eq!(annotations.len(), 2); + assert!(annotations.iter().all(|annotation| annotation.is_archived)); +} + +#[tokio::test] +async fn batch_archive_annotations_propagates_grpc_error() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_batch_archive_annotations() + .returning(|_| Err(Status::not_found("no such annotation"))); + + let (service, _h) = service_with_mock(mock).await; + + let error = service + .batch_archive_annotations(vec!["ann1".into()]) + .await + .expect_err("expected batch archive error"); + + assert!( + error + .to_string() + .contains("failed to batch archive annotations") + ); +} + +#[tokio::test] +async fn update_annotations_collects_failures_and_continues() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_update_annotation().times(3).returning(|req| { + let annotation = req.into_inner().annotation.unwrap(); + if annotation.annotation_id == "ann2" { + return Err(Status::not_found("no such annotation")); + } + Ok(Response::new(UpdateAnnotationResponse { + annotation: Some(annotation), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let outcome = service + .update_annotations( + vec!["ann1".into(), "ann2".into(), "ann3".into()], + Some("renamed".into()), + None, + None, + None, + None, + None, + None, + None, + None, + None, + ) + .await + .expect("bulk update failed"); + + let updated_ids = outcome + .annotations + .iter() + .map(|annotation| annotation.annotation_id.as_str()) + .collect::>(); + assert_eq!(updated_ids, ["ann1", "ann3"]); + assert_eq!(outcome.failures.len(), 1); + assert_eq!(outcome.failures[0].annotation_id, "ann2"); +} + #[tokio::test] async fn update_annotation_builds_mask_from_provided_fields() { let mut mock = MockAnnotationServiceImpl::new(); diff --git a/rust/crates/sift_mcp/src/tool/annotations/mod.rs b/rust/crates/sift_mcp/src/tool/annotations/mod.rs index ef8283e4f..be9e8c9e5 100644 --- a/rust/crates/sift_mcp/src/tool/annotations/mod.rs +++ b/rust/crates/sift_mcp/src/tool/annotations/mod.rs @@ -312,8 +312,10 @@ impl SiftMcpServer { Other changes, including unarchiving, use one `UpdateAnnotation` request per annotation. Output: - - `{ \"annotations\": [Annotation, ...], \"next_step\": string }`. Each item is the post-update - state from the server and includes a `url` field with its Sift web link when the host can be derived. + - `{ \"annotations\": [Annotation, ...], \"failures\": [...], \"batch_archive_error\": object|null, + \"archive_skipped\": bool, \"next_step\": string }`. Successful annotations include a `url` field + when the host can be derived. Each individual failure includes `annotation_id`, MCP `code`, `message`, + and optional `data`. Partial failures set the tool result's `isError` flag. Parameters: - `annotation_ids`: required list of 1 to 1000 annotation ids. The same changes are applied to every @@ -337,14 +339,16 @@ impl SiftMcpServer { Errors: - `INVALID_PARAMS` if `annotation_ids` is empty, contains an empty id, exceeds 1000 ids, `state` is unrecognized, or no updatable field is set. - - `RESOURCE_NOT_FOUND` if no annotation matches one of the ids. - - `INTERNAL_ERROR` for upstream gRPC failures. + - Individual upstream failures are returned in `failures` next to successful results. Follow each + failure's guidance and retry only eligible failed ids. + - Batch-archive failures are returned in `batch_archive_error`. The archive outcome may be unknown. Guidance: - This is a write. CONFIRM every target and the full proposed values with the user before invoking — `tags`, `linked_channel_ids`, and `metadata` are REPLACE operations, not merges. - General field updates and unarchive operations issue one API request per annotation and are not - atomic. A failure can leave earlier annotations updated. Archive uses a single batch API request. + atomic. Up to 50 requests run concurrently. Archive uses a single batch API request after all + individual updates succeed; otherwise `archive_skipped` is true. - For appends, read the current annotation via `list_annotations` filtered by `annotation_id == \"\"`, then send the union. ", @@ -414,57 +418,95 @@ impl SiftMcpServer { let state = state.map(|s| parse_annotation_state(&s)).transpose()?; let metadata = metadata.map(|m| m.into_iter().map(MetadataValue::from).collect::>()); - let unarchive = (is_archived == Some(false)).then_some(false); - let has_individual_update = has_field_update || unarchive.is_some(); - - let mut annotations = Vec::new(); - if has_individual_update { - annotations.reserve(annotation_ids.len()); - for annotation_id in &annotation_ids { - let annotation = self - .annotation_service - .update_annotation( - annotation_id.clone(), - name.clone(), - description.clone(), - start_time_unix_nanos, - end_time_unix_nanos, - assigned_to_user_id.clone(), - state, - tags.clone(), - linked_channel_ids.clone(), - metadata.clone(), - unarchive, - ) - .await - .map_err(from_anyhow)?; - annotations.push(annotation); - } - } + let requested_ids = annotation_ids.clone(); + let requested_count = annotation_ids.len(); - if is_archived == Some(true) { - annotations = self - .annotation_service - .batch_archive_annotations(annotation_ids) - .await - .map_err(from_anyhow)?; - } + let outcome = self + .annotation_service + .update_annotations( + annotation_ids, + name, + description, + start_time_unix_nanos, + end_time_unix_nanos, + assigned_to_user_id, + state, + tags, + linked_channel_ids, + metadata, + is_archived, + ) + .await + .map_err(from_anyhow)?; - let updated_count = annotations.len(); + let updated_count = outcome.annotations.len(); + let failures = outcome + .failures + .into_iter() + .map(|failure| { + let error = from_anyhow(failure.error); + serde_json::json!({ + "annotation_id": failure.annotation_id, + "code": error.code.0, + "message": error.message, + "data": error.data, + }) + }) + .collect::>(); + let failure_count = failures.len(); + let batch_archive_error = outcome.batch_archive_error.map(|error| { + let error = from_anyhow(error); + serde_json::json!({ + "annotation_ids": requested_ids, + "code": error.code.0, + "message": error.message, + "data": error.data, + }) + }); + let has_errors = failure_count > 0 || batch_archive_error.is_some(); + + let annotations = outcome.annotations; let annotations = with_urls(&annotations, |annotation| { self.url_service .build_annotation_url(&annotation.annotation_id) .ok() })?; - let next_step = format!( - "Updated {updated_count} annotation(s). Surface the new states and links to the user and confirm the \ - changes match their intent. Remember: tags, linked channels, and metadata are REPLACE operations.", - ); + let next_step = if batch_archive_error.is_some() { + format!( + "Completed field updates for {updated_count} of {requested_count} annotation(s), but batch archive \ + failed. Archive state may be unknown. Verify the targets with `list_annotations` before retrying." + ) + } else if outcome.archive_skipped { + format!( + "Updated {updated_count} of {requested_count} annotation(s); {failure_count} failed. Archive was not \ + attempted because individual updates failed. Review `failures`, follow their guidance, and retry \ + only eligible failed ids." + ) + } else if failure_count > 0 { + format!( + "Updated {updated_count} of {requested_count} annotation(s); {failure_count} failed. Successful \ + annotations are already changed. Review `failures`, follow their guidance, and retry only eligible \ + failed ids." + ) + } else { + format!( + "Updated {updated_count} annotation(s). Surface the new states and links to the user and confirm the \ + changes match their intent. Remember: tags, linked channels, and metadata are REPLACE operations." + ) + }; - let mut result = CallToolResult::structured(serde_json::json!({ + let structured = serde_json::json!({ "annotations": annotations, + "failures": failures, + "batch_archive_error": batch_archive_error, + "archive_skipped": outcome.archive_skipped, "next_step": next_step, - })); + }); + let mut result = if has_errors { + CallToolResult::structured_error(structured) + } else { + CallToolResult::structured(structured) + }; result.content = vec![ContentBlock::text(next_step)]; Ok(result) } diff --git a/rust/crates/sift_mcp/src/tool/annotations/test.rs b/rust/crates/sift_mcp/src/tool/annotations/test.rs index cd0a7ff95..d4c90b111 100644 --- a/rust/crates/sift_mcp/src/tool/annotations/test.rs +++ b/rust/crates/sift_mcp/src/tool/annotations/test.rs @@ -13,7 +13,10 @@ use tokio::task::JoinHandle; use tonic::{Response, Status, transport::Server}; use super::{AnnotationListParams, CreateAnnotationParams, UpdateAnnotationParams}; -use crate::{server::SiftMcpServer, tool::common::test_support::structured_field}; +use crate::{ + server::SiftMcpServer, + tool::common::test_support::{structured, structured_field}, +}; async fn server_with_mock(mock: MockAnnotationServiceImpl) -> (SiftMcpServer, JoinHandle<()>) { let (client, server) = tokio::io::duplex(1024); @@ -258,6 +261,38 @@ async fn update_annotation_updates_each_annotation() { assert_eq!(annotations[1]["annotationId"], "ann2"); } +#[tokio::test] +async fn update_annotation_reports_partial_failures_and_continues() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_update_annotation().times(3).returning(|req| { + let annotation = req.into_inner().annotation.unwrap(); + if annotation.annotation_id == "ann2" { + return Err(Status::not_found("no such annotation")); + } + Ok(Response::new(UpdateAnnotationResponse { + annotation: Some(annotation), + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let mut params = update_params("ann1"); + params.annotation_ids.extend(["ann2".into(), "ann3".into()]); + params.name = Some("renamed".into()); + + let resp = server + .update_annotation(Parameters(params)) + .await + .expect("partial update result should be returned"); + + assert_eq!(resp.is_error, Some(true)); + let body = structured(resp); + assert_eq!(body["annotations"][0]["annotationId"], "ann1"); + assert_eq!(body["annotations"][1]["annotationId"], "ann3"); + assert_eq!(body["failures"][0]["annotation_id"], "ann2"); + assert_eq!(body["failures"][0]["code"], ErrorCode::RESOURCE_NOT_FOUND.0); +} + #[tokio::test] async fn update_annotation_batch_archives_annotations() { let mut mock = MockAnnotationServiceImpl::new(); @@ -344,6 +379,68 @@ async fn update_annotation_updates_fields_before_archiving() { .expect("update_annotation failed"); } +#[tokio::test] +async fn update_annotation_skips_archive_after_partial_field_failure() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_update_annotation().times(2).returning(|req| { + let annotation = req.into_inner().annotation.unwrap(); + if annotation.annotation_id == "ann2" { + return Err(Status::not_found("no such annotation")); + } + Ok(Response::new(UpdateAnnotationResponse { + annotation: Some(annotation), + })) + }); + mock.expect_batch_archive_annotations().times(0); + + let (server, _h) = server_with_mock(mock).await; + + let mut params = update_params("ann1"); + params.annotation_ids.push("ann2".into()); + params.name = Some("renamed".into()); + params.is_archived = Some(true); + + let resp = server + .update_annotation(Parameters(params)) + .await + .expect("partial update result should be returned"); + + assert_eq!(resp.is_error, Some(true)); + let body = structured(resp); + assert_eq!(body["archive_skipped"], true); + assert_eq!(body["annotations"][0]["annotationId"], "ann1"); + assert_eq!(body["failures"][0]["annotation_id"], "ann2"); +} + +#[tokio::test] +async fn update_annotation_reports_opaque_batch_archive_error() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_batch_archive_annotations() + .returning(|_| Err(Status::not_found("one target was not found"))); + + let (server, _h) = server_with_mock(mock).await; + + let mut params = update_params("ann1"); + params.annotation_ids.push("ann2".into()); + params.is_archived = Some(true); + + let resp = server + .update_annotation(Parameters(params)) + .await + .expect("batch archive failure details should be returned"); + + assert_eq!(resp.is_error, Some(true)); + let body = structured(resp); + assert_eq!( + body["batch_archive_error"]["annotation_ids"], + serde_json::json!(["ann1", "ann2"]) + ); + assert_eq!( + body["batch_archive_error"]["code"], + ErrorCode::RESOURCE_NOT_FOUND.0 + ); +} + #[tokio::test] async fn update_annotation_unarchives_each_annotation() { let mut mock = MockAnnotationServiceImpl::new(); @@ -426,10 +523,13 @@ async fn update_annotation_propagates_grpc_error() { let mut params = update_params("ann1"); params.name = Some("x".into()); - let err = server + let resp = server .update_annotation(Parameters(params)) .await - .expect_err("expected error"); + .expect("failure details should be returned"); - assert_eq!(err.code, ErrorCode::RESOURCE_NOT_FOUND); + assert_eq!(resp.is_error, Some(true)); + let body = structured(resp); + assert_eq!(body["failures"][0]["annotation_id"], "ann1"); + assert_eq!(body["failures"][0]["code"], ErrorCode::RESOURCE_NOT_FOUND.0); } From 064451c0d3edbafc743a3d5f7727897f3e172277 Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Sat, 22 Aug 2026 09:46:29 -0700 Subject: [PATCH 3/7] fix: harden bulk annotation failure handling --- .../sift_mcp/src/service/annotations/mod.rs | 122 ++++++++++++------ .../sift_mcp/src/service/annotations/test.rs | 33 +++++ .../sift_mcp/src/tool/annotations/mod.rs | 45 +++++-- .../sift_mcp/src/tool/annotations/test.rs | 35 +++++ 4 files changed, 183 insertions(+), 52 deletions(-) diff --git a/rust/crates/sift_mcp/src/service/annotations/mod.rs b/rust/crates/sift_mcp/src/service/annotations/mod.rs index ff54f0fc0..fd92e7299 100644 --- a/rust/crates/sift_mcp/src/service/annotations/mod.rs +++ b/rust/crates/sift_mcp/src/service/annotations/mod.rs @@ -13,6 +13,7 @@ use sift_rs::{ }, metadata::v1::MetadataValue, }; +use tonic::Code; #[cfg(test)] mod test; @@ -29,6 +30,7 @@ pub struct AnnotationUpdateFailure { pub struct UpdateAnnotationsResult { pub annotations: Vec, pub failures: Vec, + pub not_attempted: Vec, pub batch_archive_error: Option, pub archive_skipped: bool, } @@ -50,6 +52,21 @@ where results.into_iter().map(|(_, result)| result).collect() } +fn is_backend_wide_failure(error: &Error) -> bool { + error.downcast_ref::().is_some_and(|status| { + matches!( + status.code(), + Code::ResourceExhausted + | Code::Unavailable + | Code::DeadlineExceeded + | Code::Internal + | Code::Unauthenticated + | Code::PermissionDenied + | Code::Cancelled + ) + }) +} + /// Build a protobuf `Timestamp` from Unix nanoseconds via the shared helper. fn timestamp_from_unix_nanos(nanos: i64) -> Timestamp { let (seconds, nanos) = common::unix_nanos_to_secs_and_subsec_nanos(nanos); @@ -261,50 +278,70 @@ impl AnnotationService { let mut annotations = Vec::new(); let mut failures = Vec::new(); + let mut not_attempted = Vec::new(); if has_individual_update { - let service = self.clone(); - let updates = fan_out_bounded( - annotation_ids.clone(), - MAX_CONCURRENT_ANNOTATION_UPDATES, - move |annotation_id| { - let service = service.clone(); - let name = name.clone(); - let description = description.clone(); - let assigned_to_user_id = assigned_to_user_id.clone(); - let tags = tags.clone(); - let linked_channel_ids = linked_channel_ids.clone(); - let metadata = metadata.clone(); - async move { - let result = service - .update_annotation( - annotation_id.clone(), - name, - description, - start_time_unix_nanos, - end_time_unix_nanos, - assigned_to_user_id, - state, - tags, - linked_channel_ids, - metadata, - unarchive, - ) - .await; - (annotation_id, result) + annotations.reserve(annotation_ids.len()); + for chunk_start in (0..annotation_ids.len()).step_by(MAX_CONCURRENT_ANNOTATION_UPDATES) + { + let chunk_end = + (chunk_start + MAX_CONCURRENT_ANNOTATION_UPDATES).min(annotation_ids.len()); + let service = self.clone(); + let name = name.clone(); + let description = description.clone(); + let assigned_to_user_id = assigned_to_user_id.clone(); + let tags = tags.clone(); + let linked_channel_ids = linked_channel_ids.clone(); + let metadata = metadata.clone(); + let updates = fan_out_bounded( + annotation_ids[chunk_start..chunk_end].to_vec(), + MAX_CONCURRENT_ANNOTATION_UPDATES, + move |annotation_id| { + let service = service.clone(); + let name = name.clone(); + let description = description.clone(); + let assigned_to_user_id = assigned_to_user_id.clone(); + let tags = tags.clone(); + let linked_channel_ids = linked_channel_ids.clone(); + let metadata = metadata.clone(); + async move { + let result = service + .update_annotation( + annotation_id.clone(), + name, + description, + start_time_unix_nanos, + end_time_unix_nanos, + assigned_to_user_id, + state, + tags, + linked_channel_ids, + metadata, + unarchive, + ) + .await; + (annotation_id, result) + } + }, + ) + .await; + let stop_after_batch = updates + .iter() + .any(|(_, result)| result.as_ref().is_err_and(is_backend_wide_failure)); + + for (annotation_id, result) in updates { + match result { + Ok(annotation) => annotations.push(annotation), + Err(error) => failures.push(AnnotationUpdateFailure { + annotation_id, + error, + }), } - }, - ) - .await; - - annotations.reserve(updates.len()); - for (annotation_id, result) in updates { - match result { - Ok(annotation) => annotations.push(annotation), - Err(error) => failures.push(AnnotationUpdateFailure { - annotation_id, - error, - }), + } + + if stop_after_batch { + not_attempted.extend_from_slice(&annotation_ids[chunk_end..]); + break; } } } @@ -312,7 +349,7 @@ impl AnnotationService { let mut batch_archive_error = None; let mut archive_skipped = false; if is_archived == Some(true) { - if failures.is_empty() { + if failures.is_empty() && not_attempted.is_empty() { match self.batch_archive_annotations(annotation_ids).await { Ok(archived) => annotations = archived, Err(error) => batch_archive_error = Some(error), @@ -325,6 +362,7 @@ impl AnnotationService { Ok(UpdateAnnotationsResult { annotations, failures, + not_attempted, batch_archive_error, archive_skipped, }) diff --git a/rust/crates/sift_mcp/src/service/annotations/test.rs b/rust/crates/sift_mcp/src/service/annotations/test.rs index 294599617..c5050d409 100644 --- a/rust/crates/sift_mcp/src/service/annotations/test.rs +++ b/rust/crates/sift_mcp/src/service/annotations/test.rs @@ -419,6 +419,39 @@ async fn update_annotations_collects_failures_and_continues() { assert_eq!(outcome.failures[0].annotation_id, "ann2"); } +#[tokio::test] +async fn update_annotations_stops_after_backend_wide_failure() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_update_annotation() + .times(50) + .returning(|_| Err(Status::resource_exhausted("slow down"))); + + let (service, _h) = service_with_mock(mock).await; + let annotation_ids = (0..60) + .map(|index| format!("ann{index}")) + .collect::>(); + + let outcome = service + .update_annotations( + annotation_ids.clone(), + Some("renamed".into()), + None, + None, + None, + None, + None, + None, + None, + None, + None, + ) + .await + .expect("bulk update failed"); + + assert_eq!(outcome.failures.len(), 50); + assert_eq!(outcome.not_attempted, annotation_ids[50..]); +} + #[tokio::test] async fn update_annotation_builds_mask_from_provided_fields() { let mut mock = MockAnnotationServiceImpl::new(); diff --git a/rust/crates/sift_mcp/src/tool/annotations/mod.rs b/rust/crates/sift_mcp/src/tool/annotations/mod.rs index be9e8c9e5..97cad9dbe 100644 --- a/rust/crates/sift_mcp/src/tool/annotations/mod.rs +++ b/rust/crates/sift_mcp/src/tool/annotations/mod.rs @@ -313,9 +313,10 @@ impl SiftMcpServer { Output: - `{ \"annotations\": [Annotation, ...], \"failures\": [...], \"batch_archive_error\": object|null, - \"archive_skipped\": bool, \"next_step\": string }`. Successful annotations include a `url` field - when the host can be derived. Each individual failure includes `annotation_id`, MCP `code`, `message`, - and optional `data`. Partial failures set the tool result's `isError` flag. + \"not_attempted\": [string, ...], \"archive_skipped\": bool, \"next_step\": string }`. Successful + annotations include a `url` field when the host can be derived. Each individual failure includes + `annotation_id`, MCP `code`, `message`, and optional `data`. Partial failures set the tool result's + `isError` flag. Parameters: - `annotation_ids`: required list of 1 to 1000 annotation ids. The same changes are applied to every @@ -341,14 +342,16 @@ impl SiftMcpServer { unrecognized, or no updatable field is set. - Individual upstream failures are returned in `failures` next to successful results. Follow each failure's guidance and retry only eligible failed ids. + - Backend-wide failures stop later batches. Their ids are returned in `not_attempted` without an API + request. - Batch-archive failures are returned in `batch_archive_error`. The archive outcome may be unknown. Guidance: - This is a write. CONFIRM every target and the full proposed values with the user before invoking — `tags`, `linked_channel_ids`, and `metadata` are REPLACE operations, not merges. - General field updates and unarchive operations issue one API request per annotation and are not - atomic. Up to 50 requests run concurrently. Archive uses a single batch API request after all - individual updates succeed; otherwise `archive_skipped` is true. + atomic. Requests run in batches of up to 50. A backend-wide failure stops later batches. Archive uses + a single batch API request after all individual updates succeed; otherwise `archive_skipped` is true. - For appends, read the current annotation via `list_annotations` filtered by `annotation_id == \"\"`, then send the union. ", @@ -454,6 +457,8 @@ impl SiftMcpServer { }) .collect::>(); let failure_count = failures.len(); + let not_attempted = outcome.not_attempted; + let not_attempted_count = not_attempted.len(); let batch_archive_error = outcome.batch_archive_error.map(|error| { let error = from_anyhow(error); serde_json::json!({ @@ -463,7 +468,8 @@ impl SiftMcpServer { "data": error.data, }) }); - let has_errors = failure_count > 0 || batch_archive_error.is_some(); + let has_errors = + failure_count > 0 || not_attempted_count > 0 || batch_archive_error.is_some(); let annotations = outcome.annotations; let annotations = with_urls(&annotations, |annotation| { @@ -477,10 +483,24 @@ impl SiftMcpServer { failed. Archive state may be unknown. Verify the targets with `list_annotations` before retrying." ) } else if outcome.archive_skipped { + if not_attempted_count > 0 { + format!( + "Updated {updated_count} of {requested_count} annotation(s); {failure_count} failed and \ + {not_attempted_count} were not attempted after a backend-wide failure. Archive was not \ + attempted. Review `failures` before retrying eligible failed and not-attempted ids." + ) + } else { + format!( + "Updated {updated_count} of {requested_count} annotation(s); {failure_count} failed. Archive was \ + not attempted because individual updates failed. Review `failures`, follow their guidance, and \ + retry only eligible failed ids." + ) + } + } else if not_attempted_count > 0 { format!( - "Updated {updated_count} of {requested_count} annotation(s); {failure_count} failed. Archive was not \ - attempted because individual updates failed. Review `failures`, follow their guidance, and retry \ - only eligible failed ids." + "Updated {updated_count} of {requested_count} annotation(s); {failure_count} failed and \ + {not_attempted_count} were not attempted after a backend-wide failure. Review `failures` before \ + retrying eligible failed and not-attempted ids." ) } else if failure_count > 0 { format!( @@ -498,6 +518,7 @@ impl SiftMcpServer { let structured = serde_json::json!({ "annotations": annotations, "failures": failures, + "not_attempted": not_attempted, "batch_archive_error": batch_archive_error, "archive_skipped": outcome.archive_skipped, "next_step": next_step, @@ -507,7 +528,11 @@ impl SiftMcpServer { } else { CallToolResult::structured(structured) }; - result.content = vec![ContentBlock::text(next_step)]; + if has_errors { + result.content.push(ContentBlock::text(next_step)); + } else { + result.content = vec![ContentBlock::text(next_step)]; + } Ok(result) } } diff --git a/rust/crates/sift_mcp/src/tool/annotations/test.rs b/rust/crates/sift_mcp/src/tool/annotations/test.rs index d4c90b111..1b06d1db3 100644 --- a/rust/crates/sift_mcp/src/tool/annotations/test.rs +++ b/rust/crates/sift_mcp/src/tool/annotations/test.rs @@ -286,11 +286,46 @@ async fn update_annotation_reports_partial_failures_and_continues() { .expect("partial update result should be returned"); assert_eq!(resp.is_error, Some(true)); + let content = serde_json::to_string(&resp.content).expect("content should serialize"); + assert!(content.contains("ann2")); + assert!(content.contains("no such annotation")); let body = structured(resp); assert_eq!(body["annotations"][0]["annotationId"], "ann1"); assert_eq!(body["annotations"][1]["annotationId"], "ann3"); assert_eq!(body["failures"][0]["annotation_id"], "ann2"); assert_eq!(body["failures"][0]["code"], ErrorCode::RESOURCE_NOT_FOUND.0); + assert_eq!(body["not_attempted"], serde_json::json!([])); +} + +#[tokio::test] +async fn update_annotation_reports_ids_not_attempted_after_backend_wide_failure() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_update_annotation() + .times(50) + .returning(|_| Err(Status::resource_exhausted("slow down"))); + + let (server, _h) = server_with_mock(mock).await; + + let mut params = update_params("ann0"); + params + .annotation_ids + .extend((1..60).map(|index| format!("ann{index}"))); + params.name = Some("renamed".into()); + + let resp = server + .update_annotation(Parameters(params)) + .await + .expect("partial update result should be returned"); + + assert_eq!(resp.is_error, Some(true)); + let content = serde_json::to_string(&resp.content).expect("content should serialize"); + assert!(content.contains("ann0")); + assert!(content.contains("ann59")); + let body = structured(resp); + assert_eq!(body["failures"].as_array().unwrap().len(), 50); + assert_eq!(body["not_attempted"].as_array().unwrap().len(), 10); + assert_eq!(body["not_attempted"][0], "ann50"); + assert_eq!(body["not_attempted"][9], "ann59"); } #[tokio::test] From 85ee24a9bf97b0d99888e55604181329f8e42b6e Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Wed, 26 Aug 2026 01:14:27 -0700 Subject: [PATCH 4/7] fix: unarchive annotations through the batch RPC and stop double-emitting error payloads --- .../sift_mcp/src/service/annotations/mod.rs | 53 ++++++--- .../sift_mcp/src/service/annotations/test.rs | 2 - .../sift_mcp/src/tool/annotations/mod.rs | 28 ++--- .../sift_mcp/src/tool/annotations/test.rs | 112 +++++++++++++++--- 4 files changed, 144 insertions(+), 51 deletions(-) diff --git a/rust/crates/sift_mcp/src/service/annotations/mod.rs b/rust/crates/sift_mcp/src/service/annotations/mod.rs index fd92e7299..7565147b2 100644 --- a/rust/crates/sift_mcp/src/service/annotations/mod.rs +++ b/rust/crates/sift_mcp/src/service/annotations/mod.rs @@ -7,9 +7,9 @@ use sift_rs::{ SiftChannel, annotations::v1::{ Annotation, AnnotationLinkedChannel, AnnotationState, AnnotationType, - BatchArchiveAnnotationsRequest, CreateAnnotationRequest, ListAnnotationsRequest, - ListAnnotationsResponse, UpdateAnnotationRequest, annotation_linked_channel, - annotation_service_client::AnnotationServiceClient, + BatchArchiveAnnotationsRequest, BatchUnarchiveAnnotationsRequest, CreateAnnotationRequest, + ListAnnotationsRequest, ListAnnotationsResponse, UpdateAnnotationRequest, + annotation_linked_channel, annotation_service_client::AnnotationServiceClient, }, metadata::v1::MetadataValue, }; @@ -249,6 +249,30 @@ impl AnnotationService { Ok(resp.annotations) } + pub async fn batch_unarchive_annotations( + &self, + annotation_ids: Vec, + ) -> Result> { + let channel = self.channel.clone(); + let resp = with_retry(&self.policy, move || { + let channel = channel.clone(); + let annotation_ids = annotation_ids.clone(); + async move { + let mut client = AnnotationServiceClient::new(channel); + client + .batch_unarchive_annotations(BatchUnarchiveAnnotationsRequest { + annotation_ids, + }) + .await + .map(|resp| resp.into_inner()) + } + }) + .await + .context("failed to batch unarchive annotations")?; + + Ok(resp.annotations) + } + #[allow(clippy::too_many_arguments)] pub async fn update_annotations( &self, @@ -273,14 +297,11 @@ impl AnnotationService { || tags.is_some() || linked_channel_ids.is_some() || metadata.is_some(); - let unarchive = (is_archived == Some(false)).then_some(false); - let has_individual_update = has_field_update || unarchive.is_some(); - let mut annotations = Vec::new(); let mut failures = Vec::new(); let mut not_attempted = Vec::new(); - if has_individual_update { + if has_field_update { annotations.reserve(annotation_ids.len()); for chunk_start in (0..annotation_ids.len()).step_by(MAX_CONCURRENT_ANNOTATION_UPDATES) { @@ -317,7 +338,6 @@ impl AnnotationService { tags, linked_channel_ids, metadata, - unarchive, ) .await; (annotation_id, result) @@ -348,9 +368,14 @@ impl AnnotationService { let mut batch_archive_error = None; let mut archive_skipped = false; - if is_archived == Some(true) { + if is_archived.is_some() { if failures.is_empty() && not_attempted.is_empty() { - match self.batch_archive_annotations(annotation_ids).await { + let result = if is_archived == Some(true) { + self.batch_archive_annotations(annotation_ids).await + } else { + self.batch_unarchive_annotations(annotation_ids).await + }; + match result { Ok(archived) => annotations = archived, Err(error) => batch_archive_error = Some(error), } @@ -372,7 +397,7 @@ impl AnnotationService { /// `protos/sift/annotations/v1/annotations.proto::UpdateAnnotationRequest` the /// updatable fields are `name`, `description`, `start_time`, `end_time`, /// `assigned_to_user_id`, `state`, `tags`, `legend_config`, `linked_channels`, - /// `metadata`, and `is_archived`. This service exposes all but `legend_config`. + /// and `metadata`. This service exposes all but `legend_config`. /// /// `tags`, `linked_channels`, and `metadata` use REPLACE semantics — passing /// `Some(vec![])` clears the field. @@ -389,7 +414,6 @@ impl AnnotationService { tags: Option>, linked_channel_ids: Option>, metadata: Option>, - is_archived: Option, ) -> Result { let mut annotation = Annotation { annotation_id, @@ -433,11 +457,6 @@ impl AnnotationService { annotation.metadata = v; paths.push("metadata".to_string()); } - if let Some(v) = is_archived { - annotation.is_archived = v; - paths.push("is_archived".to_string()); - } - let channel = self.channel.clone(); let resp = with_retry(&self.policy, move || { let channel = channel.clone(); diff --git a/rust/crates/sift_mcp/src/service/annotations/test.rs b/rust/crates/sift_mcp/src/service/annotations/test.rs index c5050d409..743d3b250 100644 --- a/rust/crates/sift_mcp/src/service/annotations/test.rs +++ b/rust/crates/sift_mcp/src/service/annotations/test.rs @@ -488,7 +488,6 @@ async fn update_annotation_builds_mask_from_provided_fields() { Some(vec!["important".to_string()]), None, None, - None, ) .await .expect("update_annotation failed"); @@ -516,7 +515,6 @@ async fn update_annotation_propagates_grpc_error() { None, None, None, - None, ) .await .expect_err("expected error"); diff --git a/rust/crates/sift_mcp/src/tool/annotations/mod.rs b/rust/crates/sift_mcp/src/tool/annotations/mod.rs index 97cad9dbe..564929729 100644 --- a/rust/crates/sift_mcp/src/tool/annotations/mod.rs +++ b/rust/crates/sift_mcp/src/tool/annotations/mod.rs @@ -308,8 +308,8 @@ impl SiftMcpServer { #[tool( name = "update_annotation", description = " - Update one or more existing annotations. Uses `annotations/v1 BatchArchiveAnnotations` when archiving. - Other changes, including unarchiving, use one `UpdateAnnotation` request per annotation. + Update one or more existing annotations. Uses `annotations/v1 BatchArchiveAnnotations` when archiving + and `BatchUnarchiveAnnotations` when unarchiving. Output: - `{ \"annotations\": [Annotation, ...], \"failures\": [...], \"batch_archive_error\": object|null, @@ -320,7 +320,8 @@ impl SiftMcpServer { Parameters: - `annotation_ids`: required list of 1 to 1000 annotation ids. The same changes are applied to every - annotation. + annotation. This replaces the former `annotation_id` parameter; pass a single annotation as a + one-element list. - `name`: optional new name. - `description`: optional new description. - `start_time_unix_nanos` / `end_time_unix_nanos`: optional new time bounds in Unix nanoseconds. @@ -331,9 +332,9 @@ impl SiftMcpServer { Pass `[]` to clear. Bit-field and calculated-channel links are not exposed here. - `metadata`: optional; REPLACES the full metadata list. Each entry is `{ \"name\": \"\", \"value\": }`. Pass `[]` to clear. - - `is_archived`: optional archive state. `true` uses one batch-archive request. `false` is included in - each annotation's individual update request. When `true` is combined with other fields, annotations - are updated before archiving. + - `is_archived`: optional archive state. `true` uses one batch-archive request; `false` uses one + batch-unarchive request. When combined with other fields, annotations are updated before their + archive state changes. At least one updatable field must be set; otherwise the tool returns `INVALID_PARAMS`. @@ -344,14 +345,15 @@ impl SiftMcpServer { failure's guidance and retry only eligible failed ids. - Backend-wide failures stop later batches. Their ids are returned in `not_attempted` without an API request. - - Batch-archive failures are returned in `batch_archive_error`. The archive outcome may be unknown. + - Batch archive or unarchive failures are returned in `batch_archive_error`. The archive outcome may + be unknown. Guidance: - This is a write. CONFIRM every target and the full proposed values with the user before invoking — `tags`, `linked_channel_ids`, and `metadata` are REPLACE operations, not merges. - - General field updates and unarchive operations issue one API request per annotation and are not - atomic. Requests run in batches of up to 50. A backend-wide failure stops later batches. Archive uses - a single batch API request after all individual updates succeed; otherwise `archive_skipped` is true. + - General field updates issue one API request per annotation and are not atomic. Requests run in + batches of up to 50. A backend-wide failure stops later batches. Archive and unarchive use a single + batch API request after all individual updates succeed; otherwise `archive_skipped` is true. - For appends, read the current annotation via `list_annotations` filtered by `annotation_id == \"\"`, then send the union. ", @@ -528,11 +530,7 @@ impl SiftMcpServer { } else { CallToolResult::structured(structured) }; - if has_errors { - result.content.push(ContentBlock::text(next_step)); - } else { - result.content = vec![ContentBlock::text(next_step)]; - } + result.content = vec![ContentBlock::text(next_step)]; Ok(result) } } diff --git a/rust/crates/sift_mcp/src/tool/annotations/test.rs b/rust/crates/sift_mcp/src/tool/annotations/test.rs index 1b06d1db3..3d160ebb3 100644 --- a/rust/crates/sift_mcp/src/tool/annotations/test.rs +++ b/rust/crates/sift_mcp/src/tool/annotations/test.rs @@ -5,8 +5,9 @@ use std::sync::{ use rmcp::{handler::server::wrapper::Parameters, model::ErrorCode}; use sift_rs::annotations::v1::{ - Annotation, BatchArchiveAnnotationsResponse, CreateAnnotationResponse, ListAnnotationsResponse, - UpdateAnnotationResponse, annotation_service_server::AnnotationServiceServer, + Annotation, BatchArchiveAnnotationsResponse, BatchUnarchiveAnnotationsResponse, + CreateAnnotationResponse, ListAnnotationsResponse, UpdateAnnotationResponse, + annotation_service_server::AnnotationServiceServer, }; use sift_test_util::{grpc::memory_sift_channel, mock::annotations::v1::MockAnnotationServiceImpl}; use tokio::task::JoinHandle; @@ -287,8 +288,9 @@ async fn update_annotation_reports_partial_failures_and_continues() { assert_eq!(resp.is_error, Some(true)); let content = serde_json::to_string(&resp.content).expect("content should serialize"); - assert!(content.contains("ann2")); - assert!(content.contains("no such annotation")); + assert_eq!(resp.content.len(), 1); + assert!(!content.contains("ann2")); + assert!(!content.contains("no such annotation")); let body = structured(resp); assert_eq!(body["annotations"][0]["annotationId"], "ann1"); assert_eq!(body["annotations"][1]["annotationId"], "ann3"); @@ -319,8 +321,9 @@ async fn update_annotation_reports_ids_not_attempted_after_backend_wide_failure( assert_eq!(resp.is_error, Some(true)); let content = serde_json::to_string(&resp.content).expect("content should serialize"); - assert!(content.contains("ann0")); - assert!(content.contains("ann59")); + assert_eq!(resp.content.len(), 1); + assert!(!content.contains("ann0")); + assert!(!content.contains("ann59")); let body = structured(resp); assert_eq!(body["failures"].as_array().unwrap().len(), 50); assert_eq!(body["not_attempted"].as_array().unwrap().len(), 10); @@ -477,35 +480,88 @@ async fn update_annotation_reports_opaque_batch_archive_error() { } #[tokio::test] -async fn update_annotation_unarchives_each_annotation() { +async fn update_annotation_batch_unarchives_annotations() { let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_batch_unarchive_annotations() + .withf(|req| req.get_ref().annotation_ids == ["ann1", "ann2"]) + .returning(|_| { + Ok(Response::new(BatchUnarchiveAnnotationsResponse { + annotations: vec![ + Annotation { + annotation_id: "ann1".into(), + is_archived: false, + ..Default::default() + }, + Annotation { + annotation_id: "ann2".into(), + is_archived: false, + ..Default::default() + }, + ], + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let mut params = update_params("ann1"); + params.annotation_ids.push("ann2".into()); + params.is_archived = Some(false); + + let resp = server + .update_annotation(Parameters(params)) + .await + .expect("update_annotation failed"); + + let annotations = structured_field(resp, "annotations"); + assert_eq!(annotations.as_array().unwrap().len(), 2); +} + +#[tokio::test] +async fn update_annotation_updates_fields_before_unarchiving() { + let mut mock = MockAnnotationServiceImpl::new(); + let updated_count = Arc::new(AtomicUsize::new(0)); + + let update_count = Arc::clone(&updated_count); mock.expect_update_annotation() .times(2) - .withf(|req| { - let req = req.get_ref(); - !req.annotation.as_ref().unwrap().is_archived - && req.update_mask.as_ref().unwrap().paths == ["is_archived"] - }) - .returning(|req| { + .returning(move |req| { + update_count.fetch_add(1, Ordering::SeqCst); let annotation = req.into_inner().annotation.unwrap(); Ok(Response::new(UpdateAnnotationResponse { annotation: Some(annotation), })) }); + let unarchive_count = Arc::clone(&updated_count); + mock.expect_batch_unarchive_annotations() + .withf(move |_| unarchive_count.load(Ordering::SeqCst) == 2) + .returning(|req| { + let annotations = req + .into_inner() + .annotation_ids + .into_iter() + .map(|annotation_id| Annotation { + annotation_id, + is_archived: false, + ..Default::default() + }) + .collect(); + Ok(Response::new(BatchUnarchiveAnnotationsResponse { + annotations, + })) + }); + let (server, _h) = server_with_mock(mock).await; let mut params = update_params("ann1"); params.annotation_ids.push("ann2".into()); + params.name = Some("renamed".into()); params.is_archived = Some(false); - let resp = server + server .update_annotation(Parameters(params)) .await .expect("update_annotation failed"); - - let annotations = structured_field(resp, "annotations"); - assert_eq!(annotations.as_array().unwrap().len(), 2); } #[tokio::test] @@ -523,6 +579,28 @@ async fn update_annotation_rejects_empty_ids() { assert_eq!(err.code, ErrorCode::INVALID_PARAMS); } +#[tokio::test] +async fn update_annotation_rejects_more_than_1000_ids_without_requests() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_update_annotation().times(0); + mock.expect_batch_archive_annotations().times(0); + + let (server, _h) = server_with_mock(mock).await; + + let mut params = update_params("ann0"); + params + .annotation_ids + .extend((1..=1_000).map(|index| format!("ann{index}"))); + params.name = Some("renamed".into()); + + let err = server + .update_annotation(Parameters(params)) + .await + .expect_err("expected error"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + #[tokio::test] async fn update_annotation_rejects_empty_id() { let (server, _h) = server_with_mock(MockAnnotationServiceImpl::new()).await; From eb59e6ff1a611f76cd0142aa547750195fb8f9c1 Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Wed, 26 Aug 2026 10:20:18 -0700 Subject: [PATCH 5/7] fix: unarchive annotations per id with bounded fan-out --- .../sift_mcp/src/service/annotations/mod.rs | 97 ++++++++----- .../sift_mcp/src/tool/annotations/mod.rs | 51 +++---- .../sift_mcp/src/tool/annotations/test.rs | 134 +++++++++++++----- 3 files changed, 193 insertions(+), 89 deletions(-) diff --git a/rust/crates/sift_mcp/src/service/annotations/mod.rs b/rust/crates/sift_mcp/src/service/annotations/mod.rs index 7565147b2..dcacb9ad8 100644 --- a/rust/crates/sift_mcp/src/service/annotations/mod.rs +++ b/rust/crates/sift_mcp/src/service/annotations/mod.rs @@ -7,8 +7,8 @@ use sift_rs::{ SiftChannel, annotations::v1::{ Annotation, AnnotationLinkedChannel, AnnotationState, AnnotationType, - BatchArchiveAnnotationsRequest, BatchUnarchiveAnnotationsRequest, CreateAnnotationRequest, - ListAnnotationsRequest, ListAnnotationsResponse, UpdateAnnotationRequest, + BatchArchiveAnnotationsRequest, CreateAnnotationRequest, ListAnnotationsRequest, + ListAnnotationsResponse, UnarchiveAnnotationRequest, UpdateAnnotationRequest, annotation_linked_channel, annotation_service_client::AnnotationServiceClient, }, metadata::v1::MetadataValue, @@ -249,30 +249,6 @@ impl AnnotationService { Ok(resp.annotations) } - pub async fn batch_unarchive_annotations( - &self, - annotation_ids: Vec, - ) -> Result> { - let channel = self.channel.clone(); - let resp = with_retry(&self.policy, move || { - let channel = channel.clone(); - let annotation_ids = annotation_ids.clone(); - async move { - let mut client = AnnotationServiceClient::new(channel); - client - .batch_unarchive_annotations(BatchUnarchiveAnnotationsRequest { - annotation_ids, - }) - .await - .map(|resp| resp.into_inner()) - } - }) - .await - .context("failed to batch unarchive annotations")?; - - Ok(resp.annotations) - } - #[allow(clippy::too_many_arguments)] pub async fn update_annotations( &self, @@ -368,13 +344,9 @@ impl AnnotationService { let mut batch_archive_error = None; let mut archive_skipped = false; - if is_archived.is_some() { + if is_archived == Some(true) { if failures.is_empty() && not_attempted.is_empty() { - let result = if is_archived == Some(true) { - self.batch_archive_annotations(annotation_ids).await - } else { - self.batch_unarchive_annotations(annotation_ids).await - }; + let result = self.batch_archive_annotations(annotation_ids).await; match result { Ok(archived) => annotations = archived, Err(error) => batch_archive_error = Some(error), @@ -382,6 +354,47 @@ impl AnnotationService { } else { archive_skipped = true; } + } else if is_archived == Some(false) && failures.is_empty() && not_attempted.is_empty() { + annotations.clear(); + annotations.reserve(annotation_ids.len()); + for chunk_start in (0..annotation_ids.len()).step_by(MAX_CONCURRENT_ANNOTATION_UPDATES) + { + let chunk_end = + (chunk_start + MAX_CONCURRENT_ANNOTATION_UPDATES).min(annotation_ids.len()); + let service = self.clone(); + let unarchives = fan_out_bounded( + annotation_ids[chunk_start..chunk_end].to_vec(), + MAX_CONCURRENT_ANNOTATION_UPDATES, + move |annotation_id| { + let service = service.clone(); + async move { + let result = service.unarchive_annotation(annotation_id.clone()).await; + (annotation_id, result) + } + }, + ) + .await; + let stop_after_batch = unarchives + .iter() + .any(|(_, result)| result.as_ref().is_err_and(is_backend_wide_failure)); + + for (annotation_id, result) in unarchives { + match result { + Ok(annotation) => annotations.push(annotation), + Err(error) => failures.push(AnnotationUpdateFailure { + annotation_id, + error, + }), + } + } + + if stop_after_batch { + not_attempted.extend_from_slice(&annotation_ids[chunk_end..]); + break; + } + } + } else if is_archived == Some(false) { + archive_skipped = true; } Ok(UpdateAnnotationsResult { @@ -479,4 +492,24 @@ impl AnnotationService { resp.annotation .ok_or_else(|| anyhow!("update_annotation response missing annotation")) } + + pub async fn unarchive_annotation(&self, annotation_id: String) -> Result { + let channel = self.channel.clone(); + let resp = with_retry(&self.policy, move || { + let channel = channel.clone(); + let annotation_id = annotation_id.clone(); + async move { + let mut client = AnnotationServiceClient::new(channel); + client + .unarchive_annotation(UnarchiveAnnotationRequest { annotation_id }) + .await + .map(|resp| resp.into_inner()) + } + }) + .await + .context("failed to unarchive annotation")?; + + resp.annotation + .ok_or_else(|| anyhow!("unarchive_annotation response missing annotation")) + } } diff --git a/rust/crates/sift_mcp/src/tool/annotations/mod.rs b/rust/crates/sift_mcp/src/tool/annotations/mod.rs index 564929729..0b967b03b 100644 --- a/rust/crates/sift_mcp/src/tool/annotations/mod.rs +++ b/rust/crates/sift_mcp/src/tool/annotations/mod.rs @@ -22,6 +22,14 @@ mod test; const MAX_UPDATE_ANNOTATIONS: usize = 1_000; +fn upstream_error_message(error: &anyhow::Error) -> String { + if let Some(status) = error.downcast_ref::() { + format!("{}: {}", status.code(), status.message()) + } else { + error.to_string() + } +} + #[derive(Debug, Deserialize, JsonSchema)] pub struct AnnotationListParams { pub(crate) filter: String, @@ -309,14 +317,16 @@ impl SiftMcpServer { name = "update_annotation", description = " Update one or more existing annotations. Uses `annotations/v1 BatchArchiveAnnotations` when archiving - and `BatchUnarchiveAnnotations` when unarchiving. + and one `UnarchiveAnnotation` request per annotation when unarchiving. Output: - `{ \"annotations\": [Annotation, ...], \"failures\": [...], \"batch_archive_error\": object|null, \"not_attempted\": [string, ...], \"archive_skipped\": bool, \"next_step\": string }`. Successful annotations include a `url` field when the host can be derived. Each individual failure includes - `annotation_id`, MCP `code`, `message`, and optional `data`. Partial failures set the tool result's - `isError` flag. + `annotation_id` and an upstream `message`. `batch_archive_error` includes `annotation_ids` and an + upstream `message` only for a failed `BatchArchiveAnnotations` request. Partial failures set the tool result's + `isError` flag. `archive_skipped` is true when the requested archive-state change was not attempted + because field updates failed. Parameters: - `annotation_ids`: required list of 1 to 1000 annotation ids. The same changes are applied to every @@ -333,27 +343,27 @@ impl SiftMcpServer { - `metadata`: optional; REPLACES the full metadata list. Each entry is `{ \"name\": \"\", \"value\": }`. Pass `[]` to clear. - `is_archived`: optional archive state. `true` uses one batch-archive request; `false` uses one - batch-unarchive request. When combined with other fields, annotations are updated before their - archive state changes. + `UnarchiveAnnotation` request per annotation. When combined with other fields, annotations are + updated before their archive state changes. At least one updatable field must be set; otherwise the tool returns `INVALID_PARAMS`. Errors: - `INVALID_PARAMS` if `annotation_ids` is empty, contains an empty id, exceeds 1000 ids, `state` is unrecognized, or no updatable field is set. - - Individual upstream failures are returned in `failures` next to successful results. Follow each - failure's guidance and retry only eligible failed ids. + - Individual upstream failures, including unarchive failures, are returned in `failures` next to + successful results. Follow each failure's guidance and retry only eligible failed ids. - Backend-wide failures stop later batches. Their ids are returned in `not_attempted` without an API request. - - Batch archive or unarchive failures are returned in `batch_archive_error`. The archive outcome may - be unknown. + - Batch archive failures are returned in `batch_archive_error`. The archive outcome may be unknown. Guidance: - This is a write. CONFIRM every target and the full proposed values with the user before invoking — `tags`, `linked_channel_ids`, and `metadata` are REPLACE operations, not merges. - - General field updates issue one API request per annotation and are not atomic. Requests run in - batches of up to 50. A backend-wide failure stops later batches. Archive and unarchive use a single - batch API request after all individual updates succeed; otherwise `archive_skipped` is true. + - General field updates and unarchive issue one API request per annotation and are not atomic. Requests + run in batches of up to 50. A backend-wide failure stops later batches. Archive uses one batch API + request. Requested archive-state changes begin only after all individual updates succeed; otherwise + `archive_skipped` is true. - For appends, read the current annotation via `list_annotations` filtered by `annotation_id == \"\"`, then send the union. ", @@ -425,7 +435,6 @@ impl SiftMcpServer { let metadata = metadata.map(|m| m.into_iter().map(MetadataValue::from).collect::>()); let requested_ids = annotation_ids.clone(); let requested_count = annotation_ids.len(); - let outcome = self .annotation_service .update_annotations( @@ -449,12 +458,9 @@ impl SiftMcpServer { .failures .into_iter() .map(|failure| { - let error = from_anyhow(failure.error); serde_json::json!({ "annotation_id": failure.annotation_id, - "code": error.code.0, - "message": error.message, - "data": error.data, + "message": upstream_error_message(&failure.error), }) }) .collect::>(); @@ -462,12 +468,9 @@ impl SiftMcpServer { let not_attempted = outcome.not_attempted; let not_attempted_count = not_attempted.len(); let batch_archive_error = outcome.batch_archive_error.map(|error| { - let error = from_anyhow(error); serde_json::json!({ "annotation_ids": requested_ids, - "code": error.code.0, - "message": error.message, - "data": error.data, + "message": upstream_error_message(&error), }) }); let has_errors = @@ -482,18 +485,18 @@ impl SiftMcpServer { let next_step = if batch_archive_error.is_some() { format!( "Completed field updates for {updated_count} of {requested_count} annotation(s), but batch archive \ - failed. Archive state may be unknown. Verify the targets with `list_annotations` before retrying." + failed. Archive state may be unknown. Verify the targets with `list_annotations` before retrying.", ) } else if outcome.archive_skipped { if not_attempted_count > 0 { format!( "Updated {updated_count} of {requested_count} annotation(s); {failure_count} failed and \ - {not_attempted_count} were not attempted after a backend-wide failure. Archive was not \ + {not_attempted_count} were not attempted after a backend-wide failure. The archive state change was not \ attempted. Review `failures` before retrying eligible failed and not-attempted ids." ) } else { format!( - "Updated {updated_count} of {requested_count} annotation(s); {failure_count} failed. Archive was \ + "Updated {updated_count} of {requested_count} annotation(s); {failure_count} failed. The archive state change was \ not attempted because individual updates failed. Review `failures`, follow their guidance, and \ retry only eligible failed ids." ) diff --git a/rust/crates/sift_mcp/src/tool/annotations/test.rs b/rust/crates/sift_mcp/src/tool/annotations/test.rs index 3d160ebb3..04cbfffbe 100644 --- a/rust/crates/sift_mcp/src/tool/annotations/test.rs +++ b/rust/crates/sift_mcp/src/tool/annotations/test.rs @@ -5,8 +5,8 @@ use std::sync::{ use rmcp::{handler::server::wrapper::Parameters, model::ErrorCode}; use sift_rs::annotations::v1::{ - Annotation, BatchArchiveAnnotationsResponse, BatchUnarchiveAnnotationsResponse, - CreateAnnotationResponse, ListAnnotationsResponse, UpdateAnnotationResponse, + Annotation, BatchArchiveAnnotationsResponse, CreateAnnotationResponse, ListAnnotationsResponse, + UnarchiveAnnotationResponse, UpdateAnnotationResponse, annotation_service_server::AnnotationServiceServer, }; use sift_test_util::{grpc::memory_sift_channel, mock::annotations::v1::MockAnnotationServiceImpl}; @@ -295,7 +295,10 @@ async fn update_annotation_reports_partial_failures_and_continues() { assert_eq!(body["annotations"][0]["annotationId"], "ann1"); assert_eq!(body["annotations"][1]["annotationId"], "ann3"); assert_eq!(body["failures"][0]["annotation_id"], "ann2"); - assert_eq!(body["failures"][0]["code"], ErrorCode::RESOURCE_NOT_FOUND.0); + assert_eq!( + body["failures"][0]["message"], + "Some requested entity was not found: no such annotation" + ); assert_eq!(body["not_attempted"], serde_json::json!([])); } @@ -451,7 +454,7 @@ async fn update_annotation_skips_archive_after_partial_field_failure() { } #[tokio::test] -async fn update_annotation_reports_opaque_batch_archive_error() { +async fn update_annotation_reports_batch_archive_error() { let mut mock = MockAnnotationServiceImpl::new(); mock.expect_batch_archive_annotations() .returning(|_| Err(Status::not_found("one target was not found"))); @@ -474,30 +477,67 @@ async fn update_annotation_reports_opaque_batch_archive_error() { serde_json::json!(["ann1", "ann2"]) ); assert_eq!( - body["batch_archive_error"]["code"], - ErrorCode::RESOURCE_NOT_FOUND.0 + body["batch_archive_error"]["message"], + "Some requested entity was not found: one target was not found" ); + assert!(body["batch_archive_error"].get("code").is_none()); +} + +#[tokio::test] +async fn update_annotation_reports_partial_unarchive_failures() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_unarchive_annotation() + .times(2) + .returning(|req| { + let annotation_id = req.into_inner().annotation_id; + if annotation_id == "ann2" { + return Err(Status::not_found("no such annotation")); + } + Ok(Response::new(UnarchiveAnnotationResponse { + annotation: Some(Annotation { + annotation_id, + is_archived: false, + ..Default::default() + }), + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let mut params = update_params("ann1"); + params.annotation_ids.push("ann2".into()); + params.is_archived = Some(false); + + let resp = server + .update_annotation(Parameters(params)) + .await + .expect("partial unarchive result should be returned"); + + assert_eq!(resp.is_error, Some(true)); + let body = structured(resp); + assert_eq!(body["annotations"][0]["annotationId"], "ann1"); + assert_eq!(body["failures"][0]["annotation_id"], "ann2"); + assert_eq!( + body["failures"][0]["message"], + "Some requested entity was not found: no such annotation" + ); + assert_eq!(body["batch_archive_error"], serde_json::Value::Null); + assert!(body["failures"][0].get("code").is_none()); } #[tokio::test] async fn update_annotation_batch_unarchives_annotations() { let mut mock = MockAnnotationServiceImpl::new(); - mock.expect_batch_unarchive_annotations() - .withf(|req| req.get_ref().annotation_ids == ["ann1", "ann2"]) - .returning(|_| { - Ok(Response::new(BatchUnarchiveAnnotationsResponse { - annotations: vec![ - Annotation { - annotation_id: "ann1".into(), - is_archived: false, - ..Default::default() - }, - Annotation { - annotation_id: "ann2".into(), - is_archived: false, - ..Default::default() - }, - ], + mock.expect_unarchive_annotation() + .times(2) + .returning(|req| { + let annotation_id = req.into_inner().annotation_id; + Ok(Response::new(UnarchiveAnnotationResponse { + annotation: Some(Annotation { + annotation_id, + is_archived: false, + ..Default::default() + }), })) }); @@ -514,6 +554,35 @@ async fn update_annotation_batch_unarchives_annotations() { let annotations = structured_field(resp, "annotations"); assert_eq!(annotations.as_array().unwrap().len(), 2); + assert_eq!(annotations[0]["annotationId"], "ann1"); + assert_eq!(annotations[1]["annotationId"], "ann2"); +} + +#[tokio::test] +async fn update_annotation_reports_unarchive_ids_not_attempted_after_backend_wide_failure() { + let mut mock = MockAnnotationServiceImpl::new(); + mock.expect_unarchive_annotation() + .times(50) + .returning(|_| Err(Status::resource_exhausted("slow down"))); + + let (server, _h) = server_with_mock(mock).await; + + let mut params = update_params("ann1"); + params + .annotation_ids + .extend((2..=51).map(|index| format!("ann{index}"))); + params.is_archived = Some(false); + + let resp = server + .update_annotation(Parameters(params)) + .await + .expect("partial unarchive result should be returned"); + + assert_eq!(resp.is_error, Some(true)); + let body = structured(resp); + assert_eq!(body["failures"].as_array().unwrap().len(), 50); + assert_eq!(body["not_attempted"], serde_json::json!(["ann51"])); + assert_eq!(body["batch_archive_error"], serde_json::Value::Null); } #[tokio::test] @@ -533,21 +602,17 @@ async fn update_annotation_updates_fields_before_unarchiving() { }); let unarchive_count = Arc::clone(&updated_count); - mock.expect_batch_unarchive_annotations() + mock.expect_unarchive_annotation() + .times(2) .withf(move |_| unarchive_count.load(Ordering::SeqCst) == 2) .returning(|req| { - let annotations = req - .into_inner() - .annotation_ids - .into_iter() - .map(|annotation_id| Annotation { + let annotation_id = req.into_inner().annotation_id; + Ok(Response::new(UnarchiveAnnotationResponse { + annotation: Some(Annotation { annotation_id, is_archived: false, ..Default::default() - }) - .collect(); - Ok(Response::new(BatchUnarchiveAnnotationsResponse { - annotations, + }), })) }); @@ -644,5 +709,8 @@ async fn update_annotation_propagates_grpc_error() { assert_eq!(resp.is_error, Some(true)); let body = structured(resp); assert_eq!(body["failures"][0]["annotation_id"], "ann1"); - assert_eq!(body["failures"][0]["code"], ErrorCode::RESOURCE_NOT_FOUND.0); + assert_eq!( + body["failures"][0]["message"], + "Some requested entity was not found: no such annotation" + ); } From 8dfa82c408a5d3e78e4b75c0d8ea3ad3cea56ec1 Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang Date: Wed, 26 Aug 2026 10:27:34 -0700 Subject: [PATCH 6/7] fix: unarchive annotations by clearing the delete date per annotation --- .../sift_mcp/src/service/annotations/mod.rs | 16 ++- .../sift_mcp/src/tool/annotations/mod.rs | 6 +- .../sift_mcp/src/tool/annotations/test.rs | 102 +++++++++--------- 3 files changed, 68 insertions(+), 56 deletions(-) diff --git a/rust/crates/sift_mcp/src/service/annotations/mod.rs b/rust/crates/sift_mcp/src/service/annotations/mod.rs index dcacb9ad8..dcf067bfc 100644 --- a/rust/crates/sift_mcp/src/service/annotations/mod.rs +++ b/rust/crates/sift_mcp/src/service/annotations/mod.rs @@ -8,8 +8,8 @@ use sift_rs::{ annotations::v1::{ Annotation, AnnotationLinkedChannel, AnnotationState, AnnotationType, BatchArchiveAnnotationsRequest, CreateAnnotationRequest, ListAnnotationsRequest, - ListAnnotationsResponse, UnarchiveAnnotationRequest, UpdateAnnotationRequest, - annotation_linked_channel, annotation_service_client::AnnotationServiceClient, + ListAnnotationsResponse, UpdateAnnotationRequest, annotation_linked_channel, + annotation_service_client::AnnotationServiceClient, }, metadata::v1::MetadataValue, }; @@ -493,6 +493,7 @@ impl AnnotationService { .ok_or_else(|| anyhow!("update_annotation response missing annotation")) } + #[allow(deprecated)] // The backend requires this deprecated field to unarchive annotations. pub async fn unarchive_annotation(&self, annotation_id: String) -> Result { let channel = self.channel.clone(); let resp = with_retry(&self.policy, move || { @@ -501,7 +502,16 @@ impl AnnotationService { async move { let mut client = AnnotationServiceClient::new(channel); client - .unarchive_annotation(UnarchiveAnnotationRequest { annotation_id }) + .update_annotation(UpdateAnnotationRequest { + annotation: Some(Annotation { + annotation_id, + deleted_date: None, + ..Default::default() + }), + update_mask: Some(FieldMask { + paths: vec!["deleted_date".into()], + }), + }) .await .map(|resp| resp.into_inner()) } diff --git a/rust/crates/sift_mcp/src/tool/annotations/mod.rs b/rust/crates/sift_mcp/src/tool/annotations/mod.rs index 0b967b03b..5e48e73e7 100644 --- a/rust/crates/sift_mcp/src/tool/annotations/mod.rs +++ b/rust/crates/sift_mcp/src/tool/annotations/mod.rs @@ -317,7 +317,7 @@ impl SiftMcpServer { name = "update_annotation", description = " Update one or more existing annotations. Uses `annotations/v1 BatchArchiveAnnotations` when archiving - and one `UnarchiveAnnotation` request per annotation when unarchiving. + and one `UpdateAnnotation` request per annotation when unarchiving. Output: - `{ \"annotations\": [Annotation, ...], \"failures\": [...], \"batch_archive_error\": object|null, @@ -343,8 +343,8 @@ impl SiftMcpServer { - `metadata`: optional; REPLACES the full metadata list. Each entry is `{ \"name\": \"\", \"value\": }`. Pass `[]` to clear. - `is_archived`: optional archive state. `true` uses one batch-archive request; `false` uses one - `UnarchiveAnnotation` request per annotation. When combined with other fields, annotations are - updated before their archive state changes. + per-annotation update request that clears the annotation's delete date. When combined with other + fields, annotations are updated before their archive state changes. At least one updatable field must be set; otherwise the tool returns `INVALID_PARAMS`. diff --git a/rust/crates/sift_mcp/src/tool/annotations/test.rs b/rust/crates/sift_mcp/src/tool/annotations/test.rs index 04cbfffbe..a0f00ad9d 100644 --- a/rust/crates/sift_mcp/src/tool/annotations/test.rs +++ b/rust/crates/sift_mcp/src/tool/annotations/test.rs @@ -6,8 +6,7 @@ use std::sync::{ use rmcp::{handler::server::wrapper::Parameters, model::ErrorCode}; use sift_rs::annotations::v1::{ Annotation, BatchArchiveAnnotationsResponse, CreateAnnotationResponse, ListAnnotationsResponse, - UnarchiveAnnotationResponse, UpdateAnnotationResponse, - annotation_service_server::AnnotationServiceServer, + UpdateAnnotationResponse, annotation_service_server::AnnotationServiceServer, }; use sift_test_util::{grpc::memory_sift_channel, mock::annotations::v1::MockAnnotationServiceImpl}; use tokio::task::JoinHandle; @@ -484,23 +483,26 @@ async fn update_annotation_reports_batch_archive_error() { } #[tokio::test] +#[allow(deprecated)] // The backend requires this deprecated field to unarchive annotations. async fn update_annotation_reports_partial_unarchive_failures() { let mut mock = MockAnnotationServiceImpl::new(); - mock.expect_unarchive_annotation() - .times(2) - .returning(|req| { - let annotation_id = req.into_inner().annotation_id; - if annotation_id == "ann2" { - return Err(Status::not_found("no such annotation")); - } - Ok(Response::new(UnarchiveAnnotationResponse { - annotation: Some(Annotation { - annotation_id, - is_archived: false, - ..Default::default() - }), - })) - }); + mock.expect_update_annotation().times(2).returning(|req| { + let req = req.into_inner(); + assert_eq!(req.update_mask.unwrap().paths, ["deleted_date"]); + let annotation = req.annotation.unwrap(); + assert_eq!(annotation.deleted_date, None); + let annotation_id = annotation.annotation_id; + if annotation_id == "ann2" { + return Err(Status::not_found("no such annotation")); + } + Ok(Response::new(UpdateAnnotationResponse { + annotation: Some(Annotation { + annotation_id, + is_archived: false, + ..Default::default() + }), + })) + }); let (server, _h) = server_with_mock(mock).await; @@ -526,20 +528,22 @@ async fn update_annotation_reports_partial_unarchive_failures() { } #[tokio::test] +#[allow(deprecated)] // The backend requires this deprecated field to unarchive annotations. async fn update_annotation_batch_unarchives_annotations() { let mut mock = MockAnnotationServiceImpl::new(); - mock.expect_unarchive_annotation() - .times(2) - .returning(|req| { - let annotation_id = req.into_inner().annotation_id; - Ok(Response::new(UnarchiveAnnotationResponse { - annotation: Some(Annotation { - annotation_id, - is_archived: false, - ..Default::default() - }), - })) - }); + mock.expect_update_annotation().times(2).returning(|req| { + let req = req.into_inner(); + assert_eq!(req.update_mask.unwrap().paths, ["deleted_date"]); + let annotation = req.annotation.unwrap(); + assert_eq!(annotation.deleted_date, None); + Ok(Response::new(UpdateAnnotationResponse { + annotation: Some(Annotation { + annotation_id: annotation.annotation_id, + is_archived: false, + ..Default::default() + }), + })) + }); let (server, _h) = server_with_mock(mock).await; @@ -559,11 +563,15 @@ async fn update_annotation_batch_unarchives_annotations() { } #[tokio::test] +#[allow(deprecated)] // The backend requires this deprecated field to unarchive annotations. async fn update_annotation_reports_unarchive_ids_not_attempted_after_backend_wide_failure() { let mut mock = MockAnnotationServiceImpl::new(); - mock.expect_unarchive_annotation() - .times(50) - .returning(|_| Err(Status::resource_exhausted("slow down"))); + mock.expect_update_annotation().times(50).returning(|req| { + let req = req.into_inner(); + assert_eq!(req.update_mask.unwrap().paths, ["deleted_date"]); + assert_eq!(req.annotation.unwrap().deleted_date, None); + Err(Status::resource_exhausted("slow down")) + }); let (server, _h) = server_with_mock(mock).await; @@ -586,36 +594,30 @@ async fn update_annotation_reports_unarchive_ids_not_attempted_after_backend_wid } #[tokio::test] +#[allow(deprecated)] // The backend requires this deprecated field to unarchive annotations. async fn update_annotation_updates_fields_before_unarchiving() { let mut mock = MockAnnotationServiceImpl::new(); let updated_count = Arc::new(AtomicUsize::new(0)); let update_count = Arc::clone(&updated_count); mock.expect_update_annotation() - .times(2) + .times(4) .returning(move |req| { - update_count.fetch_add(1, Ordering::SeqCst); - let annotation = req.into_inner().annotation.unwrap(); + let req = req.into_inner(); + let paths = req.update_mask.unwrap().paths; + let annotation = req.annotation.unwrap(); + if paths == ["deleted_date"] { + assert_eq!(update_count.load(Ordering::SeqCst), 2); + assert_eq!(annotation.deleted_date, None); + } else { + assert_eq!(paths, ["name"]); + update_count.fetch_add(1, Ordering::SeqCst); + } Ok(Response::new(UpdateAnnotationResponse { annotation: Some(annotation), })) }); - let unarchive_count = Arc::clone(&updated_count); - mock.expect_unarchive_annotation() - .times(2) - .withf(move |_| unarchive_count.load(Ordering::SeqCst) == 2) - .returning(|req| { - let annotation_id = req.into_inner().annotation_id; - Ok(Response::new(UnarchiveAnnotationResponse { - annotation: Some(Annotation { - annotation_id, - is_archived: false, - ..Default::default() - }), - })) - }); - let (server, _h) = server_with_mock(mock).await; let mut params = update_params("ann1"); From d91adacf872c6e4cee479d56a8f8b1675a16000b Mon Sep 17 00:00:00 2001 From: Evan Frawley-Tsang <159062208+evan-sift@users.noreply.github.com> Date: Thu, 27 Aug 2026 12:39:03 -0700 Subject: [PATCH 7/7] rust(chore): agent skill refresh and sift-cli 0.5.0 release prep (#763) Co-authored-by: Liam Neville --- rust/crates/sift_cli/CHANGELOG.md | 26 ++++++ rust/crates/sift_cli/Cargo.toml | 2 +- .../sift_cli/assets/skills/sift/SKILL.md | 93 ++++++++++++++----- 3 files changed, 99 insertions(+), 22 deletions(-) diff --git a/rust/crates/sift_cli/CHANGELOG.md b/rust/crates/sift_cli/CHANGELOG.md index d8d0acbce..090d08c0f 100644 --- a/rust/crates/sift_cli/CHANGELOG.md +++ b/rust/crates/sift_cli/CHANGELOG.md @@ -7,6 +7,32 @@ This project adheres to [Semantic Versioning](http://semver.org/). ### What's New +## [v0.5.0] - August 26, 2026 + +### What's New + +- Added MCP tools for managing calculated channels: `list_calculated_channels`, + `list_calculated_channel_versions`, `create_calculated_channel`, + `update_calculated_channel`, `archive_calculated_channel`, and + `unarchive_calculated_channel`. +- `get_data` now serves saved calculated channels. A name in `channel_names` + with no raw-channel match resolves as an active saved calculated channel for + the asset and run; unresolvable names are reported explicitly. +- `get_data` now accepts `asset_id` as an alternative to `asset_name`; exactly + one must be set. +- Added `preview_rule`, which dry-runs a saved rule or an ad-hoc draft rule + config against a run without persisting anything. +- Added MCP tools for managing user-defined functions: + `list_user_defined_functions`, `list_user_defined_function_versions`, + `create_user_defined_function`, `update_user_defined_function`, + `archive_user_defined_function`, and `unarchive_user_defined_function`. +- `update_annotation` now requires `annotation_ids` instead of `annotation_id`, + a breaking change for existing callers; pass a one-element list for one + annotation. It updates 1 to 1000 annotations per call with per-ID failure + reporting, and its new `is_archived` parameter archives or unarchives + annotations. +- Refreshed the bundled Sift agent skill to cover the expanded MCP tool surface. + ## [v0.4.4] - August 24, 2026 ### What's New diff --git a/rust/crates/sift_cli/Cargo.toml b/rust/crates/sift_cli/Cargo.toml index 933525bd9..f1b096a04 100644 --- a/rust/crates/sift_cli/Cargo.toml +++ b/rust/crates/sift_cli/Cargo.toml @@ -3,7 +3,7 @@ test-reports = ["sift_mcp/test-reports"] [package] name = "sift_cli" -version = "0.4.4" +version = "0.5.0" authors.workspace = true edition.workspace = true categories.workspace = true diff --git a/rust/crates/sift_cli/assets/skills/sift/SKILL.md b/rust/crates/sift_cli/assets/skills/sift/SKILL.md index de9eab083..47f56b3cb 100644 --- a/rust/crates/sift_cli/assets/skills/sift/SKILL.md +++ b/rust/crates/sift_cli/assets/skills/sift/SKILL.md @@ -1,19 +1,21 @@ --- name: sift description: >- - Use when working with Sift: ingesting or importing time-series data, - querying assets/runs/channels/users, exporting data, decimating or running - SQL over data, opening a view in the Sift Explore web app, writing code that - integrates with Sift, installing, updating, or diagnosing the Sift agent - integration, or looking up how Sift works in its product and API - documentation. Covers the Sift MCP server (started by `sift-cli mcp`), the - `sift-cli` itself, the Sift REST API over cURL, the Sift Python library - (`sift_client`), and the Sift Rust streaming library (`sift_stream`). + Use for Sift tasks: ingesting or importing time-series data, querying + assets/runs/channels/users, managing calculated channels, rules, and + user-defined functions, exporting or decimating data, running SQL over data, + opening a view in Sift Explore, writing code that integrates with Sift, + installing, updating, or diagnosing the Sift agent integration, or looking + up how Sift works in its product and API documentation. Covers the Sift MCP + server (started by `sift-cli mcp`), `sift-cli`, the Sift REST API over cURL, + the Sift Python library (`sift_client`), and the Sift Rust streaming library + (`sift_stream`). Triggers include phrases like "import this file into Sift", "stream data to Sift", "list assets/runs/channels", "runs I created", "runs a teammate - created", "export a run", "query Sift", "graph", "plot", "visualize", "open - in Explore", "write code to integrate with Sift", "how does X work in Sift", - "what does this endpoint do", or "look up the Sift API reference". + created", "export a run", "query Sift", "graph", "plot", "visualize", "open in + Explore", "write code to integrate with Sift", "how does X work in Sift", + "what does this endpoint do", "list calculated channels", "preview a rule", + or "look up the Sift API reference". ---