Anthropic extended thinking, opt-in via provider thinking_budget field

Thinking/redacted blocks replay verbatim in tool loops via the _anthropic sidecar.
Streaming thinking_delta/signature_delta surface as reasoning; sampling knobs dropped.
Live-verified tool loop + streaming on claude-haiku-4-5.

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),
             )
         )

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",

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"