diff --git a/README.ko.md b/README.ko.md index 6c0d94d..dbfaf35 100644 --- a/README.ko.md +++ b/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 diff --git a/README.md b/README.md index 62e3f8c..027454f 100644 --- a/README.md +++ b/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 diff --git a/docs/agent_orchestration_summary_contract.ko.md b/docs/agent_orchestration_summary_contract.ko.md index 3295af3..e65fbbe 100644 --- a/docs/agent_orchestration_summary_contract.ko.md +++ b/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 diff --git a/docs/agent_orchestration_summary_contract.md b/docs/agent_orchestration_summary_contract.md index 6b103db..ba8933b 100644 --- a/docs/agent_orchestration_summary_contract.md +++ b/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 diff --git a/scripts/check_edgeenv_runtime_feed_contract.py b/scripts/check_edgeenv_runtime_feed_contract.py index 3df3169..2539dc3 100644 --- a/scripts/check_edgeenv_runtime_feed_contract.py +++ b/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 {} diff --git a/src/inferedge_orchestrator/sustained.py b/src/inferedge_orchestrator/sustained.py index 709f98a..9fe2167 100644 --- a/src/inferedge_orchestrator/sustained.py +++ b/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", diff --git a/tests/test_multi_workload_sustained.py b/tests/test_multi_workload_sustained.py index 4efa7f5..3ba2fbe 100644 --- a/tests/test_multi_workload_sustained.py +++ b/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,