feat(planner): bind offline frequency evidence and preserve o11y fallbacks

control_plane/Cargo.toml

@@ -93,16 +93,16 @@ asap_types.workspace = true
 # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this*
 # repo is a real one. Vendored locally instead of chased upstream -- see
 # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`.
-planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" }
-asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" }
+planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "abb2f20e27091ac0c60715dc42b9b59c75b3447c" }
+asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "abb2f20e27091ac0c60715dc42b9b59c75b3447c" }
 
 # L1 adoption (design-target-architecture.md Part B): the PromQL front
 # end itself, replacing control_plane's own query_parser/promql.rs.
 # Pinned via `rev`, not a floating branch reference. Same rev as
 # `planner-types`/`asap-aware-mapping` above -- these three MUST move
 # together (two revs of the same upstream repo's types in one workspace
 # resolve to distinct Rust types that won't unify).
-asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" }
+asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "abb2f20e27091ac0c60715dc42b9b59c75b3447c" }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }

control_plane/docs/offline-sketch-evidence.md

@@ -0,0 +1,86 @@
+# Offline sketch evidence replay
+
+Audience: developers evaluating planner integration without a deployed data plane.
+
+`ControlPlaneCostModel::with_offline_evidence` accepts the planner's validated
+offline provider. Candidate ordering compares update CPU nanoseconds only when
+every candidate has compatible evidence for the exact parameters returned by
+this deployment's sizing policy. Missing, stale, incompatible, or ambiguous
+measurements preserve the existing order. Offline errors never change formal
+accuracy guarantees. Existing physical and lifecycle costs retain their units;
+CPU nanoseconds are not added to legacy dimensionless costs.
+
+For explicit integer-key point-frequency queries,
+`with_offline_frequency_comparison(evidence, request)` additionally compares
+query-matched sketch measurements against an exact snapshot baseline. The
+request supplies the observed mean-error budget, number of reads, retained
+state count, horizon and CPU/memory weights. This is a fixed-snapshot offline
+comparison: the caller asserts the recorded integer-key distribution and probe
+population apply. The mean-error threshold applies to that recorded population,
+not to each queried key or future live data.
+
+The backend restricts candidates to measured power-of-two CMS layouts before
+comparison and supplies its own formal minimum parameters from the tighter
+workload and query accuracy. Observed error may select a larger measured sketch,
+but never weakens formal sizing. Missing evidence, an unacceptable error budget,
+or an exact winner yields `PassThrough`, preserving exact execution. Unfiltered
+legacy frequency totals and `count_over_time` never receive point-frequency
+error acceptance. `offline_frequency_recommendation(payload)` exposes the same
+decision and rejection reasons used by the binder.
+
+The typed frequency example reads real comparison artifacts without starting a
+data plane, using the planner revision pinned in this backend:
+
+```bash
+cargo run -p control_plane --example offline_frequency_plan -- \
+  comparison-evidence.json comparison-request.json 7 0.01
+```
+
+It binds a named integer-key source and reports the chosen parameters, exact
+alternative, cost estimates and preserved point readout. It does not certify
+that an existing Prometheus metric or deployed materialization uses that source.
+
+Run the PromQL replay using the pinned planner dependency:
+
+```bash
+cargo run -p control_plane --example offline_planner_replay -- \
+  o11y_bench_promql.txt planner-evidence.json context.json > control-plane-o11y.json
+```
+
+For development against unpublished planner changes, the optional local-checkout
+wrapper supplies source patches and records the checkout revisions:
+
+```bash
+python3 tools/run-offline-planner-replay.py \
+  --planner /path/to/ASAPPlanner \
+  --queries /path/to/ASAPPlanner/crates/frontend-promql/tests/observability/data/o11y_bench_promql.txt \
+  --evidence /path/to/planner-evidence.json \
+  --context /path/to/context.json \
+  --output /tmp/control-plane-o11y.json
+```
+
+The wrapper supplies local Cargo source patches for all three planner crates,
+preserving a single set of IR types. It also changes Cargo.lock; normal use of
+the pinned dependency requires no source patches. The normal backend sibling
+dependencies (ASAPCollector and asap_sketchlib)
+must remain available at the paths in its workspace manifests.
+
+The replay calls the real control-plane parser and typed summary binder for
+exact, default, and empirical modes. It records per-query rejection/fallback,
+selected summary states, matching update/state evidence, provenance, and elapsed
+planning time. The selected offline context is an explicit simulation assumption;
+it is not an assertion that an o11y metric has that measured distribution.
+Source scans beneath summaries count as raw subtrees, so they are not themselves
+evidence that the whole query fell back. Root fallback is a separate field.
+Bare selectors, sort roots and comparison/filter roots retain the complete
+original query as `KeepPreAsap`. They bind successfully while preserving label
+predicates, ordering and filtering. Executable query compilation marks these
+roots `ExactFallback` and requests no summary materializations; successful
+binding therefore does not mean they are served by the warm tier.
+
+This is binding coverage, not successful deployment compilation or execution.
+No collectors or query servers start. Point-frequency benchmark errors are
+exported as observations with `error_applies_to_current_query: false`.
+Whole-plan resource savings remain null without matched raw/residual physical
+operator evidence. Unsupported summary binary operators lower to explicit warm
+tier fallback; they are not silently executed with different semantics.

control_plane/examples/offline_frequency_plan.rs

@@ -0,0 +1,81 @@
+//! Bind a typed fixed-snapshot point-frequency query using offline evidence.
+use std::rc::Rc;
+
+use anyhow::{bail, Result};
+use asap_aware_mapping::empirical_comparison::{
+    OfflineComparisonEvidence, OfflineComparisonRequest,
+};
+use control_plane::{
+    physical::post_asap::{bind_query_expr_with_cost_model, cost_model::ControlPlaneCostModel},
+    planner_selection::frequency,
+    types_v2::AccuracyTarget,
+};
+use planner_types::pre_asap::{AggIntent, Column, DataType, QueryExpr, Reduction, Schema, Source};
+use serde_json::json;
+
+fn main() -> Result<()> {
+    let args: Vec<_> = std::env::args().skip(1).collect();
+    if args.len() != 4 {
+        bail!("usage: offline_frequency_plan COMPARISON.json REQUEST.json INTEGER_KEY EPSILON");
+    }
+    let evidence: OfflineComparisonEvidence =
+        serde_json::from_str(&std::fs::read_to_string(&args[0])?)?;
+    let request: OfflineComparisonRequest =
+        serde_json::from_str(&std::fs::read_to_string(&args[1])?)?;
+    let item: i64 = args[2].parse()?;
+    let epsilon: f64 = args[3].parse()?;
+    if !epsilon.is_finite() || epsilon <= 0.0 || epsilon >= 1.0 {
+        bail!("epsilon must be finite and strictly between zero and one");
+    }
+    let accuracy = AccuracyTarget::Epsilon(epsilon);
+    let model = ControlPlaneCostModel::new(accuracy.clone())
+        .with_offline_frequency_comparison(evidence, request.clone());
+    let intent = frequency(accuracy, Some(("key".into(), item.to_string())));
+    let AggIntent::Extension { payload, .. } = &intent else {
+        unreachable!()
+    };
+    let recommendation = model.offline_frequency_recommendation(payload);
+    let query = QueryExpr::Aggregate {
+        reduction: Reduction::PerEntity,
+        measures: vec![intent],
+        output_names: vec![],
+        having: None,
+        child: Rc::new(QueryExpr::Scan {
+            source: Source::TimeSeries {
+                metric: "offline_integer_snapshot".into(),
+            },
+            predicates: vec![],
+            schema: Schema::with_time_index(
+                vec![
+                    Column::new("ts", DataType::Timestamp, false),
+                    Column::new("key", DataType::Int64, false),
+                    Column::new("value", DataType::Float64, false),
+                ],
+                0,
+                vec![],
+            ),
+        }),
+    };
+    let bound = bind_query_expr_with_cost_model(&query, &model)?;
+    let root_raw_fallback = matches!(&bound,
+        control_plane::physical::post_asap::PhysicalExpr::Committed(
+            control_plane::physical::post_asap::PostAsapPlan::Summary(node)
+        ) if matches!(node.expr, planner_types::post_asap::SummaryExpr::KeepPreAsap(_)));
+    let (recommendation, unavailable_reason) = match recommendation {
+        Ok(value) => (Some(value), None),
+        Err(reason) => (None, Some(reason)),
+    };
+    serde_json::to_writer_pretty(
+        std::io::stdout(),
+        &json!({
+            "scope":"typed offline point-frequency recommendation and control-plane binding",
+            "request":request, "item":item, "formal_epsilon":epsilon,
+            "recommendation":recommendation,"unavailable_reason":unavailable_reason,
+        "bound_plan":format!("{bound:#?}"), "root_raw_fallback":root_raw_fallback,
+            "limitations":["The caller asserts the fixed-snapshot integer-key benchmark context",
+                "Observed mean error applies to the recorded offline probe population, not an individual key guarantee",
+                "No materializations deployed or data-plane execution performed"]
+        }),
+    )?;
+    Ok(())
+}

control_plane/examples/offline_planner_replay.rs

@@ -0,0 +1,154 @@
+//! Offline control-plane binding, without starting collectors or query servers.
+use std::{collections::HashSet, rc::Rc, time::Instant};
+
+use anyhow::{bail, Context, Result};
+use asap_aware_mapping::empirical_cost::{
+    EmpiricalEvidenceProvider, EvidenceArtifact, EvidenceContext,
+};
+use control_plane::{
+    physical::post_asap::{
+        bind_query_expr_with_cost_model, cost_model::ControlPlaneCostModel, PhysicalExpr,
+        PostAsapPlan,
+    },
+    query_parser::parse_query_expr_canonical,
+    types_v2::AccuracyTarget,
+};
+use planner_types::post_asap::{SummaryExpr, SummaryFamilyType, SummaryNode};
+use serde_json::{json, Value};
+
+fn inspect(
+    node: &Rc<SummaryNode>,
+    model: &ControlPlaneCostModel,
+    seen: &mut HashSet<usize>,
+    states: &mut Vec<Value>,
+    raw: &mut usize,
+) {
+    if !seen.insert(Rc::as_ptr(node) as usize) {
+        return;
+    }
+    match &node.expr {
+        SummaryExpr::KeepPreAsap(_) => *raw += 1,
+        SummaryExpr::SummaryAgg { family, child, .. } => {
+            if let SummaryFamilyType::Sketch(kind, _) = family {
+                let lookup = model
+                    .offline_evidence()
+                    .map(|provider| provider.lookup(kind.algorithm(), kind.params()));
+                let (measurement, reason) = match lookup {
+                    Some(Ok(row)) => (
+                        Some(json!({
+                            "record_id": row.id,
+                            "provenance": row.provenance,
+                            "update_cpu_ns": row.metrics.resources.cpu.update_cpu_ns,
+                            "retained_bytes": row.metrics.resources.retained_memory_bytes,
+                            "offline_error_observation": row.error,
+                            "error_applies_to_current_query": false,
+                        })),
+                        None,
+                    ),
+                    Some(Err(error)) => (None, Some(error.to_string())),
+                    None => (None, Some("offline evidence not supplied".into())),
+                };
+                states.push(json!({"algorithm":kind.algorithm(), "params":kind.params(), "measurement":measurement, "unavailable_reason":reason}));
+            } else {
+                states.push(json!({"exact_family":format!("{family:?}")}));
+            }
+            inspect(child, model, seen, states, raw);
+        }
+        SummaryExpr::SummaryEstimate { summary_input, .. }
+        | SummaryExpr::SummaryDelete { summary_input, .. } => {
+            inspect(summary_input, model, seen, states, raw)
+        }
+        SummaryExpr::SummaryMerge { children } => {
+            for child in children {
+                inspect(child, model, seen, states, raw);
+            }
+        }
+        SummaryExpr::SummaryJoin {
+            outer: lhs,
+            inner: rhs,
+            ..
+        }
+        | SummaryExpr::SummarySubtract {
+            left: lhs,
+            right: rhs,
+        }
+        | SummaryExpr::BinaryOp { lhs, rhs, .. } => {
+            inspect(lhs, model, seen, states, raw);
+            inspect(rhs, model, seen, states, raw);
+        }
+    }
+}
+
+fn main() -> Result<()> {
+    let args: Vec<_> = std::env::args().skip(1).collect();
+    if args.len() != 1 && args.len() != 3 {
+        bail!("usage: offline_planner_replay QUERIES.txt [EVIDENCE.json CONTEXT.json]");
+    }
+    let corpus = std::fs::read_to_string(&args[0])?;
+    let evidence = if args.len() == 3 {
+        let artifact: EvidenceArtifact = serde_json::from_str(&std::fs::read_to_string(&args[1])?)?;
+        let context: EvidenceContext = serde_json::from_str(&std::fs::read_to_string(&args[2])?)?;
+        Some((artifact, context))
+    } else {
+        None
+    };
+    let mut rows = Vec::new();
+    let mut modes = vec!["exact", "default"];
+    if evidence.is_some() {
+        modes.push("empirical");
+    }
+    for mode in modes {
+        let accuracy = if mode == "exact" {
+            AccuracyTarget::Exact
+        } else {
+            AccuracyTarget::Epsilon(0.01)
+        };
+        let mut model = ControlPlaneCostModel::new(accuracy.clone());
+        if mode == "empirical" {
+            let (artifact, context) = evidence.as_ref().context("missing evidence")?;
+            model = model.with_offline_evidence(EmpiricalEvidenceProvider::new(
+                artifact.clone(),
+                context.clone(),
+            )?);
+        }
+        for query in corpus
+            .lines()
+            .map(str::trim)
+            .filter(|q| !q.is_empty() && !q.starts_with('#'))
+        {
+            let start = Instant::now();
+            let result = parse_query_expr_canonical(query, accuracy.clone()).and_then(|expr| {
+                bind_query_expr_with_cost_model(&expr, &model).map_err(Into::into)
+            });
+            let elapsed_ns = start.elapsed().as_nanos();
+            let result = match result {
+                Ok(PhysicalExpr::Committed(PostAsapPlan::Summary(node))) => {
+                    let mut states = Vec::new();
+                    let mut raw = 0;
+                    inspect(&node, &model, &mut HashSet::new(), &mut states, &mut raw);
+                    json!({"status":"bound", "states":states, "raw_subtrees":raw,
+                        "root_raw_fallback":matches!(node.expr, SummaryExpr::KeepPreAsap(_)),
+                        "summary_plan":format!("{node:#?}")})
+                }
+                Ok(plan) => json!({"status":"other_physical_plan", "plan":format!("{plan:?}")}),
+                Err(error) => json!({"status":"rejected", "reason":error.to_string()}),
+            };
+            rows.push(
+                json!({"query":query,"mode":mode,"planning_elapsed_ns":elapsed_ns,"result":result}),
+            );
+        }
+    }
+    serde_json::to_writer_pretty(
+        std::io::stdout(),
+        &json!({
+            "schema_version":1,
+            "evaluation":"offline control-plane parser and typed summary binder",
+            "corpus_path":args[0],
+            "offline_context":evidence.as_ref().map(|(_, context)|context),
+            "limitations":["No deployed execution or measured end-to-end speedup", "Binding does not establish executable placement or complete physical cost", "Offline point-frequency errors do not establish current query guarantees", "raw_subtrees includes necessary source scans beneath summaries"],
+            "estimated_end_to_end_savings":null,
+            "rows":rows
+        }),
+    )?;
+    Ok(())
+}