diff --git a/src/cli_core/structured_io.py b/src/cli_core/structured_io.py index df966382d..f4c362eb6 100644 --- a/src/cli_core/structured_io.py +++ b/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 diff --git a/src/entrypoints/headless.py b/src/entrypoints/headless.py index 0282425b0..7f301ead7 100644 --- a/src/entrypoints/headless.py +++ b/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( diff --git a/src/providers/anthropic_provider.py b/src/providers/anthropic_provider.py index e01b69abf..51093fe04 100644 --- a/src/providers/anthropic_provider.py +++ b/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]): diff --git a/src/providers/base.py b/src/providers/base.py index f7b5716f7..7c8ab95dc 100644 --- a/src/providers/base.py +++ b/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] diff --git a/src/query/query.py b/src/query/query.py index c9216a9f6..12a24be97 100644 --- a/src/query/query.py +++ b/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)) diff --git a/src/tool_system/tool_search.py b/src/tool_system/tool_search.py index 1e29fff75..65540b1ec 100644 --- a/src/tool_system/tool_search.py +++ b/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 diff --git a/src/tool_system/tools/bash/bash_tool.py b/src/tool_system/tools/bash/bash_tool.py index 172acabd1..0a6915e1c 100644 --- a/src/tool_system/tools/bash/bash_tool.py +++ b/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, diff --git a/tests/test_advisor_request_wiring.py b/tests/test_advisor_request_wiring.py index 3f0ba180c..61c5140f1 100644 --- a/tests/test_advisor_request_wiring.py +++ b/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) diff --git a/tests/test_bash_timeout_vs_esc.py b/tests/test_bash_timeout_vs_esc.py index b9e0b19fb..9326d1e06 100644 --- a/tests/test_bash_timeout_vs_esc.py +++ b/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 diff --git a/tests/test_cli_core.py b/tests/test_cli_core.py index 4be03e6c4..a3e2d2978 100644 --- a/tests/test_cli_core.py +++ b/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) diff --git a/tests/test_providers.py b/tests/test_providers.py index c24137f08..4aad06c1f 100644 --- a/tests/test_providers.py +++ b/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")