Mirror policy pressure in EdgeEnv feed

README.ko.md

@@ -165,6 +165,12 @@ flowchart LR
   candidate context 필수 필드를 검증한다. 또한 `aiguard_evidence_candidates`가
   `runtime_queue_overload`, `runtime_thermal_instability`를 유지하는지 확인해
   downstream diagnosis/report fixture의 deterministic anomaly 경계를 보존한다.
+- policy-pressure context는 operation timeline에서
+  `candidate_context.operation.policy_pressure_summary`로 mirror되며,
+  `role=supplemental`, `scheduler_owner=orchestrator`, `decision_owner=lab`,
+  `not_a_deployment_decision=true` marker를 함께 둔다. 이를 통해 EdgeEnv가
+  같은 scheduler-pressure evidence를 AIGuard/Lab handoff로 보존하더라도
+  Orchestrator가 decision owner가 되지 않는다.
 
 ## Implementation Map
 

README.md

@@ -168,6 +168,12 @@ The boundary is intentional:
   `aiguard_evidence_candidates` to keep `runtime_queue_overload` and
   `runtime_thermal_instability` so downstream diagnosis/report fixtures keep
   the same deterministic anomaly boundary.
+- Policy-pressure context is mirrored from the operation timeline into
+  `candidate_context.operation.policy_pressure_summary` with
+  `role=supplemental`, `scheduler_owner=orchestrator`, `decision_owner=lab`,
+  and `not_a_deployment_decision=true` so EdgeEnv can preserve the same
+  scheduler-pressure evidence for AIGuard/Lab handoff without making
+  Orchestrator a decision owner.
 
 ## Implementation Map
 

docs/agent_orchestration_summary_contract.ko.md

@@ -294,6 +294,12 @@ drop과 load-shedding drop을 stale/backlog evidence로 분류하고,
 이 summary는 AIGuard/Lab/Env가 raw timeline을 훑기 전에 보는 additive
 navigation aid이며 `drop_events`, `queue_depth_timeline`, `latency_timeline`,
 `policy_decision_log`를 대체하지 않고 deployment decision도 아닙니다.
+EdgeEnv handoff에서는 같은 policy-pressure payload를
+`candidate_context.operation.policy_pressure_summary`로도 mirror하고,
+`role=supplemental`, `scheduler_owner=orchestrator`, `decision_owner=lab`,
+`not_a_deployment_decision=true` marker를 함께 둡니다. 이를 통해 downstream
+AIGuard/Lab check가 scheduler-pressure context를 보존하더라도 decision으로
+취급하지 않게 합니다.
 또한 feed는
 `downstream_guard_alignment.producer_lineage_evidence_type=edgeenv_orchestrator_producer_lineage`
 를 포함하므로 AIGuard/Lab이 producer-lineage reasoning을 queue/thermal operation

docs/agent_orchestration_summary_contract.md

@@ -294,6 +294,11 @@ including `tasks_with_stale_drop`, reason counts, and `review_stale_drop` hints.
 This summary is an additive navigation aid for AIGuard/Lab/Env; it does not
 replace `drop_events`, `queue_depth_timeline`, `latency_timeline`, or
 `policy_decision_log`, and it is not a deployment decision.
+For EdgeEnv handoff, the same policy-pressure payload is also mirrored as
+`candidate_context.operation.policy_pressure_summary` with `role=supplemental`,
+`scheduler_owner=orchestrator`, `decision_owner=lab`, and
+`not_a_deployment_decision=true` so downstream AIGuard/Lab checks can preserve
+scheduler-pressure context without treating it as a decision.
 The feed also carries
 `downstream_guard_alignment.producer_lineage_evidence_type=edgeenv_orchestrator_producer_lineage`
 so AIGuard/Lab can validate producer-lineage reasoning separately from

scripts/check_edgeenv_runtime_feed_contract.py

@@ -56,6 +56,7 @@ def main(argv: list[str] | None = None) -> int:
     operation = candidate_context.get("operation") or {}
     latency_budget_protection = operation.get("latency_budget_protection") or {}
     operation_timeline_summary = operation.get("operation_timeline_summary") or {}
+    policy_pressure_summary = operation.get("policy_pressure_summary") or {}
     guard_alignment = feed.get("downstream_guard_alignment") or {}
     print("EdgeEnv runtime telemetry feed contract passed.")
     print(
@@ -111,7 +112,11 @@ def main(argv: list[str] | None = None) -> int:
         affected = operation_timeline_summary.get("affected_tasks") or {}
         review_hints = operation_timeline_summary.get("review_hints") or []
         stale_drop = operation_timeline_summary.get("stale_drop") or {}
-        policy_pressure = operation_timeline_summary.get("policy_pressure") or {}
+        policy_pressure = (
+            policy_pressure_summary
+            or operation_timeline_summary.get("policy_pressure")
+            or {}
+        )
         scheduler_fairness = operation_timeline_summary.get(
             "scheduler_fairness"
         ) or {}

src/inferedge_orchestrator/sustained.py

@@ -338,6 +338,9 @@ def _edgeenv_runtime_telemetry_feed(
         ),
         "operation_timeline_summary": _operation_timeline_summary(report, config),
     }
+    operation["policy_pressure_summary"] = operation[
+        "operation_timeline_summary"
+    ]["policy_pressure"]
     operation["latency_budget_protection"] = _latency_budget_protection_context(
         config,
         report,
@@ -495,8 +498,18 @@ def validate_edgeenv_runtime_telemetry_feed(
             raise ValueError(
                 "edgeenv_runtime_telemetry_feed.candidate_context.operation."
                 "operation_timeline_summary must be an object"
-            )
+        )
         _validate_operation_timeline_summary(operation_timeline_summary)
+    policy_pressure_summary = candidate_context["operation"].get(
+        "policy_pressure_summary"
+    )
+    if policy_pressure_summary is not None:
+        if not isinstance(policy_pressure_summary, dict):
+            raise ValueError(
+                "edgeenv_runtime_telemetry_feed.candidate_context.operation."
+                "policy_pressure_summary must be an object"
+            )
+        _validate_policy_pressure_summary(policy_pressure_summary)
     stale_drop_summary = candidate_context["operation"].get("stale_drop_summary")
     if stale_drop_summary is not None:
         if not isinstance(stale_drop_summary, dict):
@@ -750,6 +763,11 @@ def _validate_policy_pressure_summary(payload: dict[str, Any]) -> None:
             "policy_pressure_summary.schema_version must be "
             f"{POLICY_PRESSURE_SUMMARY_SCHEMA}"
         )
+    if payload.get("role") != "supplemental":
+        raise ValueError(
+            "edgeenv_runtime_telemetry_feed.candidate_context.operation."
+            "policy_pressure_summary.role must be supplemental"
+        )
     if payload.get("operation_context_role") != "supplemental":
         raise ValueError(
             "edgeenv_runtime_telemetry_feed.candidate_context.operation."
@@ -1619,6 +1637,7 @@ def _policy_pressure_summary(report: dict[str, Any]) -> dict[str, Any]:
     )
     return {
         "schema_version": POLICY_PRESSURE_SUMMARY_SCHEMA,
+        "role": "supplemental",
         "operation_context_role": "supplemental",
         "scheduler_owner": "orchestrator",
         "decision_owner": "lab",

tests/test_multi_workload_sustained.py

@@ -136,6 +136,7 @@ def test_run_multi_workload_sustained_writes_profile_summary(tmp_path) -> None:
     assert timeline["policy"]["first_decision"]["queue_depth_snapshot"]
     policy_pressure = timeline["policy_pressure"]
     assert policy_pressure["schema_version"] == POLICY_PRESSURE_SUMMARY_SCHEMA
+    assert policy_pressure["role"] == "supplemental"
     assert policy_pressure["operation_context_role"] == "supplemental"
     assert policy_pressure["scheduler_owner"] == "orchestrator"
     assert policy_pressure["decision_owner"] == "lab"
@@ -293,6 +294,7 @@ def test_run_multi_workload_sustained_writes_profile_summary(tmp_path) -> None:
         candidate["operation"]["operation_timeline_summary"]["policy_pressure"]
         == policy_pressure
     )
+    assert candidate["operation"]["policy_pressure_summary"] == policy_pressure
     assert candidate["operation"]["stale_drop_summary"] == stale_drop
     assert candidate["operation"]["operation_risk_rollup"] == risk_rollup
     assert candidate["operation"]["scheduler_fairness_summary"] == fairness
@@ -647,8 +649,8 @@ def test_write_edgeenv_runtime_telemetry_feed_requires_policy_pressure_schema(
         frames=4,
     )
     report["edgeenv_runtime_telemetry_feed"]["candidate_context"]["operation"][
-        "operation_timeline_summary"
-    ]["policy_pressure"]["decision_owner"] = "orchestrator"
+        "policy_pressure_summary"
+    ]["decision_owner"] = "orchestrator"
 
     with pytest.raises(
         ValueError,