feat(core): Agent Adapter protocol definition - #429
Conversation
…409) Introduces the foundational types for Phase 4 Agent Adapter Architecture: - AgentAdapter Protocol (runtime_checkable) for structural subtyping - AgentContext dataclass with all context CodeFrame assembles - AgentResult dataclass for engine outcomes - AgentEvent dataclass for progress streaming - AgentResultStatus enum (completed/failed/blocked/timeout) - AdapterTokenUsage lightweight dataclass (decoupled from billing model) Closes #409
WalkthroughThis PR adds Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
Comment |
Code Review - PR 429: Agent Adapter Protocol DefinitionThis is a clean, well-scoped Phase 4 foundation. The design rationale is sound - a Issues / Suggestions1. AgentEvent.metadata missing type parameter Bare 2. stream_events() / execute() relationship is undocumented The Protocol does not clarify when 3. requires_api_key dict contract is undocumented What do the keys and values represent? The FakeAdapter in tests returns 4. attempt field semantics
5. Intra-method imports in tests All 23 tests import from within each test method rather than at module top. Module-level imports are standard practice and make test failures easier to distinguish from import failures. Consider consolidating. Minor Observations (non-blocking)
What is Working Well
Verdict: Solid foundation. Key actionable items are the |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@codeframe/core/agent_adapter.py`:
- Around line 82-94: The AgentAdapter protocol doesn't match implementations:
change the protocol to declare run(...) instead of execute(...), remove
stream_events() and requires_api_key from the core AgentAdapter, and create two
small optional protocols—StreamableAgent (with stream_events() ->
Iterator[AgentEvent]) and ApiKeyAgent (with requires_api_key: bool or def
requires_api_key() -> bool)—so BuiltinReactAdapter, BuiltinPlanAdapter, and
SubprocessAdapter will be recognized as implementing AgentAdapter (and can opt
into StreamableAgent/ApiKeyAgent if they support those features); update any
type checks or isinstance usages to reference the new optional protocols where
streaming or API-key behavior is required.
In `@tests/core/test_agent_adapter.py`:
- Around line 10-13: This module is missing the required v2 pytest marker; add
an import for pytest and declare a module-level marker variable named pytestmark
set to pytest.mark.v2 (i.e., add "import pytest" and "pytestmark =
pytest.mark.v2") so the entire test file is marked as v2; locate this in
tests/core/test_agent_adapter.py near the top-level imports to apply the marker
for the whole module.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 6ecfc7a1-d52d-4d3b-9382-9e46e602df60
📒 Files selected for processing (2)
codeframe/core/agent_adapter.pytests/core/test_agent_adapter.py
| @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]: ... |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
python - <<'PY'
from typing import Protocol, runtime_checkable
`@runtime_checkable`
class BaseAdapter(Protocol):
def execute(self) -> int: ...
def stream_events(self): ...
class NoStream:
def execute(self) -> int:
return 1
class WithStream:
def execute(self) -> int:
return 1
def stream_events(self):
return iter(())
print("NoStream:", isinstance(NoStream(), BaseAdapter))
print("WithStream:", isinstance(WithStream(), BaseAdapter))
PYRepository: frankbria/codeframe
Length of output: 95
🏁 Script executed:
# First, locate and read the agent_adapter.py file to verify the current protocol
fd agent_adapter.py --type fRepository: frankbria/codeframe
Length of output: 210
🏁 Script executed:
# Search for AgentAdapter implementations to see if any omit stream_events
rg "class.*AgentAdapter" -A 10 --type pyRepository: frankbria/codeframe
Length of output: 3286
🏁 Script executed:
# Check how stream_events is used across the codebase
rg "stream_events" --type py -B 2 -A 2Repository: frankbria/codeframe
Length of output: 2599
🏁 Script executed:
# Read the actual agent_adapter.py file to see the complete protocol definition
cat -n codeframe/core/agent_adapter.py | head -100Repository: frankbria/codeframe
Length of output: 3469
🏁 Script executed:
# Check the test that validates protocol compliance without stream_events
cat -n tests/core/test_agent_adapter.py | grep -A 15 "PartialAdapter"Repository: frankbria/codeframe
Length of output: 1144
🏁 Script executed:
# Look at how AgentAdapter is actually used in the codebase
rg "isinstance.*AgentAdapter" --type py -B 2 -A 2Repository: frankbria/codeframe
Length of output: 6108
🏁 Script executed:
# Check if any existing adapter implementations skip stream_events
rg "class.*Adapter" codeframe/core/adapters --type py -A 30 | grep -E "(class|def stream_events)" | head -40Repository: frankbria/codeframe
Length of output: 816
🏁 Script executed:
# Look at how stream_events is actually called in the codebase
rg "\.stream_events\(\)" --type py -B 3 -A 3Repository: frankbria/codeframe
Length of output: 546
🏁 Script executed:
# Check the adapters subdirectory structure
ls -la codeframe/core/adapters/Repository: frankbria/codeframe
Length of output: 630
🏁 Script executed:
# Check if there's any design documentation or objectives about streaming being optional
fd -e md -e txt | xargs rg -l "streaming.*optional|optional.*stream" 2>/dev/nullRepository: frankbria/codeframe
Length of output: 72
🏁 Script executed:
# Check if there are any comments in the adapters about streaming being optional
rg "optional|streaming" codeframe/core/adapters --type py -B 2 -A 2 | head -50Repository: frankbria/codeframe
Length of output: 1364
🏁 Script executed:
# Read the README to check for design objectives about optional streaming
cat README.md | head -100Repository: frankbria/codeframe
Length of output: 3939
🏁 Script executed:
# Also check if there's a CLAUDE.md or design doc that mentions streaming
fd -e md | xargs grep -l "objective\|design\|streaming" 2>/dev/null | head -10Repository: frankbria/codeframe
Length of output: 385
🏁 Script executed:
# Check the actual implementation of stream_events across existing adapters
for file in codeframe/core/adapters/*.py; do echo "=== $file ==="; grep -A 5 "def stream_events" "$file" 2>/dev/null || echo "No stream_events found"; doneRepository: frankbria/codeframe
Length of output: 560
🏁 Script executed:
# Search for stream_events implementations more broadly
rg "def stream_events" --type py -B 3 -A 5Repository: frankbria/codeframe
Length of output: 1106
🏁 Script executed:
# Check the builtin adapter more carefully for stream_events
cat -n codeframe/core/adapters/builtin.py | head -150Repository: frankbria/codeframe
Length of output: 6191
🏁 Script executed:
# Check the subprocess adapter which is the base class
cat -n codeframe/core/adapters/subprocess_adapter.py | head -100Repository: frankbria/codeframe
Length of output: 4019
🏁 Script executed:
# Check if BuiltinReactAdapter actually implements stream_events anywhere
cat -n codeframe/core/adapters/builtin.py | tail -100Repository: frankbria/codeframe
Length of output: 4288
🏁 Script executed:
# Check if these adapters implement requires_api_key property
rg "requires_api_key" --type py -B 2 -A 3Repository: frankbria/codeframe
Length of output: 3102
🏁 Script executed:
# Look at a test that checks isinstance for these actual adapters
cat -n tests/core/adapters/test_builtin.py | head -80Repository: frankbria/codeframe
Length of output: 3549
🏁 Script executed:
# Check if BuiltinReactAdapter implements execute method (not just run)
rg "def execute" codeframe/core/adapters/builtin.pyRepository: frankbria/codeframe
Length of output: 45
🏁 Script executed:
# Check if there are any methods we're missing in the builtin adapters
cat -n codeframe/core/adapters/builtin.py | grep -E "def |@property"Repository: frankbria/codeframe
Length of output: 520
🏁 Script executed:
# Run the actual test to see if it passes
cd codeframe && python -m pytest tests/core/adapters/test_builtin.py::TestBuiltinReactAdapter::test_conforms_to_protocol -v 2>&1 || echo "Test failed or pytest not available"Repository: frankbria/codeframe
Length of output: 1035
🏁 Script executed:
# Find the actual test file location
find . -name "*test_builtin.py" -type fRepository: frankbria/codeframe
Length of output: 100
🏁 Script executed:
# Check if there's an execute method defined somewhere else or if run is being aliased
rg "def execute|execute.*=" codeframe/core/adapters/ --type pyRepository: frankbria/codeframe
Length of output: 45
🏁 Script executed:
# Look for where AgentAdapter methods are actually being used in the tests
cat -n tests/core/adapters/test_builtin.py | grep -A 5 "conforms_to_protocol"Repository: frankbria/codeframe
Length of output: 705
🏁 Script executed:
# Check if BuiltinReactAdapter has stream_events and requires_api_key somewhere
grep -n "stream_events\|requires_api_key\|execute" codeframe/core/adapters/builtin.pyRepository: frankbria/codeframe
Length of output: 45
🏁 Script executed:
# Run the actual test to see what happens
python -m pytest ./tests/core/adapters/test_builtin.py::TestBuiltinReactAdapter::test_conforms_to_protocol -xvs 2>&1 | head -50Repository: frankbria/codeframe
Length of output: 2625
🏁 Script executed:
# Check if maybe there's a base class or mixin being used
head -20 codeframe/core/adapters/builtin.pyRepository: frankbria/codeframe
Length of output: 761
🏁 Script executed:
# Check if BuiltinReactAdapter is missing execute, stream_events, requires_api_key
python3 << 'EOF'
from typing import Protocol, runtime_checkable, Iterator
`@runtime_checkable`
class AgentAdapter(Protocol):
def execute(self): ...
def stream_events(self) -> Iterator: ...
`@property`
def name(self) -> str: ...
`@property`
def requires_api_key(self) -> dict: ...
class BuiltinReactAdapterMock:
`@property`
def name(self) -> str:
return "react"
def run(self):
pass
# Missing: execute, stream_events, requires_api_key
adapter = BuiltinReactAdapterMock()
print("BuiltinReactAdapter conforms to AgentAdapter:", isinstance(adapter, AgentAdapter))
print("\nMissing members:")
print("- execute:", not hasattr(adapter, 'execute'))
print("- stream_events:", not hasattr(adapter, 'stream_events'))
print("- requires_api_key:", not hasattr(adapter, 'requires_api_key'))
EOFRepository: frankbria/codeframe
Length of output: 195
🏁 Script executed:
# Let's check what the actual protocol definition looks like and what adapters implement
python3 << 'EOF'
import sys
sys.path.insert(0, '/home/jailuser/git')
# Check what methods BuiltinReactAdapter actually has
from codeframe.core.adapters.builtin import BuiltinReactAdapter
print("BuiltinReactAdapter methods and properties:")
for attr in dir(BuiltinReactAdapter):
if not attr.startswith('_'):
print(f" {attr}")
# Check specifically for protocol required members
required = ['execute', 'stream_events', 'name', 'requires_api_key']
actual = set(dir(BuiltinReactAdapter))
for member in required:
status = "✓" if member in actual else "✗"
print(f"{status} {member}")
EOFRepository: frankbria/codeframe
Length of output: 801
🏁 Script executed:
# Check if there's any dynamic method assignment or aliasing happening
rg "execute.*=|setattr.*execute" codeframe/core/adapters/ --type pyRepository: frankbria/codeframe
Length of output: 45
🏁 Script executed:
# Check if the adapters actually define execute or if they're supposed to alias run
cat -n codeframe/core/adapters/subprocess_adapter.py | grep -A 20 "def run"Repository: frankbria/codeframe
Length of output: 901
🏁 Script executed:
# Let's check what the protocol actually expects vs what's being used
rg "\.execute\(" codeframe/core --type py -B 2 -A 2 | head -40Repository: frankbria/codeframe
Length of output: 1785
🏁 Script executed:
# Look at the complete protocol definition to see if there's something with aliases or __getattr__
cat -n codeframe/core/agent_adapter.pyRepository: frankbria/codeframe
Length of output: 3469
🏁 Script executed:
# Check if there are two different AgentAdapter definitions or if one is importing from another
find codeframe -name "*.py" -type f -exec grep -l "class AgentAdapter" {} \;Repository: frankbria/codeframe
Length of output: 135
🏁 Script executed:
# Look at what the comment about the "linked objective" might refer to - check for design docs
cat CLAUDE.md 2>/dev/null | grep -i "stream\|optional" | head -20Repository: frankbria/codeframe
Length of output: 1293
The AgentAdapter protocol requires methods that no adapter implementations actually provide.
The protocol at lines 82–100 mandates execute(), stream_events(), name, and requires_api_key. However, all actual adapters (BuiltinReactAdapter, BuiltinPlanAdapter, SubprocessAdapter) implement run() instead of execute(), and are missing stream_events() and requires_api_key entirely. This means isinstance(adapter, AgentAdapter) would return False for every adapter in the codebase, despite test claims of protocol conformance. The protocol definition needs to match actual implementations—either rename execute() to run(), make stream_events() optional by splitting it into a separate protocol, or update adapters to provide the required surface.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@codeframe/core/agent_adapter.py` around lines 82 - 94, The AgentAdapter
protocol doesn't match implementations: change the protocol to declare run(...)
instead of execute(...), remove stream_events() and requires_api_key from the
core AgentAdapter, and create two small optional protocols—StreamableAgent (with
stream_events() -> Iterator[AgentEvent]) and ApiKeyAgent (with requires_api_key:
bool or def requires_api_key() -> bool)—so BuiltinReactAdapter,
BuiltinPlanAdapter, and SubprocessAdapter will be recognized as implementing
AgentAdapter (and can opt into StreamableAgent/ApiKeyAgent if they support those
features); update any type checks or isinstance usages to reference the new
optional protocols where streaming or API-key behavior is required.
Code Review: Agent Adapter Protocol DefinitionThis is a well-scoped and well-executed implementation. The protocol layer is clean and minimal, exactly what Phase 4 needs as a foundation. Strengths
Issues and Suggestions
Overall Solid foundation for the adapter architecture. Items 1, 4, and 6 are easy fixes. Items 2, 3, and 5 are design questions that can land in follow-up issues (410-417). Ready to approve once item 6 is addressed. |
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@tests/core/test_agent_adapter.py`:
- Around line 257-265: The test currently treats stream_events as a required
attribute but streaming is optional; modify test_partial_implementation_fails so
it verifies failure only when truly required members are missing: keep execute
but remove stream_events from the “missing” list and instead omit a genuinely
required attribute such as name or requires_api_key (or both) on PartialAdapter,
and assert that an instance of that incomplete PartialAdapter is not an
AgentAdapter; reference test_partial_implementation_fails, AgentAdapter,
PartialAdapter, stream_events, name, and requires_api_key when making the
change.
- Around line 243-284: Add a real compatibility test that exercises the existing
ReactAgent adaptation path: import ReactAgent (the concrete agent
implementation), instantiate it, and assert it satisfies the AgentAdapter
protocol just like FakeAdapter does—i.e., assert isinstance(ReactAgent(),
AgentAdapter) and also call execute(...) and stream_events() on the ReactAgent
instance and assert the returned types are AgentResult and AgentEvent
respectively (matching the existing tests
test_compliant_class_satisfies_protocol, test_execute_returns_agent_result, and
test_stream_events_yields_agent_events).
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 208a8b1a-0add-4878-a3ee-9178052a6733
📒 Files selected for processing (1)
tests/core/test_agent_adapter.py
| 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) |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Add at least one compatibility test against the real ReactAgent adaptation path.
These assertions only prove the protocol against FakeAdapter. The acceptance criteria for this PR explicitly call out ensuring the existing ReactAgent can be adapted, and that can still regress silently with the current suite.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tests/core/test_agent_adapter.py` around lines 243 - 284, Add a real
compatibility test that exercises the existing ReactAgent adaptation path:
import ReactAgent (the concrete agent implementation), instantiate it, and
assert it satisfies the AgentAdapter protocol just like FakeAdapter does—i.e.,
assert isinstance(ReactAgent(), AgentAdapter) and also call execute(...) and
stream_events() on the ReactAgent instance and assert the returned types are
AgentResult and AgentEvent respectively (matching the existing tests
test_compliant_class_satisfies_protocol, test_execute_returns_agent_result, and
test_stream_events_yields_agent_events).
| 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) |
There was a problem hiding this comment.
This test hard-codes stream_events as mandatory.
Issue #409 describes streaming support as optional, but this negative case rejects adapters that omit stream_events. That bakes the wrong contract into the suite and will fail valid non-streaming adapters.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tests/core/test_agent_adapter.py` around lines 257 - 265, The test currently
treats stream_events as a required attribute but streaming is optional; modify
test_partial_implementation_fails so it verifies failure only when truly
required members are missing: keep execute but remove stream_events from the
“missing” list and instead omit a genuinely required attribute such as name or
requires_api_key (or both) on PartialAdapter, and assert that an instance of
that incomplete PartialAdapter is not an AgentAdapter; reference
test_partial_implementation_fails, AgentAdapter, PartialAdapter, stream_events,
name, and requires_api_key when making the change.
1 participant
Summary
Implements #409: [Phase 4] Agent Adapter Protocol Definition
AgentAdapterProtocol (runtime_checkable) for structural subtyping — any coding agent just needs the right methodsAgentContextdataclass captures all context CodeFrame assembles (PRD, tech stack, preferences, files, gates)AgentResultdataclass for engine outcomes (status, files changed, errors, blockers)AgentEventdataclass for progress streaming during executionAgentResultStatusenum: completed/failed/blocked/timeoutAdapterTokenUsagelightweight dataclass — decoupled from the heavy billingTokenUsagemodelAcceptance Criteria
AgentAdapterProtocol defined withexecute(),stream_events(),name,requires_api_keyAgentContextcaptures all context CodeFrame currently assemblesAgentResultcaptures outcome, modified files, errors, blockersAgentEventsupports progress streamingReactAgentcan be wrapped to satisfy the protocol (design validation via tests)Test Plan
Implementation Notes
AdapterTokenUsageinstead of importingTokenUsagefrommodels.py— the existing model is a heavy Pydantic BaseModel with billing fields (agent_id,project_id,estimated_cost_usd) that would force adapter implementors to know about CodeFrame internals__init__.pyexports — consumers import directly fromcodeframe.core.agent_adapter, matching how other core modules are consumed@runtime_checkableenablingisinstance()checks at adapter registration timeCloses #409
Summary by CodeRabbit
New Features
Tests