agent: Control listener thread dies on an unexpected handler exception

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)

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

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)