Add policy pressure timeline summary

docs/operation_control.ko.md

@@ -32,13 +32,16 @@ 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 경계
 
 Orchestrator는 `edgeenv_runtime_telemetry_feed`를 통해 supplemental operation
 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를 제공할 수 있다.

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.

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: "

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
 
 

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] = {}