From 6bf246ce39270a3e99a8d92ea71afced69a1efeb Mon Sep 17 00:00:00 2001 From: Claude Code Date: Wed, 14 Oct 2026 11:47:19 +0000 Subject: [PATCH] Harden agentctl listener against unexpected handler failures One failing request no longer stops the agentctl-listener thread: unexpected exceptions are logged as control_request_failed events and answered with {ok: false} on the same connection. send_request maps oversized or unparsable replies to ControlUnavailableError so agentctl exits 2. Co-Authored-By: Claude Code --- simple_coding_agent/__main__.py | 8 +- simple_coding_agent/control_server.py | 123 +++++++++++++++++++------- tests/test_control_socket.py | 87 ++++++++++++++++++ 3 files changed, 183 insertions(+), 35 deletions(-) diff --git a/simple_coding_agent/__main__.py b/simple_coding_agent/__main__.py index 05039ba..90d473d 100644 --- a/simple_coding_agent/__main__.py +++ b/simple_coding_agent/__main__.py @@ -137,7 +137,13 @@ def _attempt_sink(issue_number: int) -> AttemptArchive | None: issue_number_provider=lambda: lifecycle.active_issue_number, ) restore_interruption_handler = interruption_handler.install() - control_server = ControlServer(socket_path_for(config.data_dir), lifecycle) + control_server = ControlServer( + socket_path_for(config.data_dir), + lifecycle, + event_log=lambda event, detail="", level="INFO", issue_number=None: logger.emit( + event, phase="control", detail=detail, level=level, issue_number=issue_number + ), + ) _start_control_endpoint(logger, control_server) try: lifecycle.run_forever(stop=lambda: False) diff --git a/simple_coding_agent/control_server.py b/simple_coding_agent/control_server.py index cc56e06..ee53dd9 100644 --- a/simple_coding_agent/control_server.py +++ b/simple_coding_agent/control_server.py @@ -71,9 +71,16 @@ class ControlUnavailableError(ConnectionError): class ControlServer: """Serve one agent instance's control surface on a Unix domain socket.""" - def __init__(self, socket_path: Path, control: ControlSurface) -> None: + def __init__( + self, + socket_path: Path, + control: ControlSurface, + *, + event_log: Callable[..., None] | None = None, + ) -> None: self._socket_path = socket_path self._control = control + self._event_log = event_log self._listener: socket.socket | None = None self._thread: threading.Thread | None = None self._stopped = threading.Event() @@ -140,6 +147,38 @@ def stop(self) -> None: except OSError: pass + def _log_control_error(self, detail: str) -> None: + """Record one failed control request as a structured event, if wired.""" + + event_log = self._event_log + if event_log is None: + return + try: + event_log("control_request_failed", detail=detail, level="ERROR") + except Exception: + pass + + def _safe_repository(self) -> str: + try: + return self._repository() + except Exception: + return "unknown" + + def _send_failure(self, connection: socket.socket, error: object) -> None: + """Answer one failed request without ever raising out of the listener.""" + + try: + _send( + connection, + { + "ok": False, + "repository": self._safe_repository(), + "error": f"Control request failed: {error}", + }, + ) + except Exception: + pass + def _serve(self) -> None: listener = self._listener if listener is None: @@ -153,7 +192,14 @@ def _serve(self) -> None: except OSError: return try: - self._handle(connection) + try: + self._handle(connection) + except Exception as error: + # One failing request must never stop the listener: log + # the failure and answer on the same connection when it + # is still writable, then accept the next connection. + self._log_control_error(f"{type(error).__name__}: {error}") + self._send_failure(connection, error) finally: try: connection.close() @@ -170,7 +216,7 @@ def _handle(self, connection: socket.socket) -> None: connection, { "ok": False, - "repository": self._repository(), + "repository": self._safe_repository(), "error": "Request must be one JSON object per line.", }, ) @@ -180,41 +226,45 @@ def _handle(self, connection: socket.socket) -> None: connection, { "ok": False, - "repository": self._repository(), + "repository": self._safe_repository(), "error": "Request must be a JSON object.", }, ) return operation = request.get("op") - if operation == "stop": - self._reply_stop(connection, request) - elif operation == "stop_after": - self._reply_stop_after(connection, request) - elif operation == "next_issue": - self._reply_next_issue(connection, request) - elif operation == "resume": - self._reply_resume(connection, request) - elif operation == "handoff": - self._reply_handoff(connection, request) - elif operation == "recovery_retry": - self._reply_recovery_retry(connection, request) - elif operation == "recovery_release": - self._reply_recovery_release(connection, request) - elif operation == "errors_reset": - self._reply_errors_reset(connection, request) - elif operation == "status": - self._reply_status(connection) - elif operation == "command": - self._reply_command(connection, request) - else: - _send( - connection, - { - "ok": False, - "repository": self._repository(), - "error": f"Unknown operation: {operation!r}.", - }, - ) + try: + if operation == "stop": + self._reply_stop(connection, request) + elif operation == "stop_after": + self._reply_stop_after(connection, request) + elif operation == "next_issue": + self._reply_next_issue(connection, request) + elif operation == "resume": + self._reply_resume(connection, request) + elif operation == "handoff": + self._reply_handoff(connection, request) + elif operation == "recovery_retry": + self._reply_recovery_retry(connection, request) + elif operation == "recovery_release": + self._reply_recovery_release(connection, request) + elif operation == "errors_reset": + self._reply_errors_reset(connection, request) + elif operation == "status": + self._reply_status(connection) + elif operation == "command": + self._reply_command(connection, request) + else: + _send( + connection, + { + "ok": False, + "repository": self._safe_repository(), + "error": f"Unknown operation: {operation!r}.", + }, + ) + except Exception as error: + self._log_control_error(f"{operation}: {type(error).__name__}: {error}") + self._send_failure(connection, error) def _reply_stop(self, connection: socket.socket, request: dict) -> None: from simple_coding_agent.control import PayloadMismatchError, RequestIdError @@ -491,13 +541,18 @@ def send_request(socket_path: Path, request: dict) -> dict: f"Cannot connect to the agent process at {socket_path}: {error}. " "The agent may not be running in this instance." ) from error + encoded = (json.dumps(request) + "\n").encode() try: - client.sendall((json.dumps(request) + "\n").encode()) + client.sendall(encoded) raw = _read_line(client) except OSError as error: raise ControlUnavailableError( f"Lost connection to the agent process at {socket_path}: {error}" ) from error + except ValueError as error: + raise ControlUnavailableError( + f"Agent process at {socket_path} returned an unreadable reply." + ) from error finally: try: client.close() diff --git a/tests/test_control_socket.py b/tests/test_control_socket.py index 8752f7f..67a46c2 100644 --- a/tests/test_control_socket.py +++ b/tests/test_control_socket.py @@ -307,3 +307,90 @@ def test_failed_resume_keeps_its_request_id_for_retry( err = capsys.readouterr().err assert "req-retry-r" in err assert "--request-id req-retry-r" in err + + +# --------------------------------------------------------------------------- +# listener resilience (unexpected handler failures never stop the listener) +# --------------------------------------------------------------------------- + + +class _FailingControl: + """Control surface whose handlers raise outside the expected categories.""" + + def __init__(self) -> None: + self.events: list[tuple[str, str, str]] = [] + self.repository = "owner/repo" + + def _log(self, event: str, detail: str = "", level: str = "INFO") -> None: + self.events.append((event, detail, level)) + + def control_status(self, repository: str | None = None): + raise KeyError("boom-checkpoint") + + def get_command(self, request_id: str): + raise TypeError("boom-lookup") + + def submit_stop(self, request_id: str, *, has_active_attempt: bool = False): + raise KeyError("boom-submit") + + +def test_unexpected_status_failure_replies_false_and_keeps_serving( + tmp_path: Path, +) -> None: + from simple_coding_agent.control_server import ControlServer, send_request + + control = _FailingControl() + socket_path = tmp_path / "state" / "agentctl.sock" + server = ControlServer(socket_path, control, event_log=control._log) # noqa: SLF001 + server.start() + try: + reply = send_request(socket_path, {"op": "status"}) + assert reply["ok"] is False + assert "boom-checkpoint" in reply["error"] + + assert server._thread is not None and server._thread.is_alive() # noqa: SLF001 + + second = send_request(socket_path, {"op": "stop", "request_id": "req-1"}) + assert second["ok"] is False + assert "boom-submit" in second["error"] + + assert server._thread is not None and server._thread.is_alive() # noqa: SLF001 + assert [event for event, _, _ in control.events] == [ + "control_request_failed", + "control_request_failed", + ] + assert all(level == "ERROR" for _, _, level in control.events) + finally: + server.stop() + + +def test_oversized_reply_surfaces_as_a_control_error(tmp_path: Path) -> None: + import threading + + from simple_coding_agent.control_server import send_request + + socket_path = tmp_path / "state" / "agentctl.sock" + socket_path.parent.mkdir(parents=True, exist_ok=True) + listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + listener.bind(str(socket_path)) + listener.listen(1) + + def serve_once() -> None: + connection, _ = listener.accept() + with connection: + data = b"" + while not data.endswith(b"\n"): + chunk = connection.recv(65536) + if not chunk: + break + data += chunk + connection.sendall(b'{"ok": true, "pad": "' + b"x" * (2**20) + b'"}\n') + + thread = threading.Thread(target=serve_once, daemon=True) + thread.start() + try: + with pytest.raises(ControlUnavailableError): + send_request(socket_path, {"op": "status"}) + finally: + listener.close() + thread.join(timeout=10)