[4.D.6] Abstract streaming layer (streaming_chat.py) to use LLMProvider

[frankbria]

Parent

Part of #542 — Multi-provider LLM support

Depends on

#543 (OpenAI adapter), #544 (factory wiring)

What

codeframe/core/adapters/streaming_chat.py is currently hardwired to anthropic.AsyncAnthropic().messages.stream(). This is used by the ClaudeCodeAdapter and interactive session flows. Abstract it behind LLMProvider.

Current problem

streaming_chat.py directly uses:

• anthropic.AsyncAnthropic().messages.stream()
• Anthropic-specific tool_call chunking format
• Anthropic-specific extended_thinking blocks

Target state

• StreamingChatAdapter takes a LLMProvider (or async_stream() callable)
• Provider-specific streaming details stay in each provider implementation
• Callers of StreamingChatAdapter are unaffected

Files to modify

• codeframe/core/adapters/streaming_chat.py
• codeframe/adapters/llm/base.py — ensure stream() / async_stream() is in the ABC
• codeframe/adapters/llm/anthropic.py — move streaming logic here from streaming_chat.py
• codeframe/adapters/llm/openai.py — implement streaming
• tests/core/adapters/test_streaming_chat.py

Notes

• Extended thinking (anthropic.ExtendedThinking) is Anthropic-only — wrap behind a provider capability check (provider.supports("extended_thinking")) rather than removing it. If provider doesn't support it, silently skip.
• Streaming chunk format differs between providers — normalize to a common StreamChunk datatype in base.py
• If streaming is complex to generalize, it's acceptable to make it a # provider-specific path with clear extension points rather than forcing premature abstraction

Acceptance criteria

• streaming_chat.py has no import anthropic
• Streaming works with Anthropic provider (regression test)
• Streaming works with OpenAI provider
• Extended thinking gracefully no-ops on non-Anthropic providers
• Existing streaming tests pass

[coderabbitai]

🔗 Related PRs

#512 - feat(core): StreamingChatAdapter for token-by-token Anthropic relay (#503) [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!

[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

• streaming_chat.py directly instantiates anthropic.AsyncAnthropic via a lazy _client property and processes Anthropic-specific event types (content_block_start, content_block_delta, thinking_delta, input_json_delta, message_stop).
• LLMProvider in base.py has a synchronous stream() with a fallback implementation, but no async_stream() method exists yet.
• AnthropicProvider in anthropic.py has a synchronous stream() using self.client.messages.stream() (sync client), not the async client used in streaming_chat.py.
• MockProvider mirrors the sync stream() pattern.
• The existing tests in tests/core/test_streaming_chat.py patch adapter._async_client directly — they will need updating to patch through the provider instead.
• session_chat_ws.py instantiates StreamingChatAdapter with session_id, db_repo, and workspace_path — the constructor signature change must remain backward-compatible or the router must be updated.

Approach

Introduce a StreamChunk dataclass and an async_stream() abstract method on LLMProvider. Move all Anthropic-specific async streaming logic into AnthropicProvider. Implement OpenAI async streaming in openai.py. Refactor StreamingChatAdapter to accept an LLMProvider and delegate the raw stream to it, keeping all ChatEvent orchestration (tool loops, history, persistence) in the adapter. The session_chat_ws.py router constructs the provider and passes it in.

---

Implementation Steps

1. Add StreamChunk and async_stream() to file:codeframe/adapters/llm/base.py

• Define a StreamChunk dataclass with fields:
  • type: str — one of "text_delta", "thinking_delta", "tool_use_start", "tool_use_stop", "message_stop"
  • text: Optional[str] — populated for text_delta and thinking_delta
  • tool_id: Optional[str], tool_name: Optional[str], tool_input: Optional[dict] — populated for tool_use_start
  • input_tokens: Optional[int], output_tokens: Optional[int], stop_reason: Optional[str] — populated for message_stop
• Add an async_stream() abstract method to LLMProvider with signature:

  async_stream(messages, system, tools, model, max_tokens, interrupt_event) -> AsyncIterator[StreamChunk]
  

  • tools accepts list[dict] (already-serialized format, matching _TOOLS_FOR_API in streaming_chat.py)
  • model: str is passed explicitly (the adapter owns model selection for streaming)
  • Default implementation raises NotImplementedError (unlike stream() which has a fallback)
• Add a supports(capability: str) -> bool method to LLMProvider with a default returning False. This is the capability check gate for extended_thinking.

classDiagram
    class LLMProvider {
        +complete(...)
        +stream(...)
        +async_stream(...) AsyncIterator~StreamChunk~
        +supports(capability) bool
        +get_model(purpose) str
    }
    class StreamChunk {
        +type: str
        +text: Optional[str]
        +tool_id: Optional[str]
        +tool_name: Optional[str]
        +tool_input: Optional[dict]
        +input_tokens: Optional[int]
        +output_tokens: Optional[int]
        +stop_reason: Optional[str]
    }
    LLMProvider --> StreamChunk : yields

Loading

---

2. Implement async_stream() in file:codeframe/adapters/llm/anthropic.py

• Add an _async_client property (lazy, mirrors the existing _client pattern but uses AsyncAnthropic).
• Implement async_stream():
  • Open async with self._async_client.messages.stream(...) using the passed model, system, messages, tools, max_tokens.
  • If self.supports("extended_thinking") is True, include the thinking betas parameter (wrap in a try/except so it degrades gracefully if the SDK version doesn't support it).
  • Translate each Anthropic SDK event into a StreamChunk:
    • content_block_start with block.type == "tool_use" → StreamChunk(type="tool_use_start", tool_id=..., tool_name=..., tool_input=...)
    • content_block_delta with delta.type == "text_delta" → StreamChunk(type="text_delta", text=delta.text)
    • content_block_delta with delta.type == "thinking_delta" → StreamChunk(type="thinking_delta", text=delta.thinking)
    • content_block_delta with delta.type == "input_json_delta" → StreamChunk(type="tool_input_delta") (pass-through; tool input is rebuilt from final message)
    • content_block_stop → StreamChunk(type="tool_use_stop")
    • message_stop → call stream.get_final_message(), emit StreamChunk(type="message_stop", stop_reason=..., input_tokens=..., output_tokens=..., tool_inputs_by_id=...) — include the final content blocks so the adapter can rebuild tool inputs.
  • Override supports() to return True for "extended_thinking".

---

3. Implement async_stream() in file:codeframe/adapters/llm/openai.py

• Create codeframe/adapters/llm/openai.py if it does not exist (check file:codeframe/adapters/llm/__init__.py — it is not currently exported, so create it).
• Implement OpenAIProvider(LLMProvider) with:
  • Lazy _async_client using openai.AsyncOpenAI (verify openai is in pyproject.toml before referencing).
  • complete() using the sync client.
  • async_stream() using await self._async_client.chat.completions.create(..., stream=True):
    • Translate OpenAI SSE chunks to StreamChunk:
      • delta.content → StreamChunk(type="text_delta", text=...)
      • delta.tool_calls → StreamChunk(type="tool_use_start", ...) on first appearance of a tool call index, StreamChunk(type="tool_input_delta", ...) for subsequent deltas
      • finish_reason present → StreamChunk(type="message_stop", stop_reason=..., input_tokens=..., output_tokens=...) using chunk.usage if available.
  • supports() returns False (no extended thinking).

---

4. Refactor file:codeframe/core/adapters/streaming_chat.py

This is the core change. The goal: no import anthropic, all provider-specific logic delegated.

• Constructor: Replace api_key / model / lazy _async_client with a provider: LLMProvider parameter. Keep session_id, db_repo, workspace_path. Remove _api_key, _async_client, _client property, and _DEFAULT_MODEL.

  • For backward compatibility with session_chat_ws.py (which passes no provider today), accept provider: Optional[LLMProvider] = None and default-construct an AnthropicProvider() if None — but prefer updating the router (step 5) to pass the provider explicitly.

• _stream_turn(): Replace the async with self._client.messages.stream(...) block with:

  async for chunk in self._provider.async_stream(
      messages=current_messages,
      system=<system_prompt>,
      tools=_TOOLS_FOR_API,
      model=self._model,   # or derive from provider
      max_tokens=4096,
      interrupt_event=interrupt_event,
  ):
  

  • Map StreamChunk types to ChatEvent types:

  StreamChunk.type	ChatEvent emitted
  text_delta	ChatEventType.TEXT_DELTA
  thinking_delta	ChatEventType.THINKING
  tool_use_start	ChatEventType.TOOL_USE_START
  tool_use_stop	append to pending_tool_calls
  message_stop	ChatEventType.COST_UPDATE + set stop_reason

• Extended thinking: Before calling async_stream(), check self._provider.supports("extended_thinking"). If True, pass an extended_thinking=True kwarg (or a flag the provider understands). The provider handles the Anthropic-specific betas parameter internally.

• _rebuild_tool_inputs(): Move this helper into AnthropicProvider or keep it in streaming_chat.py but feed it from the message_stop StreamChunk (which carries the final content blocks).

• _estimate_cost(): Keep in streaming_chat.py for now (it's already provider-agnostic by model prefix). Alternatively, move cost estimation into each provider's message_stop chunk — either approach is acceptable.

• Remove the _DEFAULT_MODEL constant and the model constructor parameter; the provider's model_selector owns model selection. If the caller needs to override the model, they configure the ModelSelector on the provider.

---

5. Update file:codeframe/ui/routers/session_chat_ws.py

• In _run_streaming_adapter(), construct an AnthropicProvider() (or read provider type from app config/env) and pass it to StreamingChatAdapter:

  from codeframe.adapters.llm.anthropic import AnthropicProvider
  provider = AnthropicProvider()
  adapter = StreamingChatAdapter(
      session_id=session_id,
      db_repo=db_repo,
      workspace_path=workspace_path,
      provider=provider,
  )

• This keeps the router as the composition root and avoids the adapter needing to know which provider to instantiate.

---

6. Update file:codeframe/adapters/llm/__init__.py

• Export StreamChunk from base.py.
• Export OpenAIProvider once openai.py is created.
• Update get_provider() to handle "openai" as a valid provider_type.

---

7. Update file:tests/core/test_streaming_chat.py

• Replace all patch.object(adapter, "_async_client") patches with a mock LLMProvider whose async_stream() is an AsyncMock yielding StreamChunk objects.
• Update TestStreamingChatAdapterInit:
  • test_raises_if_no_api_key → this check now lives in AnthropicProvider.__init__, not StreamingChatAdapter. Either test it there, or test that passing a provider with no key raises at provider construction time.
  • test_accepts_api_key_from_env and test_default_model → adapt to the new constructor (pass a MockProvider or AnthropicProvider with a fake key).
• Add a MockProvider-based async_stream() to file:codeframe/adapters/llm/mock.py that yields configurable StreamChunk sequences, so tests can drive the full _stream_turn() loop without touching Anthropic internals.

sequenceDiagram
    participant WS as session_chat_ws
    participant SCA as StreamingChatAdapter
    participant AP as AnthropicProvider
    participant SDK as anthropic.AsyncAnthropic

    WS->>SCA: send_message(content, history, interrupt)
    SCA->>AP: async_stream(messages, system, tools, model, max_tokens)
    AP->>SDK: messages.stream(...)
    SDK-->>AP: SDK events (content_block_delta, etc.)
    AP-->>SCA: StreamChunk(type="text_delta", text="...")
    SCA-->>WS: ChatEvent(TEXT_DELTA)
    AP-->>SCA: StreamChunk(type="message_stop", stop_reason="end_turn", ...)
    SCA-->>WS: ChatEvent(COST_UPDATE)
    SCA-->>WS: ChatEvent(DONE)

Loading

Import In IDE

🤖 Prompt for AI Agents

## Observations

- `streaming_chat.py` directly instantiates `anthropic.AsyncAnthropic` via a lazy `_client` property and processes Anthropic-specific event types (`content_block_start`, `content_block_delta`, `thinking_delta`, `input_json_delta`, `message_stop`).
- `LLMProvider` in `base.py` has a synchronous `stream()` with a fallback implementation, but **no `async_stream()` method** exists yet.
- `AnthropicProvider` in `anthropic.py` has a synchronous `stream()` using `self.client.messages.stream()` (sync client), not the async client used in `streaming_chat.py`.
- `MockProvider` mirrors the sync `stream()` pattern.
- The existing tests in `tests/core/test_streaming_chat.py` patch `adapter._async_client` directly — they will need updating to patch through the provider instead.
- `session_chat_ws.py` instantiates `StreamingChatAdapter` with `session_id`, `db_repo`, and `workspace_path` — the constructor signature change must remain backward-compatible or the router must be updated.

## Approach

Introduce a `StreamChunk` dataclass and an `async_stream()` abstract method on `LLMProvider`. Move all Anthropic-specific async streaming logic into `AnthropicProvider`. Implement OpenAI async streaming in `openai.py`. Refactor `StreamingChatAdapter` to accept an `LLMProvider` and delegate the raw stream to it, keeping all `ChatEvent` orchestration (tool loops, history, persistence) in the adapter. The `session_chat_ws.py` router constructs the provider and passes it in.

---

## Implementation Steps

### 1. Add `StreamChunk` and `async_stream()` to `file:codeframe/adapters/llm/base.py`

- Define a `StreamChunk` dataclass with fields:
  - `type: str` — one of `"text_delta"`, `"thinking_delta"`, `"tool_use_start"`, `"tool_use_stop"`, `"message_stop"`
  - `text: Optional[str]` — populated for `text_delta` and `thinking_delta`
  - `tool_id: Optional[str]`, `tool_name: Optional[str]`, `tool_input: Optional[dict]` — populated for `tool_use_start`
  - `input_tokens: Optional[int]`, `output_tokens: Optional[int]`, `stop_reason: Optional[str]` — populated for `message_stop`
- Add an `async_stream()` abstract method to `LLMProvider` with signature:
  ```
  async_stream(messages, system, tools, model, max_tokens, interrupt_event) -> AsyncIterator[StreamChunk]
  ```
  - `tools` accepts `list[dict]` (already-serialized format, matching `_TOOLS_FOR_API` in `streaming_chat.py`)
  - `model: str` is passed explicitly (the adapter owns model selection for streaming)
  - Default implementation raises `NotImplementedError` (unlike `stream()` which has a fallback)
- Add a `supports(capability: str) -> bool` method to `LLMProvider` with a default returning `False`. This is the capability check gate for `extended_thinking`.

```mermaid
classDiagram
    class LLMProvider {
        +complete(...)
        +stream(...)
        +async_stream(...) AsyncIterator~StreamChunk~
        +supports(capability) bool
        +get_model(purpose) str
    }
    class StreamChunk {
        +type: str
        +text: Optional[str]
        +tool_id: Optional[str]
        +tool_name: Optional[str]
        +tool_input: Optional[dict]
        +input_tokens: Optional[int]
        +output_tokens: Optional[int]
        +stop_reason: Optional[str]
    }
    LLMProvider --> StreamChunk : yields
```

---

### 2. Implement `async_stream()` in `file:codeframe/adapters/llm/anthropic.py`

- Add an `_async_client` property (lazy, mirrors the existing `_client` pattern but uses `AsyncAnthropic`).
- Implement `async_stream()`:
  - Open `async with self._async_client.messages.stream(...)` using the passed `model`, `system`, `messages`, `tools`, `max_tokens`.
  - If `self.supports("extended_thinking")` is `True`, include the `thinking` betas parameter (wrap in a try/except so it degrades gracefully if the SDK version doesn't support it).
  - Translate each Anthropic SDK event into a `StreamChunk`:
    - `content_block_start` with `block.type == "tool_use"` → `StreamChunk(type="tool_use_start", tool_id=..., tool_name=..., tool_input=...)`
    - `content_block_delta` with `delta.type == "text_delta"` → `StreamChunk(type="text_delta", text=delta.text)`
    - `content_block_delta` with `delta.type == "thinking_delta"` → `StreamChunk(type="thinking_delta", text=delta.thinking)`
    - `content_block_delta` with `delta.type == "input_json_delta"` → `StreamChunk(type="tool_input_delta")` (pass-through; tool input is rebuilt from final message)
    - `content_block_stop` → `StreamChunk(type="tool_use_stop")`
    - `message_stop` → call `stream.get_final_message()`, emit `StreamChunk(type="message_stop", stop_reason=..., input_tokens=..., output_tokens=..., tool_inputs_by_id=...)` — include the final content blocks so the adapter can rebuild tool inputs.
  - Override `supports()` to return `True` for `"extended_thinking"`.

---

### 3. Implement `async_stream()` in `file:codeframe/adapters/llm/openai.py`

- Create `codeframe/adapters/llm/openai.py` if it does not exist (check `file:codeframe/adapters/llm/__init__.py` — it is not currently exported, so create it).
- Implement `OpenAIProvider(LLMProvider)` with:
  - Lazy `_async_client` using `openai.AsyncOpenAI` (verify `openai` is in `pyproject.toml` before referencing).
  - `complete()` using the sync client.
  - `async_stream()` using `await self._async_client.chat.completions.create(..., stream=True)`:
    - Translate OpenAI SSE chunks to `StreamChunk`:
      - `delta.content` → `StreamChunk(type="text_delta", text=...)`
      - `delta.tool_calls` → `StreamChunk(type="tool_use_start", ...)` on first appearance of a tool call index, `StreamChunk(type="tool_input_delta", ...)` for subsequent deltas
      - `finish_reason` present → `StreamChunk(type="message_stop", stop_reason=..., input_tokens=..., output_tokens=...)` using `chunk.usage` if available.
  - `supports()` returns `False` (no extended thinking).

---

### 4. Refactor `file:codeframe/core/adapters/streaming_chat.py`

This is the core change. The goal: **no `import anthropic`**, all provider-specific logic delegated.

- **Constructor**: Replace `api_key` / `model` / lazy `_async_client` with a `provider: LLMProvider` parameter. Keep `session_id`, `db_repo`, `workspace_path`. Remove `_api_key`, `_async_client`, `_client` property, and `_DEFAULT_MODEL`.
  - For backward compatibility with `session_chat_ws.py` (which passes no provider today), accept `provider: Optional[LLMProvider] = None` and default-construct an `AnthropicProvider()` if `None` — but prefer updating the router (step 5) to pass the provider explicitly.
- **`_stream_turn()`**: Replace the `async with self._client.messages.stream(...)` block with:
  ```
  async for chunk in self._provider.async_stream(
      messages=current_messages,
      system=<system_prompt>,
      tools=_TOOLS_FOR_API,
      model=self._model,   # or derive from provider
      max_tokens=4096,
      interrupt_event=interrupt_event,
  ):
  ```
  - Map `StreamChunk` types to `ChatEvent` types:

  | `StreamChunk.type` | `ChatEvent` emitted |
  |---|---|
  | `text_delta` | `ChatEventType.TEXT_DELTA` |
  | `thinking_delta` | `ChatEventType.THINKING` |
  | `tool_use_start` | `ChatEventType.TOOL_USE_START` |
  | `tool_use_stop` | append to `pending_tool_calls` |
  | `message_stop` | `ChatEventType.COST_UPDATE` + set `stop_reason` |

- **Extended thinking**: Before calling `async_stream()`, check `self._provider.supports("extended_thinking")`. If `True`, pass an `extended_thinking=True` kwarg (or a flag the provider understands). The provider handles the Anthropic-specific betas parameter internally.
- **`_rebuild_tool_inputs()`**: Move this helper into `AnthropicProvider` or keep it in `streaming_chat.py` but feed it from the `message_stop` `StreamChunk` (which carries the final content blocks).
- **`_estimate_cost()`**: Keep in `streaming_chat.py` for now (it's already provider-agnostic by model prefix). Alternatively, move cost estimation into each provider's `message_stop` chunk — either approach is acceptable.
- Remove the `_DEFAULT_MODEL` constant and the `model` constructor parameter; the provider's `model_selector` owns model selection. If the caller needs to override the model, they configure the `ModelSelector` on the provider.

---

### 5. Update `file:codeframe/ui/routers/session_chat_ws.py`

- In `_run_streaming_adapter()`, construct an `AnthropicProvider()` (or read provider type from app config/env) and pass it to `StreamingChatAdapter`:
  ```python
  from codeframe.adapters.llm.anthropic import AnthropicProvider
  provider = AnthropicProvider()
  adapter = StreamingChatAdapter(
      session_id=session_id,
      db_repo=db_repo,
      workspace_path=workspace_path,
      provider=provider,
  )
  ```
- This keeps the router as the composition root and avoids the adapter needing to know which provider to instantiate.

---

### 6. Update `file:codeframe/adapters/llm/__init__.py`

- Export `StreamChunk` from `base.py`.
- Export `OpenAIProvider` once `openai.py` is created.
- Update `get_provider()` to handle `"openai"` as a valid `provider_type`.

---

### 7. Update `file:tests/core/test_streaming_chat.py`

- Replace all `patch.object(adapter, "_async_client")` patches with a mock `LLMProvider` whose `async_stream()` is an `AsyncMock` yielding `StreamChunk` objects.
- Update `TestStreamingChatAdapterInit`:
  - `test_raises_if_no_api_key` → this check now lives in `AnthropicProvider.__init__`, not `StreamingChatAdapter`. Either test it there, or test that passing a provider with no key raises at provider construction time.
  - `test_accepts_api_key_from_env` and `test_default_model` → adapt to the new constructor (pass a `MockProvider` or `AnthropicProvider` with a fake key).
- Add a `MockProvider`-based `async_stream()` to `file:codeframe/adapters/llm/mock.py` that yields configurable `StreamChunk` sequences, so tests can drive the full `_stream_turn()` loop without touching Anthropic internals.

```mermaid
sequenceDiagram
    participant WS as session_chat_ws
    participant SCA as StreamingChatAdapter
    participant AP as AnthropicProvider
    participant SDK as anthropic.AsyncAnthropic

    WS->>SCA: send_message(content, history, interrupt)
    SCA->>AP: async_stream(messages, system, tools, model, max_tokens)
    AP->>SDK: messages.stream(...)
    SDK-->>AP: SDK events (content_block_delta, etc.)
    AP-->>SCA: StreamChunk(type="text_delta", text="...")
    SCA-->>WS: ChatEvent(TEXT_DELTA)
    AP-->>SCA: StreamChunk(type="message_stop", stop_reason="end_turn", ...)
    SCA-->>WS: ChatEvent(COST_UPDATE)
    SCA-->>WS: ChatEvent(DONE)
```

---

Execution Information

Branch: main
Commit: f61db8d

---

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