[Phase 2.5-F] Verify ReAct engine works via API routes

[frankbria]

Summary

Verify that the ReAct engine can be invoked through the existing v2 API routes, not just the CLI. This ensures the server layer (Phase 2) works with the new engine.

Acceptance Criteria

• API endpoint accepts engine parameter:
  • `POST /api/v2/tasks/{id}/execute` accepts `engine` field in request body
  • Default: `"plan"` (backward compatible)
  • Optional: `"react"` to use new engine
• `codeframe/ui/models.py` updated with engine field in execution request model
• `codeframe/ui/routers/tasks_v2.py` passes engine to runtime
• SSE streaming (`/api/v2/tasks/{id}/stream`) works with ReAct engine events
• Phase-based events (EXPLORING, CREATING, etc.) appear in SSE stream
• Integration test: API call triggers ReAct execution and streams events
• All existing API tests pass
• All existing CLI tests pass

Implementation Notes

• The v2 routers are thin adapters over core — the change is adding the engine param passthrough
• SSE streaming in `streaming_v2.py` should already work if events use the existing event format
• Test with curl or httpx, not browser

Dependencies

• Phase F CLI validation: [Phase 2.5-F] End-to-end CLI validation with cf-test project #353 (CLI must work first)
• Existing: v2 API routers, SSE streaming

Reference

• `docs/AGENT_V3_UNIFIED_PLAN.md` (CLI-first, then API)
• `codeframe/ui/routers/tasks_v2.py` (existing task execution endpoint)
• `codeframe/ui/routers/streaming_v2.py` (SSE streaming)

[traycerai]

Plan

Observations

The codebase currently uses a single "plan" engine (Agent + Planner architecture in file:codeframe/core/agent.py and file:codeframe/core/planner.py). The v2 API routes in file:codeframe/ui/routers/tasks_v2.py delegate to file:codeframe/core/runtime.py for task execution, which calls execute_agent(). The SSE streaming infrastructure in file:codeframe/ui/routers/streaming_v2.py already supports structured events (ProgressEvent, CompletionEvent, etc.) via EventPublisher. No ReAct engine implementation exists yet—this ticket prepares the API layer for future engine selection.

Approach

The implementation follows a thin adapter pattern: add an optional engine parameter to the execution request models and pass it through the API → runtime → agent execution chain. The default value "plan" maintains backward compatibility with existing behavior. The SSE streaming layer requires no changes since it already handles generic ExecutionEvent types. The focus is on parameter plumbing and validation, not engine implementation, as the ReAct engine will be added separately in Phase 2.5-F.

Implementation Steps

1. Update Request Models

File: file:codeframe/ui/models.py

Add engine field to execution-related request models:

• Locate StartExecutionRequest class (around line 77-100)
• Add new field after retry_count:

  engine: str = Field(
      "plan",
      description="Execution engine: 'plan' (default, current planner-based) or 'react' (new ReAct engine)"
  )

• Add validation to ensure only valid engine values are accepted:

  @model_validator(mode="after")
  def validate_engine(self):
      valid_engines = {"plan", "react"}
      if self.engine not in valid_engines:
          raise ValueError(f"Invalid engine: {self.engine}. Must be one of {valid_engines}")
      return self

2. Update API Route Handler

File: file:codeframe/ui/routers/tasks_v2.py

Modify the start_execution endpoint (line 480-552) to accept and pass through the engine parameter:

• In the conductor.start_batch() call (around line 528-535), add engine parameter:

  batch = conductor.start_batch(
      workspace,
      task_ids=task_ids,
      strategy=body.strategy,
      max_parallel=body.max_parallel,
      retry_count=body.retry_count,
      on_failure="continue",
      engine=body.engine,  # NEW: pass engine parameter
  )

• Update the start_single_task endpoint (line 555-650) similarly:

  • Add engine query parameter after verbose:

    engine: str = Query("plan", description="Execution engine (plan or react)")

  • Pass to runtime.execute_agent() call (around line 607-613):

    runtime.execute_agent(
        workspace,
        run,
        dry_run=dry_run,
        verbose=verbose,
        event_publisher=publisher,
        engine=engine,  # NEW
    )

3. Update Conductor Module

File: file:codeframe/core/conductor.py

Modify start_batch() function signature to accept and store engine parameter:

• Update function signature (around line 408):

  def start_batch(
      workspace: Workspace,
      task_ids: list[str],
      strategy: str = "serial",
      max_parallel: int = 4,
      retry_count: int = 0,
      on_failure: str = "continue",
      dry_run: bool = False,
      engine: str = "plan",  # NEW parameter
  ) -> BatchRun:

• Add engine field to BatchRun dataclass (around line 238-280):

  @dataclass
  class BatchRun:
      # ... existing fields ...
      engine: str = "plan"  # NEW: execution engine to use

• Store engine in batch record when calling _save_batch() (around line 450-480)

• Pass engine to _execute_single_task() calls in _execute_serial() and _execute_parallel() functions (around lines 1143-1605)

• Update _execute_single_task() to pass engine to runtime.execute_agent() (around line 1400-1450)

4. Update Runtime Module

File: file:codeframe/core/runtime.py

Modify execute_agent() function to accept engine parameter:

• Update function signature (line 591-599):

  def execute_agent(
      workspace: Workspace,
      run: Run,
      dry_run: bool = False,
      debug: bool = False,
      verbose: bool = False,
      fix_coordinator: Optional["GlobalFixCoordinator"] = None,
      event_publisher: Optional["EventPublisher"] = None,
      engine: str = "plan",  # NEW parameter
  ) -> "AgentState":

• Add engine selection logic before creating Agent instance (around line 660-671):

  # Select execution engine
  if engine == "react":
      # Future: Use ReAct engine when available
      # For now, raise NotImplementedError to fail gracefully
      raise NotImplementedError(
          "ReAct engine not yet implemented. Use engine='plan' (default) for now."
      )
  elif engine != "plan":
      raise ValueError(f"Unknown engine: {engine}. Valid options: 'plan', 'react'")
  
  # Use plan engine (current implementation)
  agent = Agent(...)

5. Update Database Schema

File: file:codeframe/core/workspace.py or migration script

Add engine column to runs table and batches table:

• For runs table:

  ALTER TABLE runs ADD COLUMN engine TEXT DEFAULT 'plan';

• For batches table (if it exists):

  ALTER TABLE batches ADD COLUMN engine TEXT DEFAULT 'plan';

• Update _save_batch() in conductor.py to persist engine field

• Update _row_to_batch() to hydrate engine field from database

6. Verify SSE Streaming Compatibility

File: file:codeframe/ui/routers/streaming_v2.py

No changes required—verify that existing event streaming works:

• The event_stream_generator() function (line 84-149) already handles generic ExecutionEvent types
• The format_sse_event() function (line 54-66) serializes events using model_dump_json()
• Phase-based events (EXPLORING, CREATING, etc.) will be emitted as ProgressEvent with appropriate phase values
• The ReAct engine (when implemented) should emit events using the same ExecutionEvent types

7. Add Integration Test

File: file:tests/api/test_tasks_v2_engine.py (new file)

Create integration test to verify engine parameter handling:

"""Integration tests for engine parameter in v2 API."""

import pytest
from fastapi.testclient import TestClient

def test_execute_with_plan_engine(client, workspace, task_id):
    """Test execution with default 'plan' engine."""
    response = client.post(
        "/api/v2/tasks/execute",
        json={
            "task_ids": [task_id],
            "strategy": "serial",
            "engine": "plan"
        },
        headers={"workspace-path": workspace.path}
    )
    assert response.status_code == 200
    assert response.json()["success"] is True

def test_execute_with_react_engine_not_implemented(client, workspace, task_id):
    """Test execution with 'react' engine raises NotImplementedError."""
    response = client.post(
        "/api/v2/tasks/execute",
        json={
            "task_ids": [task_id],
            "strategy": "serial",
            "engine": "react"
        },
        headers={"workspace-path": workspace.path}
    )
    # Should fail gracefully with 500 and NotImplementedError message
    assert response.status_code == 500
    assert "not yet implemented" in response.json()["detail"].lower()

def test_execute_with_invalid_engine(client, workspace, task_id):
    """Test execution with invalid engine value."""
    response = client.post(
        "/api/v2/tasks/execute",
        json={
            "task_ids": [task_id],
            "strategy": "serial",
            "engine": "invalid"
        },
        headers={"workspace-path": workspace.path}
    )
    assert response.status_code == 400
    assert "invalid engine" in response.json()["detail"].lower()

def test_sse_stream_with_engine_parameter(client, workspace, task_id):
    """Test SSE streaming works with engine parameter."""
    # Start execution with engine parameter
    client.post(
        f"/api/v2/tasks/{task_id}/start",
        params={"execute": True, "engine": "plan"},
        headers={"workspace-path": workspace.path}
    )
    
    # Connect to SSE stream
    with client.stream("GET", f"/api/v2/tasks/{task_id}/stream") as response:
        events = []
        for line in response.iter_lines():
            if line.startswith("data:"):
                events.append(json.loads(line[5:]))
        
        # Verify we received progress events
        progress_events = [e for e in events if e["event_type"] == "progress"]
        assert len(progress_events) > 0

8. Update Existing Tests

Ensure all existing API and CLI tests pass:

• File: file:tests/api/test_multi_agent_api.py

  • Verify tests don't break with new optional parameter
  • Add assertions that default engine is "plan"

• File: file:tests/cli/test_v2_cli_integration.py

  • Verify CLI tests still pass (CLI may not expose engine parameter yet per dependency [Phase 2.5-F] End-to-end CLI validation with cf-test project #353)

• File: file:tests/integration/test_worker_agent_execution.py

  • Update to explicitly pass engine="plan" if needed

9. Documentation Updates

File: file:docs/PHASE_2_DEVELOPER_GUIDE.md or API documentation

Document the new engine parameter:

## Execution Engine Selection

The v2 API supports multiple execution engines:

- `plan` (default): Traditional planner-based execution (Agent + Planner)
- `react` (experimental): ReAct-style reasoning and acting engine

### API Usage

```bash
# Use default plan engine
POST /api/v2/tasks/execute
{
  "task_ids": ["task-123"],
  "strategy": "serial",
  "engine": "plan"  # optional, defaults to "plan"
}

# Use ReAct engine (when available)
POST /api/v2/tasks/execute
{
  "task_ids": ["task-123"],
  "strategy": "serial",
  "engine": "react"
}

SSE Streaming

Both engines emit the same event types:

• progress: Phase transitions (planning, execution, verification)
• output: Command output
• blocker: Human input needed
• completion: Task finished
• error: Execution error

## Sequence Diagram

```mermaid
sequenceDiagram
    participant Client
    participant API as tasks_v2.py
    participant Conductor
    participant Runtime
    participant Agent as Agent/ReAct Engine
    participant SSE as streaming_v2.py
    
    Client->>API: POST /api/v2/tasks/execute<br/>{engine: "react"}
    API->>API: Validate engine parameter
    API->>Conductor: start_batch(engine="react")
    Conductor->>Runtime: execute_agent(engine="react")
    
    alt engine == "react"
        Runtime->>Agent: Create ReAct engine instance
    else engine == "plan"
        Runtime->>Agent: Create Agent instance (default)
    end
    
    Agent->>SSE: Emit ProgressEvent(phase="EXPLORING")
    SSE->>Client: SSE: data: {event_type: "progress", phase: "EXPLORING"}
    
    Agent->>SSE: Emit ProgressEvent(phase="CREATING")
    SSE->>Client: SSE: data: {event_type: "progress", phase: "CREATING"}
    
    Agent->>SSE: Emit CompletionEvent(status="completed")
    SSE->>Client: SSE: data: {event_type: "completion"}
    
    Runtime-->>Conductor: AgentState
    Conductor-->>API: BatchRun
    API-->>Client: {success: true, batch_id: "..."}

Validation Checklist

• StartExecutionRequest model has engine field with validation
• POST /api/v2/tasks/execute accepts and validates engine parameter
• POST /api/v2/tasks/{id}/start accepts engine query parameter
• conductor.start_batch() accepts and stores engine parameter
• runtime.execute_agent() accepts engine parameter and validates it
• Database schema includes engine column in runs and batches tables
• SSE streaming works with both engines (verified via existing event types)
• Integration test covers engine parameter validation and execution
• All existing API tests pass without modification
• All existing CLI tests pass (CLI may not expose engine yet)
• Documentation updated with engine parameter usage

Import In IDE

🤖 Prompt for AI Agents

## Observations

The codebase currently uses a single "plan" engine (Agent + Planner architecture in `file:codeframe/core/agent.py` and `file:codeframe/core/planner.py`). The v2 API routes in `file:codeframe/ui/routers/tasks_v2.py` delegate to `file:codeframe/core/runtime.py` for task execution, which calls `execute_agent()`. The SSE streaming infrastructure in `file:codeframe/ui/routers/streaming_v2.py` already supports structured events (ProgressEvent, CompletionEvent, etc.) via EventPublisher. No ReAct engine implementation exists yet—this ticket prepares the API layer for future engine selection.

## Approach

The implementation follows a thin adapter pattern: add an optional `engine` parameter to the execution request models and pass it through the API → runtime → agent execution chain. The default value "plan" maintains backward compatibility with existing behavior. The SSE streaming layer requires no changes since it already handles generic ExecutionEvent types. The focus is on parameter plumbing and validation, not engine implementation, as the ReAct engine will be added separately in Phase 2.5-F.

## Implementation Steps

### 1. Update Request Models

**File:** `file:codeframe/ui/models.py`

Add `engine` field to execution-related request models:

- Locate `StartExecutionRequest` class (around line 77-100)
- Add new field after `retry_count`:
  ```python
  engine: str = Field(
      "plan",
      description="Execution engine: 'plan' (default, current planner-based) or 'react' (new ReAct engine)"
  )
  ```
- Add validation to ensure only valid engine values are accepted:
  ```python
  @model_validator(mode="after")
  def validate_engine(self):
      valid_engines = {"plan", "react"}
      if self.engine not in valid_engines:
          raise ValueError(f"Invalid engine: {self.engine}. Must be one of {valid_engines}")
      return self
  ```

### 2. Update API Route Handler

**File:** `file:codeframe/ui/routers/tasks_v2.py`

Modify the `start_execution` endpoint (line 480-552) to accept and pass through the engine parameter:

- In the `conductor.start_batch()` call (around line 528-535), add `engine` parameter:
  ```python
  batch = conductor.start_batch(
      workspace,
      task_ids=task_ids,
      strategy=body.strategy,
      max_parallel=body.max_parallel,
      retry_count=body.retry_count,
      on_failure="continue",
      engine=body.engine,  # NEW: pass engine parameter
  )
  ```

- Update the `start_single_task` endpoint (line 555-650) similarly:
  - Add `engine` query parameter after `verbose`:
    ```python
    engine: str = Query("plan", description="Execution engine (plan or react)")
    ```
  - Pass to `runtime.execute_agent()` call (around line 607-613):
    ```python
    runtime.execute_agent(
        workspace,
        run,
        dry_run=dry_run,
        verbose=verbose,
        event_publisher=publisher,
        engine=engine,  # NEW
    )
    ```

### 3. Update Conductor Module

**File:** `file:codeframe/core/conductor.py`

Modify `start_batch()` function signature to accept and store engine parameter:

- Update function signature (around line 408):
  ```python
  def start_batch(
      workspace: Workspace,
      task_ids: list[str],
      strategy: str = "serial",
      max_parallel: int = 4,
      retry_count: int = 0,
      on_failure: str = "continue",
      dry_run: bool = False,
      engine: str = "plan",  # NEW parameter
  ) -> BatchRun:
  ```

- Add `engine` field to `BatchRun` dataclass (around line 238-280):
  ```python
  @dataclass
  class BatchRun:
      # ... existing fields ...
      engine: str = "plan"  # NEW: execution engine to use
  ```

- Store engine in batch record when calling `_save_batch()` (around line 450-480)

- Pass engine to `_execute_single_task()` calls in `_execute_serial()` and `_execute_parallel()` functions (around lines 1143-1605)

- Update `_execute_single_task()` to pass engine to `runtime.execute_agent()` (around line 1400-1450)

### 4. Update Runtime Module

**File:** `file:codeframe/core/runtime.py`

Modify `execute_agent()` function to accept engine parameter:

- Update function signature (line 591-599):
  ```python
  def execute_agent(
      workspace: Workspace,
      run: Run,
      dry_run: bool = False,
      debug: bool = False,
      verbose: bool = False,
      fix_coordinator: Optional["GlobalFixCoordinator"] = None,
      event_publisher: Optional["EventPublisher"] = None,
      engine: str = "plan",  # NEW parameter
  ) -> "AgentState":
  ```

- Add engine selection logic before creating Agent instance (around line 660-671):
  ```python
  # Select execution engine
  if engine == "react":
      # Future: Use ReAct engine when available
      # For now, raise NotImplementedError to fail gracefully
      raise NotImplementedError(
          "ReAct engine not yet implemented. Use engine='plan' (default) for now."
      )
  elif engine != "plan":
      raise ValueError(f"Unknown engine: {engine}. Valid options: 'plan', 'react'")
  
  # Use plan engine (current implementation)
  agent = Agent(...)
  ```

### 5. Update Database Schema

**File:** `file:codeframe/core/workspace.py` or migration script

Add `engine` column to `runs` table and `batches` table:

- For `runs` table:
  ```sql
  ALTER TABLE runs ADD COLUMN engine TEXT DEFAULT 'plan';
  ```

- For `batches` table (if it exists):
  ```sql
  ALTER TABLE batches ADD COLUMN engine TEXT DEFAULT 'plan';
  ```

- Update `_save_batch()` in conductor.py to persist engine field
- Update `_row_to_batch()` to hydrate engine field from database

### 6. Verify SSE Streaming Compatibility

**File:** `file:codeframe/ui/routers/streaming_v2.py`

No changes required—verify that existing event streaming works:

- The `event_stream_generator()` function (line 84-149) already handles generic `ExecutionEvent` types
- The `format_sse_event()` function (line 54-66) serializes events using `model_dump_json()`
- Phase-based events (EXPLORING, CREATING, etc.) will be emitted as `ProgressEvent` with appropriate `phase` values
- The ReAct engine (when implemented) should emit events using the same `ExecutionEvent` types

### 7. Add Integration Test

**File:** `file:tests/api/test_tasks_v2_engine.py` (new file)

Create integration test to verify engine parameter handling:

```python
"""Integration tests for engine parameter in v2 API."""

import pytest
from fastapi.testclient import TestClient

def test_execute_with_plan_engine(client, workspace, task_id):
    """Test execution with default 'plan' engine."""
    response = client.post(
        "/api/v2/tasks/execute",
        json={
            "task_ids": [task_id],
            "strategy": "serial",
            "engine": "plan"
        },
        headers={"workspace-path": workspace.path}
    )
    assert response.status_code == 200
    assert response.json()["success"] is True

def test_execute_with_react_engine_not_implemented(client, workspace, task_id):
    """Test execution with 'react' engine raises NotImplementedError."""
    response = client.post(
        "/api/v2/tasks/execute",
        json={
            "task_ids": [task_id],
            "strategy": "serial",
            "engine": "react"
        },
        headers={"workspace-path": workspace.path}
    )
    # Should fail gracefully with 500 and NotImplementedError message
    assert response.status_code == 500
    assert "not yet implemented" in response.json()["detail"].lower()

def test_execute_with_invalid_engine(client, workspace, task_id):
    """Test execution with invalid engine value."""
    response = client.post(
        "/api/v2/tasks/execute",
        json={
            "task_ids": [task_id],
            "strategy": "serial",
            "engine": "invalid"
        },
        headers={"workspace-path": workspace.path}
    )
    assert response.status_code == 400
    assert "invalid engine" in response.json()["detail"].lower()

def test_sse_stream_with_engine_parameter(client, workspace, task_id):
    """Test SSE streaming works with engine parameter."""
    # Start execution with engine parameter
    client.post(
        f"/api/v2/tasks/{task_id}/start",
        params={"execute": True, "engine": "plan"},
        headers={"workspace-path": workspace.path}
    )
    
    # Connect to SSE stream
    with client.stream("GET", f"/api/v2/tasks/{task_id}/stream") as response:
        events = []
        for line in response.iter_lines():
            if line.startswith("data:"):
                events.append(json.loads(line[5:]))
        
        # Verify we received progress events
        progress_events = [e for e in events if e["event_type"] == "progress"]
        assert len(progress_events) > 0
```

### 8. Update Existing Tests

Ensure all existing API and CLI tests pass:

- **File:** `file:tests/api/test_multi_agent_api.py`
  - Verify tests don't break with new optional parameter
  - Add assertions that default engine is "plan"

- **File:** `file:tests/cli/test_v2_cli_integration.py`
  - Verify CLI tests still pass (CLI may not expose engine parameter yet per dependency #353)

- **File:** `file:tests/integration/test_worker_agent_execution.py`
  - Update to explicitly pass `engine="plan"` if needed

### 9. Documentation Updates

**File:** `file:docs/PHASE_2_DEVELOPER_GUIDE.md` or API documentation

Document the new engine parameter:

```markdown
## Execution Engine Selection

The v2 API supports multiple execution engines:

- `plan` (default): Traditional planner-based execution (Agent + Planner)
- `react` (experimental): ReAct-style reasoning and acting engine

### API Usage

```bash
# Use default plan engine
POST /api/v2/tasks/execute
{
  "task_ids": ["task-123"],
  "strategy": "serial",
  "engine": "plan"  # optional, defaults to "plan"
}

# Use ReAct engine (when available)
POST /api/v2/tasks/execute
{
  "task_ids": ["task-123"],
  "strategy": "serial",
  "engine": "react"
}
```

### SSE Streaming

Both engines emit the same event types:
- `progress`: Phase transitions (planning, execution, verification)
- `output`: Command output
- `blocker`: Human input needed
- `completion`: Task finished
- `error`: Execution error
```

## Sequence Diagram

```mermaid
sequenceDiagram
    participant Client
    participant API as tasks_v2.py
    participant Conductor
    participant Runtime
    participant Agent as Agent/ReAct Engine
    participant SSE as streaming_v2.py
    
    Client->>API: POST /api/v2/tasks/execute<br/>{engine: "react"}
    API->>API: Validate engine parameter
    API->>Conductor: start_batch(engine="react")
    Conductor->>Runtime: execute_agent(engine="react")
    
    alt engine == "react"
        Runtime->>Agent: Create ReAct engine instance
    else engine == "plan"
        Runtime->>Agent: Create Agent instance (default)
    end
    
    Agent->>SSE: Emit ProgressEvent(phase="EXPLORING")
    SSE->>Client: SSE: data: {event_type: "progress", phase: "EXPLORING"}
    
    Agent->>SSE: Emit ProgressEvent(phase="CREATING")
    SSE->>Client: SSE: data: {event_type: "progress", phase: "CREATING"}
    
    Agent->>SSE: Emit CompletionEvent(status="completed")
    SSE->>Client: SSE: data: {event_type: "completion"}
    
    Runtime-->>Conductor: AgentState
    Conductor-->>API: BatchRun
    API-->>Client: {success: true, batch_id: "..."}
```

## Validation Checklist

- [ ] `StartExecutionRequest` model has `engine` field with validation
- [ ] `POST /api/v2/tasks/execute` accepts and validates `engine` parameter
- [ ] `POST /api/v2/tasks/{id}/start` accepts `engine` query parameter
- [ ] `conductor.start_batch()` accepts and stores `engine` parameter
- [ ] `runtime.execute_agent()` accepts `engine` parameter and validates it
- [ ] Database schema includes `engine` column in `runs` and `batches` tables
- [ ] SSE streaming works with both engines (verified via existing event types)
- [ ] Integration test covers engine parameter validation and execution
- [ ] All existing API tests pass without modification
- [ ] All existing CLI tests pass (CLI may not expose engine yet)
- [ ] Documentation updated with engine parameter usage

---

Execution Information

Branch: main
Commit: 99f0ea9

---

💡 Tips

Supported Commands (Inside Comments)

• Use @traycerai generate to iterate on the previous version of the implementation plan.

Supported Commands (Inside Description)

• Add @traycerai ignore anywhere in the ticket description to prevent this ticket from being processed.
• Add @traycerai branch:<branch-name> anywhere in the ticket description to specify the target branch for the implementation plan.

Community

• Join our Discord Community to get help, request features, and share feedback.
• Follow us on X/Twitter for updates and announcements.