feat(eval): add Legacy SR acquisition matrix

Extend the canonical Legacy collector to deploy hash-bound symbolic policies across training, generalization, and ablation scenarios with reproducible execution provenance.

Co-authored-by: Cursor <cursoragent@cursor.com>

src/drl_pinball/legacy_test/acquire.py

@@ -5,7 +5,10 @@ import argparse
 import hashlib
 import json
 import os
+import platform
 import shutil
+import subprocess
+import sys
 from collections import deque
 from pathlib import Path
 from typing import Any, Callable
@@ -21,17 +24,26 @@ from drl_pinball.acquisition import (
     write_phase_cycle_artifacts,
 )
 from .cases import get_case
+from SR_analysis.configs import get_scene
+from SR_analysis.stage_3_validate import SymbolicPolicy, load_formula_pair
+from SR_analysis.utils.data_contracts import CAUSAL_ALIGNMENT
 from .metrics import erase_reward_terms, frozen_reference_comparison, reward_terms, sha256_file
 from .core.dtw_metrics import gen_target_states_at
 from .runtime import erase_policy_observation, load_policy_norm, policy_observation, reset_runtime, run_historical_interval
 
 CASE_NAME, ROLE = "karman_re100", "controlled"  # compatibility defaults
 PERIODIC_CASES = ("karman_re50", "karman_re100", "karman_re200", "karman_re400",
-                  "illusion_075L", "illusion_1L", "illusion_15L")
+                  "illusion_075L", "illusion_1L", "illusion_15L",
+                  "karman_re25", "karman_re70", "karman_re150", "karman_re300",
+                  "illusion_05L", "illusion_06L", "illusion_08L", "illusion_12L", "illusion_2L")
 SUPPORTED_CASES = PERIODIC_CASES + ("steady", "vortex_lamb", "vortex_taylor", "erase")
-PERIODIC_ROLES = ("controlled", "target", "zero")
+PERIODIC_ROLES = ("controlled", "target", "zero", "sr")
 STEADY_ROLES = ("target", "constant", "zero")
 ROLES = PERIODIC_ROLES
+SR_VARIANTS = ("k_front0", "k_rear0", "k_rear1", "i_front0", "i_front1", "i_rear0", "i_rear1")
+SR_VARIANT_CASE = {**{name: "karman_re100" for name in SR_VARIANTS[:3]},
+                   **{name: "illusion_15L" for name in SR_VARIANTS[3:]}}
+SR_ROLE_DIR = {name: f"sr_{name}" for name in SR_VARIANTS}
 WARMUP_INTERVALS, COLLECT_BOUNDARIES = 480, 160
 NX, U0, SI, CYCLE_LENGTH = 1280, 0.01, 800, 30  # compatibility defaults
 MIN_FREE_BYTES = 4 * 1024**3
@@ -47,6 +59,18 @@ def _atomic_json(path: Path, payload: dict[str, Any]) -> None:
     os.replace(temporary, path)
 
 
+def _execution_provenance(repo_root: Path):
+    try:
+        git_sha = subprocess.run(["git", "-C", str(repo_root), "rev-parse", "HEAD"], check=True, capture_output=True, text=True).stdout.strip()
+        git_dirty = bool(subprocess.run(["git", "-C", str(repo_root), "status", "--porcelain"], check=True, capture_output=True, text=True).stdout)
+    except (OSError, subprocess.CalledProcessError):
+        git_sha, git_dirty = None, None
+    return {"git_sha": git_sha, "git_dirty": git_dirty,
+            "environment": {"python": sys.version, "platform": platform.platform(), "numpy": np.__version__,
+                            "conda_env": os.environ.get("CONDA_DEFAULT_ENV"),
+                            "cuda_visible_devices": os.environ.get("CUDA_VISIBLE_DEVICES")}}
+
+
 def _identity(path: Path) -> dict[str, Any]:
     return {"path": str(path.resolve()), "sha256": sha256_file(str(path)), "bytes": path.stat().st_size}
 
@@ -103,6 +127,7 @@ def _role_semantics(role: str) -> str:
         "target": "live builder-generated target trajectory; actions, rewards, and policy normalization unavailable",
         "zero": "physical-zero/uncontrolled trajectory using counter-bias normalized action; frozen norm used only for native reward evaluation",
         "constant": "established steady constant control at case.action_bias*U0 from normalized action zero",
+        "sr": "causal symbolic-regression policy using hash-bound article formulas",
     }[role]
 
 
@@ -118,7 +143,53 @@ def _controlled_reference(role, bundle, columns, conv_len):
 def _case_roles(case):
     if case.name not in SUPPORTED_CASES:
         raise ValueError(f"unsupported Legacy acquisition case {case.name!r}")
-    return STEADY_ROLES if case.scene == "steady" else PERIODIC_ROLES
+    if case.scene == "steady": return STEADY_ROLES
+    if case.scene in ("vortex", "erase"): return ("controlled", "target", "zero")
+    return PERIODIC_ROLES if case.model is not None else ("target", "zero", "sr")
+
+
+def _sr_formula_paths(repo_root: Path, case, variant: str | None = None):
+    if case.scene not in ("karman", "illusion") or case.sr_scene is None:
+        raise ValueError(f"case {case.name!r} has no SR deployment contract")
+    if variant is not None and SR_VARIANT_CASE.get(variant) != case.name:
+        raise ValueError(f"SR variant {variant!r} is restricted to {SR_VARIANT_CASE.get(variant)!r}")
+    family = case.scene
+    canonical = repo_root / "src/SR_analysis/results/runs" / f"article-refit-{family}-topology-a-20260718/formulas"
+    paths = {"front": canonical / "joint_front.json", "rear": canonical / "joint_rear_shared_upper.json"}
+    if variant is not None:
+        head = "front" if "front" in variant else "rear"
+        term = variant[-1]
+        paths[head] = repo_root / "src/SR_analysis/results/runs/article-ablation-formulas-v2-20260718/formulas" / f"{family}_{head}__delete_t{term}.json"
+    return paths
+
+
+def _assert_sr_contract(case, cfg, data):
+    expected = {"scene_id": case.scene, "re_code": case.re_code,
+                "sample_interval": case.sample_interval, "action_scale": case.action_scale,
+                "action_bias": tuple(case.action_bias), "n_objects_env": 7 if case.scene == "karman" else 6,
+                "obs_slice": (2, 14) if case.scene == "karman" else (0, 12)}
+    for key, value in expected.items():
+        actual = tuple(cfg[key]) if key in ("action_bias", "obs_slice") else cfg[key]
+        if actual != value:
+            raise ValueError(f"SR/Legacy contract mismatch for {key}: {actual!r} != {value!r}")
+    if case.scene == "illusion" and float(cfg["target_radius"]) != float(case.target_radius_l):
+        raise ValueError("SR/Legacy target geometry contract mismatch")
+    config = data["config"]
+    if (int(config["sample_interval"]), int(config["n_obj_total"])) != (case.sample_interval, expected["n_objects_env"]):
+        raise ValueError("SR/Legacy built runtime contract mismatch")
+
+
+def _resolve_sr(repo_root: Path, case, variant: str | None):
+    cfg = get_scene(case.sr_scene)
+    paths = _sr_formula_paths(repo_root, case, variant)
+    pair = load_formula_pair(paths["front"], paths["rear"])
+    harmonics_path = (repo_root / "src/SR_analysis/data/illusion" / case.sr_scene / "target_harmonics.json"
+                      if case.scene == "illusion" else None)
+    harmonics = None
+    if harmonics_path is not None:
+        with harmonics_path.open(encoding="utf-8") as stream: harmonics = json.load(stream)
+    return {"cfg": cfg, "pair": pair, "formula_paths": paths, "variant": variant,
+            "harmonics": harmonics, "harmonics_path": harmonics_path}
 
 
 def _resolve_bundle(
@@ -129,7 +200,7 @@ def _resolve_bundle(
     _case_roles(case)
     frozen = repo_root / "src" / "SR_analysis" / "data" / case.scene / (case.reference_case or case.name)
     needs_model = case.model is not None and role == "controlled"
-    needs_frozen_policy = (case.model is not None and role != "target" and
+    needs_frozen_policy = (case.model is not None and role not in ("target", "sr") and
                            case.scene != "erase" and
                            not (case.scene == "vortex" and role == "zero"))
     norm_path = frozen / "norm.json" if needs_frozen_policy else None
@@ -248,7 +319,7 @@ def _observation_slices(case, role, boundary_obs):
 
 
 def _collect_role(role, data, model, ff, policy_norm, scratch: Path,
-                  capture_field: Callable = _capture_boundary_field, case_name: str = CASE_NAME):
+                  capture_field: Callable = _capture_boundary_field, case_name: str = CASE_NAME, sr_policy=None):
     case = get_case(case_name)
     if role not in _case_roles(case):
         raise ValueError(f"unknown acquisition role {role!r} for {case.name}")
@@ -268,11 +339,17 @@ def _collect_role(role, data, model, ff, policy_norm, scratch: Path,
     field_buffer = None
     total = WARMUP_INTERVALS + COLLECT_BOUNDARIES
     harmonics = data.get("target_harmonics")
+    if role == "sr" and sr_policy is None: raise ValueError("sr role requires a symbolic policy")
+    raw_for_policy = np.asarray(data["norm"]["save_states"], np.float64)[-1].copy() if role == "sr" else None
     steady = case.scene == "steady"
     for interval in range(total):
         if role == "controlled":
             action, _ = model.predict(obs, deterministic=True)
             action, command = _command(case, action, expected_objects, ff.DATA_TYPE)
+        elif role == "sr":
+            omega, _, _ = sr_policy.action(raw_for_policy, interval)
+            action = (np.asarray(omega, np.float64) / U0 - np.asarray(case.action_bias)) / case.action_scale
+            action, command = _command(case, action, expected_objects, ff.DATA_TYPE)
         elif role == "zero":
             action = -np.asarray(case.action_bias, np.float32) / np.float32(case.action_scale)
             action, command = _command(case, action, expected_objects, ff.DATA_TYPE)
@@ -286,6 +363,8 @@ def _collect_role(role, data, model, ff, policy_norm, scratch: Path,
             action = np.full(3, np.nan, np.float32); command = np.zeros(expected_objects, ff.DATA_TYPE)
         _run_interval(ff, si, command)
         boundary_obs = np.asarray(ff.obs, np.float32).copy()
+        if role == "sr":
+            sr_policy.observe(np.asarray(command[-3:], np.float64)); raw_for_policy = boundary_obs[:12] if case.scene == "illusion" else boundary_obs[2:14]
         sensors, forces = _observation_slices(case, role, boundary_obs)
         if role == "target":
             effective = np.full(3, np.nan, np.float32)
@@ -297,7 +376,7 @@ def _collect_role(role, data, model, ff, policy_norm, scratch: Path,
                 terms = {name: np.nan for name in ("reward", "reward_cd", "reward_cl", "native_legacy_dtw")}
             else:
                 terms = reward_terms(case, data["target_states"], harmonics, np.asarray(fifo), policy_norm["force_norm_fact"], interval)
-            if not steady:
+            if not steady and role != "sr":
                 target_force = gen_target_states_at(interval + 1, harmonics)[:2] if harmonics is not None else None
                 obs = policy_observation(raw, policy_norm, target_force=target_force)
             effective = np.asarray(ff.current_effective_action(), np.float32)[-3:].copy()
@@ -378,6 +457,17 @@ PHASE_FIELD_KEYS = {
 }
 
 
+def _sr_metadata(sr):
+    if sr is None: return None
+    pair = sr["pair"]
+    files = {name: _identity(path) for name, path in sr["formula_paths"].items()}
+    return {"variant": sr["variant"], "formula_ids": {name: doc["artifact_id"] for name, doc in (("front", pair.front), ("rear", pair.rear))},
+            "formula_files": files, "pair_hash": pair.pair_hash,
+            "deployment_expression_hashes": {"front": pair.front["deployment_expression_hash"], "rear": pair.rear["deployment_expression_hash"]},
+            "alignment": CAUSAL_ALIGNMENT, "initialization": "both action lags seeded from cfg.policy_init_action; first action uses current raw post-FIFO state; observe after interval",
+            "config": _canonical_json_value(sr["cfg"]), "harmonics_file": _optional_file_provenance(sr["harmonics_path"])}
+
+
 def _finalize(role_dir: Path, scratch: Path, rows, field_buffer, data, bundle, storage, role="controlled"):
     case = bundle["case"]; columns = _columns(rows)
     phase_meta = {}
@@ -446,6 +536,7 @@ def _finalize(role_dir: Path, scratch: Path, rows, field_buffer, data, bundle, s
         "field_contract": {"quantity": "q/RHO_ref", "RHO_ref": 1, "source": "public current_step_velocity_field q/U0 multiplied by field_config.velocity U0"},
         "field_count": 1 if case.scene == "steady" else 8, "nonperiodic": case.scene == "steady",
         "role_semantics": _role_semantics(role),
+        "sr_deployment": _sr_metadata(bundle.get("sr")) if role == "sr" else None,
         "timeline_contract": "V5 lattice_step is an absolute solver count; Legacy physical_time is derived only from relative Legacy lattice_step",
         "source_provenance": {
             "model_file": _identity(bundle["model_path"]) if role == "controlled" else None,
@@ -461,6 +552,7 @@ def _finalize(role_dir: Path, scratch: Path, rows, field_buffer, data, bundle, s
                                  ("builder_recomputed" if role != "target" else "not_applicable")),
         "scratch_cleanup": {"complete": not scratch.exists(), "path": str(scratch)},
         "resolved_optane_path": str(storage["resolved_output_root"]), "phase_variables": list(phase_values),
+        "execution_provenance": _execution_provenance(Path(__file__).resolve().parents[3]),
         **phase_meta})
 
 
@@ -484,8 +576,10 @@ def _validate_staged_role(role_dir: Path, case_name: str = CASE_NAME) -> None:
 
 def _acquire_standard(role, *, case_name=CASE_NAME, output_root=None, overwrite=False, device_id=0,
                  repo_root=None, storage_validator=validate_output_storage,
-                 runtime_factory=None, finalizer=_finalize, **unused):
+                 runtime_factory=None, finalizer=_finalize, variant=None, **unused):
     case = get_case(case_name)
+    if variant is not None and role != "sr": raise ValueError("--variant is valid only with role sr")
+    if variant is not None and variant not in SR_VARIANTS: raise ValueError(f"unknown SR variant {variant!r}")
     if role not in _case_roles(case): raise ValueError(f"unknown acquisition role {role!r} for {case_name}")
     repo = Path(repo_root) if repo_root is not None else Path(__file__).resolve().parents[3]
     storage = storage_validator(repo_mapping=default_reproduction_mapping(repo),
@@ -494,7 +588,9 @@ def _acquire_standard(role, *, case_name=CASE_NAME, output_root=None, overwrite=
         bundle = _resolve_bundle(repo)
     else:
         bundle = _resolve_bundle(repo, case_name, role)
-    prepared = prepare_role_output(storage, "legacy", case_name, role, overwrite=overwrite)
+    sr = _resolve_sr(repo, case, variant) if role == "sr" else None
+    output_role = SR_ROLE_DIR[variant] if variant is not None else role
+    prepared = prepare_role_output(storage, "legacy", case_name, output_role, overwrite=overwrite)
     scratch = create_scratch(prepared["scratch_root"])
     ff = None
     try:
@@ -502,9 +598,14 @@ def _acquire_standard(role, *, case_name=CASE_NAME, output_root=None, overwrite=
             data, model, ff = _default_runtime(bundle["case"], bundle, device_id, role)
         else:
             data, model, ff = runtime_factory(bundle["case"], bundle, device_id, role)
+        if sr is not None:
+            _assert_sr_contract(case, sr["cfg"], data)
+            if sr["harmonics"] is not None: data["target_harmonics"] = sr["harmonics"]
+            bundle["sr"] = sr
         policy_norm = (None if role == "target" else
-                       (data["norm"] if bundle["case"].scene == "steady" else bundle["norm"]))
+                       (data["norm"] if role == "sr" or bundle["case"].scene == "steady" else bundle["norm"]))
-        rows, fields = _collect_role(role, data, model, ff, policy_norm, scratch, case_name=case_name)
+        policy = SymbolicPolicy(sr["pair"], sr["cfg"], sr["harmonics"]) if sr is not None else None
+        rows, fields = _collect_role(role, data, model, ff, policy_norm, scratch, case_name=case_name, sr_policy=policy)
         finalizer(prepared["role_dir"], scratch, rows, fields, data, bundle, storage, role)
         _validate_staged_role(prepared["role_dir"], case_name)
         return publish_role_output(prepared)
@@ -792,26 +893,43 @@ def _acquire_special(role,case_name,output_root,overwrite,device_id,repo_root,st
     finally:
         if ff is not None and hasattr(ff,"close"): ff.close()
 
-def acquire_role(role, *, case_name=CASE_NAME, output_root=None, overwrite=False, device_id=0, repo_root=None, storage_validator=validate_output_storage, runtime_factory=None, finalizer=_finalize, vortex_y_offset_l0=0):
+def acquire_role(role, *, case_name=CASE_NAME, output_root=None, overwrite=False, device_id=0, repo_root=None, storage_validator=validate_output_storage, runtime_factory=None, finalizer=_finalize, vortex_y_offset_l0=0, variant=None):
     case=get_case(case_name)
+    if variant is not None and role != "sr": raise ValueError("variant is valid only for the sr role")
+    if variant is not None and variant not in SR_VARIANTS: raise ValueError(f"unknown SR variant {variant!r}")
     if role not in _case_roles(case): raise ValueError(f"unknown acquisition role {role!r} for {case_name}")
     if case.scene in ("vortex","erase"):
         return _acquire_special(role,case_name,output_root,overwrite,device_id,repo_root,storage_validator,runtime_factory,vortex_y_offset_l0)
-    return _acquire_standard(role,case_name=case_name,output_root=output_root,overwrite=overwrite,device_id=device_id,repo_root=repo_root,storage_validator=storage_validator,runtime_factory=runtime_factory,finalizer=finalizer)
+    return _acquire_standard(role,case_name=case_name,output_root=output_root,overwrite=overwrite,device_id=device_id,repo_root=repo_root,storage_validator=storage_validator,runtime_factory=runtime_factory,finalizer=finalizer,variant=variant)
 
 def acquire_controlled(**kwargs):
     return acquire_role("controlled", **kwargs)
 
 
+def campaign_matrix():
+    training = PERIODIC_CASES[:7]
+    generalization = PERIODIC_CASES[7:]
+    rows = [(name, "sr", None) for name in training]
+    rows.extend((name, role, None) for name in generalization for role in ("target", "zero", "sr"))
+    rows.extend((case, "sr", variant) for variant, case in SR_VARIANT_CASE.items())
+    return rows
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(description="Acquire Legacy full-matrix evidence")
-    parser.add_argument("case", choices=SUPPORTED_CASES)
+    parser.add_argument("case", choices=SUPPORTED_CASES, nargs="?")
-    parser.add_argument("--role", choices=ROLES + ("constant",), required=True)
+    parser.add_argument("--role", choices=ROLES + ("constant",))
+    parser.add_argument("--variant", choices=SR_VARIANTS)
+    parser.add_argument("--list-campaign", action="store_true")
     parser.add_argument("--device-id", type=int, default=0)
     parser.add_argument("--vortex-y-offset-l0", type=int, default=0)
     parser.add_argument("--output-root", type=Path); parser.add_argument("--overwrite", action="store_true")
     args = parser.parse_args()
-    path = acquire_role(args.role, case_name=args.case, output_root=args.output_root, overwrite=args.overwrite, device_id=args.device_id, vortex_y_offset_l0=args.vortex_y_offset_l0)
+    if args.list_campaign:
+        for case, role, variant in campaign_matrix(): print(case, role, variant or "canonical")
+        print(f"count={len(campaign_matrix())}"); return 0
+    if args.case is None or args.role is None: parser.error("case and --role are required unless --list-campaign is used")
+    path = acquire_role(args.role, case_name=args.case, output_root=args.output_root, overwrite=args.overwrite, device_id=args.device_id, vortex_y_offset_l0=args.vortex_y_offset_l0, variant=args.variant)
     print(path); return 0
 
 

src/drl_pinball/legacy_test/cases.py

@@ -16,12 +16,19 @@ class Case:
     target_radius_l: float | None = None
     vortex_type: str | None = None
     reference_case: str | None = None
+    sr_scene: str | None = None
 
 CASES = {
-    **{f"karman_re{re}": Case(f"karman_re{re}", "karman", f"d1a3o12_re{re}", re_code=float(re)) for re in (50, 100, 200, 400)},
+    **{f"karman_re{re}": Case(f"karman_re{re}", "karman", f"d1a3o12_re{re}", re_code=float(re), sr_scene=f"karman_re{re}") for re in (50, 100, 200, 400)},
-    "illusion_075L": Case("illusion_075L", "illusion", "d1a3o14_250525_imit_075L_2U_400S", sample_interval=400, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=0.75, reference_case="illusion_0.75L"),
+    "illusion_075L": Case("illusion_075L", "illusion", "d1a3o14_250525_imit_075L_2U_400S", sample_interval=400, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=0.75, reference_case="illusion_0.75L", sr_scene="illusion_0.75L"),
-    "illusion_1L": Case("illusion_1L", "illusion", "d1a3o14_250525_imit_1L_2U_600S", sample_interval=600, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=1.0),
+    "illusion_1L": Case("illusion_1L", "illusion", "d1a3o14_250525_imit_1L_2U_600S", sample_interval=600, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=1.0, sr_scene="illusion_1L"),
-    "illusion_15L": Case("illusion_15L", "illusion", "d1a3o14_250525_imit_15L_2U", action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=1.5, reference_case="illusion_1.5L"),
+    "illusion_15L": Case("illusion_15L", "illusion", "d1a3o14_250525_imit_15L_2U", action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=1.5, reference_case="illusion_1.5L", sr_scene="illusion_1.5L"),
+    **{f"karman_re{re}": Case(f"karman_re{re}", "karman", None, re_code=float(re), sr_scene=f"karman_re{re}") for re in (25, 70, 150, 300)},
+    "illusion_05L": Case("illusion_05L", "illusion", None, sample_interval=400, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=0.5, reference_case="illusion_0.5L", sr_scene="illusion_0.5L"),
+    "illusion_06L": Case("illusion_06L", "illusion", None, sample_interval=400, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=0.6, reference_case="illusion_0.6L", sr_scene="illusion_0.6L"),
+    "illusion_08L": Case("illusion_08L", "illusion", None, sample_interval=600, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=0.8, reference_case="illusion_0.8L", sr_scene="illusion_0.8L"),
+    "illusion_12L": Case("illusion_12L", "illusion", None, sample_interval=600, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=1.2, reference_case="illusion_1.2L", sr_scene="illusion_1.2L"),
+    "illusion_2L": Case("illusion_2L", "illusion", None, sample_interval=800, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=2.0, sr_scene="illusion_2L"),
     "vortex_lamb": Case("vortex_lamb", "vortex", "vortex_lamb", steps=150, action_scale=4.0, vortex_type="lamb"),
     "vortex_taylor": Case("vortex_taylor", "vortex", "vortex_taylor", steps=150, action_scale=4.0, vortex_type="taylor"),
     "erase": Case("erase", "erase", "d1a3o12_250729_250326_erase", sample_interval=600, action_scale=8.0, action_bias=(0.0, -8.0, 8.0), conv_len=36),

src/drl_pinball/legacy_test/tests/test_acquire_matrix.py

@@ -7,12 +7,15 @@ import src.drl_pinball.legacy_test.acquire as acquire
 def test_allowlist_roles_and_case_sample_intervals():
     assert acquire.SUPPORTED_CASES == (
         "karman_re50", "karman_re100", "karman_re200", "karman_re400",
-        "illusion_075L", "illusion_1L", "illusion_15L", "steady", "vortex_lamb", "vortex_taylor", "erase",
+        "illusion_075L", "illusion_1L", "illusion_15L",
+        "karman_re25", "karman_re70", "karman_re150", "karman_re300",
+        "illusion_05L", "illusion_06L", "illusion_08L", "illusion_12L", "illusion_2L",
+        "steady", "vortex_lamb", "vortex_taylor", "erase",
     )
     assert [acquire.get_case(name).sample_interval for name in
             ("karman_re50", "illusion_075L", "illusion_1L", "illusion_15L", "steady")] == [800, 400, 600, 800, 800]
     assert acquire._case_roles(acquire.get_case("steady")) == ("target", "constant", "zero")
-    assert acquire._case_roles(acquire.get_case("illusion_075L")) == ("controlled", "target", "zero")
+    assert acquire._case_roles(acquire.get_case("illusion_075L")) == ("controlled", "target", "zero", "sr")
 
 
 def test_scene_observation_slices_are_exact():

src/drl_pinball/legacy_test/tests/test_acquire_roles.py

@@ -230,3 +230,4 @@ def test_role_failure_transaction_preserves_existing(monkeypatch, tmp_path, role
 def test_cli_exposes_all_roles():
     source = Path(acquire.__file__).read_text(encoding="utf-8")
     assert 'choices=ROLES' in source
+    assert '--variant' in source and '--list-campaign' in source

src/drl_pinball/legacy_test/tests/test_acquire_sr.py

@@ -0,0 +1,90 @@
+from pathlib import Path
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+
+import src.drl_pinball.legacy_test.acquire as acquire
+
+
+def test_generalization_registry_has_no_ppo_and_exact_sr_mapping():
+    expected = {
+        "karman_re25": ("karman_re25", 25, 800), "karman_re70": ("karman_re70", 70, 800),
+        "karman_re150": ("karman_re150", 150, 800), "karman_re300": ("karman_re300", 300, 800),
+        "illusion_05L": ("illusion_0.5L", 100, 400), "illusion_06L": ("illusion_0.6L", 100, 400),
+        "illusion_08L": ("illusion_0.8L", 100, 600), "illusion_12L": ("illusion_1.2L", 100, 600),
+        "illusion_2L": ("illusion_2L", 100, 800),
+    }
+    for name, (scene, re_code, si) in expected.items():
+        case = acquire.get_case(name)
+        assert case.model is None and case.sr_scene == scene
+        assert (case.re_code, case.sample_interval) == (re_code, si)
+        assert acquire._case_roles(case) == ("target", "zero", "sr")
+        with pytest.raises(ValueError, match="unknown acquisition role"):
+            acquire.acquire_role("controlled", case_name=name)
+
+
+def test_formula_authority_and_variant_mapping():
+    root = Path(acquire.__file__).resolve().parents[3]
+    canonical = acquire._sr_formula_paths(root, acquire.get_case("karman_re70"))
+    assert "article-refit-karman-topology-a-20260718" in str(canonical["front"])
+    pair = acquire._sr_formula_paths(root, acquire.get_case("karman_re100"), "k_rear1")
+    assert pair["front"] == canonical["front"]
+    assert pair["rear"].name == "karman_rear__delete_t1.json"
+    illusion = acquire._sr_formula_paths(root, acquire.get_case("illusion_15L"), "i_front0")
+    assert illusion["front"].name == "illusion_front__delete_t0.json"
+    assert "article-refit-illusion-topology-a-20260718" in str(illusion["rear"])
+    with pytest.raises(ValueError, match="restricted"):
+        acquire._sr_formula_paths(root, acquire.get_case("karman_re50"), "k_front0")
+
+
+def test_campaign_matrix_is_exact_41_unique_output_roles():
+    matrix = acquire.campaign_matrix()
+    assert len(matrix) == 41
+    keys = [(case, acquire.SR_ROLE_DIR[variant] if variant else role) for case, role, variant in matrix]
+    assert len(keys) == len(set(keys))
+    assert sum(variant is not None for _, _, variant in matrix) == 7
+    assert sum(acquire.get_case(case).model is None for case, _, _ in matrix) == 27
+
+
+def test_sr_symbolic_timing_conversion_and_historical_reset(monkeypatch, tmp_path):
+    case = acquire.get_case("karman_re100")
+    class Policy:
+        def __init__(self): self.events = []
+        def action(self, raw, step):
+            self.events.append(("action", np.asarray(raw).copy(), step))
+            return np.array([.01, -.02, .03]), None, None
+        def observe(self, omega): self.events.append(("observe", np.asarray(omega).copy()))
+    class Flow:
+        DATA_TYPE = np.float32
+        def __init__(self): self.action=np.ones(7,np.float32); self.obs=np.arange(14,dtype=np.float32); self.commands=[]
+        def restore_ddf(self): pass
+        def apply_ddf(self): pass
+        def run(self, steps, command):
+            assert np.array_equal(self.action, np.zeros(7,np.float32))
+            self.commands.append(command.copy()); self.action=command.copy(); self.obs += 1
+        def current_effective_action(self): return self.action
+    data={"config":{"nx":1280,"ny":1,"sample_interval":800,"n_obj_total":7},
+          "norm":{"save_states":np.vstack((np.zeros(12),np.arange(12))).astype(np.float32)},
+          "target_states":np.zeros((30,6),np.float32)}
+    monkeypatch.setattr(acquire,"WARMUP_INTERVALS",0); monkeypatch.setattr(acquire,"COLLECT_BOUNDARIES",1)
+    monkeypatch.setattr(acquire,"reward_terms",lambda *a:{"reward":1.,"reward_cd":1.,"reward_cl":1.,"native_legacy_dtw":1.})
+    policy=Policy(); flow=Flow()
+    rows,_=acquire._collect_role("sr",data,None,flow,{"force_norm_fact":1.},tmp_path,
+        capture_field=lambda *a:{"ux":np.zeros((1,1280)),"uy":np.zeros((1,1280))},sr_policy=policy)
+    assert policy.events[0][0] == "action" and np.array_equal(policy.events[0][1], np.arange(12))
+    assert policy.events[1][0] == "observe"
+    assert np.allclose(flow.commands[0][-3:], [.01,-.02,.03])
+    expected=(np.array([1.,-2.,3.])-np.asarray(case.action_bias))/case.action_scale
+    assert np.allclose(rows[0]["action_normalized"],expected)
+
+
+def test_sr_metadata_binds_formula_and_harmonics(tmp_path):
+    front=tmp_path/"front.json"; rear=tmp_path/"rear.json"; harmonics=tmp_path/"harmonics.json"
+    front.write_text("front"); rear.write_text("rear"); harmonics.write_text("[]")
+    pair=SimpleNamespace(pair_hash="p",front={"artifact_id":"f","deployment_expression_hash":"df"},rear={"artifact_id":"r","deployment_expression_hash":"dr"})
+    meta=acquire._sr_metadata({"pair":pair,"formula_paths":{"front":front,"rear":rear},"variant":None,
+                              "cfg":{"policy_init_action":(0,-4,4)},"harmonics_path":harmonics})
+    assert meta["pair_hash"] == "p" and meta["alignment"] == "causal_post_state_to_next_action"
+    assert all(len(item["sha256"]) == 64 for item in meta["formula_files"].values())
+    assert meta["harmonics_file"]["file_identity"]["sha256"]

src/drl_pinball/legacy_test/tests/test_contracts.py

@@ -18,7 +18,7 @@ from src.drl_pinball.legacy_test.runtime import (
 )
 
 def test_config_matrix():
-    assert set(CASES)=={"karman_re50","karman_re100","karman_re200","karman_re400","illusion_075L","illusion_1L","illusion_15L","vortex_lamb","vortex_taylor","erase","steady"}
+    assert set(CASES)=={"karman_re50","karman_re100","karman_re200","karman_re400","karman_re25","karman_re70","karman_re150","karman_re300","illusion_075L","illusion_1L","illusion_15L","illusion_05L","illusion_06L","illusion_08L","illusion_12L","illusion_2L","vortex_lamb","vortex_taylor","erase","steady"}
     assert [CASES[f"karman_re{x}"].re_code for x in (50,100,200,400)] == [50,100,200,400]
     assert [CASES[x].sample_interval for x in ("illusion_075L","illusion_1L","illusion_15L")] == [400,600,800]
     assert get_case("erase").model == "d1a3o12_250729_250326_erase"