"""Tests for session analyzer — digest builder and LLM-based analysis.""" import json import subprocess from pathlib import Path from unittest.mock import MagicMock, patch import pytest from headroom.learn.analyzer import ( SessionAnalyzer, _build_digest, _call_cli_llm, _call_llm, _detect_default_model, _parse_llm_response, _strip_fenced_json, ) from headroom.learn.models import ( AnalysisResult, ErrorCategory, ProjectInfo, RecommendationTarget, SessionData, SessionEvent, ToolCall, ) def _project() -> ProjectInfo: return ProjectInfo( name="test-project", project_path=Path("/tmp/test-project"), data_path=Path("/tmp/test-data"), ) def _tc( name: str = "Bash", input_data: dict | None = None, output: str = "ok", is_error: bool = False, error_category: ErrorCategory = ErrorCategory.UNKNOWN, msg_index: int = 0, output_bytes: int = 0, ) -> ToolCall: return ToolCall( name=name, tool_call_id=f"tc_{msg_index}", input_data=input_data or {}, output=output, is_error=is_error, error_category=error_category, msg_index=msg_index, output_bytes=output_bytes or len(output), ) # ============================================================================= # Digest Builder Tests # ============================================================================= class TestDigestBuilder: def test_includes_project_info(self): project = _project() sessions = [SessionData(session_id="s1", tool_calls=[_tc()])] digest = _build_digest(project, sessions) assert "test-project" in digest assert "/tmp/test-project" in digest def test_includes_session_stats(self): sessions = [ SessionData( session_id="abc123", tool_calls=[_tc(msg_index=0), _tc(msg_index=1, is_error=True, output="Error!")], ) ] digest = _build_digest(_project(), sessions) assert "abc123" in digest assert "2 calls" in digest assert "1 failure" in digest def test_includes_tool_call_details(self): sessions = [ SessionData( session_id="s1", tool_calls=[ _tc( name="Read", input_data={"file_path": "/src/foo.py"}, output="contents", msg_index=0, ), _tc( name="Bash", input_data={"command": "python3 run.py"}, output="ModuleNotFoundError", is_error=True, error_category=ErrorCategory.MODULE_NOT_FOUND, msg_index=1, ), ], ) ] digest = _build_digest(_project(), sessions) assert "/src/foo.py" in digest assert "python3 run.py" in digest assert "ERROR" in digest assert "ModuleNotFoundError" in digest def test_includes_user_messages(self): tc = _tc(msg_index=0) events = [ SessionEvent(type="tool_call", msg_index=0, tool_call=tc), SessionEvent(type="user_message", msg_index=1, text="Use uv run instead"), ] sessions = [SessionData(session_id="s1", tool_calls=[tc], events=events)] digest = _build_digest(_project(), sessions) assert "USER:" in digest assert "Use uv run instead" in digest def test_includes_subagent_summaries(self): events = [ SessionEvent( type="agent_summary", msg_index=0, agent_tool_count=150, agent_tokens=60000, agent_prompt="Explore all test files", ), ] sessions = [SessionData(session_id="s1", events=events)] digest = _build_digest(_project(), sessions) assert "SUBAGENT" in digest assert "150 tool calls" in digest assert "Explore all test files" in digest def test_includes_interruptions(self): events = [ SessionEvent( type="interruption", msg_index=0, text="[Request interrupted by user]", ), ] sessions = [SessionData(session_id="s1", events=events)] digest = _build_digest(_project(), sessions) assert "INTERRUPTED" in digest def test_empty_sessions(self): digest = _build_digest(_project(), []) assert "0 sessions" in digest or "test-project" in digest # ============================================================================= # LLM Response Parser Tests # ============================================================================= class TestLLMResponseParser: def test_parses_context_file_rules(self): raw = { "context_file_rules": [ { "section": "Environment", "content": "- Use `uv run python` instead of `python3`", "estimated_tokens_saved": 800, "evidence_count": 5, } ], "memory_file_rules": [], } recs = _parse_llm_response(raw) assert len(recs) == 1 assert recs[0].target == RecommendationTarget.CONTEXT_FILE assert recs[0].section == "Environment" assert "uv run python" in recs[0].content assert recs[0].estimated_tokens_saved == 800 assert recs[0].evidence_count == 5 def test_parses_memory_file_rules(self): raw = { "context_file_rules": [], "memory_file_rules": [ { "section": "User Preferences", "content": "- Do not auto-execute curl commands", "estimated_tokens_saved": 500, "evidence_count": 3, } ], } recs = _parse_llm_response(raw) assert len(recs) == 1 assert recs[0].target == RecommendationTarget.MEMORY_FILE assert "curl" in recs[0].content def test_sorts_by_token_savings(self): raw = { "context_file_rules": [ { "section": "Paths", "content": "- Use correct paths", "estimated_tokens_saved": 200, "evidence_count": 2, }, { "section": "Environment", "content": "- Use uv", "estimated_tokens_saved": 1000, "evidence_count": 5, }, ], "memory_file_rules": [], } recs = _parse_llm_response(raw) assert recs[0].estimated_tokens_saved == 1000 assert recs[1].estimated_tokens_saved == 200 def test_handles_missing_fields(self): raw = { "context_file_rules": [ {"section": "Env", "content": "- stuff"}, {"section": "", "content": ""}, # should be skipped {"not_a_real_field": True}, # should be skipped ], "memory_file_rules": [], } recs = _parse_llm_response(raw) assert len(recs) == 1 def test_handles_empty_response(self): recs = _parse_llm_response({}) assert recs == [] def test_handles_non_dict_entries(self): raw = {"context_file_rules": ["not a dict", 42], "memory_file_rules": []} recs = _parse_llm_response(raw) assert recs == [] # ============================================================================= # Full Analyzer Integration Tests (mocked LLM) # ============================================================================= class TestSessionAnalyzer: def test_empty_sessions_no_llm_call(self): """No failures + no events → no LLM call, empty result.""" analyzer = SessionAnalyzer() result = analyzer.analyze(_project(), []) assert result.total_calls == 0 assert result.total_failures == 0 assert result.recommendations == [] @patch("headroom.learn.analyzer._call_llm") def test_calls_llm_with_digest(self, mock_call_llm: MagicMock): mock_call_llm.return_value = { "context_file_rules": [ { "section": "Environment", "content": "- Use uv run python", "estimated_tokens_saved": 800, "evidence_count": 3, } ], "memory_file_rules": [], } analyzer = SessionAnalyzer(model="test-model") sessions = [ SessionData( session_id="s1", tool_calls=[ _tc(msg_index=0, is_error=True, output="ModuleNotFoundError"), _tc(msg_index=1), ], ) ] result = analyzer.analyze(_project(), sessions) mock_call_llm.assert_called_once() assert result.total_calls == 2 assert result.total_failures == 1 assert len(result.recommendations) == 1 assert "uv run python" in result.recommendations[0].content @patch("headroom.learn.analyzer._call_llm") def test_handles_llm_failure_gracefully(self, mock_call_llm: MagicMock): mock_call_llm.side_effect = RuntimeError("API key not set") analyzer = SessionAnalyzer(model="test-model") sessions = [ SessionData( session_id="s1", tool_calls=[_tc(msg_index=0, is_error=True, output="error")], ) ] result = analyzer.analyze(_project(), sessions) # Stats should still work, just no recommendations assert result.total_calls == 1 assert result.total_failures == 1 assert result.recommendations == [] @patch("headroom.learn.analyzer._call_llm") def test_passes_events_to_digest(self, mock_call_llm: MagicMock): """User messages and subagent events should appear in the digest.""" mock_call_llm.return_value = {"context_file_rules": [], "memory_file_rules": []} tc = _tc(msg_index=0, is_error=True, output="error") events = [ SessionEvent(type="tool_call", msg_index=0, tool_call=tc), SessionEvent(type="user_message", msg_index=1, text="use venv python"), ] sessions = [SessionData(session_id="s1", tool_calls=[tc], events=events)] analyzer = SessionAnalyzer(model="test-model") analyzer.analyze(_project(), sessions) # Check that the digest passed to the LLM includes user message call_args = mock_call_llm.call_args digest = call_args[0][0] # first positional arg assert "use venv python" in digest # ============================================================================= # Model Auto-Detection # ============================================================================= class TestDetectDefaultModel: def test_anthropic_key(self, monkeypatch): monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test") monkeypatch.delenv("OPENAI_API_KEY", raising=False) monkeypatch.delenv("GEMINI_API_KEY", raising=False) assert _detect_default_model() == "claude-sonnet-4-6" def test_openai_key(self, monkeypatch): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.setenv("OPENAI_API_KEY", "sk-test") monkeypatch.delenv("GEMINI_API_KEY", raising=False) assert _detect_default_model() == "gpt-4o" def test_gemini_key(self, monkeypatch): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_KEY", raising=False) monkeypatch.setenv("GEMINI_API_KEY", "test") assert _detect_default_model() == "gemini/gemini-2.0-flash" def test_anthropic_preferred_over_openai(self, monkeypatch): monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test") monkeypatch.setenv("OPENAI_API_KEY", "sk-test") assert _detect_default_model() == "claude-sonnet-4-6" def test_no_keys_no_cli_raises(self, monkeypatch): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_KEY", raising=False) monkeypatch.delenv("GEMINI_API_KEY", raising=False) monkeypatch.setattr("headroom.learn.analyzer.shutil.which", lambda _name: None) with pytest.raises(RuntimeError, match="No LLM API key found"): _detect_default_model() def test_cli_fallback_claude(self, monkeypatch): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_KEY", raising=False) monkeypatch.delenv("GEMINI_API_KEY", raising=False) monkeypatch.setattr( "headroom.learn.analyzer.shutil.which", lambda name: f"/usr/bin/{name}" if name == "claude" else None, ) assert _detect_default_model() == "claude-cli" def test_cli_fallback_gemini(self, monkeypatch): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_KEY", raising=False) monkeypatch.delenv("GEMINI_API_KEY", raising=False) monkeypatch.setattr( "headroom.learn.analyzer.shutil.which", lambda name: f"/usr/bin/{name}" if name == "gemini" else None, ) assert _detect_default_model() == "gemini-cli" def test_cli_fallback_codex(self, monkeypatch): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_KEY", raising=False) monkeypatch.delenv("GEMINI_API_KEY", raising=False) monkeypatch.setattr( "headroom.learn.analyzer.shutil.which", lambda name: f"/usr/bin/{name}" if name == "codex" else None, ) assert _detect_default_model() == "codex-cli" def test_api_key_preferred_over_cli(self, monkeypatch): monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test") monkeypatch.setattr( "headroom.learn.analyzer.shutil.which", lambda name: f"/usr/bin/{name}" if name == "claude" else None, ) assert _detect_default_model() == "claude-sonnet-4-6" def test_env_var_selects_gemini(self, monkeypatch): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_KEY", raising=False) monkeypatch.delenv("GEMINI_API_KEY", raising=False) monkeypatch.setenv("HEADROOM_LEARN_CLI", "gemini") assert _detect_default_model() == "gemini-cli" def test_env_var_selects_codex(self, monkeypatch): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_KEY", raising=False) monkeypatch.delenv("GEMINI_API_KEY", raising=False) monkeypatch.setenv("HEADROOM_LEARN_CLI", "codex") assert _detect_default_model() == "codex-cli" def test_env_var_invalid_raises(self, monkeypatch): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_KEY", raising=False) monkeypatch.delenv("GEMINI_API_KEY", raising=False) monkeypatch.setenv("HEADROOM_LEARN_CLI", "unknown-tool") with pytest.raises(ValueError, match="not a supported CLI"): _detect_default_model() def test_api_key_preferred_over_env_var(self, monkeypatch): monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test") monkeypatch.setenv("HEADROOM_LEARN_CLI", "gemini") assert _detect_default_model() == "claude-sonnet-4-6" def test_env_var_preferred_over_auto_detect(self, monkeypatch): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_KEY", raising=False) monkeypatch.delenv("GEMINI_API_KEY", raising=False) monkeypatch.setenv("HEADROOM_LEARN_CLI", "codex") monkeypatch.setattr( "headroom.learn.analyzer.shutil.which", lambda name: f"/usr/bin/{name}" if name == "claude" else None, ) # codex selected via env var, even though claude is in PATH assert _detect_default_model() == "codex-cli" # ============================================================================= # CLI LLM Backend # ============================================================================= class TestStripFencedJson: def test_raw_json(self): result = _strip_fenced_json('{"key": "value"}') assert result == {"key": "value"} def test_fenced_json(self): raw = '```json\n{"key": "value"}\n```' result = _strip_fenced_json(raw) assert result == {"key": "value"} def test_fenced_no_language_tag(self): raw = '```\n{"key": "value"}\n```' result = _strip_fenced_json(raw) assert result == {"key": "value"} def test_whitespace_padding(self): raw = ' \n```json\n{"key": "value"}\n```\n ' result = _strip_fenced_json(raw) assert result == {"key": "value"} def test_invalid_json_raises(self): with pytest.raises(json.JSONDecodeError): _strip_fenced_json("not json at all") class TestCallCliLlm: @patch("headroom.learn.analyzer.subprocess.run") def test_claude_cli_success(self, mock_run: MagicMock): mock_run.return_value = MagicMock( returncode=0, stdout='{"context_file_rules": [], "memory_file_rules": []}', stderr="", ) result = _call_cli_llm("test digest", "claude-cli") assert result == {"context_file_rules": [], "memory_file_rules": []} mock_run.assert_called_once() cmd = mock_run.call_args[0][0] assert cmd == ["claude", "-p"] # Prompt passed via stdin, not as an argument assert mock_run.call_args.kwargs.get("input") is not None @patch("headroom.learn.analyzer.subprocess.run") def test_codex_cli_uses_exec(self, mock_run: MagicMock): mock_run.return_value = MagicMock( returncode=0, stdout='{"context_file_rules": [], "memory_file_rules": []}', stderr="", ) result = _call_cli_llm("test digest", "codex-cli") assert result == {"context_file_rules": [], "memory_file_rules": []} cmd = mock_run.call_args[0][0] assert cmd == ["codex", "exec"] @patch("headroom.learn.analyzer.subprocess.run") def test_gemini_cli_uses_p_flag(self, mock_run: MagicMock): mock_run.return_value = MagicMock( returncode=0, stdout='{"context_file_rules": [], "memory_file_rules": []}', stderr="", ) _call_cli_llm("test digest", "gemini-cli") cmd = mock_run.call_args[0][0] assert cmd == ["gemini", "-p"] @patch("headroom.learn.analyzer.subprocess.run") def test_cli_nonzero_exit_raises(self, mock_run: MagicMock): mock_run.return_value = MagicMock( returncode=1, stdout="", stderr="Error: auth required", ) with pytest.raises(RuntimeError, match="failed.*exit 1"): _call_cli_llm("test digest", "claude-cli") @patch("headroom.learn.analyzer.subprocess.run") def test_cli_stderr_truncated_in_error(self, mock_run: MagicMock): long_stderr = "x" * 5000 mock_run.return_value = MagicMock( returncode=1, stdout="", stderr=long_stderr, ) with pytest.raises(RuntimeError) as exc_info: _call_cli_llm("test digest", "claude-cli") # Full 5000-char stderr should not appear in the error message assert long_stderr not in str(exc_info.value) def test_unknown_cli_model_raises(self): with pytest.raises(ValueError, match="Unknown CLI model"): _call_cli_llm("test digest", "unknown-cli") @patch("headroom.learn.analyzer.subprocess.run") def test_fenced_output_parsed(self, mock_run: MagicMock): mock_run.return_value = MagicMock( returncode=0, stdout='```json\n{"context_file_rules": [], "memory_file_rules": []}\n```', stderr="", ) result = _call_cli_llm("test digest", "claude-cli") assert result == {"context_file_rules": [], "memory_file_rules": []} @patch("headroom.learn.analyzer.subprocess.run") def test_cli_not_installed_raises(self, mock_run: MagicMock): mock_run.side_effect = FileNotFoundError("No such file or directory: 'codex'") with pytest.raises(RuntimeError, match="not found in PATH"): _call_cli_llm("test digest", "codex-cli") @patch("headroom.learn.analyzer.subprocess.run") def test_timeout_raises_runtime_error(self, mock_run: MagicMock): mock_run.side_effect = subprocess.TimeoutExpired(cmd=["claude", "-p"], timeout=120) with pytest.raises(RuntimeError, match="did not respond within"): _call_cli_llm("test digest", "claude-cli") @patch("headroom.learn.analyzer.subprocess.run") def test_unparseable_output_raises_with_context(self, mock_run: MagicMock): mock_run.return_value = MagicMock( returncode=0, stdout="This is not JSON at all", stderr="", ) with pytest.raises(RuntimeError, match="unparseable output"): _call_cli_llm("test digest", "claude-cli") class TestCallLlmRouting: @patch("headroom.learn.analyzer._call_cli_llm") def test_routes_cli_model_to_cli_backend(self, mock_cli: MagicMock): mock_cli.return_value = {"context_file_rules": [], "memory_file_rules": []} result = _call_llm("test digest", "claude-cli") mock_cli.assert_called_once_with("test digest", "claude-cli") assert result == {"context_file_rules": [], "memory_file_rules": []} @patch("headroom.learn.analyzer._call_cli_llm") def test_routes_codex_cli(self, mock_cli: MagicMock): mock_cli.return_value = {} _call_llm("digest", "codex-cli") mock_cli.assert_called_once_with("digest", "codex-cli") # ============================================================================= # Legacy Compatibility # ============================================================================= class TestFailureAnalyzerCompat: @patch("headroom.learn.analyzer._call_llm") def test_legacy_alias_works(self, mock_call_llm: MagicMock): from headroom.learn.analyzer import FailureAnalyzer mock_call_llm.return_value = {"context_file_rules": [], "memory_file_rules": []} analyzer = FailureAnalyzer() result = analyzer.analyze(_project(), []) assert isinstance(result, AnalysisResult)