Merge pull request #457 from Integration-Automation/fix/crash-and-interruption-hardening

Prevent crashes and silent automation interruptions

Author: JE-Chen

je_auto_control/utils/chatops/router.py

@@ -144,8 +144,8 @@ def _dispatch_argv(self, argv: List[str],
             return spec.handler(rest, context)
         except ChatOpsError as error:
             return CommandResult(text=f"{name}: {error}", succeeded=False)
-        except (RuntimeError, OSError, ValueError, TypeError,
-                AutoControlException, sqlite3.Error) as error:
+        except (RuntimeError, OSError, ValueError, TypeError, LookupError,
+                AttributeError, AutoControlException, sqlite3.Error) as error:
             return CommandResult(
                 text=f"{name} failed: {type(error).__name__}: {error}",
                 succeeded=False,

je_auto_control/utils/hotkey/hotkey_daemon.py

@@ -16,6 +16,7 @@
 from dataclasses import dataclass
 from typing import Callable, Dict, FrozenSet, List, Optional, Tuple
 
+from je_auto_control.utils.exception.exceptions import AutoControlException
 from je_auto_control.utils.json.json_file import read_action_json
 from je_auto_control.utils.logging.logging_instance import autocontrol_logger
 from je_auto_control.utils.run_history.artifact_manager import (
@@ -186,7 +187,11 @@ def _fire_binding(self, binding_id: str) -> None:
         try:
             actions = read_action_json(match.script_path)
             self._execute(actions)
-        except (OSError, ValueError, RuntimeError) as error:
+        except (OSError, ValueError, RuntimeError, AutoControlException) as error:
+            # AutoControlException covers the common cases — a missing/renamed
+            # script (AutoControlJsonActionException) or an action that raises
+            # (image/window not found). Without it the exception escaped the
+            # backend run-loop and silently killed the whole hotkey daemon.
             status = STATUS_ERROR
             error_text = repr(error)
             autocontrol_logger.error("hotkey %s failed: %r",

je_auto_control/utils/observer/observer.py

@@ -26,7 +26,11 @@
 _ALL_EVENTS = (EVENT_APPEAR, EVENT_VANISH, EVENT_CHANGE)
 
 # Errors a predicate/handler may raise that must not kill the poll loop.
+# LookupError/StopIteration/ArithmeticError cover user callbacks that index a
+# dict/list, exhaust an iterator, or divide — an uncaught one kills the daemon
+# thread and silently stops every rule.
 _RULE_ERRORS = (OSError, RuntimeError, ValueError, AttributeError, TypeError,
+                LookupError, StopIteration, ArithmeticError,
                 AutoControlException)
 _UNSET = object()
 

je_auto_control/utils/remote_desktop/host_service.py

@@ -135,13 +135,13 @@ def run_daemon(config: HostServiceConfig) -> None:
                 session_id, multi.session_count(),
             )
             time.sleep(config.poll_interval_s)
-        except (signaling_client.SignalingError, OSError, RuntimeError) as error:
-            autocontrol_logger.warning("host_service loop: %r", error)
-            time.sleep(min(30.0, config.poll_interval_s * 5))
         except KeyboardInterrupt:
             autocontrol_logger.info("host_service: shutting down")
             multi.stop_all()
             return
+        except Exception as error:  # noqa: BLE001  # reason: a daemon must survive ANY transient error (signaling/aiortc/av/ValueError) and retry, not exit the loop
+            autocontrol_logger.warning("host_service loop: %r", error)
+            time.sleep(min(30.0, config.poll_interval_s * 5))
 
 
 # --- service installation helpers ----------------------------------------

je_auto_control/utils/remote_desktop/webrtc_host.py

@@ -24,9 +24,7 @@
 from je_auto_control.utils.remote_desktop.fingerprint import (
     load_or_create_host_fingerprint,
 )
-from je_auto_control.utils.remote_desktop.input_dispatch import (
-    InputDispatchError, dispatch_input,
-)
+from je_auto_control.utils.remote_desktop.input_dispatch import dispatch_input
 from je_auto_control.utils.remote_desktop.permissions import SessionPermissions
 from je_auto_control.utils.remote_desktop.rate_limit import (
     RateLimitConfig, RateLimiter,
@@ -976,7 +974,7 @@ def _dispatch_input_safely(self, payload: Any) -> None:
             return
         try:
             self._dispatch(payload)
-        except InputDispatchError as error:
+        except Exception as error:  # noqa: BLE001  # reason: isolation boundary — a malformed/failing remote input must not kill the channel bridge (dispatch can raise AutoControl*/OSError, not just InputDispatchError)
             autocontrol_logger.warning("input dispatch: %r", error)
 
     def _send_ctrl(self, payload: Mapping[str, Any]) -> None:

je_auto_control/utils/rest_api/rest_server.py

@@ -93,6 +93,10 @@ class _RestRequestHandler(BaseHTTPRequestHandler):
     """Stdlib request handler — delegates to gate + route table."""
 
     server_version = "AutoControlREST/2.0"
+    # socketserver applies this to the connection socket in setup(); it bounds
+    # every read (the body is read before the auth gate) so a client that
+    # declares a Content-Length then stalls cannot pin a worker thread forever.
+    timeout = 30.0
 
     def log_message(self, format, *args) -> None:  # noqa: A002  # pylint: disable=redefined-builtin  # reason: stdlib BaseHTTPRequestHandler override
         autocontrol_logger.info("rest-api %s - %s",

je_auto_control/utils/socket_server/auto_control_socket_server.py

@@ -56,10 +56,20 @@ def _close() -> None:
                      daemon=True).start()
 
 
+_HANDLER_TIMEOUT_S = 30.0
+
+
 class TCPServerHandler(socketserver.BaseRequestHandler):
 
     def handle(self) -> None:
-        command_string = _read_command(self.request)
+        try:
+            self.request.settimeout(_HANDLER_TIMEOUT_S)
+            command_string = _read_command(self.request)
+        except OSError as error:
+            # A client that connects and never sends a terminator would
+            # otherwise block this handler thread forever; the timeout drops it.
+            autocontrol_logger.info("socket command read dropped: %r", error)
+            return
         socket = self.request
         autocontrol_logger.info("command is: %s", command_string)
         if command_string == "quit_server":
@@ -93,12 +103,18 @@ def handle(self) -> None:
 class TCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
     """Threaded TCP command server for AutoControl.
 
+    ``daemon_threads`` so a stalled handler thread never blocks interpreter
+    exit (and, with the per-handler read timeout, stalled clients are dropped
+    rather than accumulating non-daemon threads).
+
     ``close_flag`` used to live here. It was written on quit_server and read
     by nobody in the tree — a dead flag standing in for the ``server_close()``
     that was actually missing. Ask the socket instead: ``server.socket``
     is closed once quit_server has run.
     """
 
+    daemon_threads = True
+
 
 def start_autocontrol_socket_server(host: str = "127.0.0.1", port: int = 9938) -> TCPServer:
     """

je_auto_control/utils/triggers/webhook_server.py

@@ -50,6 +50,10 @@
 
 _DEFAULT_BIND = "127.0.0.1"
 _MAX_BODY_BYTES = 1 << 20  # 1 MiB cap
+# Automation is serialised (one script drives the shared input devices at a
+# time). Bound how long a concurrent webhook waits for the running script so a
+# long/stuck one can't pile up handler threads indefinitely — reject as busy.
+_FIRE_LOCK_TIMEOUT_S = 120.0
 # Cap how much we'll drain from a rejected request so a hostile client
 # can't make us spin reading a multi-GiB body. 4× the body cap covers
 # typical "client sent slightly too much" cases; beyond that we close
@@ -106,6 +110,9 @@ class _WebhookHandler(BaseHTTPRequestHandler):
     """HTTP handler dispatched by :class:`WebhookTriggerServer`."""
 
     server_version = "AutoControlWebhook/1.0"
+    # Bound every read so a stalled client (valid Content-Length then dribble)
+    # can't pin a worker thread forever.
+    timeout = 30.0
 
     # Signature must mirror BaseHTTPRequestHandler.log_message exactly,
     # including the parameter name 'format' — pylint W0221 trips on
@@ -196,6 +203,12 @@ def _dispatch(self, method: str) -> None:
             ),
         }
         run_id = registry.fire(trigger, payload)
+        if run_id is None:
+            self._send_json(
+                HTTPStatus.SERVICE_UNAVAILABLE,
+                {"fired": False, "error": "automation busy, retry later"},
+            )
+            return
         self._send_json(HTTPStatus.OK, {"run_id": run_id, "fired": True})
 
     def do_GET(self) -> None:  # noqa: N802 - http.server contract
@@ -303,8 +316,19 @@ def authorize(self, trigger: WebhookTrigger,
 
     def fire(self, trigger: WebhookTrigger,
              payload: Dict[str, Any]) -> Optional[int]:
-        """Run the trigger's script with ``payload`` seeded into variables."""
-        with self._fire_lock:
+        """Run the trigger's script with ``payload`` seeded into variables.
+
+        Returns the run id, or ``None`` if the automation stayed busy past
+        :data:`_FIRE_LOCK_TIMEOUT_S` (the caller answers 503 rather than
+        blocking this handler thread forever).
+        """
+        if not self._fire_lock.acquire(timeout=_FIRE_LOCK_TIMEOUT_S):
+            autocontrol_logger.warning(
+                "webhook %s rejected: automation busy for >%.0fs",
+                trigger.webhook_id, _FIRE_LOCK_TIMEOUT_S,
+            )
+            return None
+        try:
             run_id = default_history_store.start_run(
                 SOURCE_TRIGGER, f"webhook:{trigger.webhook_id}",
                 trigger.script_path,
@@ -337,6 +361,8 @@ def fire(self, trigger: WebhookTrigger,
                     live.fired += 1
                     live.last_status = 200 if status == STATUS_OK else 500
             return run_id
+        finally:
+            self._fire_lock.release()
 
     def start(self, host: str = _DEFAULT_BIND, port: int = 0) -> Tuple[str, int]:
         """Start the HTTP server; idempotent if already running."""

je_auto_control/utils/usbip/server.py

@@ -116,6 +116,10 @@ def _accept_loop(self) -> None:
                 target=self._handle_client, args=(client_sock,),
                 name="usbip-client", daemon=True,
             )
+            # Drop finished workers so the list doesn't grow without bound over
+            # a long session with many short-lived connections (each dead Thread
+            # object would otherwise be retained until stop()).
+            self._workers[:] = [w for w in self._workers if w.is_alive()]
             self._workers.append(worker)
             worker.start()
 

je_auto_control/utils/visual_match/visual_match.py

@@ -12,14 +12,36 @@
 (grab the screen) is device-bound. OpenCV + NumPy come in via the project's
 ``je_open_cv`` dependency and are imported lazily. Imports no ``PySide6``.
 """
+import functools
 from dataclasses import asdict, dataclass
 from typing import Any, Dict, List, Optional, Sequence
 
+from je_auto_control.utils.exception.exceptions import AutoControlScreenException
+
 # cv2 method name -> the OpenCV constant is resolved lazily in _method().
 _METHOD_NAMES = ("ccoeff_normed", "ccorr_normed", "sqdiff_normed")
 ImageSource = Any
 
 
+def _contain_cv2_error(fn):
+    """Convert OpenCV's ``cv2.error`` into a contained AutoControlScreenException.
+
+    A degenerate template/mask makes ``cv2.matchTemplate``/``minMaxLoc`` raise
+    ``cv2.error`` — a bare ``Exception`` subclass that is NOT in the executor's
+    containment tuple, so it would escape and abort the whole automation run
+    instead of being recorded as a failed match step.
+    """
+    @functools.wraps(fn)
+    def wrapper(*args, **kwargs):
+        import cv2
+        try:
+            return fn(*args, **kwargs)
+        except cv2.error as error:
+            raise AutoControlScreenException(
+                f"{fn.__name__} failed: {error}") from error
+    return wrapper
+
+
 @dataclass(frozen=True)
 class Match:
     """One template match: top-left (x, y), size, correlation score, scale."""
@@ -107,6 +129,7 @@ def _resize(template, scale: float):
     return cv2.resize(template, new_size)
 
 
+@_contain_cv2_error
 def _score_map(template: ImageSource, haystack: Optional[ImageSource] = None, *,
                region: Optional[Sequence[int]] = None,
                method: str = "ccoeff_normed", scale: float = 1.0):
@@ -128,6 +151,7 @@ def _score_map(template: ImageSource, haystack: Optional[ImageSource] = None, *,
     return result, tmpl
 
 
+@_contain_cv2_error
 def match_template(template: ImageSource, *, haystack: Optional[ImageSource] = None,
                    region: Optional[Sequence[int]] = None,
                    scales: Sequence[float] = (1.0,), min_score: float = 0.8,
@@ -201,6 +225,7 @@ def _select_candidates(result, min_score: float, width: int, height: int,
             for x, y, s in zip(xs, ys, scores)]
 
 
+@_contain_cv2_error
 def match_template_all(template: ImageSource, *,
                        haystack: Optional[ImageSource] = None,
                        region: Optional[Sequence[int]] = None,
@@ -285,6 +310,7 @@ def _masked_scores(template: ImageSource, mask: Optional[ImageSource],
     return np.nan_to_num(result, nan=0.0, posinf=0.0, neginf=0.0), tmpl
 
 
+@_contain_cv2_error
 def match_masked(template: ImageSource, *, mask: Optional[ImageSource] = None,
                  haystack: Optional[ImageSource] = None,
                  region: Optional[Sequence[int]] = None,
@@ -308,6 +334,7 @@ def match_masked(template: ImageSource, *, mask: Optional[ImageSource] = None,
                  round(float(max_val), 4), 1.0)
 
 
+@_contain_cv2_error
 def match_masked_all(template: ImageSource, *, mask: Optional[ImageSource] = None,
                      haystack: Optional[ImageSource] = None,
                      region: Optional[Sequence[int]] = None,