feat(core): Agent Adapter protocol definition

codeframe/core/agent_adapter.py

@@ -0,0 +1,100 @@
+"""Agent adapter protocol for CodeFRAME.
+
+Defines the interface that any coding agent (Claude Code, Codex, Aider, built-in)
+must implement to be used as a CodeFrame execution engine.
+
+This module is headless - no FastAPI or HTTP dependencies.
+"""
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from enum import Enum
+from pathlib import Path
+from typing import Iterator, Protocol, runtime_checkable
+
+
+class AgentResultStatus(str, Enum):
+    """Terminal status from an agent execution."""
+
+    COMPLETED = "completed"
+    FAILED = "failed"
+    BLOCKED = "blocked"
+    TIMEOUT = "timeout"
+
+
+@dataclass
+class AdapterTokenUsage:
+    """Lightweight token usage for adapter results."""
+
+    input_tokens: int
+    output_tokens: int
+    model: str | None = None
+    cost_usd: float | None = None
+
+    @property
+    def total_tokens(self) -> int:
+        return self.input_tokens + self.output_tokens
+
+
+@dataclass
+class AgentContext:
+    """Everything CodeFrame provides to an execution engine."""
+
+    task_id: str
+    task_title: str
+    task_description: str
+    prd_content: str | None = None
+    tech_stack: str | None = None
+    project_preferences: str | None = None
+    relevant_files: list[str] = field(default_factory=list)
+    file_contents: dict[str, str] = field(default_factory=dict)
+    blocker_history: list[str] = field(default_factory=list)
+    dependency_context: str | None = None
+    verification_gates: list[str] = field(default_factory=list)
+    attempt: int = 0
+    previous_errors: list[str] = field(default_factory=list)
+
+
+@dataclass
+class AgentResult:
+    """What every execution engine returns to CodeFrame."""
+
+    status: AgentResultStatus
+    summary: str
+    files_modified: list[str] = field(default_factory=list)
+    files_created: list[str] = field(default_factory=list)
+    error: str | None = None
+    blocker_question: str | None = None
+    token_usage: AdapterTokenUsage | None = None
+    duration_ms: int = 0
+
+
+@dataclass
+class AgentEvent:
+    """Progress event yielded during agent execution."""
+
+    type: str
+    message: str
+    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    metadata: dict = field(default_factory=dict)
+
+
+@runtime_checkable
+class AgentAdapter(Protocol):
+    """Interface for any coding agent that CodeFrame can orchestrate."""
+
+    def execute(
+        self,
+        task_prompt: str,
+        workspace_path: Path,
+        context: AgentContext,
+        timeout_ms: int = 3_600_000,
+    ) -> AgentResult: ...
+
+    def stream_events(self) -> Iterator[AgentEvent]: ...
+
+    @property
+    def name(self) -> str: ...
+
+    @property
+    def requires_api_key(self) -> dict[str, str]: ...

tests/core/test_agent_adapter.py

@@ -0,0 +1,284 @@
+"""Tests for AgentAdapter protocol and supporting types.
+
+Validates:
+- Dataclass construction with defaults and full params
+- AgentResultStatus enum values
+- AgentAdapter protocol compliance via @runtime_checkable
+- Streaming iterator contract
+"""
+
+import pytest
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Iterator
+
+pytestmark = pytest.mark.v2
+
+
+class TestAgentResultStatus:
+    """AgentResultStatus enum covers all terminal states."""
+
+    def test_has_completed(self):
+        from codeframe.core.agent_adapter import AgentResultStatus
+        assert AgentResultStatus.COMPLETED.value == "completed"
+
+    def test_has_failed(self):
+        from codeframe.core.agent_adapter import AgentResultStatus
+        assert AgentResultStatus.FAILED.value == "failed"
+
+    def test_has_blocked(self):
+        from codeframe.core.agent_adapter import AgentResultStatus
+        assert AgentResultStatus.BLOCKED.value == "blocked"
+
+    def test_has_timeout(self):
+        from codeframe.core.agent_adapter import AgentResultStatus
+        assert AgentResultStatus.TIMEOUT.value == "timeout"
+
+    def test_is_str_enum(self):
+        from codeframe.core.agent_adapter import AgentResultStatus
+        assert isinstance(AgentResultStatus.COMPLETED, str)
+
+
+class TestAdapterTokenUsage:
+    """Lightweight token usage dataclass."""
+
+    def test_minimal_construction(self):
+        from codeframe.core.agent_adapter import AdapterTokenUsage
+        usage = AdapterTokenUsage(input_tokens=100, output_tokens=50)
+        assert usage.input_tokens == 100
+        assert usage.output_tokens == 50
+        assert usage.model is None
+        assert usage.cost_usd is None
+
+    def test_full_construction(self):
+        from codeframe.core.agent_adapter import AdapterTokenUsage
+        usage = AdapterTokenUsage(
+            input_tokens=1000,
+            output_tokens=500,
+            model="claude-sonnet-4-20250514",
+            cost_usd=0.015,
+        )
+        assert usage.model == "claude-sonnet-4-20250514"
+        assert usage.cost_usd == 0.015
+
+    def test_total_tokens(self):
+        from codeframe.core.agent_adapter import AdapterTokenUsage
+        usage = AdapterTokenUsage(input_tokens=100, output_tokens=50)
+        assert usage.total_tokens == 150
+
+
+class TestAgentContext:
+    """AgentContext captures all context CodeFrame provides to engines."""
+
+    def test_minimal_construction(self):
+        from codeframe.core.agent_adapter import AgentContext
+        ctx = AgentContext(
+            task_id="task-1",
+            task_title="Implement feature X",
+            task_description="Add X to the system",
+        )
+        assert ctx.task_id == "task-1"
+        assert ctx.prd_content is None
+        assert ctx.tech_stack is None
+        assert ctx.project_preferences is None
+        assert ctx.relevant_files == []
+        assert ctx.file_contents == {}
+        assert ctx.blocker_history == []
+        assert ctx.dependency_context is None
+        assert ctx.verification_gates == []
+        assert ctx.attempt == 0
+        assert ctx.previous_errors == []
+
+    def test_full_construction(self):
+        from codeframe.core.agent_adapter import AgentContext
+        ctx = AgentContext(
+            task_id="task-42",
+            task_title="Fix auth bug",
+            task_description="Session tokens expire too early",
+            prd_content="# Auth PRD\nTokens should last 24h",
+            tech_stack="Python with FastAPI",
+            project_preferences="Use ruff for linting",
+            relevant_files=["auth.py", "tests/test_auth.py"],
+            file_contents={"auth.py": "def login(): pass"},
+            blocker_history=["Previous: needed DB access"],
+            dependency_context="Task-41 created the auth module",
+            verification_gates=["ruff", "pytest"],
+            attempt=2,
+            previous_errors=["ImportError: no module named jwt"],
+        )
+        assert ctx.task_id == "task-42"
+        assert len(ctx.relevant_files) == 2
+        assert ctx.attempt == 2
+        assert len(ctx.previous_errors) == 1
+
+    def test_list_defaults_are_independent(self):
+        """Ensure default_factory creates independent lists (no shared mutable state)."""
+        from codeframe.core.agent_adapter import AgentContext
+        ctx1 = AgentContext(task_id="1", task_title="A", task_description="A")
+        ctx2 = AgentContext(task_id="2", task_title="B", task_description="B")
+        ctx1.relevant_files.append("file.py")
+        assert ctx2.relevant_files == []
+
+
+class TestAgentResult:
+    """AgentResult captures outcome from any engine."""
+
+    def test_minimal_construction(self):
+        from codeframe.core.agent_adapter import AgentResult, AgentResultStatus
+        result = AgentResult(
+            status=AgentResultStatus.COMPLETED,
+            summary="Added feature X",
+        )
+        assert result.status == AgentResultStatus.COMPLETED
+        assert result.files_modified == []
+        assert result.files_created == []
+        assert result.error is None
+        assert result.blocker_question is None
+        assert result.token_usage is None
+        assert result.duration_ms == 0
+
+    def test_failed_result(self):
+        from codeframe.core.agent_adapter import AgentResult, AgentResultStatus
+        result = AgentResult(
+            status=AgentResultStatus.FAILED,
+            summary="Could not implement",
+            error="ImportError: missing dependency",
+            duration_ms=5000,
+        )
+        assert result.status == AgentResultStatus.FAILED
+        assert result.error is not None
+
+    def test_blocked_result_with_question(self):
+        from codeframe.core.agent_adapter import AgentResult, AgentResultStatus
+        result = AgentResult(
+            status=AgentResultStatus.BLOCKED,
+            summary="Need clarification on auth approach",
+            blocker_question="Should we use JWT or session cookies?",
+        )
+        assert result.blocker_question is not None
+
+    def test_result_with_token_usage(self):
+        from codeframe.core.agent_adapter import (
+            AdapterTokenUsage, AgentResult, AgentResultStatus,
+        )
+        result = AgentResult(
+            status=AgentResultStatus.COMPLETED,
+            summary="Done",
+            token_usage=AdapterTokenUsage(input_tokens=1000, output_tokens=500),
+            files_modified=["auth.py"],
+            files_created=["tests/test_auth.py"],
+            duration_ms=12000,
+        )
+        assert result.token_usage.total_tokens == 1500
+        assert result.files_modified == ["auth.py"]
+        assert result.duration_ms == 12000
+
+
+class TestAgentEvent:
+    """AgentEvent supports progress streaming."""
+
+    def test_minimal_construction(self):
+        from codeframe.core.agent_adapter import AgentEvent
+        event = AgentEvent(type="progress", message="Working on step 1")
+        assert event.type == "progress"
+        assert event.message == "Working on step 1"
+        assert isinstance(event.timestamp, datetime)
+        assert event.metadata == {}
+
+    def test_with_metadata(self):
+        from codeframe.core.agent_adapter import AgentEvent
+        ts = datetime(2026, 3, 9, tzinfo=timezone.utc)
+        event = AgentEvent(
+            type="file_changed",
+            message="Modified auth.py",
+            timestamp=ts,
+            metadata={"file": "auth.py", "lines_changed": 15},
+        )
+        assert event.timestamp == ts
+        assert event.metadata["lines_changed"] == 15
+
+    def test_event_types_are_strings(self):
+        from codeframe.core.agent_adapter import AgentEvent
+        for event_type in ("progress", "file_changed", "command_run", "error"):
+            event = AgentEvent(type=event_type, message="test")
+            assert event.type == event_type
+
+
+class TestAgentAdapterProtocol:
+    """AgentAdapter protocol compliance via @runtime_checkable."""
+
+    def _make_compliant_class(self):
+        """Create a minimal class that satisfies AgentAdapter."""
+        from codeframe.core.agent_adapter import (
+            AgentContext, AgentEvent, AgentResult, AgentResultStatus,
+        )
+
+        class FakeAdapter:
+            def execute(
+                self,
+                task_prompt: str,
+                workspace_path: Path,
+                context: AgentContext,
+                timeout_ms: int = 3_600_000,
+            ) -> AgentResult:
+                return AgentResult(
+                    status=AgentResultStatus.COMPLETED,
+                    summary="fake",
+                    duration_ms=100,
+                )
+
+            def stream_events(self) -> Iterator[AgentEvent]:
+                yield AgentEvent(type="progress", message="working")
+
+            @property
+            def name(self) -> str:
+                return "fake"
+
+            @property
+            def requires_api_key(self) -> dict[str, str]:
+                return {}
+
+        return FakeAdapter
+
+    def test_compliant_class_satisfies_protocol(self):
+        from codeframe.core.agent_adapter import AgentAdapter
+        FakeAdapter = self._make_compliant_class()
+        adapter = FakeAdapter()
+        assert isinstance(adapter, AgentAdapter)
+
+    def test_non_compliant_class_fails(self):
+        from codeframe.core.agent_adapter import AgentAdapter
+
+        class NotAnAdapter:
+            pass
+
+        assert not isinstance(NotAnAdapter(), AgentAdapter)
+
+    def test_partial_implementation_fails(self):
+        from codeframe.core.agent_adapter import AgentAdapter
+
+        class PartialAdapter:
+            def execute(self, task_prompt, workspace_path, context, timeout_ms=0):
+                pass
+            # Missing: stream_events, name, requires_api_key
+
+        assert not isinstance(PartialAdapter(), AgentAdapter)
+
+    def test_execute_returns_agent_result(self):
+        from codeframe.core.agent_adapter import (
+            AgentContext, AgentResult, AgentResultStatus,
+        )
+        FakeAdapter = self._make_compliant_class()
+        adapter = FakeAdapter()
+        ctx = AgentContext(task_id="1", task_title="Test", task_description="Test")
+        result = adapter.execute("do something", Path("/tmp"), ctx)
+        assert isinstance(result, AgentResult)
+        assert result.status == AgentResultStatus.COMPLETED
+
+    def test_stream_events_yields_agent_events(self):
+        from codeframe.core.agent_adapter import AgentEvent
+        FakeAdapter = self._make_compliant_class()
+        adapter = FakeAdapter()
+        events_list = list(adapter.stream_events())
+        assert len(events_list) == 1
+        assert isinstance(events_list[0], AgentEvent)