Fix headless thinking replay and benchmark efficiency

src/cli_core/structured_io.py

@@ -61,7 +61,8 @@ class PartialTextEvent(HeadlessEvent):
 @dataclass
 class AssistantEvent(HeadlessEvent):
     type: str = "assistant"
-    text: str = ""
+    text: str | None = None
+    message: dict[str, Any] | None = None
 
 
 @dataclass

src/entrypoints/headless.py

@@ -512,6 +512,38 @@ def _persist(msg: Any) -> None:
                                     msg.content,
                                     usage=getattr(msg, "usage", None),
                                 )
+                                # Claude Code's stream-json exposes signed
+                                # thinking blocks as assistant content events.
+                                # Emit only the private blocks here; the visible
+                                # final text keeps its existing single
+                                # AssistantEvent below.
+                                if writer is not None and msg.role == "assistant":
+                                    from src.types.content_blocks import (
+                                        RedactedThinkingBlock,
+                                        ThinkingBlock,
+                                        content_block_to_dict,
+                                    )
+
+                                    blocks = (
+                                        msg.content
+                                        if isinstance(msg.content, list)
+                                        else []
+                                    )
+                                    thinking = [
+                                        content_block_to_dict(block)
+                                        for block in blocks
+                                        if isinstance(
+                                            block,
+                                            (ThinkingBlock, RedactedThinkingBlock),
+                                        )
+                                    ]
+                                    if thinking:
+                                        writer.write(AssistantEvent(
+                                            message={
+                                                "role": "assistant",
+                                                "content": thinking,
+                                            }
+                                        ))
                             except Exception:
                                 import logging
                                 logging.getLogger(__name__).exception(

src/providers/anthropic_provider.py

@@ -390,9 +390,32 @@ def _build_chat_response(self, response: Any) -> ChatResponse:
         content_text = ""
         tool_uses: list[dict[str, Any]] = []
         raw_content_blocks: list[dict[str, Any]] = []
+        thinking_blocks: list[dict[str, Any]] = []
 
         for block in response.content:
             block_type = getattr(block, "type", "text")
+            if block_type in ("thinking", "redacted_thinking"):
+                dump = getattr(block, "model_dump", None)
+                if callable(dump):
+                    thinking_blocks.append(
+                        _strip_none_deep(dict(dump(exclude_none=True)))
+                    )
+                elif block_type == "thinking":
+                    thinking_blocks.append({
+                        "type": "thinking",
+                        "thinking": str(getattr(block, "thinking", "")),
+                        **(
+                            {"signature": str(getattr(block, "signature"))}
+                            if getattr(block, "signature", None) is not None
+                            else {}
+                        ),
+                    })
+                else:
+                    thinking_blocks.append({
+                        "type": "redacted_thinking",
+                        "data": str(getattr(block, "data", "")),
+                    })
+                continue
             block_name = getattr(block, "name", None)
             is_advisor = block_type == "advisor_tool_result" or (
                 block_type == "server_tool_use" and block_name == "advisor"
@@ -433,6 +456,7 @@ def _build_chat_response(self, response: Any) -> ChatResponse:
             finish_reason=str(getattr(response, "stop_reason", "stop")),
             tool_uses=tool_uses if tool_uses else None,
             raw_content_blocks=raw_content_blocks or None,
+            thinking_blocks=thinking_blocks or None,
         )
 
     def _client_for_request(self, kwargs: dict[str, Any]):

src/providers/base.py

@@ -41,6 +41,10 @@ class ChatResponse:
     # when the response had no such blocks. See
     # ``src/utils/advisor.py`` for the policy.
     raw_content_blocks: Optional[list[dict[str, Any]]] = None
+    # Anthropic extended-thinking blocks must be replayed byte-for-byte on
+    # follow-up tool turns.  ``thinking`` text may be empty/redacted while the
+    # signature carries the opaque state the API validates.
+    thinking_blocks: Optional[list[dict[str, Any]]] = None
 
 
 MessageInput: TypeAlias = ChatMessage | dict[str, Any]

src/query/query.py

@@ -857,8 +857,12 @@ async def _call_model_sync(
                         block_types.append(str(type(b).__name__))
                 logger.warning("[DIAG]   msg[%d] role=%s  blocks=%s", i, role, block_types)
     _t0 = time.monotonic()
+    from ..tool_system.tool_search import filter_tools_for_request
+
+    provider_model = getattr(provider, "model", None) or ""
+    request_tools = filter_tools_for_request(tools, provider_model, api_messages)
     tool_schemas = []
-    for tool in tools:
+    for tool in request_tools:
         # Filter out internal/hidden tools (is_enabled=False) so they
         # don't leak into the API tools[] alongside the advisor schema
         # we append below. Some callers pass an unfiltered tool list
@@ -1243,6 +1247,15 @@ def _do_provider_call():
     assistant_blocks: list[Any] = []
     tool_use_blocks: list[ToolUseBlock] = []
 
+    # Signed Anthropic thinking must precede the text/tool blocks exactly as
+    # returned.  Keeping it in message history both preserves reasoning state
+    # across tool turns and makes it available to stream-json/session logs.
+    if response.thinking_blocks:
+        from ..types.content_blocks import content_block_from_dict
+
+        for raw in response.thinking_blocks:
+            assistant_blocks.append(content_block_from_dict(raw))
+
     if response.content:
         assistant_blocks.append(TextBlock(text=response.content))
 

src/tool_system/tool_search.py

@@ -165,9 +165,9 @@ def is_deferred_tool(tool: Tool) -> bool:
     A tool is deferred if it's an MCP tool or has should_defer=True.
     Mirrors TS isDeferredTool from toolSearch.ts.
     """
-    if tool.is_mcp:
+    if getattr(tool, "is_mcp", False):
         return True
-    if tool.should_defer:
+    if getattr(tool, "should_defer", False):
         return True
     return False
 

src/tool_system/tools/bash/bash_tool.py

@@ -144,8 +144,26 @@ def _run_bash_with_abort(
     # output that buffered before the signal landed.
     try:
         stdout, stderr = proc.communicate(timeout=_KILL_REAP_TIMEOUT_S)
-    except subprocess.TimeoutExpired:
-        stdout, stderr = "", ""
+    except subprocess.TimeoutExpired as exc:
+        # A user command can intentionally detach a descendant without using
+        # ``run_in_background`` (for example ``nohup server ... &``). The
+        # shell exits, but the descendant's intermediary shell may retain our
+        # pipe, so communicate cannot observe EOF. TimeoutExpired still
+        # carries everything already read; preserve it instead of replacing a
+        # successful command's output with "(Bash completed with no output)".
+        def _captured(value: Any) -> str:
+            if isinstance(value, bytes):
+                return value.decode(errors="replace")
+            return value if isinstance(value, str) else ""
+
+        stdout = _captured(exc.output)
+        stderr = _captured(exc.stderr)
+        for pipe in (proc.stdout, proc.stderr):
+            if pipe is not None:
+                try:
+                    pipe.close()
+                except OSError:
+                    pass
 
     return _BashRunResult(
         returncode=proc.returncode if proc.returncode is not None else -1,

tests/test_advisor_request_wiring.py

@@ -28,6 +28,7 @@
 from src.providers.anthropic_provider import AnthropicProvider
 from src.providers.base import ChatResponse
 from src.query.query import _call_model_sync
+from src.types.content_blocks import ThinkingBlock
 from src.types.messages import AssistantMessage, UserMessage
 
 
@@ -184,6 +185,59 @@ def prompt(self) -> str:
         self.assertEqual(tools[-1]["name"], "advisor")
         self.assertEqual(tools[-1]["model"], "claude-opus-4-6")
 
+    def test_request_filter_excludes_undiscovered_deferred_tool(self) -> None:
+        cap = _Capture()
+        provider = _stub_provider_class(AnthropicProvider, cap)
+
+        class _FakeTool:
+            input_schema = {"type": "object", "properties": {}}
+            is_mcp = False
+
+            def __init__(self, name: str, should_defer: bool) -> None:
+                self.name = name
+                self.should_defer = should_defer
+
+            def prompt(self) -> str:
+                return self.name
+
+        with patch.dict(os.environ, {
+            "ENABLE_TOOL_SEARCH": "true",
+            "CLAUDE_CODE_DISABLE_EXPERIMENTAL_BETAS": "",
+        }):
+            _run(provider, [UserMessage(content="x")], tools=[
+                _FakeTool("Read", False),
+                _FakeTool("EnterPlanMode", True),
+            ])
+
+        names = [tool["name"] for tool in cap.call_kwargs["tools"]]
+        self.assertIn("Read", names)
+        self.assertNotIn("EnterPlanMode", names)
+
+    def test_signed_thinking_is_kept_in_assistant_history(self) -> None:
+        cap = _Capture()
+        provider = _stub_provider_class(AnthropicProvider, cap)
+
+        def fake_chat_stream_response(*args, **kwargs):
+            return ChatResponse(
+                content="",
+                model="claude-opus-4-8",
+                usage={"input_tokens": 1, "output_tokens": 1},
+                finish_reason="tool_use",
+                tool_uses=[{"id": "t1", "name": "Bash", "input": {"command": "true"}}],
+                thinking_blocks=[{
+                    "type": "thinking",
+                    "thinking": "",
+                    "signature": "opaque-signature",
+                }],
+            )
+
+        provider.chat_stream_response = fake_chat_stream_response
+        messages, _ = _run(provider, [UserMessage(content="x")])
+
+        assistant = next(m for m in messages if isinstance(m, AssistantMessage))
+        thinking = next(b for b in assistant.content if isinstance(b, ThinkingBlock))
+        self.assertEqual(thinking.signature, "opaque-signature")
+
     def test_instructions_appended_to_string_system_prompt(self) -> None:
         cap = _Capture()
         provider = _stub_provider_class(AnthropicProvider, cap)

tests/test_bash_timeout_vs_esc.py

@@ -246,3 +246,20 @@ def test_run_bash_with_abort_natural_exit_sets_neither_flag(tmp_path: Path) -> N
     assert result.timed_out is False
     assert result.returncode == 0
     assert "hello" in result.stdout
+
+
+def test_detached_descendant_does_not_discard_captured_stdout(tmp_path: Path) -> None:
+    """A detached child may keep the output pipe open after Bash exits."""
+    result = _run_bash_with_abort(
+        [
+            "bash",
+            "-lc",
+            "nohup sh -c 'sleep 4' >/dev/null 2>&1 & echo server-started",
+        ],
+        cwd=str(tmp_path),
+        timeout_s=10,
+        abort_signal=None,
+    )
+
+    assert result.returncode == 0
+    assert "server-started" in result.stdout

tests/test_cli_core.py

@@ -171,6 +171,22 @@ def test_stream_json_writer_escapes_line_terminators_in_output():
     assert payload["text"] == "x\u2028y"
 
 
+def test_assistant_event_can_emit_signed_thinking_message():
+    event = AssistantEvent(message={
+        "role": "assistant",
+        "content": [{
+            "type": "thinking",
+            "thinking": "",
+            "signature": "opaque-signature",
+        }],
+    })
+
+    payload = event.to_dict()
+
+    assert "text" not in payload
+    assert payload["message"]["content"][0]["signature"] == "opaque-signature"
+
+
 def test_stream_json_writer_accepts_plain_dict():
     buf = io.StringIO()
     writer = StreamJsonWriter(buf)

tests/test_providers.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import unittest
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 from src.providers import get_provider_class
@@ -71,6 +72,38 @@ def test_custom_model(self):
         provider = AnthropicProvider(api_key="test_key", model="claude-3-opus-20240229")
         self.assertEqual(provider.model, "claude-3-opus-20240229")
 
+    def test_build_response_preserves_signed_thinking_blocks(self):
+        provider = AnthropicProvider(api_key="test_key")
+        response = SimpleNamespace(
+            content=[
+                SimpleNamespace(
+                    type="thinking",
+                    thinking="",
+                    signature="opaque-signature",
+                ),
+                SimpleNamespace(type="redacted_thinking", data="opaque-data"),
+                SimpleNamespace(type="text", text="done"),
+            ],
+            model="claude-opus-4-8",
+            usage=SimpleNamespace(input_tokens=10, output_tokens=5),
+            stop_reason="end_turn",
+        )
+
+        result = provider._build_chat_response(response)
+
+        self.assertEqual(result.content, "done")
+        self.assertEqual(
+            result.thinking_blocks,
+            [
+                {
+                    "type": "thinking",
+                    "thinking": "",
+                    "signature": "opaque-signature",
+                },
+                {"type": "redacted_thinking", "data": "opaque-data"},
+            ],
+        )
+
     def test_get_available_models(self):
         """Test getting available models."""
         provider = AnthropicProvider(api_key="test_key")