diff --git a/coworker/providers/anthropic_provider.py b/coworker/providers/anthropic_provider.py index fb264fb6..e01dd33d 100644 --- a/coworker/providers/anthropic_provider.py +++ b/coworker/providers/anthropic_provider.py @@ -10,6 +10,11 @@ from chat.completions in ways the converters must absorb: - Tool results are `tool_result` blocks that must ALL land in the single next user message — N consecutive `role:"tool"` messages collapse into one user message here. - `max_tokens` is required. +- Extended thinking (opt-in via the provider profile's `thinking_budget` field): responses + carry `thinking`/`redacted_thinking` blocks that MUST be replayed verbatim (signatures + and all) ahead of the same turn's tool_use blocks when returning tool results — they ride + the canonical assistant message as the `_anthropic` sidecar and are reattached here. The + thinking text also lands on `AssistantTurn.reasoning` for display. """ from __future__ import annotations @@ -48,8 +53,12 @@ _SETTINGS_WHITELIST = { "top_k", "stop_sequences", "metadata", + "thinking", } +# Sampling knobs the API rejects alongside extended thinking (temperature must stay 1). +_THINKING_INCOMPATIBLE = ("temperature", "top_p", "top_k") + _DATA_URL_RE = re.compile( r"^data:(image/[a-z0-9.+-]+);base64,(.+)$", re.IGNORECASE | re.DOTALL ) @@ -193,6 +202,9 @@ def convert_messages( converted.append({"role": "user", "content": blocks}) elif role == "assistant": blocks = [] + # Replay thinking/redacted_thinking blocks VERBATIM, ahead of the turn's own + # blocks — required whenever the turn's tool calls are being answered. + blocks.extend((message.get("_anthropic") or {}).get("blocks") or []) text = message.get("content") if isinstance(text, str) and text: blocks.append({"type": "text", "text": text}) @@ -257,6 +269,19 @@ def convert_tools(tools: Optional[list[dict[str, Any]]]) -> list[dict[str, Any]] return converted +def _reasoning_text(thinking_blocks: list[dict[str, Any]]) -> Optional[str]: + """Display text for the GUI's disclosure — thinking text only (redacted stays opaque).""" + text = "".join( + b.get("thinking", "") for b in thinking_blocks if b.get("type") == "thinking" + ) + return text or None + + +def _thinking_extras(thinking_blocks: list[dict[str, Any]]) -> dict[str, Any]: + """Raw blocks → the `_anthropic` sidecar convert_messages replays (empty when none).""" + return {"_anthropic": {"blocks": thinking_blocks}} if thinking_blocks else {} + + class AnthropicProvider(ProviderClient): def __init__( self, @@ -265,14 +290,17 @@ class AnthropicProvider(ProviderClient): default_model: str = "claude-sonnet-4-6", api_key: Optional[str] = None, secrets: Any = None, + thinking_budget: Optional[int] = None, ): # Mirrors OpenAIProvider: the SDK client is built lazily so engines can be assembled # before any key exists; the key resolves at call time (explicit → env → SecretStore). - # Tests inject a `client` directly. + # Tests inject a `client` directly. `thinking_budget` (tokens, from the provider + # profile's optional field) opts every request into extended thinking. self._client = client self._api_key = api_key self._secrets = secrets self.default_model = default_model + self.thinking_budget = thinking_budget or 0 def _ensure_client(self) -> Any: if self._client is None: @@ -301,6 +329,20 @@ class AnthropicProvider(ProviderClient): stop = settings["stop"] settings["stop_sequences"] = [stop] if isinstance(stop, str) else list(stop) filtered = {k: v for k, v in settings.items() if k in _SETTINGS_WHITELIST} + if self.thinking_budget > 0: + filtered.setdefault( + "thinking", + {"type": "enabled", "budget_tokens": self.thinking_budget}, + ) + thinking = filtered.get("thinking") or {} + if thinking.get("type") == "enabled": + # Budget must fit under max_tokens, and sampling knobs are rejected. + budget = int(thinking.get("budget_tokens") or 0) + floor = max(DEFAULT_MAX_TOKENS, budget + 4096) + if int(filtered.get("max_tokens") or 0) <= budget: + filtered["max_tokens"] = floor + for key in _THINKING_INCOMPATIBLE: + filtered.pop(key, None) filtered.setdefault("max_tokens", DEFAULT_MAX_TOKENS) kwargs: dict[str, Any] = {"model": model, "messages": converted, **filtered} if system: @@ -324,6 +366,7 @@ class AnthropicProvider(ProviderClient): text_parts: list[str] = [] tool_calls: list[ToolCall] = [] + thinking_blocks: list[dict[str, Any]] = [] for block in getattr(response, "content", None) or []: kind = getattr(block, "type", None) if kind == "text": @@ -336,12 +379,29 @@ class AnthropicProvider(ProviderClient): arguments=dict(getattr(block, "input", None) or {}), ) ) + elif kind == "thinking": + thinking_blocks.append( + { + "type": "thinking", + "thinking": getattr(block, "thinking", "") or "", + "signature": getattr(block, "signature", "") or "", + } + ) + elif kind == "redacted_thinking": + thinking_blocks.append( + { + "type": "redacted_thinking", + "data": getattr(block, "data", "") or "", + } + ) stop_reason = getattr(response, "stop_reason", None) return AssistantTurn( text="".join(text_parts) or None, tool_calls=tool_calls, finish_reason=_STOP_REASON_MAP.get(stop_reason, stop_reason), raw=response, + reasoning=_reasoning_text(thinking_blocks), + extras=_thinking_extras(thinking_blocks), ) def capabilities(self, model: str) -> ModelCapabilities: @@ -363,18 +423,34 @@ class AnthropicProvider(ProviderClient): text_parts: list[str] = [] tool_accum: dict[int, dict[str, str]] = {} + # Thinking blocks accumulate per stream index and must be replayed verbatim later, + # so both the text and the signature_delta tail are collected (in block order). + thinking_accum: dict[int, dict[str, Any]] = {} stop_reason = None for event in client.messages.create(**kwargs): kind = getattr(event, "type", None) if kind == "content_block_start": block = getattr(event, "content_block", None) - if getattr(block, "type", None) == "tool_use": + block_kind = getattr(block, "type", None) + if block_kind == "tool_use": tool_accum[getattr(event, "index", 0)] = { "id": getattr(block, "id", "") or "", "name": getattr(block, "name", "") or "", "json": "", } + elif block_kind == "thinking": + thinking_accum[getattr(event, "index", 0)] = { + "type": "thinking", + "thinking": getattr(block, "thinking", "") or "", + "signature": getattr(block, "signature", "") or "", + } + elif block_kind == "redacted_thinking": + # Arrives whole — opaque data, no deltas. + thinking_accum[getattr(event, "index", 0)] = { + "type": "redacted_thinking", + "data": getattr(block, "data", "") or "", + } elif kind == "content_block_delta": delta = getattr(event, "delta", None) delta_kind = getattr(delta, "type", None) @@ -387,7 +463,18 @@ class AnthropicProvider(ProviderClient): acc = tool_accum.get(getattr(event, "index", 0)) if acc is not None: acc["json"] += getattr(delta, "partial_json", "") or "" - # thinking/signature deltas are ignored + elif delta_kind == "thinking_delta": + acc = thinking_accum.get(getattr(event, "index", 0)) + thought = getattr(delta, "thinking", "") or "" + if acc is not None and thought: + acc["thinking"] += thought + yield StreamChunk(reasoning_delta=thought) + elif delta_kind == "signature_delta": + acc = thinking_accum.get(getattr(event, "index", 0)) + if acc is not None: + acc["signature"] = (acc.get("signature") or "") + ( + getattr(delta, "signature", "") or "" + ) elif kind == "message_delta": reason = getattr(getattr(event, "delta", None), "stop_reason", None) if reason: @@ -401,11 +488,14 @@ class AnthropicProvider(ProviderClient): id=acc["id"], name=acc["name"], arguments=_parse_args(acc["json"]) ) ) + thinking_blocks = [thinking_accum[i] for i in sorted(thinking_accum)] yield StreamChunk( turn=AssistantTurn( text="".join(text_parts) or None, tool_calls=tool_calls, finish_reason=_STOP_REASON_MAP.get(stop_reason, stop_reason), + reasoning=_reasoning_text(thinking_blocks), + extras=_thinking_extras(thinking_blocks), ) ) diff --git a/coworker/providers/registry.py b/coworker/providers/registry.py index 05897277..ccf13d2a 100644 --- a/coworker/providers/registry.py +++ b/coworker/providers/registry.py @@ -107,7 +107,13 @@ def _build_anthropic(profile: dict[str, Any], secrets: Any) -> ProviderClient: # Key resolution stays in AnthropicProvider/resolve_api_key (explicit → env → SecretStore), # deferred to first call so the provider can be built before a key exists. api_key = ((profile or {}).get("api_key") or "").strip() or None - return AnthropicProvider(api_key=api_key, secrets=secrets) + try: + thinking_budget = int(str((profile or {}).get("thinking_budget") or "").strip()) + except ValueError: + thinking_budget = 0 + return AnthropicProvider( + api_key=api_key, secrets=secrets, thinking_budget=thinking_budget + ) def _build_gemini(profile: dict[str, Any], secrets: Any) -> ProviderClient: @@ -219,6 +225,13 @@ DESCRIPTORS: list[ProviderDescriptor] = [ secret=True, placeholder="sk-ant-…", ), + ProviderField( + "thinking_budget", + "Extended thinking budget (tokens, optional)", + required=False, + placeholder="e.g. 8192 — blank = off", + help="Turns on Claude's extended thinking for every request, with this token budget. The thought process shows in the transcript.", + ), ], build=_build_anthropic, recommended_model="claude-fable-5", diff --git a/tests/test_anthropic_provider.py b/tests/test_anthropic_provider.py index 8433a0ab..c6a5d81d 100644 --- a/tests/test_anthropic_provider.py +++ b/tests/test_anthropic_provider.py @@ -526,3 +526,100 @@ def test_convert_malformed_file_part_becomes_text_note(): "type": "text", "text": "[unsupported file attachment]", } + + +# -- extended thinking (model-layer roadmap item 4, phase 3) ------------------------ + + +def test_thinking_budget_enables_thinking_and_drops_sampling_knobs(): + client = _FakeClient(response=_text_response()) + provider = AnthropicProvider(client=client, thinking_budget=8192) + provider.complete( + model="claude-fable-5", + messages=[{"role": "user", "content": "x"}], + temperature=0.2, + top_p=0.9, + ) + assert client.kwargs["thinking"] == {"type": "enabled", "budget_tokens": 8192} + assert client.kwargs["max_tokens"] > 8192 + assert "temperature" not in client.kwargs and "top_p" not in client.kwargs + # Off by default: no budget → no thinking param. + client2 = _FakeClient(response=_text_response()) + AnthropicProvider(client=client2).complete( + model="claude-fable-5", messages=[{"role": "user", "content": "x"}] + ) + assert "thinking" not in client2.kwargs + + +def test_complete_parses_thinking_blocks_into_reasoning_and_sidecar(): + response = SimpleNamespace( + content=[ + SimpleNamespace(type="thinking", thinking="pondering...", signature="SIG1"), + SimpleNamespace(type="redacted_thinking", data="OPAQUE"), + SimpleNamespace(type="text", text="the answer"), + ], + stop_reason="end_turn", + ) + provider = AnthropicProvider(client=_FakeClient(response=response), thinking_budget=4096) + turn = provider.complete(model="claude-fable-5", messages=[{"role": "user", "content": "x"}]) + assert turn.text == "the answer" + assert turn.reasoning == "pondering..." # redacted stays out of the display text + assert turn.extras["_anthropic"]["blocks"] == [ + {"type": "thinking", "thinking": "pondering...", "signature": "SIG1"}, + {"type": "redacted_thinking", "data": "OPAQUE"}, + ] + + +def test_stream_accumulates_thinking_and_signature(): + events = [ + SimpleNamespace( + type="content_block_start", + index=0, + content_block=SimpleNamespace(type="thinking", thinking="", signature=""), + ), + _delta(0, type="thinking_delta", thinking="step one. "), + _delta(0, type="thinking_delta", thinking="step two."), + _delta(0, type="signature_delta", signature="SIGSTREAM"), + SimpleNamespace(type="content_block_stop", index=0), + SimpleNamespace( + type="content_block_start", + index=1, + content_block=SimpleNamespace(type="text"), + ), + _delta(1, type="text_delta", text="done"), + SimpleNamespace( + type="message_delta", delta=SimpleNamespace(stop_reason="end_turn") + ), + ] + provider = AnthropicProvider(client=_FakeClient(events=events), thinking_budget=4096) + chunks = list(provider.stream(model="m", messages=[{"role": "user", "content": "x"}])) + assert [c.reasoning_delta for c in chunks if c.reasoning_delta] == ["step one. ", "step two."] + final = chunks[-1].turn + assert final.text == "done" and final.reasoning == "step one. step two." + assert final.extras["_anthropic"]["blocks"] == [ + {"type": "thinking", "thinking": "step one. step two.", "signature": "SIGSTREAM"} + ] + + +def test_convert_replays_thinking_blocks_ahead_of_tool_use(): + _, msgs = convert_messages( + [ + {"role": "user", "content": "go"}, + { + "role": "assistant", + "content": "", + "_anthropic": { + "blocks": [ + {"type": "thinking", "thinking": "plan", "signature": "S"}, + ] + }, + "tool_calls": [ + {"id": "t1", "type": "function", "function": {"name": "a", "arguments": "{}"}}, + ], + }, + {"role": "tool", "tool_call_id": "t1", "content": "ok"}, + ] + ) + assistant = msgs[1]["content"] + assert assistant[0] == {"type": "thinking", "thinking": "plan", "signature": "S"} + assert assistant[1]["type"] == "tool_use"