From fcd0bcfab31de90a0f1d1949c9c2a9f04214e0f9 Mon Sep 17 00:00:00 2001 From: hyeokjun32 Date: Mon, 22 Jun 2026 14:29:52 +0900 Subject: [PATCH] Preserve Orchestrator pressure window handoff --- README.md | 16 +- docs/ko/README.md | 16 +- docs/runtime-telemetry-history.md | 10 ++ inferedge_env/result/lab_handoff.py | 138 ++++++++++++++++++ inferedge_env/result/telemetry_history.py | 120 +++++++++++++++ .../test_runtime_intelligence_lab_handoff.py | 75 ++++++++++ tests/test_runtime_telemetry_history.py | 77 ++++++++++ 7 files changed, 446 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 899aad0..ee70426 100644 --- a/README.md +++ b/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 ` 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 diff --git a/docs/ko/README.md b/docs/ko/README.md index 2b34b63..07d196b 100644 --- a/docs/ko/README.md +++ b/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한다: diff --git a/docs/runtime-telemetry-history.md b/docs/runtime-telemetry-history.md index b1ec6db..913a5d1 100644 --- a/docs/runtime-telemetry-history.md +++ b/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 diff --git a/inferedge_env/result/lab_handoff.py b/inferedge_env/result/lab_handoff.py index d7ddc10..b759204 100644 --- a/inferedge_env/result/lab_handoff.py +++ b/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 = ( diff --git a/inferedge_env/result/telemetry_history.py b/inferedge_env/result/telemetry_history.py index 83d4dd5..7f34869 100644 --- a/inferedge_env/result/telemetry_history.py +++ b/inferedge_env/result/telemetry_history.py @@ -68,6 +68,9 @@ ORCHESTRATOR_WORKER_HEALTH_TREND_SCHEMA_VERSION = ( "inferedge-orchestrator-worker-health-trend-v1" ) +ORCHESTRATOR_PRESSURE_WINDOW_SUMMARY_SCHEMA_VERSION = ( + "inferedge-orchestrator-pressure-window-summary-v1" +) class RuntimeTelemetryHistoryError(ValueError): @@ -329,6 +332,7 @@ def inspect_runtime_telemetry_history( policy_pressure_summary_run_ids = _policy_pressure_summary_run_ids(payload) stale_drop_summary_run_ids = _stale_drop_summary_run_ids(payload) worker_health_trend_run_ids = _worker_health_trend_run_ids(payload) + pressure_window_summary_run_ids = _pressure_window_summary_run_ids(payload) timestamps = [ entry.get("telemetry_timestamp") for entry in runs @@ -383,6 +387,7 @@ def inspect_runtime_telemetry_history( "policy_pressure_summary_run_ids": policy_pressure_summary_run_ids, "stale_drop_summary_run_ids": stale_drop_summary_run_ids, "worker_health_trend_run_ids": worker_health_trend_run_ids, + "pressure_window_summary_run_ids": pressure_window_summary_run_ids, "first_telemetry_timestamp": min(timestamps) if timestamps else None, "last_telemetry_timestamp": max(timestamps) if timestamps else None, "execution_sequence_ids": sequence_ids, @@ -595,6 +600,27 @@ def _worker_health_trend_run_ids(payload: dict[str, Any]) -> list[str]: return run_ids +def _pressure_window_summary_run_ids(payload: dict[str, Any]) -> list[str]: + run_ids: list[str] = [] + for entry in payload.get("runs", []): + if not isinstance(entry, dict): + continue + context = entry.get("orchestrator_operation_context") + if _has_pressure_window_summary(context): + run_id = entry.get("run_id") + if isinstance(run_id, str): + run_ids.append(run_id) + for item in payload.get("missing_telemetry", []): + if not isinstance(item, dict): + continue + context = item.get("orchestrator_operation_context") + if _has_pressure_window_summary(context): + run_id = item.get("run_id") + if isinstance(run_id, str): + run_ids.append(run_id) + return run_ids + + def _has_operation_risk_rollup(value: Any) -> bool: if not isinstance(value, dict): return False @@ -648,6 +674,17 @@ def _has_worker_health_trend(value: Any) -> bool: ) +def _has_pressure_window_summary(value: Any) -> bool: + if not isinstance(value, dict): + return False + operation = _candidate_operation_context(value) + timeline = operation.get("operation_timeline_summary") + return ( + isinstance(timeline, dict) + and isinstance(timeline.get("pressure_window"), dict) + ) + + def _candidate_operation_context(value: dict[str, Any]) -> dict[str, Any]: candidate_context = value.get("candidate_context") if not isinstance(candidate_context, dict): @@ -1364,6 +1401,89 @@ def _validate_orchestrator_operation_timeline_summary( summary.get("worker_health_trend"), source=source, ) + _validate_orchestrator_pressure_window_summary( + summary.get("pressure_window"), + source=source, + ) + return summary + + +def _validate_orchestrator_pressure_window_summary( + value: Any, + *, + source: Path, +) -> dict[str, Any] | None: + if value is None: + return None + if not isinstance(value, dict): + raise RuntimeTelemetryHistoryError( + "Orchestrator telemetry feed pressure_window must be " + f"an object: {source}" + ) + summary = deepcopy(value) + 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 summary.get(key) != expected: + raise RuntimeTelemetryHistoryError( + "Orchestrator telemetry feed pressure_window." + f"{key} must be {expected}: {source}" + ) + if summary.get("not_a_deployment_decision") is not True: + raise RuntimeTelemetryHistoryError( + "Orchestrator telemetry feed pressure_window." + f"not_a_deployment_decision must be true: {source}" + ) + for field in ( + "overload_backlog_threshold", + "window_count", + "longest_window_cycles", + "peak_total_queue_depth", + "policy_decision_count", + ): + field_value = summary.get(field) + if field_value is not None and ( + type(field_value) is not int or field_value < 0 + ): + raise RuntimeTelemetryHistoryError( + "Orchestrator telemetry feed pressure_window." + f"{field} must be a non-negative integer when present: {source}" + ) + for field in ("peak_window", "longest_window"): + field_value = summary.get(field) + if field_value is not None and not isinstance(field_value, dict): + raise RuntimeTelemetryHistoryError( + "Orchestrator telemetry feed pressure_window." + f"{field} must be an object when present: {source}" + ) + windows = summary.get("windows") + if windows is not None and ( + not isinstance(windows, list) + or not all(isinstance(item, dict) for item in windows) + ): + raise RuntimeTelemetryHistoryError( + "Orchestrator telemetry feed pressure_window." + f"windows must be an object list when present: {source}" + ) + for field in ( + "limited_tasks", + "protected_tasks", + "fallback_tasks", + "pressure_reasons", + ): + field_value = summary.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 RuntimeTelemetryHistoryError( + "Orchestrator telemetry feed pressure_window." + f"{field} must be a string list when present: {source}" + ) return summary diff --git a/tests/test_runtime_intelligence_lab_handoff.py b/tests/test_runtime_intelligence_lab_handoff.py index 3b6bfdb..b6ffb01 100644 --- a/tests/test_runtime_intelligence_lab_handoff.py +++ b/tests/test_runtime_intelligence_lab_handoff.py @@ -18,6 +18,7 @@ ) from inferedge_env.result.telemetry_history import ( ORCHESTRATOR_EDGEENV_AIGUARD_EVIDENCE_CANDIDATES, + ORCHESTRATOR_PRESSURE_WINDOW_SUMMARY_SCHEMA_VERSION, ORCHESTRATOR_PRODUCER_LINEAGE_AIGUARD_EVIDENCE_TYPE, ORCHESTRATOR_TELEMETRY_FEED_ARTIFACT_ROLE, ORCHESTRATOR_TELEMETRY_FEED_PRODUCER_CONTRACT, @@ -124,6 +125,7 @@ def test_runtime_intelligence_lab_handoff_manifest_records_producer_contracts( assert payload["lab_bundle_alignment"]["optional_aiguard_evidence_types"] == [ "stale_frame_risk", "edgeenv_orchestrator_stale_drop_summary", + "edgeenv_orchestrator_pressure_window_summary", ] assert payload["lab_bundle_alignment"]["optional_aiguard_source_traceability"] == { "context_role": "read_only_optional_source_traceability", @@ -262,6 +264,8 @@ def test_runtime_intelligence_lab_handoff_manifest_records_producer_contracts( "orchestrator_stale_drop_summary_run_ids": ["candidate"], "orchestrator_worker_health_trend_present": True, "orchestrator_worker_health_trend_run_ids": ["candidate"], + "orchestrator_pressure_window_summary_present": True, + "orchestrator_pressure_window_summary_run_ids": ["candidate"], "duration_traceability_present": True, "duration_traceability_run_ids": ["candidate"], "duration_sources": ["entrypoint_requested_frames"], @@ -1164,6 +1168,32 @@ def test_runtime_intelligence_lab_handoff_rejects_worker_health_trend_as_decisio ) +def test_runtime_intelligence_lab_handoff_rejects_pressure_window_as_decision( + tmp_path, +): + baseline_path, candidate_path, regression_path, history_path = _write_handoff_files( + tmp_path + ) + regression = json.loads(regression_path.read_text(encoding="utf-8")) + regression["runtime_telemetry_context"]["candidate"][ + "orchestrator_operation_context" + ]["candidate_context"]["operation"]["operation_timeline_summary"][ + "pressure_window" + ]["decision_owner"] = "orchestrator" + regression_path.write_text(json.dumps(regression), encoding="utf-8") + + with pytest.raises( + RuntimeIntelligenceLabHandoffError, + match="pressure_window.decision_owner must be lab", + ): + build_runtime_intelligence_lab_handoff_manifest( + baseline_result_path=baseline_path, + candidate_result_path=candidate_path, + edgeenv_regression_report_path=regression_path, + telemetry_history_path=history_path, + ) + + def test_runtime_intelligence_lab_handoff_rejects_policy_pressure_mirror_drift( tmp_path, ): @@ -1507,6 +1537,7 @@ def _operation_timeline_summary_payload() -> dict: "policy_pressure": _policy_pressure_summary_payload(), "stale_drop": _stale_drop_summary_payload(), "worker_health_trend": _worker_health_trend_payload(), + "pressure_window": _pressure_window_payload(), "affected_tasks": { "deadline_missed": ["vision_agent"], "fallback": ["voice_command_agent"], @@ -1515,6 +1546,7 @@ def _operation_timeline_summary_payload() -> dict: "constrained": [], "stale_drop": ["vision_agent"], "policy_pressure": ["vision_agent", "voice_command_agent"], + "pressure_window": ["vision_agent", "voice_command_agent"], }, "review_hints": [ "review_queue_pressure", @@ -1524,10 +1556,53 @@ def _operation_timeline_summary_payload() -> dict: "review_stale_drop", "review_policy_pressure", "review_worker_health_trend", + "review_sustained_pressure_window", ], } +def _pressure_window_payload() -> dict: + return { + "schema_version": ORCHESTRATOR_PRESSURE_WINDOW_SUMMARY_SCHEMA_VERSION, + "operation_context_role": "supplemental", + "scheduler_owner": "orchestrator", + "decision_owner": "lab", + "not_a_deployment_decision": True, + "source": "queue_depth_timeline+policy_decision_log", + "first_read": "review_sustained_pressure_window", + "overload_backlog_threshold": 5, + "window_count": 1, + "longest_window_cycles": 2, + "peak_total_queue_depth": 7, + "peak_window": { + "start_cycle": 3, + "end_cycle": 4, + "peak_total_queue_depth": 7, + }, + "longest_window": { + "start_cycle": 3, + "end_cycle": 4, + "duration_cycles": 2, + }, + "windows": [ + { + "start_cycle": 3, + "end_cycle": 4, + "duration_cycles": 2, + "peak_total_queue_depth": 7, + } + ], + "limited_tasks": ["vision_agent", "voice_command_agent"], + "protected_tasks": ["safety_monitor_agent"], + "fallback_tasks": ["voice_command_agent"], + "pressure_reasons": ["queue_backlog_threshold_exceeded"], + "policy_decision_count": 2, + "interpretation": ( + "Pressure window preserved as Lab review context only." + ), + } + + def _worker_health_trend_payload() -> dict: return { "schema_version": ORCHESTRATOR_WORKER_HEALTH_TREND_SCHEMA_VERSION, diff --git a/tests/test_runtime_telemetry_history.py b/tests/test_runtime_telemetry_history.py index e512c2a..45521ef 100644 --- a/tests/test_runtime_telemetry_history.py +++ b/tests/test_runtime_telemetry_history.py @@ -17,6 +17,7 @@ ORCHESTRATOR_EDGEENV_OPERATION_CONTEXT_ROLE, ORCHESTRATOR_EDGEENV_REQUIRED_CANDIDATE_FIELDS, ORCHESTRATOR_PRODUCER_LINEAGE_AIGUARD_EVIDENCE_TYPE, + ORCHESTRATOR_PRESSURE_WINDOW_SUMMARY_SCHEMA_VERSION, ORCHESTRATOR_REMOTE_OPERATION_BOUNDARY, ORCHESTRATOR_REMOTE_RUNTIME_EVENT_SUMMARY_ROLE, ORCHESTRATOR_TELEMETRY_FEED_ARTIFACT_ROLE, @@ -404,6 +405,7 @@ def test_runtime_telemetry_history_preserves_operation_risk_summary( assert summary["replay"]["policy_pressure_summary_run_ids"] == ["candidate"] assert summary["replay"]["stale_drop_summary_run_ids"] == ["candidate"] assert summary["replay"]["worker_health_trend_run_ids"] == ["candidate"] + assert summary["replay"]["pressure_window_summary_run_ids"] == ["candidate"] assert summary["replay"]["orchestrator_context_run_ids"] == ["candidate"] @@ -450,6 +452,7 @@ def test_runtime_telemetry_history_preserves_operation_risk_rollup( assert summary["replay"]["policy_pressure_summary_run_ids"] == ["candidate"] assert summary["replay"]["stale_drop_summary_run_ids"] == ["candidate"] assert summary["replay"]["worker_health_trend_run_ids"] == ["candidate"] + assert summary["replay"]["pressure_window_summary_run_ids"] == ["candidate"] def test_runtime_telemetry_history_rejects_operation_risk_rollup_as_decision( @@ -597,6 +600,35 @@ def test_runtime_telemetry_history_rejects_worker_health_trend_as_decision( build_runtime_telemetry_history(edgeenv_root, orchestrator_feeds=[feed_path]) +def test_runtime_telemetry_history_rejects_pressure_window_as_decision( + tmp_path, + bench_config, + target_profile, + config_files, +): + edgeenv_root = tmp_path / ".edgeenv" + _write_registered_run( + edgeenv_root, + bench_config, + target_profile, + config_files, + run_id="candidate", + runtime_telemetry=_runtime_telemetry_payload(sequence_id=2), + ) + feed = _orchestrator_feed_payload("candidate") + feed["candidate_context"]["operation"]["operation_timeline_summary"][ + "pressure_window" + ]["decision_owner"] = "orchestrator" + feed_path = tmp_path / "orchestrator-feed.json" + feed_path.write_text(json.dumps(feed), encoding="utf-8") + + with pytest.raises( + RuntimeTelemetryHistoryError, + match="pressure_window.decision_owner must be lab", + ): + build_runtime_telemetry_history(edgeenv_root, orchestrator_feeds=[feed_path]) + + def test_runtime_telemetry_history_rejects_policy_pressure_mirror_drift( tmp_path, bench_config, @@ -1757,6 +1789,7 @@ def _operation_timeline_summary_payload() -> dict: "policy_pressure": _policy_pressure_summary_payload(), "stale_drop": _stale_drop_summary_payload(), "worker_health_trend": _worker_health_trend_payload(), + "pressure_window": _pressure_window_payload(), "affected_tasks": { "deadline_missed": ["vision_agent"], "fallback": ["voice_command_agent"], @@ -1765,6 +1798,7 @@ def _operation_timeline_summary_payload() -> dict: "constrained": [], "stale_drop": ["vision_agent"], "policy_pressure": ["vision_agent", "voice_command_agent"], + "pressure_window": ["vision_agent", "voice_command_agent"], }, "review_hints": [ "review_queue_pressure", @@ -1774,7 +1808,50 @@ def _operation_timeline_summary_payload() -> dict: "review_stale_drop", "review_policy_pressure", "review_worker_health_trend", + "review_sustained_pressure_window", + ], + } + + +def _pressure_window_payload() -> dict: + return { + "schema_version": ORCHESTRATOR_PRESSURE_WINDOW_SUMMARY_SCHEMA_VERSION, + "operation_context_role": "supplemental", + "scheduler_owner": "orchestrator", + "decision_owner": "lab", + "not_a_deployment_decision": True, + "source": "queue_depth_timeline+policy_decision_log", + "first_read": "review_sustained_pressure_window", + "overload_backlog_threshold": 5, + "window_count": 1, + "longest_window_cycles": 2, + "peak_total_queue_depth": 7, + "peak_window": { + "start_cycle": 3, + "end_cycle": 4, + "peak_total_queue_depth": 7, + }, + "longest_window": { + "start_cycle": 3, + "end_cycle": 4, + "duration_cycles": 2, + }, + "windows": [ + { + "start_cycle": 3, + "end_cycle": 4, + "duration_cycles": 2, + "peak_total_queue_depth": 7, + } ], + "limited_tasks": ["vision_agent", "voice_command_agent"], + "protected_tasks": ["safety_monitor_agent"], + "fallback_tasks": ["voice_command_agent"], + "pressure_reasons": ["queue_backlog_threshold_exceeded"], + "policy_decision_count": 2, + "interpretation": ( + "Pressure window preserved as Lab review context only." + ), }