diff --git a/.github/workflows/mvp-ci.yml b/.github/workflows/mvp-ci.yml
index 8cfc1d22e..925b83a98 100644
--- a/.github/workflows/mvp-ci.yml
+++ b/.github/workflows/mvp-ci.yml
@@ -85,9 +85,18 @@ jobs:
           CARGO_NET_GIT_FETCH_WITH_CLI: "true"
           # Give background flusher/sealer waits headroom on hosted runners.
           ASAP_TEST_TIMEOUT_SCALE: "4"
+          ASAP_LEVEL1_ARTIFACT_DIR: ${{ github.workspace }}/artifacts/issue754-level1
         working-directory: ASAPQuery-backend
         # Persistence tests wait on background flushers with bounded deadlines.
         # Match scripts/e2e.sh to avoid competing flushers exhausting those waits.
         run: |
           export ASAP_E2E_CONTROL_PLANE_BIN="$PWD/target/debug/control_plane"
           cargo test --workspace --locked -- --test-threads=1
+
+      - name: Upload issue 754 level-1 plans
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: issue754-level1-plans
+          path: artifacts/issue754-level1
+          if-no-files-found: ignore
diff --git a/control_plane/tests/issue754_level1.rs b/control_plane/tests/issue754_level1.rs
new file mode 100644
index 000000000..2fb329dad
--- /dev/null
+++ b/control_plane/tests/issue754_level1.rs
@@ -0,0 +1,1322 @@
+//! Level 1: inspect every supported candidate and binding rejection, without cost selection.
+use asap_types::sds::SummaryOperator;
+use control_plane::physical::compiler::{
+    CompiledPhysicalPlan, DeploymentPlanCompiler, QueryFrontend,
+};
+use control_plane::physical::executable_binding::validate_query_plan;
+use control_plane::physical::workload_cost::{
+    compile_candidates_for_pricing, enumerate_exact_and_materialized_candidates,
+    CandidateEvaluationStatus,
+};
+use control_plane::query_plan::QueryPlanNode;
+use planner_types::post_asap::{ExactKind, SketchAlgorithm, SummaryFamilyType};
+use serde_json::{json, Value};
+
+#[path = "support/issue754_workload.rs"]
+mod workload;
+use workload::Suite;
+
+/// Level 1 asserts *membership*: which shapes the candidate inventory must
+/// expose, and which it must refuse and why. It never names a winner, because
+/// ranking is #742's contract and this layer prices nothing.
+///
+/// A required shape declares how it must resolve. `MustBind` shapes have to
+/// reach `AwaitingQuote`; `MustReject` shapes have to appear in the inventory
+/// and then fail admission for the stated policy reason. The same shape can be
+/// `MustReject` under the strict fixture and `MustBind` under the certified
+/// one, so the fixture, not a second test path, carries the difference.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+enum Resolution {
+    MustBind,
+    MustReject(PolicyReason),
+}
+
+/// Which fixture a query is being planned under. The strict issue-754 generator
+/// certifies nothing; its certified companion separates the top-k boundary.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+enum Fixture {
+    Strict,
+    Certified,
+}
+
+/// Reasons a fixture may legitimately refuse an otherwise well-formed
+/// candidate. Anything outside this set is a defect, not a policy decision.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+enum PolicyReason {
+    /// The fixture supplies no evidence that certifies the summary readout.
+    NoCertifiedGuarantee,
+    /// The realized family cannot meet the query's declared accuracy target.
+    AccuracyTargetUnmet,
+    /// A workload candidate proposed sharing one deployed output between
+    /// queries that need different semantics from it. Refusing it is correct,
+    /// not a bug: the sharing would silently change what one of the queries
+    /// computes. Only the batch path can reach this, which is why the
+    /// single-query fixtures never produce it.
+    ConflictingSharedOutput,
+}
+
+impl PolicyReason {
+    fn matches(self, reason: &str) -> bool {
+        match self {
+            Self::NoCertifiedGuarantee => reason.contains("accuracy guarantee"),
+            Self::AccuracyTargetUnmet => reason.contains("does not satisfy"),
+            Self::ConflictingSharedOutput => {
+                reason.contains("one deployed output cannot have different semantic definitions")
+            }
+        }
+    }
+
+    const ALL: &'static [Self] = &[
+        Self::NoCertifiedGuarantee,
+        Self::AccuracyTargetUnmet,
+        Self::ConflictingSharedOutput,
+    ];
+}
+
+/// Sketch/heap families each query must expose, and how the given fixture is
+/// required to resolve them.
+///
+/// The inventory is the same either way -- Planner exposes the heap candidates
+/// regardless. What changes is the evidence: under `Strict` the fixture
+/// certifies nothing, so every heap family must be present *and* refused; under
+/// `Certified` the same families must bind. Asserting both directions is what
+/// keeps the refusal meaningful, since a heap that quietly left the inventory
+/// would otherwise satisfy the strict fixture on its own.
+fn required_summary_shapes(name: &str, fixture: Fixture) -> Vec<(&'static str, Resolution)> {
+    let families: &[&str] = match name {
+        "spatial-topk" => &["CountSketchWithHeap"],
+        "topk-rate" => &["CmsWithHeap", "CountSketchWithHeap"],
+        _ => &[],
+    };
+    families
+        .iter()
+        .map(|family| {
+            let resolution = match fixture {
+                Fixture::Strict => Resolution::MustReject(PolicyReason::NoCertifiedGuarantee),
+                Fixture::Certified => Resolution::MustBind,
+            };
+            (*family, resolution)
+        })
+        .collect()
+}
+
+/// Only explicit admission policies may reject a candidate. Loss of operator
+/// semantics or an incompatible Planner result schema is a regression.
+fn assert_rejection_is_accounted_for(name: &str, reason: &str) {
+    assert!(PolicyReason::ALL.iter().any(|policy| policy.matches(reason)),
+        "{name}: unexpected binding rejection: {reason}. Fix the compiler or input binding; do not allowlist the defect.");
+}
+
+struct ExpectedPlan {
+    family: Option<ExpectedFamily>,
+    partitioning: &'static str,
+    readout: &'static str,
+    root_operation: Option<&'static str>,
+}
+
+enum ExpectedFamily {
+    Exact(ExactKind),
+    QuantileSketch,
+}
+
+// Expectations describe semantic structure, never a cost-selected winner.
+fn expected_plan(name: &str) -> ExpectedPlan {
+    match name {
+        "spatial-sum" => ExpectedPlan {
+            family: Some(ExpectedFamily::Exact(ExactKind::Sum)),
+            partitioning: "grouped",
+            readout: "sum",
+            root_operation: None,
+        },
+        "spatial-topk" => ExpectedPlan {
+            family: None,
+            partitioning: "",
+            readout: "",
+            root_operation: None,
+        },
+        "spatial-quantile" => ExpectedPlan {
+            family: Some(ExpectedFamily::QuantileSketch),
+            partitioning: "grouped",
+            readout: "quantile",
+            root_operation: None,
+        },
+        "temporal-sum" => ExpectedPlan {
+            family: Some(ExpectedFamily::Exact(ExactKind::Sum)),
+            partitioning: "per_entity",
+            readout: "sum",
+            root_operation: None,
+        },
+        "temporal-quantile" => ExpectedPlan {
+            family: Some(ExpectedFamily::QuantileSketch),
+            partitioning: "per_entity",
+            readout: "quantile",
+            root_operation: None,
+        },
+        "temporal-rate" => ExpectedPlan {
+            family: Some(ExpectedFamily::Exact(ExactKind::Rate)),
+            partitioning: "per_entity",
+            readout: "rate",
+            root_operation: None,
+        },
+        "grouped-rate" => ExpectedPlan {
+            family: Some(ExpectedFamily::Exact(ExactKind::Rate)),
+            partitioning: "per_entity",
+            readout: "rate",
+            root_operation: Some("aggregate"),
+        },
+        "grouped-temporal-sum" => ExpectedPlan {
+            family: Some(ExpectedFamily::Exact(ExactKind::Sum)),
+            partitioning: "grouped",
+            readout: "sum",
+            root_operation: None,
+        },
+        "topk-rate" => ExpectedPlan {
+            family: Some(ExpectedFamily::Exact(ExactKind::Rate)),
+            partitioning: "per_entity",
+            readout: "rate",
+            root_operation: Some("limit"),
+        },
+        "quantile-ratio" => ExpectedPlan {
+            family: None,
+            partitioning: "",
+            readout: "",
+            root_operation: None,
+        },
+        other => panic!("no level-1 plan expectation for {other}"),
+    }
+}
+
+fn family_matches(expected: &ExpectedFamily, actual: &SummaryFamilyType) -> bool {
+    match (expected, actual) {
+        (ExpectedFamily::Exact(expected), SummaryFamilyType::ExactAggregate(actual, _)) => {
+            expected == actual
+        }
+        (ExpectedFamily::QuantileSketch, SummaryFamilyType::Sketch(kind, _)) => {
+            matches!(
+                kind.algorithm(),
+                SketchAlgorithm::DDSketch | SketchAlgorithm::Kll
+            )
+        }
+        _ => false,
+    }
+}
+
+// Inspect the Planner expression, since the adapter only stores sort direction.
+fn assert_rate_sort_expression(dag: &asap_types::executable_plan::OwnedPostAsapDag) {
+    for sort in &dag.nodes {
+        let Some(spec) = sort.payload["operation"].get("Sort") else {
+            continue;
+        };
+        let inputs: Vec<_> = dag
+            .edges
+            .iter()
+            .filter(|edge| edge.consumer == sort.id)
+            .collect();
+        assert_eq!(inputs.len(), 1, "rate ranking requires one producer");
+        let producer = dag
+            .nodes
+            .iter()
+            .find(|node| node.id == inputs[0].producer)
+            .unwrap();
+        assert_eq!(
+            producer.payload["operation"], "FinalizeExactAccumulator",
+            "ranking must consume finalized per-series rates"
+        );
+        let fields = producer.output_schema["fields"].as_array().unwrap();
+        let value = fields
+            .iter()
+            .position(|field| field["name"] == "value")
+            .unwrap();
+        assert_eq!(
+            spec["keys"],
+            json!([{
+                "ascending": false, "expr": {"Column": value}, "nulls_first": false
+            }]),
+            "TopK must rank rate values, not timestamps or labels"
+        );
+        let partition = fields
+            .iter()
+            .position(|field| field["name"] == "label_0")
+            .unwrap();
+        assert_eq!(spec["partition_by"], json!([partition]));
+    }
+}
+
+// A descending sort over a timestamp or label must fail the Level 1 contract.
+/// Compile the topk-rate query and return every ranking DAG the inventory
+/// exposes. Reading a committed export instead would pin the contract to plans
+/// generated by an older Planner revision; candidate identities do not survive
+/// a Planner bump, so the fixture has to be produced by the build under test.
+fn compiled_topk_rate_sort_dags() -> Vec<asap_types::executable_plan::OwnedPostAsapDag> {
+    let case = workload::suite()
+        .queries
+        .into_iter()
+        .find(|case| case.name == "topk-rate")
+        .expect("topk-rate left the issue-754 suite");
+    let (request, environment) = workload::input(&case)
+        .into_physical_compilation_request()
+        .unwrap();
+    let mut dags = Vec::new();
+    for candidate in enumerate_exact_and_materialized_candidates(request).unwrap() {
+        let Ok(plan) = DeploymentPlanCompiler.compile_promql(candidate, environment.clone()) else {
+            continue;
+        };
+        dags.extend(
+            plan.query_plan
+                .selected_dags
+                .values()
+                .filter(|dag| {
+                    dag.nodes
+                        .iter()
+                        .any(|node| node.payload["operation"].get("Sort").is_some())
+                })
+                .cloned(),
+        );
+    }
+    assert!(!dags.is_empty(), "topk-rate exposes no ranking DAG");
+    dags
+}
+
+#[test]
+fn topk_rate_sort_contract_rejects_wrong_value_expression() {
+    let dag = compiled_topk_rate_sort_dags().remove(0);
+    assert_rate_sort_expression(&dag);
+    for column in [0, 2] {
+        let mut wrong = dag.clone();
+        let sort = wrong
+            .nodes
+            .iter_mut()
+            .find(|node| node.payload["operation"].get("Sort").is_some())
+            .unwrap();
+        sort.payload["operation"]["Sort"]["keys"][0]["expr"] = json!({"Column": column});
+        assert!(std::panic::catch_unwind(|| assert_rate_sort_expression(&wrong)).is_err());
+    }
+}
+
+// Assert the selected native operator and its typed key/group positions, not just its name.
+fn assert_selection_fragment(node: &Value, expected: &str) {
+    use asap_physical_operators::physical_planner::CompiledPhysicalDag;
+    assert_eq!(node["op"], "physical_fragment");
+    let bytes: Vec<u8> = serde_json::from_value(node["dag"].clone()).unwrap();
+    let plan = CompiledPhysicalDag::decode(&bytes).unwrap();
+    let contracts = plan.input_contracts().collect::<Vec<_>>();
+    assert_eq!(contracts.len(), 1);
+    let schema = &contracts[0].1.schema;
+    let grouping = schema
+        .fields
+        .iter()
+        .position(|field| field.name == "label_0")
+        .unwrap();
+    let sample = schema
+        .fields
+        .iter()
+        .position(|field| {
+            field.dtype == SummaryFamilyType::Plain(planner_types::pre_asap::DataType::Float64)
+        })
+        .unwrap();
+    let encoded: Value = serde_json::from_slice(&bytes).unwrap();
+    let root = &encoded["nodes"][plan.roots()[0].to_string()]["Operator"];
+    assert_eq!(root["inputs"], json!([contracts[0].0]));
+    assert_eq!(plan.operator_name(plan.roots()[0]), Some(expected));
+    assert_eq!(
+        root["operator"]["kind"][expected],
+        if expected == "Limit" {
+            json!({"n":3,"offset":0,"groups":[grouping]})
+        } else {
+            json!({"keys":[{"column":sample,"descending":true,"nulls_first":false}],"groups":[grouping]})
+        }
+    );
+}
+
+fn assert_native_ranking(installed: &asap_types::query_plan::QueryPlanEntry) {
+    let physical = if installed.physical_vector_binding().is_some() {
+        installed.recover_vector_physical_dag().unwrap()
+    } else {
+        installed.recover_population_physical_dag().unwrap()
+    };
+    let inputs = physical.input_contracts().collect::<Vec<_>>();
+    assert_eq!(
+        inputs.len(),
+        1,
+        "spatial ranking binds one complete population"
+    );
+    let input = &inputs[0].1.schema;
+    assert!(input
+        .fields
+        .iter()
+        .any(|field| field.name == "$promql_series_identity"));
+    let value_column = input
+        .fields
+        .iter()
+        .position(|field| field.name == "value")
+        .unwrap();
+    let group_column = input
+        .fields
+        .iter()
+        .position(|field| field.name == "label_0")
+        .unwrap();
+    let program: Value = serde_json::from_slice(&physical.encode().unwrap()).unwrap();
+    let operations: Vec<_> = program["nodes"]
+        .as_object()
+        .unwrap()
+        .values()
+        .filter_map(|node| node.get("Operator"))
+        .collect();
+    assert_eq!(operations.len(), 2);
+    let sort = operations
+        .iter()
+        .find(|node| node["operator"]["kind"].get("Sort").is_some())
+        .unwrap();
+    assert_eq!(
+        sort["operator"]["kind"]["Sort"]["keys"],
+        json!([{"column":value_column,"descending":true,"nulls_first":false}])
+    );
+    assert_eq!(
+        sort["operator"]["kind"]["Sort"]["groups"],
+        json!([group_column])
+    );
+    let limit = operations
+        .iter()
+        .find(|node| node["operator"]["kind"].get("Limit").is_some())
+        .unwrap();
+    assert_eq!(
+        limit["operator"]["kind"]["Limit"],
+        json!({"n":3,"offset":0,"groups":[group_column]})
+    );
+    assert_eq!(sort["inputs"], json!([inputs[0].0]));
+    let sort_id: u64 = program["nodes"]
+        .as_object()
+        .unwrap()
+        .iter()
+        .find(|(_, node)| node.get("Operator") == Some(*sort))
+        .unwrap()
+        .0
+        .parse()
+        .unwrap();
+    assert_eq!(limit["inputs"], json!([sort_id]));
+}
+
+// Check the computation on each side of the persisted frontier, not just the
+// presence of Sum: per-series Rate must be finalized before grouped aggregation.
+fn assert_native_grouped_rate(plan: &CompiledPhysicalPlan) -> bool {
+    let entry = plan.query_plan.entries.values().next().unwrap();
+    let query = entry.recover_vector_physical_dag().unwrap();
+    let query: Value = serde_json::from_slice(&query.encode().unwrap()).unwrap();
+    let installed = &plan.precompute_plan.executable_dags[&entry.query_id];
+    let stored = !installed.native_programs.is_empty();
+    let maintenance;
+    let aggregation = if stored {
+        assert_eq!(installed.native_programs.len(), 1);
+        let sink = *installed.native_programs.keys().next().unwrap();
+        maintenance = serde_json::from_slice::<Value>(
+            &installed
+                .native_program(sink)
+                .unwrap()
+                .unwrap()
+                .encode()
+                .unwrap(),
+        )
+        .unwrap();
+        &maintenance
+    } else {
+        &query
+    };
+    let nodes = aggregation["nodes"].as_object().unwrap();
+    let sum = nodes
+        .values()
+        .filter_map(|node| node.get("Operator"))
+        .find(|op| op["operator"]["kind"].get("SummaryBuild").is_some())
+        .expect("grouped Rate candidate must explicitly build Sum");
+    let build = &sum["operator"]["kind"]["SummaryBuild"];
+    assert!(build["family"].to_string().contains("Sum"));
+    assert_eq!(sum["inputs"].as_array().unwrap().len(), 1);
+    let producer = &nodes[&sum["inputs"][0].to_string()];
+    let fields = producer
+        .get("Input")
+        .map(|input| &input["schema"]["fields"])
+        .unwrap_or(&producer["Operator"]["operator"]["output"]["fields"])
+        .as_array()
+        .unwrap();
+    let groups: Vec<_> = build["groups"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|column| {
+            fields[column.as_u64().unwrap() as usize]["name"]
+                .as_str()
+                .unwrap()
+        })
+        .collect();
+    assert_eq!(groups, ["label_0"]);
+    let value = build["value"].as_u64().unwrap() as usize;
+    assert_eq!(fields[value]["dtype"], json!({"Plain":"float64"}));
+    if stored {
+        let rate = &producer["Operator"]["operator"]["kind"]["Readout"];
+        assert_eq!(rate["statistic"], "Rate");
+        assert_eq!(rate["parameters"]["logical_lookback_ms"], "60000");
+        assert!(!query.to_string().contains("SummaryBuild"));
+        assert!(query.to_string().contains("Readout"));
+        assert!(
+            plan.precompute_plan
+                .materializations
+                .iter()
+                .all(|state| state.window_size == 60 && state.slide_interval == 10),
+            "stored Sum must cover each 60-second query window at the fixture's 10-second cadence"
+        );
+    } else {
+        assert!(
+            producer.get("Input").is_some(),
+            "Sum consumes the bound Rate vector"
+        );
+        let wire = serde_json::to_value(entry).unwrap();
+        let bound_nodes = wire["nodes"].as_object().unwrap();
+        let root = &bound_nodes[&wire["root"].to_string()];
+        assert_eq!(root["op"], "physical");
+        assert_eq!(root["inputs"].as_array().unwrap().len(), 1);
+        let rate = &bound_nodes[&root["inputs"][0].to_string()];
+        assert_eq!(rate["op"], "exact_readout");
+        assert_eq!(rate["readout"], "rate");
+        let state = &bound_nodes[&rate["input"].to_string()];
+        assert_eq!(state["op"], "read_materialization");
+        assert_eq!(state["binding"]["readout_lookback_ms"], 60_000);
+        assert_eq!(state["binding"]["output_grouping"]["mode"], "per_entity");
+    }
+    stored
+}
+
+fn assert_candidate_plan(name: &str, plan: &CompiledPhysicalPlan) -> Option<String> {
+    if name == "topk-rate" {
+        for dag in plan.query_plan.selected_dags.values() {
+            assert_rate_sort_expression(dag);
+        }
+    }
+    for materialization in &plan.precompute_plan.materializations {
+        let definition = plan
+            .summary_catalog
+            .outputs
+            .get(&materialization.policy_fingerprint().into())
+            .expect("precompute producer has no catalog definition");
+        let semantics = &plan.summary_catalog.definitions[&definition.definition_id];
+        if let asap_types::summary_semantics::SummarySemantics::Planner { fragment } =
+            &semantics.semantics
+        {
+            assert!(
+                fragment.dataset_identity.is_some(),
+                "persisted Planner output lacks dataset identity"
+            );
+            assert_eq!(
+                fragment.dataset_identity,
+                plan.precompute_plan.ingest.dataset_identity
+            );
+        }
+        assert_eq!(
+            semantics.id().unwrap(),
+            definition.definition_id,
+            "{name}: stored output's semantic identity must match its persisted description"
+        );
+        let writer = plan
+            .precompute_plan
+            .schemas
+            .iter()
+            .find(|schema| {
+                schema.materialization.fingerprint() == materialization.policy_fingerprint()
+            })
+            .unwrap();
+        assert_eq!(
+            writer.stored_output_reference.definition_id,
+            definition.definition_id
+        );
+        let descriptor =
+            &plan.summary_catalog.summary_descriptors[&definition.summary_descriptor_id];
+        let SummaryOperator::Configured { family, .. } = &descriptor.operator else {
+            panic!("{name}: producer catalog descriptor lacks Planner family");
+        };
+        assert_eq!(
+            family,
+            &materialization.accumulator_spec().unwrap().family,
+            "{name}: precompute producer and catalog disagree about Planner family"
+        );
+    }
+    if name == "grouped-rate"
+        && plan
+            .query_plan
+            .entries
+            .values()
+            .next()
+            .unwrap()
+            .physical_vector_binding()
+            .is_some()
+    {
+        assert_native_grouped_rate(plan);
+        return None;
+    }
+    let mut expected = expected_plan(name);
+    let artifact = serde_json::to_value(plan).unwrap();
+    let entries = artifact["query_plan"]["entries"].as_object().unwrap();
+    assert_eq!(entries.len(), 1, "{name}: expected one query plan");
+    let entry = entries.values().next().unwrap();
+    let nodes = entry["nodes"].as_object().unwrap();
+    let mut node = &nodes[&entry["root"].as_u64().unwrap().to_string()];
+    let materializations = artifact["precompute_plan"]["materializations"]
+        .as_array()
+        .unwrap();
+    if node["operator"]["kind"] == "current_series"
+        && matches!(name, "spatial-sum" | "spatial-quantile")
+    {
+        assert_eq!(nodes.len(), 1);
+        assert!(materializations.is_empty());
+        let population = &node["operator"]["population"];
+        assert_eq!(population["metric"], "data");
+        assert_eq!(
+            population["grouping"],
+            json!({"labels":["label_0"],"without":false})
+        );
+        assert_eq!(population["lookback_ms"], 5000);
+        assert_eq!(
+            node["operator"]["readout"],
+            if name == "spatial-sum" {
+                json!({"kind":"sum"})
+            } else {
+                json!({"kind":"quantile","q":0.9})
+            }
+        );
+        return None;
+    }
+    if name == "spatial-topk" {
+        assert_eq!(nodes.len(), 1, "{name}: unexpected query nodes");
+        assert_eq!(node["op"], "logical", "{name}: expected a local readout");
+        assert_eq!(node["operator"]["kind"], "current_series");
+        assert_eq!(node["operator"]["population"]["metric"], "data");
+        assert_eq!(
+            node["operator"]["population"]["grouping"],
+            json!({"labels":["label_0"],"without":false})
+        );
+        assert_eq!(node["operator"]["population"]["max_k"], 3);
+        assert_eq!(node["operator"]["readout"], json!({"kind":"snapshot"}));
+        let installed = plan.query_plan.entries.values().next().unwrap();
+        assert_native_ranking(installed);
+        assert!(
+            materializations.is_empty(),
+            "{name}: current-series readout has no summary producer"
+        );
+        return None;
+    }
+    if name == "grouped-temporal-sum" && node["op"] == "logical" {
+        // Both frontiers are legal: grouped maintained state, or maintained
+        // per-series temporal state followed by a query-side grouped Sum.
+        expected.partitioning = "per_entity";
+        expected.root_operation = Some("aggregate");
+    }
+    if name == "quantile-ratio" {
+        if node["op"] == "exact_fallback" {
+            return Some("quantile-ratio: expected two q=0.9/q=0.5 quantile sketch readouts followed by local division; Planner emitted exact fallback".into());
+        }
+        assert_eq!(node["op"], "binary", "{name}: expected local division");
+        assert!(
+            matches!(node["operator"].as_str(), Some("div" | "Div")),
+            "{name}: wrong binary operator"
+        );
+        let inputs = node["inputs"].as_array().unwrap();
+        assert_eq!(inputs.len(), 2);
+        for (input, q) in inputs.iter().zip([0.9, 0.5]) {
+            let readout = &nodes[&input.to_string()];
+            assert_eq!(readout["op"], "summary_estimate");
+            assert_eq!(readout["query"], json!({"kind":"quantile","q":q}));
+            let leaf = &nodes[&readout["input"].to_string()];
+            assert_eq!(leaf["op"], "read_materialization");
+            assert_eq!(leaf["binding"]["output_grouping"]["mode"], "per_entity");
+            assert_eq!(leaf["binding"]["readout_lookback_ms"], 60_000);
+        }
+        assert!(!materializations.is_empty());
+        assert!(plan.precompute_plan.materializations.iter().all(|summary| {
+            family_matches(
+                &ExpectedFamily::QuantileSketch,
+                &summary.accumulator_spec().unwrap().family,
+            ) && summary.metric == "data"
+                && summary.window_size == 60
+        }));
+        return None;
+    }
+    let native_rate = matches!(name, "topk-rate" | "temporal-rate") && node["op"] == "physical";
+    if native_rate {
+        let installed = plan.query_plan.entries.values().next().unwrap();
+        if name == "topk-rate" {
+            assert_native_ranking(installed);
+        } else {
+            // A complete Rate frontier can itself be the selected physical output.
+            let physical = installed.recover_vector_physical_dag().unwrap();
+            let sources = physical
+                .input_contracts()
+                .map(|(id, _)| id)
+                .collect::<Vec<_>>();
+            assert_eq!(sources.len(), 1);
+            assert_eq!(physical.roots(), sources.as_slice());
+            assert_eq!(physical.operator_name(sources[0]), Some("Input"));
+        }
+        let inputs = node["inputs"].as_array().unwrap();
+        assert_eq!(inputs.len(), 1);
+        node = &nodes[&inputs[0].to_string()];
+        expected.root_operation = None;
+    }
+    let Some(family) = expected.family.as_ref() else {
+        panic!("{name}: no physical plan contract");
+    };
+    if expected.root_operation == Some("limit") && node["op"] == "physical_fragment" {
+        assert_selection_fragment(node, "Limit");
+        let inputs = node["inputs"].as_array().unwrap();
+        assert_eq!(inputs.len(), 1);
+        node = &nodes[&inputs[0].to_string()];
+        assert_selection_fragment(node, "Sort");
+        let inputs = node["inputs"].as_array().unwrap();
+        assert_eq!(inputs.len(), 1);
+        node = &nodes[&inputs[0].to_string()];
+    } else if let Some(operation) = expected.root_operation {
+        assert_eq!(node["op"], "logical", "{name}: missing root operator");
+        assert_eq!(
+            node["operator"]["kind"], operation,
+            "{name}: wrong root operator"
+        );
+        if operation == "aggregate" {
+            assert_eq!(node["operator"]["operation"], "sum");
+        } else {
+            assert_eq!(node["operator"]["n"], 3, "{name}: wrong grouped limit");
+            assert_eq!(node["operator"]["offset"], 0);
+        }
+        assert_eq!(
+            node["operator"]["grouping"],
+            json!({"labels":["label_0"],"without":false}),
+            "{name}: wrong grouping"
+        );
+        let inputs = node["inputs"].as_array().unwrap();
+        assert_eq!(inputs.len(), 1, "{name}: root must have one input");
+        node = &nodes[&inputs[0].to_string()];
+        if operation == "limit" {
+            assert_eq!(node["op"], "logical");
+            assert_eq!(node["operator"]["kind"], "sort");
+            assert_eq!(node["operator"]["descending"], true);
+            assert_eq!(
+                node["operator"]["grouping"],
+                json!({"labels":["label_0"],"without":false})
+            );
+            let inputs = node["inputs"].as_array().unwrap();
+            assert_eq!(inputs.len(), 1);
+            node = &nodes[&inputs[0].to_string()];
+        }
+    }
+    assert_eq!(
+        nodes.len(),
+        if native_rate {
+            3
+        } else if expected.root_operation == Some("limit") {
+            4
+        } else if expected.root_operation.is_some() {
+            3
+        } else {
+            2
+        },
+        "{name}: unexpected DAG nodes: {nodes:?}"
+    );
+    if expected.readout == "quantile" {
+        assert_eq!(
+            node["op"], "summary_estimate",
+            "{name}: missing sketch readout"
+        );
+        assert_eq!(
+            node["query"],
+            json!({"kind":"quantile","q":0.9}),
+            "{name}: wrong quantile"
+        );
+    } else {
+        assert_eq!(node["op"], "exact_readout", "{name}: wrong readout node");
+        assert_eq!(node["readout"], expected.readout, "{name}: wrong readout");
+    }
+    let leaf = &nodes[&node["input"].to_string()];
+    assert_eq!(
+        leaf["op"], "read_materialization",
+        "{name}: missing summary read"
+    );
+    assert_eq!(
+        materializations.len(),
+        1,
+        "{name}: expected one summary producer"
+    );
+    let summary = &materializations[0];
+    let actual_family = plan.precompute_plan.materializations[0]
+        .accumulator_spec()
+        .unwrap()
+        .family;
+    assert!(
+        family_matches(family, &actual_family),
+        "{name}: wrong Planner family: {actual_family:?}"
+    );
+    assert_eq!(summary["metric"], "data", "{name}: wrong source metric");
+    assert_eq!(
+        summary["partitioning"], expected.partitioning,
+        "{name}: wrong population partitioning"
+    );
+    let spatial = expected.partitioning == "grouped";
+    // Grouping does not shorten a temporal range. window_size is the semantic
+    // window; the selected window_layout independently specifies stored panes.
+    if name == "grouped-temporal-sum" {
+        assert_eq!(leaf["binding"]["readout_lookback_ms"], 60_000);
+        assert_eq!(
+            leaf["binding"]["window_ms"],
+            plan.precompute_plan.materializations[0].stored_window_ms()
+        );
+    }
+    assert_eq!(
+        summary["window_size"],
+        if spatial && name != "grouped-temporal-sum" {
+            5
+        } else {
+            60
+        },
+        "{name}: wrong summary window"
+    );
+    assert_eq!(
+        leaf["binding"]["output_grouping"]["mode"],
+        if spatial { "reduce" } else { "per_entity" },
+        "{name}: wrong read grouping"
+    );
+    if spatial {
+        assert_eq!(summary["grouping_labels"]["labels"], json!(["label_0"]));
+        assert_eq!(
+            leaf["binding"]["output_grouping"]["keys"],
+            json!(["label_0"])
+        );
+    } else {
+        assert_eq!(
+            leaf["binding"]["readout_lookback_ms"], 60_000,
+            "{name}: wrong PromQL range"
+        );
+    }
+    None
+}
+
+/// The same ten expressions used by level 2 must compile to typed, connected plans.
+#[test]
+fn issue754_queries_have_valid_physical_plans() {
+    let suite: Suite = workload::suite();
+    assert_eq!(suite.queries.len(), 10, "the issue-754 contract changed");
+    for case in suite.queries {
+        let expected = expected_plan(&case.name);
+        let input = workload::input(&case);
+        let (request, environment) = input.clone().into_physical_compilation_request().unwrap();
+        if case.name == "spatial-quantile" {
+            // The fixture's admissible sketch families must reach deployment
+            // costing; the initially preferred family is not the inventory.
+            let mut families = std::collections::BTreeSet::new();
+            for forest in &request.planner_candidate_forests {
+                for query in forest {
+                    let dag =
+                        planner_types::post_asap::compile_executable_dag(&query.selected_plan_root)
+                            .unwrap();
+                    for node in dag.nodes {
+                        if let planner_types::post_asap::ExecutableOperatorPayload::SummaryAgg {
+                            family: SummaryFamilyType::Sketch(kind, _),
+                            ..
+                        } = node.payload
+                        {
+                            families.insert(format!("{:?}", kind.algorithm()));
+                        }
+                    }
+                }
+            }
+            assert!(
+                families.contains("Kll")
+                    || request
+                        .planner_selection_trace
+                        .iter()
+                        .flat_map(|trace| trace["groups"].as_array().into_iter().flatten())
+                        .flat_map(|group| group["rejected"].as_array().into_iter().flatten())
+                        .any(|rejection| rejection["description"]
+                            .as_str()
+                            .is_some_and(|s| s.contains("Kll"))
+                            && rejection["reason"]
+                                .as_str()
+                                .is_some_and(|s| s.contains("does not satisfy"))),
+                "a missing KLL candidate needs an explicit accuracy rejection"
+            );
+            let mut relaxed = input.clone();
+            relaxed.query_workload.repeating_queries.as_mut().unwrap()[0]
+                .requirements
+                .accuracy = planner_types::workload::AccuracyRequirement::Explicit(
+                planner_types::types::AccuracyTarget::Epsilon(0.05),
+            );
+            let (relaxed, _) = relaxed.into_physical_compilation_request().unwrap();
+            let roots = relaxed
+                .planner_candidate_forests
+                .iter()
+                .flatten()
+                .map(|query| format!("{:?}", query.selected_plan_root))
+                .collect::<Vec<_>>();
+            assert!(
+                roots.iter().any(|root| root.contains("Kll")),
+                "admissible KLL disappeared before deployment costing"
+            );
+            assert!(
+                roots.iter().any(|root| root.contains("DDSketch")),
+                "admissible DDSketch disappeared before deployment costing"
+            );
+            assert!(
+                families.contains("DDSketch"),
+                "DDSketch disappeared before deployment costing"
+            );
+            assert!(
+                request.planner_selection_trace.iter().any(|trace| trace
+                    ["computation_search_scope"]["joint_workload_search_exhaustive"]
+                    == false),
+                "root substitutions must not be reported as exhaustive joint search"
+            );
+        }
+        let heap_families = required_summary_shapes(&case.name, Fixture::Strict);
+        let mut heap_roots =
+            heap_families
+                .iter()
+                .map(|(family, resolution)| {
+                    let trace = request
+                        .planner_selection_trace
+                        .iter()
+                        .find(|trace| {
+                            trace["stage"] == "planner.physical_candidate"
+                                && trace["physical_dag"].to_string().contains(family)
+                        })
+                        .unwrap_or_else(|| {
+                            panic!("{}: Planner must expose native {family}", case.name)
+                        });
+                    let program =
+                        asap_physical_operators::physical_planner::CompiledPhysicalDag::decode(
+                            &serde_json::to_vec(&trace["physical_dag"]).unwrap(),
+                        )
+                        .unwrap();
+                    assert_eq!(program.input_contracts().count(), 1);
+                    let encoded = trace["physical_dag"].to_string();
+                    assert!(
+                        encoded.contains("KeyedSummaryBuild") && encoded.contains("KeyedReadout")
+                    );
+                    assert!(encoded.contains("$promql_series_identity"));
+                    assert!(
+                        !encoded.contains("CurrentSeries"),
+                        "the bound source supplies this evaluation's vector"
+                    );
+                    assert!(trace["guarantee"].to_string().contains("topk_max_distinct_items"),
+                "fixture lacks an enforced bound; cardinality estimates cannot certify a heap");
+                    (
+                        trace["logical_root_id"].as_str().unwrap().to_owned(),
+                        *resolution,
+                    )
+                })
+                .collect::<Vec<_>>();
+        if case.name == "topk-rate" {
+            for (family, resolution) in &heap_families {
+                let trace = request
+                    .planner_selection_trace
+                    .iter()
+                    .find(|trace| {
+                        trace["stage"] == "planner.physical_candidate"
+                            && trace["physical_candidate"].to_string().contains(family)
+                    })
+                    .unwrap_or_else(|| panic!("missing fixed-window {family} candidate"));
+                let split = asap_physical_operators::physical_planner::PhysicalCandidate::decode(
+                    &serde_json::to_vec(&trace["physical_candidate"]).unwrap(),
+                )
+                .unwrap();
+                let maintenance =
+                    String::from_utf8(split.precompute.as_ref().unwrap().encode().unwrap())
+                        .unwrap();
+                let query = String::from_utf8(split.query.encode().unwrap()).unwrap();
+                assert!(maintenance.contains("Rate") && maintenance.contains("KeyedSummaryBuild"));
+                assert!(query.contains("KeyedReadout") && !query.contains("KeyedSummaryBuild"));
+                assert_eq!(split.materialized_outputs.len(), 1);
+                assert!(split
+                    .query
+                    .input_contracts()
+                    .all(|(id, _)| split.materialized_outputs.contains_key(&id)));
+                heap_roots.push((
+                    trace["logical_root_id"].as_str().unwrap().to_owned(),
+                    *resolution,
+                ));
+            }
+        }
+        let candidates = enumerate_exact_and_materialized_candidates(request).unwrap();
+        let mut valid_plans = Vec::new();
+        let mut grouped_rate_placements = std::collections::BTreeSet::new();
+        let (_, admission) = compile_candidates_for_pricing(
+            candidates.clone(),
+            environment.clone(),
+            QueryFrontend::PromQl,
+        );
+        assert_eq!(admission.len(), candidates.len());
+        for result in &admission {
+            assert!(
+                result.total_cost.is_none(),
+                "Level 1 must not price candidates"
+            );
+            match result.status {
+                CandidateEvaluationStatus::AwaitingQuote => assert!(result.plan_id.is_some()),
+                CandidateEvaluationStatus::CompilationFailed => {
+                    let reason = result.unavailable_reason.as_deref().unwrap_or_default();
+                    assert!(
+                        !reason.is_empty(),
+                        "{}: bind failure without a reason",
+                        case.name
+                    );
+                    assert_rejection_is_accounted_for(&case.name, reason);
+                }
+                ref status => panic!("unexpected pre-pricing status: {status:?}"),
+            }
+        }
+        if let Ok(directory) = std::env::var("ASAP_LEVEL1_ARTIFACT_DIR") {
+            // Mirror the committed layout: `admission/` is the contract, so a
+            // downloaded CI artifact drops straight onto the repository copy.
+            let admission_dir = std::path::Path::new(&directory).join("admission");
+            std::fs::create_dir_all(&admission_dir).unwrap();
+            std::fs::write(
+                admission_dir.join(format!("{}.admission.json", case.name)),
+                serde_json::to_vec_pretty(&admission).unwrap(),
+            )
+            .unwrap();
+        }
+        let mut errors = Vec::new();
+        for (candidate_index, candidate) in candidates.into_iter().enumerate() {
+            match DeploymentPlanCompiler.compile_promql(candidate.clone(), environment.clone()) {
+                Ok(plan) => {
+                    let entry = plan.query_plan.lookup(&case.expr).unwrap();
+                    assert_eq!(entry.canonical_query, case.expr);
+                    assert!(
+                        entry.nodes.contains_key(&entry.root),
+                        "query root must exist"
+                    );
+                    if let Some(installed) =
+                        plan.precompute_plan.executable_dags.get(&entry.query_id)
+                    {
+                        installed.validate().expect("typed DAG is valid");
+                        validate_query_plan(installed, entry).expect("DAG/query bindings agree");
+                    } else {
+                        assert!(
+                            plan.precompute_plan.materializations.is_empty(),
+                            "summary plan must retain its Planner DAG"
+                        );
+                    }
+                    {
+                        let local = entry.nodes.values().all(|node| !matches!(node,
+                            QueryPlanNode::ExactFallback { .. } | QueryPlanNode::Logical {
+                                operator: control_plane::query_plan::residual::ResidualQueryOperator::ExactSubquery { .. }
+                                    | control_plane::query_plan::residual::ResidualQueryOperator::CandidateExactSubquery { .. }, .. }));
+                        if local {
+                            assert_eq!(
+                                assert_candidate_plan(&case.name, &plan),
+                                None,
+                                "every admitted local candidate must preserve query semantics"
+                            );
+                        }
+                    }
+                    if case.name == "grouped-rate" && entry.physical_vector_binding().is_some() {
+                        grouped_rate_placements.insert(assert_native_grouped_rate(&plan));
+                    }
+                    let dot = control_plane::physical::plan_dot::render(&plan);
+                    assert!(dot.contains("PrecomputePlan") && dot.contains("QueryPlan:"));
+                    if let Ok(directory) = std::env::var("ASAP_LEVEL1_ARTIFACT_DIR") {
+                        let base = std::path::Path::new(&directory)
+                            .join("candidates")
+                            .join(format!("{}-{candidate_index}", case.name));
+                        std::fs::create_dir_all(base.parent().unwrap()).unwrap();
+                        std::fs::write(
+                            base.with_extension("json"),
+                            serde_json::to_vec_pretty(&plan).unwrap(),
+                        )
+                        .unwrap();
+                        std::fs::write(base.with_extension("dot"), &dot).unwrap();
+                    }
+                    valid_plans.push(plan);
+                }
+                Err(error) => errors.push(error.to_string()),
+            }
+        }
+        if case.name == "grouped-rate" {
+            assert_eq!(
+                grouped_rate_placements,
+                std::collections::BTreeSet::from([false, true]),
+                "Planner must expose query-time and precomputed grouped Rate/Sum: {errors:?}"
+            );
+        }
+        assert!(
+            !valid_plans.is_empty(),
+            "{} has no valid physical plan: {errors:?}",
+            case.name
+        );
+        if let Some(family) = expected.family.as_ref() {
+            assert!(
+                valid_plans.iter().any(|plan| {
+                    plan.precompute_plan
+                        .materializations
+                        .iter()
+                        .any(|m| family_matches(family, &m.accumulator_spec().unwrap().family))
+                        && plan.query_plan.entries.values().all(|entry| {
+                            entry.nodes.values().any(|node| {
+                                matches!(node, QueryPlanNode::ReadMaterialization { .. })
+                            }) && !entry
+                                .nodes
+                                .values()
+                                .any(|node| matches!(node, QueryPlanNode::ExactFallback { .. }))
+                        })
+                }),
+                "{} lacks a readable summary candidate: {errors:?}",
+                case.name
+            );
+        }
+        for (root_id, resolution) in heap_roots {
+            let heap = admission
+                .iter()
+                .filter(|candidate| candidate.logical_root_ids.contains(&root_id))
+                .collect::<Vec<_>>();
+            assert!(
+                !heap.is_empty(),
+                "physical heap candidate disappeared before admission"
+            );
+            match resolution {
+                Resolution::MustBind => assert!(
+                    heap.iter().all(|candidate| candidate.status
+                        == CandidateEvaluationStatus::AwaitingQuote
+                        && candidate.plan_id.is_some()
+                        && candidate.total_cost.is_none()),
+                    "certified shape must reach pricing unpriced: {heap:?}"
+                ),
+                Resolution::MustReject(policy) => assert!(
+                    heap.iter().all(|candidate| candidate.status
+                        == CandidateEvaluationStatus::CompilationFailed
+                        && candidate.total_cost.is_none()
+                        && candidate
+                            .unavailable_reason
+                            .as_deref()
+                            .is_some_and(|reason| policy.matches(reason))),
+                    "missing proof must be an explicit {policy:?} admission failure: {heap:?}"
+                ),
+            }
+        }
+    }
+}
+
+/// The other half of the membership contract: under a fixture that *can*
+/// certify a heap, the shapes the strict fixture must refuse have to bind.
+///
+/// This is what keeps `MustReject(NoCertifiedGuarantee)` honest. Without it,
+/// a heap candidate that vanished from the inventory, or one refused for some
+/// unrelated reason, would still satisfy the strict fixture.
+#[test]
+fn certified_fixture_admits_the_heaps_the_strict_fixture_refuses() {
+    for name in ["spatial-topk", "topk-rate"] {
+        let case = workload::suite()
+            .queries
+            .into_iter()
+            .find(|case| case.name == name)
+            .unwrap_or_else(|| panic!("{name} left the issue-754 suite"));
+        let (request, environment) = workload::certified_topk_input(&case)
+            .into_physical_compilation_request()
+            .unwrap();
+        let shapes = required_summary_shapes(name, Fixture::Certified);
+        let heap_roots: Vec<(String, Resolution)> = shapes
+            .iter()
+            .map(|(family, resolution)| {
+                let trace = request
+                    .planner_selection_trace
+                    .iter()
+                    .find(|trace| {
+                        trace["stage"] == "planner.physical_candidate"
+                            && trace["physical_dag"].to_string().contains(family)
+                    })
+                    .unwrap_or_else(|| panic!("{name}: Planner must expose native {family}"));
+                (
+                    trace["logical_root_id"].as_str().unwrap().to_owned(),
+                    *resolution,
+                )
+            })
+            .collect();
+        let candidates = enumerate_exact_and_materialized_candidates(request).unwrap();
+        let (_, admission) =
+            compile_candidates_for_pricing(candidates, environment, QueryFrontend::PromQl);
+        assert!(
+            !admission.iter().any(|candidate| candidate
+                .unavailable_reason
+                .as_deref()
+                .is_some_and(|reason| PolicyReason::NoCertifiedGuarantee.matches(reason))),
+            "{name}: certified evidence must remove every uncertified-readout refusal"
+        );
+        for (root_id, resolution) in heap_roots {
+            let heap: Vec<_> = admission
+                .iter()
+                .filter(|candidate| candidate.logical_root_ids.contains(&root_id))
+                .collect();
+            assert!(
+                !heap.is_empty(),
+                "{name}: heap candidate left the inventory"
+            );
+            assert_eq!(
+                resolution,
+                Resolution::MustBind,
+                "the certified fixture declares binding shapes"
+            );
+            assert!(
+                heap.iter().any(|candidate| candidate.status
+                    == CandidateEvaluationStatus::AwaitingQuote
+                    && candidate.plan_id.is_some()
+                    && candidate.total_cost.is_none()),
+                "{name}: certified heap must reach pricing, unpriced: {heap:?}"
+            );
+        }
+    }
+}
+
+/// Ensembles retain every query and coherent shared producer bindings in each
+/// exposed workload candidate. This does not claim exhaustive joint search.
+#[test]
+fn ensembles_preserve_all_queries_and_shared_output_identity() {
+    for (name, cases) in workload::ensembles() {
+        let (request, env) = workload::ensemble_input(&cases)
+            .into_physical_compilation_request()
+            .unwrap();
+        assert_eq!(request.queries.len(), cases.len());
+        let request_traces = request.planner_selection_trace.clone();
+        let candidates = enumerate_exact_and_materialized_candidates(request).unwrap();
+        let (_, admission) =
+            compile_candidates_for_pricing(candidates.clone(), env.clone(), QueryFrontend::PromQl);
+        assert_eq!(admission.len(), candidates.len());
+        // The same discipline as the single-query path. A workload candidate's
+        // rejection is attributed to the ensemble, since it covers every query
+        // in it: either a declared policy refusal or a recorded defect, never
+        // an unexplained bind failure.
+        for result in &admission {
+            assert!(
+                result.total_cost.is_none(),
+                "{name}: Level 1 must not price workload candidates"
+            );
+            if result.status == CandidateEvaluationStatus::CompilationFailed {
+                let reason = result.unavailable_reason.as_deref().unwrap_or_default();
+                assert!(!reason.is_empty(), "{name}: bind failure without a reason");
+                assert_rejection_is_accounted_for(&name, reason);
+            }
+        }
+        // Membership holds for the batch too: a heap family that Planner exposes
+        // for a member query must still appear in the workload inventory, and
+        // must still be refused for the declared reason.
+        for case in &cases {
+            for (family, resolution) in required_summary_shapes(&case.name, Fixture::Strict) {
+                let Some(trace) = request_traces.iter().find(|trace| {
+                    trace["stage"] == "planner.physical_candidate"
+                        && trace["physical_dag"].to_string().contains(family)
+                }) else {
+                    panic!("{name}: {} lost its native {family}", case.name);
+                };
+                let root_id = trace["logical_root_id"].as_str().unwrap().to_owned();
+                let heap: Vec<_> = admission
+                    .iter()
+                    .filter(|candidate| candidate.logical_root_ids.contains(&root_id))
+                    .collect();
+                assert!(
+                    !heap.is_empty(),
+                    "{name}: {} heap candidate left the workload inventory",
+                    case.name
+                );
+                let Resolution::MustReject(policy) = resolution else {
+                    panic!("the strict fixture declares refusing shapes");
+                };
+                assert!(
+                    heap.iter().all(|candidate| candidate.status
+                        == CandidateEvaluationStatus::CompilationFailed
+                        && candidate.total_cost.is_none()
+                        && candidate
+                            .unavailable_reason
+                            .as_deref()
+                            .is_some_and(|reason| policy.matches(reason))),
+                    "{name}: {} heap must stay an explicit {policy:?} refusal in the workload: {heap:?}",
+                    case.name
+                );
+            }
+        }
+        let mut bound = 0;
+        let mut shared = false;
+        for (index, candidate) in candidates.into_iter().enumerate() {
+            let Ok(plan) = DeploymentPlanCompiler.compile_promql(candidate, env.clone()) else {
+                assert_eq!(
+                    admission[index].status,
+                    CandidateEvaluationStatus::CompilationFailed
+                );
+                assert!(admission[index]
+                    .unavailable_reason
+                    .as_ref()
+                    .is_some_and(|s| !s.is_empty()));
+                continue;
+            };
+            bound += 1;
+            assert_eq!(plan.query_plan.entries.len(), cases.len());
+            assert_eq!(
+                plan.precompute_plan.ingest.dataset_identity.as_ref(),
+                Some(&env.dataset_identity)
+            );
+            for definition in plan.summary_catalog.definitions.values() {
+                if let asap_types::summary_semantics::SummarySemantics::Planner { fragment } =
+                    &definition.semantics
+                {
+                    assert_eq!(
+                        fragment.dataset_identity.as_ref(),
+                        Some(&env.dataset_identity)
+                    );
+                }
+            }
+            let mut consumers = std::collections::BTreeMap::new();
+            for case in &cases {
+                let entry = plan
+                    .query_plan
+                    .lookup(&case.expr)
+                    .expect("ensemble query disappeared");
+                assert!(entry.nodes.contains_key(&entry.root));
+                if let Some(dag) = plan.precompute_plan.executable_dags.get(&entry.query_id) {
+                    dag.validate().unwrap();
+                    validate_query_plan(dag, entry).unwrap();
+                }
+                for binding in entry.materialization_bindings() {
+                    let reference = &binding.stored_output_reference;
+                    let output = &plan.summary_catalog.outputs[&reference.stored_output_id];
+                    assert_eq!(reference.definition_id, output.definition_id);
+                    consumers
+                        .entry(reference.stored_output_id)
+                        .or_insert_with(std::collections::BTreeSet::new)
+                        .insert(entry.query_id.clone());
+                }
+            }
+            let producers: std::collections::BTreeSet<_> = plan
+                .precompute_plan
+                .materializations
+                .iter()
+                .map(|m| m.policy_fingerprint())
+                .collect();
+            assert_eq!(producers.len(), plan.precompute_plan.materializations.len());
+            shared |= consumers.values().any(|readers| readers.len() > 1);
+            if let Ok(directory) = std::env::var("ASAP_LEVEL1_ARTIFACT_DIR") {
+                let root = std::path::Path::new(&directory)
+                    .join("ensembles")
+                    .join(&name);
+                std::fs::create_dir_all(&root).unwrap();
+                std::fs::write(
+                    root.join(format!("candidate-{index}.json")),
+                    serde_json::to_vec_pretty(&plan).unwrap(),
+                )
+                .unwrap();
+                std::fs::write(
+                    root.join(format!("candidate-{index}.dot")),
+                    control_plane::physical::plan_dot::render(&plan),
+                )
+                .unwrap();
+                std::fs::write(
+                    root.join("admission.json"),
+                    serde_json::to_vec_pretty(&admission).unwrap(),
+                )
+                .unwrap();
+            }
+        }
+        assert!(bound > 0, "{name}: no bound ensemble candidate");
+        if name == "shared-rate" {
+            assert!(shared, "Rate consumers never share a stored producer");
+        }
+    }
+}
diff --git a/control_plane/tests/support/issue754_workload.rs b/control_plane/tests/support/issue754_workload.rs
new file mode 100644
index 000000000..f4b85282d
--- /dev/null
+++ b/control_plane/tests/support/issue754_workload.rs
@@ -0,0 +1,199 @@
+//! Shared query/accuracy fixture; contains no candidate prices or expected winner.
+//!
+//! This module is `#[path]`-included by each level's test binary, so every
+//! binary compiles all of it while using only the parts that level needs.
+//! Unused-code lints are therefore about the including binary, not about the
+//! fixture, and under `-D warnings` they would fail a level for helpers that
+//! another level relies on.
+#![allow(dead_code)]
+use control_plane::physical::compiler::BackendLocalPlanningInput;
+use serde::Deserialize;
+use serde_json::{json, Value};
+
+#[derive(Deserialize)]
+pub struct Suite {
+    pub queries: Vec<Case>,
+}
+#[derive(Deserialize)]
+pub struct Case {
+    pub name: String,
+    pub expr: String,
+}
+
+pub fn suite() -> Suite {
+    serde_yaml::from_str(include_str!(
+        "../../../promql-compliance/suites/issue-754.yaml"
+    ))
+    .unwrap()
+}
+
+pub fn input(case: &Case) -> BackendLocalPlanningInput {
+    let mut snapshot: Value = serde_json::from_str(include_str!(
+        "../../../docs/examples/asapquery-planning-snapshot.json"
+    ))
+    .unwrap();
+    snapshot["query_workload"]["repeating_queries"][0]["query"] = case.expr.clone().into();
+    if case.name == "quantile-ratio" {
+        // The issue-754 generator defines the entire positive finite input
+        // population. Supply its domain contract rather than certifying a
+        // ratio from sample observations or weakening the admission rule.
+        let fixture: Value = serde_yaml::from_str(include_str!(
+            "../../../promql-compliance/datasets/issue-754.yaml"
+        ))
+        .unwrap();
+        let mut lower = f64::INFINITY;
+        let mut upper = f64::NEG_INFINITY;
+        let mut count = 0u64;
+        for series in fixture["series"].as_array().unwrap() {
+            let g = &series["generated_samples"];
+            let n = |k: &str| g[k].as_f64().unwrap();
+            assert!(n("multiplier") > 0.0 && n("modulo") > 0.0 && n("base") > 0.0);
+            lower = lower.min(n("multiplier") * n("base"));
+            upper = upper.max(n("multiplier") * (n("base") + n("modulo")));
+            count += ((n("end_offset_seconds") - n("start_offset_seconds")) / n("step_seconds"))
+                .round() as u64
+                + 1;
+        }
+        let root = control_plane::query_parser::parse_query_expr_canonical(
+            &case.expr,
+            planner_types::types::AccuracyTarget::EpsilonDelta {
+                epsilon: 0.01,
+                delta: 0.01,
+            },
+        )
+        .unwrap();
+        let planner_types::pre_asap::QueryExpr::BinaryOp { lhs, rhs, .. } = root else {
+            panic!("ratio fixture");
+        };
+        snapshot["implementation"]["data_snapshot_id"] = json!("issue-754-level1");
+        snapshot["implementation"]["accuracy_evidence"][&case.expr] = json!({
+            "query_string":case.expr,"data_snapshot_id":"issue-754-level1",
+            "data_workload":snapshot["data_workload"],"source":"issue-754-finite-generator",
+            "observed_at_unix_ms":9500,"valid_for_ms":60000,
+            "quantile_operand_domains":([lhs,rhs].into_iter().map(|operand| json!({
+                "operand":operand,"lower":lower,"upper":upper,"max_samples":count,
+                "contract":"complete finite issue-754 generator population"})).collect::<Vec<_>>())
+        });
+    }
+    serde_json::from_value(snapshot).unwrap()
+}
+
+/// Certified companion fixture for the two top-k queries.
+///
+/// The issue-754 generator cannot certify a heap: its per-series value domains
+/// overlap, so no scalar `topk_selected_lower_bound > topk_excluded_upper_bound`
+/// holds across the validity window, and these queries evaluate in `real_time`
+/// scope rather than at one instant. `issue-754-certified-topk` gives each
+/// series a decade of its own so the domains are disjoint by construction.
+///
+/// Every number below is derived from that dataset, never chosen: the bounds
+/// come from the same `multiplier * base` .. `multiplier * (base + modulo)`
+/// domain this file already uses for the quantile operands, and the distinct
+/// item count is the series count. The assertion at the end is what makes this
+/// a contract rather than a guess -- if the dataset stops separating, the
+/// fixture fails instead of certifying something false.
+#[allow(dead_code)] // Shared fixture module: only the admission suite needs certification.
+pub fn certified_topk_input(case: &Case) -> BackendLocalPlanningInput {
+    const K: usize = 3;
+    let mut snapshot: Value = serde_json::from_str(include_str!(
+        "../../../docs/examples/asapquery-planning-snapshot.json"
+    ))
+    .unwrap();
+    snapshot["query_workload"]["repeating_queries"][0]["query"] = case.expr.clone().into();
+    let fixture: Value = serde_yaml::from_str(include_str!(
+        "../../../promql-compliance/datasets/issue-754-certified-topk.yaml"
+    ))
+    .unwrap();
+    let series = fixture["series"].as_array().unwrap();
+    let mut groups: std::collections::BTreeMap<String, Vec<(f64, f64)>> = Default::default();
+    for entry in series {
+        let g = &entry["generated_samples"];
+        let n = |k: &str| g[k].as_f64().unwrap();
+        assert!(n("multiplier") > 0.0 && n("modulo") > 0.0 && n("base") > 0.0);
+        groups
+            .entry(entry["labels"]["label_0"].as_str().unwrap().to_owned())
+            .or_default()
+            .push((
+                n("multiplier") * n("base"),
+                n("multiplier") * (n("base") + n("modulo")),
+            ));
+    }
+    let mut selected_lower = f64::INFINITY;
+    let mut excluded_upper = f64::NEG_INFINITY;
+    for domains in groups.values_mut() {
+        domains.sort_by(|a, b| b.0.total_cmp(&a.0));
+        for (rank, (lower, upper)) in domains.iter().enumerate() {
+            if rank < K {
+                selected_lower = selected_lower.min(*lower);
+            } else {
+                excluded_upper = excluded_upper.max(*upper);
+            }
+        }
+    }
+    assert!(
+        selected_lower > excluded_upper,
+        "certified fixture does not separate the top-{K} boundary: \
+         selected down to {selected_lower}, excluded up to {excluded_upper}"
+    );
+    let data_workload = snapshot["data_workload"].clone();
+    snapshot["implementation"]["data_snapshot_id"] = json!("issue-754-certified-topk");
+    snapshot["implementation"]["accuracy_evidence"][&case.expr] = json!({
+        "query_string": case.expr,
+        "data_snapshot_id": "issue-754-certified-topk",
+        "data_workload": data_workload,
+        "source": "issue-754-certified-topk-generator",
+        "observed_at_unix_ms": 9500,
+        "valid_for_ms": 60000,
+        "topk_max_distinct_items": series.len() as u64,
+        "topk_selected_lower_bound": selected_lower,
+        "topk_excluded_upper_bound": excluded_upper,
+        "topk_interval_failure_probability": 0.0,
+    });
+    serde_json::from_value(snapshot).unwrap()
+}
+
+pub fn ensembles() -> Vec<(String, Vec<Case>)> {
+    let groups: [(&str, &[&str]); 3] = [
+        (
+            "shared-rate",
+            &["temporal-rate", "grouped-rate", "topk-rate"],
+        ),
+        ("shared-quantiles", &["temporal-quantile", "quantile-ratio"]),
+        ("all-ten", &[]),
+    ];
+    groups
+        .into_iter()
+        .map(|(name, names)| {
+            (
+                name.to_owned(),
+                suite()
+                    .queries
+                    .into_iter()
+                    .filter(|case| names.is_empty() || names.contains(&case.name.as_str()))
+                    .collect(),
+            )
+        })
+        .collect()
+}
+
+pub fn ensemble_input(cases: &[Case]) -> BackendLocalPlanningInput {
+    let mut combined = serde_json::to_value(input(&cases[0])).unwrap();
+    let mut queries = Vec::new();
+    let mut evidence = serde_json::Map::new();
+    for case in cases {
+        let wire = serde_json::to_value(input(case)).unwrap();
+        queries.extend(
+            wire["query_workload"]["repeating_queries"]
+                .as_array()
+                .unwrap()
+                .clone(),
+        );
+        if let Some(items) = wire["implementation"]["accuracy_evidence"].as_object() {
+            evidence.extend(items.clone());
+        }
+    }
+    combined["query_workload"]["repeating_queries"] = json!(queries);
+    combined["implementation"]["data_snapshot_id"] = json!("issue-754-level1");
+    combined["implementation"]["accuracy_evidence"] = json!(evidence);
+    serde_json::from_value(combined).unwrap()
+}
diff --git a/docs/design_docs/planning-test-layers.md b/docs/design_docs/planning-test-layers.md
new file mode 100644
index 000000000..93dcc0c7b
--- /dev/null
+++ b/docs/design_docs/planning-test-layers.md
@@ -0,0 +1,79 @@
+# Current planning and execution validation
+
+Audience: implementers and reviewers.
+
+The current scope is a complete deterministic path from a workload to correct
+execution. Backend uses explicit synthetic prices for candidate selection. Online
+ERP collection, feedback-driven replanning and deployment switching are deferred.
+
+| PR | Input | Assertion |
+| --- | --- | --- |
+| #728: structure | workload and declared execution/accuracy contracts | Membership: required shapes appear in the inventory and resolve as declared (bind, or refuse for a named policy reason); every other rejection is a recorded defect with a fixed occurrence count. No prices, no winner |
+| #742: ranking | same candidates plus synthetic complete quotes | Selection follows costs, reverses with costs, and excludes unavailable candidates |
+| #775: execution | workload, synthetic quotes, finite fixture data | Mutate quotes to select each admitted candidate, install and execute it; validate every query result and execution provenance |
+
+```text
+workload → Planner physical candidates
+         → Backend selection with synthetic costs
+         → Deployment Plan → data plane → checked results
+```
+
+Planner owns computation semantics and physical candidate construction. Backend
+owns deployment selection and binding. Serving loads the selected typed plan;
+it must not silently re-plan it at startup. The executor's existing operator,
+shared-producer, window, bound-SDS and recovery tests remain part of the baseline.
+Synthetic pricing does not relax query accuracy admission or SDS identity checks.
+
+## Workload ensembles
+
+Each layer tests the ten individual queries plus three workload ensembles:
+shared-rate (temporal-rate, grouped-rate, topk-rate), shared-quantiles
+(temporal-quantile, quantile-ratio), and all ten queries together. Structure checks
+preserve every query, validate shared output identities, and apply the same
+rejection discipline as the single-query path, with its own recorded defect
+counts. Sharing one deployed output between queries that need different
+semantics from it is refused as a matter of policy, not recorded as a defect. Ranking checks change
+prices for whole workload candidates. Execution installs each selected workload
+once, ingests its fixture once, and checks every query in that workload.
+
+“All candidates” means the supported inventory exposed by Planner and admitted
+under the fixture's contracts, not an exhaustive Cartesian product of hypothetical
+query plans. Binding rejections remain visible; prices cannot override them.
+
+Required shapes are asserted in both directions: the strict fixture must refuse
+the heap candidates for a named policy reason, and a certified companion fixture
+must admit the same shapes to pricing. A refusal asserted alone would also pass
+if the candidate had vanished.
+
+A rejection is not self-justifying. Level 1 separates *policy* refusals — the
+fixture declining to certify a candidate, which are asserted as required
+outcomes — from *defects*, which are bugs recorded with exact occurrence counts
+so that neither a new instance nor a silent fix can pass unnoticed. Level 1's
+enumerated candidate plans and ensembles are CI artifacts, not repository
+content: they are fully derived, and their identities change with the Planner
+pin. Only the per-query admission reports are committed.
+
+## Synthetic cost scope
+
+Prices are deterministic test inputs, not resource measurements. The execution
+fixture quotes every component of each compilable candidate. A candidate requiring
+external execution is marked infeasible for the local-execution fixture. Backend
+selects from that inventory using its production quote selection path. Retain the
+quoted snapshot, selected plan and installed plan for inspection. Contract tests
+must reject missing prices, a changed cost-model identity or a different installed
+generation. Human plan approval remains a separate review.
+
+## Deferred work
+
+ERP means Error–Resource Profile. In future, offline benchmark evidence or online
+measurements may inform candidate costs. Query-time merged sketch statistics can
+supply merge/readout resource observations, but do not alone establish accuracy.
+An online loop would additionally require valid observation scope, update policy,
+replanning triggers and safe deployment replacement. None is required now.
+
+#776 (statistics), #777 (accuracy evidence), #778 (resource measurements) and #759
+(real-evidence selection audit) are follow-up PRs, not prerequisites for the
+current #728 → #742 → #775 path. Their contract tests are not live telemetry or
+production validation. Independent measurement correctness fixes may be reviewed
+separately. The broader real-trace experiment remains deferred; its prior CPU
+objective is not a gate for the current fixture execution milestone.
diff --git a/docs/evaluation/issue754-human-review/README.md b/docs/evaluation/issue754-human-review/README.md
new file mode 100644
index 000000000..e9d99641a
--- /dev/null
+++ b/docs/evaluation/issue754-human-review/README.md
@@ -0,0 +1,129 @@
+# Issue #754 plans for human review
+
+Level 1 asserts *membership*, not a winner: for each query it requires that the
+candidate inventory expose particular shapes, and that the shapes this fixture
+declines to certify be refused for a named policy reason. It prices nothing, so
+no file here records a cost or a selected plan. Ranking is #742; observed
+selection quality is Level 3.
+
+## What is committed, and what is not
+
+| Path | Contents | Regenerated by |
+| --- | --- | --- |
+| `admission/*.admission.json` | The contract. One report per query: every candidate's id, status and, where it failed, the exact reason. `total_cost` is `null` throughout. | The Level-1 test |
+| `source-commit.txt` | The Backend revision the reports were produced from. | By hand, on regeneration |
+
+Enumerated candidate plans (`candidates/`) and workload ensembles (`ensembles/`)
+are **no longer committed**. They ran to roughly 94,000 lines, they are fully
+derived from the test, and every identity in them changes whenever the Planner
+pin moves. CI uploads them on every run as the `issue754-level1-plans` artifact,
+including on failure. Download that artifact when a specific rejection needs
+explaining; its `admission/` directory drops straight onto this one.
+
+## Identities are pinned, and they move
+
+Candidate, physical-candidate and logical-root IDs identify the generated
+computations and plans. A Planner or binding change can change those identities,
+the candidate inventory, or admission results. Regenerate before citing an ID;
+use the pinned source revision to reproduce the report.
+
+Backend source revision is in [source-commit.txt](source-commit.txt). The
+Planner revision is whatever `planner-types` pins in the workspace `Cargo.toml`;
+read it from there rather than from a number copied into prose, which is how the
+previous revision in this file came to be wrong.
+
+## Current admission result
+
+| Query | Candidates | Bound | Refused |
+| --- | --- | --- | --- |
+| grouped-rate | 5 | 5 | — |
+| grouped-temporal-sum | 4 | 4 | — |
+| quantile-ratio | 3 | 3 | — |
+| spatial-quantile | 4 | 4 | — |
+| spatial-sum | 4 | 4 | — |
+| spatial-topk | 4 | 3 | 1 no certified guarantee |
+| temporal-quantile | 3 | 3 | — |
+| temporal-rate | 5 | 5 | — |
+| temporal-sum | 3 | 3 | — |
+| topk-rate | 8 | 4 | 4 no certified guarantee |
+Only explicit admission policy refusals remain in these fixtures. Unexpected binding failures fail Level 1.
+
+**Policy refusals** are the fixture declining to certify a candidate, and they
+are asserted as required outcomes. The heap candidates are the whole of this
+category: Planner exposes `CountSketchWithHeap` for `spatial-topk` and both
+`CmsWithHeap` and `CountSketchWithHeap` for `topk-rate`, above a bound per-series
+Rate readout. This fixture supplies no distinct-item bound and no score
+separation, so each one is refused before pricing. That refusal is the
+assertion — the shapes must be present *and* rejected.
+
+The refusal is only meaningful if the same shapes bind when the evidence
+exists, so Level 1 asserts both directions. `issue-754` itself cannot supply
+that evidence: scoped evidence requires `topk_selected_lower_bound >
+topk_excluded_upper_bound`, and under its own per-series value domain the third-
+and fourth-ranked series overlap in both groups (group `a` selects down to 40
+while excluding a series reaching 130; group `b` selects down to 360 while
+excluding one reaching 850). The pointwise ordering never changes, but these
+queries evaluate in `real_time` scope, so the bound must hold across the whole
+validity window rather than at one instant.
+
+`promql-compliance/datasets/issue-754-certified-topk.yaml` is the companion
+that can. Each series occupies a decade of its own with `modulo: 1`, so the
+domains are disjoint by construction and the separation is *derived* from the
+dataset rather than chosen: selected down to 100, excluded up to 21, across
+both groups at once, with 8 distinct items.
+`certified_fixture_admits_the_heaps_the_strict_fixture_refuses` plans the two
+top-k queries against it and requires every heap family to reach pricing, and
+requires no uncertified-readout refusal to remain. Without that test, a heap
+candidate that silently left the inventory would still satisfy the strict
+fixture.
+
+The three ensembles (`shared-rate`, `shared-quantiles`, `all-ten`) are planned
+as a single workload input rather than one query at a time, and they carry the
+same discipline. The batch path reaches one refusal the single-query path
+cannot: a workload candidate that would share one deployed output between
+queries needing different semantics from it. Refusing that is correct, so it is
+recorded as a policy reason rather than a defect.
+
+**Binding defects are not allowed.** Planner query-result finalization and
+Backend typed bindings now preserve the selected semantics. The former fragment
+mismatch and accumulator-output schema failures have been fixed, and their
+allowlist has been removed. The same rejection checks apply to single queries
+and the three query ensembles.
+
+## Review order
+
+1. Check source, value transformations, grouping, windows and readouts.
+2. Compare each refusal reason with the query's accuracy requirement, and check
+   which of the two categories above it belongs to.
+3. Check persisted boundaries, definition IDs and requested pane coverage.
+4. For topk-rate, check the actual rate-value sort expression and partition keys.
+
+Candidate discovery preserves each root's admitted computations. Deployment
+evaluates single-root substitutions in a preferred workload context; it does not
+exhaustively enumerate joint workload combinations.
+
+| Query | PromQL |
+| --- | --- |
+| grouped-rate | `sum by (label_0) (rate(data[1m]))` |
+| grouped-temporal-sum | `sum by (label_0) (sum_over_time(data[1m]))` |
+| quantile-ratio | `quantile_over_time(0.9, data[1m]) / quantile_over_time(0.5, data[1m])` |
+| spatial-quantile | `quantile by (label_0) (0.9, data)` |
+| spatial-sum | `sum by (label_0) (data)` |
+| spatial-topk | `topk by (label_0) (3, data)` |
+| temporal-quantile | `quantile_over_time(0.9, data[1m])` |
+| temporal-rate | `rate(data[1m])` |
+| temporal-sum | `sum_over_time(data[1m])` |
+| topk-rate | `topk by (label_0) (3, rate(data[1m]))` |
+
+## Reproduce
+
+```sh
+ASAP_LEVEL1_ARTIFACT_DIR=/tmp/issue754-plans \
+  cargo test -p control_plane --test issue754_level1 --locked -- --test-threads=1
+```
+
+This writes `admission/` plus the uncommitted `candidates/` and `ensembles/`
+trees.
+
+Actual execution and recovery evidence is maintained in the downstream
+[bound SDS validation report](https://github.com/ProjectASAP/ASAPQuery-backend/blob/test/issue754-level3/docs/evaluation/bound-sds-2026-09-26/README.md).
diff --git a/docs/evaluation/issue754-human-review/admission/grouped-rate.admission.json b/docs/evaluation/issue754-human-review/admission/grouped-rate.admission.json
new file mode 100644
index 000000000..a3cd324de
--- /dev/null
+++ b/docs/evaluation/issue754-human-review/admission/grouped-rate.admission.json
@@ -0,0 +1,62 @@
+[
+  {
+    "candidate_id": "asap-explain-v1:candidate:111f0952136afd6e948171d6e16b8c676d34b55547bdf99843205593a97ef47f",
+    "logical_root_ids": [
+      "asap-explain-v1:root:011978ff0d3f5eb2719aa1fb05c25431e30bce7ce501a9ee95a58f95e1e77984"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:47825c1d8d1596ca28581a9c3911012f43f2e69e5375fed4703a123d48ac3e51",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 5849823772717530049,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:5720ec70a85747b7ed172ed0dea5e64452c43638184ffce632bdc0b547cb945f",
+    "logical_root_ids": [
+      "asap-explain-v1:root:4c88a5e7a225f5a46dbf69d5abfd7df6277d8adbb6d3e6d551b2773fff417860"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:4300a782bb2a5d4045f46fb57d992a8834c6069a5f735549f0507caa98252eb8",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 741363842036532054,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:9a6da4a496cd79bb65f8e11a6d3a14a1f8d25bc1198298df6f2a2b56fa20b225",
+    "logical_root_ids": [
+      "asap-explain-v1:root:6daba5f3c9b44ad12632739d6780f55e616a9653fbd1525b02168d0e453a4845"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:6f31f22b23159cfb14664439fe5e367d047968ad47d2d96d0ebfbd89e951442d",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 6192272502608540012,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:f1d848dd9b94a31003252e3e67de1fc831e8cd65e6816ac62bf5f0e1153924a6",
+    "logical_root_ids": [
+      "asap-explain-v1:root:7d4dc0a7304f936ee0d09795aa2989c5d0580dd075f7d5d006bacfb95eea6105"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:405478c7249bbf80125586a7fffa2568d84b797d0b354692be3fde55dad7a3f8",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 15282866737310247190,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:ec910d95d9d6da4f5d07f294f742290d18ab508bdafc9f350322591dc453d664",
+    "logical_root_ids": [
+      "asap-explain-v1:root:9208d81129d425f445ab8d757e7ad0e4d98cb582785eaf654560a7c6df5bdf35"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:7ee64b28e1a319d45bfa1b89940e164be12e84c751323db4b70ce5923f2eed60",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 3349874915222028658,
+    "total_cost": null,
+    "unavailable_reason": null
+  }
+]
\ No newline at end of file
diff --git a/docs/evaluation/issue754-human-review/admission/grouped-temporal-sum.admission.json b/docs/evaluation/issue754-human-review/admission/grouped-temporal-sum.admission.json
new file mode 100644
index 000000000..75af47516
--- /dev/null
+++ b/docs/evaluation/issue754-human-review/admission/grouped-temporal-sum.admission.json
@@ -0,0 +1,50 @@
+[
+  {
+    "candidate_id": "asap-explain-v1:candidate:47249a7520d1624e50b877075f6f97aa5e1cb072e6f172910c2a5e8db82eff76",
+    "logical_root_ids": [
+      "asap-explain-v1:root:448e345cb243745ba15d18995976d0c9a66cae45011ff106e159e97f293aaab2"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:bfd305549407f3e5ef862a6064f74690ec29b7bdebf66c7fdd296048b54e3c10",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 17508439989252383247,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:6514e86127465662ab44d368ad43c83e682d9cc7fad42da7fc84907094dcdac4",
+    "logical_root_ids": [
+      "asap-explain-v1:root:cc3eb5e15d1bcaf981dca5a4450f7a550709d52f0b87d0f68842e61de9f7f4fa"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:ef727cded34086f495ca0f9930fef6197e967ce7e22a3ad9f7045efe389ae141",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 375190931036314940,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:0c7f183a76d8e52946122044e047540dd4340df864d51d9d43e0448226eddaa3",
+    "logical_root_ids": [
+      "asap-explain-v1:root:afed30a788e560b32b6a5e3d72e6aca13b8cd09c2fc184e950bc6a6f68ec7e57"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:5c44b4cb533eb97040236079b2ded3a65af5484c45409bd7b36e53ade0e95744",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 13773692493706998879,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:40dd21fdc165cc2a65b26c4c5a11e3c2da152435497b9e879db9a04015725322",
+    "logical_root_ids": [
+      "asap-explain-v1:root:b8c508ddc54d5bc05ce612f20451625866cd9c4c76d97ed47a2bfb87e47db290"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:3e1a1a41993110533f5d64a598f55da19ff8f819b21f9c03dc292d20c3742f07",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 3225737927651639557,
+    "total_cost": null,
+    "unavailable_reason": null
+  }
+]
\ No newline at end of file
diff --git a/docs/evaluation/issue754-human-review/admission/quantile-ratio.admission.json b/docs/evaluation/issue754-human-review/admission/quantile-ratio.admission.json
new file mode 100644
index 000000000..bb1f6036e
--- /dev/null
+++ b/docs/evaluation/issue754-human-review/admission/quantile-ratio.admission.json
@@ -0,0 +1,38 @@
+[
+  {
+    "candidate_id": "asap-explain-v1:candidate:4fb544af8f312840e962c54cfa2dcffbad5bb3debebf58e8b512acba1a07bee5",
+    "logical_root_ids": [
+      "asap-explain-v1:root:36df098d763a273afc79e420e66e9ffca935f037e5a1a79c935267b287a470cc"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:bdd60d919a3717a7eac447277bd55d9cb0d2489121c4c275be8535974668df1b",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 14871479243635342203,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:4a707b1f6e9ac5a5b31643a1eee202400ead6e24f3725eba0804283f1500bb1c",
+    "logical_root_ids": [
+      "asap-explain-v1:root:cb06b175d20dbe1f760083d83562229a77ffbf3b4f719dbbc422d48c54880cbf"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:3f1e2b58edfa1c0af3cb3180a0632c438f3f68a8219868aa89b4c00b3a06c2d1",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 5752356276213301876,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:7f9c6332a1d49b67039b33c02a2b4cc324cfca3271fee74cdc8f907e59bb72d2",
+    "logical_root_ids": [
+      "asap-explain-v1:root:6341ef0ca36372c08399bd313a99aab646d639ad3f55107c2e038b0339493a50"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:f87a97555e148f5d4bbdc8ce2c0c43970ce3e28f1d86d83187a4008447bb55b7",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 12971657221408630864,
+    "total_cost": null,
+    "unavailable_reason": null
+  }
+]
\ No newline at end of file
diff --git a/docs/evaluation/issue754-human-review/admission/spatial-quantile.admission.json b/docs/evaluation/issue754-human-review/admission/spatial-quantile.admission.json
new file mode 100644
index 000000000..9692c489a
--- /dev/null
+++ b/docs/evaluation/issue754-human-review/admission/spatial-quantile.admission.json
@@ -0,0 +1,50 @@
+[
+  {
+    "candidate_id": "asap-explain-v1:candidate:c9aaf94a2015bbb6bc16c6da810771953107d1316817b63788f031334b824f68",
+    "logical_root_ids": [
+      "asap-explain-v1:root:6aa3eb79ee2f717bb3be85517dfece4cef935a48534b2bf08ad7b82beb0c48a2"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:bf2ff22c472f8c253909eb82e6e056d3f3a8558176ba2f865dc241fbe3cc314d",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 3294188734724225991,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:54b2218020929f8b0d25267363baaa2c34a9b92b4cd99faa7790593bcb437ded",
+    "logical_root_ids": [
+      "asap-explain-v1:root:fd88d52b75a7faf02f8c7b0ec431ba495f22b63e62695a388b313871434ea2e6"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:d76fd2149f74284fdd27bed60f1b1d7e0b752f8ec43853271130405e945fb660",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 14256379658571227787,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:dc49674bd5b4a491c1e32421471de94183448a57e2974bf5880f4c1014f6d565",
+    "logical_root_ids": [
+      "asap-explain-v1:root:59753aa608478f315510b90d33afe3d3498b424dfbe744dbb595b0d898a16dba"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:6eb943c7680d7bc70eb41402fcebd957c600040bec07698011d74789d8b8b0af",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 18366075381595067177,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:71ac43494ecab69c9808292bc9b5f44b21cec3ff30a8433774531aff351bff81",
+    "logical_root_ids": [
+      "asap-explain-v1:root:ff6c7fa75bd824d15015c7b39434e7b7956caa545dabb6a4b0300d38ce852b13"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:d4080a0b42161499e7f65aa98f73e07b8093a1b271879644633d465cc50c09e8",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 7667691247138405742,
+    "total_cost": null,
+    "unavailable_reason": null
+  }
+]
\ No newline at end of file
diff --git a/docs/evaluation/issue754-human-review/admission/spatial-sum.admission.json b/docs/evaluation/issue754-human-review/admission/spatial-sum.admission.json
new file mode 100644
index 000000000..9d6e2dfd7
--- /dev/null
+++ b/docs/evaluation/issue754-human-review/admission/spatial-sum.admission.json
@@ -0,0 +1,50 @@
+[
+  {
+    "candidate_id": "asap-explain-v1:candidate:a8869d0a16276da98a6fd3da7ad085d0145f165a84e2129cf43a6dcfc6b631b5",
+    "logical_root_ids": [
+      "asap-explain-v1:root:cca99c7039fe48156cbb90dccd659276ff1a70de776ba616d620f99f776d5cd2"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:d953c94e0cc0efa160db48ee14b791be59a7a83a4667c52f28446629bbecea71",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 1236684965287321759,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:abf5efc0850b4b6277747e6f1854128d7d317036073049557d7bc8aa5901fe4e",
+    "logical_root_ids": [
+      "asap-explain-v1:root:7ff3b3d94ab998a2379b3ece3415130f068445e61bf76faca2580b9745c1a573"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:839f658c5abbda84fa3ff561ec9b91abdcde972208cb67b0bb151aebd56049d1",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 17875413431231421451,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:b7db63ce72644bb7114426c030589f6da8b1e031aa40c794b8d1272179443c21",
+    "logical_root_ids": [
+      "asap-explain-v1:root:3052368e6a27c889cd28866ab95cd2d31700861582303eeeeca4916ace17d75c"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:7b333a16d4d53d7be8adb6a0c1414c99f31dfd03738ab16958868fd20eb0755d",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 8064649510529993241,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:04b75c822a626cb540735532131ced917dd2268098297c88a225f235275466b7",
+    "logical_root_ids": [
+      "asap-explain-v1:root:45fc4ceb0dfc3aa52fefb47dd50844411b0ff858390366fef488ab3244f0810e"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:941c854f10c2cb6f50e13c6f540d817009530956e216850f704af41bc6eedef6",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 10907830462850005469,
+    "total_cost": null,
+    "unavailable_reason": null
+  }
+]
\ No newline at end of file
diff --git a/docs/evaluation/issue754-human-review/admission/spatial-topk.admission.json b/docs/evaluation/issue754-human-review/admission/spatial-topk.admission.json
new file mode 100644
index 000000000..b1db5f74a
--- /dev/null
+++ b/docs/evaluation/issue754-human-review/admission/spatial-topk.admission.json
@@ -0,0 +1,50 @@
+[
+  {
+    "candidate_id": "asap-explain-v1:candidate:7d3e10d14fca58a7ecf3bb41932c6a4e644eca1b9a1ebb18d6a36365c2f16c64",
+    "logical_root_ids": [
+      "asap-explain-v1:root:ce68b3daf9be1bebe2252ca6957c8880a59ef03d33608d76b51ee0ae5eab3166"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:aeb2a2838d761656a89dbcc21a02624af8c301ef5270287176f230318139422b",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 1305595822516651180,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:8750d56b757927cf51befebf02d41ed72de02bc2454d8789f943f6a70fffaedd",
+    "logical_root_ids": [
+      "asap-explain-v1:root:d3fb125ea5076ac3e0310f3fefda894380e5160a9a66a6f6cf0385cfe2fce9fb"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:49338bca64db2ee6610bfc59cb2b7f92c0ff6d0a5b63b58abf431638e2ee6919",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 5538412219857372274,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:f35b353455f28664f5d479b20dfa5d8c7d15bdb38e38c3001760ceea861e427a",
+    "logical_root_ids": [
+      "asap-explain-v1:root:3f042e8731583d1bff8e7f8fea235b22aa7b91c197f24377269b9aba882e7293"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:0ab57dcc6b6f25ddacf1dcde8778b86eb14f1ab19142418d9f4becbc87c853b4",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 2697027346875258477,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:fa96fd113280e9693ebe25570e56388d43c994536c3e011fda2a12378ad92cf4",
+    "logical_root_ids": [
+      "asap-explain-v1:root:37837d0f71bf4f773d9158fa9240f3476c7be0ac5797fb00368e60c152740a21"
+    ],
+    "physical_candidate_id": null,
+    "identity_unavailable_reason": null,
+    "status": "bind_failed",
+    "plan_id": null,
+    "total_cost": null,
+    "unavailable_reason": "query compat-query-0: selected summary readout has no certified accuracy guarantee; provide scoped evidence or use exact execution"
+  }
+]
\ No newline at end of file
diff --git a/docs/evaluation/issue754-human-review/admission/temporal-quantile.admission.json b/docs/evaluation/issue754-human-review/admission/temporal-quantile.admission.json
new file mode 100644
index 000000000..5ce895501
--- /dev/null
+++ b/docs/evaluation/issue754-human-review/admission/temporal-quantile.admission.json
@@ -0,0 +1,38 @@
+[
+  {
+    "candidate_id": "asap-explain-v1:candidate:cae8c0dc87b8c9b8a4caf0505c0e6f93331ef8a6a11738891490a699d95e9bec",
+    "logical_root_ids": [
+      "asap-explain-v1:root:f67ab58bd7aeed1db6490f0fa305404fc2d226e66af995d9033c1558f82def12"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:8928adb5cbe3827eac3eb3494c586dccf712858b4db91f92c0a0f9b5c4a39d0a",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 11818444224054941250,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:f6baac22056351e8ab54eced4b908df054269b9b37cfe712a909d715a843d51c",
+    "logical_root_ids": [
+      "asap-explain-v1:root:4c0760500525736640e1f5cfefe23f81975c1986959cfda3a8d81c8e1528780b"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:6161dde9c05e443ef70973dc8d4e4de1012c20c7e32fe66f3b14ffe596996146",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 4754089690189431057,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:685f5f0b8a54645b9149cd418dba5a87b04ed49de58596c1a979e6669b6ccee1",
+    "logical_root_ids": [
+      "asap-explain-v1:root:37f71dd62f9a9ba01d93f319ede576bb568eec80f4a14f46b7d3a3907ee344a2"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:dc7476ab614c0587fbabcff0fe9cf2eabf671ada395f234a308d9bc8c11ca102",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 11545774673163874085,
+    "total_cost": null,
+    "unavailable_reason": null
+  }
+]
\ No newline at end of file
diff --git a/docs/evaluation/issue754-human-review/admission/temporal-rate.admission.json b/docs/evaluation/issue754-human-review/admission/temporal-rate.admission.json
new file mode 100644
index 000000000..09c4bea67
--- /dev/null
+++ b/docs/evaluation/issue754-human-review/admission/temporal-rate.admission.json
@@ -0,0 +1,62 @@
+[
+  {
+    "candidate_id": "asap-explain-v1:candidate:d4da2ee8e5f050683caa56b943f360336922ce9cbd8513affd8fbc4c4e07bb21",
+    "logical_root_ids": [
+      "asap-explain-v1:root:a021345938b62e6f25a9bcf16b3f41f21fa9b26781c1ec69210e6ef284aee7d6"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:6388d959687b00185715017906261a9a5841786cd1a8b9848f2ceb7572da6272",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 4803518866474131004,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:c3b669ec93cee2e22119ee1c8ef1759d3c24959cd3875e1df9d54041d90da775",
+    "logical_root_ids": [
+      "asap-explain-v1:root:a021345938b62e6f25a9bcf16b3f41f21fa9b26781c1ec69210e6ef284aee7d6"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:d484b0f28853d7b9b5a591077b9217403663beabc23968e335f4653a67e3b1ac",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 11300228965735498754,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:63307678b5ac4b6551b16bbb6ff52a1390043de10997d737ab048e0249d7e101",
+    "logical_root_ids": [
+      "asap-explain-v1:root:20edcbe6a7c98400a1ed3d85bf6e50d0bb3815c0588613057b45bfb5d0110c2a"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:805bb8cc1e5151d09e32d793522a9a824dbb9dc6b5e2727796d658b6de9d69c7",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 27061497774794526,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:e43ed052748b0278634e65c01a0faddffe80ef2cb539e1eaaf51b4a89c043fcf",
+    "logical_root_ids": [
+      "asap-explain-v1:root:52e399207dc2e8015624fc1167db3a5ee14447eeeef748c4e0a08102e837eeff"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:1d50d68ea9aa9939fd369e190ac573184ce95b3cc598a1e4cf06dd585cb8d0d1",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 3235857638904289352,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:b29564192b3f53546a0a5677fefe6aa97f3df36acf90c61cc0262af0c349a121",
+    "logical_root_ids": [
+      "asap-explain-v1:root:c4059b9e97db0e50b6438781b6374c58e3d1da7d63164af4422480b731e8a93e"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:bdccd007f2363cca84a0a47d0cd2b6fdca05bb132415cd3ad0e8cf6ec6a08b39",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 17865984201647136845,
+    "total_cost": null,
+    "unavailable_reason": null
+  }
+]
\ No newline at end of file
diff --git a/docs/evaluation/issue754-human-review/admission/temporal-sum.admission.json b/docs/evaluation/issue754-human-review/admission/temporal-sum.admission.json
new file mode 100644
index 000000000..48790009c
--- /dev/null
+++ b/docs/evaluation/issue754-human-review/admission/temporal-sum.admission.json
@@ -0,0 +1,38 @@
+[
+  {
+    "candidate_id": "asap-explain-v1:candidate:3ce219336d9871faaf086424ef02d4343c55191100acfe6231f423a572d2282f",
+    "logical_root_ids": [
+      "asap-explain-v1:root:65d44a424a99a04654dca6661114926b74a4a68bc398ddb9057c25a6d3dc02a8"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:156ff9afdfd69bd21810916a38b8e6f058b4666ff81ffce27b13353c67f34177",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 17810161372573872165,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:f4ad48202aae7a3e0fd5fbe558b6496b2894f26898a2a739de4aff1188ff9580",
+    "logical_root_ids": [
+      "asap-explain-v1:root:68707b744b7e3c770561391fdca5a428e5693343ae6e28935b5f3c90fd2f4c44"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:00f0ebe97213f5f8ba2977920985819c9602a43e28d9fbe4fce4e29119eb5ee7",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 3167846180279825593,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:cdbc5a3b9c4c439cd0cd9346cabf84383f41ba8375965f901bed62fc32664211",
+    "logical_root_ids": [
+      "asap-explain-v1:root:f184b3151f9a3094c8969aabc655efc1c2845a30cb2045fc6fa80eb0f0340b5e"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:7156c54b17a13f028c9d7fd0c5046d0653cbdcf0c06c7477b354e9148ae84893",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 17637043547591721229,
+    "total_cost": null,
+    "unavailable_reason": null
+  }
+]
\ No newline at end of file
diff --git a/docs/evaluation/issue754-human-review/admission/topk-rate.admission.json b/docs/evaluation/issue754-human-review/admission/topk-rate.admission.json
new file mode 100644
index 000000000..925207dfe
--- /dev/null
+++ b/docs/evaluation/issue754-human-review/admission/topk-rate.admission.json
@@ -0,0 +1,98 @@
+[
+  {
+    "candidate_id": "asap-explain-v1:candidate:111c7f89045f3f95d2496526c0e639483e68e7eccd5233b5800217bb48ca75fc",
+    "logical_root_ids": [
+      "asap-explain-v1:root:49fecd6076c1f86c5d6d741f3d36823254b4063876fba5827c6230d16919ce24"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:196a6f0d69e6a659db8e260d799c43ef321c987b10f208d765f4905c4f3f00f3",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 14733843382488271694,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:ce6d33de51ec5ae0d2d7f07881dd3eb7e5b395d9785589d5cdd0837098cf2f85",
+    "logical_root_ids": [
+      "asap-explain-v1:root:d56e4eb55b93685203954c7ba13cd63e97364a54b94b7d9296c2d4fd904157b2"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:39b0b2e7a9f6c2702bafc47a781a7d11573b49c657c947541b15cd60dee3b5c7",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 15833270860732361489,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:e3a772c43298db6d5212f19233f9f799102eea82179768ed1e3485f2494b5cd7",
+    "logical_root_ids": [
+      "asap-explain-v1:root:5e56d98a238228c80a8ed1b60bacf60ae8c4cdb37e06fa1aeaa22097c4b4ddd0"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:50811f37d1645f2d71bf7dfc600357213487c408b6521f5536f5e10218fe915b",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 5379663311087285901,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:90197a1355bccfb7e3b4b8223ed1ddd7e1008175605ff1edfe54c6c0741b8de1",
+    "logical_root_ids": [
+      "asap-explain-v1:root:801d89f580d5ed0c2767f0b4c5e89e82fd56a5d4522b174f6b079df4a2237830"
+    ],
+    "physical_candidate_id": null,
+    "identity_unavailable_reason": null,
+    "status": "bind_failed",
+    "plan_id": null,
+    "total_cost": null,
+    "unavailable_reason": "query compat-query-0: selected summary readout has no certified accuracy guarantee; provide scoped evidence or use exact execution"
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:c024ce553894a4a530d5a95127501840d547bc8986567f0fa2e686a69da43e5d",
+    "logical_root_ids": [
+      "asap-explain-v1:root:af33ea0fea85318f5a3fe68d396d697db96c2fe8933dbaa94e0d857c6bc8683e"
+    ],
+    "physical_candidate_id": null,
+    "identity_unavailable_reason": null,
+    "status": "bind_failed",
+    "plan_id": null,
+    "total_cost": null,
+    "unavailable_reason": "query compat-query-0: selected summary readout has no certified accuracy guarantee; provide scoped evidence or use exact execution"
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:c14415b2ba781f81821bfa69680db6e7704f0aef682d0ba8eb195629bcf1958a",
+    "logical_root_ids": [
+      "asap-explain-v1:root:e6cbb418ec21a9245c8357c1d3693dd220377f4839fd2f5e11580bbcca4037da"
+    ],
+    "physical_candidate_id": "asap-explain-v1:physical:a7c59efbbc4961d4a4cc74aa4eebeefec7feb0a7e7e17139697c62af864c8a8d",
+    "identity_unavailable_reason": null,
+    "status": "bound",
+    "plan_id": 11938117179177087264,
+    "total_cost": null,
+    "unavailable_reason": null
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:4e9aa572b654a822240790f00df23c815365abbee13b98a5ad40f12abfd40758",
+    "logical_root_ids": [
+      "asap-explain-v1:root:d2ad4a4b530bc3673ae07790c49943b60fc8931f74efaf5cdb6241da0154b831"
+    ],
+    "physical_candidate_id": null,
+    "identity_unavailable_reason": null,
+    "status": "bind_failed",
+    "plan_id": null,
+    "total_cost": null,
+    "unavailable_reason": "query compat-query-0: selected summary readout has no certified accuracy guarantee; provide scoped evidence or use exact execution"
+  },
+  {
+    "candidate_id": "asap-explain-v1:candidate:5bfeb9e2bd2e5cb60386f05cda2aa9d0fc1787d5471f17988b4c42d2ff5da540",
+    "logical_root_ids": [
+      "asap-explain-v1:root:6531189b3ea6dc08910b5148ca99b8022ed4fd5508238828419cc7457af1537b"
+    ],
+    "physical_candidate_id": null,
+    "identity_unavailable_reason": null,
+    "status": "bind_failed",
+    "plan_id": null,
+    "total_cost": null,
+    "unavailable_reason": "query compat-query-0: selected summary readout has no certified accuracy guarantee; provide scoped evidence or use exact execution"
+  }
+]
\ No newline at end of file
diff --git a/docs/evaluation/issue754-human-review/source-commit.txt b/docs/evaluation/issue754-human-review/source-commit.txt
new file mode 100644
index 000000000..e8e03085a
--- /dev/null
+++ b/docs/evaluation/issue754-human-review/source-commit.txt
@@ -0,0 +1 @@
+066a263a06486c582a1538bccbba70b96aab131f
diff --git a/promql-compliance/datasets/issue-754-certified-topk.yaml b/promql-compliance/datasets/issue-754-certified-topk.yaml
new file mode 100644
index 000000000..207998357
--- /dev/null
+++ b/promql-compliance/datasets/issue-754-certified-topk.yaml
@@ -0,0 +1,41 @@
+# Companion to issue-754, built so that a top-k heap can actually be certified.
+#
+# issue-754 itself cannot certify one: its per-series value domains overlap, so
+# no scalar `topk_selected_lower_bound > topk_excluded_upper_bound` holds across
+# the validity window. Here each series occupies a decade of its own and the
+# modulo is 1, so the domains are disjoint by construction:
+#
+#   value_i(t) in [multiplier_i * base_i, multiplier_i * (base_i + modulo_i)]
+#
+#   group a: a1 [10,11]  a2 [100,101]  a3 [1000,1001]  a4 [10000,10001]
+#   group b: b1 [20,21]  b2 [200,201]  b3 [2000,2001]  b4 [20000,20001]
+#
+# topk by (label_0) (3, ...) selects {a2,a3,a4} and {b2,b3,b4}. The lowest
+# selected bound is 100 and the highest excluded bound is 21, so the separation
+# contract holds for both groups at once with a single scalar pair.
+name: issue-754-certified-topk
+series:
+  - metric: data
+    labels: {label_0: a, instance: a1}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 1, base: 10, modulo: 1}
+  - metric: data
+    labels: {label_0: a, instance: a2}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 1, base: 100, modulo: 1}
+  - metric: data
+    labels: {label_0: a, instance: a3}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 1, base: 1000, modulo: 1}
+  - metric: data
+    labels: {label_0: a, instance: a4}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 1, base: 10000, modulo: 1}
+  - metric: data
+    labels: {label_0: b, instance: b1}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 1, base: 20, modulo: 1}
+  - metric: data
+    labels: {label_0: b, instance: b2}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 1, base: 200, modulo: 1}
+  - metric: data
+    labels: {label_0: b, instance: b3}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 1, base: 2000, modulo: 1}
+  - metric: data
+    labels: {label_0: b, instance: b4}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 1, base: 20000, modulo: 1}
diff --git a/promql-compliance/datasets/issue-754.yaml b/promql-compliance/datasets/issue-754.yaml
new file mode 100644
index 000000000..d84b1e283
--- /dev/null
+++ b/promql-compliance/datasets/issue-754.yaml
@@ -0,0 +1,26 @@
+name: issue-754
+series:
+  - metric: data
+    labels: {label_0: a, instance: a1}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 1, base: 10, modulo: 120}
+  - metric: data
+    labels: {label_0: a, instance: a2}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 2, base: 20, modulo: 120}
+  - metric: data
+    labels: {label_0: a, instance: a3}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 3, base: 30, modulo: 120}
+  - metric: data
+    labels: {label_0: a, instance: a4}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 4, base: 40, modulo: 120}
+  - metric: data
+    labels: {label_0: b, instance: b1}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 5, base: 50, modulo: 120}
+  - metric: data
+    labels: {label_0: b, instance: b2}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 6, base: 60, modulo: 120}
+  - metric: data
+    labels: {label_0: b, instance: b3}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 7, base: 70, modulo: 120}
+  - metric: data
+    labels: {label_0: b, instance: b4}
+    generated_samples: {start_offset_seconds: 0, end_offset_seconds: 180, step_seconds: 0.1, multiplier: 8, base: 80, modulo: 120}
diff --git a/promql-compliance/suites/issue-754.yaml b/promql-compliance/suites/issue-754.yaml
new file mode 100644
index 000000000..cb04ecf19
--- /dev/null
+++ b/promql-compliance/suites/issue-754.yaml
@@ -0,0 +1,46 @@
+name: issue-754
+comparison_defaults:
+  value_tolerance:
+    relative: 0.01
+    absolute: 0.000001
+queries:
+  - name: spatial-sum
+    expr: 'sum by (label_0) (data)'
+    instant_offsets_seconds: [120, 180]
+    range: {start_offset_seconds: 120, end_offset_seconds: 180, step_seconds: 60}
+  - name: spatial-topk
+    expr: 'topk by (label_0) (3, data)'
+    instant_offsets_seconds: [120, 180]
+    range: {start_offset_seconds: 120, end_offset_seconds: 180, step_seconds: 60}
+  - name: spatial-quantile
+    expr: 'quantile by (label_0) (0.9, data)'
+    instant_offsets_seconds: [120, 180]
+    range: {start_offset_seconds: 120, end_offset_seconds: 180, step_seconds: 60}
+  - name: temporal-sum
+    expr: 'sum_over_time(data[1m])'
+    instant_offsets_seconds: [120, 180]
+    range: {start_offset_seconds: 120, end_offset_seconds: 180, step_seconds: 60}
+  - name: temporal-quantile
+    expr: 'quantile_over_time(0.9, data[1m])'
+    instant_offsets_seconds: [120, 180]
+    range: {start_offset_seconds: 120, end_offset_seconds: 180, step_seconds: 60}
+  - name: temporal-rate
+    expr: 'rate(data[1m])'
+    instant_offsets_seconds: [120, 180]
+    range: {start_offset_seconds: 120, end_offset_seconds: 180, step_seconds: 60}
+  - name: grouped-rate
+    expr: 'sum by (label_0) (rate(data[1m]))'
+    instant_offsets_seconds: [120, 180]
+    range: {start_offset_seconds: 120, end_offset_seconds: 180, step_seconds: 60}
+  - name: grouped-temporal-sum
+    expr: 'sum by (label_0) (sum_over_time(data[1m]))'
+    instant_offsets_seconds: [120, 180]
+    range: {start_offset_seconds: 120, end_offset_seconds: 180, step_seconds: 60}
+  - name: topk-rate
+    expr: 'topk by (label_0) (3, rate(data[1m]))'
+    instant_offsets_seconds: [120, 180]
+    range: {start_offset_seconds: 120, end_offset_seconds: 180, step_seconds: 60}
+  - name: quantile-ratio
+    expr: 'quantile_over_time(0.9, data[1m]) / quantile_over_time(0.5, data[1m])'
+    instant_offsets_seconds: [120, 180]
+    range: {start_offset_seconds: 120, end_offset_seconds: 180, step_seconds: 60}