fix(ci): normalize train_series wall_s for replica hash

After telemetry-rt, CPU re-exec replica agreement failed because
metrics.train_series wall_s (and its sha256 pointers) differed per run.
Zero wall_s on normalize and realign digests so two honest replicas
share one manifest_sha256 again.

Author: echobt

src/prism_challenge/evaluator/cpu_test_mode.py

@@ -147,19 +147,44 @@ def configure_cpu_reexec_test_mode(settings: Any) -> Path:
 
 
 def normalize_manifest_for_replication(manifest: dict[str, Any]) -> dict[str, Any]:
-    """Return a deep copy with volatile ``compute`` timing/memory fields fixed to a constant.
+    """Return a deep copy with volatile host-timing fields fixed to a constant.
 
-    Two honest CPU re-execs of the SAME submission differ ONLY in these host-timing fields; fixing
-    them makes the canonical manifest bytes (and thus :func:`compute_manifest_sha256`) identical,
-    which is what lets two independent worker replicas agree and be accepted (rather than disputed).
+    Two honest CPU re-execs of the SAME submission differ in host timing: the
+    classical ``compute`` wall/RSS fields plus (after telemetry-rt) per-step
+    ``metrics.train_series.points[*].wall_s`` and the linked
+    ``train_series_sha256`` digests. Fix timing so canonical manifest bytes
+    (and thus :func:`compute_manifest_sha256`) match across replicas.
     """
 
+    from .train_series import train_series_sha256
+
     normalized = copy.deepcopy(manifest)
     compute = normalized.get("compute")
     if isinstance(compute, dict):
         for field in VOLATILE_COMPUTE_FIELDS:
             if field in compute:
                 compute[field] = 0
+
+    metrics = normalized.get("metrics")
+    series_digest: str | None = None
+    if isinstance(metrics, dict):
+        raw_series = metrics.get("train_series")
+        if isinstance(raw_series, dict):
+            points = raw_series.get("points")
+            if isinstance(points, list):
+                for point in points:
+                    if isinstance(point, dict) and "wall_s" in point:
+                        point["wall_s"] = 0.0
+            series_digest = train_series_sha256(raw_series)
+            if "train_series_sha256" in metrics:
+                metrics["train_series_sha256"] = series_digest
+            if "sha256" in raw_series:
+                raw_series["sha256"] = series_digest
+
+    artifacts = normalized.get("artifacts")
+    if isinstance(artifacts, dict) and "train_series_sha256" in artifacts:
+        artifacts["train_series_sha256"] = series_digest or ("0" * 64)
+
     return normalized
 
 

tests/test_cpu_test_mode.py

@@ -94,6 +94,33 @@ def test_normalize_drops_volatile_compute_fields():
     assert manifest["compute"]["wall_clock_seconds"] == 12.5
 
 
+def test_normalize_zeros_train_series_wall_and_realigns_digest():
+    """Telemetry wall_s must not keep two honest replicas from sharing one hash."""
+    series = {
+        "schema_version": "prism_train_series.v1",
+        "points": [
+            {"i": 0, "wall_s": 0.0123, "train_ce_nats": 1.0, "tokens_seen": 8},
+            {"i": 1, "wall_s": 0.0456, "train_ce_nats": 0.9, "tokens_seen": 16},
+        ],
+    }
+    manifest = {
+        "metrics": {
+            "train_series": series,
+            "train_series_sha256": "deadbeef",
+        },
+        "artifacts": {"train_series_sha256": "deadbeef"},
+    }
+    normalized = normalize_manifest_for_replication(manifest)
+    points = normalized["metrics"]["train_series"]["points"]
+    assert all(point["wall_s"] == 0.0 for point in points)
+    assert points[0]["train_ce_nats"] == 1.0  # non-timing preserved
+    digest = normalized["metrics"]["train_series_sha256"]
+    assert digest != "deadbeef" and len(digest) == 64
+    assert normalized["artifacts"]["train_series_sha256"] == digest
+    # Input not mutated.
+    assert series["points"][0]["wall_s"] == 0.0123
+
+
 def test_two_replicas_of_same_submission_agree_on_hash(tmp_path):
     # Two independent worker hosts (distinct artifact roots) re-exec the SAME submission; the
     # normalized manifest hash MUST match so the base worker plane accepts (not disputes) them.