diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 1b271f23ef..98cc47dee0 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -190,6 +190,11 @@ jobs: continue-on-error: true run: .venv/bin/python scripts/check-reviewer-job-names.py + - name: Check LLM API boundary + id: llm-api-boundary + continue-on-error: true + run: .venv/bin/python scripts/check-llm-api-calls.py + - name: Check results if: always() run: | @@ -202,6 +207,7 @@ jobs: [ "${{ steps.workflow-secrets.outcome }}" = "failure" ] && failed="$failed workflow-secrets" [ "${{ steps.hardcoded-ports.outcome }}" = "failure" ] && failed="$failed hardcoded-ports" [ "${{ steps.reviewer-job-names.outcome }}" = "failure" ] && failed="$failed reviewer-job-names" + [ "${{ steps.llm-api-boundary.outcome }}" = "failure" ] && failed="$failed llm-api-boundary" if [ -n "$failed" ]; then echo "::error::Failed checks:$failed" exit 1 diff --git a/gateway/config_validator.py b/gateway/config_validator.py index a2562b2cb2..44fa341b47 100644 --- a/gateway/config_validator.py +++ b/gateway/config_validator.py @@ -86,7 +86,7 @@ def validate_config() -> None: ] if not domains: errors.append( - "Allowed domains file is empty (no domains configured)\n" + "Allowed domains file is empty (no domains configured)\n" # noqa: EGG200 - validation message, not an API call " At minimum, api.anthropic.com is required for private mode" ) except Exception as e: diff --git a/gateway/gateway.py b/gateway/gateway.py index b86160012f..faefee4a1e 100644 --- a/gateway/gateway.py +++ b/gateway/gateway.py @@ -3679,7 +3679,7 @@ def get_anthropic_client() -> httpx.Client: global _anthropic_client if _anthropic_client is None: _anthropic_client = httpx.Client( - base_url="https://api.anthropic.com", + base_url="https://api.anthropic.com", # noqa: EGG200 - gateway proxy client, not direct LLM call timeout=httpx.Timeout(120.0, connect=10.0), limits=httpx.Limits(max_connections=100, max_keepalive_connections=20), ) diff --git a/orchestrator/health_checks/tier2/agent_inspector.py b/orchestrator/health_checks/tier2/agent_inspector.py index 6a874860aa..416a1e4023 100644 --- a/orchestrator/health_checks/tier2/agent_inspector.py +++ b/orchestrator/health_checks/tier2/agent_inspector.py @@ -1,9 +1,11 @@ """ -AgentInspectorCheck — Tier 2 semantic health check using Claude API. +AgentInspectorCheck — Tier 2 semantic health check via sandbox container. -Sends a structured prompt with pipeline context (git log, diff stats, -agent outputs, contract state) to the Claude API and parses a JSON -verdict. Gracefully degrades to HEALTHY on API errors. +Serializes pipeline context and delegates LLM analysis to a short-lived +sandbox container running ``egg-health-inspect``. The orchestrator never +calls the Anthropic API directly — that happens inside the sandbox. + +Gracefully degrades to HEALTHY on container errors. """ from __future__ import annotations @@ -28,7 +30,6 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] return logging.getLogger(name) -import httpx from health_checks.context import PipelineHealthContext from health_checks.types import ( HealthAction, @@ -40,86 +41,8 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] logger = get_logger("orchestrator.health_checks.tier2.agent_inspector") -# Default model for agent inspection -_DEFAULT_MODEL = "claude-sonnet-4-20250514" -# API timeout in seconds -_API_TIMEOUT = 30 -# Max retries on transient failures -_MAX_RETRIES = 1 - -# System prompt for the inspector -_SYSTEM_PROMPT = """\ -You are a pipeline health inspector. Analyze the provided context about a \ -software engineering agent's work and produce a health verdict. - -Respond ONLY with a JSON object (no markdown fencing, no extra text): -{ - "status": "HEALTHY" | "DEGRADED" | "FAILED", - "reasoning": "" -} - -Rules: -- HEALTHY: Agent is making reasonable progress; commits present; no red flags. -- DEGRADED: Minor concerns — e.g. no recent commits, stale output, contract \ -tasks still pending after a long time. Use this when there's risk but not \ -certainty of failure. -- FAILED: Clear signs of stuck agent — e.g. repeated errors in outputs, \ -no commits and no drafts, contradictory state. -- Be concise. One paragraph for reasoning. -""" - - -def _build_user_prompt(context: PipelineHealthContext) -> str: - """Assemble the user prompt from pipeline context fields.""" - parts: list[str] = [] - - parts.append(f"Pipeline: {context.pipeline_id}") - parts.append(f"Phase: {context.current_phase.value}") - parts.append(f"Branch: {context.branch or 'unknown'}") - parts.append(f"Trigger: {context.trigger}") - parts.append("") - - # Git log - git_log = context.git_log - if git_log: - parts.append("## Recent Commits") - parts.append(git_log) - parts.append("") - - # Git diff stat - diff_stat = context.git_diff_stat - if diff_stat: - parts.append("## Diff Stats (vs main)") - parts.append(diff_stat) - parts.append("") - - # Agent outputs (summarize keys + truncated content) - outputs = context.agent_outputs - if outputs: - parts.append("## Agent Output Files") - for name, content in outputs.items(): - parts.append(f"### {name}") - # Cap individual output in the prompt to keep total manageable - parts.append(content[:2000]) - parts.append("") - else: - parts.append("## Agent Output Files") - parts.append("(none found)") - parts.append("") - - # Contract state - contract = context.contract - if contract: - parts.append("## Contract State") - # Serialize key fields, not the entire blob - summary: dict[str, Any] = {} - for key in ("current_phase", "acceptance_criteria", "decisions", "agent_executions"): - if key in contract: - summary[key] = contract[key] - parts.append(json.dumps(summary, indent=2, default=str)[:3000]) - parts.append("") - - return "\n".join(parts) +# Timeout for the inspector container (seconds) +_CONTAINER_TIMEOUT = 60 def _parse_verdict(text: str) -> tuple[HealthStatus, str]: @@ -160,79 +83,150 @@ def _parse_verdict(text: str) -> tuple[HealthStatus, str]: return status, reasoning -def _call_claude_api( - user_prompt: str, - *, - api_key: str | None = None, - base_url: str | None = None, - model: str | None = None, +def _serialize_context(context: PipelineHealthContext) -> dict[str, Any]: + """Serialize PipelineHealthContext into a JSON-safe dict for the inspector.""" + # Build contract summary (filtered keys) + contract = context.contract + contract_summary: dict[str, Any] = {} + if contract: + for key in ("current_phase", "acceptance_criteria", "decisions", "agent_executions"): + if key in contract: + contract_summary[key] = contract[key] + + return { + "pipeline_id": context.pipeline_id, + "current_phase": context.current_phase.value, + "branch": context.branch or "unknown", + "trigger": context.trigger, + "git_log": context.git_log, + "git_diff_stat": context.git_diff_stat, + "agent_outputs": context.agent_outputs, + "contract_summary": contract_summary, + } + + +def _run_inspector_container( + context_payload: dict[str, Any], + pipeline_id: str, ) -> str: - """Call the Anthropic Messages API via httpx. + """Spawn a sandbox container to run the inspector script. - Returns the text content of the first response block. - Raises on HTTP or timeout errors (caller handles graceful degradation). + Context is passed via the EGG_INSPECTOR_CONTEXT env var (JSON string) + to avoid file-mount complexity. Aggregate payload size is validated + against a 100KB limit before passing to the container. + + Returns the raw response text from the container's stdout. + Raises on container spawn/wait/parse failures (caller handles graceful degradation). """ - key = api_key or os.environ.get("ANTHROPIC_API_KEY", "") - url = (base_url or os.environ.get("ANTHROPIC_BASE_URL", "https://api.anthropic.com")).rstrip( - "/" + import uuid + + from container_spawner import ContainerSpawner, ContainerSpawnError + from docker_client import DockerClient, DockerClientError, get_docker_client + from models import AgentRole + + context_json = json.dumps(context_payload, default=str) + + # Guard against oversized payloads that could exceed Docker env var limits (~131KB). + # Individual fields are truncated, but aggregate size is not otherwise bounded. + max_context_bytes = 100_000 # 100KB safety margin + if len(context_json) > max_context_bytes: + raise ValueError( + f"Inspector context too large for env var: {len(context_json)} bytes " + f"(limit: {max_context_bytes} bytes)" + ) + + docker: DockerClient = get_docker_client() + spawner = ContainerSpawner(docker_client=docker) + + # Use a unique suffix to avoid container name collisions when concurrent + # health checks run for the same pipeline (e.g., PHASE_COMPLETE and + # ON_DEMAND triggered simultaneously). + unique_suffix = uuid.uuid4().hex[:8] + + # Spawn the inspector container — pass context via env var directly + spawned = spawner.spawn_agent_container( + pipeline_id=f"{pipeline_id}-inspect-{unique_suffix}", + agent_role=AgentRole.INSPECTOR, + mode=os.environ.get("EGG_NETWORK_MODE", "public"), + extra_env={ + "EGG_INSPECTOR_CONTEXT": context_json, + }, + command=["python3", "/home/egg/sandbox/bin/egg-health-inspect"], + wait_for_gateway=False, + repo_volumes={}, ) - mdl = model or os.environ.get("HEALTH_CHECK_MODEL", _DEFAULT_MODEL) - headers = { - "x-api-key": key, - "anthropic-version": "2023-06-01", - "content-type": "application/json", - } + container_id = spawned.container_info.container_id - payload = { - "model": mdl, - "max_tokens": 512, - "system": _SYSTEM_PROMPT, - "messages": [{"role": "user", "content": user_prompt}], - } + try: + # Wait for the container to exit + exit_info = docker.wait_for_container( + container_id, + timeout=_CONTAINER_TIMEOUT, + ) + + # Get the container logs (stdout contains the JSON verdict) + logs = docker.get_container_logs(container_id, tail=50) + + if exit_info.exit_code != 0: + raise RuntimeError( + f"Inspector container exited with code {exit_info.exit_code}: " + f"{logs[-500:] if logs else '(no logs)'}" + ) + + return logs - last_exc: Exception | None = None - for attempt in range(_MAX_RETRIES + 1): + finally: + # Clean up the container try: - resp = httpx.post( - f"{url}/v1/messages", - headers=headers, - json=payload, - timeout=_API_TIMEOUT, + spawner.remove_agent_container( + container_id, + force=True, + cleanup_session=True, ) - resp.raise_for_status() - body = resp.json() - # Extract text from first content block - content_blocks = body.get("content", []) - if content_blocks and isinstance(content_blocks, list): - return str(content_blocks[0].get("text", "")) - return "" - except (httpx.TimeoutException, httpx.HTTPStatusError) as exc: - last_exc = exc - if attempt < _MAX_RETRIES: - logger.warning( - "Claude API attempt failed, retrying", - attempt=attempt + 1, - error=str(exc), - ) + except (DockerClientError, ContainerSpawnError): + pass # Best effort cleanup + + +def _parse_container_output(logs: str) -> str: + """Extract the raw_response from container stdout JSON. + + The container outputs a JSON object like ``{"raw_response": "..."}``. + We extract the raw_response field for verdict parsing. + + Docker log timestamps (``2024-01-01T00:00:00.000000000Z ``) + are stripped using a heuristic: if a line starts with a digit and + contains a space before a ``{``, we treat the part after the first + space as the payload. If the timestamp format changes, the fallback + (trying the raw line as JSON) still applies. + """ + # Container logs may have timestamps prefixed — find the JSON line + for line in reversed(logs.strip().splitlines()): + # Strip Docker timestamp prefix if present (format: 2024-01-01T00:00:00.000000000Z ...) + stripped = line.strip() + if " " in stripped and stripped[0].isdigit(): + # Try stripping timestamp prefix + _, _, after = stripped.partition(" ") + if after.startswith("{"): + stripped = after + + if stripped.startswith("{"): + try: + data = json.loads(stripped) + return str(data.get("raw_response", "")) + except (json.JSONDecodeError, ValueError): continue - raise - except Exception: - raise - # Should not reach here, but satisfy type checker - if last_exc: - raise last_exc - return "" + raise ValueError(f"No valid JSON found in container output: {logs[-300:]}") class AgentInspectorCheck: - """Tier 2 health check that uses Claude to semantically inspect agent state. + """Tier 2 health check that delegates LLM analysis to a sandbox container. - Sends pipeline context (git log, diff stats, agent outputs, contract) - to the Claude API and interprets the structured verdict. + Serializes pipeline context and spawns a short-lived container running + ``egg-health-inspect``, which calls the Claude API from inside the sandbox. - On API failure, gracefully degrades to HEALTHY with a warning — + On container failure, gracefully degrades to HEALTHY with a warning — Tier 2 failures should never block the pipeline. """ @@ -247,10 +241,21 @@ class AgentInspectorCheck: ) def run(self, context: PipelineHealthContext) -> HealthResult: - """Execute the agent inspection check.""" + """Execute the agent inspection check via sandbox container.""" try: - user_prompt = _build_user_prompt(context) - response_text = _call_claude_api(user_prompt) + # Serialize context for the inspector container + context_payload = _serialize_context(context) + + # Spawn container and get response + logs = _run_inspector_container( + context_payload=context_payload, + pipeline_id=context.pipeline_id, + ) + + # Parse container output to get raw Claude response + response_text = _parse_container_output(logs) + + # Parse the verdict from Claude's response status, reasoning = _parse_verdict(response_text) logger.info( @@ -271,9 +276,9 @@ def run(self, context: PipelineHealthContext) -> HealthResult: ) except Exception as exc: - # Graceful degradation: API failure should not block pipeline + # Graceful degradation: container failure should not block pipeline logger.warning( - "Agent inspector API call failed, degrading gracefully", + "Agent inspector container failed, degrading gracefully", error=str(exc), pipeline=context.pipeline_id, ) diff --git a/orchestrator/models.py b/orchestrator/models.py index da5ff19cb4..f5ed50ef9d 100644 --- a/orchestrator/models.py +++ b/orchestrator/models.py @@ -76,6 +76,8 @@ class AgentRole(StrEnum): RISK_ANALYST = "risk_analyst" # Refine-phase roles REFINER = "refiner" + # Health check roles + INSPECTOR = "inspector" # Reviewer roles (specific subtypes) REVIEWER_CODE = "reviewer_code" REVIEWER_CONTRACT = "reviewer_contract" diff --git a/orchestrator/tests/test_health_check_tier2.py b/orchestrator/tests/test_health_check_tier2.py index 77f1f80881..9fca6ba24c 100644 --- a/orchestrator/tests/test_health_check_tier2.py +++ b/orchestrator/tests/test_health_check_tier2.py @@ -2,11 +2,11 @@ Covers: - Context assembly: contract loading, truncation, lazy evaluation -- Prompt construction: all fields present, caps respected - Verdict parsing: valid JSON, malformed JSON, missing fields, code fences -- AgentInspectorCheck: healthy/degraded/failed verdicts, API timeout, - API errors, graceful degradation, event emission +- AgentInspectorCheck: healthy/degraded/failed verdicts, container timeout, + container errors, graceful degradation, event emission - Escalation logic: runner integration with Tier 2 +- Container delegation: _serialize_context, _parse_container_output """ import json @@ -31,9 +31,9 @@ from health_checks.runner import HealthCheckRunner from health_checks.tier2.agent_inspector import ( AgentInspectorCheck, - _build_user_prompt, - _call_claude_api, + _parse_container_output, _parse_verdict, + _serialize_context, ) from health_checks.types import ( HealthAction, @@ -86,23 +86,6 @@ def _make_context( ) -def _mock_httpx_response(status_code: int = 200, json_body: dict | None = None): - """Create a mock httpx Response.""" - resp = MagicMock() - resp.status_code = status_code - resp.raise_for_status = MagicMock() - if status_code >= 400: - import httpx - - resp.raise_for_status.side_effect = httpx.HTTPStatusError( - f"HTTP {status_code}", - request=MagicMock(), - response=resp, - ) - resp.json.return_value = json_body or {} - return resp - - # =========================================================================== # 1. Context Assembly Tests # =========================================================================== @@ -273,120 +256,7 @@ def test_reads_from_agent_outputs_subdir(self, tmp_path): # =========================================================================== -# 2. Prompt Construction Tests -# =========================================================================== - - -class TestBuildUserPrompt: - """Tests for _build_user_prompt.""" - - def test_includes_pipeline_metadata(self, tmp_path): - ctx = _make_context(repo_path=tmp_path) - prompt = _build_user_prompt(ctx) - assert "issue-99" in prompt - assert "implement" in prompt - assert "egg/issue-99" in prompt - - def test_includes_git_log(self, tmp_path): - ctx = _make_context(repo_path=tmp_path) - with patch.object(ctx, "_run_git", return_value="abc1234 Add feature"): - # Force lazy loading - _ = ctx.git_log - prompt = _build_user_prompt(ctx) - assert "abc1234 Add feature" in prompt - - def test_includes_diff_stat(self, tmp_path): - ctx = _make_context(repo_path=tmp_path) - with patch.object(ctx, "_run_git", return_value="file.py | 5 ++---"): - # Force lazy loading for diff stat - _ = ctx.git_diff_stat - prompt = _build_user_prompt(ctx) - assert "file.py" in prompt - - def test_includes_agent_outputs(self, tmp_path): - state_dir = tmp_path / ".egg-state" / "drafts" - state_dir.mkdir(parents=True) - (state_dir / "plan.md").write_text("# Implementation Plan\nStep 1: Do things") - - pipeline = Pipeline( - id="issue-99", - issue_number=99, - repo=None, - branch="egg/issue-99", - mode="issue", - status=PipelineStatus.RUNNING, - current_phase=PipelinePhase.IMPLEMENT, - ) - ctx = PipelineHealthContext( - pipeline=pipeline, - repo_path=tmp_path, - trigger="phase_complete", - ) - prompt = _build_user_prompt(ctx) - assert "plan.md" in prompt - assert "Implementation Plan" in prompt - - def test_includes_contract_state(self, tmp_path): - state_dir = tmp_path / ".egg-state" / "contracts" - state_dir.mkdir(parents=True) - contract = { - "current_phase": "implement", - "acceptance_criteria": ["All tests pass"], - "agent_executions": [{"role": "coder", "status": "complete"}], - } - (state_dir / "99.json").write_text(json.dumps(contract)) - - pipeline = Pipeline( - id="issue-99", - issue_number=99, - repo=None, - branch="egg/issue-99", - mode="issue", - status=PipelineStatus.RUNNING, - current_phase=PipelinePhase.IMPLEMENT, - ) - ctx = PipelineHealthContext( - pipeline=pipeline, - repo_path=tmp_path, - trigger="phase_complete", - ) - prompt = _build_user_prompt(ctx) - assert "Contract State" in prompt - assert "implement" in prompt - - def test_empty_outputs_shows_none_found(self, tmp_path): - ctx = _make_context(repo_path=tmp_path) - prompt = _build_user_prompt(ctx) - assert "(none found)" in prompt - - def test_output_content_capped_in_prompt(self, tmp_path): - state_dir = tmp_path / ".egg-state" / "drafts" - state_dir.mkdir(parents=True) - (state_dir / "huge.md").write_text("x" * 5000) - - pipeline = Pipeline( - id="issue-99", - issue_number=99, - repo=None, - branch="egg/issue-99", - mode="issue", - status=PipelineStatus.RUNNING, - current_phase=PipelinePhase.IMPLEMENT, - ) - ctx = PipelineHealthContext( - pipeline=pipeline, - repo_path=tmp_path, - trigger="phase_complete", - ) - prompt = _build_user_prompt(ctx) - # Content in prompt should be capped at 2000 chars per output - lines_with_x = [line for line in prompt.split("\n") if "x" * 100 in line] - for line in lines_with_x: - assert len(line) <= 2001 # some tolerance - - -# =========================================================================== -# 3. Verdict Parsing Tests +# 2. Verdict Parsing Tests # =========================================================================== @@ -459,134 +329,126 @@ def test_lowercase_status_handled(self): # =========================================================================== -# 4. Claude API Call Tests +# 4. Container Output Parsing Tests # =========================================================================== -class TestCallClaudeApi: - """Tests for _call_claude_api with mocked httpx.""" - - @patch("health_checks.tier2.agent_inspector.httpx") - def test_successful_api_call(self, mock_httpx): - """Successful API call returns text content.""" - mock_resp = _mock_httpx_response( - 200, - {"content": [{"type": "text", "text": '{"status": "HEALTHY", "reasoning": "OK"}'}]}, - ) - mock_httpx.post.return_value = mock_resp +class TestParseContainerOutput: + """Tests for _parse_container_output.""" - result = _call_claude_api("test prompt", api_key="sk-ant-test123") + def test_simple_json_output(self): + """Parses plain JSON output from container.""" + logs = '{"raw_response": "{\\"status\\": \\"HEALTHY\\", \\"reasoning\\": \\"OK\\"}"}\n' + result = _parse_container_output(logs) assert '"HEALTHY"' in result - mock_httpx.post.assert_called_once() - - @patch("health_checks.tier2.agent_inspector.httpx") - def test_api_sends_correct_headers(self, mock_httpx): - """API call sends correct Anthropic headers.""" - mock_resp = _mock_httpx_response(200, {"content": [{"type": "text", "text": "ok"}]}) - mock_httpx.post.return_value = mock_resp - _call_claude_api("test", api_key="sk-ant-key", base_url="https://custom.api.com") - - call_args = mock_httpx.post.call_args - assert call_args[0][0] == "https://custom.api.com/v1/messages" - headers = call_args[1]["headers"] - assert headers["x-api-key"] == "sk-ant-key" - assert headers["anthropic-version"] == "2023-06-01" - - @patch("health_checks.tier2.agent_inspector.httpx") - def test_api_sends_correct_model(self, mock_httpx): - """API call uses specified model.""" - mock_resp = _mock_httpx_response(200, {"content": [{"type": "text", "text": "ok"}]}) - mock_httpx.post.return_value = mock_resp + def test_json_with_docker_timestamp(self): + """Parses JSON with Docker timestamp prefix.""" + logs = '2024-01-01T00:00:00.000000000Z {"raw_response": "verdict text"}\n' + result = _parse_container_output(logs) + assert result == "verdict text" + + def test_multiline_logs_finds_json(self): + """Finds JSON line among other log output.""" + logs = ( + "2024-01-01T00:00:00Z Starting inspector...\n" + "2024-01-01T00:00:01Z Processing context\n" + '2024-01-01T00:00:02Z {"raw_response": "the verdict"}\n' + ) + result = _parse_container_output(logs) + assert result == "the verdict" - _call_claude_api("test", api_key="sk-ant-key", model="claude-sonnet-4-20250514") + def test_no_json_raises_value_error(self): + """Raises ValueError when no JSON found.""" + import pytest - payload = mock_httpx.post.call_args[1]["json"] - assert payload["model"] == "claude-sonnet-4-20250514" + logs = "Just some log lines\nNo JSON here\n" + with pytest.raises(ValueError, match="No valid JSON found"): + _parse_container_output(logs) - @patch("health_checks.tier2.agent_inspector.httpx") - def test_api_timeout_raises(self, mock_httpx): - """Timeout exception propagates after retries.""" - import httpx - import pytest + def test_empty_raw_response(self): + """Returns empty string for empty raw_response.""" + logs = '{"raw_response": ""}\n' + result = _parse_container_output(logs) + assert result == "" - mock_httpx.TimeoutException = httpx.TimeoutException - mock_httpx.HTTPStatusError = httpx.HTTPStatusError - mock_httpx.post.side_effect = httpx.TimeoutException("Connection timed out") - with pytest.raises(httpx.TimeoutException): - _call_claude_api("test", api_key="sk-ant-key") - # Should have retried once (2 calls total) - assert mock_httpx.post.call_count == 2 +# =========================================================================== +# 5. Context Serialization Tests +# =========================================================================== - @patch("health_checks.tier2.agent_inspector.httpx") - def test_api_http_error_raises(self, mock_httpx): - """HTTP 500 errors propagate after retries.""" - import httpx - import pytest - mock_httpx.TimeoutException = httpx.TimeoutException - mock_httpx.HTTPStatusError = httpx.HTTPStatusError +class TestSerializeContext: + """Tests for _serialize_context.""" - mock_resp = _mock_httpx_response(500) - mock_httpx.post.return_value = mock_resp + def test_basic_fields(self, tmp_path): + """Serialized context includes all basic fields.""" + ctx = _make_context(repo_path=tmp_path) + payload = _serialize_context(ctx) - with pytest.raises(httpx.HTTPStatusError): - _call_claude_api("test", api_key="sk-ant-key") - # Should have retried once (2 calls total) - assert mock_httpx.post.call_count == 2 + assert payload["pipeline_id"] == "issue-99" + assert payload["current_phase"] == "implement" + assert payload["branch"] == "egg/issue-99" + assert payload["trigger"] == "phase_complete" - @patch("health_checks.tier2.agent_inspector.httpx") - def test_api_empty_content_blocks(self, mock_httpx): - """Empty content blocks return empty string.""" - mock_resp = _mock_httpx_response(200, {"content": []}) - mock_httpx.post.return_value = mock_resp + def test_contract_summary_filters_keys(self, tmp_path): + """Only specific contract keys appear in the serialized summary.""" + state_dir = tmp_path / ".egg-state" / "contracts" + state_dir.mkdir(parents=True) + contract = { + "current_phase": "implement", + "acceptance_criteria": ["Tests pass"], + "decisions": [], + "agent_executions": [], + "schema_version": "1.0", + "internal_stuff": "excluded", + } + (state_dir / "99.json").write_text(json.dumps(contract)) - result = _call_claude_api("test", api_key="sk-ant-key") - assert result == "" + pipeline = Pipeline( + id="issue-99", + issue_number=99, + repo=None, + branch="egg/issue-99", + mode="issue", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.IMPLEMENT, + ) + ctx = PipelineHealthContext( + pipeline=pipeline, + repo_path=tmp_path, + trigger="phase_complete", + ) + payload = _serialize_context(ctx) + summary = payload["contract_summary"] - @patch("health_checks.tier2.agent_inspector.httpx") - def test_api_uses_env_vars(self, mock_httpx): - """Falls back to env vars for config.""" - mock_resp = _mock_httpx_response(200, {"content": [{"type": "text", "text": "ok"}]}) - mock_httpx.post.return_value = mock_resp - - with patch.dict( - "os.environ", - { - "ANTHROPIC_API_KEY": "sk-ant-env-key", - "ANTHROPIC_BASE_URL": "https://env.api.com", - "HEALTH_CHECK_MODEL": "claude-3-haiku-20240307", - }, - ): - _call_claude_api("test") - - call_args = mock_httpx.post.call_args - assert call_args[0][0] == "https://env.api.com/v1/messages" - assert call_args[1]["headers"]["x-api-key"] == "sk-ant-env-key" - assert call_args[1]["json"]["model"] == "claude-3-haiku-20240307" - - @patch("health_checks.tier2.agent_inspector.httpx") - def test_retry_then_success(self, mock_httpx): - """First attempt fails, retry succeeds.""" - import httpx - - mock_httpx.TimeoutException = httpx.TimeoutException - mock_httpx.HTTPStatusError = httpx.HTTPStatusError - - success_resp = _mock_httpx_response(200, {"content": [{"type": "text", "text": "success"}]}) - mock_httpx.post.side_effect = [ - httpx.TimeoutException("timeout"), - success_resp, - ] + assert "current_phase" in summary + assert "acceptance_criteria" in summary + assert "schema_version" not in summary + assert "internal_stuff" not in summary - result = _call_claude_api("test", api_key="sk-ant-key") - assert result == "success" - assert mock_httpx.post.call_count == 2 + def test_branch_none_becomes_unknown(self, tmp_path): + """None branch is serialized as 'unknown'.""" + pipeline = Pipeline( + id="test", + issue_number=1, + repo=None, + branch=None, + mode="issue", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.IMPLEMENT, + ) + ctx = PipelineHealthContext( + pipeline=pipeline, + repo_path=tmp_path, + trigger="on_demand", + ) + payload = _serialize_context(ctx) + assert payload["branch"] == "unknown" # =========================================================================== -# 5. AgentInspectorCheck Integration Tests +# 6. AgentInspectorCheck Integration Tests (container delegation) # =========================================================================== @@ -603,9 +465,9 @@ def test_conforms_to_health_check_protocol(self): assert HealthTrigger.PHASE_COMPLETE in check.triggers assert HealthTrigger.ON_DEMAND in check.triggers - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_healthy_verdict_returns_healthy(self, mock_api, tmp_path): - mock_api.return_value = '{"status": "HEALTHY", "reasoning": "Everything looks fine."}' + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_healthy_verdict_returns_healthy(self, mock_run, tmp_path): + mock_run.return_value = '{"raw_response": "{\\"status\\": \\"HEALTHY\\", \\"reasoning\\": \\"Everything looks fine.\\"}"}\n' check = AgentInspectorCheck() ctx = _make_context(repo_path=tmp_path) @@ -617,9 +479,9 @@ def test_healthy_verdict_returns_healthy(self, mock_api, tmp_path): assert result.action == HealthAction.CONTINUE assert "Everything looks fine" in result.reasoning - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_degraded_verdict_returns_alert(self, mock_api, tmp_path): - mock_api.return_value = '{"status": "DEGRADED", "reasoning": "Stale output files."}' + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_degraded_verdict_returns_alert(self, mock_run, tmp_path): + mock_run.return_value = '{"raw_response": "{\\"status\\": \\"DEGRADED\\", \\"reasoning\\": \\"Stale output files.\\"}"}\n' check = AgentInspectorCheck() result = check.run(_make_context(repo_path=tmp_path)) @@ -628,10 +490,10 @@ def test_degraded_verdict_returns_alert(self, mock_api, tmp_path): assert result.action == HealthAction.ALERT assert "Stale output" in result.reasoning - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_failed_verdict_returns_alert_not_fail_pipeline(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_failed_verdict_returns_alert_not_fail_pipeline(self, mock_run, tmp_path): """FAILED verdict returns ALERT action (not FAIL_PIPELINE per design).""" - mock_api.return_value = '{"status": "FAILED", "reasoning": "Agent is stuck."}' + mock_run.return_value = '{"raw_response": "{\\"status\\": \\"FAILED\\", \\"reasoning\\": \\"Agent is stuck.\\"}"}\n' check = AgentInspectorCheck() result = check.run(_make_context(repo_path=tmp_path)) @@ -640,12 +502,10 @@ def test_failed_verdict_returns_alert_not_fail_pipeline(self, mock_api, tmp_path assert result.action == HealthAction.ALERT # NOT FAIL_PIPELINE assert "stuck" in result.reasoning - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_api_timeout_graceful_degradation(self, mock_api, tmp_path): - """API timeout results in HEALTHY with warning (graceful degradation).""" - import httpx - - mock_api.side_effect = httpx.TimeoutException("Connection timed out") + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_container_timeout_graceful_degradation(self, mock_run, tmp_path): + """Container timeout results in HEALTHY with warning (graceful degradation).""" + mock_run.side_effect = RuntimeError("Wait failed: timeout") check = AgentInspectorCheck() result = check.run(_make_context(repo_path=tmp_path)) @@ -655,16 +515,10 @@ def test_api_timeout_graceful_degradation(self, mock_api, tmp_path): assert "unavailable" in result.reasoning assert result.details.get("graceful_degradation") is True - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_api_http_error_graceful_degradation(self, mock_api, tmp_path): - """HTTP error results in HEALTHY with warning.""" - import httpx - - mock_api.side_effect = httpx.HTTPStatusError( - "500 Internal Server Error", - request=MagicMock(), - response=MagicMock(status_code=500), - ) + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_container_spawn_error_graceful_degradation(self, mock_run, tmp_path): + """Container spawn error results in HEALTHY with warning.""" + mock_run.side_effect = Exception("Failed to spawn container: gateway unhealthy") check = AgentInspectorCheck() result = check.run(_make_context(repo_path=tmp_path)) @@ -674,32 +528,34 @@ def test_api_http_error_graceful_degradation(self, mock_api, tmp_path): assert "unavailable" in result.reasoning assert "error" in result.details - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_malformed_response_graceful_healthy(self, mock_api, tmp_path): - """Malformed API response defaults to HEALTHY.""" - mock_api.return_value = "This is not JSON at all" + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_malformed_container_output_graceful(self, mock_run, tmp_path): + """Malformed container output triggers graceful degradation.""" + mock_run.return_value = "Not JSON at all\nJust garbage\n" check = AgentInspectorCheck() result = check.run(_make_context(repo_path=tmp_path)) + # _parse_container_output raises ValueError → caught by outer except assert result.status == HealthStatus.HEALTHY - assert "Could not parse" in result.reasoning + assert result.details.get("graceful_degradation") is True - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_empty_response_graceful_healthy(self, mock_api, tmp_path): - """Empty API response defaults to HEALTHY.""" - mock_api.return_value = "" + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_container_nonzero_exit_graceful(self, mock_run, tmp_path): + """Container exiting non-zero triggers graceful degradation.""" + mock_run.side_effect = RuntimeError("Inspector container exited with code 1") check = AgentInspectorCheck() result = check.run(_make_context(repo_path=tmp_path)) assert result.status == HealthStatus.HEALTHY + assert result.details.get("graceful_degradation") is True - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_result_includes_raw_response(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_result_includes_raw_response(self, mock_run, tmp_path): """Result details include raw_response for debugging.""" raw = '{"status": "HEALTHY", "reasoning": "OK"}' - mock_api.return_value = raw + mock_run.return_value = '{"raw_response": ' + json.dumps(raw) + "}\n" check = AgentInspectorCheck() result = check.run(_make_context(repo_path=tmp_path)) @@ -707,10 +563,12 @@ def test_result_includes_raw_response(self, mock_api, tmp_path): assert "raw_response" in result.details assert result.details["raw_response"] == raw - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_result_serializes_to_dict(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_result_serializes_to_dict(self, mock_run, tmp_path): """Result to_dict() contains all required fields for SSE.""" - mock_api.return_value = '{"status": "DEGRADED", "reasoning": "Concern."}' + mock_run.return_value = ( + '{"raw_response": "{\\"status\\": \\"DEGRADED\\", \\"reasoning\\": \\"Concern.\\"}"}\n' + ) check = AgentInspectorCheck() result = check.run(_make_context(repo_path=tmp_path)) @@ -723,10 +581,10 @@ def test_result_serializes_to_dict(self, mock_api, tmp_path): assert d["action"] == "alert" assert "timestamp" in d - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_generic_exception_graceful_degradation(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_generic_exception_graceful_degradation(self, mock_run, tmp_path): """Any unexpected exception results in HEALTHY with warning.""" - mock_api.side_effect = RuntimeError("Unexpected error") + mock_run.side_effect = RuntimeError("Unexpected error") check = AgentInspectorCheck() result = check.run(_make_context(repo_path=tmp_path)) @@ -738,7 +596,7 @@ def test_generic_exception_graceful_degradation(self, mock_api, tmp_path): # =========================================================================== -# 6. Escalation Logic Tests (Runner + Tier 2) +# 7. Escalation Logic Tests (Runner + Tier 2) # =========================================================================== @@ -950,17 +808,19 @@ def run(self, context): # =========================================================================== -# 7. Event Emission Tests +# 8. Event Emission Tests # =========================================================================== class TestEventEmission: """Tests that Tier 2 results are emitted to EventBus for SSE.""" - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_tier2_events_emitted(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_tier2_events_emitted(self, mock_run, tmp_path): """Tier 2 check results are emitted via EventBus.""" - mock_api.return_value = '{"status": "DEGRADED", "reasoning": "Concern."}' + mock_run.return_value = ( + '{"raw_response": "{\\"status\\": \\"DEGRADED\\", \\"reasoning\\": \\"Concern.\\"}"}\n' + ) runner = HealthCheckRunner() runner.register(_AlwaysHealthyTier1()) @@ -997,10 +857,12 @@ def test_tier2_events_emitted(self, mock_api, tmp_path): assert "reasoning" in payload assert payload["action"] == "alert" - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_aggregate_event_includes_tier2(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_aggregate_event_includes_tier2(self, mock_run, tmp_path): """Aggregate COMPLETED event includes Tier 2 results.""" - mock_api.return_value = '{"status": "HEALTHY", "reasoning": "OK."}' + mock_run.return_value = ( + '{"raw_response": "{\\"status\\": \\"HEALTHY\\", \\"reasoning\\": \\"OK.\\"}"}\n' + ) runner = HealthCheckRunner() runner.register(_AlwaysHealthyTier1()) @@ -1031,17 +893,17 @@ def test_aggregate_event_includes_tier2(self, mock_api, tmp_path): # =========================================================================== -# 8. Edge Cases +# 9. Edge Cases # =========================================================================== class TestEdgeCases: """Edge case tests for Tier 2 health checks.""" - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_no_api_key_graceful_degradation(self, mock_api, tmp_path): - """Missing API key still triggers graceful degradation.""" - mock_api.side_effect = Exception("No API key configured") + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_container_failure_graceful_degradation(self, mock_run, tmp_path): + """Container failure still triggers graceful degradation.""" + mock_run.side_effect = Exception("Container spawn failed") check = AgentInspectorCheck() result = check.run(_make_context(repo_path=tmp_path)) @@ -1069,12 +931,12 @@ def test_context_with_no_repo(self, tmp_path): assert ctx.agent_outputs == {} assert ctx.contract == {} - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_concurrent_safety_multiple_runs(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_concurrent_safety_multiple_runs(self, mock_run, tmp_path): """Multiple runs of the check don't share state unsafely.""" - mock_api.side_effect = [ - '{"status": "HEALTHY", "reasoning": "Run 1 OK."}', - '{"status": "DEGRADED", "reasoning": "Run 2 concern."}', + mock_run.side_effect = [ + '{"raw_response": "{\\"status\\": \\"HEALTHY\\", \\"reasoning\\": \\"Run 1 OK.\\"}"}\n', + '{"raw_response": "{\\"status\\": \\"DEGRADED\\", \\"reasoning\\": \\"Run 2 concern.\\"}"}\n', ] check = AgentInspectorCheck() @@ -1085,11 +947,11 @@ def test_concurrent_safety_multiple_runs(self, mock_api, tmp_path): assert r2.status == HealthStatus.DEGRADED assert r1 is not r2 - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_raw_response_truncated_in_details(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_raw_response_truncated_in_details(self, mock_run, tmp_path): """Raw response in details is capped at 500 chars.""" - long_response = '{"status": "HEALTHY", "reasoning": "' + "x" * 1000 + '"}' - mock_api.return_value = long_response + long_verdict = '{"status": "HEALTHY", "reasoning": "' + "x" * 1000 + '"}' + mock_run.return_value = '{"raw_response": ' + json.dumps(long_verdict) + "}\n" check = AgentInspectorCheck() result = check.run(_make_context(repo_path=tmp_path)) diff --git a/orchestrator/tests/test_health_check_tier2_tester.py b/orchestrator/tests/test_health_check_tier2_tester.py index 92e247ae80..91fadcff36 100644 --- a/orchestrator/tests/test_health_check_tier2_tester.py +++ b/orchestrator/tests/test_health_check_tier2_tester.py @@ -4,10 +4,7 @@ - Context assembly: multi-subdir agent_outputs, file extension filtering, per-file cap, git_log/git_diff_stat laziness, _run_git repo resolution -- Prompt construction: trigger field, contract key filtering, contract cap, - branch=None -- API call: non-retryable exceptions, trailing-slash URL, system prompt in - payload, max_tokens, default model +- Container delegation: container output parsing, context serialization - Event emission: per-status event types, bus=None no-op, runner catches Tier 2 exceptions - Route integration: pipeline health check endpoint invokes Tier 2 @@ -15,7 +12,6 @@ """ import json -import os import sys from pathlib import Path from unittest.mock import MagicMock, patch @@ -39,9 +35,9 @@ from health_checks.runner import HealthCheckRunner, worst_action from health_checks.tier2.agent_inspector import ( AgentInspectorCheck, - _build_user_prompt, - _call_claude_api, + _parse_container_output, _parse_verdict, + _serialize_context, ) from health_checks.types import ( HealthAction, @@ -365,121 +361,7 @@ def test_live_container_ids_empty_on_docker_error(self, tmp_path): # =========================================================================== -# 4. Prompt Construction — extended coverage -# =========================================================================== - - -class TestBuildUserPromptExtended: - """Extended tests for _build_user_prompt covering gaps.""" - - def test_trigger_field_in_prompt(self, tmp_path): - """Prompt includes the trigger value.""" - ctx = _ctx(repo_path=tmp_path, trigger="wave_complete") - prompt = _build_user_prompt(ctx) - assert "wave_complete" in prompt - - def test_branch_none_shows_unknown(self, tmp_path): - """When pipeline.branch is None, prompt shows 'unknown'.""" - pipeline = _pipeline(branch=None) - ctx = _ctx(pipeline=pipeline, repo_path=tmp_path) - prompt = _build_user_prompt(ctx) - assert "unknown" in prompt - - def test_contract_key_filtering(self, tmp_path): - """Only specific contract keys appear in the Contract State section.""" - state_dir = tmp_path / ".egg-state" / "contracts" - state_dir.mkdir(parents=True) - contract = { - "current_phase": "implement", - "acceptance_criteria": ["Tests pass"], - "decisions": [{"q": "Which DB?"}], - "agent_executions": [{"role": "coder"}], - "schema_version": "1.0", - "internal_secret": "should-not-appear", - } - (state_dir / "42.json").write_text(json.dumps(contract)) - - pipeline = _pipeline(repo=None) - ctx = PipelineHealthContext( - pipeline=pipeline, - repo_path=tmp_path, - trigger="phase_complete", - ) - prompt = _build_user_prompt(ctx) - - # Extract just the Contract State section (after "## Contract State") - contract_idx = prompt.index("## Contract State") - contract_section = prompt[contract_idx:] - - # Included keys in Contract State section - assert "current_phase" in contract_section - assert "acceptance_criteria" in contract_section - assert "decisions" in contract_section - assert "agent_executions" in contract_section - # Excluded keys should NOT be in the Contract State section - # (they may appear in Agent Output Files which shows raw file content) - assert "internal_secret" not in contract_section - assert "schema_version" not in contract_section - - def test_contract_content_capped_at_3000_in_prompt(self, tmp_path): - """Contract JSON in prompt is capped at 3000 chars.""" - state_dir = tmp_path / ".egg-state" / "contracts" - state_dir.mkdir(parents=True) - # Build a contract with a very large acceptance_criteria field - big_contract = { - "current_phase": "implement", - "acceptance_criteria": ["x" * 5000], - } - (state_dir / "42.json").write_text(json.dumps(big_contract)) - - pipeline = _pipeline(repo=None) - ctx = PipelineHealthContext( - pipeline=pipeline, - repo_path=tmp_path, - trigger="phase_complete", - ) - prompt = _build_user_prompt(ctx) - - # The contract section should be present but capped - assert "Contract State" in prompt - # Find the JSON blob in the prompt — it should be <= 3000 chars - idx = prompt.index("## Contract State") - contract_section = prompt[idx:] - # The serialized JSON portion shouldn't exceed 3000 chars of the contract - # (plus section headers). Verify no single "x" run exceeds ~3000. - x_runs = [line for line in contract_section.split("\n") if "xxxxx" in line] - for line in x_runs: - assert len(line) <= 3001 - - def test_prompt_sections_order(self, tmp_path): - """Prompt has sections in correct order: metadata, commits, diff, outputs, contract.""" - state_dir = tmp_path / ".egg-state" / "drafts" - state_dir.mkdir(parents=True) - (state_dir / "plan.md").write_text("plan content") - - pipeline = _pipeline(repo=None) - ctx = PipelineHealthContext( - pipeline=pipeline, - repo_path=tmp_path, - trigger="phase_complete", - ) - with patch.object(ctx, "_run_git", return_value="abc log"): - _ = ctx.git_log - _ = ctx.git_diff_stat - - prompt = _build_user_prompt(ctx) - - # Sections should appear in order - assert ( - prompt.index("Pipeline:") < prompt.index("## Recent Commits") - or "## Recent Commits" not in prompt - ) - if "## Agent Output Files" in prompt: - assert prompt.index("## Agent Output Files") > 0 - - -# =========================================================================== -# 5. Verdict Parsing — extended coverage +# 4. Verdict Parsing — extended coverage # =========================================================================== @@ -535,119 +417,84 @@ def test_partial_json_returns_healthy(self): # =========================================================================== -# 6. API Call — extended coverage +# 5. Container Output Parsing — extended coverage # =========================================================================== -class TestCallClaudeApiExtended: - """Extended tests for _call_claude_api.""" - - @patch("health_checks.tier2.agent_inspector.httpx") - def test_non_retryable_exception_raises_immediately(self, mock_httpx): - """Generic Exception (not Timeout/HTTPStatusError) raises without retry.""" - mock_httpx.post.side_effect = ValueError("Unexpected error") - # Need to set these so the except clause can match properly - import httpx - - mock_httpx.TimeoutException = httpx.TimeoutException - mock_httpx.HTTPStatusError = httpx.HTTPStatusError - - with pytest.raises(ValueError, match="Unexpected error"): - _call_claude_api("test", api_key="sk-ant-key") - # Only 1 attempt — no retry for non-transient errors - assert mock_httpx.post.call_count == 1 - - @patch("health_checks.tier2.agent_inspector.httpx") - def test_base_url_trailing_slash_stripped(self, mock_httpx): - """Trailing slash in base_url is stripped before appending path.""" - mock_resp = MagicMock() - mock_resp.status_code = 200 - mock_resp.raise_for_status = MagicMock() - mock_resp.json.return_value = {"content": [{"type": "text", "text": "ok"}]} - mock_httpx.post.return_value = mock_resp - - _call_claude_api("test", api_key="key", base_url="https://api.example.com/") - - url_called = mock_httpx.post.call_args[0][0] - assert url_called == "https://api.example.com/v1/messages" - assert "//" not in url_called.replace("https://", "") - - @patch("health_checks.tier2.agent_inspector.httpx") - def test_payload_includes_system_prompt(self, mock_httpx): - """API payload includes the system prompt.""" - mock_resp = MagicMock() - mock_resp.status_code = 200 - mock_resp.raise_for_status = MagicMock() - mock_resp.json.return_value = {"content": [{"type": "text", "text": "ok"}]} - mock_httpx.post.return_value = mock_resp - - _call_claude_api("test prompt", api_key="key") - - payload = mock_httpx.post.call_args[1]["json"] - assert "system" in payload - assert "pipeline health inspector" in payload["system"] - - @patch("health_checks.tier2.agent_inspector.httpx") - def test_payload_max_tokens_is_512(self, mock_httpx): - """API payload requests max_tokens=512.""" - mock_resp = MagicMock() - mock_resp.status_code = 200 - mock_resp.raise_for_status = MagicMock() - mock_resp.json.return_value = {"content": [{"type": "text", "text": "ok"}]} - mock_httpx.post.return_value = mock_resp - - _call_claude_api("test", api_key="key") - - payload = mock_httpx.post.call_args[1]["json"] - assert payload["max_tokens"] == 512 - - @patch("health_checks.tier2.agent_inspector.httpx") - def test_default_model_is_claude_sonnet(self, mock_httpx): - """Default model is claude-sonnet-4-20250514 when no env var set.""" - mock_resp = MagicMock() - mock_resp.status_code = 200 - mock_resp.raise_for_status = MagicMock() - mock_resp.json.return_value = {"content": [{"type": "text", "text": "ok"}]} - mock_httpx.post.return_value = mock_resp - - # Ensure env vars are cleared - env = {k: v for k, v in os.environ.items() if k != "HEALTH_CHECK_MODEL"} - with patch.dict("os.environ", env, clear=True): - _call_claude_api("test", api_key="key") - - payload = mock_httpx.post.call_args[1]["json"] - assert payload["model"] == "claude-sonnet-4-20250514" - - @patch("health_checks.tier2.agent_inspector.httpx") - def test_timeout_value_is_30_seconds(self, mock_httpx): - """API call uses 30-second timeout.""" - mock_resp = MagicMock() - mock_resp.status_code = 200 - mock_resp.raise_for_status = MagicMock() - mock_resp.json.return_value = {"content": [{"type": "text", "text": "ok"}]} - mock_httpx.post.return_value = mock_resp - - _call_claude_api("test", api_key="key") - - call_kwargs = mock_httpx.post.call_args[1] - assert call_kwargs["timeout"] == 30 - - @patch("health_checks.tier2.agent_inspector.httpx") - def test_user_prompt_forwarded_to_api(self, mock_httpx): - """User prompt is included in the messages array.""" - mock_resp = MagicMock() - mock_resp.status_code = 200 - mock_resp.raise_for_status = MagicMock() - mock_resp.json.return_value = {"content": [{"type": "text", "text": "ok"}]} - mock_httpx.post.return_value = mock_resp - - _call_claude_api("Analyze this pipeline", api_key="key") - - payload = mock_httpx.post.call_args[1]["json"] - messages = payload["messages"] - assert len(messages) == 1 - assert messages[0]["role"] == "user" - assert messages[0]["content"] == "Analyze this pipeline" +class TestParseContainerOutputExtended: + """Extended tests for _parse_container_output.""" + + def test_json_without_timestamp(self): + """Parses JSON without any timestamp prefix.""" + logs = '{"raw_response": "test verdict"}\n' + result = _parse_container_output(logs) + assert result == "test verdict" + + def test_multiple_json_lines_picks_last(self): + """When multiple JSON lines, picks the last one (most recent).""" + logs = '{"raw_response": "first"}\n{"raw_response": "second"}\n' + result = _parse_container_output(logs) + assert result == "second" + + def test_missing_raw_response_returns_empty(self): + """Missing raw_response key returns empty string.""" + logs = '{"other_key": "value"}\n' + result = _parse_container_output(logs) + assert result == "" + + def test_invalid_json_lines_skipped(self): + """Invalid JSON lines are skipped, valid one is found.""" + logs = 'Some log output\n{broken json\n{"raw_response": "found it"}\n' + result = _parse_container_output(logs) + assert result == "found it" + + def test_empty_logs_raises(self): + """Empty logs raise ValueError.""" + with pytest.raises(ValueError, match="No valid JSON found"): + _parse_container_output("") + + def test_only_whitespace_raises(self): + """Whitespace-only logs raise ValueError.""" + with pytest.raises(ValueError, match="No valid JSON found"): + _parse_container_output(" \n \n ") + + +# =========================================================================== +# 6. Context Serialization — extended coverage +# =========================================================================== + + +class TestSerializeContextExtended: + """Extended tests for _serialize_context.""" + + def test_git_log_included(self, tmp_path): + """Serialized context includes git log.""" + ctx = _ctx(repo_path=tmp_path) + with patch.object(ctx, "_run_git", return_value="abc1234 commit msg"): + _ = ctx.git_log + payload = _serialize_context(ctx) + assert "abc1234 commit msg" in payload["git_log"] + + def test_empty_contract_gives_empty_summary(self, tmp_path): + """Empty contract produces empty contract_summary.""" + ctx = _ctx(repo_path=tmp_path) + payload = _serialize_context(ctx) + assert payload["contract_summary"] == {} + + def test_agent_outputs_included(self, tmp_path): + """Agent outputs are included in serialized context.""" + drafts = tmp_path / ".egg-state" / "drafts" + drafts.mkdir(parents=True) + (drafts / "plan.md").write_text("plan content") + + pipeline = _pipeline(repo=None) + ctx = PipelineHealthContext( + pipeline=pipeline, + repo_path=tmp_path, + trigger="phase_complete", + ) + payload = _serialize_context(ctx) + assert "plan.md" in payload["agent_outputs"] # =========================================================================== @@ -658,10 +505,12 @@ def test_user_prompt_forwarded_to_api(self, mock_httpx): class TestAgentInspectorCheckExtended: """Extended tests for AgentInspectorCheck.run().""" - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_prompt_built_from_context(self, mock_api, tmp_path): - """Verifies _build_user_prompt is called with the context.""" - mock_api.return_value = '{"status": "HEALTHY", "reasoning": "OK"}' + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_prompt_serialization_called(self, mock_run, tmp_path): + """Verifies _serialize_context is called with the context.""" + mock_run.return_value = ( + '{"raw_response": "{\\"status\\": \\"HEALTHY\\", \\"reasoning\\": \\"OK\\"}"}\n' + ) check = AgentInspectorCheck() pipeline = _pipeline(repo=None) @@ -672,16 +521,16 @@ def test_prompt_built_from_context(self, mock_api, tmp_path): ) with patch( - "health_checks.tier2.agent_inspector._build_user_prompt", - wraps=_build_user_prompt, - ) as mock_build: + "health_checks.tier2.agent_inspector._serialize_context", + wraps=_serialize_context, + ) as mock_serialize: check.run(ctx) - mock_build.assert_called_once_with(ctx) + mock_serialize.assert_called_once_with(ctx) - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_connection_error_graceful_degradation(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_connection_error_graceful_degradation(self, mock_run, tmp_path): """ConnectionError results in graceful degradation.""" - mock_api.side_effect = ConnectionError("Connection refused") + mock_run.side_effect = ConnectionError("Connection refused") check = AgentInspectorCheck() result = check.run(_ctx(repo_path=tmp_path)) @@ -691,34 +540,23 @@ def test_connection_error_graceful_degradation(self, mock_api, tmp_path): assert result.details.get("graceful_degradation") is True assert "ConnectionError" in result.reasoning - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_keyboard_interrupt_graceful_degradation(self, mock_api, tmp_path): - """Even KeyboardInterrupt-derived errors degrade gracefully.""" - mock_api.side_effect = Exception("Simulated interrupt") - - check = AgentInspectorCheck() - result = check.run(_ctx(repo_path=tmp_path)) - - assert result.status == HealthStatus.HEALTHY - assert result.details.get("graceful_degradation") is True - - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_check_name_and_tier_in_every_result(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_check_name_and_tier_in_every_result(self, mock_run, tmp_path): """Every result has correct check_name and tier regardless of outcome.""" scenarios = [ - '{"status": "HEALTHY", "reasoning": "OK"}', - '{"status": "DEGRADED", "reasoning": "Concern"}', - '{"status": "FAILED", "reasoning": "Bad"}', + '{"raw_response": "{\\"status\\": \\"HEALTHY\\", \\"reasoning\\": \\"OK\\"}"}\n', + '{"raw_response": "{\\"status\\": \\"DEGRADED\\", \\"reasoning\\": \\"Concern\\"}"}\n', + '{"raw_response": "{\\"status\\": \\"FAILED\\", \\"reasoning\\": \\"Bad\\"}"}\n', ] for response in scenarios: - mock_api.return_value = response + mock_run.return_value = response check = AgentInspectorCheck() result = check.run(_ctx(repo_path=tmp_path)) assert result.check_name == "agent_inspector" assert result.tier == HealthTier.AGENT - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_triggers_include_required_set(self, mock_api): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_triggers_include_required_set(self, mock_run): """Check triggers match the design spec.""" check = AgentInspectorCheck() assert HealthTrigger.WAVE_COMPLETE in check.triggers @@ -815,20 +653,18 @@ def test_tier2_exception_caught_by_run_single(self, tmp_path): assert "failed internally" in tier2_result.reasoning assert tier2_result.action == HealthAction.ALERT - def test_multiple_tier2_checks_all_run(self, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_multiple_tier2_checks_all_run(self, mock_run, tmp_path): """Multiple Tier 2 checks all execute when escalation triggers.""" + mock_run.return_value = '{"raw_response": "{\\"status\\": \\"HEALTHY\\", \\"reasoning\\": \\"Inspector OK\\"}"}\n' + runner = HealthCheckRunner() runner.register(_HealthyTier1()) runner.register(AgentInspectorCheck()) runner.register(_SecondTier2()) ctx = _ctx(repo_path=tmp_path, trigger="phase_complete") - - with patch( - "health_checks.tier2.agent_inspector._call_claude_api", - return_value='{"status": "HEALTHY", "reasoning": "Inspector OK"}', - ): - results = runner.run(ctx, HealthTrigger.PHASE_COMPLETE) + results = runner.run(ctx, HealthTrigger.PHASE_COMPLETE) # 1 Tier 1 + 2 Tier 2 checks assert len(results) == 3 @@ -909,10 +745,10 @@ def test_worst_action_empty_results(self): class TestEventEmissionExtended: """Extended event emission tests.""" - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_failed_verdict_emits_health_check_failed_event(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_failed_verdict_emits_health_check_failed_event(self, mock_run, tmp_path): """FAILED verdict emits HEALTH_CHECK_FAILED event type.""" - mock_api.return_value = '{"status": "FAILED", "reasoning": "Agent stuck."}' + mock_run.return_value = '{"raw_response": "{\\"status\\": \\"FAILED\\", \\"reasoning\\": \\"Agent stuck.\\"}"}\n' runner = HealthCheckRunner() runner.register(_HealthyTier1()) @@ -938,10 +774,12 @@ def test_failed_verdict_emits_health_check_failed_event(self, mock_api, tmp_path ] assert len(tier2_events) == 1 - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_healthy_verdict_emits_health_check_completed_event(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_healthy_verdict_emits_health_check_completed_event(self, mock_run, tmp_path): """HEALTHY verdict emits HEALTH_CHECK_COMPLETED event type for per-check.""" - mock_api.return_value = '{"status": "HEALTHY", "reasoning": "OK."}' + mock_run.return_value = ( + '{"raw_response": "{\\"status\\": \\"HEALTHY\\", \\"reasoning\\": \\"OK.\\"}"}\n' + ) runner = HealthCheckRunner() runner.register(_HealthyTier1()) @@ -972,10 +810,12 @@ def test_healthy_verdict_emits_health_check_completed_event(self, mock_api, tmp_ names = {kw["data"]["check_name"] for _, kw in completed_events} assert "agent_inspector" in names - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_degraded_verdict_emits_health_check_degraded_event(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_degraded_verdict_emits_health_check_degraded_event(self, mock_run, tmp_path): """DEGRADED verdict emits HEALTH_CHECK_DEGRADED event type.""" - mock_api.return_value = '{"status": "DEGRADED", "reasoning": "Stale."}' + mock_run.return_value = ( + '{"raw_response": "{\\"status\\": \\"DEGRADED\\", \\"reasoning\\": \\"Stale.\\"}"}\n' + ) runner = HealthCheckRunner() runner.register(_HealthyTier1()) @@ -1012,10 +852,10 @@ def test_no_event_bus_does_not_crash(self, tmp_path): assert len(results) == 1 assert results[0].status == HealthStatus.HEALTHY - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_event_payload_has_required_sse_fields(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_event_payload_has_required_sse_fields(self, mock_run, tmp_path): """Event payload includes all fields needed for SSE clients.""" - mock_api.return_value = '{"status": "DEGRADED", "reasoning": "No commits."}' + mock_run.return_value = '{"raw_response": "{\\"status\\": \\"DEGRADED\\", \\"reasoning\\": \\"No commits.\\"}"}\n' runner = HealthCheckRunner() runner.register(_HealthyTier1()) @@ -1307,10 +1147,12 @@ def test_health_result_is_frozen(self): class TestRunnerWithRealInspector: """Integration tests with the actual AgentInspectorCheck.""" - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_full_run_phase_complete(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_full_run_phase_complete(self, mock_run, tmp_path): """Full PHASE_COMPLETE run with healthy Tier 1 + real AgentInspectorCheck.""" - mock_api.return_value = '{"status": "HEALTHY", "reasoning": "All good."}' + mock_run.return_value = ( + '{"raw_response": "{\\"status\\": \\"HEALTHY\\", \\"reasoning\\": \\"All good.\\"}"}\n' + ) runner = HealthCheckRunner() runner.register(_HealthyTier1()) @@ -1325,10 +1167,10 @@ def test_full_run_phase_complete(self, mock_api, tmp_path): assert results[1].check_name == "agent_inspector" assert results[1].status == HealthStatus.HEALTHY - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_full_run_wave_complete_degraded_triggers_inspector(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_full_run_wave_complete_degraded_triggers_inspector(self, mock_run, tmp_path): """WAVE_COMPLETE with degraded Tier 1 triggers the real AgentInspectorCheck.""" - mock_api.return_value = '{"status": "DEGRADED", "reasoning": "Inspector concern."}' + mock_run.return_value = '{"raw_response": "{\\"status\\": \\"DEGRADED\\", \\"reasoning\\": \\"Inspector concern.\\"}"}\n' runner = HealthCheckRunner() runner.register(_DegradedTier1()) @@ -1343,12 +1185,10 @@ def test_full_run_wave_complete_degraded_triggers_inspector(self, mock_api, tmp_ assert inspector_result.status == HealthStatus.DEGRADED assert inspector_result.action == HealthAction.ALERT - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_full_run_api_failure_still_returns_results(self, mock_api, tmp_path): - """API failure during run still returns a result (graceful degradation).""" - import httpx - - mock_api.side_effect = httpx.TimeoutException("timeout") + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_full_run_container_failure_still_returns_results(self, mock_run, tmp_path): + """Container failure during run still returns a result (graceful degradation).""" + mock_run.side_effect = RuntimeError("Container spawn failed") runner = HealthCheckRunner() runner.register(_HealthyTier1()) @@ -1362,10 +1202,10 @@ def test_full_run_api_failure_still_returns_results(self, mock_api, tmp_path): assert inspector_result.status == HealthStatus.HEALTHY # graceful degradation assert "unavailable" in inspector_result.reasoning - @patch("health_checks.tier2.agent_inspector._call_claude_api") - def test_full_run_on_demand(self, mock_api, tmp_path): + @patch("health_checks.tier2.agent_inspector._run_inspector_container") + def test_full_run_on_demand(self, mock_run, tmp_path): """ON_DEMAND always runs Tier 2 with real AgentInspectorCheck.""" - mock_api.return_value = '{"status": "HEALTHY", "reasoning": "On-demand check OK."}' + mock_run.return_value = '{"raw_response": "{\\"status\\": \\"HEALTHY\\", \\"reasoning\\": \\"On-demand check OK.\\"}"}\n' runner = HealthCheckRunner() runner.register(_HealthyTier1()) diff --git a/sandbox/bin/egg-health-inspect b/sandbox/bin/egg-health-inspect new file mode 100755 index 0000000000..ecd86e3d50 --- /dev/null +++ b/sandbox/bin/egg-health-inspect @@ -0,0 +1,177 @@ +#!/usr/bin/env python3 +""" +Health inspector script — runs inside the sandbox container. + +Reads pipeline context JSON from the EGG_INSPECTOR_CONTEXT env var (preferred), +a file path in EGG_INSPECTOR_CONTEXT_PATH, or stdin. Calls the Claude API to +produce a health verdict and outputs a JSON verdict to stdout. + +Exit codes: + 0 — verdict produced successfully + 1 — error (context missing, API failure, etc.) +""" + +from __future__ import annotations + +import json +import os +import sys + +import httpx + +# Default model for agent inspection +_DEFAULT_MODEL = "claude-sonnet-4-20250514" +# API timeout in seconds +_API_TIMEOUT = 30 +# Max retries on transient failures +_MAX_RETRIES = 1 + +# System prompt for the inspector +_SYSTEM_PROMPT = """\ +You are a pipeline health inspector. Analyze the provided context about a \ +software engineering agent's work and produce a health verdict. + +Respond ONLY with a JSON object (no markdown fencing, no extra text): +{ + "status": "HEALTHY" | "DEGRADED" | "FAILED", + "reasoning": "" +} + +Rules: +- HEALTHY: Agent is making reasonable progress; commits present; no red flags. +- DEGRADED: Minor concerns — e.g. no recent commits, stale output, contract \ +tasks still pending after a long time. Use this when there's risk but not \ +certainty of failure. +- FAILED: Clear signs of stuck agent — e.g. repeated errors in outputs, \ +no commits and no drafts, contradictory state. +- Be concise. One paragraph for reasoning. +""" + + +def _read_context() -> dict: + """Read inspector context from env var, file, or stdin.""" + # Preferred: context passed directly via env var (avoids file mounting) + context_env = os.environ.get("EGG_INSPECTOR_CONTEXT") + if context_env: + return json.loads(context_env) + # Fallback: file path + context_path = os.environ.get("EGG_INSPECTOR_CONTEXT_PATH") + if context_path: + with open(context_path) as f: + return json.load(f) + # Last resort: stdin + return json.load(sys.stdin) + + +def _call_claude_api(user_prompt: str) -> str: + """Call the Anthropic Messages API and return text response.""" + api_key = os.environ.get("ANTHROPIC_API_KEY", "") + base_url = os.environ.get("ANTHROPIC_BASE_URL", "https://api.anthropic.com").rstrip("/") + model = os.environ.get("HEALTH_CHECK_MODEL", _DEFAULT_MODEL) + + headers = { + "x-api-key": api_key, + "anthropic-version": "2023-06-01", + "content-type": "application/json", + } + + payload = { + "model": model, + "max_tokens": 512, + "system": _SYSTEM_PROMPT, + "messages": [{"role": "user", "content": user_prompt}], + } + + last_exc: Exception | None = None + for attempt in range(_MAX_RETRIES + 1): + try: + resp = httpx.post( + f"{base_url}/v1/messages", + headers=headers, + json=payload, + timeout=_API_TIMEOUT, + ) + resp.raise_for_status() + body = resp.json() + content_blocks = body.get("content", []) + if content_blocks and isinstance(content_blocks, list): + return str(content_blocks[0].get("text", "")) + return "" + except (httpx.TimeoutException, httpx.HTTPStatusError) as exc: + last_exc = exc + if attempt < _MAX_RETRIES: + print( + f"API attempt {attempt + 1} failed: {exc}", + file=sys.stderr, + ) + continue + raise + except Exception: + raise + + if last_exc: + raise last_exc + return "" + + +def main() -> int: + """Read context, call Claude API, output verdict JSON.""" + try: + context = _read_context() + except Exception as exc: + print(f"Failed to read context: {exc}", file=sys.stderr) + return 1 + + # Build user prompt from context fields + parts: list[str] = [] + parts.append(f"Pipeline: {context.get('pipeline_id', 'unknown')}") + parts.append(f"Phase: {context.get('current_phase', 'unknown')}") + parts.append(f"Branch: {context.get('branch', 'unknown')}") + parts.append(f"Trigger: {context.get('trigger', 'unknown')}") + parts.append("") + + if context.get("git_log"): + parts.append("## Recent Commits") + parts.append(context["git_log"]) + parts.append("") + + if context.get("git_diff_stat"): + parts.append("## Diff Stats (vs main)") + parts.append(context["git_diff_stat"]) + parts.append("") + + if context.get("agent_outputs"): + parts.append("## Agent Output Files") + for name, content in context["agent_outputs"].items(): + parts.append(f"### {name}") + parts.append(str(content)[:2000]) + parts.append("") + else: + parts.append("## Agent Output Files") + parts.append("(none found)") + parts.append("") + + if context.get("contract_summary"): + parts.append("## Contract State") + parts.append(json.dumps(context["contract_summary"], indent=2, default=str)[:3000]) + parts.append("") + + user_prompt = "\n".join(parts) + + try: + response_text = _call_claude_api(user_prompt) + except Exception as exc: + print(f"API call failed: {exc}", file=sys.stderr) + return 1 + + # Output verdict to stdout + json.dump( + {"raw_response": response_text}, + sys.stdout, + ) + print() # trailing newline + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/check-llm-api-calls.py b/scripts/check-llm-api-calls.py new file mode 100755 index 0000000000..6f1219de19 --- /dev/null +++ b/scripts/check-llm-api-calls.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 +""" +Lint check: Ensure LLM API calls only happen inside the sandbox. + +The orchestrator, gateway, and shared modules must NEVER call the Anthropic +API directly. LLM calls must be delegated to sandbox containers — this +maintains the security boundary between the orchestrator (which has Docker +and pipeline credentials) and the LLM (which processes untrusted prompts). + +Detection patterns (AST-based for Python files): + - ``import anthropic`` / ``from anthropic import ...`` + - String literals containing ``api.anthropic.com`` + - ``os.environ.get("ANTHROPIC_API_KEY")`` or ``os.environ["ANTHROPIC_API_KEY"]`` + +Known gap: + Indirect API calls via ``urllib.request`` (without importing the anthropic + SDK or referencing ``api.anthropic.com`` in a detectable way) are not + caught. Suppress with ``# noqa: EGG200`` where needed. + +Suppression: + # noqa: EGG200 - + +Usage: + python3 scripts/check-llm-api-calls.py + +Exit codes: + 0 - No violations found + 1 - Found violations +""" + +from __future__ import annotations + +import ast +import sys +from pathlib import Path + +NOQA_CODE = "EGG200" + +# Directories to scan (relative to repo root) +SCAN_DIRS = ("orchestrator", "gateway", "shared") + +# Directories/patterns to skip +SKIP_DIRS = {".venv", ".git", "__pycache__", "node_modules", ".mypy_cache"} +SKIP_SUFFIXES = ("_test.py", "test_.py") + + +def _should_skip_path(path: Path) -> bool: + """Return True if path should be excluded from scanning.""" + parts = path.parts + # Skip excluded directories + if any(part in SKIP_DIRS for part in parts): + return True + # Skip test files + name = path.name + if name.startswith("test_") or name.endswith("_test.py"): + return True + # Skip files in test directories + if "tests" in parts: + return True + return False + + +class LLMApiVisitor(ast.NodeVisitor): + """AST visitor that detects direct LLM API usage.""" + + def __init__(self, source_lines: list[str]): + self.source_lines = source_lines + self.violations: list[tuple[int, str]] = [] + + def _has_noqa(self, lineno: int) -> bool: + """Check if a line has a noqa: EGG200 comment.""" + if 1 <= lineno <= len(self.source_lines): + line = self.source_lines[lineno - 1] + return f"noqa: {NOQA_CODE}" in line + return False + + def visit_Import(self, node: ast.Import) -> None: + """Detect ``import anthropic``.""" + if self._has_noqa(node.lineno): + self.generic_visit(node) + return + for alias in node.names: + if alias.name == "anthropic" or alias.name.startswith("anthropic."): + self.violations.append( + (node.lineno, f"Direct import of Anthropic SDK: import {alias.name}") + ) + self.generic_visit(node) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + """Detect ``from anthropic import ...``.""" + if self._has_noqa(node.lineno): + self.generic_visit(node) + return + if node.module and (node.module == "anthropic" or node.module.startswith("anthropic.")): + self.violations.append( + (node.lineno, f"Direct import from Anthropic SDK: from {node.module} import ...") + ) + self.generic_visit(node) + + def visit_Constant(self, node: ast.Constant) -> None: + """Detect string literals referencing the Anthropic API.""" + if not isinstance(node.value, str): + self.generic_visit(node) + return + if self._has_noqa(node.lineno): + self.generic_visit(node) + return + + val = node.value + if "api.anthropic.com" in val: + self.violations.append( + (node.lineno, f"Anthropic API URL in string literal: ...{val[:80]}...") + ) + self.generic_visit(node) + + def visit_Call(self, node: ast.Call) -> None: + """Detect os.environ access for ANTHROPIC_API_KEY.""" + if self._has_noqa(node.lineno): + self.generic_visit(node) + return + + # Detect os.environ.get("ANTHROPIC_API_KEY") or os.environ["ANTHROPIC_API_KEY"] + func = node.func + if isinstance(func, ast.Attribute) and func.attr == "get": + if isinstance(func.value, ast.Attribute) and func.value.attr == "environ": + if isinstance(func.value.value, ast.Name) and func.value.value.id == "os": + if node.args: + first_arg = node.args[0] + if ( + isinstance(first_arg, ast.Constant) + and isinstance(first_arg.value, str) + and first_arg.value == "ANTHROPIC_API_KEY" + ): + self.violations.append( + ( + node.lineno, + "Direct access to ANTHROPIC_API_KEY via os.environ.get()", + ) + ) + + self.generic_visit(node) + + def visit_Subscript(self, node: ast.Subscript) -> None: + """Detect os.environ["ANTHROPIC_API_KEY"].""" + if self._has_noqa(node.lineno): + self.generic_visit(node) + return + + if isinstance(node.value, ast.Attribute) and node.value.attr == "environ": + if isinstance(node.value.value, ast.Name) and node.value.value.id == "os": + if isinstance(node.slice, ast.Constant) and node.slice.value == "ANTHROPIC_API_KEY": + self.violations.append( + (node.lineno, "Direct access to ANTHROPIC_API_KEY via os.environ[]") + ) + + self.generic_visit(node) + + +def check_python_file(file_path: Path) -> list[tuple[int, str]]: + """Parse a Python file and return list of (lineno, description) violations.""" + try: + content = file_path.read_text() + tree = ast.parse(content, filename=str(file_path)) + lines = content.split("\n") + visitor = LLMApiVisitor(lines) + visitor.visit(tree) + return visitor.violations + except SyntaxError as e: + print(f"Warning: Could not parse {file_path}: {e}", file=sys.stderr) + return [] + except Exception as e: + print(f"Warning: Could not read {file_path}: {e}", file=sys.stderr) + return [] + + +def main() -> int: + """Run all checks and report violations.""" + script_dir = Path(__file__).resolve().parent + repo_root = script_dir.parent + + all_violations: list[tuple[str, list[tuple[int, str]]]] = [] + + for scan_dir in SCAN_DIRS: + dir_path = repo_root / scan_dir + if not dir_path.is_dir(): + continue + + for py_file in dir_path.rglob("*.py"): + if _should_skip_path(py_file): + continue + + rel = str(py_file.relative_to(repo_root)) + violations = check_python_file(py_file) + if violations: + all_violations.append((rel, violations)) + + if all_violations: + print("ERROR: Found direct LLM API usage outside the sandbox!\n") + print("=" * 70) + print("LLM API calls must ONLY happen inside sandbox containers.") + print("The orchestrator/gateway/shared must delegate to sandbox.") + print("=" * 70) + print() + + for file_path, violations in sorted(all_violations): + print(f"File: {file_path}") + for lineno, desc in sorted(violations): + print(f" Line {lineno}: {desc}") + print() + + print("How to fix:") + print(" 1. Move LLM API calls to a sandbox script (e.g. sandbox/bin/)") + print(" 2. Use ContainerSpawner to delegate from the orchestrator") + print(" 3. If this is a false positive, suppress with:") + print(f" # noqa: {NOQA_CODE} - ") + print() + print(" See orchestrator/health_checks/tier2/agent_inspector.py for") + print(" an example of the sandbox delegation pattern.") + print() + + return 1 + else: + print("OK: No direct LLM API usage found outside sandbox") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/tests/__init__.py b/scripts/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/scripts/tests/test_check_llm_api_calls.py b/scripts/tests/test_check_llm_api_calls.py new file mode 100644 index 0000000000..0524ca7369 --- /dev/null +++ b/scripts/tests/test_check_llm_api_calls.py @@ -0,0 +1,193 @@ +"""Tests for the LLM API boundary linter (scripts/check-llm-api-calls.py). + +Verifies that the linter correctly detects: +- Direct Anthropic SDK imports +- Anthropic API URL string literals +- ANTHROPIC_API_KEY environment access +- Proper suppression via noqa: EGG200 +""" + +import sys +import textwrap +from pathlib import Path + +# Add scripts directory to path so we can import the linter +_scripts_path = Path(__file__).parent.parent +if str(_scripts_path) not in sys.path: + sys.path.insert(0, str(_scripts_path)) + +# Import after path manipulation (module has hyphens, use importlib) +import importlib.util + +spec = importlib.util.spec_from_file_location( + "check_llm_api_calls", + _scripts_path / "check-llm-api-calls.py", +) +check_llm_api_calls = importlib.util.module_from_spec(spec) +spec.loader.exec_module(check_llm_api_calls) + + +def _write_and_check(tmp_path: Path, code: str) -> list[tuple[int, str]]: + """Write code to a temp Python file and check it.""" + py_file = tmp_path / "test_file.py" + py_file.write_text(textwrap.dedent(code)) + return check_llm_api_calls.check_python_file(py_file) + + +class TestDetectsAnthropicImports: + """Tests that direct Anthropic SDK imports are detected.""" + + def test_import_anthropic(self, tmp_path): + violations = _write_and_check(tmp_path, "import anthropic\n") + assert len(violations) == 1 + assert "import" in violations[0][1].lower() + + def test_from_anthropic_import(self, tmp_path): + violations = _write_and_check(tmp_path, "from anthropic import Client\n") + assert len(violations) == 1 + assert "import" in violations[0][1].lower() + + def test_import_anthropic_submodule(self, tmp_path): + violations = _write_and_check(tmp_path, "import anthropic.resources\n") + assert len(violations) == 1 + + def test_from_anthropic_submodule(self, tmp_path): + violations = _write_and_check(tmp_path, "from anthropic.types import Message\n") + assert len(violations) == 1 + + def test_unrelated_import_clean(self, tmp_path): + violations = _write_and_check(tmp_path, "import json\nimport os\n") + assert len(violations) == 0 + + +class TestDetectsApiUrls: + """Tests that Anthropic API URL string literals are detected.""" + + def test_api_url_in_string(self, tmp_path): + code = 'url = "https://api.anthropic.com/v1/messages"\n' + violations = _write_and_check(tmp_path, code) + assert len(violations) == 1 + assert "api.anthropic.com" in violations[0][1] + + def test_api_url_in_httpx_call(self, tmp_path): + code = 'resp = httpx.post("https://api.anthropic.com/v1/messages", json={})\n' + violations = _write_and_check(tmp_path, code) + assert len(violations) == 1 + + def test_no_false_positive_on_unrelated_urls(self, tmp_path): + code = 'url = "https://api.github.com/repos"\n' + violations = _write_and_check(tmp_path, code) + assert len(violations) == 0 + + +class TestDetectsApiKeyAccess: + """Tests that ANTHROPIC_API_KEY environment access is detected.""" + + def test_os_environ_get(self, tmp_path): + code = 'key = os.environ.get("ANTHROPIC_API_KEY")\n' + violations = _write_and_check(tmp_path, code) + assert len(violations) == 1 + assert "ANTHROPIC_API_KEY" in violations[0][1] + + def test_os_environ_subscript(self, tmp_path): + code = 'key = os.environ["ANTHROPIC_API_KEY"]\n' + violations = _write_and_check(tmp_path, code) + assert len(violations) == 1 + assert "ANTHROPIC_API_KEY" in violations[0][1] + + def test_os_environ_get_other_key_clean(self, tmp_path): + code = 'val = os.environ.get("HOME")\n' + violations = _write_and_check(tmp_path, code) + assert len(violations) == 0 + + +class TestNoqaSuppression: + """Tests that noqa: EGG200 suppresses violations.""" + + def test_noqa_suppresses_import(self, tmp_path): + code = "import anthropic # noqa: EGG200 - test helper\n" + violations = _write_and_check(tmp_path, code) + assert len(violations) == 0 + + def test_noqa_suppresses_url(self, tmp_path): + code = 'url = "https://api.anthropic.com" # noqa: EGG200 - constant for docs\n' + violations = _write_and_check(tmp_path, code) + assert len(violations) == 0 + + def test_noqa_suppresses_env_get(self, tmp_path): + code = 'key = os.environ.get("ANTHROPIC_API_KEY") # noqa: EGG200 - forwarding\n' + violations = _write_and_check(tmp_path, code) + assert len(violations) == 0 + + def test_wrong_noqa_code_does_not_suppress(self, tmp_path): + code = "import anthropic # noqa: EGG100 - wrong code\n" + violations = _write_and_check(tmp_path, code) + assert len(violations) == 1 + + +class TestCleanCode: + """Tests that clean code passes without violations.""" + + def test_normal_python_code(self, tmp_path): + code = """\ + import json + import os + from pathlib import Path + + def hello(): + return "world" + """ + violations = _write_and_check(tmp_path, code) + assert len(violations) == 0 + + def test_httpx_to_other_apis(self, tmp_path): + code = """\ + import httpx + resp = httpx.post("https://api.openai.com/v1/chat", json={}) + """ + violations = _write_and_check(tmp_path, code) + assert len(violations) == 0 + + def test_syntax_error_returns_empty(self, tmp_path): + """Files with syntax errors return empty violations (not crash).""" + py_file = tmp_path / "broken.py" + py_file.write_text("def broken(\n") + violations = check_llm_api_calls.check_python_file(py_file) + assert len(violations) == 0 + + +class TestPathFiltering: + """Tests for _should_skip_path.""" + + def test_skip_venv(self): + p = Path(".venv/lib/python3.11/anthropic.py") + assert check_llm_api_calls._should_skip_path(p) is True + + def test_skip_test_files(self): + p = Path("orchestrator/tests/test_inspector.py") + assert check_llm_api_calls._should_skip_path(p) is True + + def test_skip_test_prefix(self): + p = Path("orchestrator/test_something.py") + assert check_llm_api_calls._should_skip_path(p) is True + + def test_allow_source_files(self): + p = Path("orchestrator/health_checks/tier2/agent_inspector.py") + assert check_llm_api_calls._should_skip_path(p) is False + + def test_skip_pycache(self): + p = Path("orchestrator/__pycache__/module.py") + assert check_llm_api_calls._should_skip_path(p) is True + + +class TestMultipleViolations: + """Tests that multiple violations in one file are all detected.""" + + def test_multiple_patterns(self, tmp_path): + code = """\ + import anthropic + url = "https://api.anthropic.com/v1/messages" + key = os.environ.get("ANTHROPIC_API_KEY") + """ + violations = _write_and_check(tmp_path, code) + assert len(violations) == 3 diff --git a/shared/egg_config/configs/llm.py b/shared/egg_config/configs/llm.py index 9dd19ca824..5b848405ff 100644 --- a/shared/egg_config/configs/llm.py +++ b/shared/egg_config/configs/llm.py @@ -84,12 +84,12 @@ def health_check(self, timeout: float = 5.0) -> HealthCheckResult: import json import urllib.request - base_url = self.anthropic_base_url or "https://api.anthropic.com" + base_url = self.anthropic_base_url or "https://api.anthropic.com" # noqa: EGG200 - config default for sandbox use start = time.time() # Use a minimal messages request to verify API key # Note: This is a billable API call (minimal cost with max_tokens=1) - req = urllib.request.Request( + req = urllib.request.Request( # noqa: EGG200 - config health check used by sandbox containers f"{base_url}/v1/messages", data=json.dumps( { @@ -164,7 +164,7 @@ def from_env(cls) -> "LLMConfig": config = cls() # Load API key from environment - config.anthropic_api_key = os.environ.get("ANTHROPIC_API_KEY", "") + config.anthropic_api_key = os.environ.get("ANTHROPIC_API_KEY", "") # noqa: EGG200 - config loader for sandbox containers # Optional settings config.anthropic_base_url = os.environ.get("ANTHROPIC_BASE_URL", "")