From 6e0b7a804f8a92e080caf3b7237164d1bb1469ec Mon Sep 17 00:00:00 2001 From: Test User Date: Mon, 9 Mar 2026 16:12:39 -0700 Subject: [PATCH] fix(core): reconcile duplicate AgentAdapter protocols into canonical location PR #408 already introduced AgentAdapter at codeframe/core/adapters/agent_adapter.py with run()-based interface used by 14 modules. PR #409 created a duplicate at codeframe/core/agent_adapter.py with execute()-based interface used by nothing. This reconciles them: - Merge new types (AgentContext, AdapterTokenUsage, AgentResultStatus) into canonical - Add token_usage/duration_ms to AgentResult, message/timestamp to AgentEvent - Remove duplicate codeframe/core/agent_adapter.py - Update tests to import from canonical location - Export new types from codeframe.core.adapters.__init__ --- codeframe/core/adapters/__init__.py | 6 + codeframe/core/adapters/agent_adapter.py | 55 ++++++- codeframe/core/agent_adapter.py | 100 ------------ tests/core/test_agent_adapter.py | 197 +++++------------------ 4 files changed, 104 insertions(+), 254 deletions(-) delete mode 100644 codeframe/core/agent_adapter.py diff --git a/codeframe/core/adapters/__init__.py b/codeframe/core/adapters/__init__.py index 3051ec27..2b80cfe2 100644 --- a/codeframe/core/adapters/__init__.py +++ b/codeframe/core/adapters/__init__.py @@ -1,7 +1,10 @@ from codeframe.core.adapters.agent_adapter import ( + AdapterTokenUsage, AgentAdapter, + AgentContext, AgentEvent, AgentResult, + AgentResultStatus, ) from codeframe.core.adapters.builtin import ( BuiltinPlanAdapter, @@ -13,9 +16,12 @@ from codeframe.core.adapters.verification_wrapper import VerificationWrapper __all__ = [ + "AdapterTokenUsage", "AgentAdapter", + "AgentContext", "AgentEvent", "AgentResult", + "AgentResultStatus", "BuiltinPlanAdapter", "BuiltinReactAdapter", "ClaudeCodeAdapter", diff --git a/codeframe/core/adapters/agent_adapter.py b/codeframe/core/adapters/agent_adapter.py index 683d087c..e92dcbc4 100644 --- a/codeframe/core/adapters/agent_adapter.py +++ b/codeframe/core/adapters/agent_adapter.py @@ -1,12 +1,61 @@ -"""Agent adapter protocol for delegating task execution to external coding agents.""" +"""Agent adapter protocol for delegating task execution to external coding agents. + +Defines the interface that any coding agent (Claude Code, Codex, Aider, built-in) +must implement to be used as a CodeFrame execution engine, plus the supporting +types for context, results, events, and token tracking. +""" from __future__ import annotations from dataclasses import dataclass, field +from datetime import datetime, timezone +from enum import Enum from pathlib import Path from typing import Callable, Literal, 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: """Result from an agent adapter execution.""" @@ -16,6 +65,8 @@ class AgentResult: modified_files: 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 @@ -24,6 +75,8 @@ class AgentEvent: type: str # "progress", "tool_call", "output", "error" data: dict = field(default_factory=dict) + message: str = "" + timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) @runtime_checkable diff --git a/codeframe/core/agent_adapter.py b/codeframe/core/agent_adapter.py deleted file mode 100644 index 8eb68856..00000000 --- a/codeframe/core/agent_adapter.py +++ /dev/null @@ -1,100 +0,0 @@ -"""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]: ... diff --git a/tests/core/test_agent_adapter.py b/tests/core/test_agent_adapter.py index 4d12373f..34f4364a 100644 --- a/tests/core/test_agent_adapter.py +++ b/tests/core/test_agent_adapter.py @@ -1,16 +1,18 @@ """Tests for AgentAdapter protocol and supporting types. +Tests the types added by #409 (AgentContext, AdapterTokenUsage, AgentResultStatus) +merged into the canonical protocol at codeframe.core.adapters.agent_adapter. + Validates: - Dataclass construction with defaults and full params - AgentResultStatus enum values -- AgentAdapter protocol compliance via @runtime_checkable -- Streaming iterator contract +- AgentContext field completeness +- AdapterTokenUsage arithmetic +- AgentResult and AgentEvent field extensions """ import pytest from datetime import datetime, timezone -from pathlib import Path -from typing import Iterator pytestmark = pytest.mark.v2 @@ -19,23 +21,23 @@ class TestAgentResultStatus: """AgentResultStatus enum covers all terminal states.""" def test_has_completed(self): - from codeframe.core.agent_adapter import AgentResultStatus + from codeframe.core.adapters.agent_adapter import AgentResultStatus assert AgentResultStatus.COMPLETED.value == "completed" def test_has_failed(self): - from codeframe.core.agent_adapter import AgentResultStatus + from codeframe.core.adapters.agent_adapter import AgentResultStatus assert AgentResultStatus.FAILED.value == "failed" def test_has_blocked(self): - from codeframe.core.agent_adapter import AgentResultStatus + from codeframe.core.adapters.agent_adapter import AgentResultStatus assert AgentResultStatus.BLOCKED.value == "blocked" def test_has_timeout(self): - from codeframe.core.agent_adapter import AgentResultStatus + from codeframe.core.adapters.agent_adapter import AgentResultStatus assert AgentResultStatus.TIMEOUT.value == "timeout" def test_is_str_enum(self): - from codeframe.core.agent_adapter import AgentResultStatus + from codeframe.core.adapters.agent_adapter import AgentResultStatus assert isinstance(AgentResultStatus.COMPLETED, str) @@ -43,7 +45,7 @@ class TestAdapterTokenUsage: """Lightweight token usage dataclass.""" def test_minimal_construction(self): - from codeframe.core.agent_adapter import AdapterTokenUsage + from codeframe.core.adapters.agent_adapter import AdapterTokenUsage usage = AdapterTokenUsage(input_tokens=100, output_tokens=50) assert usage.input_tokens == 100 assert usage.output_tokens == 50 @@ -51,7 +53,7 @@ def test_minimal_construction(self): assert usage.cost_usd is None def test_full_construction(self): - from codeframe.core.agent_adapter import AdapterTokenUsage + from codeframe.core.adapters.agent_adapter import AdapterTokenUsage usage = AdapterTokenUsage( input_tokens=1000, output_tokens=500, @@ -62,7 +64,7 @@ def test_full_construction(self): assert usage.cost_usd == 0.015 def test_total_tokens(self): - from codeframe.core.agent_adapter import AdapterTokenUsage + from codeframe.core.adapters.agent_adapter import AdapterTokenUsage usage = AdapterTokenUsage(input_tokens=100, output_tokens=50) assert usage.total_tokens == 150 @@ -71,7 +73,7 @@ class TestAgentContext: """AgentContext captures all context CodeFrame provides to engines.""" def test_minimal_construction(self): - from codeframe.core.agent_adapter import AgentContext + from codeframe.core.adapters.agent_adapter import AgentContext ctx = AgentContext( task_id="task-1", task_title="Implement feature X", @@ -90,7 +92,7 @@ def test_minimal_construction(self): assert ctx.previous_errors == [] def test_full_construction(self): - from codeframe.core.agent_adapter import AgentContext + from codeframe.core.adapters.agent_adapter import AgentContext ctx = AgentContext( task_id="task-42", task_title="Fix auth bug", @@ -113,172 +115,61 @@ def test_full_construction(self): def test_list_defaults_are_independent(self): """Ensure default_factory creates independent lists (no shared mutable state).""" - from codeframe.core.agent_adapter import AgentContext + from codeframe.core.adapters.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 +class TestAgentResultExtensions: + """Tests for #409 extensions to AgentResult (token_usage, duration_ms).""" def test_result_with_token_usage(self): - from codeframe.core.agent_adapter import ( - AdapterTokenUsage, AgentResult, AgentResultStatus, + from codeframe.core.adapters.agent_adapter import ( + AdapterTokenUsage, AgentResult, ) result = AgentResult( - status=AgentResultStatus.COMPLETED, - summary="Done", + status="completed", + output="Done", token_usage=AdapterTokenUsage(input_tokens=1000, output_tokens=500), - files_modified=["auth.py"], - files_created=["tests/test_auth.py"], + modified_files=["auth.py"], duration_ms=12000, ) assert result.token_usage.total_tokens == 1500 - assert result.files_modified == ["auth.py"] + assert result.modified_files == ["auth.py"] assert result.duration_ms == 12000 + def test_result_defaults_for_new_fields(self): + from codeframe.core.adapters.agent_adapter import AgentResult + result = AgentResult(status="completed") + assert result.token_usage is None + assert result.duration_ms == 0 -class TestAgentEvent: - """AgentEvent supports progress streaming.""" - def test_minimal_construction(self): - from codeframe.core.agent_adapter import AgentEvent +class TestAgentEventExtensions: + """Tests for #409 extensions to AgentEvent (message, timestamp).""" + + def test_event_with_message(self): + from codeframe.core.adapters.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 + def test_event_with_explicit_timestamp(self): + from codeframe.core.adapters.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}, + data={"file": "auth.py"}, ) 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 - + assert event.data["file"] == "auth.py" -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) + def test_event_defaults_for_new_fields(self): + from codeframe.core.adapters.agent_adapter import AgentEvent + event = AgentEvent(type="output") + assert event.message == "" + assert isinstance(event.timestamp, datetime)