feat(research): add CALF/LegNav comparator diagnostic

configs/benchmarks/issue_7318_calf_legnav_comparator_smoke.yaml

@@ -0,0 +1,22 @@
+schema_version: calf_legnav_comparator_config.v1
+issue: 7318
+
+# This is a one-seed, one-scenario diagnostic fixture. It is not benchmark evidence.
+candidate: ppo_issue791_best_v1
+candidate_registry: docs/context/policy_search/candidate_registry.yaml
+scenario_matrix: configs/scenarios/sets/issue_3320_occluded_emergence_live_replay.yaml
+scenario_name: issue_2756_occluded_emergence
+stage: smoke
+seed: 111
+horizon: 12
+dt_s: 0.1
+personal_space_radius_m: 1.5
+timeout_seconds: 180
+
+conditions:
+  perfect_perception:
+    ignore_fixture_visibility: true
+  sensor_limited:
+    position_noise_std_m: 0.15
+    position_noise_bound_m: 0.3
+    observation_perturbation_seed: 7318

docs/README.md

@@ -335,6 +335,7 @@ design. See `CONTRIBUTING.md` for the strict-build contract and the curated sour
 * **[SocNav Asset Setup (License-Safe)](./socnav_assets_setup.md)** - Official-source download/staging instructions for SocNav third-party datasets with validation commands
 * **[Benchmark Runner & Metrics](./benchmark.md)** - Episode schema, aggregation, metrics suite (collisions, comfort exposure, SNQI), and validation hooks
 * **[Prediction, Planning, and Runtime-Safety Diagnostic](./prediction_planning_safety_diagnostic.md)** - Fixture-only `prediction_planning_safety.v1` trace contract, split coverage, runtime event accounting, and claim boundary for issue #7317
+* **[CALF/LegNav Comparator Diagnostic](./calf_legnav_comparator.md)** - One-seed paired ideal/perception-limited policy smoke with explicit source-transfer and metric-proxy boundaries
 * **[Issue #1434 Stress/Uncertainty Coverage Schema v1](./context/issue_1434_stress_uncertainty_coverage_schema.md)** - `stress_uncertainty_coverage.v1` field contract, statistical summary tiers, coverage axes, interpretation boundaries, and fail-closed consumer rules for benchmark reports
 * **[Full Classic Interaction Benchmark](./benchmark_full_classic.md)** - Complete guide: episodes, aggregation, effect sizes, adaptive precision, plots, videos, scaling metrics
 * **[Benchmark Artifact Publication](./benchmark_artifact_publication.md)** - Public artifact policy, DOI-ready export bundles, release/Zenodo workflow

docs/calf_legnav_comparator.md

@@ -0,0 +1,108 @@
+# CALF/LegNav-inspired comparator diagnostic
+
+This page documents a narrowly scoped, one-seed Robot SF smoke for asking whether
+the existing policy-search trace can separate an ideal observation contract from a
+perception-limited contract. It is diagnostic-only evidence. It does not reproduce
+the external CALF policy, the LegNav simulator, a calibrated leg sensor, TurtleBot 4
+hardware, or real-world deployment.
+
+## Source and method boundary
+
+The source paper, [Learning Social Robot Navigation By Sensing Human Legs](https://arxiv.org/abs/2607.27922),
+describes CALF (Convolutional Attention for Leg Features), a leg-sensing policy evaluated
+in the LegNav lightweight 2-D simulator and reported with a TurtleBot 4 deployment. The
+Robot SF comparator records that method card for provenance, but imports none of its
+checkpoint, sensor calibration, gait model, simulator, embodiment, or training recipe.
+Temporal features, source action bounds, and Social Force/HSFM calibration are also
+recorded as unavailable rather than inferred from the local PPO adapter.
+
+The local question is smaller: with the same Robot SF PPO candidate, scenario, and seed,
+what changes when the policy receives the fixture's ideal state versus a fixed-shape,
+perception-limited observation? A result cannot support a CALF ranking, sensor-realism
+claim, safety claim, universal planner claim, or zero-shot transfer claim.
+
+## Reproducible fixture smoke
+
+From the repository root, run:
+
+```bash
+uv run python scripts/benchmark/run_calf_legnav_comparator_issue_7318.py \
+  --config configs/benchmarks/issue_7318_calf_legnav_comparator_smoke.yaml \
+  --output-dir /tmp/issue-7318-calf-smoke
+```
+
+The config freezes candidate `ppo_issue791_best_v1`, scenario
+`issue_2756_occluded_emergence`, seed `111`, horizon `12`, and a `1.5 m` personal-space
+threshold. The perfect-perception condition disables only the fixture's first-visible
+mask; the sensor-limited condition retains that mask and adds bounded position noise.
+The runner executes both conditions through the existing policy-search step diagnostics,
+then validates `summary.json` against
+`robot_sf/benchmark/schemas/calf_legnav_comparator.v1.json`.
+The YAML itself is checked against
+`robot_sf/benchmark/schemas/calf_legnav_comparator_config.v1.json` before either
+condition starts, including a strict finite-JSON check for numeric values.
+The report provenance also records the resolved selected scenario, its manifest
+include graph, and referenced map/route-override digests; an unreadable input
+blocks the handoff. A learned-policy trace is admitted only when both paired
+conditions report a successful load, the same model identity, and a
+`computed_resolved_file` checkpoint digest matching the model registry.
+
+Use `--dry-run` to inspect the two generated commands without executing the policy.
+Generated traces and summaries belong under ignored `output/` or a disposable temporary
+directory; they are not durable benchmark evidence by themselves.
+
+## Metric mapping
+
+The report keeps local and proxy semantics explicit:
+
+| Report field | Local source | Mapping |
+| --- | --- | --- |
+| Success and collision rates | trace outcome and collision flags | exact local |
+| Minimum human distance and personal-space compliance | simulator ground-truth distance | qualified proxy |
+| Angular jerk | second differences of commanded angular action | qualified proxy |
+| Action smoothness | successive two-channel action difference | exact local |
+| Timeout rate | trace termination fields | exact local |
+
+The external CALF/LegNav action envelope is unavailable and is never inferred. For
+the local PPO candidate, the runner records the effective `action_space`, resolved
+Robot SF kinematics, and finite linear/angular command bounds in
+`policy_action_contract`. Each emitted policy command is checked against that
+contract before environment conversion; a missing, incomplete, or out-of-range
+contract blocks the paired condition rather than allowing downstream clipping to
+look like valid policy output.
+
+For each executed action, the distance metrics use the conservative minimum of
+the available pre-step and post-step ground-truth distances. This keeps the
+shared state between adjacent rows from being counted twice while preserving
+within-step clearance violations. Every executed row must expose at least one
+finite non-negative distance field; a partial distance trace leaves both distance
+metrics unavailable instead of silently changing their denominator. Outcome
+flags must be JSON booleans; missing or malformed flags are reported as
+unavailable rather than coerced into results. Success, collision, and timeout
+must all be available before a condition can be admitted as `available`. When
+terminal `done_info.success` is present, a true row-level `is_success` flag must
+not contradict a false terminal outcome; such a contradiction leaves success and
+dependent timeout metrics unavailable. Trace rows must also have contiguous step
+identities, a complete fixed horizon or explicit terminal `done_info`, recognized
+execution-mode provenance, non-negative integer observed-actor counts, and finite
+non-negative distance values. Violations block the condition or materialize a
+schema-valid blocked handoff. The configured observation perturbation profile is
+also checked against the trace before the paired contrast is admitted.
+
+Trace and report writers use strict JSON. Non-finite diagnostic values are represented as
+`null` or reject the handoff, so they remain unavailable rather than becoming measurements.
+
+One paired episode has no uncertainty estimate. Missing observations, runner errors,
+fallback/degraded execution, or an unrecognized observation contract produce `blocked`
+or `unavailable` fields rather than fabricated zeros.
+An unavailable or mislabelled observation contract also blocks that condition's metric
+values and paired deltas, so a failed contrast cannot be read as a valid comparison.
+
+## Evidence and next proof
+
+The fixture smoke is useful for validating provenance, observation separation, and metric
+plumbing. It is not a research result about CALF or LegNav. A stronger research result
+requires a preregistered multi-seed, multi-scenario comparison with explicit sensor and
+embodiment assumptions, followed by review of whether a real manifest and source-policy
+artifacts are actually available. Until then, keep this comparator in the diagnostic
+lane and do not promote its output to a benchmark or paper-facing claim.

robot_sf/baselines/ppo.py

@@ -137,6 +137,7 @@ def __init__(
         self.config = self._parse_config(config)
         self._seed = seed
         self._model = None
+        self._resolved_model_path: Path | None = None
         self._status = "ok"
         self._fallback_reason: str | None = None
         self._predictive_foresight: PredictiveForesightEncoder | None = None
@@ -164,6 +165,7 @@ def _parse_config(self, cfg: PPOPlannerConfig | dict[str, Any]) -> PPOPlannerCon
 
     def _load_model(self) -> None:
         """Load the PPO model from disk or enter fallback mode."""
+        self._resolved_model_path = None
         if self.config.model_id is None:
             validate_no_local_model_path_value(
                 self.config.model_path,
@@ -223,6 +225,7 @@ def _load_model(self) -> None:
         try:
             # Avoid printing system info in CI/test logs
             self._model = PPO.load(str(mp), device=self.config.device, print_system_info=False)
+            self._resolved_model_path = mp
             self._status = "ok"
             self._fallback_reason = None
         except (RuntimeError, ValueError, OSError) as e:
@@ -268,6 +271,7 @@ def configure(self, config: PPOPlannerConfig | dict[str, Any]) -> None:
         """Update the planner's configuration."""
         self.config = self._parse_config(config)
         self._model = None
+        self._resolved_model_path = None
         self._initialized = False
         if not self._defer_model_loading:
             self._ensure_model_loaded()
@@ -296,17 +300,47 @@ def _init_predictive_foresight(self) -> None:
 
     # --- API -----------------------------------------------------------
     def step(self, obs: Observation | dict[str, Any]) -> dict[str, float]:
-        """Compute a planner action for the given observation.
+        """Compute a projected planner action for the given observation.
 
         Args:
             obs: SocNav-style observation or Observation instance.
 
         Returns:
-            Action dict in either velocity or unicycle format.
+            Action dict in either velocity or unicycle format, projected into
+                the configured runtime bounds.
+        """
+        return self._step(obs, project=True)
+
+    def step_raw(self, obs: Observation | dict[str, Any]) -> dict[str, float]:
+        """Compute the model-space action before runtime projection.
+
+        The returned mapping uses the same stable keys as :meth:`step`, but
+        preserves the raw model values so benchmark adapters can validate the
+        policy contract before clipping or rescaling. Goal-directed fallback
+        actions remain bounded because they are not model outputs.
+
+        Args:
+            obs: SocNav-style observation or Observation instance.
+
+        Returns:
+            Raw model action dict in either velocity or unicycle format.
+        """
+        return self._step(obs, project=False)
+
+    def _step(
+        self,
+        obs: Observation | dict[str, Any],
+        *,
+        project: bool,
+    ) -> dict[str, float]:
+        """Run one prediction while selecting raw or projected action output.
+
+        Returns:
+            Action dict in the configured action space.
         """
         self._ensure_model_loaded()
         if is_observation_mapping(obs) and self._uses_dict_observation():
-            return self._step_dict_obs(obs)
+            return self._step_dict_obs(obs, project=project)
 
         if is_observation_mapping(obs):
             obs = observation_from_mapping(obs)
@@ -319,11 +353,11 @@ def step(self, obs: Observation | dict[str, Any]) -> dict[str, float]:
                 action_vec = self._predict_action(model_obs)
                 if action_vec is None:
                     raise RuntimeError("PPO model unavailable or prediction failed")
-                return self._action_vec_to_dict_from_array(action_vec)
+                return self._action_vec_to_dict_from_array(action_vec, project=project)
             action_vec = self._predict_action(obs)
             if action_vec is None:
                 raise RuntimeError("PPO model unavailable or prediction failed")
-            return self._action_vec_to_dict(action_vec, obs)
+            return self._action_vec_to_dict(action_vec, obs, project=project)
         except (RuntimeError, ValueError, OSError):
             # Fallback for robustness on common prediction errors
             if self.config.fallback_to_goal:
@@ -338,7 +372,7 @@ def _uses_dict_observation(self) -> bool:
         """Return whether planner is configured for native dict observations."""
         return str(self.config.obs_mode).strip().lower() in {"dict", "native_dict", "multi_input"}
 
-    def _step_dict_obs(self, obs: dict[str, Any]) -> dict[str, float]:
+    def _step_dict_obs(self, obs: dict[str, Any], *, project: bool) -> dict[str, float]:
         """Predict an action from flattened dict observations expected by MultiInput PPO.
 
         Returns:
@@ -349,7 +383,7 @@ def _step_dict_obs(self, obs: dict[str, Any]) -> dict[str, float]:
             action_vec = self._predict_action(model_obs)
             if action_vec is None:
                 raise RuntimeError("PPO model unavailable or prediction failed")
-            return self._action_vec_to_dict_from_array(action_vec)
+            return self._action_vec_to_dict_from_array(action_vec, project=project)
         except (RuntimeError, ValueError, OSError):
             if self.config.fallback_to_goal:
                 if self._status != "fallback":
@@ -601,6 +635,7 @@ def _align_model_obs_dict(
                         f"expected {target_shape}",
                     )
                 arr = arr.reshape(target_shape)
+            self._validate_model_observation_value(key, arr, sub_space)
             converted[key] = arr
         if backfilled:
             logger.debug(
@@ -610,6 +645,19 @@ def _align_model_obs_dict(
             )
         return converted
 
+    @staticmethod
+    def _validate_model_observation_value(key: str, value: Any, sub_space: Any) -> None:
+        """Reject a supplied observation value outside its declared model space."""
+        contains = getattr(sub_space, "contains", None)
+        if not callable(contains):
+            return
+        try:
+            in_bounds = bool(contains(value))
+        except (TypeError, ValueError):
+            in_bounds = False
+        if not in_bounds:
+            raise ValueError(f"Observation key '{key}' is outside the model-declared space")
+
     @classmethod
     def _default_for_space(cls, sub_space: Any) -> Any:
         """Build an in-bounds default payload for a missing model observation key.
@@ -724,6 +772,7 @@ def _build_model_obs_flat_box(
                     f"got shape {tuple(flat_obs.shape)}, expected {target_shape}."
                 )
             flat_obs = flat_obs.reshape(target_shape)
+        self._validate_model_observation_value("<flat>", flat_obs, model_space)
         return flat_obs
 
     def _predictive_feature_payload(self, obs: dict[str, Any]) -> dict[str, np.ndarray]:
@@ -834,8 +883,16 @@ def _vectorize(self, obs: Observation) -> np.ndarray:
         vec = np.concatenate([rel_goal, rv, ped_flat]).astype(float)
         return vec
 
-    def _action_vec_to_dict_from_array(self, act: np.ndarray) -> dict[str, float]:
-        """Convert a raw action vector to the configured action dictionary.
+    def _action_vec_to_dict_from_array(
+        self,
+        act: np.ndarray,
+        *,
+        project: bool = True,
+    ) -> dict[str, float]:
+        """Convert an action vector to the configured action dictionary.
+
+        ``project=False`` preserves the model output for callers that need to
+        validate policy-space bounds before runtime conversion.
 
         Returns:
             Action dict in either velocity or unicycle format.
@@ -844,27 +901,35 @@ def _action_vec_to_dict_from_array(self, act: np.ndarray) -> dict[str, float]:
             # Expect [v, omega]
             v = float(act[0]) if act.size >= 1 else 0.0
             w = float(act[1]) if act.size >= 2 else 0.0
-            v = max(0.0, min(v, self.config.v_max))
-            w = max(-self.config.omega_max, min(w, self.config.omega_max))
+            if project:
+                v = max(0.0, min(v, self.config.v_max))
+                w = max(-self.config.omega_max, min(w, self.config.omega_max))
             return {"v": v, "omega": w}
         # Default velocity space: expect [vx, vy]
         vx = float(act[0]) if act.size >= 1 else 0.0
         vy = float(act[1]) if act.size >= 2 else 0.0
         # Optional clamp to v_max
-        spd = float(np.hypot(vx, vy))
-        if spd > self.config.v_max and spd > self.EPS:
-            scale = self.config.v_max / (spd + self.EPS)
-            vx *= scale
-            vy *= scale
+        if project:
+            spd = float(np.hypot(vx, vy))
+            if spd > self.config.v_max and spd > self.EPS:
+                scale = self.config.v_max / (spd + self.EPS)
+                vx *= scale
+                vy *= scale
         return {"vx": vx, "vy": vy}
 
-    def _action_vec_to_dict(self, act: np.ndarray, _obs: Observation) -> dict[str, float]:
+    def _action_vec_to_dict(
+        self,
+        act: np.ndarray,
+        _obs: Observation,
+        *,
+        project: bool = True,
+    ) -> dict[str, float]:
         """Convert raw action vector to configured action dict for Observation mode.
 
         Returns:
             Action dict in configured output space.
         """
-        return self._action_vec_to_dict_from_array(act)
+        return self._action_vec_to_dict_from_array(act, project=project)
 
     def _fallback_action(self, obs: Observation) -> dict[str, float]:
         """Return a simple goal-seeking action when PPO is unavailable.