Add a strict multi-fit ERP Figure 1 runner

control_plane/Cargo.toml

@@ -91,18 +91,18 @@ 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 = "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/ProjectASAP/ASAPPlanner", rev = "9f526a9" }
-asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "9f526a9" }
+planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "a9651cc" }
+asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "a9651cc" }
 
 # 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 = "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/ProjectASAP/ASAPPlanner", rev = "9f526a9" }
-asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "9f526a9" }
-asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "9f526a9" }
+asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "a9651cc" }
+asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "a9651cc" }
+asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "a9651cc" }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full", "test-util"] }

crates/asap_types/Cargo.toml

@@ -32,4 +32,4 @@ sha2 = "0.10"
 # exactly (`control_plane/Cargo.toml`) -- two different revs of the same
 # git dependency in one workspace resolve to two distinct Rust types that
 # won't unify.
-planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "9f526a9" }
+planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "a9651cc" }

data_plane/Cargo.toml

@@ -38,8 +38,8 @@ control_plane = { path = "../control_plane" }
 # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now
 # (not a separate `asap-ir` import). Query serving consumes the compiled
 # QueryPlan; these types are used at physical-plan compilation boundaries.
-planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "9f526a9" }
-asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "9f526a9" }
+planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "a9651cc" }
+asap-frontend-metricsql = { git = "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/ProjectASAP/ASAPPlanner", rev = "a9651cc" }
 
 # Shared external (workspace)
 serde.workspace = true
@@ -131,7 +131,7 @@ crc32fast = "1.4"
 # none of them.
 
 [dev-dependencies]
-asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "9f526a9" }
+asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "a9651cc" }
 tempfile = "3.20.0"
 criterion = { version = "0.5", features = ["html_reports"] }
 tokio-tungstenite = "0.21"

tools/erp-figure1/README.md

@@ -0,0 +1,54 @@
+# ERP Figure 1 runner
+
+`run.py` is the experiment boundary for the five Figure 1 arms:
+
+1. `autosketch_per_query`
+2. `planner_analytical`
+3. `planner_erp`
+4. `asap_no_sharing`
+5. `exact`
+
+The manifest pins a dataset SHA-256 and one candidate space, memory budget,
+accuracy requirement, window workload, multi-family `ErpShapeObservation`, and
+the complete fit/divergence/confidence selection policy from ASAPPlanner #380.
+Each arm runs in a separate process.
+The runner measures wall time, user/system CPU, and peak RSS with GNU `time`;
+the arm reports its selected plan, selected candidate records, retained state
+bytes, measured error, and any additional metrics. The runner checks that every
+selected candidate belongs to the common space and that non-exact arms meet the
+total memory and accuracy bounds. A successful arm must echo the exact contract from
+`ASAP_FIGURE1_CONTRACT_JSON`; the dataset path is supplied in
+`ASAP_FIGURE1_DATASET`. Contract drift aborts the experiment rather than
+producing a comparison.
+
+The observation must contain positive cardinality/event counts and one or more
+unique fitted families. Every fit records numeric parameters, non-negative
+goodness-of-fit, and confidence in `[0,1]`. The shared selection policy includes
+event/cardinality/parameter gates plus maximum goodness-of-fit, minimum
+confidence, and minimum confidence margin. This keeps poor or ambiguous fits
+visible to every arm instead of letting the ERP arm silently use an older
+single-family shape contract.
+
+```sh
+python3 tools/erp-figure1/run.py \
+  --manifest /path/to/immutable-figure1-manifest.json \
+  --output /path/to/new-results.json
+python3 -m unittest tools/erp-figure1/test_run.py -v
+```
+
+Each command must print one JSON object:
+
+```json
+{
+  "contract": {"the exact object supplied in ASAP_FIGURE1_CONTRACT_JSON": true},
+  "selected_plan": {"family": "cms", "width": 1024, "depth": 5},
+  "selected_candidates": [{"family": "cms", "width": 1024, "depth": 5}],
+  "metrics": {"state_bytes": 40960, "max_error": 0.007},
+  "provenance": {"revision": "..."}
+}
+```
+
+The runner does not contain sketch, Planner, or AutoSketch implementations. The
+arm commands must invoke those production/reference implementations. Failed
+arms remain in the output and make the command fail. Do not publish a Figure 1
+from smoke adapters or from arms that merely relabel a fixed CLI configuration.

tools/erp-figure1/run.py

@@ -0,0 +1,169 @@
+#!/usr/bin/env python3
+"""Run five isolated ERP evaluation arms under one immutable contract."""
+import argparse, hashlib, json, os, pathlib, subprocess, tempfile, time
+
+ARM_NAMES = ["autosketch_per_query", "planner_analytical", "planner_erp", "asap_no_sharing", "exact"]
+REQUIRED_METRICS = {"state_bytes", "max_error"}
+
+
+def canonical(value):
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False)
+
+
+def sha256_file(path):
+    digest = hashlib.sha256()
+    with open(path, "rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def validate_manifest(manifest, root):
+    if manifest.get("schema_version") != 1:
+        raise ValueError("unsupported manifest schema_version")
+    dataset = manifest["dataset"]
+    path = (root / dataset["path"]).resolve() if not pathlib.Path(dataset["path"]).is_absolute() else pathlib.Path(dataset["path"])
+    actual = sha256_file(path)
+    if actual != dataset["sha256"]:
+        raise ValueError(f"dataset checksum mismatch: expected {dataset['sha256']}, got {actual}")
+    constraints = manifest["constraints"]
+    for key in (
+        "candidate_space", "memory_budget_bytes", "accuracy", "window",
+        "erp_observation", "erp_selection_policy",
+    ):
+        if key not in constraints:
+            raise ValueError(f"missing common constraint {key}")
+    if constraints["memory_budget_bytes"] <= 0 or not constraints["candidate_space"]:
+        raise ValueError("candidate space and memory budget must be non-empty/positive")
+    validate_erp_contract(constraints["erp_observation"], constraints["erp_selection_policy"])
+    arms = manifest["arms"]
+    if [arm.get("name") for arm in arms] != ARM_NAMES:
+        raise ValueError(f"arms must be ordered exactly as {ARM_NAMES}")
+    if any(not isinstance(arm.get("command"), list) or not arm["command"] for arm in arms):
+        raise ValueError("every arm requires a non-empty argv command")
+    return path, actual, constraints
+
+
+def validate_erp_contract(observation, policy):
+    if observation.get("cardinality", 0) <= 0 or observation.get("observed_events", 0) <= 0:
+        raise ValueError("ERP observation cardinality/events must be positive")
+    fits = observation.get("fits")
+    if not isinstance(fits, list) or not fits:
+        raise ValueError("ERP observation requires at least one fitted family")
+    families = set()
+    for fit in fits:
+        family = fit.get("family")
+        if not isinstance(family, str) or not family or family in families:
+            raise ValueError("ERP fit families must be unique and non-empty")
+        families.add(family)
+        parameters = fit.get("parameters")
+        goodness, confidence = fit.get("goodness_of_fit"), fit.get("confidence")
+        if not isinstance(parameters, dict) or any(not isinstance(v, (int, float)) for v in parameters.values()):
+            raise ValueError("ERP fit parameters must be numeric")
+        if not isinstance(goodness, (int, float)) or goodness < 0:
+            raise ValueError("ERP goodness_of_fit must be non-negative")
+        if not isinstance(confidence, (int, float)) or not 0 <= confidence <= 1:
+            raise ValueError("ERP confidence must be within [0,1]")
+    fingerprint = observation.get("empirical_fingerprint")
+    if fingerprint is not None and (not isinstance(fingerprint, str) or not fingerprint):
+        raise ValueError("ERP empirical_fingerprint must be absent or non-empty")
+    required = {
+        "minimum_benchmark_events", "max_log2_cardinality_distance",
+        "max_parameter_distance", "max_goodness_of_fit", "minimum_confidence",
+        "minimum_confidence_margin",
+    }
+    if required - set(policy):
+        raise ValueError(f"ERP selection policy missing {sorted(required - set(policy))}")
+    if policy["minimum_benchmark_events"] <= 0:
+        raise ValueError("ERP minimum benchmark events must be positive")
+    for key in ("max_log2_cardinality_distance", "max_parameter_distance", "max_goodness_of_fit"):
+        if not isinstance(policy[key], (int, float)) or policy[key] < 0:
+            raise ValueError(f"ERP {key} must be non-negative")
+    for key in ("minimum_confidence", "minimum_confidence_margin"):
+        if not isinstance(policy[key], (int, float)) or not 0 <= policy[key] <= 1:
+            raise ValueError(f"ERP {key} must be within [0,1]")
+
+
+def validate_result(name, result, expected_contract, candidate_space):
+    if result.get("contract") != expected_contract:
+        raise ValueError(f"{name} did not execute the identical evaluation contract")
+    missing = REQUIRED_METRICS - set(result.get("metrics", {}))
+    if missing or "selected_plan" not in result or "selected_candidates" not in result:
+        raise ValueError(f"{name} result missing selection/metrics: {sorted(missing)}")
+    selected = result["selected_candidates"]
+    if not isinstance(selected, list) or (name != "exact" and not selected):
+        raise ValueError(f"{name} must report selected candidate records")
+    legal = {canonical(candidate) for candidate in candidate_space}
+    if any(canonical(candidate) not in legal for candidate in selected):
+        raise ValueError(f"{name} selected a candidate outside the common space")
+    metrics = result["metrics"]
+    state_bytes, error = metrics["state_bytes"], metrics["max_error"]
+    if not isinstance(state_bytes, int) or state_bytes < 0 or not isinstance(error, (int, float)):
+        raise ValueError(f"{name} reported invalid state/error metrics")
+    if name != "exact" and state_bytes > expected_contract["memory_budget_bytes"]:
+        raise ValueError(f"{name} exceeded the common memory budget")
+    accuracy = expected_contract["accuracy"]
+    if accuracy.get("metric") != "max_error" or "upper_bound" not in accuracy:
+        raise ValueError("accuracy contract must define max_error upper_bound")
+    if error > accuracy["upper_bound"]:
+        raise ValueError(f"{name} violated the common accuracy constraint")
+
+
+def run_arm(arm, expected_contract, candidate_space, cwd, dataset_path=None):
+    with tempfile.NamedTemporaryFile(prefix="asap-figure1-time-", delete=False) as timing:
+        timing_path = timing.name
+    argv = ["/usr/bin/time", "-f", '{"user_seconds":%U,"system_seconds":%S,"peak_rss_kb":%M}', "-o", timing_path, "--"] + arm["command"]
+    started = time.monotonic_ns()
+    env = os.environ.copy()
+    env["ASAP_FIGURE1_CONTRACT_JSON"] = canonical(expected_contract)
+    if dataset_path is not None:
+        env["ASAP_FIGURE1_DATASET"] = str(dataset_path)
+    proc = subprocess.run(argv, cwd=cwd, capture_output=True, text=True, env=env)
+    wall_ns = time.monotonic_ns() - started
+    try:
+        timing_lines = pathlib.Path(timing_path).read_text().splitlines()
+        resources = json.loads(timing_lines[-1])
+    finally:
+        pathlib.Path(timing_path).unlink(missing_ok=True)
+    record = {"name": arm["name"], "command": arm["command"], "exit_code": proc.returncode,
+              "measured_resources": {**resources, "wall_seconds": wall_ns / 1e9}}
+    if proc.returncode:
+        record.update({"status": "failed", "stderr": proc.stderr})
+        return record
+    result = json.loads(proc.stdout)
+    validate_result(arm["name"], result, expected_contract, candidate_space)
+    record.update({"status": "completed", "selected_plan": result["selected_plan"],
+                   "metrics": result["metrics"], "provenance": result.get("provenance", {})})
+    return record
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--manifest", required=True, type=pathlib.Path)
+    parser.add_argument("--output", required=True, type=pathlib.Path)
+    args = parser.parse_args()
+    if args.output.exists():
+        parser.error("output already exists")
+    manifest_path = args.manifest.resolve(); root = manifest_path.parent
+    manifest = json.loads(manifest_path.read_text())
+    dataset_path, dataset_sha, constraints = validate_manifest(manifest, root)
+    candidate_sha = hashlib.sha256(canonical(constraints["candidate_space"]).encode()).hexdigest()
+    contract = {"dataset_sha256": dataset_sha, "candidate_space_sha256": candidate_sha,
+                "memory_budget_bytes": constraints["memory_budget_bytes"],
+                "accuracy": constraints["accuracy"], "window": constraints["window"],
+                "erp_observation": constraints["erp_observation"],
+                "erp_selection_policy": constraints["erp_selection_policy"]}
+    report = {"schema_version": 1, "manifest_sha256": hashlib.sha256(canonical(manifest).encode()).hexdigest(),
+              "dataset": {"path": str(dataset_path), "sha256": dataset_sha}, "contract": contract,
+              "host": {"uname": list(os.uname())}, "arms": []}
+    for arm in manifest["arms"]:
+        report["arms"].append(
+            run_arm(arm, contract, constraints["candidate_space"], root, dataset_path)
+        )
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    with open(args.output, "x") as sink:
+        json.dump(report, sink, indent=2); sink.write("\n")
+    if any(row["status"] != "completed" for row in report["arms"]):
+        raise SystemExit(1)
+
+if __name__ == "__main__": main()