diff --git a/src/beaver_gateway/agents/claude.py b/src/beaver_gateway/agents/claude.py index a1846ef..b7c6c48 100644 --- a/src/beaver_gateway/agents/claude.py +++ b/src/beaver_gateway/agents/claude.py @@ -1,8 +1,13 @@ """Claude Code agent definition. -Notice the absence of a ``streaming`` field — claude-code does not emit -token-level deltas, and that fact is encoded in the type, not in a -runtime branch. +There is no ``streaming`` field, but there is +:attr:`ClaudeCodeOptions.include_partial_messages`, and the difference +matters. Token-level deltas are not a property of the agent, they are a +property of the transport: the PTY transport reads a JSONL that only +ever contains finished blocks, so it cannot stream no matter what the +agent asks for, while ``transport="stream_json"`` can. The flag lives +next to the transport that grants it rather than pretending to be a +free-standing capability. ``BaseAgent.system_prompt`` maps onto the claude CLI's ``--system-prompt`` — i.e. it really *is* the agent's system prompt, @@ -57,6 +62,9 @@ HistoryInjectionMode = Literal["native_jsonl", "concat_message"] cycle on the user-facing path (configs may be loaded without ``claude-code-api`` installed, e.g. ``--extra prod`` minus claude).""" +Transport = Literal["pty", "stream_json"] +"""Mirrors ``claude_code_api.Transport``, for the same reason.""" + class ClaudeCodeOptions(BaseModel): """Per-agent passthrough for ``claude_code_api.BackendOptions``. @@ -85,6 +93,32 @@ class ClaudeCodeOptions(BaseModel): model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True) + transport: Transport = "pty" + """Which claude process backs this agent. + + ``stream_json`` runs ``claude -p`` with stream-json on stdin and + stdout: prompts go into a pipe, events come back as they happen, and + a native ``result`` record closes each turn. ``pty`` (the default, + kept for backwards compatibility) drives the interactive TUI and + tails the session JSONL, which costs a multi-second readiness wait + per spawn — the thing ``startup_delay`` bounds — and brings the + swallowed-paste failure mode with it. + + Switching also makes the PTY-only knobs below inert + (``startup_delay`` / ``file_wait_timeout`` / + ``turn_duration_timeout`` / ``wait_for_turn_duration``) and drops + interactive-only ``extra_args`` such as ``--remote-control``, which + would take claude out of print mode. Everything else — model, + prompts, tools, MCPs, history seeding — behaves identically.""" + + include_partial_messages: bool = False + """Stream token-level deltas to the client instead of one delta per + finished block. Requires ``transport="stream_json"``; inert on the + PTY transport, whose JSONL contains only finished blocks. Off by + default because it multiplies the SSE event count — worth it for + interactive chat, pointless for a client that waits for the whole + message anyway.""" + append_system_prompt: str | None = None """Maps to claude CLI's ``--append-system-prompt``. Opting in re-attaches claude-code's full built-in prompt (persona, planning diff --git a/src/beaver_gateway/backends/claude_code.py b/src/beaver_gateway/backends/claude_code.py index dfca1b5..54b9494 100644 --- a/src/beaver_gateway/backends/claude_code.py +++ b/src/beaver_gateway/backends/claude_code.py @@ -13,11 +13,20 @@ Per :meth:`complete` we: * hand the full Anthropic-style ``messages`` list to ``ClaudeCodeBackend.complete`` — it does its own fingerprint-based session lookup, so we never need to track sessions ourselves; -* re-emit each ``AssistantMessage`` as ``content_block_start`` + - one delta + ``content_block_stop`` per content block, with - monotonically increasing indices spanning the entire turn (one - ``message_start`` … ``message_stop`` envelope per ``complete`` call); -* close the envelope on the synthesized ``ResultMessage``. +* turn its events into one ``message_start`` … ``message_stop`` + envelope per ``complete`` call, with content-block indices increasing + monotonically across the whole turn; +* close the envelope on the ``ResultMessage``. + +How the blocks inside that envelope are produced depends on the agent's +transport. By default each ``AssistantMessage`` is re-emitted as +``content_block_start`` + one delta + ``content_block_stop`` per block — +a "stream" of finished blocks, because the PTY transport has nothing +finer to offer. With ``transport="stream_json"`` *and* +``include_partial_messages``, the backend instead feeds us the real +token-level ``StreamEvent``s and we emit from those; the whole-block +records still arrive and are still kept for :class:`TurnCapture`, but +emitting from both would duplicate every block on the wire. The per-request ``system`` parameter is intentionally **ignored** — ``BackendOptions.system_prompt`` is fixed at session-spawn time, and the @@ -29,6 +38,7 @@ from __future__ import annotations import json import logging import uuid +from collections.abc import Mapping # runtime import: isinstance in _emit_stream_event from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any, Self @@ -37,6 +47,7 @@ from claude_code_api import ( BackendOptions, ClaudeCodeBackend, ResultMessage, + StreamEvent, TextBlock, ThinkingBlock, ToolUseBlock, @@ -60,7 +71,7 @@ from beaver_gateway.core.events import ( ) if TYPE_CHECKING: - from collections.abc import AsyncIterator, Iterable, Mapping + from collections.abc import AsyncIterator, Iterable from anthropic.types import MessageParam @@ -184,6 +195,8 @@ def _build_backend_options( append_system_prompt=opt.append_system_prompt, allowed_tools=allowed_tools, mcp_servers=_build_mcp_servers(agent, mcp_internal_urls), + transport=opt.transport, + include_partial_messages=opt.include_partial_messages, disallowed_tools=opt.disallowed_tools, permission_mode=opt.permission_mode, dangerously_skip_permissions=opt.dangerously_skip_permissions, @@ -215,9 +228,16 @@ class ClaudeCodeBackendAdapter: self, *, agent: ClaudeAgent, mcp_internal_urls: Mapping[str, str] ) -> None: self._agent = agent - self._backend = ClaudeCodeBackend( - _build_backend_options(agent, mcp_internal_urls) + options = _build_backend_options(agent, mcp_internal_urls) + self._backend = ClaudeCodeBackend(options) + self._streaming = ( + options.transport == "stream_json" and options.include_partial_messages ) + """Whether the backend feeds us token-level ``StreamEvent``s. + + When it does, blocks are emitted from those and the whole-block + ``AssistantMessage`` records are used only for bookkeeping — + emitting from both would duplicate every block on the wire.""" @property def agent(self) -> ClaudeAgent: @@ -229,12 +249,16 @@ class ClaudeCodeBackendAdapter: @property def live_sessions(self) -> dict[str, Any]: - """Live PTY processes keyed by claude session_id. + """Live claude processes keyed by claude session_id. Pass-through to the underlying ``ClaudeCodeBackend``. The value - type is the claude-code-api ``PtyClaudeProcess``; admin code - consumes ``captured_output()`` / ``add_output_listener`` / - ``write`` from it. Typed as ``Any`` to avoid leaking the lower + is a ``PtyClaudeProcess`` or a ``StreamClaudeProcess`` depending + on the agent's transport; both implement the + ``captured_output()`` / ``add_output_listener`` / ``write`` + surface the admin terminal consumes, so it works either way — + with the caveat that the stream transport's "terminal" is a + read-only view of the JSON event stream rather than a TUI you + can type into. Typed as ``Any`` to avoid leaking the lower layer's type into the gateway's public surface. """ return self._backend.live_sessions @@ -302,7 +326,13 @@ class ClaudeCodeBackendAdapter: async for event in self._backend.complete(msgs_list): raw_events.append(event) - if isinstance(event, AssistantMessage): + if isinstance(event, StreamEvent): + # Only ever arrives when ``self._streaming`` — the + # backend doesn't emit these otherwise. Indices are + # already rebased onto the whole turn by claude-code-api. + for ev in _emit_stream_event(event.event): + yield ev + elif isinstance(event, AssistantMessage): for block in event.content: if isinstance(block, TextBlock): n_text += 1 @@ -310,8 +340,9 @@ class ClaudeCodeBackendAdapter: n_thinking += 1 elif isinstance(block, ToolUseBlock): n_tool_use += 1 - for ev in _emit_block(block, next_index): - yield ev + if not self._streaming: + for ev in _emit_block(block, next_index): + yield ev next_index += 1 elif isinstance(event, ResultMessage): # ResultMessage is the terminal event from TurnManager @@ -349,6 +380,67 @@ class ClaudeCodeBackendAdapter: yield build_message_stop() +def _emit_stream_event(event: Mapping[str, Any]) -> Iterable[MessageStreamEvent]: + """Render one raw Anthropic streaming event through our own builders. + + The payload is already Anthropic-shaped, so forwarding it verbatim + is tempting — but it arrives as a plain dict and the frontend + serializes pydantic models, so it would have to be validated against + the SDK's ``RawMessageStreamEvent`` union anyway. Rebuilding it from + the fields we recognize is the same amount of work and fails the + same way :func:`_emit_block` already does: a block or delta type + this gateway doesn't know is dropped rather than raising mid-stream + on a claude release that added one. + + Only ``content_block_*`` events reach here — claude-code-api strips + the inner per-request message envelopes, because the caller owns the + single envelope spanning the whole turn. + """ + etype = event.get("type") + index = event.get("index") + if not isinstance(index, int): + return () + + if etype == "content_block_start": + block = event.get("content_block") + if not isinstance(block, Mapping): + return () + btype = block.get("type") + if btype == "text": + return (build_text_block_start(index),) + if btype == "thinking": + return (build_thinking_block_start(index),) + if btype == "tool_use": + return ( + build_tool_use_block_start( + index, + tool_use_id=str(block.get("id", "")), + name=str(block.get("name", "")), + ), + ) + return () + + if etype == "content_block_delta": + delta = event.get("delta") + if not isinstance(delta, Mapping): + return () + dtype = delta.get("type") + if dtype == "text_delta": + return (build_text_delta(index, str(delta.get("text", ""))),) + if dtype == "thinking_delta": + return (build_thinking_delta(index, str(delta.get("thinking", ""))),) + if dtype == "signature_delta": + return (build_signature_delta(index, str(delta.get("signature", ""))),) + if dtype == "input_json_delta": + return (build_input_json_delta(index, str(delta.get("partial_json", ""))),) + return () + + if etype == "content_block_stop": + return (build_content_block_stop(index),) + + return () + + def _emit_block( block: TextBlock | ThinkingBlock | ToolUseBlock | Any, index: int ) -> Iterable[MessageStreamEvent]: diff --git a/uv.lock b/uv.lock index 6a7c147..928f295 100644 --- a/uv.lock +++ b/uv.lock @@ -287,7 +287,7 @@ local = [ { name = "raycast-api", version = "0.1.0", source = { editable = "../raycast-api" } }, ] prod = [ - { name = "claude-code-api", version = "0.1.0", source = { git = "https://git.kotikot.com/beaver/claude-code-api.git#aa7beea1e014ba0dd8cd4980fee4b9302c57b212" } }, + { name = "claude-code-api", version = "0.1.0", source = { git = "https://git.kotikot.com/beaver/claude-code-api.git#76799179d9437dcd133a40ffcf5eecc524063c2f" } }, { name = "raycast-api", version = "0.1.0", source = { git = "https://git.kotikot.com/beaver/raycast-api.git#e73894c8e435da5c0709f92f69f11bcd0dab9afe" } }, ] @@ -419,7 +419,7 @@ wheels = [ [[package]] name = "claude-code-api" version = "0.1.0" -source = { git = "https://git.kotikot.com/beaver/claude-code-api.git#aa7beea1e014ba0dd8cd4980fee4b9302c57b212" } +source = { git = "https://git.kotikot.com/beaver/claude-code-api.git#76799179d9437dcd133a40ffcf5eecc524063c2f" } resolution-markers = [ "python_full_version >= '3.14'", "python_full_version < '3.14'",