admin/ComfyUI-OpenClaw
1
0
Fork 0
mirror of https://github.com/rookiestar28/ComfyUI-OpenClaw.git synced 2026-08-14 08:52:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: failover guard and provider error imports, add critical notes

Browse Source
This commit is contained in:
rookiestar28
2026-02-07 03:26:53 +08:00
parent 4b272a3758
commit 9a857fd09d
22 changed files with 375 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1 -1
View File
@@ -355,7 +355,7 @@ Set webhook auth env vars (see “Quick Start”) and restart ComfyUI.
### Admin Token: server-side vs UI
`OPENCLAW_ADMIN_TOKEN` is a **server-side environment variable**.
`OPENCLAW_ADMIN_TOKEN` is a **server-side environment variable**.
The Settings UI can **use** an Admin Token for authenticated requests, but **cannot set or persist** the server token.
Full setup steps: see `tests/TEST_SOP.md`.
api/config.py
+36 -3
View File
@@ -502,6 +502,12 @@ async def llm_test_handler(request: web.Request) -> web.Response:
from ..services.async_utils import run_in_thread
except ImportError:
from services.async_utils import run_in_thread
try:
# IMPORTANT: use package-relative import in ComfyUI runtime.
# CRITICAL: Missing this import causes NameError in provider error handling.
from ..services.provider_errors import ProviderHTTPError
except ImportError:
from services.provider_errors import ProviderHTTPError # type: ignore
# S26+: CSRF protection for convenience mode
admin_token_configured = bool(get_admin_token())
@@ -648,6 +654,12 @@ async def llm_chat_handler(request: web.Request) -> web.Response:
from ..services.async_utils import run_in_thread
except ImportError:
from services.async_utils import run_in_thread
try:
# IMPORTANT: use package-relative import in ComfyUI runtime.
# CRITICAL: Missing this import causes NameError in provider error handling.
from ..services.provider_errors import ProviderHTTPError
except ImportError:
from services.provider_errors import ProviderHTTPError # type: ignore
# S17: Rate Limit
if not check_rate_limit(request, "admin"):
@@ -685,8 +697,14 @@ async def llm_chat_handler(request: web.Request) -> web.Response:
if isinstance(body.get("user_message"), str)
else body.get("message") if isinstance(body.get("message"), str) else ""
)
temperature = body.get("temperature") if isinstance(body.get("temperature"), (int, float)) else 0.7
max_tokens = body.get("max_tokens") if isinstance(body.get("max_tokens"), int) else 1024
temperature = (
body.get("temperature")
if isinstance(body.get("temperature"), (int, float))
else 0.7
)
max_tokens = (
body.get("max_tokens") if isinstance(body.get("max_tokens"), int) else 1024
)
if not user_message:
return web.json_response(
@@ -716,8 +734,23 @@ async def llm_chat_handler(request: web.Request) -> web.Response:
{"ok": False, "error": str(e)},
status=400,
)
except ProviderHTTPError as e:
# IMPORTANT (recurring support issue):
# Do not swallow provider errors into a generic "llm_request_failed" without context.
# The connector can safely surface *redacted* provider messages (no prompt content)
# so users can fix misconfiguration (401/403/429, SSRF allowlist, etc.) quickly.
payload = {
"ok": False,
"error": f"{e.provider} HTTP {e.status_code}: {e.message}",
"provider": e.provider,
"status_code": e.status_code,
}
if getattr(e, "retry_after", None):
payload["retry_after"] = e.retry_after
return web.json_response(payload, status=e.status_code)
except Exception as e:
logger.error(f"LLM chat request failed: {type(e).__name__}")
# Log type + message (never log prompt content).
logger.error(f"LLM chat request failed: {type(e).__name__}: {e}")
return web.json_response(
{"ok": False, "error": "llm_request_failed"},
status=500,
api/routes.py
+2 -2
View File
@@ -41,9 +41,9 @@ if web is not None:
from ..api.config import (
config_get_handler,
config_put_handler,
llm_chat_handler,
llm_models_handler,
llm_test_handler,
llm_chat_handler,
)
from ..api.preflight_handler import inventory_handler, preflight_handler
from ..api.secrets import (
@@ -83,9 +83,9 @@ if web is not None:
from api.config import (
config_get_handler,
config_put_handler,
llm_chat_handler,
llm_models_handler,
llm_test_handler,
llm_chat_handler,
)
from api.preflight_handler import inventory_handler, preflight_handler
from api.secrets import (
api/templates.py
+1
View File
@@ -28,6 +28,7 @@ else: # pragma: no cover (test-only import mode)
logger = logging.getLogger("ComfyUI-OpenClaw.api.templates")
def _ensure_templates_api_deps_ready() -> tuple[bool, str | None]:
"""
Defensive guard against a recurring regression class:
connector/__main__.py
+11 -9
View File
@@ -30,12 +30,12 @@ def _print_security_banner(config):
Fail-closed: empty allowlists = all users treated as untrusted.
"""
has_trusted_users = bool(
config.telegram_allowed_users or
config.telegram_allowed_chats or
config.discord_allowed_users or
config.discord_allowed_channels or
config.line_allowed_users or
config.line_allowed_groups
config.telegram_allowed_users
or config.telegram_allowed_chats
or config.discord_allowed_users
or config.discord_allowed_channels
or config.line_allowed_users
or config.line_allowed_groups
)
has_admins = bool(config.admin_users)
@@ -48,7 +48,9 @@ def _print_security_banner(config):
if not has_admins:
logger.warning("⚠️ No admin users configured (OPENCLAW_CONNECTOR_ADMIN_USERS).")
logger.warning("⚠️ Admin commands (/approve, /reject, etc.) will be unavailable.")
logger.warning(
"⚠️ Admin commands (/approve, /reject, etc.) will be unavailable."
)
async def main():
@@ -75,10 +77,10 @@ async def main():
# Shared Platforms Registry
platforms = {}
# Initialize Poller
poller = ResultsPoller(config, client, platforms)
# Initialize Router with Poller
router = CommandRouter(config, client, poller=poller)
connector/config.py
+10 -4
View File
@@ -14,7 +14,7 @@ class ConnectorConfig:
# OpenClaw Connection
openclaw_url: str = "http://127.0.0.1:8188"
admin_token: Optional[str] = None # To call admin endpoints
# Results Delivery
delivery_enabled: bool = True
delivery_max_images: int = 4
@@ -66,9 +66,15 @@ def load_config() -> ConnectorConfig:
cfg.state_path = os.environ.get("OPENCLAW_CONNECTOR_STATE_PATH")
# Delivery
cfg.delivery_max_images = int(os.environ.get("OPENCLAW_CONNECTOR_DELIVERY_MAX_IMAGES", "4"))
cfg.delivery_max_bytes = int(os.environ.get("OPENCLAW_CONNECTOR_DELIVERY_MAX_BYTES", str(10 * 1024 * 1024)))
cfg.delivery_timeout_sec = int(os.environ.get("OPENCLAW_CONNECTOR_DELIVERY_TIMEOUT_SEC", "600"))
cfg.delivery_max_images = int(
os.environ.get("OPENCLAW_CONNECTOR_DELIVERY_MAX_IMAGES", "4")
)
cfg.delivery_max_bytes = int(
os.environ.get("OPENCLAW_CONNECTOR_DELIVERY_MAX_BYTES", str(10 * 1024 * 1024))
)
cfg.delivery_timeout_sec = int(
os.environ.get("OPENCLAW_CONNECTOR_DELIVERY_TIMEOUT_SEC", "600")
)
# Telegram
cfg.telegram_bot_token = os.environ.get("OPENCLAW_CONNECTOR_TELEGRAM_TOKEN")
connector/contract.py
+7 -2
View File
@@ -36,11 +36,16 @@ class Platform:
"""Stop/cleanup."""
pass
async def send_image(self, channel_id: str, image_data: bytes, filename: str = "image.png", caption: Optional[str] = None):
async def send_image(
self,
channel_id: str,
image_data: bytes,
filename: str = "image.png",
caption: Optional[str] = None,
):
"""Send an image to the channel."""
pass
async def send_message(self, channel_id: str, text: str):
"""Send a text message to the channel."""
pass
connector/llm_client.py
+9 -8
View File
@@ -17,7 +17,7 @@ logger = logging.getLogger(__name__)
class LLMClient:
"""
LLM client that fetches settings from OpenClaw backend.
Security:
- No conversation memory (stateless).
- Never logs user prompt content.
@@ -27,7 +27,7 @@ class LLMClient:
def __init__(self, openclaw_client):
"""
Initialize with OpenClawClient to fetch settings from backend.
Args:
openclaw_client: Instance of OpenClawClient for API calls.
"""
@@ -39,7 +39,7 @@ class LLMClient:
"""Fetch LLM config from OpenClaw backend."""
if self._config_cache is not None:
return self._config_cache
res = await self._client.get_openclaw_config()
if res.get("ok"):
data = res.get("data", {})
@@ -57,15 +57,15 @@ class LLMClient:
"""Check if LLM is properly configured in OpenClaw settings."""
if self._configured is not None:
return self._configured
config = await self._fetch_config()
provider = config.get("provider")
# Ollama doesn't require API key
if provider == "ollama":
self._configured = True
return True
# If backend includes an explicit flag, honor it.
if "api_key_configured" in config:
self._configured = bool(config.get("api_key_configured"))
@@ -84,7 +84,7 @@ class LLMClient:
) -> str:
"""
Send a chat request and return the assistant response.
Stateless: single system + user message per call.
No logging of user prompts for privacy.
"""
@@ -137,9 +137,10 @@ class LLMClient:
provider = config.get("provider", "openai")
model = config.get("model", "gpt-4o-mini")
base_url = config.get("base_url") or self._get_default_base_url(provider)
# Try to get API key from environment (fallback only)
import os
api_key = os.environ.get(f"OPENCLAW_{provider.upper()}_API_KEY")
endpoint = f"{base_url.rstrip('/')}/chat/completions"
connector/openclaw_client.py
+7 -9
View File
@@ -89,9 +89,7 @@ class OpenClawClient:
return result
except Exception as e:
logger.error(
f"Request failed {method} {path}: {type(e).__name__}: {e}"
)
logger.error(f"Request failed {method} {path}: {type(e).__name__}: {e}")
return {"ok": False, "error": str(e)}
finally:
if local_session:
@@ -123,9 +121,7 @@ class OpenClawClient:
"max_tokens": max_tokens,
}
# NOTE: LLM calls can exceed the default 10s HTTP timeout.
return await self._request(
"POST", "/openclaw/llm/chat", payload, timeout=120
)
return await self._request("POST", "/openclaw/llm/chat", payload, timeout=120)
async def get_health(self) -> dict:
res = await self._request("GET", "/openclaw/health")
@@ -174,17 +170,19 @@ class OpenClawClient:
# Remediation: Cancel -> Interrupt (Global)
return await self._request("POST", "/api/interrupt", {})
async def get_view(self, filename: str, subfolder: str = "", type: str = "output") -> Optional[bytes]:
async def get_view(
self, filename: str, subfolder: str = "", type: str = "output"
) -> Optional[bytes]:
"""Download image/file from ComfyUI /view endpoint."""
params = {"filename": filename, "subfolder": subfolder, "type": type}
url = f"{self.base_url}/view"
session = self.session
local_session = False
if not session:
session = _create_session()
local_session = True
try:
async with session.get(url, params=params, headers=self.headers) as resp:
if resp.status == 200:
connector/platforms/discord_gateway.py
+69 -19
View File
@@ -6,7 +6,6 @@ WebSocket connection to Discord Gateway (simplified) with Rate Limit Handling.
import asyncio
import json
import logging
import logging
import time
from typing import Optional
@@ -27,6 +26,17 @@ def _import_aiohttp():
class DiscordGateway:
GATEWAY_URL = "wss://gateway.discord.gg/?v=10&encoding=json"
# Discord Gateway Intents (bitmask).
#
# IMPORTANT (recurring support issue):
# DMs require DIRECT_MESSAGES intent. Without it, the connector will connect successfully
# (READY event) but will never receive DM MESSAGE_CREATE events, which looks like "no response".
_INTENT_GUILD_MESSAGES = 1 << 9
_INTENT_DIRECT_MESSAGES = 1 << 12
_INTENT_MESSAGE_CONTENT = 1 << 15
_INTENTS_DEFAULT = (
_INTENT_GUILD_MESSAGES | _INTENT_DIRECT_MESSAGES | _INTENT_MESSAGE_CONTENT
)
def __init__(self, config: ConnectorConfig, router: CommandRouter):
self.config = config
@@ -34,6 +44,7 @@ class DiscordGateway:
self.token = config.discord_bot_token
self.session = None
self.ws = None
self._aiohttp = None
self.heartbeat_interval = 41.25
self._seq = None
self._user_id = None
@@ -43,6 +54,10 @@ class DiscordGateway:
if aiohttp is None:
logger.warning("aiohttp not installed. Skipping Discord adapter.")
return
# IMPORTANT (recurring runtime bug):
# Do not rely on a local `aiohttp` variable outside this method.
# Other methods (_connect) need WSMsgType constants; store the module reference.
self._aiohttp = aiohttp
if not self.token:
logger.warning("Discord token not configured. Skipping.")
@@ -60,6 +75,8 @@ class DiscordGateway:
await asyncio.sleep(5)
async def _connect(self):
if self._aiohttp is None:
raise RuntimeError("aiohttp not available (DiscordGateway not initialized)")
async with self.session.ws_connect(self.GATEWAY_URL) as ws:
self.ws = ws
heartbeat_task = asyncio.create_task(self._heartbeat_loop())
@@ -68,7 +85,7 @@ class DiscordGateway:
await self._send_identify()
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
if msg.type == self._aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
self._seq = data.get("s")
op = data.get("op")
@@ -89,7 +106,7 @@ class DiscordGateway:
elif t == "MESSAGE_CREATE":
await self._process_message(data["d"])
elif msg.type == aiohttp.WSMsgType.ERROR:
elif msg.type == self._aiohttp.WSMsgType.ERROR:
break
finally:
heartbeat_task.cancel()
@@ -108,7 +125,8 @@ class DiscordGateway:
"op": 2,
"d": {
"token": self.token,
"intents": 33280,
# Requires "Message Content Intent" enabled in Discord Developer Portal.
"intents": self._INTENTS_DEFAULT,
"properties": {
"$os": "linux",
"$browser": "openclaw-connector",
@@ -125,6 +143,10 @@ class DiscordGateway:
content = message.get("content", "")
if not content:
if self.config.debug:
logger.info(
"Discord message ignored (empty content). This usually means Message Content Intent is disabled."
)
return
user_id = author.get("id")
@@ -202,7 +224,13 @@ class DiscordGateway:
break
async def send_image(self, channel_id: str, image_data: bytes, filename: str = "image.png", caption: Optional[str] = None):
async def send_image(
self,
channel_id: str,
image_data: bytes,
filename: str = "image.png",
caption: Optional[str] = None,
):
"""Send image via Discord API."""
if not self.session:
return
@@ -216,44 +244,66 @@ class DiscordGateway:
}
data = aiohttp.FormData()
if caption:
data.add_field("payload_json", json.dumps({"content": caption}))
data.add_field("files[0]", image_data, filename=filename, content_type="image/png")
# Discord expects multipart with optional `payload_json` plus `files[n]`.
# Send `payload_json` even if empty so the request shape is always consistent.
data.add_field("payload_json", json.dumps({"content": caption or ""}))
data.add_field(
"files[0]", image_data, filename=filename, content_type="image/png"
)
try:
async with self.session.post(url, headers=headers, data=data) as resp:
if resp.status != 200:
err = await resp.text()
logger.error(f"Discord send_image failed: {resp.status} {err}")
retries = 3
while retries > 0:
async with self.session.post(url, headers=headers, data=data) as resp:
if resp.status == 429:
try:
body = await resp.json()
retry_after = float(body.get("retry_after", 1))
except Exception:
retry_after = 1
logger.warning(
"Discord send_image rate-limited (429). Sleeping %.2fs",
retry_after,
)
await asyncio.sleep(retry_after)
retries -= 1
continue
if resp.status not in (200, 201):
err = await resp.text()
logger.error(f"Discord send_image failed: {resp.status} {err}")
raise RuntimeError(f"discord_send_image_failed:{resp.status}")
return
except Exception as e:
logger.error(f"Discord send_image error: {e}")
raise
async def send_message(self, channel_id: str, text: str):
"""Send text message."""
if not self.session:
return
import aiohttp
url = f"https://discord.com/api/v10/channels/{channel_id}/messages"
headers = {
"Authorization": f"Bot {self.token}",
"Content-Type": "application/json" # Explicit for JSON
"Content-Type": "application/json", # Explicit for JSON
}
# Simple Length Limit
if len(text) > 1900:
text = text[:1900] + "..."
payload = {"content": text}
try:
async with self.session.post(url, headers=headers, json=payload) as r:
if r.status != 200:
if r.status not in (200, 201):
# Ignore 429 for now in this simple implementation or copy logic?
# Copying simple logging
err = await r.text()
logger.error(f"Discord send_message failed: {r.status} {err}")
except Exception as e:
logger.error(f"Discord send_message error: {e}")
connector/platforms/line_webhook.py
+14 -9
View File
@@ -180,9 +180,7 @@ class LINEWebhookServer:
now = time.time() * 1000
cutoff = now - (self.REPLAY_WINDOW_SEC * 1000)
self._nonce_cache = {
k: v for k, v in self._nonce_cache.items() if v > cutoff
}
self._nonce_cache = {k: v for k, v in self._nonce_cache.items() if v > cutoff}
async def _process_event(self, event: dict):
"""Convert LINE event to CommandRequest and route."""
@@ -274,14 +272,22 @@ class LINEWebhookServer:
except Exception as e:
logger.error(f"LINE reply exception: {e}")
async def send_image(self, channel_id: str, image_data: bytes, filename: str = "image.png", caption: Optional[str] = None):
async def send_image(
self,
channel_id: str,
image_data: bytes,
filename: str = "image.png",
caption: Optional[str] = None,
):
"""
Send image via LINE.
NOTE: LINE requires a public HTTPS URL for images.
NOTE: LINE requires a public HTTPS URL for images.
Raw bytes upload is not supported in the standard Push API the same way.
This stub logs a warning until we implement a public hosting shim or use Imgur/S3.
"""
logger.warning("LINE send_image not implemented (requires public URL). Skipping.")
logger.warning(
"LINE send_image not implemented (requires public URL). Skipping."
)
async def send_message(self, channel_id: str, text: str):
"""Send push message."""
@@ -294,10 +300,10 @@ class LINEWebhookServer:
"Content-Type": "application/json",
"Authorization": f"Bearer {self.config.line_channel_access_token}",
}
body = {
"to": channel_id,
"messages": [{"type": "text", "text": text[:2000]}] # LINE limit handling
"messages": [{"type": "text", "text": text[:2000]}], # LINE limit handling
}
try:
@@ -307,4 +313,3 @@ class LINEWebhookServer:
logger.error(f"LINE send_message failed: {resp.status} {err}")
except Exception as e:
logger.error(f"LINE send_message error: {e}")
connector/platforms/telegram_polling.py
+10 -4
View File
@@ -176,7 +176,13 @@ class TelegramPolling:
except Exception as e:
logger.error(f"Telegram send exception: {e}")
async def send_image(self, channel_id: str, image_data: bytes, filename: str = "image.png", caption: Optional[str] = None):
async def send_image(
self,
channel_id: str,
image_data: bytes,
filename: str = "image.png",
caption: Optional[str] = None,
):
"""Send photo via Telegram sendPhoto."""
if not self.session:
return
@@ -188,9 +194,9 @@ class TelegramPolling:
data.add_field("chat_id", channel_id)
if caption:
data.add_field("caption", caption)
data.add_field("photo", image_data, filename=filename, content_type="image/png")
try:
async with self.session.post(url, data=data) as resp:
if resp.status != 200:
@@ -203,7 +209,7 @@ class TelegramPolling:
"""Send text message."""
if not self.session:
return
# Reuse internal logic logic but public
# Using simplified direct call
url = f"{self.base_url}/sendMessage"
connector/rate_limiter.py
+2 -1
View File
@@ -11,6 +11,7 @@ from typing import Dict
@dataclass
class TokenBucket:
"""Simple token bucket for rate limiting."""
capacity: float # Max tokens
refill_rate: float # Tokens per second
tokens: float = field(default=0.0)
@@ -34,7 +35,7 @@ class TokenBucket:
class RateLimiter:
"""
Per-user and per-channel rate limiter.
Default: 10 req/min per user, 30 req/min per channel.
"""
connector/results_poller.py
+45 -21
View File
@@ -24,7 +24,9 @@ class ResultsPoller:
self.config = config
self.client = client
self.platforms = platforms # map "telegram" -> TelegramPolling, etc.
self.queue = asyncio.Queue() # (prompt_id, platform_name, channel_id, sender_id)
self.queue = (
asyncio.Queue()
) # (prompt_id, platform_name, channel_id, sender_id)
self.active_polls = {} # prompt_id -> task
self.active_polls = {} # prompt_id -> task
@@ -40,7 +42,9 @@ class ResultsPoller:
self._poll_job(prompt_id, platform_name, channel_id, sender_id)
)
self.active_polls[prompt_id] = task
task.add_done_callback(lambda t, pid=prompt_id: self.active_polls.pop(pid, None))
task.add_done_callback(
lambda t, pid=prompt_id: self.active_polls.pop(pid, None)
)
finally:
self.queue.task_done()
@@ -49,16 +53,18 @@ class ResultsPoller:
logger.info("ResultsPoller stopping...")
for task in self.active_polls.values():
task.cancel()
if self.active_polls:
await asyncio.gather(*self.active_polls.values(), return_exceptions=True)
logger.info("ResultsPoller stopped.")
def track_job(self, prompt_id: str, platform_name: str, channel_id: str, sender_id: str):
def track_job(
self, prompt_id: str, platform_name: str, channel_id: str, sender_id: str
):
"""Enqueue a job for result monitoring."""
if not prompt_id:
return
logger.info(f"Tracking job {prompt_id} for {platform_name} in {channel_id}")
self.queue.put_nowait((prompt_id, platform_name, channel_id, sender_id))
@@ -68,7 +74,7 @@ class ResultsPoller:
"""Poll history with backoff until complete or timeout."""
start_time = time.time()
delay = 1.0
while (time.time() - start_time) < self.config.delivery_timeout_sec:
# Check history
try:
@@ -84,7 +90,7 @@ class ResultsPoller:
return
except Exception as e:
logger.debug(f"Poll check failed for {prompt_id}: {e}")
# Backoff
try:
await asyncio.sleep(delay)
@@ -93,7 +99,11 @@ class ResultsPoller:
delay = min(delay * 2, 15) # Cap at 15s
logger.warning(f"Job {prompt_id} timed out waiting for results.")
await self._send_text(platform_name, channel_id, f"⚠️ Job {prompt_id} timed out waiting for results.")
await self._send_text(
platform_name,
channel_id,
f"⚠️ Job {prompt_id} timed out waiting for results.",
)
async def _deliver_results(
self, prompt_id: str, job_data: dict, platform_name: str, channel_id: str
@@ -102,23 +112,29 @@ class ResultsPoller:
outputs = job_data.get("outputs", {})
if not outputs:
logger.info(f"Job {prompt_id} has no outputs.")
await self._send_text(platform_name, channel_id, f"✅ Job {prompt_id} finished (No output images).")
await self._send_text(
platform_name,
channel_id,
f"✅ Job {prompt_id} finished (No output images).",
)
return
images_to_send = []
# Flatten outputs
for node_id, node_output in outputs.items():
if "images" in node_output:
images_to_send.extend(node_output["images"])
# Check if actual images were found (filter non-outputs)
if not images_to_send:
logger.info(f"Job {prompt_id} finished but no images found.")
await self._send_text(platform_name, channel_id, f"✅ Job {prompt_id} finished (No images).")
return
logger.info(f"Job {prompt_id} finished but no images found.")
await self._send_text(
platform_name, channel_id, f"✅ Job {prompt_id} finished (No images)."
)
return
images_to_send = images_to_send[: self.config.delivery_max_images]
logger.info(f"Delivering {len(images_to_send)} images for {prompt_id}")
@@ -131,17 +147,23 @@ class ResultsPoller:
filename = img_info.get("filename")
subfolder = img_info.get("subfolder", "")
img_type = img_info.get("type", "output")
# Download content
content = await self.client.get_view(filename, subfolder, img_type)
if not content:
logger.warning(f"Failed to download {filename}")
continue
# Check size
if len(content) > self.config.delivery_max_bytes:
logger.warning(f"Image {filename} too large ({len(content)} bytes). Skipping.")
await self._send_text(platform_name, channel_id, f"⚠️ Image {filename} skipped (too large).")
logger.warning(
f"Image {filename} too large ({len(content)} bytes). Skipping."
)
await self._send_text(
platform_name,
channel_id,
f"⚠️ Image {filename} skipped (too large).",
)
continue
# Send with error handling
@@ -150,7 +172,9 @@ class ResultsPoller:
except Exception as e:
logger.error(f"Failed to deliver image to {platform_name}: {e}")
# Fallback text
await self._send_text(platform_name, channel_id, f"⚠️ Failed to send image: {filename}")
await self._send_text(
platform_name, channel_id, f"⚠️ Failed to send image: {filename}"
)
async def _send_text(self, platform_name: str, channel_id: str, text: str):
platform = self.platforms.get(platform_name)
connector/router.py
+34 -12
View File
@@ -16,14 +16,19 @@ if False: # Type hinting only
from .results_poller import ResultsPoller
from .llm_client import LLMClient
from .prompts import CHAT_SYSTEM_PROMPT, CHAT_STATUS_PROMPT
from .prompts import CHAT_STATUS_PROMPT, CHAT_SYSTEM_PROMPT
from .rate_limiter import RateLimiter
logger = logging.getLogger(__name__)
class CommandRouter:
def __init__(self, config: ConnectorConfig, client: OpenClawClient, poller: "ResultsPoller" = None):
def __init__(
self,
config: ConnectorConfig,
client: OpenClawClient,
poller: "ResultsPoller" = None,
):
self.config = config
self.client = client
self.poller = poller
@@ -62,7 +67,18 @@ class CommandRouter:
)
try:
parts = shlex.split(text)
# IMPORTANT (recurring usability bug):
# Do not use `shlex.split()` directly for ChatOps commands that may include natural
# language. In POSIX mode, `shlex` treats apostrophes (`'`) as quote delimiters, so
# common contractions like "She's" trigger "unbalanced quotes" failures.
#
# We therefore only treat *double quotes* (`"`) as quoting characters, so users can
# still do: positive_prompt="a prompt with spaces" while apostrophes remain safe.
lexer = shlex.shlex(text, posix=True)
lexer.whitespace_split = True
lexer.commenters = ""
lexer.quotes = '"'
parts = list(lexer)
except ValueError:
return CommandResponse(
text="[Error] Parsing command arguments failed (unbalanced quotes?)."
@@ -77,7 +93,11 @@ class CommandRouter:
# Telegram group commands often include the bot username suffix, e.g. `/help@mybot`.
# If we don't strip it, the command won't match our dispatch table and appears "dead"
# even though polling is working.
if (req.platform or "").lower() == "telegram" and cmd.startswith("/") and "@" in cmd:
if (
(req.platform or "").lower() == "telegram"
and cmd.startswith("/")
and "@" in cmd
):
cmd = cmd.split("@", 1)[0]
# Some users type `@bot /help` in group chats. Treat that as a command too.
@@ -286,7 +306,9 @@ class CommandRouter:
else:
prompt_id = data.get("prompt_id", "unknown")
if self.poller:
self.poller.track_job(prompt_id, req.platform, req.channel_id, req.sender_id)
self.poller.track_job(
prompt_id, req.platform, req.channel_id, req.sender_id
)
return CommandResponse(
text=f"[Job Submitted]\nID: {prompt_id}\nTemplate: {template_id}\nTrace: {trace_id}"
@@ -393,10 +415,10 @@ class CommandRouter:
# Phase 4: Show execution result
if "prompt_id" in data:
pid = data['prompt_id']
pid = data["prompt_id"]
msg += f"\nExecuted: {pid}"
if self.poller:
# Approval request might have come from different flow, but usually user invoking /approve
# Approval request might have come from different flow, but usually user invoking /approve
# wants the result. Using current req context is safest assumption for "ChatOps".
self.poller.track_job(pid, req.platform, req.channel_id, req.sender_id)
elif data.get("executed") is False:
@@ -534,7 +556,7 @@ class CommandRouter:
/chat [subcommand] <message>
Subcommands: run, template, status
Default: general chat
Security: Never auto-executes commands. Only suggests command text.
"""
llm = LLMClient(self.client)
@@ -579,7 +601,9 @@ class CommandRouter:
) -> CommandResponse:
"""Suggest a /run command based on user request."""
if not request:
return CommandResponse(text="Usage: /chat run <description of what you want>")
return CommandResponse(
text="Usage: /chat run <description of what you want>"
)
# Get available templates (simplified - could fetch from API)
templates = "txt2img, img2img, upscale (examples)"
@@ -597,9 +621,7 @@ Output only the command in a code block."""
response = await llm.chat(system_prompt, user_prompt)
return CommandResponse(text=response)
async def _chat_template(
self, llm: LLMClient, request: str
) -> CommandResponse:
async def _chat_template(self, llm: LLMClient, request: str) -> CommandResponse:
"""Generate a template JSON suggestion."""
if not request:
return CommandResponse(text="Usage: /chat template <description>")
services/llm_client.py
+16 -2
View File
@@ -496,6 +496,14 @@ class LLMClient:
eff_config, _ = get_effective_config()
max_failover_candidates = eff_config.get("max_failover_candidates", 3)
# NOTE: Keep at least 1 candidate; zero yields empty attempts and opaque errors.
# CRITICAL: Do not remove this guard. It prevents "All 0 failover candidates exhausted".
try:
max_failover_candidates = int(max_failover_candidates)
except (TypeError, ValueError):
max_failover_candidates = 3
if max_failover_candidates < 1:
max_failover_candidates = 1
# Get failover config
self.fallback_models = eff_config.get(
@@ -552,8 +560,14 @@ class LLMClient:
):
# Skip if in cooldown
if failover_state.is_cooling_down(provider, model):
logger.info(f"Skipping candidate {provider}/{model} (in cooldown)")
continue
if candidate_idx < (len(candidates_to_try) - 1):
logger.info(
f"Skipping candidate {provider}/{model} (in cooldown)"
)
continue
logger.warning(
f"Candidate {provider}/{model} is in cooldown, but no alternatives remain; attempting anyway."
)
# SSRF validation for custom base URLs
if not self._validate_candidate_url(provider, base_url):
services/providers/openai_compat.py
+8 -2
View File
@@ -109,8 +109,14 @@ def make_request(
except urllib.error.HTTPError as e:
# R14/R37: Parse retry-after from headers/body
try:
from services.provider_errors import ProviderHTTPError
from services.retry_after import get_retry_after_seconds
# IMPORTANT: ComfyUI runtime requires package-relative imports.
# CRITICAL: Do not collapse this to top-level imports; it breaks in custom_nodes.
try:
from ..provider_errors import ProviderHTTPError
from ..retry_after import get_retry_after_seconds
except ImportError:
from services.provider_errors import ProviderHTTPError
from services.retry_after import get_retry_after_seconds
# Get response headers and body
headers = dict(e.headers) if hasattr(e, "headers") else {}
services/queue_submit.py
+4 -1
View File
@@ -54,7 +54,10 @@ async def submit_prompt(
try:
from .execution_budgets import check_render_size, get_limiter
except ImportError:
from services.execution_budgets import check_render_size, get_limiter # type: ignore
from services.execution_budgets import ( # type: ignore
check_render_size,
get_limiter,
)
# R33: Check render size budget
check_render_size(prompt_workflow, trace_id=trace_id)
tests/TEST_SOP.md
+1 -1
View File
@@ -235,7 +235,7 @@ Unused keys have no effect unless the workflow contains a matching `{{key}}` pla
## Admin Token & UI Usage (SOP)
**Key rule:** `OPENCLAW_ADMIN_TOKEN` is a **server-side environment variable**.
**Key rule:** `OPENCLAW_ADMIN_TOKEN` is a **server-side environment variable**.
The UI can **use** an Admin Token for authenticated requests, but **cannot set or persist** the server token.
### Recommended setup (local only)
tests/connector/test_chat.py
+23 -8
View File
@@ -36,7 +36,9 @@ class MockOpenClawClient:
return {"ok": True, "data": {"exec_info": {"queue_remaining": 3}}}
def make_request(sender_id: str, text: str, platform: str = "telegram") -> CommandRequest:
def make_request(
sender_id: str, text: str, platform: str = "telegram"
) -> CommandRequest:
"""Helper to create CommandRequest with all required fields."""
return CommandRequest(
platform=platform,
@@ -89,7 +91,9 @@ class TestChatCommand(unittest.IsolatedAsyncioTestCase):
"""Should suggest /run with --approval for untrusted users."""
mock_llm = MagicMock()
mock_llm.is_configured = AsyncMock(return_value=True)
mock_llm.chat = AsyncMock(return_value="```\n/run txt2img --input prompt='cat' --approval\n```")
mock_llm.chat = AsyncMock(
return_value="```\n/run txt2img --input prompt='cat' --approval\n```"
)
mock_llm_cls.return_value = mock_llm
client = MockOpenClawClient()
@@ -104,7 +108,9 @@ class TestChatCommand(unittest.IsolatedAsyncioTestCase):
"""Should suggest /run without --approval for trusted users."""
mock_llm = MagicMock()
mock_llm.is_configured = AsyncMock(return_value=True)
mock_llm.chat = AsyncMock(return_value="```\n/run txt2img --input prompt='cat'\n```")
mock_llm.chat = AsyncMock(
return_value="```\n/run txt2img --input prompt='cat'\n```"
)
mock_llm_cls.return_value = mock_llm
self.config.telegram_allowed_users = [123]
@@ -122,7 +128,9 @@ class TestChatCommand(unittest.IsolatedAsyncioTestCase):
"""Should summarize status."""
mock_llm = MagicMock()
mock_llm.is_configured = AsyncMock(return_value=True)
mock_llm.chat = AsyncMock(return_value="System healthy. 1 running, 2 pending jobs.")
mock_llm.chat = AsyncMock(
return_value="System healthy. 1 running, 2 pending jobs."
)
mock_llm_cls.return_value = mock_llm
client = MockOpenClawClient()
@@ -149,7 +157,9 @@ class TestChatCommand(unittest.IsolatedAsyncioTestCase):
async def test_config_retrieval_failure(self):
"""Should handle OpenClaw config retrieval failure gracefully."""
client = MagicMock()
client.get_openclaw_config = AsyncMock(return_value={"ok": False, "error": "unreachable"})
client.get_openclaw_config = AsyncMock(
return_value={"ok": False, "error": "unreachable"}
)
client.get_health = AsyncMock(return_value={"ok": False})
client.get_jobs = AsyncMock(return_value={"ok": False})
client.get_prompt_queue = AsyncMock(return_value={"ok": False})
@@ -171,9 +181,14 @@ class TestLLMClient(unittest.IsolatedAsyncioTestCase):
client = MagicMock()
client.get_openclaw_config = AsyncMock(
return_value={"ok": True, "data": {"config": {"provider": "openai", "model": "gpt-4o"}}}
return_value={
"ok": True,
"data": {"config": {"provider": "openai", "model": "gpt-4o"}},
}
)
client.chat_llm = AsyncMock(
return_value={"ok": False, "error": "llm_request_failed"}
)
client.chat_llm = AsyncMock(return_value={"ok": False, "error": "llm_request_failed"})
llm = LLMClient(client)
result = await llm.chat("system", "user message")
@@ -190,7 +205,7 @@ class TestLLMClient(unittest.IsolatedAsyncioTestCase):
client.chat_llm = AsyncMock(return_value={"ok": True, "text": "response"})
llm = LLMClient(client)
# Verify the LLM client docstring mentions no prompt logging
# This is a design verification, not a runtime check
self.assertIn("Never logs user prompt content", LLMClient.__doc__ or "")
tests/connector/test_poller.py
+49 -27
View File
@@ -1,4 +1,3 @@
import asyncio
import unittest
from unittest.mock import AsyncMock, MagicMock, patch
@@ -9,7 +8,9 @@ from connector.results_poller import ResultsPoller
class MockPlatform(Platform):
async def send_image(self, channel_id, image_data, filename="image.png", caption=None):
async def send_image(
self, channel_id, image_data, filename="image.png", caption=None
):
pass
async def send_message(self, channel_id, text):
@@ -20,15 +21,15 @@ class TestResultsPoller(unittest.TestCase):
def setUp(self):
self.config = ConnectorConfig()
self.config.delivery_timeout_sec = 5 # short timeout for tests
self.client = MagicMock()
self.client.get_history = AsyncMock()
self.client.get_view = AsyncMock()
self.mock_platform = MockPlatform()
self.mock_platform.send_image = AsyncMock()
self.mock_platform.send_message = AsyncMock()
self.platforms = {"test_plat": self.mock_platform}
self.poller = ResultsPoller(self.config, self.client, self.platforms)
@@ -42,32 +43,48 @@ class TestResultsPoller(unittest.TestCase):
@patch("connector.results_poller.asyncio.sleep", new_callable=AsyncMock)
def test_poll_job_success(self, mock_sleep, mock_time):
mock_time.time.side_effect = [0, 1, 2, 3]
self.client.get_history.side_effect = [
{"ok": True, "data": {}},
{"ok": True, "data": {"p-1": {"outputs": {"node-1": {"images": [{"filename": "f.png", "type": "output"}]}}}}}
{
"ok": True,
"data": {
"p-1": {
"outputs": {
"node-1": {
"images": [{"filename": "f.png", "type": "output"}]
}
}
}
},
},
]
self.client.get_view.return_value = b"image_bytes"
asyncio.run(self.poller._poll_job("p-1", "test_plat", "c-1", "u-1"))
self.assertEqual(self.client.get_history.call_count, 2)
self.client.get_view.assert_called_with("f.png", "", "output")
self.mock_platform.send_image.assert_called_with("c-1", b"image_bytes", filename="f.png")
self.mock_platform.send_image.assert_called_with(
"c-1", b"image_bytes", filename="f.png"
)
@patch("connector.results_poller.time")
@patch("connector.results_poller.asyncio.sleep", new_callable=AsyncMock)
def test_poll_job_no_outputs(self, mock_sleep, mock_time):
# Scenario: Job finished, but "outputs" is empty or has no images.
mock_time.time.side_effect = [0, 1, 2]
self.client.get_history.side_effect = [
{"ok": True, "data": {"p-empty": {"outputs": {}}}} # Completed, empty outputs
{
"ok": True,
"data": {"p-empty": {"outputs": {}}},
} # Completed, empty outputs
]
asyncio.run(self.poller._poll_job("p-empty", "test_plat", "c-1", "u-1"))
self.mock_platform.send_image.assert_not_called()
self.mock_platform.send_message.assert_called_once()
args = self.mock_platform.send_message.call_args
@@ -78,32 +95,32 @@ class TestResultsPoller(unittest.TestCase):
def test_poll_job_timeout(self, mock_sleep, mock_time):
self.config.delivery_timeout_sec = 2
mock_time.time.side_effect = [0, 1, 3]
self.client.get_history.return_value = {"ok": True, "data": {}}
self.client.get_history.return_value = {"ok": True, "data": {}}
asyncio.run(self.poller._poll_job("p-timeout", "test_plat", "c-1", "u-1"))
self.mock_platform.send_image.assert_not_called()
self.mock_platform.send_message.assert_called_once()
self.assertIn("timed out", self.mock_platform.send_message.call_args[0][1])
def test_deliver_results_limits(self):
self.config.delivery_max_images = 1
job_data = {
"outputs": {
"n1": {
"images": [
{"filename": "1.png", "type": "output"},
{"filename": "2.png", "type": "output"}
{"filename": "1.png", "type": "output"},
{"filename": "2.png", "type": "output"},
]
}
}
}
self.client.get_view.return_value = b"123"
asyncio.run(self.poller._deliver_results("p-1", job_data, "test_plat", "c-1"))
self.assertEqual(self.mock_platform.send_image.call_count, 1)
args = self.mock_platform.send_image.call_args
self.assertEqual(args[1]["filename"], "1.png")
@@ -111,16 +128,21 @@ class TestResultsPoller(unittest.TestCase):
def test_deliver_results_send_failure(self):
# Scenario: Platform.send_image raises Exception
job_data = {
"outputs": { "n1": { "images": [{"filename": "f.png", "type": "output"}] } }
"outputs": {"n1": {"images": [{"filename": "f.png", "type": "output"}]}}
}
self.client.get_view.return_value = b"bytes"
self.mock_platform.send_image.side_effect = Exception("Network Error")
asyncio.run(self.poller._deliver_results("p-fail", job_data, "test_plat", "c-1"))
asyncio.run(
self.poller._deliver_results("p-fail", job_data, "test_plat", "c-1")
)
# Should catch exception and try sending fallback text
self.mock_platform.send_message.assert_called()
self.assertIn("Failed to send image", self.mock_platform.send_message.call_args[0][1])
self.assertIn(
"Failed to send image", self.mock_platform.send_message.call_args[0][1]
)
if __name__ == "__main__":
unittest.main()
tests/connector/test_security.py
+16 -1
View File
@@ -100,6 +100,19 @@ class TestRouterSecurityChecks(unittest.IsolatedAsyncioTestCase):
# Last response should be rate limit
self.assertIn("rate limit", resp.text.lower())
async def test_apostrophes_do_not_break_parsing(self):
"""Natural language apostrophes (e.g. She's) must not trigger shlex quote errors."""
client = MagicMock()
client.submit_job = AsyncMock(return_value={"ok": False, "error": "test"})
router = CommandRouter(self.config, client)
req = make_request(
"user1",
"/run z She's wearing an oversized tee seed=-1",
)
resp = await router.handle(req)
self.assertNotIn("unbalanced quotes", resp.text.lower())
class TestLineReplayProtection(unittest.TestCase):
"""Test LINE webhook replay protection."""
@@ -139,7 +152,9 @@ class TestLineReplayProtection(unittest.TestCase):
server = LINEWebhookServer(config, MagicMock())
recent_ts = int((time.time() - 10) * 1000)
body = f'{{"events": [{{"timestamp": {recent_ts}, "webhookEventId": "dup_evt"}}]}}'
body = (
f'{{"events": [{{"timestamp": {recent_ts}, "webhookEventId": "dup_evt"}}]}}'
)
# First request accepted
self.assertTrue(server._check_replay_protection(body))