rust(fix): tell callers when get_data returned decimated data

rust/crates/sift_mcp/src/prompt/mod.rs

@@ -92,8 +92,9 @@ impl SiftMcpServer {
              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 \
              automatically; do not hand-compute timestamps. Use `channel_names` with exact \
-             channel names. Choose `sample_ms` to suit the run length: decimate (e.g. 100-1000 ms) \
-             for long runs and use 0 only when raw fidelity is required. Write to a Parquet path in a \
+             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 \
+             range or the channel set rather than decimating. Write to a Parquet path in a \
              working directory.\n\
              4. Summarize with `sql` against the `get_data` output: per-channel row count, min/max/mean, \
              and null rate, plus anything needed to answer the user's question. Keep \
@@ -140,7 +141,8 @@ impl SiftMcpServer {
              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 \
-             `channel_names` and a `sample_ms` suited to the transform.\n\
+             `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 \
              MUST be `timestamp_unix_nanos` (Int64, non-null). Project it first in the SELECT and never \
              rename or drop it. For aggregations that collapse rows, bucket on a time expression \

rust/crates/sift_mcp/src/service/data/mod.rs

@@ -99,6 +99,43 @@ impl fmt::Display for NoChannelData {
 
 impl std::error::Error for NoChannelData {}
 
+/// The decimation interval the service actually applied, gathered from
+/// `Metadata.sampled_ms` on every returned data page.
+///
+/// It arrives per page rather than per request, and the API ignores `sample_ms`
+/// for data types it cannot sample, so one request can come back decimated for a
+/// double channel and raw for a string one. Tracking the range keeps that visible
+/// instead of reporting whichever page happened to land last.
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub struct AppliedSampleMs {
+    lowest: Option<u32>,
+    highest: Option<u32>,
+}
+
+impl AppliedSampleMs {
+    fn observe(&mut self, sampled_ms: u32) {
+        self.lowest = Some(self.lowest.map_or(sampled_ms, |v| v.min(sampled_ms)));
+        self.highest = Some(self.highest.map_or(sampled_ms, |v| v.max(sampled_ms)));
+    }
+
+    /// The interval the pages reported, or `None` when none did. When pages
+    /// disagree this is the largest, which is the one that decides whether a
+    /// statistic drawn from the file can be trusted.
+    pub fn highest(&self) -> Option<u32> {
+        self.highest
+    }
+
+    /// True when any page came back decimated.
+    pub fn decimated(&self) -> bool {
+        self.highest.is_some_and(|v| v > 0)
+    }
+
+    /// True when some channels were decimated and others were not.
+    pub fn mixed(&self) -> bool {
+        matches!((self.lowest, self.highest), (Some(lo), Some(hi)) if lo != hi)
+    }
+}
+
 /// What `get_data` wrote, beyond the Parquet file itself.
 #[derive(Debug)]
 pub struct DataOutput {
@@ -107,6 +144,10 @@ pub struct DataOutput {
     /// caller diffing the Parquet schema against its request is the only way to
     /// notice them otherwise.
     pub empty_channels: Vec<String>,
+    /// What the service sampled at, as opposed to what was asked for. A caller
+    /// that never learns this quotes a mean off decimated data with no way to
+    /// know it is wrong.
+    pub applied_sample_ms: AppliedSampleMs,
 }
 
 pub enum ChannelInput {
@@ -330,6 +371,7 @@ impl DataService {
 
         let mut page_token = String::new();
         let mut columns = HashMap::<ColumnName, ChannelColumn>::new();
+        let mut applied_sample_ms = AppliedSampleMs::default();
 
         loop {
             let channel = self.channel.clone();
@@ -377,6 +419,7 @@ impl DataService {
                         let Some(metadata) = metadata else {
                             bail!("unexpected missing channel page metadata");
                         };
+                        applied_sample_ms.observe(metadata.sampled_ms);
 
                         let Some(channel) = metadata.channel else {
                             bail!("unexpected missing channel from metadata");
@@ -421,6 +464,7 @@ impl DataService {
                         let Some(metadata) = metadata else {
                             bail!("unexpected missing channel page metadata");
                         };
+                        applied_sample_ms.observe(metadata.sampled_ms);
 
                         let Some(channel) = metadata.channel else {
                             bail!("unexpected missing channel from metadata");
@@ -482,6 +526,7 @@ impl DataService {
                         let Some(metadata) = metadata else {
                             bail!("unexpected missing channel page metadata");
                         };
+                        applied_sample_ms.observe(metadata.sampled_ms);
 
                         let Some(channel) = metadata.channel else {
                             bail!("unexpected missing channel from metadata");
@@ -543,6 +588,7 @@ impl DataService {
                         let Some(metadata) = metadata else {
                             bail!("unexpected missing channel page metadata");
                         };
+                        applied_sample_ms.observe(metadata.sampled_ms);
 
                         let Some(channel) = metadata.channel else {
                             bail!("unexpected missing channel from metadata");
@@ -590,6 +636,7 @@ impl DataService {
                         let Some(metadata) = metadata else {
                             bail!("unexpected missing channel page metadata");
                         };
+                        applied_sample_ms.observe(metadata.sampled_ms);
 
                         let Some(channel) = metadata.channel else {
                             bail!("unexpected missing channel from metadata");
@@ -637,6 +684,7 @@ impl DataService {
                         let Some(metadata) = metadata else {
                             bail!("unexpected missing channel page metadata");
                         };
+                        applied_sample_ms.observe(metadata.sampled_ms);
 
                         let Some(channel) = metadata.channel else {
                             bail!("unexpected missing channel from metadata");
@@ -684,6 +732,7 @@ impl DataService {
                         let Some(metadata) = metadata else {
                             bail!("unexpected missing channel page metadata");
                         };
+                        applied_sample_ms.observe(metadata.sampled_ms);
 
                         let Some(channel) = metadata.channel else {
                             bail!("unexpected missing channel from metadata");
@@ -731,6 +780,7 @@ impl DataService {
                         let Some(metadata) = metadata else {
                             bail!("unexpected missing channel page metadata");
                         };
+                        applied_sample_ms.observe(metadata.sampled_ms);
 
                         let Some(channel) = metadata.channel else {
                             bail!("unexpected missing channel from metadata");
@@ -778,6 +828,7 @@ impl DataService {
                         let Some(metadata) = metadata else {
                             bail!("unexpected missing channel page metadata");
                         };
+                        applied_sample_ms.observe(metadata.sampled_ms);
 
                         let Some(channel) = metadata.channel else {
                             bail!("unexpected missing channel from metadata");
@@ -825,6 +876,7 @@ impl DataService {
                         let Some(metadata) = metadata else {
                             bail!("unexpected missing channel page metadata");
                         };
+                        applied_sample_ms.observe(metadata.sampled_ms);
 
                         let Some(channel) = metadata.channel else {
                             bail!("unexpected missing channel from metadata");
@@ -872,6 +924,7 @@ impl DataService {
                         let Some(metadata) = metadata else {
                             bail!("unexpected missing channel page metadata");
                         };
+                        applied_sample_ms.observe(metadata.sampled_ms);
 
                         let Some(channel) = metadata.channel else {
                             bail!("unexpected missing channel from metadata");
@@ -1082,7 +1135,10 @@ impl DataService {
             .close()
             .context("failed to finalize arrow writer")?;
 
-        Ok(DataOutput { empty_channels })
+        Ok(DataOutput {
+            empty_channels,
+            applied_sample_ms,
+        })
     }
 
     fn append_null_to_builder(

rust/crates/sift_mcp/src/service/data/test.rs

@@ -109,6 +109,45 @@ fn double_page(channel_id: &str, channel_name: &str, samples: Vec<(i64, f64)>) -
     }
 }
 
+/// Like `double_page`, but sets `Metadata.sampled_ms` — the rate the service
+/// says it actually applied, which is what the caller is told about.
+fn double_page_sampled(
+    channel_id: &str,
+    channel_name: &str,
+    sampled_ms: u32,
+    samples: Vec<(i64, f64)>,
+) -> Any {
+    let values = samples
+        .into_iter()
+        .map(|(ts_nanos, value)| {
+            let (seconds, nanos) = unix_nanos_to_secs_and_subsec_nanos(ts_nanos);
+            DoubleValue {
+                timestamp: Some(Timestamp { seconds, nanos }),
+                value,
+            }
+        })
+        .collect();
+
+    let payload = DoubleValues {
+        metadata: Some(Metadata {
+            channel: Some(metadata::Channel {
+                channel_id: channel_id.into(),
+                name: channel_name.into(),
+                ..Default::default()
+            }),
+            sampled_ms,
+            ..Default::default()
+        }),
+        values,
+        extras: vec![],
+    };
+
+    Any {
+        type_url: "sift.data.v2.DoubleValues".into(),
+        value: Bytes::from(payload.encode_to_vec()),
+    }
+}
+
 fn read_parquet(buffer: Vec<u8>) -> Vec<arrow::array::RecordBatch> {
     let reader = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(buffer))
         .expect("failed to open parquet")
@@ -689,3 +728,108 @@ async fn get_data_sends_saved_calculation_query() {
         schema.field(1).name(),
     );
 }
+
+/// The service reports the rate it applied on every page. A caller that is not
+/// told it holds decimated data has no way to know a mean drawn from the file is
+/// wrong, so the rate has to survive out of `get_data` rather than be discarded.
+#[tokio::test]
+async fn get_data_reports_the_applied_sample_rate() {
+    let mut mock = MockDataServiceImpl::new();
+    mock.expect_get_data().times(1).returning(|_| {
+        Ok(Response::new(GetDataResponse {
+            data: vec![double_page_sampled(
+                "c1",
+                "temp",
+                100,
+                vec![(1_000_000_000, 10.0), (2_000_000_000, 11.0)],
+            )],
+            next_page_token: String::new(),
+        }))
+    });
+
+    let (service, _h) = service_with_mock(mock).await;
+    let mut buffer = Vec::new();
+    let output = service
+        .get_data(
+            &[raw_channel("c1")],
+            asset_range(0, 3_000_000_000),
+            100,
+            &mut buffer,
+        )
+        .await
+        .expect("get_data failed");
+
+    assert_eq!(output.applied_sample_ms.highest(), Some(100));
+    assert!(output.applied_sample_ms.decimated());
+    assert!(!output.applied_sample_ms.mixed());
+}
+
+/// A raw pull must report itself as raw, not merely as "no rate reported".
+/// `decimated` false is the assurance a caller needs before quoting a statistic.
+#[tokio::test]
+async fn get_data_reports_raw_when_nothing_was_decimated() {
+    let mut mock = MockDataServiceImpl::new();
+    mock.expect_get_data().times(1).returning(|_| {
+        Ok(Response::new(GetDataResponse {
+            data: vec![double_page_sampled(
+                "c1",
+                "temp",
+                0,
+                vec![(1_000_000_000, 10.0)],
+            )],
+            next_page_token: String::new(),
+        }))
+    });
+
+    let (service, _h) = service_with_mock(mock).await;
+    let mut buffer = Vec::new();
+    let output = service
+        .get_data(
+            &[raw_channel("c1")],
+            asset_range(0, 3_000_000_000),
+            0,
+            &mut buffer,
+        )
+        .await
+        .expect("get_data failed");
+
+    assert_eq!(output.applied_sample_ms.highest(), Some(0));
+    assert!(!output.applied_sample_ms.decimated());
+    assert!(!output.applied_sample_ms.mixed());
+}
+
+/// The API ignores `sample_ms` for data types it cannot sample, so one request
+/// can come back decimated for one channel and raw for another. Reporting a
+/// single rate would describe half the file; `mixed` is what keeps the other
+/// half from being quoted as if it matched.
+#[tokio::test]
+async fn get_data_flags_a_file_that_mixes_decimated_and_raw_channels() {
+    let mut mock = MockDataServiceImpl::new();
+    mock.expect_get_data().times(1).returning(|_| {
+        Ok(Response::new(GetDataResponse {
+            data: vec![
+                double_page_sampled("c1", "sampled", 100, vec![(1_000_000_000, 10.0)]),
+                double_page_sampled("c2", "untouched", 0, vec![(1_000_000_000, 20.0)]),
+            ],
+            next_page_token: String::new(),
+        }))
+    });
+
+    let (service, _h) = service_with_mock(mock).await;
+    let mut buffer = Vec::new();
+    let output = service
+        .get_data(
+            &[raw_channel("c1"), raw_channel("c2")],
+            asset_range(0, 3_000_000_000),
+            100,
+            &mut buffer,
+        )
+        .await
+        .expect("get_data failed");
+
+    // The highest rate is the one that decides whether the file can be trusted,
+    // so a mixed result reports the decimated half rather than the raw one.
+    assert_eq!(output.applied_sample_ms.highest(), Some(100));
+    assert!(output.applied_sample_ms.decimated());
+    assert!(output.applied_sample_ms.mixed());
+}