"""Lifecycle robustness tests for jarvis.__main__.Jarvis.""" import json import pytest from pathlib import Path from collections import deque from types import SimpleNamespace from unittest.mock import MagicMock, AsyncMock, patch from jarvis.__main__ import Jarvis, TELEMETRY_SERVICE_ERROR_DETAILS, TELEMETRY_STORAGE_ERROR_DETAILS from jarvis.presence import State from jarvis.voice_attention import VoiceAttentionConfig, VoiceAttentionController def test_stop_is_noop_when_not_started(): jarvis = Jarvis.__new__(Jarvis) jarvis._started = False Jarvis.stop(jarvis) # should not raise def test_stop_suppresses_component_errors_and_resets_started(): jarvis = Jarvis.__new__(Jarvis) jarvis._started = True jarvis._output_stream = MagicMock() jarvis._output_stream.stop.side_effect = RuntimeError("stream stop failed") jarvis._output_stream.close.side_effect = RuntimeError("stream close failed") jarvis.face_tracker = MagicMock() jarvis.face_tracker.stop.side_effect = RuntimeError("face stop failed") jarvis.hand_tracker = MagicMock() jarvis.hand_tracker.stop.side_effect = RuntimeError("hand stop failed") jarvis._use_robot_audio = True jarvis.robot = MagicMock() jarvis.robot.stop_audio.side_effect = RuntimeError("audio stop failed") jarvis.robot.disconnect.side_effect = RuntimeError("disconnect failed") jarvis.presence = MagicMock() jarvis.presence.stop.side_effect = RuntimeError("presence stop failed") jarvis.config = SimpleNamespace(motion_enabled=True) Jarvis.stop(jarvis) assert jarvis._started is False assert jarvis._output_stream is None assert jarvis.face_tracker is None assert jarvis.hand_tracker is None @pytest.mark.asyncio async def test_run_cleans_up_when_start_fails(): jarvis = Jarvis.__new__(Jarvis) jarvis.start = MagicMock(side_effect=RuntimeError("start failed")) jarvis.stop = MagicMock() jarvis._listen_task = None jarvis._tts_task = None jarvis._filler_task = None jarvis.brain = MagicMock() jarvis.brain.close = AsyncMock() with pytest.raises(RuntimeError): await Jarvis.run(jarvis) jarvis.brain.close.assert_awaited_once() jarvis.stop.assert_called_once() def test_startup_summary_lines_include_core_status(): jarvis = Jarvis.__new__(Jarvis) jarvis.robot = SimpleNamespace(sim=True) jarvis.args = SimpleNamespace(no_vision=False) jarvis.tts = None jarvis.config = SimpleNamespace( motion_enabled=True, hand_track_enabled=False, home_enabled=True, home_conversation_enabled=False, memory_enabled=True, memory_path="/tmp/memory.sqlite", persona_style="composed", startup_warnings=[], tool_allowlist=["a"], tool_denylist=["b", "c"], ) lines = Jarvis._startup_summary_lines(jarvis) joined = "\n".join(lines) assert "Mode: simulation" in joined assert "Memory: enabled" in joined assert "Config warnings: 0" in joined assert "Tool policy: allow=1 deny=2" in joined assert "Error taxonomy: total=" in joined def test_parse_memory_correction_command_forget(): command = Jarvis._parse_memory_correction_command("forget memory 42") assert command == ("memory_forget", {"memory_id": 42}) def test_parse_memory_correction_command_update(): command = Jarvis._parse_memory_correction_command("update memory 7 to Call me Captain.") assert command == ("memory_update", {"memory_id": 7, "text": "Call me Captain."}) def test_parse_memory_correction_command_rejects_ambiguous_phrases(): assert Jarvis._parse_memory_correction_command("forget this") is None def test_classify_user_intent(): assert Jarvis._classify_user_intent("What time is it?") == "answer" assert Jarvis._classify_user_intent("Turn on the kitchen lights.") == "action" assert Jarvis._classify_user_intent("Can you turn on the lights and tell me the weather?") == "hybrid" def test_looks_like_user_correction(): assert Jarvis._looks_like_user_correction("Actually, I meant the bedroom lamp.") is True assert Jarvis._looks_like_user_correction("No, I meant tomorrow morning.") is True assert Jarvis._looks_like_user_correction("What time is it?") is False def test_followup_carryover_candidate_accepts_short_slot_reply_when_previous_request_unresolved(): jarvis = Jarvis.__new__(Jarvis) jarvis._followup_carryover = { "text": "Turn on the lights in the living room.", "intent": "action", "timestamp": 100.0, "unresolved": True, } assert Jarvis._is_followup_carryover_candidate(jarvis, "the bedroom", now_ts=130.0) is True def test_followup_carryover_candidate_rejects_explicit_new_action(): jarvis = Jarvis.__new__(Jarvis) jarvis._followup_carryover = { "text": "Turn on the lights in the living room.", "intent": "action", "timestamp": 100.0, "unresolved": True, } assert Jarvis._is_followup_carryover_candidate(jarvis, "turn on the kitchen lights", now_ts=130.0) is False def test_with_followup_carryover_augments_prompt_for_unresolved_followup(): jarvis = Jarvis.__new__(Jarvis) jarvis._followup_carryover = { "text": "Set the bedroom lights to warm white.", "intent": "action", "timestamp": 200.0, "unresolved": True, } augmented, applied = Jarvis._with_followup_carryover(jarvis, "and in the office", now_ts=220.0) assert applied is True assert "Follow-up intent carryover:" in augmented assert "Previous request: Set the bedroom lights to warm white." in augmented assert "unresolved slots" in augmented def test_with_followup_carryover_skips_acknowledgement_only_replies(): jarvis = Jarvis.__new__(Jarvis) jarvis._followup_carryover = { "text": "Set the bedroom lights to warm white.", "intent": "action", "timestamp": 200.0, "unresolved": True, } augmented, applied = Jarvis._with_followup_carryover(jarvis, "okay", now_ts=220.0) assert applied is False assert augmented == "okay" def test_update_followup_carryover_tracks_resolution_status(): jarvis = Jarvis.__new__(Jarvis) jarvis._followup_carryover = {} Jarvis._update_followup_carryover(jarvis, "Turn on the porch light.", "action", resolved=False, now_ts=300.0) assert jarvis._followup_carryover["unresolved"] is True assert jarvis._followup_carryover["intent"] == "action" assert jarvis._followup_carryover["timestamp"] == 300.0 Jarvis._update_followup_carryover(jarvis, "Turn on the porch light.", "action", resolved=True, now_ts=320.0) assert jarvis._followup_carryover["unresolved"] is False Jarvis._update_followup_carryover(jarvis, "What is the weather?", "answer", resolved=None, now_ts=340.0) assert jarvis._followup_carryover["unresolved"] is False assert jarvis._followup_carryover["intent"] == "answer" def test_requires_stt_repair_for_low_confidence_action_intent(): jarvis = Jarvis.__new__(Jarvis) jarvis._stt_diagnostics = { "confidence_score": 0.2, "confidence_band": "low", "fallback_used": False, } assert Jarvis._requires_stt_repair(jarvis, "turn on the bedroom lights", "action") is True def test_requires_stt_repair_skips_answers_and_corrections(): jarvis = Jarvis.__new__(Jarvis) jarvis._stt_diagnostics = { "confidence_score": 0.2, "confidence_band": "low", "fallback_used": True, } assert Jarvis._requires_stt_repair(jarvis, "what time is it", "answer") is False assert Jarvis._requires_stt_repair(jarvis, "actually, I meant the kitchen", "action") is False def test_with_voice_profile_guidance_applies_non_default_verbosity(): jarvis = Jarvis.__new__(Jarvis) jarvis.config = SimpleNamespace(identity_default_user="operator") jarvis._voice_user_profiles = {"operator": {"verbosity": "brief"}} guided = Jarvis._with_voice_profile_guidance(jarvis, "Turn on the lights.") assert "Voice profile preference" in guided assert "concise" in guided def test_with_voice_profile_guidance_applies_tone_preference(): jarvis = Jarvis.__new__(Jarvis) jarvis.config = SimpleNamespace(identity_default_user="operator") jarvis._voice_user_profiles = {"operator": {"tone": "formal"}} guided = Jarvis._with_voice_profile_guidance(jarvis, "Turn on the lights.") assert "Voice profile preference" in guided assert "formal, composed phrasing" in guided def test_learn_voice_preferences_updates_profile_and_telemetry(): jarvis = Jarvis.__new__(Jarvis) jarvis.config = SimpleNamespace(identity_default_user="operator") jarvis._voice_user_profiles = {} jarvis._telemetry = { "preference_update_turns": 0.0, "preference_update_fields": 0.0, } jarvis._last_learned_preferences = {} jarvis._persist_runtime_state_safe = MagicMock() jarvis._publish_voice_status = MagicMock() jarvis.brain = SimpleNamespace(_memory=None) updates = Jarvis._learn_voice_preferences( jarvis, "Please be brief and speak slower.", now_ts=10.0, ) assert updates["verbosity"] == "brief" assert updates["pace"] == "slow" assert jarvis._voice_user_profiles["operator"]["verbosity"] == "brief" assert jarvis._telemetry["preference_update_turns"] == 1.0 assert jarvis._telemetry["preference_update_fields"] == 2.0 assert jarvis._last_learned_preferences["user"] == "operator" jarvis._persist_runtime_state_safe.assert_called_once() jarvis._publish_voice_status.assert_called_once() def test_requires_confirmation_respects_voice_profile_confirmation_mode(): jarvis = Jarvis.__new__(Jarvis) jarvis.config = SimpleNamespace(identity_default_user="operator") jarvis._voice_user_profiles = {"operator": {"confirmations": "minimal"}} jarvis._last_doa_speech = True jarvis.presence = SimpleNamespace( signals=SimpleNamespace(face_last_seen=None, hand_last_seen=None, doa_last_seen=None) ) assert Jarvis._requires_confirmation(jarvis, now=10.0) is False jarvis._voice_user_profiles = {"operator": {"confirmations": "strict"}} assert Jarvis._requires_confirmation(jarvis, now=10.0) is True def test_confidence_pause_respects_voice_profile_pace(): jarvis = Jarvis.__new__(Jarvis) jarvis.config = SimpleNamespace(identity_default_user="operator") jarvis._voice_user_profiles = {"operator": {"pace": "slow"}} slow = Jarvis._confidence_pause(jarvis, "All set.") jarvis._voice_user_profiles = {"operator": {"pace": "fast"}} fast = Jarvis._confidence_pause(jarvis, "All set.") assert slow > fast def test_record_conversation_trace_writes_episodic_snapshot_for_action_turn(): jarvis = Jarvis.__new__(Jarvis) jarvis._turn_trace_seq = 0 jarvis._conversation_traces = deque(maxlen=50) jarvis._episodic_timeline = deque(maxlen=50) jarvis._episode_seq = 0 jarvis._observability = None jarvis.presence = SimpleNamespace(attention_source=lambda: "face") jarvis._turn_choreography = {} Jarvis._record_conversation_trace( jarvis, user_text="Turn on the porch light.", intent_class="action", turn_started_at=0.0, stt_latency_ms=25.0, llm_first_sentence_ms=40.0, tts_first_audio_ms=55.0, response_success=True, tool_summaries=[{"name": "smart_home_control", "status": "ok", "duration_ms": 12.0}], lifecycle="completed", used_brain_response=True, ) assert len(jarvis._conversation_traces) == 1 assert len(jarvis._episodic_timeline) == 1 trace = jarvis._conversation_traces[0] assert trace["conversation_id"] == "default" assert trace["parent_turn_id"] is None assert trace["route_policy"] == {} assert trace["correction_outcome"] == "none" assert trace["final_tool_usage"]["attempted_count"] == 1 episode = jarvis._episodic_timeline[0] assert episode["intent"] == "action" assert episode["conversation_id"] == "default" assert "porch light" in episode["summary"].lower() assert episode["tool_count"] == 1 def test_record_conversation_trace_captures_route_policy_and_correction_outcome(): jarvis = Jarvis.__new__(Jarvis) jarvis._turn_trace_seq = 0 jarvis._conversation_traces = deque(maxlen=20) jarvis._episodic_timeline = deque(maxlen=20) jarvis._episode_seq = 0 jarvis._conversation_id = "conv-1" jarvis._last_trace_turn_id = 7 jarvis._observability = None jarvis.presence = SimpleNamespace(attention_source=lambda: "doa") jarvis._turn_choreography = {} Jarvis._record_conversation_trace( jarvis, user_text="Unlock the front door now.", intent_class="action", turn_started_at=0.0, stt_latency_ms=30.0, llm_first_sentence_ms=50.0, tts_first_audio_ms=70.0, response_success=True, tool_summaries=[{"name": "smart_home_control", "status": "denied", "detail": "policy"}], lifecycle="completed", used_brain_response=True, route_policy={ "starting_agent": "safety", "route_confidence": 0.2, "guardrail_correction": "low_confidence_fail_closed", }, correction_outcome="resolved", interruption_route={ "strategy": "resume", "route_confidence": 0.91, "route_source": "router", }, continuation_from_turn_id=7, ) trace = jarvis._conversation_traces[0] assert trace["conversation_id"] == "conv-1" assert trace["parent_turn_id"] == 7 assert trace["route_policy"]["starting_agent"] == "safety" assert trace["route_policy"]["route_confidence"] == 0.2 assert trace["interruption_route"]["strategy"] == "resume" assert trace["correction_outcome"] == "resolved" assert trace["final_tool_usage"]["attempted_count"] == 1 def test_conversation_latency_and_policy_decision_analytics(): jarvis = Jarvis.__new__(Jarvis) jarvis._conversation_traces = deque( [ { "latencies_ms": {"total": 400.0}, "intent": "action", "wake_mode": "wake_word", "tool_calls": [{"name": "smart_home_control"}], "policy_decisions": [{"tool": "smart_home_control", "status": "denied", "detail": "policy"}], "requester_user": "operator", "turn_id": 2, "timestamp": 2.0, "route_policy": { "route_source": "router", "router_variant": "canary", "shadow_agreement": False, "starting_agent": "safety", "first_response_strategy": "clarify", "route_confidence": 0.42, "shadow_route": { "starting_agent": "action", "first_response_strategy": "act", }, }, }, { "latencies_ms": {"total": 150.0}, "intent": "answer", "wake_mode": "always_listening", "tool_calls": [], "policy_decisions": [], "requester_user": "operator", "turn_id": 1, "timestamp": 1.0, "route_policy": { "route_source": "router", "router_variant": "primary", "shadow_agreement": True, "starting_agent": "conversation", "first_response_strategy": "answer", "route_confidence": 0.81, }, }, ], maxlen=20, ) latency = Jarvis._conversation_latency_analytics(jarvis) assert latency["sample_count"] == 2 assert latency["overall_total_ms"]["p95"] >= latency["overall_total_ms"]["p50"] assert "action" in latency["by_intent"] assert "single" in latency["by_tool_mix"] policy = Jarvis._policy_decision_analytics(jarvis) assert policy["decision_count"] == 1 assert policy["by_tool"]["smart_home_control"] == 1 assert policy["by_user"]["operator"] == 1 canary = Jarvis._router_canary_analytics(jarvis) assert canary["sample_count"] == 2 assert canary["router_decision_count"] == 2 assert canary["canary_turn_count"] == 1 assert canary["shadow_compare_count"] == 2 assert canary["shadow_disagreement_count"] == 1 assert canary["recent_disagreements"][0]["turn_id"] == 2 def test_apply_turn_choreography_sets_presence_bias_fields(): jarvis = Jarvis.__new__(Jarvis) jarvis.presence = SimpleNamespace(signals=SimpleNamespace(turn_lean=0.0, turn_tilt=0.0, turn_glance_yaw=0.0)) jarvis._turn_choreography = {} jarvis._observability = None Jarvis._apply_turn_choreography(jarvis, State.THINKING) assert jarvis.presence.signals.turn_lean == 0.5 assert jarvis.presence.signals.turn_tilt == 2.0 assert jarvis.presence.signals.turn_glance_yaw == 8.0 snapshot = Jarvis._turn_choreography_snapshot(jarvis) assert snapshot["phase"] == "thinking" assert snapshot["label"] == "think_glance_away" def test_publish_voice_status_includes_turn_choreography(): jarvis = Jarvis.__new__(Jarvis) jarvis._voice_attention = VoiceAttentionController(VoiceAttentionConfig(wake_words=["jarvis"])) jarvis.presence = SimpleNamespace( signals=SimpleNamespace(state=State.LISTENING), attention_source=lambda: "face", ) jarvis._turn_choreography = {} jarvis.config = SimpleNamespace(identity_default_user="operator") jarvis._voice_user_profiles = {} jarvis._last_learned_preferences = { "user": "operator", "updates": {"verbosity": "brief"}, "applied_at": 1.0, "source_text": "be brief", } jarvis._last_doa_angle = 15.0 jarvis._last_doa_speech = True jarvis._last_doa_update = 0.0 jarvis._active_control_preset = "custom" jarvis._observability = None with patch("jarvis.__main__.set_runtime_voice_state") as set_runtime: Jarvis._publish_voice_status(jarvis) payload = set_runtime.call_args.args[0] assert payload["presence_state"] == "listening" assert payload["turn_choreography"]["phase"] == "listening" assert payload["turn_choreography"]["label"] == "listen_lean_in" assert "stt_diagnostics" in payload assert payload["stt_diagnostics"]["confidence_band"] == "unknown" assert payload["voice_profile_user"] == "operator" assert payload["voice_profile"]["verbosity"] == "normal" assert payload["control_preset"] == "custom" assert "acoustic_scene" in payload assert "attention_confidence" in payload["acoustic_scene"] assert payload["acoustic_scene"]["attention_source"] == "face" assert payload["preference_learning"]["user"] == "operator" assert "multimodal_grounding" in payload assert "confidence_band" in payload["multimodal_grounding"] assert "modality_scores" in payload["multimodal_grounding"] def test_publish_observability_status_defaults_when_observability_disabled(): jarvis = Jarvis.__new__(Jarvis) jarvis._observability = None with patch("jarvis.__main__.set_runtime_observability_state") as set_runtime: Jarvis._publish_observability_status(jarvis) payload = set_runtime.call_args.args[0] assert payload["enabled"] is False assert payload["uptime_sec"] == 0.0 assert payload["restart_count"] == 0 assert payload["intent_metrics"]["preference_update_turns"] == 0.0 assert payload["intent_metrics"]["preference_update_fields"] == 0.0 assert payload["latency_dashboards"]["sample_count"] == 0 assert payload["policy_decision_analytics"]["decision_count"] == 0 def test_publish_observability_status_uses_default_when_snapshot_raises(): jarvis = Jarvis.__new__(Jarvis) jarvis._observability = SimpleNamespace(status_snapshot=MagicMock(side_effect=RuntimeError("boom"))) jarvis._conversation_latency_analytics = MagicMock( return_value={ "sample_count": 1, "overall_total_ms": {"p50": 10.0, "p95": 10.0, "p99": 10.0}, "by_intent": {"action": {"p50": 10.0, "p95": 10.0, "p99": 10.0}}, "by_tool_mix": {"none": {"p50": 10.0, "p95": 10.0, "p99": 10.0}}, "by_wake_mode": {"wake_word": {"p50": 10.0, "p95": 10.0, "p99": 10.0}}, } ) jarvis._policy_decision_analytics = MagicMock( return_value={ "decision_count": 2, "by_tool": {"smart_home": 2}, "by_status": {"allowed": 2}, "by_reason": {"ok": 2}, "by_user": {"owner": 2}, "by_user_tool": {"owner:smart_home": 2}, } ) with patch("jarvis.__main__.set_runtime_observability_state") as set_runtime: Jarvis._publish_observability_status(jarvis) payload = set_runtime.call_args.args[0] assert payload["enabled"] is False assert payload["intent_metrics"]["preference_update_turns"] == 0.0 assert payload["intent_metrics"]["preference_update_fields"] == 0.0 assert payload["latency_dashboards"]["sample_count"] == 1 assert payload["policy_decision_analytics"]["decision_count"] == 2 jarvis._conversation_latency_analytics.assert_called_once() jarvis._policy_decision_analytics.assert_called_once() def test_runtime_invariant_checker_auto_heals_mode_and_preset_mismatches(): jarvis = Jarvis.__new__(Jarvis) voice = VoiceAttentionController(VoiceAttentionConfig(wake_words=["jarvis"])) voice.set_mode("push_to_talk") voice.set_push_to_talk_active(False) jarvis._voice_attention = voice jarvis._active_control_preset = "broken-preset" jarvis._runtime_invariant_checked_at = 0.0 jarvis._runtime_invariant_checked_monotonic = 0.0 jarvis._runtime_invariant_violations_total = 0 jarvis._runtime_invariant_auto_heals_total = 0 jarvis._runtime_invariant_recent = deque(maxlen=40) jarvis._observability = None snapshot = Jarvis._check_runtime_invariants(jarvis, auto_heal=True) assert voice.push_to_talk_active is True assert jarvis._active_control_preset == "custom" assert snapshot["total_violations"] == 2 assert snapshot["total_auto_heals"] == 2 codes = {item["code"] for item in snapshot["recent"]} assert "push_to_talk_mode_inactive" in codes assert "invalid_control_preset" in codes def test_start_requires_sounddevice_for_local_tts_playback(): jarvis = Jarvis.__new__(Jarvis) jarvis._started = False jarvis.robot = SimpleNamespace(connect=MagicMock(), sim=True) jarvis.presence = SimpleNamespace(start=MagicMock()) jarvis.config = SimpleNamespace( motion_enabled=False, sample_rate=16000, ) jarvis.args = SimpleNamespace(no_vision=True) jarvis.tts = object() jarvis.stop = MagicMock() with patch("jarvis.__main__.sd", None), patch("jarvis.__main__._SOUNDDEVICE_IMPORT_ERROR", "PortAudio missing"): with pytest.raises(RuntimeError, match="local audio playback"): Jarvis.start(jarvis) jarvis.stop.assert_called_once() def test_error_taxonomy_doc_matches_constants(): from jarvis.tool_errors import TOOL_SERVICE_ERROR_CODES doc_path = Path(__file__).resolve().parents[1] / "docs" / "operations" / "error-taxonomy.md" text = doc_path.read_text(encoding="utf-8") start = "" end = "" assert start in text and end in text block = text.split(start, 1)[1].split(end, 1)[0] documented = {line.strip() for line in block.splitlines() if line.strip()} assert documented == TOOL_SERVICE_ERROR_CODES def test_telemetry_snapshot_averages(): jarvis = Jarvis.__new__(Jarvis) jarvis._telemetry = { "turns": 10.0, "barge_ins": 2.0, "stt_latency_total_ms": 1000.0, "stt_latency_count": 4.0, "llm_first_sentence_total_ms": 1200.0, "llm_first_sentence_count": 3.0, "tts_first_audio_total_ms": 500.0, "tts_first_audio_count": 2.0, "service_errors": 3.0, "storage_errors": 1.0, "fallback_responses": 4.0, "multimodal_turns": 4.0, "multimodal_confidence_total": 2.4, "multimodal_low_confidence_turns": 1.0, "interruption_routes_total": 3.0, "interruption_resumes": 1.0, "interruption_replaces": 2.0, "interruption_clarifies": 0.0, "interruption_route_fallbacks": 1.0, "semantic_turn_decisions_total": 5.0, "semantic_turn_waits": 2.0, "semantic_turn_commits": 3.0, "semantic_turn_fallbacks": 1.0, "llm_prompt_tokens_total": 300.0, "llm_completion_tokens_total": 120.0, "llm_total_tokens_total": 420.0, "llm_cost_usd_total": 0.08, "llm_usage_samples": 2.0, "router_canary_turns_total": 1.0, "router_shadow_comparisons_total": 2.0, "router_shadow_agreements_total": 1.0, "router_shadow_disagreements_total": 1.0, } snapshot = Jarvis._telemetry_snapshot(jarvis) assert snapshot["turns"] == 10.0 assert snapshot["barge_ins"] == 2.0 assert snapshot["avg_stt_latency_ms"] == 250.0 assert snapshot["avg_llm_first_sentence_ms"] == 400.0 assert snapshot["avg_tts_first_audio_ms"] == 250.0 assert snapshot["service_errors"] == 3.0 assert snapshot["storage_errors"] == 1.0 assert snapshot["unknown_summary_details"] == 0.0 assert snapshot["fallback_responses"] == 4.0 intent = snapshot["intent_metrics"] assert intent["answer_quality_success_rate"] == 0.0 assert intent["completion_success_rate"] == 0.0 assert intent["correction_frequency"] == 0.0 multimodal = snapshot["multimodal_metrics"] assert multimodal["turn_count"] == 4.0 assert multimodal["avg_confidence"] == 0.6 assert multimodal["low_confidence_rate"] == 0.25 interruption = snapshot["interruption_metrics"] assert interruption["route_count"] == 3.0 assert interruption["resume_count"] == 1.0 assert interruption["fallback_count"] == 1.0 semantic_turn = snapshot["semantic_turn_metrics"] assert semantic_turn["decision_count"] == 5.0 assert semantic_turn["wait_count"] == 2.0 assert semantic_turn["commit_count"] == 3.0 llm_usage = snapshot["llm_token_usage"] assert llm_usage["total_tokens_total"] == 420.0 assert llm_usage["usage_samples"] == 2.0 assert llm_usage["avg_tokens_per_usage"] == pytest.approx(210.0) canary = snapshot["router_canary_metrics"] assert canary["canary_turns_total"] == 1.0 assert canary["shadow_comparisons_total"] == 2.0 assert canary["shadow_agreement_rate"] == pytest.approx(0.5) def test_telemetry_snapshot_intent_metric_rates(): jarvis = Jarvis.__new__(Jarvis) jarvis._telemetry = { "intent_turns_total": 10.0, "intent_answer_turns": 4.0, "intent_action_turns": 3.0, "intent_hybrid_turns": 3.0, "intent_answer_total": 8.0, "intent_answer_success": 6.0, "intent_completion_total": 5.0, "intent_completion_success": 4.0, "intent_corrections": 2.0, "multimodal_turns": 2.0, "multimodal_confidence_total": 1.0, "multimodal_low_confidence_turns": 1.0, } jarvis._telemetry_error_counts = {} snapshot = Jarvis._telemetry_snapshot(jarvis) intent = snapshot["intent_metrics"] assert intent["turn_count"] == 10.0 assert intent["answer_intent_count"] == 4.0 assert intent["action_intent_count"] == 3.0 assert intent["hybrid_intent_count"] == 3.0 assert intent["answer_sample_count"] == 8.0 assert intent["completion_sample_count"] == 5.0 assert intent["answer_quality_success_rate"] == pytest.approx(0.75) assert intent["completion_success_rate"] == pytest.approx(0.8) assert intent["correction_count"] == 2.0 assert intent["correction_frequency"] == pytest.approx(0.2) multimodal = snapshot["multimodal_metrics"] assert multimodal["avg_confidence"] == pytest.approx(0.5) assert multimodal["low_confidence_rate"] == pytest.approx(0.5) def test_refresh_tool_error_counters(): jarvis = Jarvis.__new__(Jarvis) jarvis._telemetry = {"service_errors": 0.0, "storage_errors": 0.0} from unittest.mock import patch with patch("jarvis.__main__.list_summaries", return_value=[ {"status": "error", "detail": "timeout"}, {"status": "error", "detail": "storage_error"}, {"status": "ok", "detail": "noop"}, ]): Jarvis._refresh_tool_error_counters(jarvis) assert jarvis._telemetry["service_errors"] == 1.0 assert jarvis._telemetry["storage_errors"] == 1.0 assert jarvis._telemetry["unknown_summary_details"] == 0.0 def test_refresh_tool_error_counters_classifies_missing_store_as_storage(): jarvis = Jarvis.__new__(Jarvis) jarvis._telemetry = {"service_errors": 0.0, "storage_errors": 0.0} from unittest.mock import patch with patch("jarvis.__main__.list_summaries", return_value=[ {"status": "error", "detail": "missing_store"}, {"status": "error", "detail": "storage_error"}, {"status": "error", "detail": "invalid_plan"}, ]): Jarvis._refresh_tool_error_counters(jarvis) assert jarvis._telemetry["service_errors"] == 1.0 assert jarvis._telemetry["storage_errors"] == 2.0 assert jarvis._telemetry["unknown_summary_details"] == 0.0 def test_refresh_tool_error_counters_includes_network_taxonomy(): jarvis = Jarvis.__new__(Jarvis) jarvis._telemetry = {"service_errors": 0.0, "storage_errors": 0.0} from unittest.mock import patch with patch("jarvis.__main__.list_summaries", return_value=[ {"status": "error", "detail": "network_client_error"}, {"status": "error", "detail": "http_error"}, {"status": "error", "detail": "unknown_error"}, ]): Jarvis._refresh_tool_error_counters(jarvis) assert jarvis._telemetry["service_errors"] == 3.0 assert jarvis._telemetry["storage_errors"] == 0.0 assert jarvis._telemetry["unknown_summary_details"] == 0.0 def test_refresh_tool_error_counters_tracks_unknown_summary_detail_count(): jarvis = Jarvis.__new__(Jarvis) jarvis._telemetry = {"service_errors": 0.0, "storage_errors": 0.0, "unknown_summary_details": 0.0} from unittest.mock import patch with patch("jarvis.__main__.list_summaries", return_value=[ {"status": "error", "detail": "timeout"}, {"status": "error", "detail": "not_a_real_code"}, {"status": "error", "detail": "another_unknown"}, ]): Jarvis._refresh_tool_error_counters(jarvis) assert jarvis._telemetry["service_errors"] == 1.0 assert jarvis._telemetry["storage_errors"] == 0.0 assert jarvis._telemetry["unknown_summary_details"] == 2.0 def test_refresh_tool_error_counters_reports_per_code_totals(): jarvis = Jarvis.__new__(Jarvis) jarvis._telemetry = {"service_errors": 0.0, "storage_errors": 0.0, "unknown_summary_details": 0.0} from unittest.mock import patch with patch("jarvis.__main__.list_summaries", return_value=[ {"status": "error", "detail": "timeout"}, {"status": "error", "detail": "timeout"}, {"status": "error", "detail": "storage_error"}, {"status": "error", "detail": "unknown_error"}, {"status": "error", "detail": "not_a_real_code"}, {"status": "ok", "detail": "timeout"}, ]): Jarvis._refresh_tool_error_counters(jarvis) assert jarvis._telemetry_error_counts == { "storage_error": 1.0, "timeout": 2.0, "unknown_error": 1.0, } snapshot = Jarvis._telemetry_snapshot(jarvis) assert snapshot["service_error_counts"] == jarvis._telemetry_error_counts assert snapshot["unknown_summary_details"] == 1.0 def test_telemetry_snapshot_guards_non_finite_values(): jarvis = Jarvis.__new__(Jarvis) jarvis._telemetry = { "turns": float("nan"), "barge_ins": float("inf"), "stt_latency_total_ms": float("inf"), "stt_latency_count": 1.0, "llm_first_sentence_total_ms": 100.0, "llm_first_sentence_count": float("nan"), "tts_first_audio_total_ms": 100.0, "tts_first_audio_count": 0.0, "service_errors": float("nan"), "storage_errors": float("inf"), "unknown_summary_details": float("nan"), "fallback_responses": float("inf"), } jarvis._telemetry_error_counts = {"timeout": float("nan"), "auth": 2.0} snapshot = Jarvis._telemetry_snapshot(jarvis) assert snapshot["turns"] == 0.0 assert snapshot["barge_ins"] == 0.0 assert snapshot["avg_stt_latency_ms"] == 0.0 assert snapshot["avg_llm_first_sentence_ms"] == 0.0 assert snapshot["service_errors"] == 0.0 assert snapshot["storage_errors"] == 0.0 assert snapshot["unknown_summary_details"] == 0.0 assert snapshot["fallback_responses"] == 0.0 assert snapshot["service_error_counts"] == {"auth": 2.0} def test_telemetry_error_taxonomy_matches_service_error_codes(): from jarvis.tools.services import SERVICE_ERROR_CODES from jarvis.tool_errors import TOOL_STORAGE_ERROR_DETAILS assert TELEMETRY_STORAGE_ERROR_DETAILS == TOOL_STORAGE_ERROR_DETAILS assert TELEMETRY_STORAGE_ERROR_DETAILS.issubset(SERVICE_ERROR_CODES) assert TELEMETRY_SERVICE_ERROR_DETAILS == (SERVICE_ERROR_CODES - TELEMETRY_STORAGE_ERROR_DETAILS) @pytest.mark.asyncio async def test_operator_control_handler_validates_and_applies_runtime_controls(): jarvis = Jarvis.__new__(Jarvis) jarvis._voice_attention = VoiceAttentionController(VoiceAttentionConfig(wake_words=["jarvis"])) jarvis._publish_voice_status = MagicMock() jarvis._publish_skills_status = MagicMock() jarvis._save_runtime_state = MagicMock() jarvis._tts_output_enabled = True jarvis.config = SimpleNamespace( motion_enabled=True, home_enabled=True, safe_mode_enabled=False, persona_style="composed", backchannel_style="balanced", identity_default_user="operator", ) jarvis.presence = SimpleNamespace( start=MagicMock(), stop=MagicMock(), set_backchannel_style=MagicMock(), ) jarvis.brain = SimpleNamespace(_memory=None) jarvis._personality_preview_snapshot = None jarvis._voice_user_profiles = {} jarvis._active_control_preset = "custom" jarvis._skills = SimpleNamespace( discover=MagicMock(), status_snapshot=lambda: {"enabled": True}, enable_skill=lambda name: (True, f"enabled:{name}"), disable_skill=lambda name: (True, f"disabled:{name}"), ) invalid_mode = await Jarvis._operator_control_handler( jarvis, "set_wake_mode", {"mode": "invalid"}, ) assert invalid_mode["ok"] is False assert invalid_mode["error"] == "invalid_payload" tts_result = await Jarvis._operator_control_handler( jarvis, "set_tts_enabled", {"enabled": "false"}, ) assert tts_result == {"ok": True, "tts_enabled": False} assert jarvis._tts_output_enabled is False invalid_bool = await Jarvis._operator_control_handler( jarvis, "set_home_enabled", {"enabled": "maybe"}, ) assert invalid_bool["ok"] is False assert invalid_bool["error"] == "invalid_payload" with patch("jarvis.__main__.service_tools.set_safe_mode") as set_safe_mode: safe_mode_result = await Jarvis._operator_control_handler( jarvis, "set_safe_mode", {"enabled": True}, ) assert safe_mode_result == {"ok": True, "safe_mode_enabled": True} assert jarvis.config.safe_mode_enabled is True set_safe_mode.assert_called_once_with(True) persona_result = await Jarvis._operator_control_handler( jarvis, "set_persona_style", {"style": "friendly"}, ) assert persona_result == {"ok": True, "persona_style": "friendly"} assert jarvis.config.persona_style == "friendly" sleep_result = await Jarvis._operator_control_handler( jarvis, "set_sleeping", {"sleeping": True}, ) assert sleep_result == {"ok": True, "sleeping": True} assert jarvis._voice_attention.sleeping is True wake_result = await Jarvis._operator_control_handler( jarvis, "set_sleeping", {"sleeping": False}, ) assert wake_result == {"ok": True, "sleeping": False} assert jarvis._voice_attention.sleeping is False backchannel_result = await Jarvis._operator_control_handler( jarvis, "set_backchannel_style", {"style": "quiet"}, ) assert backchannel_result == {"ok": True, "backchannel_style": "quiet"} assert jarvis.config.backchannel_style == "quiet" jarvis.presence.set_backchannel_style.assert_any_call("quiet") preview_result = await Jarvis._operator_control_handler( jarvis, "preview_personality", {"persona_style": "terse", "backchannel_style": "expressive"}, ) assert preview_result["ok"] is True assert preview_result["preview_active"] is True assert preview_result["baseline"] == {"persona_style": "friendly", "backchannel_style": "quiet"} assert jarvis.config.persona_style == "terse" assert jarvis.config.backchannel_style == "expressive" rollback_result = await Jarvis._operator_control_handler( jarvis, "rollback_personality_preview", {}, ) assert rollback_result["ok"] is True assert rollback_result["rolled_back"] is True assert jarvis.config.persona_style == "friendly" assert jarvis.config.backchannel_style == "quiet" assert jarvis._personality_preview_snapshot is None preview_commit = await Jarvis._operator_control_handler( jarvis, "preview_personality", {"persona_style": "terse"}, ) assert preview_commit["ok"] is True assert jarvis.config.persona_style == "terse" assert jarvis._personality_preview_snapshot == {"persona_style": "friendly", "backchannel_style": "quiet"} commit_result = await Jarvis._operator_control_handler( jarvis, "commit_personality_preview", {}, ) assert commit_result["ok"] is True assert commit_result["committed"] is True assert commit_result["preview_active"] is False assert jarvis._personality_preview_snapshot is None set_voice_profile = await Jarvis._operator_control_handler( jarvis, "set_voice_profile", {"user": "operator", "verbosity": "brief", "confirmations": "minimal", "pace": "fast", "tone": "witty"}, ) assert set_voice_profile["ok"] is True assert set_voice_profile["profile"]["verbosity"] == "brief" assert set_voice_profile["profile"]["confirmations"] == "minimal" assert set_voice_profile["profile"]["pace"] == "fast" assert set_voice_profile["profile"]["tone"] == "witty" list_voice_profiles = await Jarvis._operator_control_handler(jarvis, "list_voice_profiles", {}) assert list_voice_profiles["ok"] is True assert list_voice_profiles["active_user"] == "operator" assert "operator" in list_voice_profiles["profiles"] clear_voice_profile = await Jarvis._operator_control_handler( jarvis, "clear_voice_profile", {"user": "operator"}, ) assert clear_voice_profile == {"ok": True, "user": "operator", "removed": True} preset_result = await Jarvis._operator_control_handler( jarvis, "apply_control_preset", {"preset": "quiet_hours"}, ) assert preset_result["ok"] is True assert preset_result["preset"] == "quiet_hours" assert jarvis._active_control_preset == "quiet_hours" export_result = await Jarvis._operator_control_handler(jarvis, "export_runtime_profile", {}) assert export_result["ok"] is True assert export_result["runtime_profile"]["active_control_preset"] == "quiet_hours" import_result = await Jarvis._operator_control_handler( jarvis, "import_runtime_profile", {"profile": {"persona_style": "friendly", "backchannel_style": "balanced", "tts_enabled": True}}, ) assert import_result["ok"] is True assert import_result["runtime_profile"]["persona_style"] == "friendly" assert jarvis._active_control_preset == "custom" unknown = await Jarvis._operator_control_handler(jarvis, "nope", {}) assert unknown["ok"] is False assert unknown["error"] == "invalid_action" assert "available_actions" in unknown assert "set_sleeping" in unknown["available_actions"] assert "set_safe_mode" in unknown["available_actions"] def test_runtime_state_persists_and_restores_runtime_controls(tmp_path): state_path = tmp_path / "runtime-state.json" save_target = Jarvis.__new__(Jarvis) save_target._runtime_state_path = state_path save_target._voice_attention = VoiceAttentionController(VoiceAttentionConfig(wake_words=["jarvis"])) save_target._voice_attention.set_mode("wake_word") save_target._voice_attention.set_timeout_profile("long") save_target._voice_attention.set_push_to_talk_active(True) save_target._voice_attention.sleeping = True save_target.config = SimpleNamespace( motion_enabled=False, home_enabled=False, safe_mode_enabled=True, persona_style="friendly", backchannel_style="expressive", identity_default_user="operator", ) save_target._voice_user_profiles = { "operator": {"verbosity": "brief", "confirmations": "minimal", "pace": "fast"} } save_target._active_control_preset = "quiet_hours" save_target._tts_output_enabled = False save_target._awaiting_confirmation = True save_target._pending_text = "arm the system" save_target._awaiting_repair_confirmation = True save_target._repair_candidate_text = "turn on the porch lights" save_target._episode_seq = 4 save_target._episodic_timeline = deque( [ { "episode_id": 4, "timestamp": 1700000000.0, "turn_id": 8, "intent": "action", "lifecycle": "completed", "summary": "Turn on the porch lights -> tools: smart_home_control", "tool_count": 1, "completion_success": True, "response_success": True, } ], maxlen=200, ) Jarvis._save_runtime_state(save_target) payload = json.loads(state_path.read_text(encoding="utf-8")) assert payload["runtime"]["motion_enabled"] is False assert payload["runtime"]["home_enabled"] is False assert payload["runtime"]["safe_mode_enabled"] is True assert payload["runtime"]["tts_enabled"] is False assert payload["runtime"]["persona_style"] == "friendly" assert payload["runtime"]["backchannel_style"] == "expressive" assert payload["runtime"]["voice_user_profiles"]["operator"]["verbosity"] == "brief" assert payload["runtime"]["active_control_preset"] == "quiet_hours" assert payload["awaiting_repair_confirmation"] is True assert payload["repair_candidate_text"] == "turn on the porch lights" assert payload["episodic_timeline_seq"] == 4 assert payload["episodic_timeline"][0]["summary"].startswith("Turn on the porch lights") load_target = Jarvis.__new__(Jarvis) load_target._runtime_state_path = state_path load_target._voice_attention = VoiceAttentionController(VoiceAttentionConfig(wake_words=["jarvis"])) load_target.config = SimpleNamespace( motion_enabled=True, home_enabled=True, safe_mode_enabled=False, persona_style="composed", backchannel_style="balanced", identity_default_user="operator", ) load_target.brain = SimpleNamespace(_memory=None) load_target.presence = SimpleNamespace(set_backchannel_style=MagicMock()) load_target._tts_output_enabled = True load_target._awaiting_confirmation = False load_target._pending_text = None load_target._awaiting_repair_confirmation = False load_target._repair_candidate_text = None load_target._voice_user_profiles = {} load_target._active_control_preset = "custom" load_target._episode_seq = 0 load_target._episodic_timeline = deque(maxlen=200) with patch("jarvis.__main__.service_tools.set_safe_mode") as set_safe_mode: Jarvis._load_runtime_state(load_target) set_safe_mode.assert_called_once_with(True) assert load_target._voice_attention.mode == "wake_word" assert load_target._voice_attention.timeout_profile == "long" assert load_target._voice_attention.push_to_talk_active is True assert load_target._voice_attention.sleeping is True assert load_target.config.motion_enabled is False assert load_target.config.home_enabled is False assert load_target.config.safe_mode_enabled is True assert load_target._tts_output_enabled is False assert load_target.config.persona_style == "friendly" assert load_target.config.backchannel_style == "expressive" assert load_target._awaiting_confirmation is True assert load_target._pending_text == "arm the system" assert load_target._awaiting_repair_confirmation is True assert load_target._repair_candidate_text == "turn on the porch lights" assert load_target._voice_user_profiles["operator"]["verbosity"] == "brief" assert load_target._active_control_preset == "quiet_hours" assert load_target._episode_seq == 4 assert len(load_target._episodic_timeline) == 1 assert load_target._episodic_timeline[0]["summary"].startswith("Turn on the porch lights") load_target.presence.set_backchannel_style.assert_called_with("expressive") def test_runtime_state_saves_personality_baseline_when_preview_active(tmp_path): state_path = tmp_path / "runtime-state-preview.json" jarvis = Jarvis.__new__(Jarvis) jarvis._runtime_state_path = state_path jarvis._voice_attention = VoiceAttentionController(VoiceAttentionConfig(wake_words=["jarvis"])) jarvis.config = SimpleNamespace( motion_enabled=True, home_enabled=True, safe_mode_enabled=False, persona_style="friendly", backchannel_style="expressive", ) jarvis._tts_output_enabled = True jarvis._awaiting_confirmation = False jarvis._pending_text = None jarvis._awaiting_repair_confirmation = False jarvis._repair_candidate_text = None jarvis._personality_preview_snapshot = { "persona_style": "composed", "backchannel_style": "balanced", } Jarvis._save_runtime_state(jarvis) payload = json.loads(state_path.read_text(encoding="utf-8")) assert payload["runtime"]["persona_style"] == "composed" assert payload["runtime"]["backchannel_style"] == "balanced"