From 7d86bff7e2f37b15f288f0e5175d973e3af60158 Mon Sep 17 00:00:00 2001 From: hyeokjun32 Date: Thu, 18 Jun 2026 21:33:02 +0900 Subject: [PATCH] Add policy pressure timeline summary --- docs/operation_control.ko.md | 3 + docs/operation_control.md | 3 + .../check_edgeenv_runtime_feed_contract.py | 14 ++ src/inferedge_orchestrator/cli.py | 12 ++ src/inferedge_orchestrator/sustained.py | 183 ++++++++++++++++++ tests/test_multi_workload_sustained.py | 72 +++++++ tests/test_scripts.py | 4 + 7 files changed, 291 insertions(+) diff --git a/docs/operation_control.ko.md b/docs/operation_control.ko.md index 1e67766..9c73ea8 100644 --- a/docs/operation_control.ko.md +++ b/docs/operation_control.ko.md @@ -32,6 +32,7 @@ validated workload | Fallback evidence | fallback decision count, fallback final status, bounded recovery marker | | Worker health | worker status, runtime event summary, resource pressure context | | Telemetry evidence | JSON operation summary, compact event rollup, policy decision reason | +| Policy pressure | 제한/보호된 task 요약, fallback policy 사용, backlog threshold 초과 marker | ## Handoff 경계 @@ -39,6 +40,8 @@ Orchestrator는 `edgeenv_runtime_telemetry_feed`를 통해 supplemental operatio context를 export할 수 있다. 이 feed는 EdgeEnv, AIGuard, Lab이 queue/deadline/fallback/resource context를 표시하는 데 도움을 주지만 ownership을 바꾸지 않는다. +operation timeline은 policy pressure summary를 포함할 수 있으며, downstream +report가 backlog 압력에서 어떤 task가 제한되거나 보호됐는지 표시하는 데 쓴다. - EdgeEnv는 registry, comparability, runtime regression evidence owner로 남는다. - AIGuard는 deterministic runtime warning evidence를 제공할 수 있다. diff --git a/docs/operation_control.md b/docs/operation_control.md index e3c3571..5b03229 100644 --- a/docs/operation_control.md +++ b/docs/operation_control.md @@ -32,12 +32,15 @@ validated workload | Fallback evidence | fallback decision count, fallback final status, bounded recovery marker | | Worker health | worker status, runtime event summary, resource pressure context | | Telemetry evidence | JSON operation summary, compact event rollup, policy decision reason | +| Policy pressure | limited/protected task summary, fallback policy use, backlog-over-threshold markers | ## Handoff Boundaries Orchestrator can export supplemental operation context through `edgeenv_runtime_telemetry_feed`. That feed can help EdgeEnv, AIGuard, and Lab show queue/deadline/fallback/resource context without changing ownership. +The operation timeline may include a policy pressure summary so downstream +reports can show which tasks were limited or protected under backlog pressure. - EdgeEnv remains the registry, comparability, and runtime regression evidence owner. diff --git a/scripts/check_edgeenv_runtime_feed_contract.py b/scripts/check_edgeenv_runtime_feed_contract.py index 3a2004d..3df3169 100644 --- a/scripts/check_edgeenv_runtime_feed_contract.py +++ b/scripts/check_edgeenv_runtime_feed_contract.py @@ -111,6 +111,7 @@ 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 {} scheduler_fairness = operation_timeline_summary.get( "scheduler_fairness" ) or {} @@ -124,6 +125,19 @@ def main(argv: list[str] | None = None) -> int: f"stale_drop_tasks={_format_list(affected.get('stale_drop'))}; " f"max_queue_wait_ms={latency.get('max_queue_wait_ms', 0)}" ) + if policy_pressure: + print( + "policy_pressure: " + f"decisions={policy_pressure.get('decision_count', 0)}; " + "limited=" + f"{_format_list(policy_pressure.get('limited_tasks'))}; " + "protected=" + f"{_format_list(policy_pressure.get('protected_tasks'))}; " + "fallback=" + f"{_format_list(policy_pressure.get('fallback_tasks'))}; " + "markers=" + f"{_format_list(policy_pressure.get('pressure_markers'))}" + ) if scheduler_fairness: print( "scheduler_fairness: " diff --git a/src/inferedge_orchestrator/cli.py b/src/inferedge_orchestrator/cli.py index 5c0bce5..6ea19a8 100644 --- a/src/inferedge_orchestrator/cli.py +++ b/src/inferedge_orchestrator/cli.py @@ -214,6 +214,9 @@ def _run_multi_workload_sustained(args: argparse.Namespace) -> int: affected_tasks = timeline.get("affected_tasks", {}) if isinstance(timeline, dict) else {} latency = timeline.get("latency", {}) if isinstance(timeline, dict) else {} stale_drop = timeline.get("stale_drop", {}) if isinstance(timeline, dict) else {} + policy_pressure = ( + timeline.get("policy_pressure", {}) if isinstance(timeline, dict) else {} + ) review_hints = timeline.get("review_hints", []) if isinstance(timeline, dict) else [] print(f"wrote sustained telemetry: {args.output}") if args.edgeenv_feed_output: @@ -238,6 +241,15 @@ def _run_multi_workload_sustained(args: argparse.Namespace) -> int: f"stale_drop_tasks={_format_cli_list(affected_tasks.get('stale_drop'))} " f"max_queue_wait_ms={latency.get('max_queue_wait_ms', 0)}" ) + if policy_pressure: + print( + "policy-pressure: " + f"decisions={policy_pressure.get('decision_count', 0)} " + f"limited={_format_cli_list(policy_pressure.get('limited_tasks'))} " + f"protected={_format_cli_list(policy_pressure.get('protected_tasks'))} " + f"fallback={_format_cli_list(policy_pressure.get('fallback_tasks'))} " + f"markers={_format_cli_list(policy_pressure.get('pressure_markers'))}" + ) return 0 diff --git a/src/inferedge_orchestrator/sustained.py b/src/inferedge_orchestrator/sustained.py index fc59e14..709f98a 100644 --- a/src/inferedge_orchestrator/sustained.py +++ b/src/inferedge_orchestrator/sustained.py @@ -46,6 +46,9 @@ SCHEDULER_FAIRNESS_SUMMARY_SCHEMA = ( "inferedge-orchestrator-scheduler-fairness-summary-v1" ) +POLICY_PRESSURE_SUMMARY_SCHEMA = ( + "inferedge-orchestrator-policy-pressure-summary-v1" +) def apply_device_local_input_overrides( @@ -722,6 +725,14 @@ def _validate_operation_timeline_summary(payload: dict[str, Any]) -> None: "operation_timeline_summary.stale_drop must be an object" ) _validate_stale_drop_summary(stale_drop) + policy_pressure = payload.get("policy_pressure") + if policy_pressure is not None: + if not isinstance(policy_pressure, dict): + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "operation_timeline_summary.policy_pressure must be an object" + ) + _validate_policy_pressure_summary(policy_pressure) scheduler_fairness = payload.get("scheduler_fairness") if scheduler_fairness is not None: if not isinstance(scheduler_fairness, dict): @@ -732,6 +743,76 @@ def _validate_operation_timeline_summary(payload: dict[str, Any]) -> None: _validate_scheduler_fairness_summary(scheduler_fairness) +def _validate_policy_pressure_summary(payload: dict[str, Any]) -> None: + if payload.get("schema_version") != POLICY_PRESSURE_SUMMARY_SCHEMA: + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "policy_pressure_summary.schema_version must be " + f"{POLICY_PRESSURE_SUMMARY_SCHEMA}" + ) + if payload.get("operation_context_role") != "supplemental": + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "policy_pressure_summary.operation_context_role must be supplemental" + ) + if payload.get("scheduler_owner") != "orchestrator": + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "policy_pressure_summary.scheduler_owner must be orchestrator" + ) + if payload.get("decision_owner") != "lab": + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "policy_pressure_summary.decision_owner must be lab" + ) + if payload.get("not_a_deployment_decision") is not True: + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "policy_pressure_summary.not_a_deployment_decision must be true" + ) + for field in ( + "decision_count", + "fallback_decision_count", + "max_total_backlog_before", + "max_backlog_over_threshold", + ): + value = payload.get(field) + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + f"policy_pressure_summary.{field} must be a non-negative integer" + ) + if not isinstance(payload.get("decision_reason_counts"), dict): + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "policy_pressure_summary.decision_reason_counts must be an object" + ) + for field in ( + "limited_tasks", + "protected_tasks", + "fallback_tasks", + "pressure_markers", + ): + value = payload.get(field) + if not isinstance(value, list) or not all( + isinstance(item, str) and item for item in value + ): + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + f"policy_pressure_summary.{field} must be a string list" + ) + thresholds = payload.get("backlog_thresholds") + if not isinstance(thresholds, list) or not all( + isinstance(item, int) and not isinstance(item, bool) and item >= 0 + for item in thresholds + ): + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "policy_pressure_summary.backlog_thresholds must be a " + "non-negative integer list" + ) + + def _validate_stale_drop_summary(payload: dict[str, Any]) -> None: if payload.get("schema_version") != STALE_DROP_SUMMARY_SCHEMA: raise ValueError( @@ -1418,6 +1499,7 @@ def _operation_timeline_summary( }, "latency": _latency_timeline_summary(report), "policy": _policy_timeline_summary(report), + "policy_pressure": _policy_pressure_summary(report), "stale_drop": stale_drop, "scheduler_fairness": _scheduler_fairness_summary(config, report), "affected_tasks": { @@ -1488,6 +1570,107 @@ def _policy_timeline_summary(report: dict[str, Any]) -> dict[str, Any]: } +def _policy_pressure_summary(report: dict[str, Any]) -> dict[str, Any]: + decisions = _dict_list(report.get("policy_decision_log")) + runtime_event_summary = _dict_value(report.get("runtime_event_summary")) + limited_tasks: list[str] = [] + protected_tasks: list[str] = [] + fallback_tasks: list[str] = [] + reason_counts: dict[str, int] = {} + max_total_backlog_before = 0 + max_backlog_over_threshold = 0 + backlog_thresholds: list[int] = [] + + for decision in decisions: + limited_task = decision.get("limited_task") + if isinstance(limited_task, str) and limited_task: + if limited_task not in limited_tasks: + limited_tasks.append(limited_task) + if ( + bool(decision.get("fallback_used")) + and limited_task not in fallback_tasks + ): + fallback_tasks.append(limited_task) + protected_task = decision.get("protected_task") + if isinstance(protected_task, str) and protected_task not in protected_tasks: + protected_tasks.append(protected_task) + reason = decision.get("decision_reason") or decision.get("reason") + if isinstance(reason, str) and reason: + reason_counts[reason] = reason_counts.get(reason, 0) + 1 + total_backlog = _non_negative_int_value( + decision.get("total_backlog_before") + ) + threshold = _non_negative_int_value(decision.get("backlog_threshold")) + max_total_backlog_before = max(max_total_backlog_before, total_backlog) + if threshold: + if threshold not in backlog_thresholds: + backlog_thresholds.append(threshold) + max_backlog_over_threshold = max( + max_backlog_over_threshold, + max(total_backlog - threshold, 0), + ) + + pressure_markers = _policy_pressure_markers( + decision_count=len(decisions), + fallback_tasks=fallback_tasks, + limited_tasks=limited_tasks, + max_backlog_over_threshold=max_backlog_over_threshold, + runtime_event_summary=runtime_event_summary, + ) + return { + "schema_version": POLICY_PRESSURE_SUMMARY_SCHEMA, + "operation_context_role": "supplemental", + "scheduler_owner": "orchestrator", + "decision_owner": "lab", + "not_a_deployment_decision": True, + "source": "policy_decision_log+runtime_event_summary", + "first_read": ( + "review_policy_pressure_context" + if pressure_markers + else "policy_pressure_nominal" + ), + "decision_count": len(decisions), + "decision_reason_counts": reason_counts, + "limited_tasks": limited_tasks, + "protected_tasks": protected_tasks, + "fallback_tasks": fallback_tasks, + "fallback_decision_count": _non_negative_int_value( + runtime_event_summary.get("fallback_decision_count") + ), + "backlog_thresholds": backlog_thresholds, + "max_total_backlog_before": max_total_backlog_before, + "max_backlog_over_threshold": max_backlog_over_threshold, + "pressure_markers": pressure_markers, + "interpretation": ( + "Policy pressure is supplemental evidence showing which scheduler " + "decisions limited or protected work under backlog pressure; Lab " + "remains the final deployment decision owner." + ), + } + + +def _policy_pressure_markers( + *, + decision_count: int, + fallback_tasks: list[str], + limited_tasks: list[str], + max_backlog_over_threshold: int, + runtime_event_summary: dict[str, Any], +) -> list[str]: + markers: list[str] = [] + if decision_count: + markers.append("policy_decision_present") + if max_backlog_over_threshold > 0: + markers.append("backlog_exceeded_threshold") + if fallback_tasks: + markers.append("fallback_policy_used") + if limited_tasks: + markers.append("workload_limited_by_policy") + if _positive_int(runtime_event_summary.get("scheduler_delay_event_count")): + markers.append("scheduler_delay_present") + return markers + + def _stale_drop_summary(report: dict[str, Any]) -> dict[str, Any]: events = _dict_list(report.get("drop_events")) stale_reason_counts: dict[str, int] = {} diff --git a/tests/test_multi_workload_sustained.py b/tests/test_multi_workload_sustained.py index fe5cfec..4efa7f5 100644 --- a/tests/test_multi_workload_sustained.py +++ b/tests/test_multi_workload_sustained.py @@ -21,6 +21,7 @@ LATENCY_BUDGET_PROTECTION_SCHEMA, MULTI_WORKLOAD_SCHEMA, OPERATION_TIMELINE_SUMMARY_SCHEMA, + POLICY_PRESSURE_SUMMARY_SCHEMA, SCHEDULER_FAIRNESS_SUMMARY_SCHEMA, STALE_DROP_SUMMARY_SCHEMA, apply_device_local_input_overrides, @@ -133,6 +134,43 @@ def test_run_multi_workload_sustained_writes_profile_summary(tmp_path) -> None: "queue_backlog_threshold_exceeded" ) 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["operation_context_role"] == "supplemental" + assert policy_pressure["scheduler_owner"] == "orchestrator" + assert policy_pressure["decision_owner"] == "lab" + assert policy_pressure["not_a_deployment_decision"] is True + assert policy_pressure["first_read"] == "review_policy_pressure_context" + assert policy_pressure["decision_count"] == len(report["policy_decision_log"]) + assert policy_pressure["decision_reason_counts"] == { + "queue_backlog_threshold_exceeded": len(report["policy_decision_log"]) + } + assert policy_pressure["limited_tasks"] == [ + "voice_command_agent", + "vision_agent", + ] + assert policy_pressure["protected_tasks"] == ["safety_monitor_agent"] + assert policy_pressure["fallback_tasks"] == [ + "voice_command_agent", + "vision_agent", + ] + assert policy_pressure["fallback_decision_count"] == ( + report["runtime_event_summary"]["fallback_decision_count"] + ) + assert policy_pressure["max_total_backlog_before"] >= ( + policy_pressure["backlog_thresholds"][0] + ) + assert policy_pressure["max_backlog_over_threshold"] > 0 + assert set(policy_pressure["pressure_markers"]) >= { + "policy_decision_present", + "backlog_exceeded_threshold", + "fallback_policy_used", + "workload_limited_by_policy", + "scheduler_delay_present", + } + assert "Lab remains the final deployment decision owner" in ( + policy_pressure["interpretation"] + ) stale_drop = timeline["stale_drop"] assert stale_drop["schema_version"] == STALE_DROP_SUMMARY_SCHEMA assert stale_drop["operation_context_role"] == "supplemental" @@ -251,6 +289,10 @@ def test_run_multi_workload_sustained_writes_profile_summary(tmp_path) -> None: report["runtime_event_summary"]["tasks_with_fallback"] ) assert candidate["operation"]["operation_timeline_summary"] == timeline + assert ( + candidate["operation"]["operation_timeline_summary"]["policy_pressure"] + == 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 @@ -589,6 +631,32 @@ def test_write_edgeenv_runtime_telemetry_feed_requires_stale_drop_schema( write_edgeenv_runtime_telemetry_feed(report, tmp_path / "feed.json") +def test_write_edgeenv_runtime_telemetry_feed_requires_policy_pressure_schema( + tmp_path, +) -> None: + config = OrchestratorConfig.from_dict( + json.loads( + Path("configs/agent_multi_workload_sustained_local.json").read_text( + encoding="utf-8" + ) + ) + ) + report = write_multi_workload_sustained( + config, + output=tmp_path / "report.json", + frames=4, + ) + report["edgeenv_runtime_telemetry_feed"]["candidate_context"]["operation"][ + "operation_timeline_summary" + ]["policy_pressure"]["decision_owner"] = "orchestrator" + + with pytest.raises( + ValueError, + match="policy_pressure_summary.decision_owner must be lab", + ): + write_edgeenv_runtime_telemetry_feed(report, tmp_path / "feed.json") + + def test_validate_edgeenv_runtime_telemetry_feed_requires_device_local_producer( tmp_path, ) -> None: @@ -741,6 +809,10 @@ def test_cli_run_multi_workload_sustained_writes_edgeenv_feed_output( assert "stale_drop=" in captured.out assert "stale_drop_tasks=" in captured.out assert "max_queue_wait_ms=" in captured.out + assert "policy-pressure: decisions=" in captured.out + assert "limited=" in captured.out + assert "protected=" in captured.out + assert "markers=" in captured.out report = json.loads(output.read_text(encoding="utf-8")) feed = json.loads(feed_output.read_text(encoding="utf-8")) assert feed == report["edgeenv_runtime_telemetry_feed"] diff --git a/tests/test_scripts.py b/tests/test_scripts.py index 12d755d..43b5afa 100644 --- a/tests/test_scripts.py +++ b/tests/test_scripts.py @@ -92,6 +92,10 @@ def test_edgeenv_runtime_feed_contract_checker_passes_device_local_feed( assert "stale_drop=" in out assert "stale_drop_tasks=" in out assert "max_queue_wait_ms=" in out + assert "policy_pressure:" in out + assert "decisions=" in out + assert "limited=" in out + assert "markers=" in out assert "scheduler_fairness:" in out assert "starvation_risk=" in out assert "degraded=" in out