feat(control-plane): scope evidence and compute ERP/analytical workload costs

README.md

@@ -302,13 +302,16 @@ dot -Tsvg target/readme-evidence/selected.dot \
   -o target/readme-evidence/selected.svg
 ```
 
-Candidate discovery accepts the checked-in unquoted templates. Deployment and
-selected-plan inspection require `ASAPQUERY_PLANNING_SNAPSHOT` to point to a
-snapshot with complete, valid workload cost evidence. Prepare that input using
-the [cost evidence workflow](docs/examples/workload-cost-evidence.md).
-There is one snapshot compiler: it compares complete executable alternatives,
-including exact fallback. Materialization IDs are definitions, not physical SIDs.
-For `--metricsql`, collect quotes for the MetricsQL frontend.
+Candidate discovery and deployment accept snapshots without external workload
+quotes. The backend combines applicable ERP resources or analytical estimates
+with data size, query frequency and physical-plan structure, then compares
+complete candidate costs, including exact fallback. The selected plan includes
+the resource breakdown and assumptions in `cost_comparison`.
+
+Point `ASAPQUERY_PLANNING_SNAPSHOT` to your workload snapshot with current data
+and capability inputs. Optional calibrated provider quotes can override automatic
+costing through the [cost evidence workflow](docs/examples/workload-cost-evidence.md).
+Materialization IDs are definitions, not physical SIDs.
 
 ## Prometheus runbook
 

control_plane/src/main.rs

@@ -203,8 +203,12 @@ struct CompileAndPublishPhysicalPlanRequest {
     target_collector_ids: Vec<String>,
     capability_snapshot_id: String,
     #[serde(default)]
+    data_snapshot_id: Option<String>,
+    #[serde(default)]
     evidence: HashMap<String, physical::compiler::TopKMembershipEvidence>,
     #[serde(default)]
+    accuracy_evidence: HashMap<String, physical::compiler::ScopedAccuracyEvidence>,
+    #[serde(default)]
     exact_composition_costs:
         HashMap<String, Vec<physical::post_asap::cost_model::ExactCompositionCostEvidence>>,
     #[serde(default)]
@@ -588,7 +592,7 @@ fn compile_physical_plan_request(
             legacy_query_source: planner_types::pre_asap::Source::TimeSeries {
                 metric: query.metric,
             },
-            query_lookback_seconds: query.window_secs,
+            query_lookback_ms: query.window_secs.saturating_mul(1_000),
             group_by_labels: query.group_by,
             accuracy_target: query.accuracy,
             summary_lifecycle_inputs: query.lifecycle,
@@ -597,16 +601,54 @@ fn compile_physical_plan_request(
         });
     }
 
-    let planner_selection_trace = match physical::compiler::select_logical_roots_with_trace(
-        &mut queries,
-        canonical_roots.clone(),
-        &request.evidence,
-        &request.exact_composition_costs,
-        request.erp.as_ref(),
-    ) {
-        Ok(trace) => trace,
-        Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string().into())),
-    };
+    let scoped_snapshot_id = request.data_snapshot_id.as_deref().or_else(|| {
+        request
+            .workload_cost_evidence
+            .as_ref()
+            .map(|evidence| evidence.data_snapshot_id.as_str())
+    });
+    if request.data_snapshot_id.as_ref().is_some_and(|id| {
+        request
+            .workload_cost_evidence
+            .as_ref()
+            .is_some_and(|evidence| evidence.data_snapshot_id != *id)
+    }) {
+        return Err((
+            StatusCode::UNPROCESSABLE_ENTITY,
+            "accuracy evidence data snapshot differs from workload cost evidence".into(),
+        ));
+    }
+    for (query_id, evidence) in &request.accuracy_evidence {
+        let Some(query) = queries.iter().find(|query| &query.query_id == query_id) else {
+            return Err((
+                StatusCode::UNPROCESSABLE_ENTITY,
+                format!("accuracy evidence names unknown query {query_id}").into(),
+            ));
+        };
+        evidence
+            .validate(
+                query_id,
+                &query.query_string,
+                &request.data_workload,
+                scoped_snapshot_id,
+                now,
+                request.max_evidence_age_ms,
+            )
+            .map_err(|error| (StatusCode::UNPROCESSABLE_ENTITY, error.to_string().into()))?;
+    }
+    let planner_selection_trace =
+        match physical::compiler::select_logical_roots_with_scoped_evidence_and_trace(
+            &mut queries,
+            canonical_roots.clone(),
+            &request.evidence,
+            &request.accuracy_evidence,
+            &request.exact_composition_costs,
+            request.erp.as_ref(),
+            now,
+        ) {
+            Ok(trace) => trace,
+            Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string().into())),
+        };
 
     for (query, model) in queries.iter_mut().zip(window_models) {
         physical::compiler::prepare_window_implementations(query, &model, request.target, 0)
@@ -620,6 +662,7 @@ fn compile_physical_plan_request(
         queries,
         allow_mixed_summary_and_exact_execution: request.target
             == physical::compiler::PhysicalDeploymentTarget::BackendLocalRemoteWrite,
+        require_backend_local_execution: false,
         enabled_materialization_keys: None,
         topk_membership_evidence_by_query_id: request.evidence,
         exact_composition_costs: request.exact_composition_costs,
@@ -685,7 +728,7 @@ fn compile_physical_plan_request(
                 )
             }
         },
-        None => frontend.compile(compilation_request, environment),
+        None => physical::workload_cost::select_candidates(candidates, environment, None, frontend),
     };
     let bundle = match compiled {
         Ok(bundle) => bundle,
@@ -926,7 +969,7 @@ mod api_tests {
             "target": "backend_local_remote_write",
             "queries": [{
                 "query_id": query.query_id, "query_string": query.query_string,
-                "metric": metric, "window_secs": query.query_lookback_seconds, "accuracy": query.accuracy_target,
+                "metric": metric, "window_secs": query.query_lookback_ms / 1_000, "accuracy": query.accuracy_target,
                 "lifecycle": query.summary_lifecycle_inputs, "evaluation_phase_ms": 0, "window_cost_model": { "implementation_id": "test", "cost": query.window_realization_candidates[0].cost }
             }],
             "collector_ids": [], "capability_snapshot_id": "test",
@@ -938,6 +981,15 @@ mod api_tests {
         let (plan, collectors, _, _, _) =
             compile_physical_plan_request(request, false, QueryFrontend::PromQl).unwrap();
         let plan = plan.unwrap();
+        let report = plan
+            .cost_comparison
+            .as_ref()
+            .expect("HTTP must compare automatically priced candidates");
+        assert_eq!(report.model_version, "backend-workload-resources-v1");
+        assert!(report
+            .candidate_evaluations
+            .iter()
+            .any(|candidate| candidate.automatic_cost.is_some()));
         assert!(collectors.is_empty());
         assert!(plan.collector_plans.is_empty());
         assert_eq!(