diff --git a/docs/agent_orchestration_summary_contract.ko.md b/docs/agent_orchestration_summary_contract.ko.md index b6c85e1..3295af3 100644 --- a/docs/agent_orchestration_summary_contract.ko.md +++ b/docs/agent_orchestration_summary_contract.ko.md @@ -140,6 +140,11 @@ Top-level summary: scheduler-delay signal, affected task, worker health를 compact supplemental risk context로 요약해 AIGuard/Lab/EdgeEnv가 사용할 수 있게 합니다. `decision_owner=lab`과 `not_a_deployment_decision=true`를 명시합니다. +- `scheduler_fairness_summary`: sustained operation evidence 안의 additive + context로, 보호된 high-priority task, scheduler-delay 또는 starvation risk가 + 있는 task, degraded/constrained worker, task별 fairness context를 + 요약합니다. Orchestrator가 소유하는 supplemental scheduling evidence이며 + Lab deployment decision이나 EdgeEnv regression judgement가 아닙니다. - `runtime_event_timeline`: queue snapshot, drop, scheduler selection, execution, policy decision, resource snapshot을 순서대로 남기는 event log입니다. execution event는 backlog/delay 확인을 위한 additive diff --git a/docs/agent_orchestration_summary_contract.md b/docs/agent_orchestration_summary_contract.md index 2a1f598..6b103db 100644 --- a/docs/agent_orchestration_summary_contract.md +++ b/docs/agent_orchestration_summary_contract.md @@ -140,6 +140,11 @@ Additive operation-health fields: counts, scheduler-delay signals, affected tasks, and worker health into a compact supplemental risk context for AIGuard/Lab/EdgeEnv. It explicitly records `decision_owner=lab` and `not_a_deployment_decision=true`. +- `scheduler_fairness_summary`: additive context inside sustained operation + evidence that summarizes protected high-priority tasks, tasks with + scheduler-delay or starvation risk, degraded/constrained workers, and + per-task fairness context. It is Orchestrator-owned supplemental scheduling + evidence, not a Lab deployment decision or EdgeEnv regression judgement. - `runtime_event_timeline`: ordered event log for queue snapshots, drops, scheduler selections, executions, policy decisions, and resource snapshots. Execution events include additive `scheduler_delay_cycles` and diff --git a/examples/telemetry/README.ko.md b/examples/telemetry/README.ko.md index 24e44ae..c449c4b 100644 --- a/examples/telemetry/README.ko.md +++ b/examples/telemetry/README.ko.md @@ -15,7 +15,7 @@ portfolio evidence를 확인할 수 있도록 제공한다. | Question | Start with | Why | | --- | --- | --- | | scheduler가 overload에서 high-priority work를 보호하는가? | `phase3_overload_sample.json` | baseline vs scheduled p95 latency와 low-priority drop을 보여준다. | -| 3-agent sustained path가 downstream review용 scheduler-delay evidence를 기록하는가? | `agent_scheduler_delay_sample.json` | 지연된 execution, policy/drop reason count, AIGuard/Lab signal name을 보여준다. | +| 3-agent sustained path가 downstream review용 scheduler-delay evidence와 fairness context를 기록하는가? | `agent_scheduler_delay_sample.json` | 지연된 execution, scheduler fairness/starvation context, policy/drop reason count, AIGuard/Lab signal name을 보여준다. | | remote dispatch starter evidence가 production retry control을 주장하지 않고 bounded fallback recovery를 보여주는가? | `remote_fallback_recovery_sample.json` | primary failure, fallback recovery, compact runtime event summary, starter boundary field를 보여준다. | | 어떤 sample이 local CI가 아니라 기존 Jetson evidence를 전제로 하는가? | `jetson_*_sample.json` files | portable CI output이 아니라 curated physical-device 또는 TensorRT-backed evidence snapshot이다. | @@ -24,7 +24,7 @@ portfolio evidence를 확인할 수 있도록 제공한다. | File | What it shows | | --- | --- | | `phase3_overload_sample.json` | synthetic FIFO baseline과 scheduler/load-shedding 비교. detector p95 end-to-end latency가 `782.0ms`에서 `8.0ms`로 개선되고 low-priority classifier work가 drop된다. | -| `agent_scheduler_delay_sample.json` | 3-agent sustained high-load config에서 추출한 curated excerpt. `scheduler_delay_event_count`, 지연된 execution event, policy/drop reason count, downstream AIGuard/Lab signal name을 보여준다. | +| `agent_scheduler_delay_sample.json` | 3-agent sustained high-load config에서 추출한 curated excerpt. `scheduler_delay_event_count`, `scheduler_fairness_summary`, 지연된 execution event, policy/drop reason count, downstream AIGuard/Lab signal name을 보여준다. | | `remote_fallback_recovery_sample.json` | remote dispatch starter에서 primary HTTP starter `connection_error`, 제한된 fallback worker recovery, retry/fallback plan field, downstream AIGuard/Lab signal name을 보여주는 curated excerpt. | | `jetson_smoke_dummy_sample.json` | Jetson dummy smoke path의 telemetry schema. task count, drop event, result event, scheduler decision, resource snapshot을 보여준다. | | `jetson_onnx_smoke_sample.json` | ONNX Runtime worker smoke path의 telemetry schema. result event metadata, output shape `[1, 2]`, resource snapshot을 보여준다. | @@ -41,6 +41,7 @@ sample은 다음 telemetry signal을 포함한다. - drop events - overload 또는 policy decisions - scheduler delay event count와 queue wait evidence +- scheduler fairness / starvation context - remote dispatch starter failure/fallback recovery evidence - result events - resource snapshots diff --git a/examples/telemetry/README.md b/examples/telemetry/README.md index 60640c9..66b5db6 100644 --- a/examples/telemetry/README.md +++ b/examples/telemetry/README.md @@ -15,7 +15,7 @@ They are not benchmark artifacts. Raw runtime reports remain ignored under | Question | Start with | Why | | --- | --- | --- | | Does the scheduler protect high-priority work under overload? | `phase3_overload_sample.json` | Shows baseline vs scheduled p95 latency and low-priority drops. | -| Did the 3-agent sustained path record scheduler-delay evidence for downstream review? | `agent_scheduler_delay_sample.json` | Shows delayed execution, policy/drop reason counts, and AIGuard/Lab signal names. | +| Did the 3-agent sustained path record scheduler-delay evidence and fairness context for downstream review? | `agent_scheduler_delay_sample.json` | Shows delayed execution, scheduler fairness/starvation context, policy/drop reason counts, and AIGuard/Lab signal names. | | Does remote dispatch starter evidence show bounded fallback recovery without claiming production retry control? | `remote_fallback_recovery_sample.json` | Shows primary failure, fallback recovery, compact runtime event summary, and starter boundary fields. | | Which samples require prior Jetson evidence rather than local CI? | `jetson_*_sample.json` files | They are curated physical-device or TensorRT-backed evidence snapshots, not portable CI outputs. | @@ -24,7 +24,7 @@ They are not benchmark artifacts. Raw runtime reports remain ignored under | File | What it shows | | --- | --- | | `phase3_overload_sample.json` | Synthetic FIFO baseline vs scheduler/load-shedding comparison. The detector p95 end-to-end latency improves from `782.0ms` to `8.0ms`, while low-priority classifier work is dropped. | -| `agent_scheduler_delay_sample.json` | Curated excerpt from the 3-agent sustained high-load config showing `scheduler_delay_event_count`, a delayed execution event, policy/drop reason counts, and the downstream AIGuard/Lab signal names. | +| `agent_scheduler_delay_sample.json` | Curated excerpt from the 3-agent sustained high-load config showing `scheduler_delay_event_count`, `scheduler_fairness_summary`, a delayed execution event, policy/drop reason counts, and the downstream AIGuard/Lab signal names. | | `remote_fallback_recovery_sample.json` | Curated remote dispatch starter excerpt showing primary HTTP starter `connection_error`, bounded fallback worker recovery, retry/fallback plan fields, and downstream AIGuard/Lab signal names. | | `jetson_smoke_dummy_sample.json` | Telemetry schema from the Jetson dummy smoke path: task counts, drop events, result events, scheduler decisions, and resource snapshots. | | `jetson_onnx_smoke_sample.json` | Telemetry schema from the ONNX Runtime worker smoke path: result event metadata, output shape `[1, 2]`, and resource snapshots. | @@ -41,6 +41,7 @@ The samples cover these telemetry signals: - drop events - overload or policy decisions - scheduler delay event counts and queue wait evidence +- scheduler fairness / starvation context - remote dispatch starter failure/fallback recovery evidence - result events - resource snapshots diff --git a/examples/telemetry/agent_scheduler_delay_sample.json b/examples/telemetry/agent_scheduler_delay_sample.json index e888078..ffe34ae 100644 --- a/examples/telemetry/agent_scheduler_delay_sample.json +++ b/examples/telemetry/agent_scheduler_delay_sample.json @@ -29,6 +29,68 @@ "queue_overflow_drop_oldest": 12 } }, + "scheduler_fairness_summary": { + "schema_version": "inferedge-orchestrator-scheduler-fairness-summary-v1", + "operation_context_role": "supplemental", + "scheduler_owner": "orchestrator", + "decision_owner": "lab", + "not_a_deployment_decision": true, + "protected_high_priority_tasks": [ + "safety_monitor_agent" + ], + "tasks_with_starvation_risk": [ + "vision_agent" + ], + "tasks_with_scheduler_delay": [ + "vision_agent" + ], + "tasks_with_degradation": [ + "voice_command_agent" + ], + "first_read": "review_scheduler_fairness_context", + "task_fairness": { + "vision_agent": { + "priority": 80, + "executed_count": 3, + "dropped_count": 12, + "fallback_count": 0, + "scheduler_delay_event_count": 3, + "max_scheduler_delay_cycles": 3, + "health_state": "constrained", + "starvation_risk": true, + "starvation_reasons": [ + "scheduler_delay_present", + "multi_cycle_scheduler_delay" + ] + }, + "voice_command_agent": { + "priority": 50, + "executed_count": 1, + "dropped_count": 9, + "fallback_count": 9, + "scheduler_delay_event_count": 0, + "max_scheduler_delay_cycles": 0, + "health_state": "degraded", + "starvation_risk": true, + "starvation_reasons": [ + "fallback_policy_used", + "worker_degraded" + ] + }, + "safety_monitor_agent": { + "priority": 100, + "executed_count": 16, + "dropped_count": 0, + "fallback_count": 0, + "scheduler_delay_event_count": 0, + "max_scheduler_delay_cycles": 0, + "health_state": "healthy", + "starvation_risk": false, + "starvation_reasons": [] + } + }, + "interpretation": "Scheduler fairness is supplemental operation evidence; Lab remains the final deployment decision owner." + }, "delayed_execution_sample": { "event_type": "execution", "agent_id": "vision_agent", diff --git a/scripts/check_edgeenv_runtime_feed_contract.py b/scripts/check_edgeenv_runtime_feed_contract.py index 3e8dd88..3a2004d 100644 --- a/scripts/check_edgeenv_runtime_feed_contract.py +++ b/scripts/check_edgeenv_runtime_feed_contract.py @@ -111,6 +111,9 @@ 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 {} + scheduler_fairness = operation_timeline_summary.get( + "scheduler_fairness" + ) or {} print( "operation_timeline: " f"review_hints={_format_list(review_hints)}; " @@ -121,6 +124,16 @@ 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 scheduler_fairness: + print( + "scheduler_fairness: " + "protected=" + f"{_format_list(scheduler_fairness.get('protected_high_priority_tasks'))}; " + "starvation_risk=" + f"{_format_list(scheduler_fairness.get('tasks_with_starvation_risk'))}; " + "degraded=" + f"{_format_list(scheduler_fairness.get('tasks_with_degradation'))}" + ) return 0 diff --git a/src/inferedge_orchestrator/sustained.py b/src/inferedge_orchestrator/sustained.py index 4a18d05..fc59e14 100644 --- a/src/inferedge_orchestrator/sustained.py +++ b/src/inferedge_orchestrator/sustained.py @@ -43,6 +43,9 @@ "queue_overflow_drop_oldest": "stale_queue_overflow", "load_shedding_backlog_threshold_exceeded": "load_shedding_stale_backlog", } +SCHEDULER_FAIRNESS_SUMMARY_SCHEMA = ( + "inferedge-orchestrator-scheduler-fairness-summary-v1" +) def apply_device_local_input_overrides( @@ -268,7 +271,7 @@ def _multi_workload_summary( **_local_profile_signals(report), **_producer_source_signals(report), }, - "operation_timeline_summary": _operation_timeline_summary(report), + "operation_timeline_summary": _operation_timeline_summary(report, config), "operation_risk_rollup": operation_risk_rollup, "next_validation_step": _next_validation_step(config), } @@ -320,6 +323,7 @@ def _edgeenv_runtime_telemetry_feed( {}, ), "operation_risk_rollup": report.get("operation_risk_rollup", {}), + "scheduler_fairness_summary": _scheduler_fairness_summary(config, report), "tasks_with_deadline_miss": runtime_event_summary.get( "tasks_with_deadline_miss", [], @@ -329,7 +333,7 @@ def _edgeenv_runtime_telemetry_feed( "tasks_with_scheduler_delay", [], ), - "operation_timeline_summary": _operation_timeline_summary(report), + "operation_timeline_summary": _operation_timeline_summary(report, config), } operation["latency_budget_protection"] = _latency_budget_protection_context( config, @@ -498,6 +502,16 @@ def validate_edgeenv_runtime_telemetry_feed( "stale_drop_summary must be an object" ) _validate_stale_drop_summary(stale_drop_summary) + scheduler_fairness_summary = candidate_context["operation"].get( + "scheduler_fairness_summary" + ) + if scheduler_fairness_summary is not None: + if not isinstance(scheduler_fairness_summary, dict): + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "scheduler_fairness_summary must be an object" + ) + _validate_scheduler_fairness_summary(scheduler_fairness_summary) if not isinstance(candidate_context.get("resource"), dict): raise ValueError( "edgeenv_runtime_telemetry_feed.candidate_context.resource must be " @@ -708,6 +722,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) + scheduler_fairness = payload.get("scheduler_fairness") + if scheduler_fairness is not None: + if not isinstance(scheduler_fairness, dict): + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "operation_timeline_summary.scheduler_fairness must be an object" + ) + _validate_scheduler_fairness_summary(scheduler_fairness) def _validate_stale_drop_summary(payload: dict[str, Any]) -> None: @@ -761,6 +783,66 @@ def _validate_stale_drop_summary(payload: dict[str, Any]) -> None: ) +def _validate_scheduler_fairness_summary(payload: dict[str, Any]) -> None: + if payload.get("schema_version") != SCHEDULER_FAIRNESS_SUMMARY_SCHEMA: + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "scheduler_fairness_summary.schema_version must be " + f"{SCHEDULER_FAIRNESS_SUMMARY_SCHEMA}" + ) + if payload.get("operation_context_role") != "supplemental": + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "scheduler_fairness_summary.operation_context_role must be supplemental" + ) + if payload.get("scheduler_owner") != "orchestrator": + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "scheduler_fairness_summary.scheduler_owner must be orchestrator" + ) + if payload.get("decision_owner") != "lab": + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "scheduler_fairness_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." + "scheduler_fairness_summary.not_a_deployment_decision must be true" + ) + for field in ( + "protected_high_priority_tasks", + "tasks_with_starvation_risk", + "tasks_with_scheduler_delay", + "tasks_with_degradation", + ): + 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"scheduler_fairness_summary.{field} must be a string list" + ) + task_fairness = payload.get("task_fairness") + if not isinstance(task_fairness, dict) or not task_fairness: + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "scheduler_fairness_summary.task_fairness must be a non-empty object" + ) + for task_name, task_context in task_fairness.items(): + if not isinstance(task_name, str) or not task_name: + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "scheduler_fairness_summary.task_fairness keys must be strings" + ) + if not isinstance(task_context, dict): + raise ValueError( + "edgeenv_runtime_telemetry_feed.candidate_context.operation." + "scheduler_fairness_summary.task_fairness values must be objects" + ) + + def _validate_edgeenv_producer_context(producer: dict[str, Any]) -> None: if producer.get("operation_context_role") != "supplemental": raise ValueError( @@ -1066,6 +1148,16 @@ def _positive_int(value: Any) -> bool: return isinstance(value, int) and not isinstance(value, bool) and value > 0 +def _non_negative_int_value(value: Any) -> int: + if isinstance(value, bool): + return 0 + if isinstance(value, int): + return max(value, 0) + if isinstance(value, float): + return max(int(value), 0) + return 0 + + def _scenario_identity(config: OrchestratorConfig) -> dict[str, str]: labels = { "normal": { @@ -1288,7 +1380,10 @@ def _producer_source_signals(report: dict[str, Any]) -> dict[str, Any]: } -def _operation_timeline_summary(report: dict[str, Any]) -> dict[str, Any]: +def _operation_timeline_summary( + report: dict[str, Any], + config: OrchestratorConfig | None = None, +) -> dict[str, Any]: queue_summary = _dict_value(report.get("queue_state_summary")) runtime_event_summary = _dict_value(report.get("runtime_event_summary")) worker_health = _dict_value(report.get("worker_health_snapshot")) @@ -1324,6 +1419,7 @@ def _operation_timeline_summary(report: dict[str, Any]) -> dict[str, Any]: "latency": _latency_timeline_summary(report), "policy": _policy_timeline_summary(report), "stale_drop": stale_drop, + "scheduler_fairness": _scheduler_fairness_summary(config, report), "affected_tasks": { "deadline_missed": _string_list( runtime_event_summary.get("tasks_with_deadline_miss") @@ -1502,6 +1598,175 @@ def _operation_timeline_review_hints( return hints or ["operation_timeline_nominal"] +def _scheduler_fairness_summary( + config: OrchestratorConfig | None, + report: dict[str, Any], +) -> dict[str, Any]: + runtime_event_summary = _dict_value(report.get("runtime_event_summary")) + task_event_summary = _dict_value(runtime_event_summary.get("task_event_summary")) + worker_health = _dict_value(report.get("worker_health_snapshot")) + workers = _dict_value(worker_health.get("workers")) + task_names = _configured_task_names(config, workers, task_event_summary) + task_fairness: dict[str, Any] = {} + tasks_with_starvation_risk: list[str] = [] + tasks_with_scheduler_delay: list[str] = [] + tasks_with_degradation: list[str] = [] + + for task_name in task_names: + worker = _dict_value(workers.get(task_name)) + task_events = _dict_value(task_event_summary.get(task_name)) + task_config = _task_config_by_name(config, task_name) + context = _task_fairness_context(task_name, task_config, worker, task_events) + task_fairness[task_name] = context + if context["starvation_risk"]: + tasks_with_starvation_risk.append(task_name) + if _positive_int(context.get("scheduler_delay_event_count")): + tasks_with_scheduler_delay.append(task_name) + if context.get("health_state") in {"constrained", "degraded"}: + tasks_with_degradation.append(task_name) + + return { + "schema_version": SCHEDULER_FAIRNESS_SUMMARY_SCHEMA, + "operation_context_role": "supplemental", + "scheduler_owner": "orchestrator", + "decision_owner": "lab", + "not_a_deployment_decision": True, + "source": "task_config+worker_health_snapshot+runtime_event_summary", + "protected_high_priority_tasks": _protected_high_priority_tasks( + config, + task_fairness, + ), + "tasks_with_starvation_risk": tasks_with_starvation_risk, + "tasks_with_scheduler_delay": tasks_with_scheduler_delay, + "tasks_with_degradation": tasks_with_degradation, + "first_read": ( + "review_scheduler_fairness_context" + if tasks_with_starvation_risk or tasks_with_scheduler_delay + else "scheduler_fairness_nominal" + ), + "task_fairness": task_fairness, + "interpretation": ( + "Scheduler fairness is supplemental operation evidence for " + "reviewing protected, delayed, degraded, or starved workloads; " + "Lab remains the final deployment decision owner." + ), + } + + +def _configured_task_names( + config: OrchestratorConfig | None, + workers: dict[str, Any], + task_event_summary: dict[str, Any], +) -> list[str]: + names: list[str] = [] + if config is not None: + for task in config.tasks: + if task.name not in names: + names.append(task.name) + for source in (workers, task_event_summary): + for task_name in source: + if isinstance(task_name, str) and task_name and task_name not in names: + names.append(task_name) + return names + + +def _task_config_by_name( + config: OrchestratorConfig | None, + task_name: str, +) -> TaskConfig | None: + if config is None: + return None + for task in config.tasks: + if task.name == task_name: + return task + return None + + +def _task_fairness_context( + task_name: str, + task: TaskConfig | None, + worker: dict[str, Any], + task_events: dict[str, Any], +) -> dict[str, Any]: + executed_count = _non_negative_int_value(worker.get("executed_count")) + dropped_count = _non_negative_int_value(worker.get("dropped_count")) + fallback_count = _non_negative_int_value(worker.get("fallback_count")) + scheduler_delay_count = _non_negative_int_value( + task_events.get("scheduler_delay_event_count") + ) + max_delay_cycles = _non_negative_int_value( + task_events.get("max_scheduler_delay_cycles") + ) + starvation_reasons = _starvation_reasons( + executed_count=executed_count, + dropped_count=dropped_count, + fallback_count=fallback_count, + scheduler_delay_count=scheduler_delay_count, + max_delay_cycles=max_delay_cycles, + health_state=worker.get("health_state"), + ) + context = { + "agent_id": task.agent_id if task else worker.get("agent_id"), + "agent_type": task.agent_type if task else worker.get("agent_type"), + "priority": task.priority if task else worker.get("priority"), + "latency_budget_ms": task.latency_budget_ms if task else None, + "executed_count": executed_count, + "dropped_count": dropped_count, + "fallback_count": fallback_count, + "scheduler_delay_event_count": scheduler_delay_count, + "max_scheduler_delay_cycles": max_delay_cycles, + "health_state": worker.get("health_state"), + "operation_risk_summary": worker.get("operation_risk_summary"), + "starvation_risk": bool(starvation_reasons), + "starvation_reasons": starvation_reasons, + } + if task is None: + context["task_name"] = task_name + return context + + +def _starvation_reasons( + *, + executed_count: int, + dropped_count: int, + fallback_count: int, + scheduler_delay_count: int, + max_delay_cycles: int, + health_state: Any, +) -> list[str]: + reasons: list[str] = [] + if executed_count == 0 and dropped_count > 0: + reasons.append("dropped_without_execution") + if scheduler_delay_count > 0: + reasons.append("scheduler_delay_present") + if max_delay_cycles >= 3: + reasons.append("multi_cycle_scheduler_delay") + if fallback_count > 0: + reasons.append("fallback_policy_used") + if health_state == "degraded": + reasons.append("worker_degraded") + return reasons + + +def _protected_high_priority_tasks( + config: OrchestratorConfig | None, + task_fairness: dict[str, Any], +) -> list[str]: + if config is None or not config.tasks: + return [] + highest_priority = max(task.priority for task in config.tasks) + protected: list[str] = [] + for task in config.tasks: + context = task_fairness.get(task.name, {}) + if not isinstance(context, dict): + continue + if task.priority == highest_priority and _positive_int( + context.get("executed_count") + ): + protected.append(task.name) + return protected + + def _worker_names_with_health_state( workers: dict[str, Any], health_state: str, diff --git a/tests/test_multi_workload_sustained.py b/tests/test_multi_workload_sustained.py index 007e61c..fe5cfec 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, + SCHEDULER_FAIRNESS_SUMMARY_SCHEMA, STALE_DROP_SUMMARY_SCHEMA, apply_device_local_input_overrides, load_tegrastats_timeline, @@ -158,6 +159,23 @@ def test_run_multi_workload_sustained_writes_profile_summary(tmp_path) -> None: assert timeline["affected_tasks"]["stale_drop"] == ( stale_drop["tasks_with_stale_drop"] ) + fairness = timeline["scheduler_fairness"] + assert fairness["schema_version"] == SCHEDULER_FAIRNESS_SUMMARY_SCHEMA + assert fairness["operation_context_role"] == "supplemental" + assert fairness["scheduler_owner"] == "orchestrator" + assert fairness["decision_owner"] == "lab" + assert fairness["not_a_deployment_decision"] is True + assert fairness["protected_high_priority_tasks"] == ["safety_monitor_agent"] + assert "voice_command_agent" in fairness["tasks_with_scheduler_delay"] + assert "voice_command_agent" in fairness["tasks_with_starvation_risk"] + assert "voice_command_agent" in fairness["tasks_with_degradation"] + assert fairness["task_fairness"]["voice_command_agent"]["starvation_risk"] is True + assert "scheduler_delay_present" in ( + fairness["task_fairness"]["voice_command_agent"]["starvation_reasons"] + ) + assert "Lab remains the final deployment decision owner" in ( + fairness["interpretation"] + ) assert "review_queue_pressure" in timeline["review_hints"] assert "review_scheduler_delay" in timeline["review_hints"] assert "review_stale_drop" in timeline["review_hints"] @@ -235,6 +253,7 @@ def test_run_multi_workload_sustained_writes_profile_summary(tmp_path) -> None: assert candidate["operation"]["operation_timeline_summary"] == timeline assert candidate["operation"]["stale_drop_summary"] == stale_drop assert candidate["operation"]["operation_risk_rollup"] == risk_rollup + assert candidate["operation"]["scheduler_fairness_summary"] == fairness protection = candidate["operation"]["latency_budget_protection"] assert protection["schema_version"] == LATENCY_BUDGET_PROTECTION_SCHEMA assert protection["operation_context_role"] == "supplemental" diff --git a/tests/test_sample_telemetry_artifacts.py b/tests/test_sample_telemetry_artifacts.py index 73d8306..510804f 100644 --- a/tests/test_sample_telemetry_artifacts.py +++ b/tests/test_sample_telemetry_artifacts.py @@ -43,6 +43,27 @@ def test_agent_scheduler_delay_sample_records_downstream_signal() -> None: "queue_overflow_drop_oldest": 12, } + fairness = sample["scheduler_fairness_summary"] # type: ignore[index] + assert fairness["schema_version"] == ( # type: ignore[index] + "inferedge-orchestrator-scheduler-fairness-summary-v1" + ) + assert fairness["scheduler_owner"] == "orchestrator" # type: ignore[index] + assert fairness["decision_owner"] == "lab" # type: ignore[index] + assert fairness["not_a_deployment_decision"] is True # type: ignore[index] + assert fairness["protected_high_priority_tasks"] == [ # type: ignore[index] + "safety_monitor_agent" + ] + assert fairness["tasks_with_starvation_risk"] == [ # type: ignore[index] + "vision_agent" + ] + assert fairness["tasks_with_scheduler_delay"] == [ # type: ignore[index] + "vision_agent" + ] + assert "voice_command_agent" in fairness["tasks_with_degradation"] # type: ignore[index] + assert "Lab remains the final deployment decision owner" in ( # type: ignore[index] + fairness["interpretation"] + ) + delayed = sample["delayed_execution_sample"] # type: ignore[index] assert delayed["event_type"] == "execution" assert delayed["scheduler_delay_cycles"] == 3 diff --git a/tests/test_scripts.py b/tests/test_scripts.py index d41392b..12d755d 100644 --- a/tests/test_scripts.py +++ b/tests/test_scripts.py @@ -92,6 +92,9 @@ 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 "scheduler_fairness:" in out + assert "starvation_risk=" in out + assert "degraded=" in out def test_edgeenv_runtime_feed_contract_checker_fails_bad_guard_alignment(