fix(control-plane): reject cost evidence quoting another dataset

control_plane/src/physical/compiler.rs

@@ -648,6 +648,14 @@ pub enum CompileError {
     Query { query_id: String, reason: String },
     #[error("query {query_id}: TopK evidence is stale or invalid: {reason}")]
     InvalidEvidence { query_id: String, reason: String },
+    #[error(
+        "cost evidence for dataset {}/{} cannot price deployment for {}/{}",
+        found.namespace, found.dataset, expected.namespace, expected.dataset
+    )]
+    CostEvidenceDataset {
+        expected: planner_types::post_asap::LogicalDatasetIdentity,
+        found: planner_types::post_asap::LogicalDatasetIdentity,
+    },
     #[error("query {query_id}: lifecycle planning failed: {reason}")]
     Lifecycle { query_id: String, reason: String },
     #[error("failed to construct a physical materialization: {0}")]

control_plane/src/physical/workload_cost.rs

@@ -445,6 +445,17 @@ impl WorkloadCostEvidence {
         {
             return Err(invalid("missing, future or stale evidence generation"));
         }
+        // One generation prices one dataset; a foreign quote taints all of it.
+        if let Some(quote) = self
+            .quotes
+            .iter()
+            .find(|quote| quote.manifest.dataset_identity != env.dataset_identity)
+        {
+            return Err(CompileError::CostEvidenceDataset {
+                expected: env.dataset_identity.clone(),
+                found: quote.manifest.dataset_identity.clone(),
+            });
+        }
         Ok(())
     }
 
@@ -1806,6 +1817,40 @@ mod tests {
         assert!(error.contains("cost evidence compiler mismatch"), "{error}");
     }
 
+    #[test]
+    fn evidence_quoting_another_dataset_is_rejected_whole() {
+        let (candidates, env, evidence) = quoted();
+        assert!(select_lowest_cost_candidate(candidates.clone(), env.clone(), &evidence).is_ok());
+        let last = evidence.quotes.len() - 1;
+        for alter in [
+            |id: &mut planner_types::post_asap::LogicalDatasetIdentity| {
+                id.namespace.push_str("-other")
+            },
+            |id: &mut planner_types::post_asap::LogicalDatasetIdentity| {
+                id.dataset.push_str("-other")
+            },
+        ] {
+            let mut foreign = evidence.clone();
+            alter(&mut foreign.quotes[last].manifest.dataset_identity);
+            let found = foreign.quotes[last].manifest.dataset_identity.clone();
+            for error in [
+                select_lowest_cost_candidate(candidates.clone(), env.clone(), &foreign),
+                select_lowest_cost_metricsql_candidate(candidates.clone(), env.clone(), &foreign),
+            ]
+            .map(Result::unwrap_err)
+            {
+                assert!(
+                    matches!(
+                        &error,
+                        CompileError::CostEvidenceDataset { expected, found: f }
+                            if expected == &env.dataset_identity && f == &found
+                    ),
+                    "{error}"
+                );
+            }
+        }
+    }
+
     #[test]
     fn snapshot_supports_automatic_cost_and_roundtrips_quote_override() {
         let mut snapshot = fixture();