Preserve Orchestrator pressure window handoff

README.md

@@ -377,6 +377,15 @@ matching runs as `worker_health_trend_run_ids` in inspect output and
 health trend is preserved as scheduler-owned operation context for Lab/AIGuard
 review; it is not a comparability field, regression threshold, or deployment
 decision.
+If the operation timeline carries `pressure_window`, EdgeEnv validates
+`schema_version=inferedge-orchestrator-pressure-window-summary-v1`,
+`operation_context_role=supplemental`, `scheduler_owner=orchestrator`,
+`decision_owner=lab`, and `not_a_deployment_decision=true`, then reports
+matching runs as `pressure_window_summary_run_ids` in inspect output and
+`orchestrator_pressure_window_summary_run_ids` in the Lab handoff summary. The
+pressure-window block is preserved as reviewer navigation for sustained
+overload intervals; it does not become an EdgeEnv regression gate or a
+deployment decision.
 Use `edgeenv runs telemetry inspect-history <path>` to validate and summarize
 that replay artifact before attaching it to a regression report. Add
 `--require-device-local-producer` when the handoff must prove that preserved
@@ -475,9 +484,10 @@ AIGuard `scheduler_delay_pattern`, and `remote_fallback_recovery_sample.json`
 can map to `remote_execution_recovered_by_fallback` before Lab renders the
 corresponding report markers.
 The alignment block separately declares optional AIGuard evidence types
-`stale_frame_risk` and `edgeenv_orchestrator_stale_drop_summary` for newer
-sustained Orchestrator stale-drop context. They are optional so EdgeEnv can
-preserve stale-drop evidence when present without rejecting older queue/thermal
+`stale_frame_risk`, `edgeenv_orchestrator_stale_drop_summary`, and
+`edgeenv_orchestrator_pressure_window_summary` for newer sustained
+Orchestrator context. They are optional so EdgeEnv can preserve stale-drop or
+pressure-window evidence when present without rejecting older queue/thermal
 feeds or changing Lab's required Runtime Intelligence bundle set.
 `lab_bundle_alignment.optional_aiguard_source_traceability` mirrors the
 AIGuard optional-present source artifact and regeneration command as read-only

docs/ko/README.md

@@ -176,6 +176,15 @@ timeline `policy_pressure` block이 있으면 EdgeEnv는
 아니다. direct `policy_pressure_summary`와 timeline `policy_pressure`가 둘 다
 있으면 handoff 중 mirror drift가 생기지 않도록 두 block이 정확히 일치해야
 한다.
+보존된 operation timeline에 Orchestrator `pressure_window` block이 있으면
+EdgeEnv는
+`schema_version=inferedge-orchestrator-pressure-window-summary-v1`,
+`operation_context_role=supplemental`, `scheduler_owner=orchestrator`,
+`decision_owner=lab`, `not_a_deployment_decision=true` marker를 검증하고
+inspect에서는 `pressure_window_summary_run_ids`, Lab handoff에서는
+`orchestrator_pressure_window_summary_run_ids`로 traceability를 노출한다.
+이는 sustained overload window를 reviewer가 빠르게 찾기 위한 navigation
+context이며 EdgeEnv regression gate나 deployment decision이 아니다.
 또한 `lab_bundle_alignment.external_aiguard_required_evidence_types`에
 `runtime_history_seed_run_config_traceability`와
 `edgeenv_orchestrator_operation_risk_rollup`,
@@ -192,9 +201,10 @@ block은 이 선언이 AIGuard `check-edgeenv-handoff-alignment`와 Lab Runtime
 Intelligence bundle manifest gate에서 검증된다는 점도 기록한다.
 별도의 `lab_bundle_alignment.optional_aiguard_evidence_types`는 최신
 sustained Orchestrator stale-drop context에서 AIGuard가 만들 수 있는
-`stale_frame_risk`와 `edgeenv_orchestrator_stale_drop_summary`를 선언한다.
-이 둘은 optional이므로 기존 queue/thermal feed나 Lab required bundle set을
-깨뜨리지 않는다.
+`stale_frame_risk`, `edgeenv_orchestrator_stale_drop_summary`,
+`edgeenv_orchestrator_pressure_window_summary`를 선언한다. 이 항목들은
+optional이므로 기존 queue/thermal feed나 Lab required bundle set을 깨뜨리지
+않는다.
 `lab_bundle_alignment.optional_aiguard_source_traceability`는 AIGuard
 optional-present source artifact와 재생성 명령을 read-only metadata로
 mirror한다:

docs/runtime-telemetry-history.md

@@ -179,6 +179,16 @@ them as `orchestrator_worker_health_trend_run_ids`. This preserves
 worker-health state counts, per-task health context, degraded/constrained
 workers, and review hints as supplemental operation evidence only; EdgeEnv does
 not turn worker health into a regression threshold or deployment decision.
+If the same operation timeline carries Orchestrator `pressure_window`, EdgeEnv
+validates
+`schema_version=inferedge-orchestrator-pressure-window-summary-v1`,
+`operation_context_role=supplemental`, `scheduler_owner=orchestrator`,
+`decision_owner=lab`, and `not_a_deployment_decision=true`. The inspect summary
+reports matching runs as `pressure_window_summary_run_ids`, and Lab handoff
+mirrors them as `orchestrator_pressure_window_summary_run_ids`. This preserves
+the overload-threshold window count, longest window, peak queue depth,
+limited/protected/fallback tasks, and `review_sustained_pressure_window`
+first-read marker as reviewer navigation context only.
 For device-local handoff smokes, `inspect-history` can enforce that lineage with
 `--require-device-local-producer`. The stricter check fails when the preserved
 history artifact has no Orchestrator context, or when preserved

inferedge_env/result/lab_handoff.py

@@ -12,6 +12,7 @@
     ORCHESTRATOR_EDGEENV_OPERATION_CONTEXT_ROLE,
     ORCHESTRATOR_EDGEENV_REQUIRED_CANDIDATE_FIELDS,
     ORCHESTRATOR_POLICY_PRESSURE_SUMMARY_SCHEMA_VERSION,
+    ORCHESTRATOR_PRESSURE_WINDOW_SUMMARY_SCHEMA_VERSION,
     ORCHESTRATOR_PRODUCER_LINEAGE_AIGUARD_EVIDENCE_TYPE,
     ORCHESTRATOR_STALE_DROP_SUMMARY_SCHEMA_VERSION,
     ORCHESTRATOR_TELEMETRY_FEED_ARTIFACT_ROLE,
@@ -89,6 +90,7 @@
 LAB_BUNDLE_OPTIONAL_AIGUARD_EVIDENCE_TYPES = (
     "stale_frame_risk",
     "edgeenv_orchestrator_stale_drop_summary",
+    "edgeenv_orchestrator_pressure_window_summary",
 )
 LAB_BUNDLE_OPTIONAL_AIGUARD_SOURCE_TRACEABILITY_CONTEXT_ROLE = (
     "read_only_optional_source_traceability"
@@ -417,6 +419,10 @@ def _validate_orchestrator_context(
         operation_context,
         regression_path=regression_path,
     )
+    _validate_orchestrator_pressure_window_context(
+        operation_context,
+        regression_path=regression_path,
+    )
 
 
 def _validate_orchestrator_producer_markers(
@@ -691,6 +697,7 @@ def _edgeenv_report_summary(regression_report: dict[str, Any]) -> dict[str, Any]
     policy_pressure_summary_run_ids = _policy_pressure_summary_run_ids(context)
     stale_drop_summary_run_ids = _stale_drop_summary_run_ids(context)
     worker_health_trend_run_ids = _worker_health_trend_run_ids(context)
+    pressure_window_summary_run_ids = _pressure_window_summary_run_ids(context)
     fixture_matrix_summary = _fixture_matrix_summary(
         regression_report.get("fixture_matrix_context")
     )
@@ -758,6 +765,12 @@ def _edgeenv_report_summary(regression_report: dict[str, Any]) -> dict[str, Any]
             worker_health_trend_run_ids
         ),
         "orchestrator_worker_health_trend_run_ids": worker_health_trend_run_ids,
+        "orchestrator_pressure_window_summary_present": bool(
+            pressure_window_summary_run_ids
+        ),
+        "orchestrator_pressure_window_summary_run_ids": (
+            pressure_window_summary_run_ids
+        ),
         "duration_traceability_present": bool(
             duration_traceability["run_ids"]
         ),
@@ -1132,6 +1145,31 @@ def append_if_present(run_context: Any) -> None:
     return run_ids
 
 
+def _pressure_window_summary_run_ids(context: Any) -> list[str]:
+    if not isinstance(context, dict):
+        return []
+    run_ids: list[str] = []
+
+    def append_if_present(run_context: Any) -> None:
+        if not isinstance(run_context, dict):
+            return
+        operation_context = run_context.get("orchestrator_operation_context")
+        if not _has_pressure_window_summary(operation_context):
+            return
+        run_id = run_context.get("run_id")
+        if isinstance(run_id, str) and run_id and run_id not in run_ids:
+            run_ids.append(run_id)
+
+    append_if_present(context.get("baseline"))
+    append_if_present(context.get("candidate"))
+    history = context.get("history")
+    if isinstance(history, dict):
+        for section in ("runs", "missing_telemetry"):
+            for entry in history.get(section, []):
+                append_if_present(entry)
+    return run_ids
+
+
 def _has_task_event_rollup(operation_context: Any) -> bool:
     if not isinstance(operation_context, dict):
         return False
@@ -1204,6 +1242,17 @@ def _has_worker_health_trend(operation_context: Any) -> bool:
     )
 
 
+def _has_pressure_window_summary(operation_context: Any) -> bool:
+    if not isinstance(operation_context, dict):
+        return False
+    operation = _candidate_operation_context(operation_context)
+    timeline = operation.get("operation_timeline_summary")
+    return (
+        isinstance(timeline, dict)
+        and isinstance(timeline.get("pressure_window"), dict)
+    )
+
+
 def _candidate_operation_context(operation_context: dict[str, Any]) -> dict[str, Any]:
     candidate_context = operation_context.get("candidate_context")
     if not isinstance(candidate_context, dict):
@@ -1402,6 +1451,20 @@ def _validate_orchestrator_worker_health_trend_context(
         )
 
 
+def _validate_orchestrator_pressure_window_context(
+    operation_context: dict[str, Any],
+    *,
+    regression_path: Path,
+) -> None:
+    operation = _candidate_operation_context(operation_context)
+    timeline = operation.get("operation_timeline_summary")
+    if isinstance(timeline, dict):
+        _validate_orchestrator_pressure_window_summary(
+            timeline.get("pressure_window"),
+            regression_path=regression_path,
+        )
+
+
 def _validate_policy_pressure_mirror_match(
     policy_pressure_summary: dict[str, Any],
     timeline_policy_pressure: Any,
@@ -1578,6 +1641,81 @@ def _validate_orchestrator_worker_health_trend(
             )
 
 
+def _validate_orchestrator_pressure_window_summary(
+    value: Any,
+    *,
+    regression_path: Path,
+) -> None:
+    if value is None:
+        return
+    if not isinstance(value, dict):
+        raise RuntimeIntelligenceLabHandoffError(
+            f"pressure_window must be an object: {regression_path}"
+        )
+    expected_pairs = {
+        "schema_version": ORCHESTRATOR_PRESSURE_WINDOW_SUMMARY_SCHEMA_VERSION,
+        "operation_context_role": ORCHESTRATOR_EDGEENV_OPERATION_CONTEXT_ROLE,
+        "scheduler_owner": "orchestrator",
+        "decision_owner": "lab",
+    }
+    for key, expected in expected_pairs.items():
+        if value.get(key) != expected:
+            raise RuntimeIntelligenceLabHandoffError(
+                f"pressure_window.{key} must be {expected}: {regression_path}"
+            )
+    if value.get("not_a_deployment_decision") is not True:
+        raise RuntimeIntelligenceLabHandoffError(
+            f"pressure_window.not_a_deployment_decision must be true: "
+            f"{regression_path}"
+        )
+    for field in (
+        "overload_backlog_threshold",
+        "window_count",
+        "longest_window_cycles",
+        "peak_total_queue_depth",
+        "policy_decision_count",
+    ):
+        field_value = value.get(field)
+        if field_value is not None and (
+            type(field_value) is not int or field_value < 0
+        ):
+            raise RuntimeIntelligenceLabHandoffError(
+                f"pressure_window.{field} must be a non-negative integer "
+                f"when present: {regression_path}"
+            )
+    for field in ("peak_window", "longest_window"):
+        field_value = value.get(field)
+        if field_value is not None and not isinstance(field_value, dict):
+            raise RuntimeIntelligenceLabHandoffError(
+                f"pressure_window.{field} must be an object when present: "
+                f"{regression_path}"
+            )
+    windows = value.get("windows")
+    if windows is not None and (
+        not isinstance(windows, list)
+        or not all(isinstance(item, dict) for item in windows)
+    ):
+        raise RuntimeIntelligenceLabHandoffError(
+            f"pressure_window.windows must be an object list when present: "
+            f"{regression_path}"
+        )
+    for field in (
+        "limited_tasks",
+        "protected_tasks",
+        "fallback_tasks",
+        "pressure_reasons",
+    ):
+        field_value = value.get(field)
+        if field_value is not None and (
+            not isinstance(field_value, list)
+            or not all(isinstance(item, str) for item in field_value)
+        ):
+            raise RuntimeIntelligenceLabHandoffError(
+                f"pressure_window.{field} must be a string list when present: "
+                f"{regression_path}"
+            )
+
+
 def _lab_bundle_alignment(files: dict[str, str]) -> dict[str, Any]:
     produced_file_keys = tuple(sorted(files))
     required_file_keys = (