diff --git a/.gitignore b/.gitignore index 8a06626..030dc28 100644 --- a/.gitignore +++ b/.gitignore @@ -24,6 +24,7 @@ openclaw_state/ *.log test_output.txt test_auth_out.txt +connector_state.json* .DS_Store .vscode/ .idea/ diff --git a/README.md b/README.md index ae01882..c5e2411 100644 --- a/README.md +++ b/README.md @@ -213,14 +213,16 @@ Callback delivery allowlist (sidecar HTTP adapter): ## Templates -Templates live in `data/templates/` and are loaded from `data/templates/manifest.json`. +Templates live in `data/templates/`. -- Only templates listed in the manifest are usable. -- Each template declares `allowed_inputs` and optional defaults. +- Any `data/templates/.json` file is runnable (template ID = filename stem). +- `data/templates/manifest.json` is optional metadata (e.g. defaults). - Rendering performs **strict placeholder substitution**: - Only exact string values matching `{{key}}` are replaced - Partial substitutions (e.g. `"foo {{bar}}"`) are intentionally not supported +For the full step-by-step guide (where to put exported workflow JSON, how to author `manifest.json`, how to verify `/openclaw/templates`, and how to use `/run`), see `tests/TEST_SOP.md`. + ## Execution Budgets Queue submissions are protected by concurrency caps and render size budgets (`services/execution_budgets.py`). diff --git a/__init__.py b/__init__.py index f7d6b58..89adc14 100644 --- a/__init__.py +++ b/__init__.py @@ -147,15 +147,17 @@ def _register_routes_once(): source="unknown", ): """Submit function for scheduler and trigger-triggered runs.""" - from .services.idempotency_store import get_store + # NOTE: Use IdempotencyStore API (check_and_record/update_prompt_id). + # Avoid legacy get_store/get/set usage; wrong API here breaks route registration at runtime. + from .services.idempotency_store import IdempotencyStore from .services.queue_submit import submit_prompt from .services.templates import get_template_service # Check idempotency - store = get_store() - existing = store.get(idempotency_key) - if existing: - return {"prompt_id": existing.get("prompt_id"), "deduped": True} + store = IdempotencyStore() + is_dup, existing_prompt_id = store.check_and_record(idempotency_key) + if is_dup: + return {"prompt_id": existing_prompt_id, "deduped": True} # Render template tmpl_svc = get_template_service() @@ -174,7 +176,7 @@ def _register_routes_once(): # Store for dedupe if result.get("prompt_id"): - store.set(idempotency_key, {"prompt_id": result["prompt_id"]}) + store.update_prompt_id(idempotency_key, result["prompt_id"]) return result diff --git a/api/routes.py b/api/routes.py index 08ce3b8..df13586 100644 --- a/api/routes.py +++ b/api/routes.py @@ -21,6 +21,7 @@ PACK_NAME = PACK_VERSION = PACK_START_TIME = LOG_FILE = get_api_key = None # ty metrics = tail_log = require_observability_access = check_rate_limit = trace_store = None # type: ignore webhook_handler = webhook_submit_handler = webhook_validate_handler = capabilities_handler = preflight_handler = None # type: ignore config_get_handler = config_put_handler = llm_test_handler = llm_models_handler = None # type: ignore +templates_list_handler = None # type: ignore secrets_status_handler = secrets_put_handler = secrets_delete_handler = None # type: ignore list_checkpoints_handler = create_checkpoint_handler = get_checkpoint_handler = delete_checkpoint_handler = None # type: ignore redact_text = None # type: ignore @@ -49,6 +50,7 @@ if web is not None: secrets_put_handler, secrets_status_handler, ) + from ..api.templates import templates_list_handler from ..api.webhook import webhook_handler from ..api.webhook_submit import webhook_submit_handler from ..api.webhook_validate import webhook_validate_handler @@ -89,6 +91,7 @@ if web is not None: secrets_put_handler, secrets_status_handler, ) + from api.templates import templates_list_handler from api.webhook import webhook_handler from api.webhook_submit import webhook_submit_handler from api.webhook_validate import webhook_validate_handler @@ -456,6 +459,11 @@ def register_routes(server) -> None: f"{prefix}/llm/models", llm_models_handler, ), # F20+: Remote model list (best-effort) + ( + "GET", + f"{prefix}/templates", + templates_list_handler, + ), # F29: Template quick list for chat connectors ( "POST", f"{prefix}/preflight", diff --git a/api/schedules.py b/api/schedules.py index 3ce4210..bfc1941 100644 --- a/api/schedules.py +++ b/api/schedules.py @@ -12,10 +12,16 @@ from ..services.scheduler.models import Schedule, TriggerType from ..services.scheduler.storage import get_schedule_store from ..services.templates import is_template_allowed -try: +# Import discipline: +# - ComfyUI runtime: package-relative imports only (prevents collisions with other custom nodes). +# - Unit tests: allow top-level fallbacks. +# +# IMPORTANT: Avoid a broad `try/except ImportError` here. Falling back to `services.*` in ComfyUI +# can silently import another pack's module and break auth/approval semantics. +if __package__ and "." in __package__: from ..services.webhook_auth import AuthError -except ImportError: - from services.webhook_auth import AuthError +else: # pragma: no cover (test-only import mode) + from services.webhook_auth import AuthError # type: ignore logger = logging.getLogger("ComfyUI-OpenClaw.api.schedules") @@ -112,8 +118,8 @@ class ScheduleHandlers: template_id = data["template_id"] if not self._template_checker(template_id): return web.json_response( - {"error": f"template_id '{template_id}' is not in allowlist"}, - status=403, + {"error": f"template_id '{template_id}' not found"}, + status=404, ) # Build schedule @@ -165,8 +171,8 @@ class ScheduleHandlers: template_id = data["template_id"] if not self._template_checker(template_id): return web.json_response( - {"error": f"template_id '{template_id}' is not in allowlist"}, - status=403, + {"error": f"template_id '{template_id}' not found"}, + status=404, ) existing.template_id = template_id diff --git a/api/templates.py b/api/templates.py new file mode 100644 index 0000000..0b540d8 --- /dev/null +++ b/api/templates.py @@ -0,0 +1,112 @@ +""" +Templates API (F29 support). + +Provides a lightweight endpoint to list template IDs that can be used +with `/openclaw/triggers/fire` (and by chat connectors via `/run ...`). +""" + +from __future__ import annotations + +import logging + +try: + from aiohttp import web # type: ignore +except ImportError: # pragma: no cover (optional for unit tests) + web = None # type: ignore + +# Import discipline: +# - ComfyUI runtime: package-relative imports only (prevents collisions with other custom nodes). +# - Unit tests: allow top-level fallbacks. +if __package__ and "." in __package__: + from ..services.access_control import require_observability_access + from ..services.rate_limit import check_rate_limit + from ..services.templates import get_template_service +else: # pragma: no cover (test-only import mode) + from services.access_control import require_observability_access # type: ignore + from services.rate_limit import check_rate_limit # type: ignore + from services.templates import get_template_service # type: ignore + +logger = logging.getLogger("ComfyUI-OpenClaw.api.templates") + +def _ensure_templates_api_deps_ready() -> tuple[bool, str | None]: + """ + Defensive guard against a recurring regression class: + if import discipline is broken, these symbols can become `None`, + causing aiohttp to emit noisy tracebacks and clients to see ERR_INVALID_RESPONSE. + """ + missing = [] + if not callable(require_observability_access): + missing.append("require_observability_access") + if not callable(check_rate_limit): + missing.append("check_rate_limit") + if not callable(get_template_service): + missing.append("get_template_service") + if missing: + return ( + False, + "Backend not fully initialized (missing route dependencies: " + + ", ".join(missing) + + ").", + ) + return True, None + + +async def templates_list_handler(request: web.Request) -> web.Response: + """ + GET /openclaw/templates (legacy: /moltbot/templates) + + Returns the templates visible to the backend (file-based discovery + optional manifest metadata). + This is safe to expose under the observability boundary because it contains + no secrets and does not return workflow bodies. + """ + if web is None: + raise RuntimeError("aiohttp not available") + + ok, init_error = _ensure_templates_api_deps_ready() + if not ok: + return web.json_response({"ok": False, "error": init_error}, status=500) + + allowed, error = require_observability_access(request) + if not allowed: + return web.json_response({"ok": False, "error": error}, status=403) + + # Reuse the admin bucket to avoid unbounded enumeration from remote callers. + if not check_rate_limit(request, "admin"): + return web.json_response( + {"ok": False, "error": "Rate limit exceeded"}, + status=429, + headers={"Retry-After": "60"}, + ) + + try: + svc = get_template_service() + items = [] + # Prefer runtime discovery (file-based templates) so operators don't need + # to maintain a separate allowlist file. + for template_id in svc.get_debug_info().get("discovered_template_ids", []): # type: ignore[call-arg] + cfg = svc.get_template_config(template_id) # type: ignore[arg-type] + if cfg is None: + continue + items.append( + { + "id": template_id, + "allowed_inputs": list(cfg.allowed_inputs or []), + "defaults": dict(cfg.defaults or {}), + } + ) + items.sort(key=lambda x: x["id"]) + resp: dict = {"ok": True, "templates": items, "count": len(items)} + + # Optional diagnostics. This reveals absolute paths, so keep it opt-in. + debug = request.query.get("debug", "").strip() in ("1", "true", "yes") + if debug: + try: + resp["debug"] = svc.get_debug_info() # type: ignore[attr-defined] + except Exception: + # If TemplateService interface changes, don't break the endpoint. + resp["debug"] = {"error": "debug_info_unavailable"} + + return web.json_response(resp) + except Exception as e: + logger.exception("Failed to list templates") + return web.json_response({"ok": False, "error": str(e)}, status=500) diff --git a/api/triggers.py b/api/triggers.py index b441e97..da15931 100644 --- a/api/triggers.py +++ b/api/triggers.py @@ -12,17 +12,26 @@ from typing import Optional from aiohttp import web -try: +# Import discipline: +# - ComfyUI runtime: this pack is loaded as a package; MUST use package-relative imports to avoid +# collisions with other custom nodes or other top-level modules named `services`. +# - Unit tests: modules may be imported as top-level (e.g. `api.*`), so allow top-level fallbacks. +# +# IMPORTANT (recurring production bug): +# Do NOT wrap these imports in a broad `try/except ImportError` without checking `__package__`. +# If the pack is loaded in a way that makes relative imports fail, falling back to `from services...` +# can silently import the WRONG module (another custom node or ComfyUI-adjacent package), causing +# template allowlists to appear "missing" even when `data/templates/manifest.json` is correct. +if __package__ and "." in __package__: from ..services.execution_budgets import BudgetExceededError from ..services.templates import is_template_allowed from ..services.trace import generate_trace_id from ..services.webhook_auth import AuthError -except ImportError: - # Fallback for ComfyUI's non-package loader or ad-hoc imports. - from services.execution_budgets import BudgetExceededError - from services.templates import is_template_allowed - from services.trace import generate_trace_id - from services.webhook_auth import AuthError +else: # pragma: no cover (test-only import mode) + from services.execution_budgets import BudgetExceededError # type: ignore + from services.templates import is_template_allowed # type: ignore + from services.trace import generate_trace_id # type: ignore + from services.webhook_auth import AuthError # type: ignore logger = logging.getLogger("ComfyUI-OpenClaw.api.triggers") @@ -117,8 +126,8 @@ class TriggerHandlers: # Check template allowlist if not self._template_checker(template_id): return web.json_response( - {"error": f"template_id '{template_id}' is not in allowlist"}, - status=403, + {"error": f"template_id '{template_id}' not found"}, + status=404, ) # Extract optional fields @@ -162,7 +171,17 @@ class TriggerHandlers: callback: Optional[dict], ) -> web.Response: """Create an approval request instead of immediate execution.""" - from services.approvals import ApprovalSource, get_approval_service + # IMPORTANT (recurring production bug): + # In ComfyUI runtime, do NOT import `services.*` as a fallback here. + # If another custom node exposes a top-level `services` package, you'll import the wrong + # module and create hard-to-debug runtime mismatches (approvals/allowlists/etc). + if __package__ and "." in __package__: + from ..services.approvals import ApprovalSource, get_approval_service + else: # pragma: no cover (test-only import mode) + from services.approvals import ( # type: ignore + ApprovalSource, + get_approval_service, + ) service = get_approval_service() @@ -265,7 +284,14 @@ async def execute_approved_trigger( Raises: ValueError: If approval not found or not approved """ - from services.approvals import ApprovalStatus, get_approval_service + # IMPORTANT: See note above about avoiding `services.*` imports in ComfyUI runtime. + if __package__ and "." in __package__: + from ..services.approvals import ApprovalStatus, get_approval_service + else: # pragma: no cover (test-only import mode) + from services.approvals import ( # type: ignore + ApprovalStatus, + get_approval_service, + ) service = get_approval_service() approval = service.get(approval_id) diff --git a/api/webhook_submit.py b/api/webhook_submit.py index d20a41d..926f792 100644 --- a/api/webhook_submit.py +++ b/api/webhook_submit.py @@ -8,7 +8,14 @@ import logging from aiohttp import web -try: +# Import discipline: +# - ComfyUI runtime: package-relative imports only (prevents collisions with other custom nodes). +# - Unit tests: allow top-level fallbacks. +# +# IMPORTANT (recurring production bug): +# Do NOT wrap these imports in a broad `try/except ImportError`. In ComfyUI, that can silently +# import another pack's top-level `services` module and break allowlists/auth in surprising ways. +if __package__ and "." in __package__: from ..models.schemas import MAX_BODY_SIZE, WebhookJobRequest from ..services.callback_delivery import start_callback_watch from ..services.execution_budgets import BudgetExceededError @@ -20,19 +27,18 @@ try: from ..services.trace import get_effective_trace_id from ..services.trace_store import trace_store from ..services.webhook_auth import require_auth -except ImportError: - # Handle path issues for testing or different contexts +else: # pragma: no cover (test-only import mode) from models.schemas import MAX_BODY_SIZE, WebhookJobRequest - from services.callback_delivery import start_callback_watch - from services.execution_budgets import BudgetExceededError - from services.idempotency_store import IdempotencyStore - from services.metrics import metrics - from services.queue_submit import submit_prompt - from services.rate_limit import check_rate_limit - from services.templates import get_template_service - from services.trace import get_effective_trace_id - from services.trace_store import trace_store - from services.webhook_auth import require_auth + from services.callback_delivery import start_callback_watch # type: ignore + from services.execution_budgets import BudgetExceededError # type: ignore + from services.idempotency_store import IdempotencyStore # type: ignore + from services.metrics import metrics # type: ignore + from services.queue_submit import submit_prompt # type: ignore + from services.rate_limit import check_rate_limit # type: ignore + from services.templates import get_template_service # type: ignore + from services.trace import get_effective_trace_id # type: ignore + from services.trace_store import trace_store # type: ignore + from services.webhook_auth import require_auth # type: ignore logger = logging.getLogger("ComfyUI-OpenClaw.api.webhook_submit") diff --git a/api/webhook_validate.py b/api/webhook_validate.py index 1b15f5d..de4081d 100644 --- a/api/webhook_validate.py +++ b/api/webhook_validate.py @@ -28,7 +28,14 @@ try: except ModuleNotFoundError: # pragma: no cover web = None # type: ignore -try: +# Import discipline: +# - ComfyUI runtime: package-relative imports only (prevents collisions with other custom nodes). +# - Unit tests: allow top-level fallbacks. +# +# IMPORTANT (recurring production bug): +# Do NOT wrap these imports in a broad `try/except ImportError`. In ComfyUI, that can silently +# import another pack's top-level `services` module and break allowlists/auth in surprising ways. +if __package__ and "." in __package__: from ..models.schemas import MAX_BODY_SIZE, WebhookJobRequest from ..services.execution_budgets import BudgetExceededError, check_render_size from ..services.metrics import metrics @@ -36,14 +43,17 @@ try: from ..services.templates import get_template_service from ..services.trace import get_effective_trace_id from ..services.webhook_auth import require_auth -except ImportError: +else: # pragma: no cover (test-only import mode) from models.schemas import MAX_BODY_SIZE, WebhookJobRequest - from services.execution_budgets import BudgetExceededError, check_render_size - from services.metrics import metrics - from services.rate_limit import check_rate_limit - from services.templates import get_template_service - from services.trace import get_effective_trace_id - from services.webhook_auth import require_auth + from services.execution_budgets import ( # type: ignore + BudgetExceededError, + check_render_size, + ) + from services.metrics import metrics # type: ignore + from services.rate_limit import check_rate_limit # type: ignore + from services.templates import get_template_service # type: ignore + from services.trace import get_effective_trace_id # type: ignore + from services.webhook_auth import require_auth # type: ignore logger = logging.getLogger("ComfyUI-OpenClaw.api.webhook_validate") diff --git a/connector/__main__.py b/connector/__main__.py index 89d93bb..c51babb 100644 --- a/connector/__main__.py +++ b/connector/__main__.py @@ -12,6 +12,7 @@ from .openclaw_client import OpenClawClient from .platforms.discord_gateway import DiscordGateway from .platforms.line_webhook import LINEWebhookServer from .platforms.telegram_polling import TelegramPolling +from .results_poller import ResultsPoller from .router import CommandRouter # Configure logging @@ -24,7 +25,7 @@ logger = logging.getLogger("connector") async def main(): - logger.info("Initializing OpenClaw Connector (Phase 3)...") + logger.info("Initializing OpenClaw Connector (Phase 5)...") # 1. Config try: @@ -42,14 +43,25 @@ async def main(): client = OpenClawClient(config) await client.start() # Start session - router = CommandRouter(config, client) + # Shared Platforms Registry + platforms = {} + + # Initialize Poller + poller = ResultsPoller(config, client, platforms) + + # Initialize Router with Poller + router = CommandRouter(config, client, poller=poller) tasks = [] + # Start Poller + tasks.append(asyncio.create_task(poller.start())) + line_server = None # 3. Platforms if config.telegram_bot_token: tg = TelegramPolling(config, router) + platforms["telegram"] = tg tasks.append(asyncio.create_task(tg.start())) else: logger.info( @@ -58,12 +70,14 @@ async def main(): if config.discord_bot_token: dc = DiscordGateway(config, router) + platforms["discord"] = dc tasks.append(asyncio.create_task(dc.start())) else: logger.info("Discord not configured (OPENCLAW_CONNECTOR_DISCORD_TOKEN missing)") if config.line_channel_secret and config.line_channel_access_token: line_server = LINEWebhookServer(config, router) + platforms["line"] = line_server await line_server.start() # If only LINE is active, tasks will be empty. Add a sleeper to keep loop alive. if not tasks: @@ -100,6 +114,8 @@ async def main(): finally: if line_server: await line_server.stop() + if poller: + await poller.stop() await client.close() logger.info("Connector stopped.") diff --git a/connector/config.py b/connector/config.py index 89d2efb..8ad15e6 100644 --- a/connector/config.py +++ b/connector/config.py @@ -14,6 +14,12 @@ 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 + delivery_max_bytes: int = 10 * 1024 * 1024 # 10MB + delivery_timeout_sec: int = 600 # Telegram telegram_bot_token: Optional[str] = None @@ -53,6 +59,11 @@ def load_config() -> ConnectorConfig: cfg.debug = os.environ.get("OPENCLAW_CONNECTOR_DEBUG", "0") == "1" 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")) + # Telegram cfg.telegram_bot_token = os.environ.get("OPENCLAW_CONNECTOR_TELEGRAM_TOKEN") if t_users := os.environ.get("OPENCLAW_CONNECTOR_TELEGRAM_ALLOWED_USERS"): diff --git a/connector/contract.py b/connector/contract.py index 5427c2e..eb37ce5 100644 --- a/connector/contract.py +++ b/connector/contract.py @@ -23,3 +23,24 @@ class CommandResponse: text: str files: List[str] = field(default_factory=list) # Local paths to upload buttons: List[dict] = field(default_factory=list) # Simple quick replies + + +class Platform: + """Abstract base class for chat platforms.""" + + async def start(self): + """Start the platform connection/polling.""" + pass + + async def stop(self): + """Stop/cleanup.""" + pass + + 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 + diff --git a/connector/openclaw_client.py b/connector/openclaw_client.py index 1f1afbe..6ac9f46 100644 --- a/connector/openclaw_client.py +++ b/connector/openclaw_client.py @@ -142,6 +142,31 @@ 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]: + """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: + return await resp.read() + else: + logger.warning(f"get_view failed: {resp.status}") + return None + except Exception as e: + logger.error(f"get_view error: {e}") + return None + finally: + if local_session: + await session.close() + # --- Approvals --- async def get_approvals(self) -> dict: diff --git a/connector/platforms/discord_gateway.py b/connector/platforms/discord_gateway.py index a824b10..9267913 100644 --- a/connector/platforms/discord_gateway.py +++ b/connector/platforms/discord_gateway.py @@ -6,7 +6,9 @@ WebSocket connection to Discord Gateway (simplified) with Rate Limit Handling. import asyncio import json import logging +import logging import time +from typing import Optional from ..config import ConnectorConfig from ..contract import CommandRequest, CommandResponse @@ -135,12 +137,10 @@ class DiscordGateway: if channel_id in self.config.discord_allowed_channels: is_allowed = True - if not is_allowed: - if self.config.debug: - logger.debug( - f"Ignored Discord message user={user_id} chan={channel_id}" - ) - return + if not is_allowed and self.config.debug: + logger.debug( + f"Untrusted Discord message user={user_id} chan={channel_id} (will require approval)" + ) # Build Request req = CommandRequest( @@ -201,3 +201,59 @@ class DiscordGateway: ) break + + 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 + + import aiohttp + + url = f"https://discord.com/api/v10/channels/{channel_id}/messages" + headers = { + "Authorization": f"Bot {self.token}", + # Do NOT set Content-Type; FormData handling does it + } + + 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") + + 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}") + except Exception as e: + logger.error(f"Discord send_image error: {e}") + + 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 + } + + # 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: + # 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}") + diff --git a/connector/platforms/line_webhook.py b/connector/platforms/line_webhook.py index 6c7de91..5922bbf 100644 --- a/connector/platforms/line_webhook.py +++ b/connector/platforms/line_webhook.py @@ -153,17 +153,16 @@ class LINEWebhookServer: is_allowed = True if not is_allowed: - # Remediation: Explicit logging for empty list or reject - msg = f"Ignored LINE message from user={user_id} in channel={channel_id}." + # Informational only: untrusted messages are accepted but will require approval. + msg = f"Untrusted LINE message from user={user_id} in channel={channel_id}." if ( not self.config.line_allowed_users and not self.config.line_allowed_groups ): - msg += " (Allow lists are empty! Configure OPENCLAW_CONNECTOR_LINE_ALLOWED_USERS/GROUPS)" + msg += " (Allow lists are empty; all users will require approval)" else: - msg += " (Not in allowlist)" + msg += " (Not in allowlist; approval required)" logger.warning(msg) - return req = CommandRequest( platform="line", @@ -214,3 +213,38 @@ 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): + """ + Send image via LINE. + 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.") + + async def send_message(self, channel_id: str, text: str): + """Send push message.""" + aiohttp, _ = _import_aiohttp_web() + if not aiohttp or not self.session: + return + + url = "https://api.line.me/v2/bot/message/push" + headers = { + "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 + } + + try: + async with self.session.post(url, headers=headers, json=body) as resp: + if resp.status != 200: + err = await resp.text() + logger.error(f"LINE send_message failed: {resp.status} {err}") + except Exception as e: + logger.error(f"LINE send_message error: {e}") + diff --git a/connector/platforms/telegram_polling.py b/connector/platforms/telegram_polling.py index fb404b2..87566b1 100644 --- a/connector/platforms/telegram_polling.py +++ b/connector/platforms/telegram_polling.py @@ -6,6 +6,7 @@ Long-polling implementation for Telegram Bot API. import asyncio import logging import time +from typing import Optional from ..config import ConnectorConfig from ..contract import CommandRequest, CommandResponse @@ -62,12 +63,28 @@ class TelegramPolling: async with self.session.get(url, params=params) as resp: if resp.status != 200: - logger.error(f"Telegram API Error {resp.status}") + # IMPORTANT (debuggability): Telegram frequently returns actionable details in the body + # for non-200 responses (e.g. 409 conflict: "terminated by other getUpdates request", + # or "webhook is active"). Log the response text in debug mode to speed up diagnosis. + try: + body = await resp.text() + except Exception: + body = "" + if body: + logger.error(f"Telegram API Error {resp.status}: {body}") + else: + logger.error(f"Telegram API Error {resp.status}") await asyncio.sleep(5) return data = await resp.json() if not data.get("ok"): + # Telegram sometimes returns `ok=false` with a useful description even on 200. + # Keep logs concise, but include enough context to fix config issues quickly. + desc = ( + data.get("description") or data.get("error_code") or "unknown_error" + ) + logger.warning(f"Telegram API returned ok=false: {desc}") return updates = data.get("result", []) @@ -97,12 +114,10 @@ class TelegramPolling: if chat_id in self.config.telegram_allowed_chats: is_allowed = True - if not is_allowed: - if self.config.debug: - logger.debug( - f"Ignored Telegram message from unauthorized user={user_id} chat={chat_id}" - ) - return + if not is_allowed and self.config.debug: + logger.debug( + f"Untrusted Telegram message user={user_id} chat={chat_id} (will require approval)" + ) # Build Request req = CommandRequest( @@ -139,3 +154,44 @@ 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): + """Send photo via Telegram sendPhoto.""" + if not self.session: + return + + import aiohttp # Lazy import safe here as we have session + + url = f"{self.base_url}/sendPhoto" + data = aiohttp.FormData() + 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: + err = await resp.text() + logger.error(f"Telegram send_image failed: {resp.status} {err}") + except Exception as e: + logger.error(f"Telegram send_image error: {e}") + + async def send_message(self, channel_id: str, text: str): + """Send text message.""" + if not self.session: + return + + # Reuse internal logic logic but public + # Using simplified direct call + url = f"{self.base_url}/sendMessage" + payload = {"chat_id": channel_id, "text": text} + try: + async with self.session.post(url, json=payload) as r: + if r.status != 200: + err = await r.text() + logger.error(f"Telegram send_message failed: {r.status} {err}") + except Exception as e: + logger.error(f"Telegram send_message error: {e}") + diff --git a/connector/results_poller.py b/connector/results_poller.py new file mode 100644 index 0000000..925ca4d --- /dev/null +++ b/connector/results_poller.py @@ -0,0 +1,161 @@ +import asyncio +import logging +import time +from typing import Dict, Optional + +from .config import ConnectorConfig +from .contract import Platform +from .openclaw_client import OpenClawClient + +logger = logging.getLogger(__name__) + + +class ResultsPoller: + """ + Polls ComfyUI history for completed jobs and triggers delivery. + """ + + def __init__( + self, + config: ConnectorConfig, + client: OpenClawClient, + platforms: Dict[str, Platform], + ): + 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.active_polls = {} # prompt_id -> task + self.active_polls = {} # prompt_id -> task + + async def start(self): + """Start the main queue consumer.""" + logger.info("ResultsPoller started.") + while True: + item = await self.queue.get() + try: + prompt_id, platform_name, channel_id, sender_id = item + # Spawn a background poll for this job + task = asyncio.create_task( + 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)) + finally: + self.queue.task_done() + + async def stop(self): + """Graceful shutdown.""" + 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): + """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)) + + async def _poll_job( + self, prompt_id: str, platform_name: str, channel_id: str, sender_id: str + ): + """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: + hist = await self.client.get_history(prompt_id) + if hist.get("ok"): + # ComfyUI /history/{prompt_id} -> { "prompt_id": { ... } } + data = hist.get("data", {}) + if prompt_id in data: + job_data = data[prompt_id] + await self._deliver_results( + prompt_id, job_data, platform_name, channel_id + ) + return + except Exception as e: + logger.debug(f"Poll check failed for {prompt_id}: {e}") + + # Backoff + try: + await asyncio.sleep(delay) + except asyncio.CancelledError: + raise + 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.") + + async def _deliver_results( + self, prompt_id: str, job_data: dict, platform_name: str, channel_id: str + ): + """Download images and send to platform.""" + 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).") + 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 + + images_to_send = images_to_send[: self.config.delivery_max_images] + logger.info(f"Delivering {len(images_to_send)} images for {prompt_id}") + + platform = self.platforms.get(platform_name) + if not platform: + logger.error(f"Platform {platform_name} not loaded.") + return + + for img_info in images_to_send: + 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).") + continue + + # Send with error handling + try: + await platform.send_image(channel_id, content, filename=filename) + 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}") + + async def _send_text(self, platform_name: str, channel_id: str, text: str): + platform = self.platforms.get(platform_name) + if platform: + try: + await platform.send_message(channel_id, text) + except Exception as e: + logger.error(f"Failed to send text to {platform_name}: {e}") diff --git a/connector/router.py b/connector/router.py index 71f3d7a..23306a3 100644 --- a/connector/router.py +++ b/connector/router.py @@ -12,13 +12,17 @@ from .contract import CommandRequest, CommandResponse from .openclaw_client import OpenClawClient from .state import ConnectorState +if False: # Type hinting only + from .results_poller import ResultsPoller + logger = logging.getLogger(__name__) class CommandRouter: - def __init__(self, config: ConnectorConfig, client: OpenClawClient): + def __init__(self, config: ConnectorConfig, client: OpenClawClient, poller: "ResultsPoller" = None): self.config = config self.client = client + self.poller = poller self.state = ConnectorState(path=self.config.state_path) async def handle(self, req: CommandRequest) -> CommandResponse: @@ -42,7 +46,7 @@ class CommandRouter: handlers = { ("/status", "status"): (self._handle_status, False), ("/help", "help", "/start"): (self._handle_help, False), - ("/run", "run"): (self._handle_run, True), + ("/run", "run"): (self._handle_run, False), ("/interrupt", "interrupt", "/cancel", "cancel", "/stop"): ( self._handle_interrupt, True, @@ -90,6 +94,50 @@ class CommandRouter: def _is_admin(self, user_id: str) -> bool: return str(user_id) in self.config.admin_users + def _is_trusted(self, req: CommandRequest) -> bool: + """ + Trusted users can execute /run immediately. + Untrusted users are routed to approval flow. + """ + if self._is_admin(req.sender_id): + return True + + platform = (req.platform or "").lower() + sender_id = str(req.sender_id) + channel_id = str(req.channel_id) + + if platform == "telegram": + try: + uid = int(sender_id) + except Exception: + uid = None + try: + cid = int(channel_id) + except Exception: + cid = None + if uid is not None and uid in self.config.telegram_allowed_users: + return True + if cid is not None and cid in self.config.telegram_allowed_chats: + return True + return False + + if platform == "discord": + if sender_id in self.config.discord_allowed_users: + return True + if channel_id in self.config.discord_allowed_channels: + return True + return False + + if platform == "line": + if sender_id in self.config.line_allowed_users: + return True + if channel_id in self.config.line_allowed_groups: + return True + return False + + # Unknown platform: trust only admins + return False + # --- Handlers --- async def _handle_status( @@ -130,11 +178,11 @@ class CommandRouter: ) # Parse flags - require_approval = False + explicit_approval = False clean_args = [] for arg in args: if arg in ("--require-approval", "--approval", "-a"): - require_approval = True + explicit_approval = True else: clean_args.append(arg) @@ -148,6 +196,9 @@ class CommandRouter: k, v = arg.split("=", 1) inputs[k.strip()] = v.strip() + trusted = self._is_trusted(req) + require_approval = explicit_approval or (not trusted) + res = await self.client.submit_job( template_id, inputs, require_approval=require_approval ) @@ -163,6 +214,9 @@ class CommandRouter: return CommandResponse(text=msg) 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) + return CommandResponse( text=f"[Job Submitted]\nID: {prompt_id}\nTemplate: {template_id}\nTrace: {trace_id}" ) @@ -232,7 +286,12 @@ class CommandRouter: # Phase 4: Show execution result if "prompt_id" in data: - msg += f"\nExecuted: {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 + # 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: msg += "\n(Not Executed)" if err := data.get("execution_error"): @@ -297,7 +356,7 @@ class CommandRouter: text=( "OpenClaw Connector\n" "/status - Check system health and queue\n" - "/run