Backend: Streaming chat adapter for Anthropic SDK (token-by-token relay)

[frankbria]

Parent Issue

Part of #500 — depends on #501, used by #502

Context

The existing chat.py endpoint calls agent.chat(user_message) synchronously and returns a complete string response. For live interactive sessions, the frontend needs token-by-token streaming so the UI updates as the model generates. This issue builds a streaming adapter that wraps the Anthropic SDK streaming API and emits typed events for the WebSocket relay.

Existing Code to Build On

• codeframe/core/adapters/claude_code.py — existing Claude Code adapter (subprocess-based)
• codeframe/adapters/llm/ — existing LLM adapter layer
• codeframe/core/streaming.py — EventPublisher async event distribution pattern

What to Build

New file: codeframe/core/adapters/streaming_chat.py

from dataclasses import dataclass
from typing import AsyncIterator
from enum import Enum

class ChatEventType(str, Enum):
    TEXT_DELTA   = "text_delta"
    TOOL_USE_START = "tool_use_start"
    TOOL_RESULT  = "tool_result"
    THINKING     = "thinking"
    COST_UPDATE  = "cost_update"
    DONE         = "done"
    ERROR        = "error"

@dataclass
class ChatEvent:
    type: ChatEventType
    content: str | None = None
    tool_name: str | None = None
    tool_input: dict | None = None
    cost_usd: float | None = None
    input_tokens: int | None = None
    output_tokens: int | None = None

class StreamingChatAdapter:
    """
    Wraps the Anthropic SDK streaming API.
    Yields ChatEvent objects as the model generates.
    Supports interrupt via asyncio.Event.
    Persists messages to session_messages table after each turn.
    """

    def __init__(self, session_id: str, model: str, db_repo):
        ...

    async def send_message(
        self,
        content: str,
        history: list[dict],
        interrupt_event: asyncio.Event | None = None,
    ) -> AsyncIterator[ChatEvent]:
        """
        Stream a single turn.
        Uses anthropic.AsyncAnthropic().messages.stream().
        Yields ChatEvent per SDK event type:
          - RawContentBlockDeltaEvent → TEXT_DELTA or THINKING
          - RawContentBlockStartEvent (tool_use) → TOOL_USE_START
          - tool execution result → TOOL_RESULT
          - MessageStopEvent → COST_UPDATE + DONE
        Checks interrupt_event between chunks; if set, stops and closes stream.
        Persists user message and complete assistant turn to session_messages.
        """
        ...

Tool execution

For interactive sessions, run a limited safe tool set (read files, search, list directory). Do not execute shell commands or write files without explicit user confirmation — keep the scope conservative for v1.

Tools to support initially:

• Read (read a file)
• Glob (find files by pattern)
• Grep (search file contents)

Each tool call yields TOOL_USE_START → executes → yields TOOL_RESULT.

Message history management

• Load session_messages from DB at session start to reconstruct context
• Append new user message before the API call
• Append complete assistant response after the stream ends
• Truncate history if context window approaches limit (use tiktoken already in deps)

Acceptance Criteria

• StreamingChatAdapter.send_message() yields TEXT_DELTA events as tokens arrive (not buffered)
• Tool calls yield TOOL_USE_START with tool name + input, then TOOL_RESULT with output
• THINKING events emitted when extended thinking is enabled
• COST_UPDATE event emitted at turn end with correct token counts
• interrupt_event.set() causes the stream to stop within 1 turn
• Messages persisted to session_messages after each complete turn
• History reconstructed correctly from DB on adapter init
• Unit tests in tests/core/test_streaming_chat.py using mocked Anthropic client

Notes

• Use anthropic.AsyncAnthropic() — do not use the synchronous client
• Model defaults to claude-sonnet-4-6 if not specified in session
• Check ANTHROPIC_API_KEY env var; raise clear error if missing

[traycerai]

Warning

Ticket Assist will be discontinued on May 1, 2026. After that date, the feature will no longer be available. We suggest installing the Traycer IDE Extension for continued AI-powered, spec-driven development. Thank you for choosing Traycer!

Plan

Observations

• Conversation history is stored in the memory table with category='conversation', accessed via db.get_conversation(project_id) and db.create_memory(...) — there is no dedicated session_messages table; the existing memory-based pattern is what db_repo maps to.
• codeframe/adapters/llm/anthropic.py uses the synchronous Anthropic client; the new adapter must use anthropic.AsyncAnthropic.
• codeframe/adapters/llm/base.py defines Tool, ToolCall, ToolResult — the new adapter's tool definitions should mirror this schema.
• codeframe/core/tools.py already implements read_file, list_files (Glob equivalent), and search_codebase (Grep equivalent) as sync functions — these can be reused as the safe tool set.
• tiktoken and anthropic are both confirmed in pyproject.toml dependencies.

Approach

Create codeframe/core/adapters/streaming_chat.py as a self-contained async adapter that wraps anthropic.AsyncAnthropic().messages.stream(). It reuses the existing DB memory pattern for persistence and the existing tool executor functions from codeframe/core/tools.py for the safe tool set. No new DB tables or repositories are needed.

---

Implementation Steps

1. Define ChatEventType and ChatEvent in codeframe/core/adapters/streaming_chat.py

Create the new file. Define ChatEventType as a str + Enum with values: TEXT_DELTA, TOOL_USE_START, TOOL_RESULT, THINKING, COST_UPDATE, DONE, ERROR. Define ChatEvent as a dataclass with fields: type: ChatEventType, content: str | None, tool_name: str | None, tool_input: dict | None, cost_usd: float | None, input_tokens: int | None, output_tokens: int | None.

---

2. Define the safe tool set for StreamingChatAdapter

Inside the same file, define three Tool objects (using Tool from codeframe/adapters/llm/base.py) mirroring the schemas already in codeframe/core/tools.py:

Adapter Tool Name	Maps to handler in codeframe/core/tools.py
read_file	_execute_read_file
list_files	_execute_list_files
search_codebase	_execute_search_codebase

Store these as a module-level constant STREAMING_SAFE_TOOLS: list[Tool].

---

3. Implement StreamingChatAdapter.__init__

Constructor signature: __init__(self, session_id: str, model: str, db_repo, workspace_path: Path).

• Validate ANTHROPIC_API_KEY env var is set; raise a clear ValueError if missing (follow the pattern in AnthropicProvider.__init__).
• Store session_id, model (default to "claude-sonnet-4-20250514" matching the existing DEFAULT_PLANNING_MODEL), db_repo, and workspace_path.
• Lazily instantiate anthropic.AsyncAnthropic(api_key=...) — store as self._async_client.

---

4. Implement _load_history helper

A method that calls db_repo.get_conversation(session_id) and converts the memory rows into the Anthropic messages list format ([{"role": ..., "content": ...}]), following the same key-parsing logic as LeadAgent.get_conversation_history() in codeframe/agents/lead_agent.py (lines 178–201).

---

5. Implement _truncate_history helper

Use tiktoken (already a dependency) to count tokens in the history list. If the total token count approaches a configurable limit (e.g., 180,000 tokens for claude-sonnet-4), drop the oldest non-system messages from the front of the list until within budget. This mirrors the truncation concern described in the ticket.

---

6. Implement _persist_turn helper

An async method that saves a user message and a complete assistant response to the DB using db_repo.create_memory(...) with category="conversation", following the exact key pattern user_{index} / assistant_{index+1} used in LeadAgent.chat() (lines 321–342 of lead_agent.py).

---

7. Implement _execute_tool helper

An async method that dispatches a tool call to the appropriate sync handler from codeframe/core/tools.py using asyncio.to_thread(execute_tool, tool_call, self._workspace_path) (reusing the existing execute_tool dispatcher). Returns the string result.

---

8. Implement send_message as an AsyncGenerator

async def send_message(
    self,
    content: str,
    history: list[dict],
    interrupt_event: asyncio.Event | None = None,
) -> AsyncIterator[ChatEvent]:

The flow:

sequenceDiagram
    participant Caller
    participant StreamingChatAdapter
    participant AnthropicSDK
    participant ToolExecutor

    Caller->>StreamingChatAdapter: send_message(content, history)
    StreamingChatAdapter->>StreamingChatAdapter: _truncate_history(history + user_msg)
    StreamingChatAdapter->>AnthropicSDK: AsyncAnthropic.messages.stream(...)
    loop SDK events
        AnthropicSDK-->>StreamingChatAdapter: RawContentBlockDeltaEvent (text)
        StreamingChatAdapter-->>Caller: yield ChatEvent(TEXT_DELTA)
        AnthropicSDK-->>StreamingChatAdapter: RawContentBlockDeltaEvent (thinking)
        StreamingChatAdapter-->>Caller: yield ChatEvent(THINKING)
        AnthropicSDK-->>StreamingChatAdapter: RawContentBlockStartEvent (tool_use)
        StreamingChatAdapter-->>Caller: yield ChatEvent(TOOL_USE_START)
        StreamingChatAdapter->>ToolExecutor: _execute_tool(tool_call)
        ToolExecutor-->>StreamingChatAdapter: result
        StreamingChatAdapter-->>Caller: yield ChatEvent(TOOL_RESULT)
        Note over StreamingChatAdapter: Check interrupt_event.is_set()
        StreamingChatAdapter->>AnthropicSDK: stream.close() if interrupted
    end
    AnthropicSDK-->>StreamingChatAdapter: MessageStopEvent → usage stats
    StreamingChatAdapter-->>Caller: yield ChatEvent(COST_UPDATE, input_tokens, output_tokens)
    StreamingChatAdapter-->>Caller: yield ChatEvent(DONE)
    StreamingChatAdapter->>StreamingChatAdapter: _persist_turn(user_msg, full_response)

Loading

Key implementation details:

• Open the stream with async with self._async_client.messages.stream(model=..., messages=..., tools=STREAMING_SAFE_TOOLS, ...).
• Iterate async for event in stream and dispatch on event.type:
  • "content_block_delta" with delta.type == "text_delta" → yield TEXT_DELTA
  • "content_block_delta" with delta.type == "thinking_delta" → yield THINKING
  • "content_block_start" with content_block.type == "tool_use" → buffer tool name + id; yield TOOL_USE_START
  • "content_block_stop" when a tool_use block just closed → call _execute_tool, yield TOOL_RESULT, then feed the result back into the next API turn
  • "message_stop" → read stream.get_final_message().usage for token counts; yield COST_UPDATE then DONE
• After each chunk, check interrupt_event.is_set() and break + close the stream if set.
• Wrap the entire body in try/except and yield ChatEvent(ERROR, content=str(exc)) on failure.
• Accumulate the full assistant text across all TEXT_DELTA events; call _persist_turn after DONE is yielded.

---

9. Handle multi-turn tool loops

When a tool result is produced, the adapter must re-enter the stream with the tool result appended to the message list (Anthropic requires tool results to be sent back as a user message with tool_result content blocks). This means the send_message method internally loops: stream → tool call → stream again, until stop_reason == "end_turn". Use the same _convert_messages pattern from AnthropicProvider._convert_messages() in codeframe/adapters/llm/anthropic.py (lines 161–212) to format tool result messages correctly.

---

10. Export from codeframe/core/adapters/__init__.py

Add StreamingChatAdapter, ChatEvent, and ChatEventType to the exports in file:codeframe/core/adapters/__init__.py.

Import In IDE

🤖 Prompt for AI Agents

## Observations

- Conversation history is stored in the `memory` table with `category='conversation'`, accessed via `db.get_conversation(project_id)` and `db.create_memory(...)` — there is **no** dedicated `session_messages` table; the existing memory-based pattern is what `db_repo` maps to.
- `codeframe/adapters/llm/anthropic.py` uses the **synchronous** `Anthropic` client; the new adapter must use `anthropic.AsyncAnthropic`.
- `codeframe/adapters/llm/base.py` defines `Tool`, `ToolCall`, `ToolResult` — the new adapter's tool definitions should mirror this schema.
- `codeframe/core/tools.py` already implements `read_file`, `list_files` (Glob equivalent), and `search_codebase` (Grep equivalent) as sync functions — these can be reused as the safe tool set.
- `tiktoken` and `anthropic` are both confirmed in `pyproject.toml` dependencies.

## Approach

Create `codeframe/core/adapters/streaming_chat.py` as a self-contained async adapter that wraps `anthropic.AsyncAnthropic().messages.stream()`. It reuses the existing DB memory pattern for persistence and the existing tool executor functions from `codeframe/core/tools.py` for the safe tool set. No new DB tables or repositories are needed.

---

## Implementation Steps

### 1. Define `ChatEventType` and `ChatEvent` in `codeframe/core/adapters/streaming_chat.py`

Create the new file. Define `ChatEventType` as a `str` + `Enum` with values: `TEXT_DELTA`, `TOOL_USE_START`, `TOOL_RESULT`, `THINKING`, `COST_UPDATE`, `DONE`, `ERROR`. Define `ChatEvent` as a `dataclass` with fields: `type: ChatEventType`, `content: str | None`, `tool_name: str | None`, `tool_input: dict | None`, `cost_usd: float | None`, `input_tokens: int | None`, `output_tokens: int | None`.

---

### 2. Define the safe tool set for `StreamingChatAdapter`

Inside the same file, define three `Tool` objects (using `Tool` from `codeframe/adapters/llm/base.py`) mirroring the schemas already in `codeframe/core/tools.py`:

| Adapter Tool Name | Maps to handler in `codeframe/core/tools.py` |
|---|---|
| `read_file` | `_execute_read_file` |
| `list_files` | `_execute_list_files` |
| `search_codebase` | `_execute_search_codebase` |

Store these as a module-level constant `STREAMING_SAFE_TOOLS: list[Tool]`.

---

### 3. Implement `StreamingChatAdapter.__init__`

Constructor signature: `__init__(self, session_id: str, model: str, db_repo, workspace_path: Path)`.

- Validate `ANTHROPIC_API_KEY` env var is set; raise a clear `ValueError` if missing (follow the pattern in `AnthropicProvider.__init__`).
- Store `session_id`, `model` (default to `"claude-sonnet-4-20250514"` matching the existing `DEFAULT_PLANNING_MODEL`), `db_repo`, and `workspace_path`.
- Lazily instantiate `anthropic.AsyncAnthropic(api_key=...)` — store as `self._async_client`.

---

### 4. Implement `_load_history` helper

A method that calls `db_repo.get_conversation(session_id)` and converts the memory rows into the Anthropic messages list format (`[{"role": ..., "content": ...}]`), following the same key-parsing logic as `LeadAgent.get_conversation_history()` in `codeframe/agents/lead_agent.py` (lines 178–201).

---

### 5. Implement `_truncate_history` helper

Use `tiktoken` (already a dependency) to count tokens in the history list. If the total token count approaches a configurable limit (e.g., 180,000 tokens for claude-sonnet-4), drop the oldest non-system messages from the front of the list until within budget. This mirrors the truncation concern described in the ticket.

---

### 6. Implement `_persist_turn` helper

An `async` method that saves a user message and a complete assistant response to the DB using `db_repo.create_memory(...)` with `category="conversation"`, following the exact key pattern `user_{index}` / `assistant_{index+1}` used in `LeadAgent.chat()` (lines 321–342 of `lead_agent.py`).

---

### 7. Implement `_execute_tool` helper

An `async` method that dispatches a tool call to the appropriate sync handler from `codeframe/core/tools.py` using `asyncio.to_thread(execute_tool, tool_call, self._workspace_path)` (reusing the existing `execute_tool` dispatcher). Returns the string result.

---

### 8. Implement `send_message` as an `AsyncGenerator`

```
async def send_message(
    self,
    content: str,
    history: list[dict],
    interrupt_event: asyncio.Event | None = None,
) -> AsyncIterator[ChatEvent]:
```

The flow:

```mermaid
sequenceDiagram
    participant Caller
    participant StreamingChatAdapter
    participant AnthropicSDK
    participant ToolExecutor

    Caller->>StreamingChatAdapter: send_message(content, history)
    StreamingChatAdapter->>StreamingChatAdapter: _truncate_history(history + user_msg)
    StreamingChatAdapter->>AnthropicSDK: AsyncAnthropic.messages.stream(...)
    loop SDK events
        AnthropicSDK-->>StreamingChatAdapter: RawContentBlockDeltaEvent (text)
        StreamingChatAdapter-->>Caller: yield ChatEvent(TEXT_DELTA)
        AnthropicSDK-->>StreamingChatAdapter: RawContentBlockDeltaEvent (thinking)
        StreamingChatAdapter-->>Caller: yield ChatEvent(THINKING)
        AnthropicSDK-->>StreamingChatAdapter: RawContentBlockStartEvent (tool_use)
        StreamingChatAdapter-->>Caller: yield ChatEvent(TOOL_USE_START)
        StreamingChatAdapter->>ToolExecutor: _execute_tool(tool_call)
        ToolExecutor-->>StreamingChatAdapter: result
        StreamingChatAdapter-->>Caller: yield ChatEvent(TOOL_RESULT)
        Note over StreamingChatAdapter: Check interrupt_event.is_set()
        StreamingChatAdapter->>AnthropicSDK: stream.close() if interrupted
    end
    AnthropicSDK-->>StreamingChatAdapter: MessageStopEvent → usage stats
    StreamingChatAdapter-->>Caller: yield ChatEvent(COST_UPDATE, input_tokens, output_tokens)
    StreamingChatAdapter-->>Caller: yield ChatEvent(DONE)
    StreamingChatAdapter->>StreamingChatAdapter: _persist_turn(user_msg, full_response)
```

**Key implementation details:**

- Open the stream with `async with self._async_client.messages.stream(model=..., messages=..., tools=STREAMING_SAFE_TOOLS, ...)`.
- Iterate `async for event in stream` and dispatch on `event.type`:
  - `"content_block_delta"` with `delta.type == "text_delta"` → yield `TEXT_DELTA`
  - `"content_block_delta"` with `delta.type == "thinking_delta"` → yield `THINKING`
  - `"content_block_start"` with `content_block.type == "tool_use"` → buffer tool name + id; yield `TOOL_USE_START`
  - `"content_block_stop"` when a tool_use block just closed → call `_execute_tool`, yield `TOOL_RESULT`, then feed the result back into the next API turn
  - `"message_stop"` → read `stream.get_final_message().usage` for token counts; yield `COST_UPDATE` then `DONE`
- After each chunk, check `interrupt_event.is_set()` and `break` + close the stream if set.
- Wrap the entire body in `try/except` and yield `ChatEvent(ERROR, content=str(exc))` on failure.
- Accumulate the full assistant text across all `TEXT_DELTA` events; call `_persist_turn` after `DONE` is yielded.

---

### 9. Handle multi-turn tool loops

When a tool result is produced, the adapter must re-enter the stream with the tool result appended to the message list (Anthropic requires tool results to be sent back as a `user` message with `tool_result` content blocks). This means the `send_message` method internally loops: stream → tool call → stream again, until `stop_reason == "end_turn"`. Use the same `_convert_messages` pattern from `AnthropicProvider._convert_messages()` in `codeframe/adapters/llm/anthropic.py` (lines 161–212) to format tool result messages correctly.

---

### 10. Export from `codeframe/core/adapters/__init__.py`

Add `StreamingChatAdapter`, `ChatEvent`, and `ChatEventType` to the exports in `file:codeframe/core/adapters/__init__.py`.

---

Execution Information

Branch: main
Commit: cb074e3

---

💡 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.

[coderabbitai]

🔗 Related PRs

#257 - feat(discovery): implement AI-powered Socratic questioning system [merged]
#364 - feat: phase-based event emission for ReactAgent progress reporting [merged]
#370 - feat: token budget tracking and message compaction for ReactAgent (#352) [merged]
#429 - feat(core): Agent Adapter protocol definition [merged]
#437 - feat(adapters): Codex engine adapter with app-server JSON-RPC protocol [merged]

---

📝 Issue Planner

_{Check the box below or use the @coderabbitai plan command to generate an implementation plan and prompts that you can use with your favorite coding assistant.}

• Create Plan

---

🧪 Issue enrichment is currently in open beta.

You can configure auto-planning by selecting labels in the issue_enrichment configuration.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord!