refactor(imports): consolidate hotspot fallback helpers

api/presets.py

@@ -8,29 +8,41 @@ import os
 import time
 from typing import Optional
 
-try:
-    from ..services.access_control import require_admin_token, resolve_token_info
-    from ..services.aiohttp_compat import import_aiohttp_web
-    from ..services.endpoint_manifest import (
-        AuthTier,
-        RiskTier,
-        RoutePlane,
-        endpoint_metadata,
-    )
-    from ..services.presets import Preset, preset_store
-    from ..services.tenant_context import TenantBoundaryError, request_tenant_scope
-except ImportError:
-    # Fallback for ComfyUI's non-package loader or ad-hoc imports.
-    from services.access_control import require_admin_token, resolve_token_info
-    from services.aiohttp_compat import import_aiohttp_web
-    from services.endpoint_manifest import (
-        AuthTier,
-        RiskTier,
-        RoutePlane,
-        endpoint_metadata,
-    )
-    from services.presets import Preset, preset_store
-    from services.tenant_context import TenantBoundaryError, request_tenant_scope
+if __package__ and "." in __package__:
+    from ..services.import_fallback import import_attrs_dual
+else:
+    from services.import_fallback import import_attrs_dual  # type: ignore
+
+(require_admin_token, resolve_token_info) = import_attrs_dual(
+    __package__,
+    "..services.access_control",
+    "services.access_control",
+    ("require_admin_token", "resolve_token_info"),
+)
+(import_aiohttp_web,) = import_attrs_dual(
+    __package__,
+    "..services.aiohttp_compat",
+    "services.aiohttp_compat",
+    ("import_aiohttp_web",),
+)
+(AuthTier, RiskTier, RoutePlane, endpoint_metadata) = import_attrs_dual(
+    __package__,
+    "..services.endpoint_manifest",
+    "services.endpoint_manifest",
+    ("AuthTier", "RiskTier", "RoutePlane", "endpoint_metadata"),
+)
+(Preset, preset_store) = import_attrs_dual(
+    __package__,
+    "..services.presets",
+    "services.presets",
+    ("Preset", "preset_store"),
+)
+(TenantBoundaryError, request_tenant_scope) = import_attrs_dual(
+    __package__,
+    "..services.tenant_context",
+    "services.tenant_context",
+    ("TenantBoundaryError", "request_tenant_scope"),
+)
 
 logger = logging.getLogger("ComfyUI-OpenClaw.api.presets")
 web = import_aiohttp_web()

api/tools.py

@@ -8,33 +8,47 @@ from __future__ import annotations
 import json
 import logging
 
-try:
-    from ..services.access_control import require_admin_token, resolve_token_info
-    from ..services.aiohttp_compat import import_aiohttp_web
-    from ..services.audit import emit_audit_event
-    from ..services.tool_runner import get_tool_runner, is_tools_enabled
-except ImportError:
-    from services.access_control import require_admin_token  # type: ignore
-    from services.access_control import resolve_token_info  # type: ignore
-    from services.aiohttp_compat import import_aiohttp_web  # type: ignore
-    from services.audit import emit_audit_event  # type: ignore
-    from services.tool_runner import get_tool_runner, is_tools_enabled
-
-# R98: Endpoint Metadata
 if __package__ and "." in __package__:
-    from ..services.endpoint_manifest import (
-        AuthTier,
-        RiskTier,
-        RoutePlane,
-        endpoint_metadata,
-    )
+    from ..services.import_fallback import import_attrs_dual
 else:
-    from services.endpoint_manifest import (
-        AuthTier,
-        RiskTier,
-        RoutePlane,
-        endpoint_metadata,
-    )
+    from services.import_fallback import import_attrs_dual  # type: ignore
+
+(require_admin_token, resolve_token_info) = import_attrs_dual(
+    __package__,
+    "..services.access_control",
+    "services.access_control",
+    ("require_admin_token", "resolve_token_info"),
+)
+(import_aiohttp_web,) = import_attrs_dual(
+    __package__,
+    "..services.aiohttp_compat",
+    "services.aiohttp_compat",
+    ("import_aiohttp_web",),
+)
+(emit_audit_event,) = import_attrs_dual(
+    __package__,
+    "..services.audit",
+    "services.audit",
+    ("emit_audit_event",),
+)
+(get_tool_runner, is_tools_enabled) = import_attrs_dual(
+    __package__,
+    "..services.tool_runner",
+    "services.tool_runner",
+    ("get_tool_runner", "is_tools_enabled"),
+)
+(AuthTier, RiskTier, RoutePlane, endpoint_metadata) = import_attrs_dual(
+    __package__,
+    "..services.endpoint_manifest",
+    "services.endpoint_manifest",
+    ("AuthTier", "RiskTier", "RoutePlane", "endpoint_metadata"),
+)
+(check_surface,) = import_attrs_dual(
+    __package__,
+    "..services.surface_guard",
+    "services.surface_guard",
+    ("check_surface",),
+)
 
 logger = logging.getLogger("ComfyUI-OpenClaw.api.tools")
 web = import_aiohttp_web()
@@ -87,11 +101,6 @@ async def tools_run_handler(request: web.Request) -> web.Response:
     Requires: Admin Token.
     """
     # S62: Block tool execution in public+split mode
-    try:
-        # CRITICAL: package-relative import must stay first in ComfyUI runtime.
-        from ..services.surface_guard import check_surface
-    except ImportError:
-        from services.surface_guard import check_surface  # type: ignore
     blocked = check_surface("tool_execution", request)
     if blocked:
         return blocked

api/webhook.py

@@ -14,43 +14,68 @@ from __future__ import annotations
 import json
 import logging
 
-try:
-    from .errors import APIError, ErrorCode, create_error_response
-except ImportError:
-    # Build-time / Test fallback
-    from api.errors import APIError, ErrorCode, create_error_response
+if __package__ and "." in __package__:
+    from ..services.import_fallback import import_attrs_dual
+else:
+    from services.import_fallback import import_attrs_dual  # type: ignore
 
-try:
-    from ..models.schemas import MAX_BODY_SIZE, WebhookJobRequest
-    from ..services.aiohttp_compat import import_aiohttp_web
-    from ..services.metrics import metrics
-    from ..services.rate_limit import build_rate_limit_response, check_rate_limit
-    from ..services.trace import get_effective_trace_id
-    from ..services.webhook_auth import get_auth_summary, require_auth
-except ImportError:
-    from models.schemas import MAX_BODY_SIZE, WebhookJobRequest
-    from services.aiohttp_compat import import_aiohttp_web  # type: ignore
-    from services.metrics import metrics
-    from services.rate_limit import build_rate_limit_response, check_rate_limit
-    from services.trace import get_effective_trace_id
-    from services.webhook_auth import get_auth_summary, require_auth
-
-try:
-    from ..services.diagnostics_flags import diagnostics
-    from ..services.endpoint_manifest import (
-        AuthTier,
-        RiskTier,
-        RoutePlane,
-        endpoint_metadata,
-    )
-except ImportError:
-    from services.diagnostics_flags import diagnostics
-    from services.endpoint_manifest import (
-        AuthTier,
-        RiskTier,
-        RoutePlane,
-        endpoint_metadata,
-    )
+(APIError, ErrorCode, create_error_response) = import_attrs_dual(
+    __package__,
+    ".errors",
+    "api.errors",
+    ("APIError", "ErrorCode", "create_error_response"),
+)
+(
+    MAX_BODY_SIZE,
+    WebhookJobRequest,
+) = import_attrs_dual(
+    __package__,
+    "..models.schemas",
+    "models.schemas",
+    ("MAX_BODY_SIZE", "WebhookJobRequest"),
+)
+(import_aiohttp_web,) = import_attrs_dual(
+    __package__,
+    "..services.aiohttp_compat",
+    "services.aiohttp_compat",
+    ("import_aiohttp_web",),
+)
+(metrics,) = import_attrs_dual(
+    __package__,
+    "..services.metrics",
+    "services.metrics",
+    ("metrics",),
+)
+(build_rate_limit_response, check_rate_limit) = import_attrs_dual(
+    __package__,
+    "..services.rate_limit",
+    "services.rate_limit",
+    ("build_rate_limit_response", "check_rate_limit"),
+)
+(get_effective_trace_id,) = import_attrs_dual(
+    __package__,
+    "..services.trace",
+    "services.trace",
+    ("get_effective_trace_id",),
+)
+(get_auth_summary, require_auth) = import_attrs_dual(
+    __package__,
+    "..services.webhook_auth",
+    "services.webhook_auth",
+    ("get_auth_summary", "require_auth"),
+)
+(diagnostics,) = import_attrs_dual(
+    __package__,
+    "..services.diagnostics_flags",
+    "services.diagnostics_flags",
+    ("diagnostics",),
+)
+(AuthTier, RiskTier, RoutePlane, endpoint_metadata) = import_attrs_dual(
+    __package__,
+    "..services.endpoint_manifest",
+    "services.endpoint_manifest",
+    ("AuthTier", "RiskTier", "RoutePlane", "endpoint_metadata"),
+)
 
 # R46: Scoped logger for safe-by-default redaction
 logger = diagnostics.get_logger("ComfyUI-OpenClaw.api.webhook", "webhook")