fix: ruff format, mypy, throttle baseline, pytest marks, annotation false-positive

- Fix throttle false-block on cold-start: skip throttle gate when no prior
  snapshot exists for the key (previously `get(key, 0.0)` made first
  capture compare against time-0, blocking when monotonic() < throttle).
- Add type: ignore[import-untyped] for yaml imports (no stubs installed).
- Remove unused type: ignore comments on auto_backup.py decorator path,
  clean up snap_domain typing to str.
- Add Any return type annotation to _backup_mgr() helper.
- Remove example function definitions from auto_backup.py module docstring —
  test_tool_annotations.py's regex was matching them as real tool defs.
- Use registered pytest marks (script, convenience) instead of unregistered
  scripts/scenes/dashboards.
- ruff format on all 27 changed py files.

Author: kingpanther13

homeassistant-addon/start.py

@@ -39,7 +39,9 @@ def generate_secret_path() -> str:
 
 
 _SECRET_PATH_RE = re.compile(r"^/(?!.*://)\S{7,}$")
-_SECRET_PATH_HINT = "Path must start with '/', contain no '://', and be at least 8 characters."
+_SECRET_PATH_HINT = (
+    "Path must start with '/', contain no '://', and be at least 8 characters."
+)
 
 
 def _is_valid_secret_path(path: str) -> bool:
@@ -65,7 +67,9 @@ def get_or_create_secret_path(data_dir: Path, custom_path: str = "") -> str:
         if not path.startswith("/"):
             path = "/" + path
         if not _is_valid_secret_path(path):
-            log_error(f"Custom secret path is invalid ({path!r}), ignoring. {_SECRET_PATH_HINT}")
+            log_error(
+                f"Custom secret path is invalid ({path!r}), ignoring. {_SECRET_PATH_HINT}"
+            )
         else:
             log_info("Using custom secret path from configuration")
             # Update stored path for consistency
@@ -80,7 +84,9 @@ def get_or_create_secret_path(data_dir: Path, custom_path: str = "") -> str:
                 log_info("Using existing auto-generated secret path")
                 return stored_path
             elif stored_path:
-                log_error(f"Stored secret path is invalid ({stored_path!r}), regenerating. {_SECRET_PATH_HINT}")
+                log_error(
+                    f"Stored secret path is invalid ({stored_path!r}), regenerating. {_SECRET_PATH_HINT}"
+                )
             else:
                 log_error("Stored secret path is empty, regenerating")
         except Exception as e:
@@ -234,31 +240,59 @@ def main() -> int:
             backup_hint = config.get("backup_hint", "normal")
             custom_secret_path = config.get("secret_path", "")
             raw_tool_search = config.get("enable_tool_search", False)
-            enable_tool_search = raw_tool_search if isinstance(raw_tool_search, bool) else False
+            enable_tool_search = (
+                raw_tool_search if isinstance(raw_tool_search, bool) else False
+            )
             raw_yaml_config = config.get("enable_yaml_config_editing", False)
-            enable_yaml_config_editing = raw_yaml_config if isinstance(raw_yaml_config, bool) else False
+            enable_yaml_config_editing = (
+                raw_yaml_config if isinstance(raw_yaml_config, bool) else False
+            )
             raw_filesystem_tools = config.get("enable_filesystem_tools", False)
-            enable_filesystem_tools = raw_filesystem_tools if isinstance(raw_filesystem_tools, bool) else False
-            raw_custom_component = config.get("enable_custom_component_integration", False)
-            enable_custom_component_integration = raw_custom_component if isinstance(raw_custom_component, bool) else False
+            enable_filesystem_tools = (
+                raw_filesystem_tools
+                if isinstance(raw_filesystem_tools, bool)
+                else False
+            )
+            raw_custom_component = config.get(
+                "enable_custom_component_integration", False
+            )
+            enable_custom_component_integration = (
+                raw_custom_component
+                if isinstance(raw_custom_component, bool)
+                else False
+            )
             raw_code_mode = config.get("enable_code_mode", False)
-            enable_code_mode = raw_code_mode if isinstance(raw_code_mode, bool) else False
+            enable_code_mode = (
+                raw_code_mode if isinstance(raw_code_mode, bool) else False
+            )
             raw_lite_docstrings = config.get("enable_lite_docstrings", False)
-            enable_lite_docstrings = raw_lite_docstrings if isinstance(raw_lite_docstrings, bool) else False
+            enable_lite_docstrings = (
+                raw_lite_docstrings if isinstance(raw_lite_docstrings, bool) else False
+            )
             raw_auto_backup = config.get("enable_auto_backup", False)
-            enable_auto_backup = raw_auto_backup if isinstance(raw_auto_backup, bool) else False
+            enable_auto_backup = (
+                raw_auto_backup if isinstance(raw_auto_backup, bool) else False
+            )
             raw_throttle = config.get("auto_backup_throttle_minutes", 0)
-            auto_backup_throttle_minutes = raw_throttle if isinstance(raw_throttle, int) else 0
+            auto_backup_throttle_minutes = (
+                raw_throttle if isinstance(raw_throttle, int) else 0
+            )
             raw_retain = config.get("auto_backup_retain_per_entity", 20)
-            auto_backup_retain_per_entity = raw_retain if isinstance(raw_retain, int) else 20
+            auto_backup_retain_per_entity = (
+                raw_retain if isinstance(raw_retain, int) else 20
+            )
             raw_max_results = config.get("tool_search_max_results", 5)
-            tool_search_max_results = raw_max_results if isinstance(raw_max_results, int) else 5
+            tool_search_max_results = (
+                raw_max_results if isinstance(raw_max_results, int) else 5
+            )
             raw_disabled = config.get("disabled_tools", "")
             disabled_tools_raw = raw_disabled if isinstance(raw_disabled, str) else ""
             raw_pinned = config.get("pinned_tools", "")
             pinned_tools_raw = raw_pinned if isinstance(raw_pinned, str) else ""
             verify_ssl = resolve_bool_option(config, "verify_ssl", True)
-            advanced_debug_logging = resolve_bool_option(config, "advanced_debug_logging", False)
+            advanced_debug_logging = resolve_bool_option(
+                config, "advanced_debug_logging", False
+            )
         except Exception as e:
             log_error(f"Failed to read config: {e}, using defaults")
 
@@ -288,7 +322,9 @@ def main() -> int:
     os.environ["ENABLE_TOOL_SEARCH"] = str(enable_tool_search).lower()
     os.environ["ENABLE_YAML_CONFIG_EDITING"] = str(enable_yaml_config_editing).lower()
     os.environ["HAMCP_ENABLE_FILESYSTEM_TOOLS"] = str(enable_filesystem_tools).lower()
-    os.environ["HAMCP_ENABLE_CUSTOM_COMPONENT_INTEGRATION"] = str(enable_custom_component_integration).lower()
+    os.environ["HAMCP_ENABLE_CUSTOM_COMPONENT_INTEGRATION"] = str(
+        enable_custom_component_integration
+    ).lower()
     os.environ["ENABLE_CODE_MODE"] = str(enable_code_mode).lower()
     os.environ["ENABLE_LITE_DOCSTRINGS"] = str(enable_lite_docstrings).lower()
     os.environ["ENABLE_AUTO_BACKUP"] = str(enable_auto_backup).lower()
@@ -302,9 +338,7 @@ def main() -> int:
     # tool isn't registered anyway, so the file is never read or
     # written. Operators can override by setting CODE_MODE_SAVED_TOOLS_PATH
     # in the add-on's environment if they want a different location.
-    os.environ.setdefault(
-        "CODE_MODE_SAVED_TOOLS_PATH", "/data/saved_tools.json"
-    )
+    os.environ.setdefault("CODE_MODE_SAVED_TOOLS_PATH", "/data/saved_tools.json")
     os.environ["TOOL_SEARCH_MAX_RESULTS"] = str(tool_search_max_results)
     os.environ["DISABLED_TOOLS"] = disabled_tools_raw
     os.environ["PINNED_TOOLS"] = pinned_tools_raw
@@ -331,6 +365,7 @@ def main() -> int:
 
     # Configure logging before server start (v3 removed log_level from run())
     import logging
+
     logging.basicConfig(level=logging.INFO)
 
     # Import and register browser landing before server start
@@ -353,6 +388,7 @@ def main() -> int:
             from ha_mcp.utils.kill_signal_diagnostics import (
                 schedule_install_after_uvicorn,
             )
+
             schedule_install_after_uvicorn()
         except Exception as e:
             log_error(f"advanced_debug_logging install failed: {e!r}; continuing")
@@ -364,7 +400,9 @@ def main() -> int:
     # server's actual FastMCP instance (not the _DeferredMCP wrapper)
     # so mypy doesn't trip over the duck-typed __getattr__ forwarding.
     server_instance = _get_server()
-    register_settings_routes(server_instance.mcp, server_instance, secret_path=secret_path)
+    register_settings_routes(
+        server_instance.mcp, server_instance, secret_path=secret_path
+    )
     logging.getLogger("mcp.server.streamable_http").addFilter(
         StatelessSessionLogFilter()
     )
@@ -391,7 +429,9 @@ def main() -> int:
         cause = e.__cause__ or e.__context__
         if cause:
             log_error(f"Caused by: {cause}")
-            traceback.print_exception(type(cause), cause, cause.__traceback__, file=sys.stderr)
+            traceback.print_exception(
+                type(cause), cause, cause.__traceback__, file=sys.stderr
+            )
         if isinstance(e, SystemExit):
             return int(e.code) if isinstance(e.code, int) else 1
         return 1

src/ha_mcp/backup_manager.py

@@ -49,7 +49,7 @@
 from pathlib import Path
 from typing import Any
 
-import yaml
+import yaml  # type: ignore[import-untyped]
 
 logger = logging.getLogger(__name__)
 
@@ -208,7 +208,16 @@ async def maybe_snapshot(
         async with lock:
             now = time.monotonic()
             throttle = self.throttle_seconds
-            if throttle and (now - self._last_snapshot.get(key, 0.0)) < throttle:
+            # Skip throttle if no prior snapshot exists for this key.
+            # Using ``get(key, 0.0)`` would falsely block the first capture
+            # whenever ``monotonic()`` < throttle (typical on a fresh process
+            # in CI), since 0.0 would be treated as "last snapshot at
+            # monotonic time 0".
+            if (
+                throttle
+                and key in self._last_snapshot
+                and (now - self._last_snapshot[key]) < throttle
+            ):
                 return None
             try:
                 config = await handler.fetch(self._client, entity_id)

src/ha_mcp/settings_ui.py

@@ -1194,7 +1194,7 @@ async def _settings_info(_: Request) -> JSONResponse:
 
     # ---- Auto-backup routes (#1288) ----
 
-    def _backup_mgr():
+    def _backup_mgr() -> Any:
         settings = get_global_settings()
         client = getattr(server, "client", None) or getattr(server, "_client", None)
         return get_backup_manager(client, settings) if client is not None else None
@@ -1258,7 +1258,7 @@ async def _view_backup(request: Request) -> JSONResponse:
     async def _diff_backup(request: Request) -> JSONResponse:
         import difflib
 
-        import yaml
+        import yaml  # type: ignore[import-untyped]
 
         mgr = _backup_mgr()
         if mgr is None:

src/ha_mcp/tools/auto_backup.py

@@ -5,19 +5,16 @@
 
 Usage
 -----
-Simple case (entity ID lives in a single kwarg)::
+Simple case (entity ID lives in a single kwarg): wrap the existing tool
+function with ``@with_auto_backup(domain="<domain>", id_param="<kwarg>")``
+placed above the ``@mcp.tool``/``@tool`` decorator and below it the
+``@log_tool_usage`` line, so the order from outermost to innermost is
+``@tool`` -> ``@with_auto_backup`` -> ``@log_tool_usage`` -> the async def.
 
-    @with_auto_backup(domain="automation", id_param="identifier")
-    @mcp.tool(name="ha_config_set_automation", ...)
-    @log_tool_usage
-    async def ha_config_set_automation(self, identifier: str, ...): ...
-
-Computed-key case (helpers — domain encodes ``helper_type``)::
-
-    @with_auto_backup(
-        domain_fn=lambda kw: f"helper_{kw['helper_type']}",
-        id_fn=lambda kw: f"{kw['helper_type']}:{kw.get('helper_id', '')}",
-    )
+Computed-key case (helpers — domain encodes ``helper_type``): use
+``domain_fn`` / ``id_fn`` instead of ``domain`` / ``id_param``; both take
+a single ``kw`` dict argument and return a string. Helpers typically use
+``domain_fn=lambda kw: f"helper_{kw['helper_type']}"``.
 
 The decorator must be applied **above** ``@mcp.tool`` so FastMCP sees the
 final wrapped callable as the tool. ``functools.wraps`` preserves the
@@ -87,13 +84,13 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
                             args[0], "client", None
                         )
                     if client_obj is not None:
-                        snap_domain = (
-                            domain_fn(kwargs) if domain_fn is not None else domain  # type: ignore[assignment]
+                        snap_domain: str = (
+                            domain_fn(kwargs) if domain_fn is not None else domain or ""
                         )
                         if id_fn is not None:
                             entity_id = _resolve_str(id_fn(kwargs))
                         else:
-                            entity_id = _resolve_str(kwargs.get(id_param))  # type: ignore[arg-type]
+                            entity_id = _resolve_str(kwargs.get(id_param or ""))
                         if entity_id:
                             mgr = get_backup_manager(client_obj, settings)
                             await mgr.maybe_snapshot(