From 819c8414eb635745098abdbbbd615580b6ca2ae8 Mon Sep 17 00:00:00 2001 From: Bill Raj-Derouin Date: Mon, 12 Oct 2026 20:45:44 -0700 Subject: [PATCH] rust(fix): exclude archived runs from get_data run_name lookup and add run_id get_data resolved run_name with limit 1 and no archive filter, so an archived run sharing a name with the active one could silently supply the data in its place. - Name lookup now filters is_archived == false and fetches two runs. More than one active match fails with INVALID_PARAMS listing the candidate run ids. A miss points at an archived run only when one holds the name. - New run_id param, mutually exclusive with run_name. It is the only way to fetch an archived run. Blank run selectors are rejected. - The result carries run_id, plus run_archived for an archived run, and the unresolved calculated channel report names the run by id. - explore_asset, analyze_run and derive_and_upload list active runs only and pass run_id to get_data, asking the user when a name matches more than one run. - Tests cover an archived duplicate, multiple active matches, run_id on an archived run, both not-found messages, and the new validation errors. Fixes ENG-15763 --- rust/crates/sift_cli/CHANGELOG.md | 7 + rust/crates/sift_mcp/src/prompt/mod.rs | 26 +- rust/crates/sift_mcp/src/tool/data/mod.rs | 207 ++++++++-- rust/crates/sift_mcp/src/tool/data/test.rs | 415 ++++++++++++++++++++- 4 files changed, 596 insertions(+), 59 deletions(-) diff --git a/rust/crates/sift_cli/CHANGELOG.md b/rust/crates/sift_cli/CHANGELOG.md index 59dea0a8a2..417351c1d0 100644 --- a/rust/crates/sift_cli/CHANGELOG.md +++ b/rust/crates/sift_cli/CHANGELOG.md @@ -11,6 +11,13 @@ This project adheres to [Semantic Versioning](http://semver.org/). `source` for the run or asset an artifact was built from, and `derived_from` only for another artifact. Agents no longer send `derived_from` for a run and get the call rejected. +- `get_data` no longer matches archived runs by `run_name`. Before, an archived + run with the same name as an active one could supply the data in its place. + A name that matches more than one active run now fails and lists the + candidate run ids. +- `get_data` accepts `run_id` in place of `run_name`. It is the only way to + fetch an archived run. The result names the run in `run_id` and sets + `run_archived` when the run is archived. ## [v0.7.3] - October 7, 2026 diff --git a/rust/crates/sift_mcp/src/prompt/mod.rs b/rust/crates/sift_mcp/src/prompt/mod.rs index 160bbf19fc..66a2efd4f7 100644 --- a/rust/crates/sift_mcp/src/prompt/mod.rs +++ b/rust/crates/sift_mcp/src/prompt/mod.rs @@ -47,13 +47,15 @@ impl SiftMcpServer { (`name == \"{asset}\"`); if nothing comes back, fall back to a substring match \ (`name.matches(\"{asset}\")`). When several assets match, ask the user which one they \ mean before continuing.\n\ - 2. List recent runs with `list_runs` filtered by `asset_id == \"\"`, \ - ordered `start_time desc`. Run and channel namespaces are per-asset, so always scope by \ - `asset_id` instead of listing everything.\n\ + 2. List recent runs with `list_runs` filtered by \ + `asset_id == \"\" && is_archived == false`, ordered `start_time desc`. \ + Run and channel namespaces are per-asset, so always scope by `asset_id` instead of \ + listing everything.\n\ 3. List channels with `list_channels` filtered by `asset_id == \"\"`.\n\ 4. Summarize for the user: the resolved asset (name and id), its most recent runs (name, \ - start/stop, duration), and the channel inventory grouped by data type. Surface the exact \ - run and channel names so they can be reused with `get_data`.\n\n\ + start/stop, duration), and the channel inventory grouped by data type. Surface each run's \ + `run_id` with its name, and the exact channel names, so they can be reused with \ + `get_data`.\n\n\ This step is discovery only. Do not pull sample data." ); @@ -87,10 +89,11 @@ impl SiftMcpServer { {question_line}\ Steps:\n\ 1. Resolve the asset and run. Use `list_assets` (`name == \"{asset}\"`) for the \ - asset_id, then `list_runs` (`name == \"{run}\" && asset_id == \"...\"`) to confirm the run.\n\ + asset_id, then `list_runs` (`name == \"{run}\" && asset_id == \"...\" && is_archived == false`) to \ + confirm the run and take its `run_id`. If more than one run matches, ask the user which one.\n\ 2. Confirm the target channels exist with `list_channels` scoped by `asset_id`. If no \ channels were named, choose a sensible set and tell the user which you picked.\n\ - 3. Pull data with `get_data`. Pass `run_name` so the run's start/stop bounds apply \ + 3. Pull data with `get_data`. Pass `run_id` so the run's start/stop bounds apply \ automatically; do not hand-compute timestamps. Use `channel_names` with exact \ channel names. Pass `sample_ms = 0`: this flow ends in statistics, and statistics taken \ off decimated data are wrong. If a long run makes the pull too large, narrow the time \ @@ -138,9 +141,12 @@ impl SiftMcpServer { Source asset: \"{source_asset}\"\nSource run: \"{source_run}\"\n\ Transform: \"{transform}\"\n{target_asset_line}\n{target_run_line}\n\n\ Steps:\n\ - 1. Resolve the source asset and run with `list_assets` and `list_runs`. Identify the \ - channels the transform needs via `list_channels` scoped by `asset_id`.\n\ - 2. Extract with `get_data`, passing `run_name` so the run bounds apply. Choose \ + 1. Resolve the source asset and run. Use `list_assets` (`name == \"{source_asset}\"`) for \ + the asset_id, then `list_runs` (`name == \"{source_run}\" && asset_id == \"...\" && \ + is_archived == false`) to confirm the run and take its `run_id`. If more than one run \ + matches, ask the user which one. Identify the channels the transform needs via \ + `list_channels` scoped by `asset_id`.\n\ + 2. Extract with `get_data`, passing `run_id` so the run bounds apply. Choose \ `channel_names`, and pass `sample_ms = 0` so the transform runs on real samples rather \ than a decimated approximation of them.\n\ 3. Apply the transform with `sql`. CRITICAL: column 0 of any dataset uploaded to Sift \ diff --git a/rust/crates/sift_mcp/src/tool/data/mod.rs b/rust/crates/sift_mcp/src/tool/data/mod.rs index 820fef2cd9..631e9afd89 100644 --- a/rust/crates/sift_mcp/src/tool/data/mod.rs +++ b/rust/crates/sift_mcp/src/tool/data/mod.rs @@ -17,7 +17,7 @@ use rmcp::{ use serde::Deserialize; use serde_json::Value; -use sift_rs::metadata::v1::MetadataValue; +use sift_rs::{assets::v1::Asset, metadata::v1::MetadataValue, runs::v2::Run}; use crate::{ error::{self, from_anyhow}, @@ -157,6 +157,7 @@ pub struct GetDataParams { asset_name: Option, asset_id: Option, run_name: Option, + run_id: Option, start_time_unix_nanos: Option, end_time_unix_nanos: Option, sample_ms: u32, @@ -167,6 +168,12 @@ pub struct GetDataParams { output: PathBuf, } +/// Which run scopes a `get_data` call, before it is resolved on the asset. +enum RunSelector { + Id(String), + Name(String), +} + #[derive(Debug, Deserialize, JsonSchema)] pub struct SqlParams { inputs: Vec, @@ -247,6 +254,8 @@ impl SiftMcpServer { always what was requested: `sample_ms` is ignored for data types that cannot be sampled. Both keys are ALWAYS present; `sample_ms_applied` is null when no page reported a rate. `mixed_sample_rates` appears when some channels came back decimated and others raw. + - `run_id` names the run that scoped the query, when one did. `run_archived` is `true` when that + run is archived. Parameters: - `asset_name`: optional, exact asset name (not a pattern). Mutually exclusive with `asset_id`; @@ -254,9 +263,11 @@ impl SiftMcpServer { - `asset_id`: optional, asset UUID. Mutually exclusive with `asset_name`; exactly one of the two MUST be set. Runs reference assets by id (`list_runs` returns `asset_id`, not a name), so pass that id here directly instead of guessing a name. - - `run_name`: optional, exact run name within the asset. When provided, the run's start/stop bounds are - used as the time range; `start_time_unix_nanos` and/or `end_time_unix_nanos` may narrow either side. - When omitted, BOTH `start_time_unix_nanos` and `end_time_unix_nanos` are required. + - `run_name`: optional, exact name of an active (non-archived) run on the asset. The run's start/stop + bounds become the time range; `start_time_unix_nanos`/`end_time_unix_nanos` may narrow either side. + Set at most one of `run_name` and `run_id`; with neither, BOTH time bounds are required. + - `run_id`: optional, run UUID from `list_runs`. Same as `run_name`, but also reaches archived runs. + Prefer it when known: names are not unique. - `sample_ms`: decimation interval in milliseconds. Set this to `0` unless the data will be used EXCLUSIVELY to generate a visualization. Decimation applies LTTB, which selects shape-defining extremes rather than representative samples: counts, means, standard deviations and trends @@ -292,8 +303,10 @@ impl SiftMcpServer { channel does not apply when the asset is outside its scope or lacks a channel its expression references. Verify the name with `list_calculated_channels` filtered by `asset_id`. - `INVALID_PARAMS` if neither `asset_name` nor `asset_id` is set, or if both are set. - - `INVALID_PARAMS` if `run_name` is absent and the full time range is not supplied, if not exactly - one channel selector is set, if a selection array is empty, or if a channel ID is blank. + - `INVALID_PARAMS` if both `run_name` and `run_id` are set or either is blank, if `run_name` + matches more than one active run (the message lists their ids), if neither is set and the full + time range is not supplied, if not exactly one channel selector is set, if a selection array is + empty, or if a channel ID is blank. - `INVALID_PARAMS` if the channel selection matches more than 200 channels and would be incomplete. Narrow `channel_regex`, pass explicit `channel_names` or `channel_ids`, or split the request. @@ -333,6 +346,7 @@ impl SiftMcpServer { asset_name, asset_id, run_name, + run_id, channel_names, channel_regex, channel_id, @@ -343,14 +357,37 @@ impl SiftMcpServer { output, }) = params; - if run_name.is_none() && (start_time_unix_nanos.is_none() || end_time_unix_nanos.is_none()) + if run_name + .iter() + .chain(&run_id) + .any(|value| value.trim().is_empty()) { return Err(ErrorData::invalid_params( - "start_time_unix_nanos and end_time_unix_nanos are required when run_name is not provided", + "`run_name` and `run_id` must not be blank; omit the one you are not using", None, )); } + let run_selector = match (run_name, run_id) { + (Some(_), Some(_)) => { + return Err(ErrorData::invalid_params( + "set at most one of `run_name` or `run_id`, not both", + None, + )); + } + (Some(name), None) => Some(RunSelector::Name(name)), + (None, Some(id)) => Some(RunSelector::Id(id)), + (None, None) => { + if start_time_unix_nanos.is_none() || end_time_unix_nanos.is_none() { + return Err(ErrorData::invalid_params( + "start_time_unix_nanos and end_time_unix_nanos are required when neither run_name nor run_id is provided", + None, + )); + } + None + } + }; + if [ channel_names.is_some(), channel_regex.is_some(), @@ -441,29 +478,8 @@ impl SiftMcpServer { ErrorData::resource_not_found(format!("{asset_label} not found"), None) })?; - let run = match run_name.as_deref() { - Some(name) => { - let filter = format!( - "name == \"{}\" && asset_id == \"{}\"", - cel_escape(name), - asset.asset_id, - ); - let run = self - .run_service - .list_runs(filter, None, Some(1)) - .await - .map_err(from_anyhow)? - .items - .into_iter() - .next() - .ok_or_else(|| { - ErrorData::resource_not_found( - format!("run '{name}' not found for asset '{}'", asset.name), - None, - ) - })?; - Some(run) - } + let run = match run_selector { + Some(selector) => Some(self.resolve_run(&asset, selector).await?), None => None, }; @@ -576,7 +592,7 @@ impl SiftMcpServer { } let unresolved_report = (!unresolved.is_empty()) - .then(|| unresolved_message(&unresolved, &asset.name, run_name.as_deref())); + .then(|| unresolved_message(&unresolved, &asset.name, run.as_ref())); if channel_inputs.is_empty() { return Err(ErrorData::resource_not_found( @@ -598,6 +614,17 @@ impl SiftMcpServer { .map(|entry| entry.name.clone()) .collect::>(); + // Only an explicit `run_id` reaches an archived run; say so in the + // result so the caller does not present it as the active one. + let run_id = run.as_ref().map(|run| run.run_id.clone()); + let archived_note = run.as_ref().filter(|run| run.is_archived).map(|run| { + format!( + " Run `{}` (id `{}`) is ARCHIVED. Tell the user this data comes from an \ + archived run.", + run.name, run.run_id, + ) + }); + let time_range = match run { Some(run) => TimeRange::Run { run: Box::new(run), @@ -663,6 +690,9 @@ impl SiftMcpServer { let mut next_step = format!("Wrote channel data to `{output_str}`. Inform the user where the data lives."); + if let Some(note) = &archived_note { + next_step.push_str(note); + } // The caller asked for a rate; the service reports what it actually // applied, and the two are not the same thing for a data type it cannot // sample. Saying so here is the difference between an agent that knows @@ -727,6 +757,12 @@ impl SiftMcpServer { if applied.mixed() { body.insert("mixed_sample_rates".to_string(), Value::from(true)); } + if let Some(run_id) = run_id { + body.insert("run_id".to_string(), Value::from(run_id)); + } + if archived_note.is_some() { + body.insert("run_archived".to_string(), Value::from(true)); + } if !unresolved.is_empty() { body.insert( @@ -969,16 +1005,117 @@ impl SiftMcpServer { } } +impl SiftMcpServer { + /// Resolve a run selector to a run on `asset`. + async fn resolve_run(&self, asset: &Asset, selector: RunSelector) -> Result { + match selector { + // An explicit id is honored even when the run is archived. + RunSelector::Id(id) => { + let filter = format!( + "run_id == \"{}\" && asset_id == \"{}\"", + cel_escape(&id), + asset.asset_id, + ); + self.run_service + .list_runs(filter, None, Some(1)) + .await + .map_err(from_anyhow)? + .items + .into_iter() + .next() + .ok_or_else(|| { + ErrorData::resource_not_found( + format!("run with id '{id}' not found for asset '{}'", asset.name), + None, + ) + }) + } + // Names are not unique. An archived duplicate must never stand in + // for the active run, and two active matches are ambiguous, so the + // lookup excludes archived runs and fetches two to detect a tie. + RunSelector::Name(name) => { + let filter = format!( + "name == \"{}\" && asset_id == \"{}\" && is_archived == false", + cel_escape(&name), + asset.asset_id, + ); + let page = self + .run_service + .list_runs(filter, None, Some(2)) + .await + .map_err(from_anyhow)?; + if page.items.len() > 1 { + let ids = page + .items + .iter() + .map(|r| r.run_id.clone()) + .collect::>(); + let more = if page.has_more { " and more" } else { "" }; + return Err(ErrorData::invalid_params( + format!( + "run name '{name}' matches more than one active run on asset '{}' \ + (ids {}{more}); pick one with `list_runs` and pass it as `run_id`", + asset.name, + common::name_list(&ids), + ), + None, + )); + } + if let Some(run) = page.items.into_iter().next() { + return Ok(run); + } + + // Point at an archived run only when one holds the name, so a + // typo does not send the caller after archived data. The check + // is best effort: if it fails, report the plain miss. + let archived_filter = format!( + "name == \"{}\" && asset_id == \"{}\" && is_archived == true", + cel_escape(&name), + asset.asset_id, + ); + let archived = self + .run_service + .list_runs(archived_filter, None, Some(1)) + .await + .ok() + .and_then(|page| page.items.into_iter().next()); + let hint = match archived { + Some(run) => format!( + "; an archived run has that name (id '{}'). Archived runs are not matched \ + by name; pass it as `run_id` only if the user asked for archived data", + run.run_id + ), + None => "; check the name with `list_runs`".to_string(), + }; + Err(ErrorData::resource_not_found( + format!( + "no active run named '{name}' found for asset '{}'{hint}", + asset.name + ), + None, + )) + } + } + } +} + /// One phrasing for calculated channels that yielded no data, shared by the /// partial-result report and the all-unresolved error so the caller reads the /// same wording either way. fn unresolved_message( unresolved: &[UnresolvedCalculation], asset_name: &str, - run_name: Option<&str>, + run: Option<&Run>, ) -> String { - let scope = match run_name { - Some(run) => format!("asset '{asset_name}' (run '{run}')"), + // The id and archived flag tell an archived run from an active one that + // shares its name. + let scope = match run { + Some(run) => format!( + "asset '{asset_name}' ({}run '{}', id '{}')", + if run.is_archived { "archived " } else { "" }, + run.name, + run.run_id, + ), None => format!("asset '{asset_name}'"), }; let items = unresolved diff --git a/rust/crates/sift_mcp/src/tool/data/test.rs b/rust/crates/sift_mcp/src/tool/data/test.rs index 97af67512b..86221e51c7 100644 --- a/rust/crates/sift_mcp/src/tool/data/test.rs +++ b/rust/crates/sift_mcp/src/tool/data/test.rs @@ -99,6 +99,7 @@ fn get_data_params(channel_regex: &str) -> Parameters { asset_name: Some("bench".into()), asset_id: None, run_name: None, + run_id: None, start_time_unix_nanos: Some(0), end_time_unix_nanos: Some(1), sample_ms: 0, @@ -604,6 +605,7 @@ fn named_params( asset_name: Some("bench".into()), asset_id: None, run_name: run_name.map(String::from), + run_id: None, start_time_unix_nanos: Some(0), end_time_unix_nanos: Some(3_000_000_000), sample_ms: 0, @@ -636,23 +638,29 @@ fn raw_channel_mock(names: &[&str]) -> MockChannelServiceImpl { mock } +/// A run spanning 0 to 3 seconds. +fn run_named(run_id: &str, name: &str, is_archived: bool) -> Run { + Run { + run_id: run_id.into(), + name: name.into(), + is_archived, + start_time: Some(Timestamp { + seconds: 0, + nanos: 0, + }), + stop_time: Some(Timestamp { + seconds: 3, + nanos: 0, + }), + ..Default::default() + } +} + fn one_run_mock() -> MockRunServiceImpl { let mut runs = MockRunServiceImpl::new(); runs.expect_list_runs().returning(|_| { Ok(Response::new(ListRunsResponse { - runs: vec![Run { - run_id: "run-1".into(), - name: "hotfire-3".into(), - start_time: Some(Timestamp { - seconds: 0, - nanos: 0, - }), - stop_time: Some(Timestamp { - seconds: 3, - nanos: 0, - }), - ..Default::default() - }], + runs: vec![run_named("run-1", "hotfire-3", false)], next_page_token: String::new(), })) }); @@ -870,7 +878,7 @@ async fn get_data_reports_unresolved_calculated_channel() { .to_string(); assert!( next_step.contains("calculated channel data was not resolved for asset 'bench'") - && next_step.contains("hotfire-3") + && next_step.contains("(run 'hotfire-3', id 'run-1')") && next_step.contains("chamber_delta"), "next_step must name what did not resolve, and for which asset and run: {next_step}", ); @@ -1178,6 +1186,7 @@ async fn get_data_reports_channel_names_that_matched_nothing() { asset_name: Some("bench".into()), asset_id: None, run_name: None, + run_id: None, start_time_unix_nanos: Some(0), end_time_unix_nanos: Some(2_000_000_000), sample_ms: 0, @@ -1247,6 +1256,7 @@ async fn get_data_reports_no_unmatched_names_for_a_regex_selection() { asset_name: Some("bench".into()), asset_id: None, run_name: None, + run_id: None, start_time_unix_nanos: Some(0), end_time_unix_nanos: Some(2_000_000_000), sample_ms: 0, @@ -1310,6 +1320,7 @@ async fn get_data_reports_matched_channels_that_returned_no_samples() { asset_name: Some("bench".into()), asset_id: None, run_name: None, + run_id: None, start_time_unix_nanos: Some(0), end_time_unix_nanos: Some(2_000_000_000), sample_ms: 0, @@ -1378,6 +1389,7 @@ async fn no_data_error_reports_both_the_empty_and_the_unmatched_channels() { asset_name: Some("bench".into()), asset_id: None, run_name: None, + run_id: None, start_time_unix_nanos: Some(0), end_time_unix_nanos: Some(2_000_000_000), sample_ms: 0, @@ -1561,6 +1573,7 @@ async fn get_data_persist_failure_keeps_the_channel_report() { asset_name: Some("bench".into()), asset_id: None, run_name: None, + run_id: None, start_time_unix_nanos: Some(0), end_time_unix_nanos: Some(2_000_000_000), sample_ms: 0, @@ -1617,6 +1630,7 @@ async fn get_data_leaves_no_file_when_the_data_call_fails() { asset_name: Some("bench".into()), asset_id: None, run_name: None, + run_id: None, start_time_unix_nanos: Some(0), end_time_unix_nanos: Some(2_000_000_000), sample_ms: 0, @@ -1674,6 +1688,7 @@ async fn get_data_leaves_no_file_when_the_request_is_cancelled() { asset_name: Some("bench".into()), asset_id: None, run_name: None, + run_id: None, start_time_unix_nanos: Some(0), end_time_unix_nanos: Some(2_000_000_000), sample_ms: 0, @@ -1764,6 +1779,7 @@ async fn get_data_writes_into_a_directory_that_does_not_exist_yet() { asset_name: Some("bench".into()), asset_id: None, run_name: None, + run_id: None, start_time_unix_nanos: Some(0), end_time_unix_nanos: Some(2_000_000_000), sample_ms: 0, @@ -1778,3 +1794,374 @@ async fn get_data_writes_into_a_directory_that_does_not_exist_yet() { assert!(nested.exists()); } + +/// A data service that asserts every query is scoped to `run_id` and returns +/// one sample for `ch-0`. +fn run_scoped_data_mock(run_id: &'static str) -> MockDataServiceImpl { + let mut data = MockDataServiceImpl::new(); + data.expect_get_data() + .times(1) + .withf(move |req| { + let queries = &req.get_ref().queries; + !queries.is_empty() + && queries.iter().all(|query| match &query.query { + Some(QueryKind::Channel(channel)) => channel.run_id.as_deref() == Some(run_id), + _ => false, + }) + }) + .returning(|_| { + Ok(Response::new(GetDataResponse { + data: vec![double_page("ch-0", "pressure", vec![(1_000_000_000, 1.0)])], + next_page_token: String::new(), + })) + }); + data +} + +/// `pressure` scoped by a run selector alone, with no explicit time range. +fn run_params( + run_name: Option<&str>, + run_id: Option<&str>, + output: PathBuf, +) -> Parameters { + let Parameters(params) = named_params(&["pressure"], run_name, output); + Parameters(GetDataParams { + run_id: run_id.map(String::from), + start_time_unix_nanos: None, + end_time_unix_nanos: None, + ..params + }) +} + +/// An archived run that shares its name with the active one must not stand in +/// for it: the name lookup excludes archived runs, so the data comes from the +/// active run. +#[tokio::test] +async fn get_data_run_name_skips_an_archived_duplicate() { + let mut runs = MockRunServiceImpl::new(); + runs.expect_list_runs() + .times(1) + .withf(|req| { + let req = req.get_ref(); + req.filter == "name == \"hotfire-3\" && asset_id == \"asset-1\" && is_archived == false" + && req.page_size == 2 + }) + .returning(|_| { + Ok(Response::new(ListRunsResponse { + runs: vec![run_named("run-active", "hotfire-3", false)], + next_page_token: String::new(), + })) + }); + + let dir = TempDir::new("sift-mcp-run-archived-dup").expect("temp dir"); + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + raw_channel_mock(&["pressure"]), + runs, + MockCalculatedChannelServiceImpl::new(), + run_scoped_data_mock("run-active"), + ) + .await; + + let body = structured( + server + .get_data(run_params( + Some("hotfire-3"), + None, + dir.path().join("out.parquet"), + )) + .await + .expect("get_data should fetch the active run"), + ); + assert_eq!(body["run_id"], "run-active"); + assert!(body.get("run_archived").is_none(), "{body}"); +} + +/// Two active runs with the same name are ambiguous. Picking either one could +/// ship the wrong run's data, so the call fails and names both candidates. +#[tokio::test] +async fn get_data_run_name_rejects_multiple_active_matches() { + let mut runs = MockRunServiceImpl::new(); + runs.expect_list_runs().times(1).returning(|_| { + Ok(Response::new(ListRunsResponse { + runs: vec![ + run_named("run-a", "hotfire-3", false), + run_named("run-b", "hotfire-3", false), + ], + next_page_token: "more".into(), + })) + }); + + let dir = TempDir::new("sift-mcp-run-ambiguous").expect("temp dir"); + let output = dir.path().join("out.parquet"); + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + MockChannelServiceImpl::new(), + runs, + MockCalculatedChannelServiceImpl::new(), + MockDataServiceImpl::new(), + ) + .await; + + let err = server + .get_data(run_params(Some("hotfire-3"), None, output.clone())) + .await + .expect_err("an ambiguous run name must fail"); + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); + assert!(err.message.contains("run-a"), "{}", err.message); + assert!(err.message.contains("run-b"), "{}", err.message); + assert!(err.message.contains("and more"), "{}", err.message); + assert!(err.message.contains("`run_id`"), "{}", err.message); + assert!(!output.exists(), "no file should be written"); +} + +/// A run service where the name lookup finds no active run and the archived +/// check returns `archived`. +fn no_active_run_mock(archived: Option) -> MockRunServiceImpl { + let mut runs = MockRunServiceImpl::new(); + runs.expect_list_runs() + .times(1) + .withf(|req| req.get_ref().filter.ends_with("is_archived == false")) + .returning(|_| { + Ok(Response::new(ListRunsResponse { + runs: vec![], + next_page_token: String::new(), + })) + }); + runs.expect_list_runs() + .times(1) + .withf(|req| req.get_ref().filter.ends_with("is_archived == true")) + .returning(move |_| { + Ok(Response::new(ListRunsResponse { + runs: archived.clone().into_iter().collect(), + next_page_token: String::new(), + })) + }); + runs +} + +/// When the only run with the name is archived, the name lookup finds nothing +/// and the error names the archived run's id for `run_id`. +#[tokio::test] +async fn get_data_run_name_does_not_match_an_archived_run() { + let runs = no_active_run_mock(Some(run_named("run-archived", "hotfire-3", true))); + + let dir = TempDir::new("sift-mcp-run-only-archived").expect("temp dir"); + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + MockChannelServiceImpl::new(), + runs, + MockCalculatedChannelServiceImpl::new(), + MockDataServiceImpl::new(), + ) + .await; + + let err = server + .get_data(run_params( + Some("hotfire-3"), + None, + dir.path().join("out.parquet"), + )) + .await + .expect_err("no active run matches"); + assert_eq!(err.code, ErrorCode::RESOURCE_NOT_FOUND); + assert!(err.message.contains("run-archived"), "{}", err.message); + assert!(err.message.contains("`run_id`"), "{}", err.message); +} + +/// A name no run holds, archived or not, gets a plain miss: pointing at +/// `run_id` would send the caller looking for archived data that is not there. +#[tokio::test] +async fn get_data_run_name_not_found_does_not_suggest_archived_data() { + let dir = TempDir::new("sift-mcp-run-missing").expect("temp dir"); + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + MockChannelServiceImpl::new(), + no_active_run_mock(None), + MockCalculatedChannelServiceImpl::new(), + MockDataServiceImpl::new(), + ) + .await; + + let err = server + .get_data(run_params( + Some("hotfire-9"), + None, + dir.path().join("out.parquet"), + )) + .await + .expect_err("no run has the name"); + assert_eq!(err.code, ErrorCode::RESOURCE_NOT_FOUND); + assert!(!err.message.contains("archived"), "{}", err.message); + assert!(err.message.contains("`list_runs`"), "{}", err.message); +} + +/// The unresolved-channel report names the run by id and flags it as +/// archived, so it is not mistaken for an active run with the same name. +#[tokio::test] +async fn get_data_unresolved_report_marks_an_archived_run() { + let mut runs = MockRunServiceImpl::new(); + runs.expect_list_runs().times(1).returning(|_| { + Ok(Response::new(ListRunsResponse { + runs: vec![run_named("run-archived", "hotfire-3", true)], + next_page_token: String::new(), + })) + }); + + let dir = TempDir::new("sift-mcp-run-archived-unresolved").expect("temp dir"); + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + raw_channel_mock(&[]), + runs, + no_calculation_mock(), + MockDataServiceImpl::new(), + ) + .await; + + let err = server + .get_data(run_params( + None, + Some("run-archived"), + dir.path().join("out.parquet"), + )) + .await + .expect_err("nothing resolves"); + assert_eq!(err.code, ErrorCode::RESOURCE_NOT_FOUND); + assert!( + err.message + .contains("(archived run 'hotfire-3', id 'run-archived')"), + "{}", + err.message + ); +} + +/// `run_id` selects the run directly, archived or not, and the result says +/// when the run is archived. +#[tokio::test] +async fn get_data_run_id_fetches_an_archived_run_and_says_so() { + let mut runs = MockRunServiceImpl::new(); + runs.expect_list_runs() + .times(1) + .withf(|req| { + req.get_ref().filter == "run_id == \"run-archived\" && asset_id == \"asset-1\"" + }) + .returning(|_| { + Ok(Response::new(ListRunsResponse { + runs: vec![run_named("run-archived", "hotfire-3", true)], + next_page_token: String::new(), + })) + }); + + let dir = TempDir::new("sift-mcp-run-id").expect("temp dir"); + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + raw_channel_mock(&["pressure"]), + runs, + MockCalculatedChannelServiceImpl::new(), + run_scoped_data_mock("run-archived"), + ) + .await; + + let body = structured( + server + .get_data(run_params( + None, + Some("run-archived"), + dir.path().join("out.parquet"), + )) + .await + .expect("an explicit run_id may name an archived run"), + ); + assert_eq!(body["run_id"], "run-archived"); + assert_eq!(body["run_archived"], true); + let next_step = body["next_step"].as_str().expect("next_step"); + assert!(next_step.contains("ARCHIVED"), "{next_step}"); +} + +#[tokio::test] +async fn get_data_run_id_not_found_for_asset() { + let mut runs = MockRunServiceImpl::new(); + runs.expect_list_runs().times(1).returning(|_| { + Ok(Response::new(ListRunsResponse { + runs: vec![], + next_page_token: String::new(), + })) + }); + + let dir = TempDir::new("sift-mcp-run-id-missing").expect("temp dir"); + let (server, _h) = server_with_calculation_mocks( + one_asset_mock(), + MockChannelServiceImpl::new(), + runs, + MockCalculatedChannelServiceImpl::new(), + MockDataServiceImpl::new(), + ) + .await; + + let err = server + .get_data(run_params( + None, + Some("run-elsewhere"), + dir.path().join("out.parquet"), + )) + .await + .expect_err("a run id not on the asset must fail"); + assert_eq!(err.code, ErrorCode::RESOURCE_NOT_FOUND); + assert!(err.message.contains("run-elsewhere"), "{}", err.message); +} + +#[tokio::test] +async fn get_data_rejects_both_run_name_and_run_id() { + let (server, _h) = + server_with_mocks(MockAssetServiceImpl::new(), MockChannelServiceImpl::new()).await; + + let err = server + .get_data(run_params( + Some("hotfire-3"), + Some("run-1"), + std::env::temp_dir().join("sift-mcp-get-data-test-never-written.parquet"), + )) + .await + .expect_err("run_name and run_id are mutually exclusive"); + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +/// A blank run selector is rejected up front rather than sent as a filter or +/// counted against the other selector. +#[tokio::test] +async fn get_data_rejects_a_blank_run_selector() { + let (server, _h) = + server_with_mocks(MockAssetServiceImpl::new(), MockChannelServiceImpl::new()).await; + + for (run_name, run_id) in [(Some(""), Some("run-1")), (None, Some(" "))] { + let err = server + .get_data(run_params( + run_name, + run_id, + std::env::temp_dir().join("sift-mcp-get-data-test-never-written.parquet"), + )) + .await + .expect_err("a blank run selector must fail"); + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); + assert!(err.message.contains("blank"), "{}", err.message); + } +} + +/// Without `run_name` or `run_id` the time range must be explicit, and the +/// error names both run selectors. +#[tokio::test] +async fn get_data_requires_a_time_range_without_a_run() { + let (server, _h) = + server_with_mocks(MockAssetServiceImpl::new(), MockChannelServiceImpl::new()).await; + + let err = server + .get_data(run_params( + None, + None, + std::env::temp_dir().join("sift-mcp-get-data-test-never-written.parquet"), + )) + .await + .expect_err("no run and no time range"); + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); + assert!(err.message.contains("run_id"), "{}", err.message); +}