diff --git a/agents/bug-fix/hackbot_agents/bug_fix/agent.py b/agents/bug-fix/hackbot_agents/bug_fix/agent.py index 0135880baa..cb22d52ac6 100644 --- a/agents/bug-fix/hackbot_agents/bug_fix/agent.py +++ b/agents/bug-fix/hackbot_agents/bug_fix/agent.py @@ -19,12 +19,15 @@ ClaudeAgentOptions, ClaudeSDKClient, McpServerConfig, - ResultMessage, ) from hackbot_runtime import ActionsRecorder, AgentError, HackbotAgentResult from hackbot_runtime.actions import ACTIONS_SERVER_NAME from hackbot_runtime.actions.claude_sdk import actions_server_for, actions_to_tool_names -from hackbot_runtime.claude import Reporter +from hackbot_runtime.claude import ( + Reporter, + UnsettledResponseError, + receive_settled_response, +) from .config import ( BUGZILLA_READ_TOOLS, @@ -98,6 +101,7 @@ async def run_bug_fix( verbose: bool = False, log: Path | None = None, actions_recorder: ActionsRecorder | None = None, + background_task_timeout_s: float = 3 * 60 * 60, ) -> BugFixResult: """Triage and fix a single Bugzilla bug with a claude-agent-sdk agent. @@ -165,18 +169,19 @@ async def run_bug_fix( setting_sources=[], ) - result_msg: ResultMessage | None = None with Reporter(verbose=verbose, log_path=log) as reporter: reporter.header(f"bug {bug}") async with ClaudeSDKClient(options=options) as client: await client.query(user_prompt) - async for msg in client.receive_response(): - reporter.message(msg) - if isinstance(msg, ResultMessage): - result_msg = msg + try: + result_msg = await receive_settled_response( + client, + on_message=reporter.message, + timeout_s=background_task_timeout_s, + ) + except UnsettledResponseError as exc: + raise AgentError(f"bug {bug}: agent run did not settle: {exc}") from exc - if result_msg is None: - raise AgentError(f"bug {bug}: agent produced no result message") if result_msg.is_error: raise AgentError( f"bug {bug} triage failed: {result_msg.result or result_msg.subtype}" diff --git a/libs/agent-tools/pyproject.toml b/libs/agent-tools/pyproject.toml index 28d29790ba..9c49fceb41 100644 --- a/libs/agent-tools/pyproject.toml +++ b/libs/agent-tools/pyproject.toml @@ -15,7 +15,7 @@ bugzilla = [ "six", ] firefox = ["grizzly-framework", "prefpicker"] -claude-sdk = ["claude-agent-sdk>=0.1.30"] +claude-sdk = ["claude-agent-sdk>=0.2.30"] searchfox = ["searchfox>=0.20.3"] vcs = ["httpx"] diff --git a/libs/hackbot-runtime/hackbot_runtime/claude.py b/libs/hackbot-runtime/hackbot_runtime/claude.py index da1d6c2234..d00c6bd233 100644 --- a/libs/hackbot-runtime/hackbot_runtime/claude.py +++ b/libs/hackbot-runtime/hackbot_runtime/claude.py @@ -10,13 +10,21 @@ from __future__ import annotations +import asyncio import json +from collections.abc import Callable from pathlib import Path from claude_agent_sdk import ( + TERMINAL_TASK_STATUSES, AssistantMessage, + ClaudeSDKClient, + Message, ResultMessage, SystemMessage, + TaskNotificationMessage, + TaskStartedMessage, + TaskUpdatedMessage, TextBlock, ThinkingBlock, ToolResultBlock, @@ -25,6 +33,33 @@ ) +class UnsettledResponseError(RuntimeError): + """``receive_settled_response`` gave up before the agent's turn settled. + + ``pending`` is the ``task_id -> description`` map of deferring tasks + still open when this was raised (empty if none were ever pending — the + connection just ended with no result at all). Stored as an attribute + for callers that want to inspect it programmatically, e.g. to name the + stuck task(s) rather than just log the message. + """ + + def __init__(self, reason: str, pending: dict[str, str]): + self.reason = reason + self.pending = pending + + def __str__(self) -> str: + if not self.pending: + return self.reason + return f"{self.reason} ({len(self.pending)} task(s) still pending)" + + +# Task types whose completion the CLI itself resumes the turn for — mirrors +# claude_agent_sdk._internal.query.DEFERRING_TASK_TYPES (not public API, so +# duplicated here rather than imported). +# https://github.com/anthropics/claude-agent-sdk-python/blob/bc0c9af676d9a63ac20a98cf1b7ba4794382c3cc/src/claude_agent_sdk/_internal/query.py#L38-L52 +_DEFERRING_TASK_TYPES = frozenset({"local_agent", "local_workflow"}) + + def _truncate(s: str, n: int = 500) -> str: return s if len(s) <= n else s[:n] + f"... [{len(s) - n} more chars]" @@ -123,3 +158,66 @@ def message(self, msg) -> None: self._emit(line, always=True) if msg.is_error: self._emit(f"[done] ERROR: {msg.result}", always=True) + + +async def receive_settled_response( + client: ClaudeSDKClient, + on_message: Callable[["Message"], None] | None = None, + *, + timeout_s: float = 3600, +) -> ResultMessage: + """Drive ``client`` to a *settled* :class:`ResultMessage`. + + ``client.receive_response()`` stops at the first ``ResultMessage``, but + the CLI can emit one while a task the agent backgrounded is still + running, reporting the turn "done" prematurely (see + anthropics/claude-agent-sdk-python#1138). This drains + ``client.receive_messages()`` instead (it doesn't stop at a + ``ResultMessage``) and only returns once one arrives with no *deferring* + task (``local_agent``/``local_workflow``, see ``_DEFERRING_TASK_TYPES``) + still open — backgrounded shells and Monitor watches run forever by + design and are never waited on. A task's terminal state can arrive as + either a ``TaskNotificationMessage`` or a ``TaskUpdatedMessage``, so both + clear it. + + Args: + client: A connected client with a query already sent. + on_message: Called with each message as it streams in, before this + function's own bookkeeping. Optional. + timeout_s: Bounds the wait so a task that never settles raises + ``UnsettledResponseError`` instead of hanging. Defaults to an + hour (a full Firefox build); pass a larger value for + longer-running work. + + Raises: + UnsettledResponseError: timed out, or the connection ended before + any ``ResultMessage`` arrived. + """ + pending: dict[str, str] = {} + result_msg: ResultMessage | None = None + + try: + async with asyncio.timeout(timeout_s): + async for msg in client.receive_messages(): + if on_message is not None: + on_message(msg) + + if isinstance(msg, TaskStartedMessage): + if msg.task_type in _DEFERRING_TASK_TYPES: + pending[msg.task_id] = msg.description + elif isinstance(msg, (TaskNotificationMessage, TaskUpdatedMessage)): + if msg.status in TERMINAL_TASK_STATUSES: + pending.pop(msg.task_id, None) + elif isinstance(msg, ResultMessage): + result_msg = msg + if not pending: + return result_msg + except TimeoutError as exc: + raise UnsettledResponseError( + f"timed out after {timeout_s:.0f}s waiting for the response to settle", + pending, + ) from exc + + raise UnsettledResponseError( + "connection ended before a settled ResultMessage arrived", pending + ) diff --git a/libs/hackbot-runtime/tests/test_claude.py b/libs/hackbot-runtime/tests/test_claude.py index 0433437778..d2752f0901 100644 --- a/libs/hackbot-runtime/tests/test_claude.py +++ b/libs/hackbot-runtime/tests/test_claude.py @@ -1,6 +1,21 @@ -"""Tests for the shared claude-agent-sdk Reporter (hackbot_runtime.claude).""" +"""Tests for the shared claude-agent-sdk helpers (hackbot_runtime.claude).""" -from hackbot_runtime.claude import Reporter, _truncate +import asyncio + +import pytest +from claude_agent_sdk import ( + ResultMessage, + SystemMessage, + TaskNotificationMessage, + TaskStartedMessage, + TaskUpdatedMessage, +) +from hackbot_runtime.claude import ( + Reporter, + UnsettledResponseError, + _truncate, + receive_settled_response, +) def test_truncate_short_string_unchanged(): @@ -34,3 +49,136 @@ def test_no_log_file_when_path_is_none(tmp_path): with Reporter(verbose=True, log_path=None) as reporter: reporter.header("section") assert not list(tmp_path.iterdir()) + + +class _FakeClient: + """Replays a fixed message list from ``receive_messages()``. + + If ``hang_after`` is set, blocks forever once the list is exhausted, + simulating a connection left open with no further messages — the + situation ``timeout_s`` in ``receive_settled_response`` is meant to + bound. + """ + + def __init__(self, messages, hang_after: bool = False): + self._messages = messages + self._hang_after = hang_after + + async def receive_messages(self): + for msg in self._messages: + yield msg + if self._hang_after: + await asyncio.Event().wait() + + +def _result(is_error: bool = False, num_turns: int = 1) -> ResultMessage: + return ResultMessage( + subtype="success", + duration_ms=1, + duration_api_ms=1, + is_error=is_error, + num_turns=num_turns, + session_id="s1", + ) + + +def _task_started(task_id: str, task_type: str = "local_agent") -> TaskStartedMessage: + return TaskStartedMessage( + subtype="task_started", + data={}, + task_id=task_id, + description="do a thing", + uuid="u1", + session_id="s1", + task_type=task_type, + ) + + +def _task_notification( + task_id: str, status: str = "completed" +) -> TaskNotificationMessage: + return TaskNotificationMessage( + subtype="task_notification", + data={}, + task_id=task_id, + status=status, + output_file="", + summary="done", + uuid="u2", + session_id="s1", + ) + + +def _task_updated(task_id: str, status: str = "completed") -> TaskUpdatedMessage: + return TaskUpdatedMessage( + subtype="task_updated", + data={}, + task_id=task_id, + patch={"status": status}, + status=status, + ) + + +async def test_receive_settled_response_returns_immediately_when_nothing_pending(): + result = _result() + client = _FakeClient([result]) + seen = [] + + got = await receive_settled_response(client, on_message=seen.append) + + assert got is result + assert seen == [result] + + +async def test_receive_settled_response_keeps_draining_past_early_result(): + started = _task_started("t1") + early_result = _result(num_turns=1) + notification = _task_notification("t1") + final_result = _result(num_turns=2) + client = _FakeClient([started, early_result, notification, final_result]) + + got = await receive_settled_response(client) + + # The first ResultMessage arrived while "t1" was still open — it must be + # ignored in favor of the one that follows the task's terminal message. + assert got is final_result + + +async def test_receive_settled_response_task_updated_also_clears_pending(): + started = _task_started("t1") + early_result = _result(num_turns=1) + updated = _task_updated("t1") + final_result = _result(num_turns=2) + client = _FakeClient([started, early_result, updated, final_result]) + + got = await receive_settled_response(client) + + assert got is final_result + + +async def test_receive_settled_response_ignores_non_deferring_task_types(): + # A backgrounded shell (task_type="local_bash") can run indefinitely by + # design — the CLI itself never holds the result frame back for one, so + # neither should we. Settling immediately (rather than waiting on "t1") + # is the correct behavior here, not a race we need to rescue. + started = _task_started("t1", task_type="local_bash") + result = _result() + client = _FakeClient([started, result]) + + got = await receive_settled_response(client) + + assert got is result + + +async def test_receive_settled_response_raises_on_timeout_when_task_never_settles(): + client = _FakeClient([_task_started("t1"), _result()], hang_after=True) + + with pytest.raises(UnsettledResponseError): + await receive_settled_response(client, timeout_s=0.05) + + +async def test_receive_settled_response_raises_when_stream_ends_without_result(): + client = _FakeClient([SystemMessage(subtype="init", data={})]) + + with pytest.raises(UnsettledResponseError): + await receive_settled_response(client) diff --git a/uv.lock b/uv.lock index 152dbb598c..a4e0436590 100644 --- a/uv.lock +++ b/uv.lock @@ -89,7 +89,7 @@ vcs = [ [package.metadata] requires-dist = [ { name = "bugsy", marker = "extra == 'bugzilla'" }, - { name = "claude-agent-sdk", marker = "extra == 'claude-sdk'", specifier = ">=0.1.30" }, + { name = "claude-agent-sdk", marker = "extra == 'claude-sdk'", specifier = ">=0.2.30" }, { name = "grizzly-framework", marker = "extra == 'firefox'" }, { name = "httpx", marker = "extra == 'vcs'" }, { name = "prefpicker", marker = "extra == 'firefox'" }, @@ -4656,9 +4656,9 @@ resolution-markers = [ "python_full_version < '3.13' and platform_machine == 'x86_64' and sys_platform == 'darwin'", ] dependencies = [ - { name = "llvmlite", version = "0.36.0", source = { registry = "https://pypi.org/simple" } }, - { name = "numpy" }, - { name = "setuptools" }, + { name = "llvmlite", version = "0.36.0", source = { registry = "https://pypi.org/simple" }, marker = "platform_machine == 'x86_64' and sys_platform == 'darwin'" }, + { name = "numpy", marker = "platform_machine == 'x86_64' and sys_platform == 'darwin'" }, + { name = "setuptools", marker = "platform_machine == 'x86_64' and sys_platform == 'darwin'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/e3/7d/3d61160836e49f40913741c464f119551c15ed371c1d91ea50308495b93b/numba-0.53.1.tar.gz", hash = "sha256:9cd4e5216acdc66c4e9dab2dfd22ddb5bef151185c070d4a3cd8e78638aff5b0", size = 2213956, upload-time = "2021-03-26T09:15:50.402Z" } @@ -4678,8 +4678,8 @@ resolution-markers = [ "(python_full_version < '3.13' and platform_machine != 'x86_64' and sys_platform == 'darwin') or (python_full_version < '3.13' and sys_platform != 'darwin' and sys_platform != 'emscripten' and sys_platform != 'win32')", ] dependencies = [ - { name = "llvmlite", version = "0.48.0", source = { registry = "https://pypi.org/simple" } }, - { name = "numpy" }, + { name = "llvmlite", version = "0.48.0", source = { registry = "https://pypi.org/simple" }, marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, + { name = "numpy", marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/ae/a0/570e3dc53e5602b49108f62a13e529f1eec8bfc7ef37d49c825924dcf546/numba-0.66.0.tar.gz", hash = "sha256:b900e63a0e26c05ea9a6d5a3a5a0a177cb64c5011887bf43edb8c3ed2c38d363", size = 2806181, upload-time = "2026-07-01T23:12:46.36Z" } wheels = [ @@ -5109,7 +5109,7 @@ name = "pexpect" version = "4.9.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "ptyprocess" }, + { name = "ptyprocess", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/42/92/cc564bf6381ff43ce1f4d06852fc19a2f11d180f23dc32d9588bee2f149d/pexpect-4.9.0.tar.gz", hash = "sha256:ee7d41123f3c9911050ea2c2dac107568dc43b2d3b0c7557a33212c398ead30f", size = 166450, upload-time = "2023-11-25T09:07:26.339Z" } wheels = [ @@ -6697,8 +6697,8 @@ name = "secretstorage" version = "3.5.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "cryptography" }, - { name = "jeepney" }, + { name = "cryptography", marker = "(platform_machine != 'x86_64' and sys_platform == 'darwin') or (sys_platform != 'darwin' and sys_platform != 'emscripten' and sys_platform != 'win32')" }, + { name = "jeepney", marker = "(platform_machine != 'x86_64' and sys_platform == 'darwin') or (sys_platform != 'darwin' and sys_platform != 'emscripten' and sys_platform != 'win32')" }, ] sdist = { url = "https://files.pythonhosted.org/packages/1c/03/e834bcd866f2f8a49a85eaff47340affa3bfa391ee9912a952a1faa68c7b/secretstorage-3.5.0.tar.gz", hash = "sha256:f04b8e4689cbce351744d5537bf6b1329c6fc68f91fa666f60a380edddcd11be", size = 19884, upload-time = "2025-11-23T19:02:53.191Z" } wheels = [