Require policy pressure mirror consistency

docs/ko/README.md

@@ -170,7 +170,9 @@ timeline `policy_pressure` block이 있으면 EdgeEnv는
 `not_a_deployment_decision=true` marker를 검증하고
 `orchestrator_policy_pressure_summary_run_ids`로 traceability를 노출한다.
 이는 scheduler pressure review context이며 EdgeEnv regression gate가
-아니다.
+아니다. direct `policy_pressure_summary`와 timeline `policy_pressure`가 둘 다
+있으면 handoff 중 mirror drift가 생기지 않도록 두 block이 정확히 일치해야
+한다.
 또한 `lab_bundle_alignment.external_aiguard_required_evidence_types`에
 `runtime_history_seed_run_config_traceability`와
 `edgeenv_orchestrator_operation_risk_rollup`,

docs/runtime-telemetry-history.md

@@ -156,7 +156,9 @@ an operation timeline `policy_pressure` block, EdgeEnv validates
 `not_a_deployment_decision=true`. The inspect and handoff summaries report
 matching runs as `policy_pressure_summary_run_ids`; this is scheduler-pressure
 review context for Lab/AIGuard, not a comparability field or deployment
-decision.
+decision. When both the direct `policy_pressure_summary` and timeline
+`policy_pressure` blocks are present, they must match exactly so the mirror
+cannot drift during handoff.
 If the same operation context carries Orchestrator `stale_drop_summary` or an
 operation timeline `stale_drop` block, EdgeEnv validates
 `schema_version=inferedge-orchestrator-stale-drop-summary-v1`,

inferedge_env/result/lab_handoff.py

@@ -1276,16 +1276,42 @@ def _validate_orchestrator_policy_pressure_context(
     regression_path: Path,
 ) -> None:
     operation = _candidate_operation_context(operation_context)
+    policy_pressure_summary = operation.get("policy_pressure_summary")
     _validate_orchestrator_policy_pressure_summary(
-        operation.get("policy_pressure_summary"),
+        policy_pressure_summary,
         regression_path=regression_path,
     )
     timeline = operation.get("operation_timeline_summary")
+    timeline_policy_pressure = None
     if isinstance(timeline, dict):
+        timeline_policy_pressure = timeline.get("policy_pressure")
         _validate_orchestrator_policy_pressure_summary(
-            timeline.get("policy_pressure"),
+            timeline_policy_pressure,
             regression_path=regression_path,
         )
+    if (
+        isinstance(policy_pressure_summary, dict)
+        and timeline_policy_pressure is not None
+    ):
+        _validate_policy_pressure_mirror_match(
+            policy_pressure_summary,
+            timeline_policy_pressure,
+            regression_path=regression_path,
+        )
+
+
+def _validate_policy_pressure_mirror_match(
+    policy_pressure_summary: dict[str, Any],
+    timeline_policy_pressure: Any,
+    *,
+    regression_path: Path,
+) -> None:
+    if policy_pressure_summary != timeline_policy_pressure:
+        raise RuntimeIntelligenceLabHandoffError(
+            "policy_pressure_summary must match "
+            "operation_timeline_summary.policy_pressure when both are present: "
+            f"{regression_path}"
+        )
 
 
 def _validate_orchestrator_policy_pressure_summary(

inferedge_env/result/telemetry_history.py

@@ -1240,14 +1240,20 @@ def _validate_orchestrator_candidate_operation_context(
         operation.get("operation_risk_rollup"),
         source=source,
     )
-    _validate_orchestrator_operation_timeline_summary(
+    timeline_summary = _validate_orchestrator_operation_timeline_summary(
         operation.get("operation_timeline_summary"),
         source=source,
     )
-    _validate_orchestrator_policy_pressure_summary(
+    policy_pressure_summary = _validate_orchestrator_policy_pressure_summary(
         operation.get("policy_pressure_summary"),
         source=source,
     )
+    if timeline_summary and policy_pressure_summary:
+        _validate_policy_pressure_mirror_match(
+            policy_pressure_summary,
+            timeline_summary.get("policy_pressure"),
+            source=source,
+        )
 
 
 def _validate_orchestrator_operation_timeline_summary(
@@ -1381,6 +1387,22 @@ def _validate_orchestrator_policy_pressure_summary(
     return summary
 
 
+def _validate_policy_pressure_mirror_match(
+    policy_pressure_summary: dict[str, Any],
+    timeline_policy_pressure: Any,
+    *,
+    source: Path,
+) -> None:
+    if timeline_policy_pressure is None:
+        return
+    if policy_pressure_summary != timeline_policy_pressure:
+        raise RuntimeTelemetryHistoryError(
+            "Orchestrator telemetry feed policy_pressure_summary must match "
+            "operation_timeline_summary.policy_pressure when both are present: "
+            f"{source}"
+        )
+
+
 def _validate_orchestrator_stale_drop_summary(
     value: Any,
     *,

tests/test_runtime_intelligence_lab_handoff.py

@@ -1128,6 +1128,32 @@ def test_runtime_intelligence_lab_handoff_rejects_policy_pressure_as_decision(
         )
 
 
+def test_runtime_intelligence_lab_handoff_rejects_policy_pressure_mirror_drift(
+    tmp_path,
+):
+    baseline_path, candidate_path, regression_path, history_path = _write_handoff_files(
+        tmp_path
+    )
+    regression = json.loads(regression_path.read_text(encoding="utf-8"))
+    regression["runtime_telemetry_context"]["candidate"][
+        "orchestrator_operation_context"
+    ]["candidate_context"]["operation"]["operation_timeline_summary"][
+        "policy_pressure"
+    ]["decision_count"] = 3
+    regression_path.write_text(json.dumps(regression), encoding="utf-8")
+
+    with pytest.raises(
+        RuntimeIntelligenceLabHandoffError,
+        match="policy_pressure_summary must match",
+    ):
+        build_runtime_intelligence_lab_handoff_manifest(
+            baseline_result_path=baseline_path,
+            candidate_result_path=candidate_path,
+            edgeenv_regression_report_path=regression_path,
+            telemetry_history_path=history_path,
+        )
+
+
 def _write_handoff_files(tmp_path):
     baseline_path = tmp_path / "baseline-result.json"
     candidate_path = tmp_path / "candidate-result.json"

tests/test_runtime_telemetry_history.py

@@ -565,6 +565,35 @@ def test_runtime_telemetry_history_rejects_policy_pressure_as_decision(
         build_runtime_telemetry_history(edgeenv_root, orchestrator_feeds=[feed_path])
 
 
+def test_runtime_telemetry_history_rejects_policy_pressure_mirror_drift(
+    tmp_path,
+    bench_config,
+    target_profile,
+    config_files,
+):
+    edgeenv_root = tmp_path / ".edgeenv"
+    _write_registered_run(
+        edgeenv_root,
+        bench_config,
+        target_profile,
+        config_files,
+        run_id="candidate",
+        runtime_telemetry=_runtime_telemetry_payload(sequence_id=2),
+    )
+    feed = _orchestrator_feed_payload("candidate")
+    feed["candidate_context"]["operation"]["operation_timeline_summary"][
+        "policy_pressure"
+    ]["decision_count"] = 3
+    feed_path = tmp_path / "orchestrator-feed.json"
+    feed_path.write_text(json.dumps(feed), encoding="utf-8")
+
+    with pytest.raises(
+        RuntimeTelemetryHistoryError,
+        match="policy_pressure_summary must match",
+    ):
+        build_runtime_telemetry_history(edgeenv_root, orchestrator_feeds=[feed_path])
+
+
 def test_runtime_telemetry_history_rejects_operation_risk_summary_as_decision(
     tmp_path,
     bench_config,