test: select, install and execute every admitted query-ensemble candidate

.github/workflows/promql-compliance.yml

@@ -0,0 +1,45 @@
+name: PromQL compliance
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened, ready_for_review]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  differential:
+    runs-on: ubuntu-latest
+    timeout-minutes: 90
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          path: ASAPQuery-backend
+      - uses: actions/checkout@v4
+        with:
+          repository: ProjectASAP/ASAPCollector
+          path: ASAPCollector
+      - uses: actions/checkout@v4
+        with:
+          repository: ProjectASAP/asap_sketchlib
+          path: asap_sketchlib
+      - uses: dtolnay/rust-toolchain@stable
+      - name: Install protobuf compiler
+        run: sudo apt-get update && sudo apt-get install -y protobuf-compiler
+      - name: Test Rust compliance harness
+        working-directory: ASAPQuery-backend
+        run: cargo test --locked -p promql-compliance
+      - name: Run every differential corpus
+        working-directory: ASAPQuery-backend/promql-compliance/runner
+        run: make run-all REPORT_DIR="$GITHUB_WORKSPACE/artifacts/reports" LOGS_DIR="$GITHUB_WORKSPACE/artifacts/logs"
+      - name: Execute every admitted individual and ensemble candidate
+        working-directory: ASAPQuery-backend/promql-compliance/runner
+        run: make candidates REPORT_DIR="$GITHUB_WORKSPACE/artifacts/reports" LOGS_DIR="$GITHUB_WORKSPACE/artifacts/logs"
+      - name: Upload reports and service logs
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: promql-compliance-evidence
+          path: artifacts
+          if-no-files-found: warn

Cargo.toml

@@ -5,6 +5,7 @@ members = [
     "crates/asap_types",
     "data_plane",
     "control_plane",
+    "promql-compliance/runner",
 ]
 
 [workspace.package]

control_plane/Dockerfile

@@ -15,8 +15,7 @@ RUN apt-get update && \
     rm -rf /var/lib/apt/lists/*
 
 COPY . ASAPQuery-backend
-
-RUN cd ASAPQuery-backend && cargo build --locked --release --bin control_plane
+RUN cd ASAPQuery-backend && cargo build --locked --release --bin control_plane --bin control_plane_quote_snapshot --example compile_workload_artifact
 
 # Runtime image: binary + CA certs.
 FROM debian:bookworm-slim
@@ -25,6 +24,10 @@ RUN apt-get update && \
     rm -rf /var/lib/apt/lists/*
 COPY --from=build /src/ASAPQuery-backend/target/release/control_plane \
     /usr/local/bin/control_plane
+COPY --from=build /src/ASAPQuery-backend/target/release/control_plane_quote_snapshot \
+    /usr/local/bin/control_plane_quote_snapshot
+COPY --from=build /src/ASAPQuery-backend/target/release/examples/compile_workload_artifact \
+    /usr/local/bin/compile_workload_artifact
 
 ENV RUST_LOG=info
 # OpAMP ws 4320, controller gRPC 4321, controller HTTP 8080.

control_plane/src/bin/control_plane_quote_snapshot.rs

@@ -0,0 +1,111 @@
+//! Complete a backend-local planning snapshot with deterministic unit-cost
+//! quotes for the PromQL compliance harness. Candidate identities and demand
+//! still come from the production planner and physical compiler.
+
+use control_plane::physical::{
+    compiler::{
+        BackendLocalPlanningInput, DeploymentPlanCompiler, BACKEND_REVISION, PLANNER_REVISION,
+    },
+    workload_cost::{
+        enumerate_exact_and_materialized_candidates, manifest, WorkloadCostEvidence, WorkloadQuote,
+    },
+};
+use std::{collections::BTreeMap, path::Path};
+
+fn quote_snapshot(
+    mut snapshot: BackendLocalPlanningInput,
+) -> Result<BackendLocalPlanningInput, String> {
+    if snapshot.workload_cost_evidence.is_some() {
+        return Err("planning snapshot already contains workload_cost_evidence".into());
+    }
+    let observed_at_unix_ms = snapshot.environment.observed_at_unix_ms;
+    let valid_for_ms = snapshot.environment.max_evidence_age_ms;
+    let (request, environment) = snapshot
+        .clone()
+        .into_physical_compilation_request()
+        .map_err(|error| error.to_string())?;
+    let quotes = enumerate_exact_and_materialized_candidates(request)
+        .map_err(|error| error.to_string())?
+        .into_iter()
+        .filter_map(|candidate| {
+            let plan = DeploymentPlanCompiler
+                .compile_promql(candidate.clone(), environment.clone())
+                .ok()?;
+            // This execution suite must exercise maintained candidates when legal.
+            let unit_cost = if plan.precompute_plan.materializations.is_empty() {
+                1e12
+            } else {
+                1.0
+            };
+            let manifest = manifest(&plan, &candidate.queries).ok()?;
+            Some(WorkloadQuote {
+                unit_costs: manifest
+                    .components
+                    .keys()
+                    .map(|key| (key.clone(), unit_cost))
+                    .collect::<BTreeMap<_, _>>(),
+                manifest,
+                executable: true,
+            })
+        })
+        .collect();
+    snapshot.workload_cost_evidence = Some(WorkloadCostEvidence {
+        backend_revision: BACKEND_REVISION.into(),
+        planner_revision: PLANNER_REVISION.into(),
+        data_snapshot_id: "promql-compliance".into(),
+        model_version: "promql-compliance-unit-costs".into(),
+        observed_at_unix_ms,
+        valid_for_ms,
+        quotes,
+    });
+    Ok(snapshot)
+}
+
+fn main() -> Result<(), String> {
+    let mut args = std::env::args_os().skip(1);
+    let input = args
+        .next()
+        .ok_or("usage: control_plane_quote_snapshot INPUT OUTPUT")?;
+    let output = args
+        .next()
+        .ok_or("usage: control_plane_quote_snapshot INPUT OUTPUT")?;
+    if args.next().is_some() {
+        return Err("usage: control_plane_quote_snapshot INPUT OUTPUT".into());
+    }
+    let snapshot = std::fs::read(&input)
+        .map_err(|error| format!("read {}: {error}", Path::new(&input).display()))?;
+    let snapshot =
+        serde_json::from_slice(&snapshot).map_err(|error| format!("decode snapshot: {error}"))?;
+    let quoted = quote_snapshot(snapshot)?;
+    std::fs::write(
+        &output,
+        serde_json::to_vec_pretty(&quoted).map_err(|error| error.to_string())?,
+    )
+    .map_err(|error| format!("write {}: {error}", Path::new(&output).display()))?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn adds_complete_candidate_quotes_to_a_snapshot() {
+        let fixture =
+            include_str!("../../../docs/examples/asapquery-compatibility-demo-snapshot.json");
+        let mut snapshot: BackendLocalPlanningInput = serde_json::from_str(fixture).unwrap();
+        snapshot.workload_cost_evidence = None;
+        let quoted = quote_snapshot(snapshot).unwrap();
+        assert!(!quoted
+            .workload_cost_evidence
+            .as_ref()
+            .unwrap()
+            .quotes
+            .is_empty());
+        let plan = quoted.compile_promql().unwrap();
+        assert!(
+            !plan.precompute_plan.materializations.is_empty(),
+            "deterministic compliance quotes should select a maintained candidate"
+        );
+    }
+}

data_plane/Dockerfile

@@ -20,8 +20,11 @@ RUN apt-get update && \
     rm -rf /var/lib/apt/lists/*
 
 COPY . ASAPQuery-backend
-
-RUN cd ASAPQuery-backend && cargo build --locked --release --bin data_plane
+RUN --mount=type=cache,target=/usr/local/cargo/registry \
+    --mount=type=cache,target=/usr/local/cargo/git \
+    --mount=type=cache,target=/src/ASAPQuery-backend/target \
+    cd ASAPQuery-backend && cargo build --locked --release --bin data_plane && \
+    install -m 755 target/release/data_plane /usr/local/bin/data_plane
 
 # Runtime image: binary + CA certs.
 FROM debian:bookworm-slim
@@ -30,7 +33,7 @@ RUN apt-get update && \
     rm -rf /var/lib/apt/lists/*
 # Default --output-dir for the rolling log file.
 RUN mkdir -p /var/log/asap
-COPY --from=build /src/ASAPQuery-backend/target/release/data_plane \
+COPY --from=build /usr/local/bin/data_plane \
     /usr/local/bin/data_plane
 
 ENV RUST_LOG=info \

data_plane/src/main.rs

@@ -461,11 +461,6 @@ fn validate_profile(args: &Args) -> Result<()> {
         )
         .into());
     }
-    if !args.forward_unsupported_queries {
-        return Err(
-            "--profile asapquery requires --forward-unsupported-queries for exact fallback".into(),
-        );
-    }
     let required_horizon = (args.precompute_allowed_lateness_ms.max(0) as u64)
         .saturating_add(args.remote_write_expected_retry_interval_ms);
     if args.remote_write_dedup_horizon_ms < required_horizon {
@@ -1392,6 +1387,19 @@ mod tests {
         .unwrap();
         assert!(validate_profile(&valid).is_ok());
 
+        let no_fallback = Args::try_parse_from([
+            "data_plane",
+            "--profile",
+            "asapquery",
+            "--physical-plan",
+            "plan.json",
+        ])
+        .unwrap();
+        assert!(
+            validate_profile(&no_fallback).is_ok(),
+            "a backend-local plan must be allowed to reject unsupported queries"
+        );
+
         assert!(
             Args::try_parse_from(["data_plane", "--streaming-config", "streaming.yaml"]).is_err()
         );

data_plane/tests/asapquery_compatibility_process_e2e.rs

@@ -210,7 +210,10 @@ fn unused_port() -> u16 {
 }
 
 async fn wait_until_ready(client: &reqwest::Client, url: &str, child: &mut Child) {
-    for _ in 0..120 {
+    // Startup compiles the complete workload candidate set before serving.
+    // This is a readiness budget, not a query latency assertion.
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(120);
+    while tokio::time::Instant::now() < deadline {
         if let Some(status) = child.try_wait().expect("inspect backend process") {
             panic!("backend exited before readiness: {status}");
         }

data_plane/tests/support/issue_701_702_process.rs

@@ -346,10 +346,21 @@ fn issue_701_702_uncertified_ratios_require_exact_fallback() {
         let candidates =
             workload_cost::enumerate_exact_and_materialized_candidates(request).unwrap();
         assert!(!candidates.is_empty());
+        let mut exact_count = 0;
         for candidate in candidates {
-            let plan = DeploymentPlanCompiler
-                .compile_promql(candidate, environment.clone())
-                .unwrap();
+            let plan = match DeploymentPlanCompiler.compile_promql(candidate, environment.clone()) {
+                Ok(plan) => plan,
+                Err(error) => {
+                    assert!(
+                        error
+                            .to_string()
+                            .contains("no certified accuracy guarantee"),
+                        "{query}: {error}"
+                    );
+                    continue;
+                }
+            };
+            exact_count += 1;
             assert!(plan.precompute_plan.materializations.is_empty(), "{query}");
             assert!(
                 plan.query_plan.entries.values().all(|entry| matches!(
@@ -359,6 +370,10 @@ fn issue_701_702_uncertified_ratios_require_exact_fallback() {
                 "{query}"
             );
         }
+        assert!(
+            exact_count > 0,
+            "{query}: exact execution must remain available"
+        );
     }
 }
 

docs/design_docs/evidence-dependent-candidates.md

@@ -17,14 +17,16 @@ could prove valid or deploys one whose guarantee has never been established.
 | --- | --- | --- |
 | Construct semantic candidates | Planner | Preserve unknown guarantees; reject known-invalid evidence and impossible shapes |
 | Supply external facts | Backend | Bind evidence to the query, data population, snapshot and validity period |
-| Derive accuracy and select logical roots | Planner, under backend models and policy | Respect the root accuracy target; missing proof is not certification |
+| Derive accuracy and expose supported candidates | Planner, under supplied facts and policy | Respect the root accuracy target; missing proof is not certification |
 | Bind and admit a deployment | Backend | Verify concrete execution support and compute complete workload resource cost |
 
-The backend still invokes Planner's workload search and global selection. It
-does not introduce a second semantic optimizer. Physical binding preserves the
-selected DAG's operators, grouping, windows and dependencies; a semantic change
-requires a new selection. Single-query and workload selection use the same
-costed Planner search; the first-candidate helper has been removed.
+Planner supplies the supported computation/physical candidate inventory. Backend
+binds and prices candidates, then selects a deployment. Physical binding preserves
+operators, grouping, windows and dependencies; it does not introduce another
+semantic optimizer. The current validation milestone uses synthetic complete
+quotes followed by execution of the selected typed plan. Online ERP observations,
+feedback-driven replanning and deployment replacement are deferred. Optional
+existing ERP adapters described below are not prerequisites for this milestone.
 
 ## Decision flow
 
@@ -34,7 +36,7 @@ flowchart TD
     Validate -->|Invalid supplied evidence| Error[Reject request with reason]
     Validate -->|Valid or absent evidence| Search[Planner search retains constructible candidates]
     Search --> Inspect[Explain known and unknown candidate properties]
-    Search --> Select[Planner global selection under backend policy]
+    Search --> Select[Backend candidate binding and cost selection]
     Select --> Exact[Explicit exact fallback when no certified summary is selected]
     Select --> Bind[Bind selected logical DAG to concrete execution]
     Bind --> Admit[Check guarantees, runtime support and computed workload cost]

docs/design_docs/physical-operators.md

@@ -68,6 +68,27 @@ Every installed range evaluation uses the shared DAG execution path and reports
 its actual local-summary and external-exact work. The differential and benefit
 runners require successful local execution evidence as well as matching values.
 Routing to the ASAP endpoint alone is insufficient.
+
+## Planner physical candidates and precompute boundaries
+
+Planner #462 exposes `physical_planner::compile_candidates` and
+`select_candidate`. A candidate carries precompute/query Physical DAGs and
+typed materialized outputs. For grouped rate, Planner can compile both
+per-series Rate → stored values → query-side Sum and precompute Rate → Sum →
+stored grouped values. Counter-state inputs remain explicit maintenance
+dependencies; raw-counter Sum before Rate is not equivalent.
+
+The backend reports source/state/operator feasibility and complete workload
+costs. `implementation.require_backend_local_execution` rejects external exact
+dependencies before pricing when the deployment has no such service. The runner
+sets this requirement because its deployment disables forwarding, and declares
+actual replay rate/cadence rather than synthetic compatibility demand.
+
+Planner capability does not prove that the backend can persist every value-output
+frontier. Result-row publication, revision/coverage and retention bindings must
+be admitted explicitly; the deployment compiler must reject unsupported
+frontiers instead of moving operators.
+
 Selected Filter, Sort, Limit and semi-join fragments are compiled by Planner and persisted
 with typed input contracts. Complete query candidates include Planner readouts
 that turn accumulator state into values; internal shared and stored edges retain