| from __future__ import annotations |
|
|
| from collections import deque |
| import json |
| from pathlib import Path |
| from types import SimpleNamespace |
|
|
| import pytest |
|
|
| from jarvis.runtime_constants import ( |
| VALID_CONTROL_PRESETS, |
| VALID_OPERATOR_AUTH_MODES, |
| ) |
| from jarvis.runtime_operator_status import ( |
| normalize_operator_auth_mode, |
| operator_auth_risk, |
| operator_status_provider, |
| ) |
|
|
|
|
| def _collect_paths(value: object, *, prefix: str = "") -> set[str]: |
| paths: set[str] = set() |
| if isinstance(value, dict): |
| for key, nested in value.items(): |
| key_text = str(key) |
| path = f"{prefix}.{key_text}" if prefix else key_text |
| paths.add(path) |
| paths.update(_collect_paths(nested, prefix=path)) |
| return paths |
| if isinstance(value, list): |
| list_path = f"{prefix}[]" if prefix else "[]" |
| paths.add(list_path) |
| if value: |
| paths.update(_collect_paths(value[0], prefix=list_path)) |
| return paths |
| return paths |
|
|
|
|
| def test_normalize_operator_auth_mode_defaults_to_token_for_invalid() -> None: |
| assert normalize_operator_auth_mode("bad-mode", valid_modes=VALID_OPERATOR_AUTH_MODES) == "token" |
| assert normalize_operator_auth_mode("SESSION", valid_modes=VALID_OPERATOR_AUTH_MODES) == "session" |
|
|
|
|
| @pytest.mark.parametrize( |
| ("mode", "token_configured", "expected"), |
| [ |
| ("off", False, "high"), |
| ("off", True, "high"), |
| ("token", False, "high"), |
| ("token", True, "medium"), |
| ("session", False, "high"), |
| ("session", True, "low"), |
| ], |
| ) |
| def test_operator_auth_risk_matrix(mode: str, token_configured: bool, expected: str) -> None: |
| assert operator_auth_risk(auth_mode=mode, token_configured=token_configured) == expected |
|
|
|
|
| @pytest.mark.asyncio |
| async def test_operator_status_provider_shapes_payload_and_risk() -> None: |
| runtime = SimpleNamespace() |
| runtime.config = SimpleNamespace( |
| operator_server_enabled=True, |
| operator_server_host="127.0.0.1", |
| operator_server_port=8080, |
| operator_auth_mode="session", |
| operator_auth_token="tok", |
| persona_style="friendly", |
| backchannel_style="expressive", |
| ) |
| runtime._conversation_traces = deque([{"turn_id": 11}], maxlen=10) |
| runtime._episodic_timeline = deque([{"episode_id": 7}], maxlen=10) |
| runtime._active_control_preset = "quiet_hours" |
| runtime._personality_preview_snapshot = {"persona_style": "composed"} |
| runtime._runtime_profile_snapshot = lambda: {"wake_mode": "wake_word"} |
| runtime._runtime_invariant_snapshot = lambda: {"total_violations": 0} |
| runtime._operator_conversation_trace_provider = lambda limit=1: [{"turn_id": 11}] |
| runtime._operator_episodic_timeline_provider = lambda limit=20: [{"episode_id": 7}] |
|
|
| async def _system_status(_: dict[str, object]) -> dict[str, object]: |
| return {"content": [{"text": '{"ok": true, "service": "jarvis"}'}]} |
|
|
| status = await operator_status_provider( |
| runtime, |
| valid_operator_auth_modes=VALID_OPERATOR_AUTH_MODES, |
| valid_control_presets=VALID_CONTROL_PRESETS, |
| system_status_fn=_system_status, |
| ) |
|
|
| assert status["ok"] is True |
| assert status["operator"]["auth_mode"] == "session" |
| assert status["operator"]["auth_risk"] == "low" |
| assert status["conversation_trace"]["latest_turn_id"] == 11 |
| assert status["episodic_timeline"]["latest_episode_id"] == 7 |
| assert status["operator_controls"]["active_control_preset"] == "quiet_hours" |
| assert status["operator_controls"]["runtime_profile"]["wake_mode"] == "wake_word" |
| recommendations = status["operator_recommendations"] |
| assert recommendations["severity"] == "low" |
| assert recommendations["count"] >= 1 |
| assert recommendations["recommended"][0]["code"] == "healthy" |
|
|
|
|
| @pytest.mark.asyncio |
| async def test_operator_status_provider_normalizes_invalid_auth_mode() -> None: |
| runtime = SimpleNamespace() |
| runtime.config = SimpleNamespace( |
| operator_server_enabled=True, |
| operator_server_host="0.0.0.0", |
| operator_server_port=8080, |
| operator_auth_mode="bad-mode", |
| operator_auth_token="", |
| persona_style="friendly", |
| backchannel_style="balanced", |
| ) |
| runtime._conversation_traces = deque([], maxlen=10) |
| runtime._episodic_timeline = deque([], maxlen=10) |
| runtime._active_control_preset = "custom" |
| runtime._personality_preview_snapshot = None |
| runtime._runtime_profile_snapshot = lambda: {} |
| runtime._runtime_invariant_snapshot = lambda: {} |
| runtime._operator_conversation_trace_provider = lambda limit=1: [] |
| runtime._operator_episodic_timeline_provider = lambda limit=20: [] |
|
|
| async def _system_status(_: dict[str, object]) -> dict[str, object]: |
| return {"content": [{"text": "{}"}]} |
|
|
| status = await operator_status_provider( |
| runtime, |
| valid_operator_auth_modes=VALID_OPERATOR_AUTH_MODES, |
| valid_control_presets=VALID_CONTROL_PRESETS, |
| system_status_fn=_system_status, |
| ) |
|
|
| assert status["operator"]["auth_mode"] == "token" |
| assert status["operator"]["auth_risk"] == "high" |
| recommendations = status["operator_recommendations"] |
| assert recommendations["severity"] in {"medium", "high"} |
| assert any( |
| row["code"] == "operator_auth_risk" |
| for row in recommendations["recommended"] |
| ) |
|
|
|
|
| @pytest.mark.asyncio |
| async def test_operator_status_provider_recommends_on_health_and_checkpoint_signals() -> None: |
| runtime = SimpleNamespace() |
| runtime.config = SimpleNamespace( |
| operator_server_enabled=True, |
| operator_server_host="127.0.0.1", |
| operator_server_port=8080, |
| operator_auth_mode="token", |
| operator_auth_token="set-token", |
| persona_style="friendly", |
| backchannel_style="balanced", |
| ) |
| runtime._conversation_traces = deque([], maxlen=10) |
| runtime._episodic_timeline = deque([], maxlen=10) |
| runtime._active_control_preset = "custom" |
| runtime._personality_preview_snapshot = None |
| runtime._runtime_profile_snapshot = lambda: {} |
| runtime._runtime_invariant_snapshot = lambda: {"total_violations": 3} |
| runtime._operator_conversation_trace_provider = lambda limit=1: [] |
| runtime._operator_episodic_timeline_provider = lambda limit=20: [] |
|
|
| async def _system_status(_: dict[str, object]) -> dict[str, object]: |
| payload = { |
| "health": {"health_level": "degraded", "reasons": ["memory_error"]}, |
| "plan_preview": {"pending_count": 2}, |
| "expansion": { |
| "planner_engine": { |
| "autonomy_waiting_checkpoint_count": 1, |
| "autonomy_backlog_step_count": 3, |
| "autonomy_needs_replan_count": 1, |
| "autonomy_retry_pending_count": 2, |
| }, |
| "proactive": {"approval_pending_count": 1}, |
| }, |
| "voice_attention": { |
| "multimodal_grounding": { |
| "confidence_band": "low", |
| "overall_confidence": 0.2, |
| } |
| }, |
| "dead_letter_queue": {"pending_count": 2}, |
| "recovery_journal": {"interrupted_count": 1, "unresolved_count": 1}, |
| } |
| return {"content": [{"text": str(payload).replace("'", '"')}]} |
|
|
| status = await operator_status_provider( |
| runtime, |
| valid_operator_auth_modes=VALID_OPERATOR_AUTH_MODES, |
| valid_control_presets=VALID_CONTROL_PRESETS, |
| system_status_fn=_system_status, |
| ) |
|
|
| recommendations = status["operator_recommendations"] |
| assert recommendations["severity"] in {"medium", "high"} |
| codes = {row["code"] for row in recommendations["recommended"]} |
| assert "runtime_health_degraded" in codes |
| assert "pending_previews" in codes |
| assert "autonomy_waiting_checkpoint" in codes |
| assert "autonomy_backlog_steps" in codes |
| assert "autonomy_needs_replan" in codes |
| assert "autonomy_retry_pending" in codes |
| assert "approval_queue_pending" in codes |
| assert "dead_letter_pending" in codes |
| assert "recovery_unresolved" in codes |
| assert "multimodal_low_confidence" in codes |
|
|
|
|
| @pytest.mark.asyncio |
| async def test_operator_status_snapshot_contract_paths_stable() -> None: |
| runtime = SimpleNamespace() |
| runtime.config = SimpleNamespace( |
| operator_server_enabled=True, |
| operator_server_host="127.0.0.1", |
| operator_server_port=8080, |
| operator_auth_mode="session", |
| operator_auth_token="tok", |
| persona_style="friendly", |
| backchannel_style="balanced", |
| ) |
| runtime._conversation_traces = deque([{"turn_id": 11}], maxlen=10) |
| runtime._episodic_timeline = deque([{"episode_id": 7}], maxlen=10) |
| runtime._active_control_preset = "custom" |
| runtime._personality_preview_snapshot = {"persona_style": "composed"} |
| runtime._runtime_profile_snapshot = lambda: {"wake_mode": "wake_word"} |
| runtime._runtime_invariant_snapshot = lambda: {"total_violations": 0} |
| runtime._operator_conversation_trace_provider = lambda limit=1: [{"turn_id": 11}] |
| runtime._operator_episodic_timeline_provider = lambda limit=20: [{"episode_id": 7}] |
|
|
| async def _system_status(_: dict[str, object]) -> dict[str, object]: |
| return {"content": [{"text": '{"ok": true, "service": "jarvis"}'}]} |
|
|
| status = await operator_status_provider( |
| runtime, |
| valid_operator_auth_modes=VALID_OPERATOR_AUTH_MODES, |
| valid_control_presets=VALID_CONTROL_PRESETS, |
| system_status_fn=_system_status, |
| ) |
|
|
| snapshot_path = Path(__file__).resolve().parents[1] / "docs" / "evals" / "operator-status-snapshot.json" |
| snapshot = json.loads(snapshot_path.read_text(encoding="utf-8")) |
| required_paths = _collect_paths(snapshot) |
| actual_paths = _collect_paths(status) |
| missing = sorted(required_paths - actual_paths) |
| assert not missing |
|
|
|
|
| @pytest.mark.asyncio |
| async def test_operator_status_recommendation_codes_stress_contract() -> None: |
| runtime = SimpleNamespace() |
| runtime.config = SimpleNamespace( |
| operator_server_enabled=True, |
| operator_server_host="127.0.0.1", |
| operator_server_port=8080, |
| operator_auth_mode="token", |
| operator_auth_token="", |
| persona_style="friendly", |
| backchannel_style="balanced", |
| ) |
| runtime._conversation_traces = deque([], maxlen=10) |
| runtime._episodic_timeline = deque([], maxlen=10) |
| runtime._active_control_preset = "custom" |
| runtime._personality_preview_snapshot = None |
| runtime._runtime_profile_snapshot = lambda: {} |
| runtime._runtime_invariant_snapshot = lambda: {"total_violations": 3} |
| runtime._operator_conversation_trace_provider = lambda limit=1: [] |
| runtime._operator_episodic_timeline_provider = lambda limit=20: [] |
|
|
| async def _system_status(_: dict[str, object]) -> dict[str, object]: |
| payload = { |
| "health": {"health_level": "degraded", "reasons": ["memory_error"]}, |
| "plan_preview": {"pending_count": 2}, |
| "expansion": { |
| "planner_engine": { |
| "autonomy_waiting_checkpoint_count": 1, |
| "autonomy_backlog_step_count": 2, |
| "autonomy_needs_replan_count": 1, |
| "autonomy_retry_pending_count": 1, |
| }, |
| "proactive": {"approval_pending_count": 1}, |
| }, |
| "voice_attention": {"multimodal_grounding": {"confidence_band": "low"}}, |
| "dead_letter_queue": {"pending_count": 1}, |
| "recovery_journal": {"interrupted_count": 1, "unresolved_count": 0}, |
| } |
| return {"content": [{"text": json.dumps(payload)}]} |
|
|
| status = await operator_status_provider( |
| runtime, |
| valid_operator_auth_modes=VALID_OPERATOR_AUTH_MODES, |
| valid_control_presets=VALID_CONTROL_PRESETS, |
| system_status_fn=_system_status, |
| ) |
|
|
| codes = {row["code"] for row in status["operator_recommendations"]["recommended"]} |
| required_codes = { |
| "operator_auth_risk", |
| "runtime_health_degraded", |
| "health_reasons_present", |
| "runtime_invariants", |
| "pending_previews", |
| "autonomy_waiting_checkpoint", |
| "autonomy_backlog_steps", |
| "autonomy_needs_replan", |
| "autonomy_retry_pending", |
| "approval_queue_pending", |
| "dead_letter_pending", |
| "recovery_unresolved", |
| "multimodal_low_confidence", |
| } |
| assert required_codes.issubset(codes) |
|
|
