feat: align LINE connector Phase 3 behavior with docs (strict creds, graceful shutdown, 429 handling)

.pre-commit-config.yaml

@@ -14,6 +14,16 @@ exclude: |
   )$
 
 repos:
+  # Prevent committing internal-only docs/plans
+  - repo: local
+    hooks:
+      - id: block-sensitive-files
+        name: block sensitive files (staged)
+        entry: python scripts/precommit_block_sensitive_files.py
+        language: python
+        pass_filenames: false
+        always_run: true
+
   # Secret detection
   - repo: https://github.com/Yelp/detect-secrets
     rev: v1.4.0

README.md

@@ -305,6 +305,17 @@ python3 -m unittest discover -s tests -p "test_*.py"
 - Git install: `git pull` inside `custom_nodes/comfyui-openclaw/`, then restart ComfyUI.
 - ComfyUI-Manager install: update from Manager UI, then restart ComfyUI.
 
+## 🎮 Remote Control (Connector)
+
+OpenClaw includes a standalone **Connector** process that allows you to control your local instance securely via **Telegram** or **Discord** without exposing it to the public internet.
+
+- **Status & Queue**: Check job progress remotely.
+- **Run Jobs**: Submit templates via chat commands.
+- **Approvals**: Approve/Reject paused workflows from your phone.
+- **Secure**: Outbound-only connection; no inbound ports required.
+
+[👉 **See Setup Guide (docs/connector.md)**](docs/connector.md)
+
 ## Security
 
 Read `SECURITY.md` before exposing any endpoint beyond localhost. The project is designed to be secure-by-default (deny-by-default auth, SSRF protections, redaction, bounded outputs), but unsafe deployment can still create risk.

connector/__init__.py

@@ -0,0 +1,5 @@
+"""
+OpenClaw System Connector (F29).
+Standalone process for ChatOps (Telegram/Discord).
+"""
+__version__ = "0.1.0"

connector/__main__.py

@@ -0,0 +1,99 @@
+"""
+Connector Entrypoint (F29).
+Runs the connector process properly.
+"""
+import asyncio
+import logging
+import sys
+from .config import load_config
+from .openclaw_client import OpenClawClient
+from .router import CommandRouter
+from .platforms.telegram_polling import TelegramPolling
+from .platforms.discord_gateway import DiscordGateway
+from .platforms.line_webhook import LINEWebhookServer
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="[%(asctime)s] %(name)s %(levelname)s: %(message)s",
+    handlers=[logging.StreamHandler(sys.stdout)]
+)
+logger = logging.getLogger("connector")
+
+async def main():
+    logger.info("Initializing OpenClaw Connector (Phase 3)...")
+    
+    # 1. Config
+    try:
+        config = load_config()
+    except Exception as e:
+        logger.critical(f"Config load failed: {e}")
+        return
+
+    if config.debug:
+        logger.setLevel(logging.DEBUG)
+        logging.getLogger("connector").setLevel(logging.DEBUG)
+        logger.debug("Debug mode enabled")
+
+    # 2. Components
+    client = OpenClawClient(config)
+    await client.start() # Start session
+    
+    router = CommandRouter(config, client)
+    
+    tasks = []
+    line_server = None
+
+    # 3. Platforms
+    if config.telegram_bot_token:
+        tg = TelegramPolling(config, router)
+        tasks.append(asyncio.create_task(tg.start()))
+    else:
+        logger.info("Telegram not configured (OPENCLAW_CONNECTOR_TELEGRAM_TOKEN missing)")
+
+    if config.discord_bot_token:
+        dc = DiscordGateway(config, router)
+        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)
+        await line_server.start()
+        # If only LINE is active, tasks will be empty. Add a sleeper to keep loop alive.
+        if not tasks:
+            tasks.append(asyncio.create_task(asyncio.sleep(3600*24*365))) # Sleep forever
+    elif config.line_channel_secret:
+         logger.warning("LINE configured but Access Token missing. Skipping.")
+    else:
+        logger.info("LINE not configured (OPENCLAW_CONNECTOR_LINE_CHANNEL_SECRET missing)")
+
+    if not tasks and not line_server:
+        logger.error("No platforms configured! Set TELEGRAM_TOKEN, DISCORD_TOKEN, or LINE_SECRET.")
+        await client.close()
+        return
+
+    # 4. Run Check
+    logger.info(f"Connecting to ComfyUI at {config.openclaw_url}...")
+    health = await client.get_health()
+    if health.get("ok"):
+        logger.info("✅ ComfyUI connection verified.")
+    else:
+        logger.warning(f"⚠️ Could not reach ComfyUI on startup: {health.get('error')}")
+
+    # 5. Wait
+    try:
+        await asyncio.gather(*tasks)
+    except asyncio.CancelledError:
+        logger.info("Connector stopping...")
+    finally:
+        if line_server:
+            await line_server.stop()
+        await client.close()
+        logger.info("Connector stopped.")
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass

connector/config.py

@@ -0,0 +1,83 @@
+"""
+Connector Configuration (F29).
+Loads environment variables and validates allowlists.
+"""
+import os
+import sys
+from dataclasses import dataclass, field
+from typing import List, Optional
+
+@dataclass
+class ConnectorConfig:
+    # OpenClaw Connection
+    openclaw_url: str = "http://127.0.0.1:8188"
+    admin_token: Optional[str] = None  # To call admin endpoints
+
+    # Telegram
+    telegram_bot_token: Optional[str] = None
+    telegram_allowed_users: List[int] = field(default_factory=list)
+    telegram_allowed_chats: List[int] = field(default_factory=list)
+
+    # Discord
+    discord_bot_token: Optional[str] = None
+    discord_allowed_users: List[str] = field(default_factory=list)
+    discord_allowed_channels: List[str] = field(default_factory=list)
+
+    # LINE
+    line_channel_secret: Optional[str] = None
+    line_channel_access_token: Optional[str] = None
+    line_allowed_users: List[str] = field(default_factory=list)
+    line_allowed_groups: List[str] = field(default_factory=list)
+    line_bind_host: str = "127.0.0.1"
+    line_bind_port: int = 8099
+    line_webhook_path: str = "/line/webhook"
+
+    # Privileged Access (ID match across platforms; Telegram Int vs Discord Str handled by router)
+    admin_users: List[str] = field(default_factory=list)
+
+    # Global
+    debug: bool = False
+    state_path: Optional[str] = None
+
+def load_config() -> ConnectorConfig:
+    """Load configuration from environment variables."""
+    cfg = ConnectorConfig()
+
+    cfg.openclaw_url = os.environ.get("OPENCLAW_CONNECTOR_URL", "http://127.0.0.1:8188").rstrip("/")
+    cfg.admin_token = os.environ.get("OPENCLAW_CONNECTOR_ADMIN_TOKEN")
+    cfg.debug = os.environ.get("OPENCLAW_CONNECTOR_DEBUG", "0") == "1"
+    cfg.state_path = os.environ.get("OPENCLAW_CONNECTOR_STATE_PATH")
+
+    # Telegram
+    cfg.telegram_bot_token = os.environ.get("OPENCLAW_CONNECTOR_TELEGRAM_TOKEN")
+    if t_users := os.environ.get("OPENCLAW_CONNECTOR_TELEGRAM_ALLOWED_USERS"):
+        cfg.telegram_allowed_users = [int(u.strip()) for u in t_users.split(",") if u.strip().isdigit()]
+    if t_chats := os.environ.get("OPENCLAW_CONNECTOR_TELEGRAM_ALLOWED_CHATS"):
+        cfg.telegram_allowed_chats = [int(u.strip()) for u in t_chats.split(",") if u.strip().lstrip("-").isdigit()]
+
+    # Discord
+    cfg.discord_bot_token = os.environ.get("OPENCLAW_CONNECTOR_DISCORD_TOKEN")
+    if d_users := os.environ.get("OPENCLAW_CONNECTOR_DISCORD_ALLOWED_USERS"):
+        cfg.discord_allowed_users = [u.strip() for u in d_users.split(",") if u.strip()]
+    if d_chans := os.environ.get("OPENCLAW_CONNECTOR_DISCORD_ALLOWED_CHANNELS"):
+        cfg.discord_allowed_channels = [u.strip() for u in d_chans.split(",") if u.strip()]
+
+    # LINE
+    cfg.line_channel_secret = os.environ.get("OPENCLAW_CONNECTOR_LINE_CHANNEL_SECRET")
+    cfg.line_channel_access_token = os.environ.get("OPENCLAW_CONNECTOR_LINE_CHANNEL_ACCESS_TOKEN")
+    if l_users := os.environ.get("OPENCLAW_CONNECTOR_LINE_ALLOWED_USERS"):
+        cfg.line_allowed_users = [u.strip() for u in l_users.split(",") if u.strip()]
+    if l_groups := os.environ.get("OPENCLAW_CONNECTOR_LINE_ALLOWED_GROUPS"):
+        cfg.line_allowed_groups = [u.strip() for u in l_groups.split(",") if u.strip()]
+    
+    cfg.line_bind_host = os.environ.get("OPENCLAW_CONNECTOR_LINE_BIND", "127.0.0.1")
+    if l_port := os.environ.get("OPENCLAW_CONNECTOR_LINE_PORT"):
+        if l_port.isdigit():
+            cfg.line_bind_port = int(l_port)
+    cfg.line_webhook_path = os.environ.get("OPENCLAW_CONNECTOR_LINE_PATH", "/line/webhook")
+
+    # Admin
+    if admins := os.environ.get("OPENCLAW_CONNECTOR_ADMIN_USERS"):
+        cfg.admin_users = [u.strip() for u in admins.split(",") if u.strip()]
+
+    return cfg

connector/contract.py

@@ -0,0 +1,22 @@
+"""
+Connector Contract (F29).
+Shared data models for request/response.
+"""
+from dataclasses import dataclass, field
+from typing import List, Optional
+
+@dataclass
+class CommandRequest:
+    platform: str  # "telegram" | "discord"
+    sender_id: str
+    channel_id: str
+    username: str
+    message_id: str
+    text: str
+    timestamp: float
+
+@dataclass
+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

connector/openclaw_client.py

@@ -0,0 +1,151 @@
+"""
+OpenClaw API Client (F29 Remediation).
+Handles communication with the local ComfyUI instance.
+"""
+import aiohttp
+import logging
+import json
+import uuid
+from typing import Optional
+from .config import ConnectorConfig
+
+logger = logging.getLogger(__name__)
+
+class OpenClawClient:
+    def __init__(self, config: ConnectorConfig):
+        self.base_url = config.openclaw_url
+        self.headers = {
+            "User-Agent": "OpenClaw-Connector/0.1.0",
+        }
+        if config.admin_token:
+            self.headers["X-OpenClaw-Admin-Token"] = config.admin_token
+        
+        self.session: Optional[aiohttp.ClientSession] = None
+
+    async def start(self):
+        """Initialize shared session."""
+        if not self.session:
+            self.session = aiohttp.ClientSession()
+
+    async def close(self):
+        """Close shared session."""
+        if self.session:
+            await self.session.close()
+
+    async def _request(self, method: str, path: str, json_data: dict = None) -> dict:
+        url = f"{self.base_url}{path}"
+        session = self.session
+        
+        # Fallback if start() wasn't called (e.g. tests)
+        local_session = False
+        if not session:
+            session = aiohttp.ClientSession()
+            local_session = True
+
+        try:
+            async with session.request(method, url, headers=self.headers, json=json_data, timeout=10) as resp:
+                result = {"ok": resp.status in (200, 201, 202)}
+                
+                try:
+                    data = await resp.json()
+                    result["data"] = data
+                    if not result["ok"]:
+                        result["error"] = data.get("error") or data.get("message") or f"HTTP {resp.status}"
+                except:
+                    result["data"] = {}
+                    if not result["ok"]:
+                        try:
+                            result["error"] = f"HTTP {resp.status}: {await resp.text()}"
+                        except:
+                            result["error"] = f"HTTP {resp.status}"
+                
+                return result
+        except Exception as e:
+            logger.error(f"Request failed {method} {path}: {e}")
+            return {"ok": False, "error": str(e)}
+        finally:
+            if local_session:
+                await session.close()
+
+    # --- Observability ---
+
+    async def get_health(self) -> dict:
+        res = await self._request("GET", "/openclaw/health")
+        # Health endpoint might return nested structure, but we just want the wrapper
+        return res
+
+    async def get_prompt_queue(self) -> dict:
+        res = await self._request("GET", "/api/prompt")
+        # Standard ComfyUI api/prompt returns {exec_info: ...} on success
+        if res.get("ok"):
+             # Normalize ComfyUI response to standard wrapper if needed, 
+             # but our wrapper put body in 'data'.
+             # Return just data to be combatible with standard expectations or keep wrapper?
+             # Wrapper is {"ok": true, "data": {...}}
+             return res
+        return res
+
+    async def get_history(self, prompt_id: str) -> dict:
+        return await self._request("GET", f"/history/{prompt_id}")
+
+    async def get_trace(self, prompt_id: str) -> dict:
+        # F29 Phase 3 Introspection
+        # Admin-only typically, gives detailed execution trace/logs for a job
+        return await self._request("GET", f"/openclaw/trace/{prompt_id}")
+
+    async def get_jobs(self) -> dict:
+        # F29 Phase 3 Introspection
+        # Returns active jobs / queue summary
+        return await self._request("GET", f"/openclaw/jobs")
+
+    # --- Execution ---
+
+    async def submit_job(self, template_id: str, inputs: dict) -> dict:
+        # Remediation: Use /triggers/fire with admin token
+        data = {
+            "template_id": template_id,
+            "inputs": inputs,
+            "trace_id": str(uuid.uuid4()),
+            "require_approval": False
+        }
+        return await self._request("POST", "/openclaw/triggers/fire", data)
+
+    async def interrupt_output(self) -> dict:
+        # Remediation: Cancel -> Interrupt (Global)
+        return await self._request("POST", "/api/interrupt", {})
+
+    # --- Approvals ---
+
+    async def get_approvals(self) -> dict:
+        # Remediation: Correct endpoint and shape
+        res = await self._request("GET", "/openclaw/approvals?status=pending")
+        if res.get("ok"):
+             # Flatten: backend wrapper {"approvals": [], ...} is inside res["data"]
+             data = res.get("data", {})
+             if "approvals" in data:
+                  return {
+                      "ok": True,
+                      "items": data.get("approvals", []),
+                      "count": data.get("count"),
+                      "pending_count": data.get("pending_count"),
+                  }
+        return res
+
+    async def approve_request(self, approval_id: str) -> dict:
+        return await self._request("POST", f"/openclaw/approvals/{approval_id}/approve")
+
+    async def reject_request(self, approval_id: str, reason: str = "") -> dict:
+        return await self._request("POST", f"/openclaw/approvals/{approval_id}/reject", {"reason": reason})
+
+    # --- Schedules ---
+
+    async def get_schedules(self) -> dict:
+        res = await self._request("GET", "/openclaw/schedules")
+        if res.get("ok"):
+             # Backend returns {"schedules": [...]}
+             data = res.get("data", {})
+             return {"ok": True, "schedules": data.get("schedules", [])}
+        return res
+
+    async def run_schedule(self, schedule_id: str) -> dict:
+        return await self._request("POST", f"/openclaw/schedules/{schedule_id}/run")

connector/platforms/discord_gateway.py

@@ -0,0 +1,176 @@
+"""
+Discord Gateway Platform (F29 Remediation).
+WebSocket connection to Discord Gateway (simplified) with Rate Limit Handling.
+"""
+import aiohttp
+import asyncio
+import json
+import logging
+import time
+from ..config import ConnectorConfig
+from ..contract import CommandRequest, CommandResponse
+from ..router import CommandRouter
+
+logger = logging.getLogger(__name__)
+
+class DiscordGateway:
+    GATEWAY_URL = "wss://gateway.discord.gg/?v=10&encoding=json"
+    
+    def __init__(self, config: ConnectorConfig, router: CommandRouter):
+        self.config = config
+        self.router = router
+        self.token = config.discord_bot_token
+        self.session = None
+        self.ws = None
+        self.heartbeat_interval = 41.25
+        self._seq = None
+        self._user_id = None
+
+    async def start(self):
+        if not self.token:
+            logger.warning("Discord token not configured. Skipping.")
+            return
+
+        logger.info("Starting Discord Gateway...")
+        async with aiohttp.ClientSession() as self.session:
+            while True:
+                try:
+                    await self._connect()
+                except asyncio.CancelledError:
+                    break
+                except Exception as e:
+                    logger.error(f"Discord gateway error: {e}")
+                    await asyncio.sleep(5)
+
+    async def _connect(self):
+        async with self.session.ws_connect(self.GATEWAY_URL) as ws:
+            self.ws = ws
+            heartbeat_task = asyncio.create_task(self._heartbeat_loop())
+            
+            try:
+                await self._send_identify()
+
+                async for msg in ws:
+                    if msg.type == aiohttp.WSMsgType.TEXT:
+                        data = json.loads(msg.data)
+                        self._seq = data.get("s")
+                        op = data.get("op")
+                        t = data.get("t")
+
+                        if op == 10: # Hello
+                            self.heartbeat_interval = data["d"]["heartbeat_interval"] / 1000
+                        elif op == 11: # Heartbeat ACK
+                            pass
+                        elif op == 0: # Dispatch
+                            if t == "READY":
+                                self._user_id = data["d"]["user"]["id"]
+                                logger.info(f"Discord Connected as {data['d']['user']['username']}")
+                            elif t == "MESSAGE_CREATE":
+                                await self._process_message(data["d"])
+                                
+                    elif msg.type == aiohttp.WSMsgType.ERROR:
+                        break
+            finally:
+                heartbeat_task.cancel()
+
+    async def _heartbeat_loop(self):
+        try:
+            while True:
+                await asyncio.sleep(self.heartbeat_interval)
+                if self.ws and not self.ws.closed:
+                    await self.ws.send_json({"op": 1, "d": self._seq})
+        except asyncio.CancelledError:
+            pass
+
+    async def _send_identify(self):
+        payload = {
+            "op": 2,
+            "d": {
+                "token": self.token,
+                "intents": 33280, 
+                "properties": {
+                    "$os": "linux",
+                    "$browser": "openclaw-connector",
+                    "$device": "openclaw-connector"
+                }
+            }
+        }
+        await self.ws.send_json(payload)
+
+    async def _process_message(self, message: dict):
+        author = message.get("author", {})
+        if author.get("bot"):
+            return 
+        
+        content = message.get("content", "")
+        if not content:
+            return
+
+        user_id = author.get("id")
+        channel_id = message.get("channel_id")
+        
+        # Security Check
+        is_allowed = False
+        if user_id in self.config.discord_allowed_users:
+            is_allowed = True
+        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
+
+        # Build Request
+        req = CommandRequest(
+            platform="discord",
+            sender_id=str(user_id),
+            channel_id=str(channel_id),
+            username=author.get("username", "unknown"),
+            message_id=str(message.get("id")),
+            text=content,
+            timestamp=time.time()
+        )
+
+        try:
+            resp = await self.router.handle(req)
+            await self._send_response(channel_id, resp)
+        except Exception as e:
+            logger.exception(f"Error handling discord command: {e}")
+            await self._send_response(channel_id, CommandResponse(text="⚠️ Internal error"))
+
+    async def _send_response(self, channel_id: str, resp: CommandResponse):
+        url = f"https://discord.com/api/v10/channels/{channel_id}/messages"
+        headers = {
+            "Authorization": f"Bot {self.token}",
+            "Content-Type": "application/json"
+        }
+        
+        # Remediation: Length Limit
+        content = resp.text
+        if len(content) > 1900:
+            content = content[:1900] + "\n...(truncated)"
+            
+        payload = {"content": content}
+        
+        # Remediation: Rate Limit handling
+        retries = 3
+        while retries > 0:
+            async with self.session.post(url, headers=headers, json=payload) as r:
+                if r.status == 429: # Too Many Requests
+                    try:
+                        data = await r.json()
+                        retry_after = data.get("retry_after", 1)
+                        logger.warning(f"Discord 429 Rate Limit. Sleeping {retry_after}s")
+                        await asyncio.sleep(retry_after)
+                        retries -= 1
+                        continue
+                    except:
+                         await asyncio.sleep(1)
+                         retries -= 1
+                         continue
+
+                if r.status not in (200, 201):
+                    logger.error(f"Failed to send Discord msg: {r.status} {await r.text()}")
+                
+                break

connector/platforms/line_webhook.py

@@ -0,0 +1,172 @@
+"""
+LINE Webhook Platform Adapter (F29).
+Receives webhooks from LINE, verifies signature, and routes commands.
+"""
+import aiohttp
+import aiohttp.web
+import base64
+import hashlib
+import hmac
+import json
+import logging
+import time
+from typing import Optional
+from ..config import ConnectorConfig
+from ..contract import CommandRequest, CommandResponse
+from ..router import CommandRouter
+
+logger = logging.getLogger(__name__)
+
+class LINEWebhookServer:
+    def __init__(self, config: ConnectorConfig, router: CommandRouter):
+        self.config = config
+        self.router = router
+        self.app = aiohttp.web.Application()
+        self.app.router.add_post(self.config.line_webhook_path, self.handle_webhook)
+        self.runner = None
+        self.site = None
+
+    async def start(self):
+        """Start the webhook server."""
+        if not self.config.line_channel_secret or not self.config.line_channel_access_token:
+            logger.warning("LINE Channel Secret or Access Token missing. Skipping LINE adapter.")
+            return
+
+        logger.info(f"Starting LINE Webhook on {self.config.line_bind_host}:{self.config.line_bind_port}{self.config.line_webhook_path}")
+        self.session = aiohttp.ClientSession()
+        self.runner = aiohttp.web.AppRunner(self.app)
+        await self.runner.setup()
+        self.site = aiohttp.web.TCPSite(self.runner, self.config.line_bind_host, self.config.line_bind_port)
+        await self.site.start()
+
+    async def stop(self):
+        """Stop the server."""
+        if self.site:
+            await self.site.stop()
+        if self.runner:
+            await self.runner.cleanup()
+        if self.session:
+            await self.session.close()
+
+    async def handle_webhook(self, request: aiohttp.web.Request):
+        # 1. Signature Verification
+        body_bytes = await request.read()
+        body_text = body_bytes.decode('utf-8')
+        signature = request.headers.get("X-Line-Signature", "")
+        
+        if not self._verify_signature(body_bytes, signature):
+            logger.warning("Invalid LINE Signature")
+            return aiohttp.web.Response(status=401, text="Invalid Signature")
+
+        # 2. Parse Event
+        try:
+            payload = json.loads(body_text)
+        except json.JSONDecodeError:
+            return aiohttp.web.Response(status=400, text="Bad JSON")
+
+        events = payload.get("events", [])
+        for event in events:
+            if event.get("type") == "message" and event.get("message", {}).get("type") == "text":
+                await self._process_event(event)
+        
+        return aiohttp.web.Response(text="OK")
+
+    def _verify_signature(self, body: bytes, signature: str) -> bool:
+        """Verify X-Line-Signature using HMAC-SHA256."""
+        if not signature or not self.config.line_channel_secret:
+            return False
+            
+        secret = self.config.line_channel_secret.encode('utf-8')
+        generated = base64.b64encode(
+            hmac.new(secret, body, hashlib.sha256).digest()
+        ).decode('utf-8')
+        
+        return hmac.compare_digest(generated, signature)
+
+    async def _process_event(self, event: dict):
+        """Convert LINE event to CommandRequest and route."""
+        source = event.get("source", {})
+        user_id = source.get("userId")
+        group_id = source.get("groupId")
+        room_id = source.get("roomId") # Remediation: Support RoomId
+        
+        # Identity Logic: 
+        # For LINE, we use user_id as sender_id.
+        # channel_id: if group/room, use that ID; else use userId (DM).
+        channel_id = group_id or room_id or user_id
+        
+        text = event["message"]["text"]
+        reply_token = event.get("replyToken")
+
+        # Security allowlist
+        is_allowed = False
+        # Check User
+        if user_id and user_id in self.config.line_allowed_users:
+            is_allowed = True
+        # Check Group/Room
+        if group_id and group_id in self.config.line_allowed_groups:
+            is_allowed = True
+        if room_id and room_id in self.config.line_allowed_groups: # Treat room as group
+            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}."
+            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)"
+            else:
+               msg += " (Not in allowlist)"
+            logger.warning(msg)
+            return
+
+        req = CommandRequest(
+            platform="line",
+            sender_id=str(user_id),
+            channel_id=str(channel_id),
+            username="line_user", 
+            message_id=event.get("webhookEventId", str(time.time())),
+            text=text,
+            timestamp=event.get("timestamp", 0) / 1000
+        )
+
+        try:
+            resp = await self.router.handle(req)
+            if resp.text:
+                await self._reply_message(reply_token, resp.text)
+        except Exception as e:
+            logger.exception(f"Error handling LINE command: {e}")
+            await self._reply_message(reply_token, "[Internal Error]")
+
+    async def _reply_message(self, reply_token: str, text: str):
+        """Send reply via LINE Messaging API."""
+        if not reply_token or reply_token == "00000000000000000000000000000000": 
+            return
+
+        url = "https://api.line.me/v2/bot/message/reply"
+        headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {self.config.line_channel_access_token}"
+        }
+        
+        if len(text) > 4000:
+             text = text[:4000] + "\n...(truncated)"
+
+        body = {
+            "replyToken": reply_token,
+            "messages": [
+                {
+                    "type": "text",
+                    "text": text
+                }
+            ]
+        }
+        
+        # Remediation: Use persistent session
+        try:
+            async with self.session.post(url, headers=headers, json=body) as resp:
+                if resp.status == 429: # Check Rate Limit first
+                    logger.warning("LINE API Rate Limit Hit")
+                elif resp.status != 200:
+                    logger.error(f"Failed to send LINE reply: {resp.status} {await resp.text()}")
+        except Exception as e:
+            logger.error(f"LINE reply exception: {e}")

connector/platforms/telegram_polling.py

@@ -0,0 +1,120 @@
+"""
+Telegram Polling Platform (F29 Remediation).
+Long-polling implementation for Telegram Bot API.
+"""
+import aiohttp
+import asyncio
+import logging
+import time
+from ..config import ConnectorConfig
+from ..contract import CommandRequest, CommandResponse
+from ..router import CommandRouter
+from ..state import ConnectorState
+
+logger = logging.getLogger(__name__)
+
+class TelegramPolling:
+    def __init__(self, config: ConnectorConfig, router: CommandRouter):
+        self.config = config
+        self.router = router
+        self.state_store = ConnectorState(path=self.config.state_path)
+        self.token = config.telegram_bot_token
+        self.base_url = f"https://api.telegram.org/bot{self.token}"
+        
+        # Remediation: Load offset from persistent state
+        self.offset = self.state_store.get_offset("telegram")
+        self.session = None
+
+    async def start(self):
+        if not self.token:
+            logger.warning("Telegram token not configured. Skipping.")
+            return
+
+        logger.info(f"Starting Telegram Polling (offset={self.offset})...")
+        async with aiohttp.ClientSession() as self.session:
+            while True:
+                try:
+                    await self._poll_once()
+                except asyncio.CancelledError:
+                    break
+                except Exception as e:
+                    logger.error(f"Telegram poll error: {e}")
+                    await asyncio.sleep(5)
+
+    async def _poll_once(self):
+        url = f"{self.base_url}/getUpdates"
+        params = {"offset": self.offset, "timeout": 30}
+        
+        async with self.session.get(url, params=params) as resp:
+            if resp.status != 200:
+                logger.error(f"Telegram API Error {resp.status}")
+                await asyncio.sleep(5)
+                return
+
+            data = await resp.json()
+            if not data.get("ok"):
+                return
+
+            updates = data.get("result", [])
+            for update in updates:
+                next_offset = update["update_id"] + 1
+                if next_offset > self.offset:
+                    self.offset = next_offset
+                    # Remediation: Persist offset
+                    self.state_store.set_offset("telegram", self.offset)
+                
+                await self._process_update(update)
+
+    async def _process_update(self, update: dict):
+        message = update.get("message")
+        if not message or "text" not in message:
+            return
+
+        chat_id = message["chat"]["id"]
+        user_id = message["from"]["id"]
+        username = message["from"].get("username", "unknown")
+        text = message["text"]
+
+        # Security Check
+        is_allowed = False
+        if user_id in self.config.telegram_allowed_users:
+            is_allowed = True
+        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
+
+        # Build Request
+        req = CommandRequest(
+            platform="telegram",
+            sender_id=str(user_id),
+            channel_id=str(chat_id),
+            username=username,
+            message_id=str(message["message_id"]),
+            text=text,
+            timestamp=time.time()
+        )
+
+        try:
+            resp = await self.router.handle(req)
+            await self._send_response(chat_id, resp)
+        except Exception as e:
+            logger.exception(f"Error handling command: {e}")
+            await self._send_response(chat_id, CommandResponse(text="[Error] Internal processing error."))
+
+    async def _send_response(self, chat_id: int, resp: CommandResponse):
+        url = f"{self.base_url}/sendMessage"
+        payload = {
+            "chat_id": chat_id,
+            # Remediation: Plain text only, no parse_mode
+            "text": resp.text
+        }
+        try:
+            async with self.session.post(url, json=payload) as r:
+                if r.status != 200:
+                    logger.error(f"Failed to send Telegram response: {r.status} {await r.text()}")
+        except Exception as e:
+            logger.error(f"Telegram send exception: {e}")

connector/router.py

@@ -0,0 +1,309 @@
+"""
+Connector Router (F29 Remediation).
+Dispatches parsed commands to handlers with AST argument parsing.
+"""
+
+import logging
+import shlex
+from typing import Any, Dict, List
+
+from .config import ConnectorConfig
+from .contract import CommandRequest, CommandResponse
+from .openclaw_client import OpenClawClient
+from .state import ConnectorState
+
+logger = logging.getLogger(__name__)
+
+
+class CommandRouter:
+    def __init__(self, config: ConnectorConfig, client: OpenClawClient):
+        self.config = config
+        self.client = client
+        self.state = ConnectorState(path=self.config.state_path)
+
+    async def handle(self, req: CommandRequest) -> CommandResponse:
+        """Main dispatch loop."""
+        text = req.text.strip()
+
+        try:
+            parts = shlex.split(text)
+        except ValueError:
+            return CommandResponse(
+                text="[Error] Parsing command arguments failed (unbalanced quotes?)."
+            )
+
+        if not parts:
+            return CommandResponse(text="Empty command.")
+
+        cmd = parts[0].lower()
+        args = parts[1:]
+
+        # Dispatch Table
+        handlers = {
+            ("/status", "status"): (self._handle_status, False),
+            ("/help", "help", "/start"): (self._handle_help, False),
+            ("/run", "run"): (self._handle_run, True),
+            ("/interrupt", "interrupt", "/cancel", "cancel", "/stop"): (
+                self._handle_interrupt,
+                True,
+            ),  # Global interrupt => admin-only.
+            ("/approvals", "approvals"): (self._handle_approvals_list, True),
+            ("/approve", "approve"): (self._handle_approve, True),
+            ("/reject", "reject"): (self._handle_reject, True),
+            ("/schedules", "schedules"): (self._handle_schedules_list, True),
+            ("/schedule", "schedule"): (self._handle_schedule_subcommand, True),
+            # Phase 3 Introspection
+            ("/history", "history"): (self._handle_history, False),
+            ("/trace", "trace"): (self._handle_trace, True), # Admin only
+            ("/jobs", "jobs", "queue"): (self._handle_jobs, False),
+        }
+
+        # Find Handler
+        handler = None
+        requires_admin = False
+
+        for aliases, (func, admin_req) in handlers.items():
+            if cmd in aliases:
+                handler = func
+                requires_admin = admin_req
+                break
+
+        if not handler:
+            return CommandResponse(
+                text=f"Unknown command: {cmd}. Type /help for options."
+            )
+
+        # Admin Check
+        if requires_admin:
+            if not self._is_admin(req.sender_id):
+                return CommandResponse(
+                    text="[Access Denied] This command requires Admin privileges."
+                )
+
+        # Execute
+        try:
+            return await handler(req, args)
+        except Exception as e:
+            logger.exception(f"Command execution error {cmd}: {e}")
+            return CommandResponse(text=f"[Internal Error] {str(e)}")
+
+    def _is_admin(self, user_id: str) -> bool:
+        return str(user_id) in self.config.admin_users
+
+    # --- Handlers ---
+
+    async def _handle_status(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        health = await self.client.get_health()
+        queue = await self.client.get_prompt_queue()
+
+        # New standardized response handling
+        health_ok = health.get("ok")
+
+        status_icon = "Online" if health_ok else "Offline"
+        details = []
+
+        if health_ok:
+            data = health.get("data", {})
+            stats = data.get("stats", {})
+            details.append(f"Logs: {stats.get('logs_processed', 0)}")
+            details.append(f"Errors: {stats.get('errors_captured', 0)}")
+
+            q_res = queue.get("data", {})
+            q_rem = q_res.get("exec_info", {}).get("queue_remaining", 0)
+            details.append(f"Queue: {q_rem}")
+        else:
+            details.append(f"Error: {health.get('error')}")
+
+        return CommandResponse(
+            text=f"[{status_icon}] System Status\n"
+            + "\n".join(f"- {d}" for d in details)
+        )
+
+    async def _handle_run(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        if not args:
+            return CommandResponse(text="Usage: /run <template_id> [key=value ...]")
+
+        template_id = args[0]
+        inputs = {}
+        for arg in args[1:]:
+            if "=" in arg:
+                k, v = arg.split("=", 1)
+                inputs[k.strip()] = v.strip()
+
+        res = await self.client.submit_job(template_id, inputs)
+        if res.get("ok"):
+            data = res.get("data", {})
+            prompt_id = data.get("prompt_id", "unknown")
+            return CommandResponse(
+                text=f"[Job Submitted]\nID: {prompt_id}\nTemplate: {template_id}"
+            )
+        else:
+            err = res.get("error", "Unknown error")
+            return CommandResponse(text=f"[Submission Failed] Reason: {err}")
+
+    async def _handle_interrupt(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        # Remediation: Global Interrupt
+        res = await self.client.interrupt_output()
+        if res.get("ok"):
+            return CommandResponse(text="[Stop] Global Interrupt sent to ComfyUI.")
+        else:
+            return CommandResponse(text=f"[Stop Failed] {res.get('error')}")
+
+    async def _handle_approvals_list(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        res = await self.client.get_approvals()
+        if not res.get("ok"):
+            return CommandResponse(
+                text=f"[Error] Failed to list approvals: {res.get('error')}"
+            )
+
+        items = res.get("items", [])
+        if not items:
+            return CommandResponse(text="No pending approvals.")
+
+        pending_count = res.get("pending_count")
+        lines = []
+        for i in items:
+            # IMPORTANT (stability): the backend approval schema uses:
+            # `approval_id`, `template_id`, `status`, `requested_by`, `source`.
+            # Do not “simplify” these keys to `id/description/requester` unless you also
+            # update the backend API + all tests. This mismatch previously caused silent
+            # bad output and brittle regressions.
+            approval_id = i.get("approval_id") or i.get("id") or "unknown"
+            template_id = i.get("template_id") or "unknown"
+            status = i.get("status") or "unknown"
+            requested_by = i.get("requested_by") or "unknown"
+            source = i.get("source") or "unknown"
+
+            lines.append(
+                f"- {approval_id} [{status}] template={template_id} by={requested_by} source={source}"
+            )
+
+        header = "Pending Approvals"
+        if isinstance(pending_count, int):
+            header += f" ({pending_count})"
+        return CommandResponse(text=header + ":\n" + "\n".join(lines))
+
+    async def _handle_approve(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        if not args:
+            return CommandResponse(text="Usage: /approve <id>")
+
+        res = await self.client.approve_request(args[0])
+        if not res.get("ok"):
+            return CommandResponse(text=f"[Failed] {res.get('error')}")
+
+        return CommandResponse(text=f"[Approved] {args[0]}")
+
+    async def _handle_reject(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        if not args:
+            return CommandResponse(text="Usage: /reject <id> [reason]")
+
+        reason = " ".join(args[1:]) if len(args) > 1 else "Rejected via chat"
+        res = await self.client.reject_request(args[0], reason)
+        if not res.get("ok"):
+            return CommandResponse(text=f"[Failed] {res.get('error')}")
+
+        return CommandResponse(text=f"[Rejected] {args[0]}")
+
+    async def _handle_schedules_list(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        res = await self.client.get_schedules()
+        if not res.get("ok"):
+            return CommandResponse(text=f"[Error] {res.get('error')}")
+
+        scheds = res.get("schedules", [])
+        if not scheds:
+            return CommandResponse(text="No schedules found.")
+
+        lines = []
+        for s in scheds:
+            status = "+" if s.get("enabled") else "-"
+            lines.append(
+                f"[{status}] {s.get('id')}: {s.get('cron')} - {s.get('template_id')}"
+            )
+
+        return CommandResponse(text="Schedules:\n" + "\n".join(lines))
+
+    async def _handle_schedule_subcommand(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        if len(args) < 2:
+            return CommandResponse(text="Usage: /schedule <run|toggle> <id>")
+
+        sub = args[0].lower()
+        sid = args[1]
+
+        if sub == "run":
+            res = await self.client.run_schedule(sid)
+            if not res.get("ok"):
+                return CommandResponse(text=f"[Error] {res.get('error')}")
+            return CommandResponse(text=f"[Success] Schedule {sid} triggered manually.")
+        else:
+            return CommandResponse(text="Not implemented yet.")
+
+    async def _handle_help(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        return CommandResponse(
+            text=(
+                "OpenClaw Connector\n"
+                "/status - Check system health and queue\n"
+                "/run <template> [k=v] - Run a generation (Admin)\n"
+                "/stop - Global Interrupt (Admin)\n"
+                "/history <id> - Job details\n"
+                "/jobs - Queue summary\n"
+                "Admin Only:\n"
+                "/approvals - List pending approvals\n"
+                "/approve <id>, /reject <id>\n"
+                "/schedules, /schedule run <id>\n"
+                "/trace <id> - Execution trace"
+            )
+        )
+
+    async def _handle_history(self, req: CommandRequest, args: List[str]) -> CommandResponse:
+        if not args: return CommandResponse(text="Usage: /history <prompt_id>")
+        res = await self.client.get_history(args[0])
+        if not res.get("ok"):
+            return CommandResponse(text=f"[Error] {res.get('error')}")
+        
+        # Simple format
+        data = res.get("data", {})
+        status = data.get("status", {}).get("status_str", "unknown")
+        # Assuming backend returns a structure we can summarise
+        return CommandResponse(text=f"Job {args[0]}: {status}\nFull details: not implemented in connector view yet.")
+
+    async def _handle_trace(self, req: CommandRequest, args: List[str]) -> CommandResponse:
+        if not args: return CommandResponse(text="Usage: /trace <prompt_id>")
+        res = await self.client.get_trace(args[0])
+        if not res.get("ok"):
+            return CommandResponse(text=f"[Error] {res.get('error')}")
+        
+        # Dump trace 
+        return CommandResponse(text=f"Trace {args[0]}: {str(res.get('data'))[:1000]}...")
+
+    async def _handle_jobs(self, req: CommandRequest, args: List[str]) -> CommandResponse:
+        # Try native /openclaw/jobs first
+        res = await self.client.get_jobs()
+        if res.get("ok"):
+             # Format nice summary
+             return CommandResponse(text=f"Default Jobs View: {res.get('data')}")
+        
+        # Fallback: Queue
+        q = await self.client.get_prompt_queue()
+        if q.get("ok"):
+             rem = q.get("data", {}).get("exec_info", {}).get("queue_remaining", "?")
+             return CommandResponse(text=f"[Fallback] Queue Remaining: {rem}")
+        
+        return CommandResponse(text="[Error] Could not fetch jobs or queue.")

connector/state.py

@@ -0,0 +1,53 @@
+"""
+Connector State Management (F29 Phase 2).
+Simple JSON persistence for offsets and cancel markers.
+"""
+import json
+import logging
+import os
+from typing import Dict, List, Set
+
+logger = logging.getLogger(__name__)
+
+STATE_FILE = "connector_state.json"
+
+class ConnectorState:
+    def __init__(self, path: str = None):
+        self.path = path or STATE_FILE
+        self.data: Dict = {}
+        self.cancelled_prompts: Set[str] = set()
+        self._load()
+
+    def _load(self):
+        if os.path.exists(self.path):
+            try:
+                with open(self.path, "r", encoding="utf-8") as f:
+                    self.data = json.load(f)
+            except Exception as e:
+                logger.error(f"Failed to load state from {self.path}: {e}")
+                self.data = {}
+
+    def save(self):
+        try:
+            tmp_path = f"{self.path}.tmp"
+            with open(tmp_path, "w", encoding="utf-8") as f:
+                json.dump(self.data, f, indent=2)
+            os.replace(tmp_path, self.path)
+        except Exception as e:
+            logger.error(f"Failed to save state to {self.path}: {e}")
+
+    # Offset Management
+    def get_offset(self, platform: str) -> int:
+        return self.data.get(f"{platform}_offset", 0)
+
+    def set_offset(self, platform: str, offset: int):
+        self.data[f"{platform}_offset"] = offset
+        self.save()
+
+    # Cancel Markers (Transient)
+    def mark_cancelled(self, prompt_id: str):
+        self.cancelled_prompts.add(prompt_id)
+    
+    def is_cancelled(self, prompt_id: str) -> bool:
+        return prompt_id in self.cancelled_prompts
+

docs/connector.md

@@ -0,0 +1,121 @@
+# OpenClaw Connector
+
+The **OpenClaw Connector** (`connector`) is a standalone process that allows you to control your local ComfyUI instance remotely via chat platforms like **Telegram** and **Discord**, without exposing your ComfyUI to the public internet.
+
+## How It Works
+
+The connector runs alongside ComfyUI on your machine.
+
+1. It connects outbound to Telegram/Discord (polling/gateway).
+2. It talks to ComfyUI via `localhost`.
+3. It relays commands and status updates securely.
+
+**Security**:
+
+- **Outbound Only**: No inbound ports required.
+- **Allowlist**: Only users/chats you explicitly allow can send commands.
+- **Local Secrets**: Bot tokens are stored in your local environment, never sent to ComfyUI.
+- **Admin Boundary**: Control-plane actions call admin endpoints on the local ComfyUI instance. See `OPENCLAW_CONNECTOR_ADMIN_TOKEN` below.
+
+## Supported Platforms
+
+- **Telegram**: Long-polling (instant response).
+- **Discord**: Gateway WebSocket (instant response).
+
+## Setup
+
+### 1. Requirements
+
+- Python 3.10+
+- `aiohttp` (installed with ComfyUI-OpenClaw)
+
+### 2. Configuration
+
+Set the following environment variables (or put them in a `.env` file if you use a loader):
+
+**Common:**
+
+- `OPENCLAW_CONNECTOR_URL`: URL of your ComfyUI (default: `http://127.0.0.1:8188`)
+- `OPENCLAW_CONNECTOR_DEBUG`: Set to `1` for verbose logs.
+- `OPENCLAW_CONNECTOR_ADMIN_USERS`: Comma-separated list of user IDs allowed to run admin commands (e.g. `/run`, `/stop`, approvals, schedules).
+- `OPENCLAW_CONNECTOR_ADMIN_TOKEN`: Admin token sent to OpenClaw (`X-OpenClaw-Admin-Token`).
+
+**Admin token behavior:**
+
+- If the OpenClaw server has `OPENCLAW_ADMIN_TOKEN` configured, you must set `OPENCLAW_CONNECTOR_ADMIN_TOKEN` to the same value or admin calls will return HTTP 403.
+- If the OpenClaw server is in loopback-only convenience mode (no Admin Token configured), the connector can still call admin endpoints via `localhost` without sending a token.
+
+**Telegram:**
+
+- `OPENCLAW_CONNECTOR_TELEGRAM_TOKEN`: Your Bot Token (from @BotFather).
+- `OPENCLAW_CONNECTOR_TELEGRAM_ALLOWED_USERS`: Comma-separated list of User IDs (e.g. `123456, 789012`).
+- `OPENCLAW_CONNECTOR_TELEGRAM_ALLOWED_CHATS`: Comma-separated list of Chat/Group IDs.
+
+**Discord:**
+
+- `OPENCLAW_CONNECTOR_DISCORD_TOKEN`: Your Bot Token (from Discord Developer Portal).
+- `OPENCLAW_CONNECTOR_DISCORD_ALLOWED_USERS`: Comma-separated User IDs.
+- `OPENCLAW_CONNECTOR_DISCORD_ALLOWED_CHANNELS`: Comma-separated Channel IDs the bot should listen in.
+
+**LINE:**
+
+*(Requires Inbound Connectivity - see below)*
+
+- `OPENCLAW_CONNECTOR_LINE_CHANNEL_SECRET`: LINE Channel Secret.
+- `OPENCLAW_CONNECTOR_LINE_CHANNEL_ACCESS_TOKEN`: LINE Channel Access Token.
+- `OPENCLAW_CONNECTOR_LINE_ALLOWED_USERS`: Comma-separated User IDs (e.g. `U1234...`).
+- `OPENCLAW_CONNECTOR_LINE_ALLOWED_GROUPS`: Comma-separated Group IDs (e.g. `C1234...`).
+- `OPENCLAW_CONNECTOR_LINE_BIND`: Host to bind (default `127.0.0.1`).
+- `OPENCLAW_CONNECTOR_LINE_PORT`: Port (default `8099`).
+- `OPENCLAW_CONNECTOR_LINE_PATH`: Webhook path (default `/line/webhook`).
+
+### 3. Usage
+
+#### Running the Connector
+
+```bash
+python -m connector
+```
+
+#### LINE Webhook Setup
+
+Unlike Telegram/Discord which pull messages, LINE pushes webhooks to your connector.
+Since the connector runs on `localhost` (default port 8099), you must expose it to the internet securely.
+
+**Option A: Cloudflare Tunnel (Recommended)**
+
+1. Install `cloudflared`.
+2. Run: `cloudflared tunnel --url http://127.0.0.1:8099`
+3. Copy the generated URL (e.g. `https://random-name.trycloudflare.com`).
+4. In LINE Developers Console > Messaging API > Webhook settings:
+   - Set URL to `https://<your-tunnel>/line/webhook` (or your custom path).
+   - Enable "Use webhook".
+
+**Option B: Reverse Proxy (Nginx/Caddy)**
+
+- Configure your proxy to forward HTTPS traffic to `127.0.0.1:8099`.
+
+## Commands
+
+**General:**
+
+| Command | Description |
+| :--- | :--- |
+| `/status` | Check ComfyUI system status, logs, and queue size. |
+| `/jobs` | View active jobs and queue summary. |
+| `/history <id>` | View details of a finished job. |
+| `/help` | Show available commands. |
+| `/run <template> [k=v]` | Submit a job (via Triggers API; requires Admin). |
+| `/stop` | **Global Interrupt**: Stop all running generations. |
+
+**Admin Only:**
+*(Requires User ID in `OPENCLAW_CONNECTOR_ADMIN_USERS`)*
+
+| Command | Description |
+| :--- | :--- |
+| `/trace <id>` | View raw execution logs/trace for a job. |
+| `/approvals` | List pending approvals. |
+| `/approve <id>` | Approve a paused workflow. |
+| `/reject <id> [reason]` | Reject a workflow. |
+| `/schedules` | List schedules. |
+| `/schedule run <id>` | Trigger a schedule immediately. |

scripts/precommit_block_sensitive_files.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+import subprocess
+import sys
+
+
+SENSITIVE_PATH_PREFIXES = (
+    ".planning/",
+    ".planning\\",
+)
+
+SENSITIVE_EXACT = {
+    "ROADMAP.md",
+}
+
+
+def _get_staged_paths() -> list[str]:
+    res = subprocess.run(
+        ["git", "diff", "--cached", "--name-only"],
+        check=False,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+    if res.returncode != 0:
+        # If git isn't available for some reason, do not block the commit.
+        # This hook is a safety net, not a hard dependency.
+        return []
+    return [line.strip() for line in res.stdout.splitlines() if line.strip()]
+
+
+def main() -> int:
+    staged = _get_staged_paths()
+    blocked: list[str] = []
+
+    for path in staged:
+        if path in SENSITIVE_EXACT:
+            blocked.append(path)
+            continue
+        if path.startswith(SENSITIVE_PATH_PREFIXES):
+            blocked.append(path)
+            continue
+
+    if not blocked:
+        return 0
+
+    msg = "\n".join(
+        [
+            "[OpenClaw] Commit blocked: sensitive files are staged.",
+            "",
+            "These files must never be committed to the public repo:",
+            *[f"  - {p}" for p in blocked],
+            "",
+            "Fix:",
+            "  git restore --staged ROADMAP.md .planning",
+            "",
+            "If you intentionally need an internal-only commit, use a separate private remote/repo.",
+        ]
+    )
+    print(msg, file=sys.stderr)
+    return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+

services/preflight.py

@@ -9,6 +9,23 @@ import logging
 import time
 from typing import Any, Dict, List, Set, Tuple
 
+logger = logging.getLogger("ComfyUI-OpenClaw.services.preflight")
+
+# IMPORTANT (ComfyUI runtime wiring):
+# This module is imported both:
+# - inside a real ComfyUI runtime (where `nodes` and `folder_paths` exist), and
+# - in unit tests / tooling contexts (where they may not).
+# Keep these imports optional and keep references guarded.
+try:  # pragma: no cover (best-effort ComfyUI imports)
+    import nodes  # type: ignore
+except Exception:  # pragma: no cover
+    nodes = None  # type: ignore
+
+try:  # pragma: no cover (best-effort ComfyUI imports)
+    import folder_paths  # type: ignore
+except Exception:  # pragma: no cover
+    folder_paths = None  # type: ignore
+
 _CACHE = {}
 _CACHE_TTL = 60  # seconds
 

tests/TEST_SOP.md

@@ -18,7 +18,20 @@ pre-commit run detect-secrets --all-files
 ```bash
 pre-commit run --all-files --show-diff-on-failure
 ```
-If any hook reports “files were modified”, stage + commit those changes and re-run until this step is clean.
+**IMPORTANT (must read): pre-commit “modified files” is a failure until committed**
+- Some hooks (e.g. `end-of-file-fixer`, `trailing-whitespace`) intentionally **exit non-zero** when they auto-fix files.
+- CI will fail if those fixes are not committed.
+- Rule: keep re-running step (2) until it reports **no modified files**, and `git status --porcelain` is empty.
+
+Typical loop:
+```bash
+pre-commit run --all-files --show-diff-on-failure
+git status --porcelain
+git diff
+git add -A
+git commit -m "Apply pre-commit autofixes"
+pre-commit run --all-files --show-diff-on-failure
+```
 
 3) Backend unit tests (recommended; CI enforces)
 ```bash
@@ -27,6 +40,18 @@ MOLTBOT_STATE_DIR="$(pwd)/moltbot_state/_local_unit" python -m unittest discover
 
 4) Frontend E2E (Playwright; CI enforces)
 ```bash
+# Ensure you are using Node.js 18+ (CI uses 20).
+node -v
+
+# If you're on WSL and `node -v` is < 18, your shell may be picking up the distro Node
+# (e.g. `/usr/bin/node`) instead of your user-installed Node. If you use `nvm`, do:
+#   source ~/.nvm/nvm.sh
+#   nvm use 18.20.8
+# Then re-check:
+#   node -v
+#
+# IMPORTANT: run `npm install` with the same Node version you use for `npm test`.
+
 # One-time browser install (recommended)
 npx playwright install chromium
 

tests/connector/test_client.py

@@ -0,0 +1,95 @@
+"""
+Unit Tests for OpenClawClient (F29 Remediation Verification).
+Verifies that client constructs correct HTTP requests/payloads.
+"""
+import asyncio
+import unittest
+from unittest.mock import MagicMock, AsyncMock, patch
+
+from connector.config import ConnectorConfig
+from connector.openclaw_client import OpenClawClient
+
+
+class TestOpenClawClient(unittest.TestCase):
+    def setUp(self):
+        self.config = ConnectorConfig()
+        self.config.admin_token = "admin-secret"
+        self.client = OpenClawClient(self.config)
+
+    def _setup_mock_session(self, MockSession, json_response):
+        mock_session = MockSession.return_value
+        mock_session.close = AsyncMock() # Fix await close()
+        
+        # Response Context Manager
+        mock_resp = MagicMock()
+        mock_resp.status = 200
+        mock_resp.json = AsyncMock(return_value=json_response)
+        
+        # Enter returns the response object
+        mock_ctx = MagicMock()
+        mock_ctx.__aenter__.return_value = mock_resp
+        mock_ctx.__aexit__.return_value = None
+        
+        mock_session.request.return_value = mock_ctx
+        return mock_session
+
+    def test_submit_job_payload(self):
+        """Verify submit_job calls /openclaw/triggers/fire with admin token and correct payload."""
+        with patch('aiohttp.ClientSession') as MockSession:
+            mock_session = self._setup_mock_session(MockSession, {"ok": True, "prompt_id": "p-1"})
+            
+            asyncio.run(self.client.submit_job("tmpl-1", {"k": "v"}))
+            
+            # Assert
+            mock_session.request.assert_called_once()
+            args, kwargs = mock_session.request.call_args
+            
+            method, url = args
+            self.assertEqual(method, "POST")
+            self.assertTrue(url.endswith("/openclaw/triggers/fire"))
+            
+            headers = kwargs["headers"]
+            self.assertEqual(headers["X-OpenClaw-Admin-Token"], "admin-secret")
+            
+            data = kwargs["json"]
+            self.assertEqual(data["template_id"], "tmpl-1")
+            self.assertEqual(data["inputs"], {"k": "v"})
+            self.assertFalse(data["require_approval"])
+            self.assertTrue("trace_id" in data)
+
+    def test_interrupt_output(self):
+        """Verify interrupt calls /api/interrupt."""
+        with patch('aiohttp.ClientSession') as MockSession:
+            mock_session = self._setup_mock_session(MockSession, {})
+
+            asyncio.run(self.client.interrupt_output())
+            
+            method, url = mock_session.request.call_args[0]
+            self.assertEqual(method, "POST")
+            self.assertTrue(url.endswith("/api/interrupt"))
+
+    def test_get_approvals_query(self):
+        """Verify get_approvals uses query param and parses nested response."""
+        with patch('aiohttp.ClientSession') as MockSession:
+            backend_resp = {
+                "approvals": [{"approval_id": "apr_1", "template_id": "tmpl_x"}],
+                "count": 1,
+                "pending_count": 1,
+            }
+            mock_session = self._setup_mock_session(MockSession, backend_resp)
+
+            res = asyncio.run(self.client.get_approvals())
+            
+            # Request Check
+            method, url = mock_session.request.call_args[0]
+            self.assertEqual(method, "GET")
+            self.assertTrue("?status=pending" in url)
+            
+            # Response Parsing Check (OpenClawClient.get_approvals logic)
+            self.assertTrue(res["ok"])
+            self.assertEqual(len(res["items"]), 1)
+            self.assertEqual(res["items"][0]["approval_id"], "apr_1")
+            self.assertEqual(res["pending_count"], 1)
+
+if __name__ == "__main__":
+    unittest.main()

tests/connector/test_config.py

@@ -0,0 +1,36 @@
+"""
+Unit Tests for Connector Config (F29).
+"""
+import os
+import unittest
+from unittest.mock import patch
+from connector.config import load_config
+
+class TestConnectorConfig(unittest.TestCase):
+    def test_basic_load(self):
+        with patch.dict(os.environ, {
+            "OPENCLAW_CONNECTOR_URL": "http://localhost:5555"
+        }):
+            cfg = load_config()
+            self.assertEqual(cfg.openclaw_url, "http://localhost:5555")
+
+    def test_telegram_allowlist(self):
+        with patch.dict(os.environ, {
+            "OPENCLAW_CONNECTOR_TELEGRAM_ALLOWED_USERS": "123, 456, abc ",
+            "OPENCLAW_CONNECTOR_TELEGRAM_ALLOWED_CHATS": "-100, 200"
+        }):
+            cfg = load_config()
+            self.assertEqual(cfg.telegram_allowed_users, [123, 456])
+            self.assertEqual(cfg.telegram_allowed_chats, [-100, 200])
+
+    def test_discord_allowlist(self):
+        with patch.dict(os.environ, {
+            "OPENCLAW_CONNECTOR_DISCORD_ALLOWED_USERS": "u1,u2,,",
+            "OPENCLAW_CONNECTOR_DISCORD_ALLOWED_CHANNELS": "c1"
+        }):
+            cfg = load_config()
+            self.assertEqual(cfg.discord_allowed_users, ["u1", "u2"])
+            self.assertEqual(cfg.discord_allowed_channels, ["c1"])
+
+if __name__ == "__main__":
+    unittest.main()

tests/connector/test_line_signature.py

@@ -0,0 +1,35 @@
+"""
+Unit Tests for LINE Signature Verification (F29 Phase 3).
+"""
+import unittest
+import base64
+import hashlib
+import hmac
+from connector.config import ConnectorConfig
+from connector.platforms.line_webhook import LINEWebhookServer
+from unittest.mock import MagicMock
+
+class TestLINESignature(unittest.TestCase):
+    def setUp(self):
+        self.config = ConnectorConfig()
+        self.config.line_channel_secret = "mysecret"
+        self.router = MagicMock()
+        self.server = LINEWebhookServer(self.config, self.router)
+
+    def test_verify_valid_signature(self):
+        body = b'{"events":[]}'
+        secret = b"mysecret"
+        sig = base64.b64encode(hmac.new(secret, body, hashlib.sha256).digest()).decode('utf-8')
+        
+        self.assertTrue(self.server._verify_signature(body, sig))
+
+    def test_verify_invalid_signature(self):
+        body = b'{"events":[]}'
+        sig = "invalid_sig"
+        self.assertFalse(self.server._verify_signature(body, sig))
+
+    def test_verify_empty(self):
+        self.assertFalse(self.server._verify_signature(b"", ""))
+
+if __name__ == "__main__":
+    unittest.main()

tests/connector/test_router_phase2.py

@@ -0,0 +1,96 @@
+"""
+Unit Tests for Connector Router Phase 2 (F29 Remediation).
+"""
+import asyncio
+import unittest
+from unittest.mock import AsyncMock, MagicMock
+
+from connector.config import ConnectorConfig
+from connector.contract import CommandRequest
+from connector.router import CommandRouter
+
+
+class TestCommandRouterPhase2(unittest.TestCase):
+    def setUp(self):
+        self.config = ConnectorConfig()
+        # Admin setup
+        self.config.admin_users = ["999", "admin_user"]
+        
+        self.client = MagicMock()
+        self.client.get_health = AsyncMock(return_value={"ok": True, "data": {"stats": {}}})
+        # Standardized wrapper for queue
+        self.client.get_prompt_queue = AsyncMock(return_value={"ok": True, "data": {"exec_info": {"queue_remaining": 5}}})
+        
+        # New standardized response for submit
+        self.client.submit_job = AsyncMock(return_value={"ok": True, "data": {"prompt_id": "p-123"}})
+        
+        self.client.get_approvals = AsyncMock(
+            return_value={
+                "ok": True,
+                "pending_count": 1,
+                "items": [
+                    {
+                        "approval_id": "apr_1",
+                        "template_id": "tmpl_x",
+                        "status": "pending",
+                        "requested_by": "bob",
+                        "source": "trigger",
+                    }
+                ],
+            }
+        )
+        self.client.interrupt_output = AsyncMock(return_value={"ok": True})
+        
+        self.router = CommandRouter(self.config, self.client)
+
+    def _req(self, text, sender="123"):
+        return CommandRequest(
+            platform="test", sender_id=sender, channel_id="c1", 
+            username="u", message_id="m1", text=text, timestamp=0
+        )
+
+    def test_run_parsing(self):
+        # /run tmpl k=v "quoted"
+        req = self._req('/run my-template prompt="hello world" steps=20', sender="999")
+        resp = asyncio.run(self.router.handle(req))
+        self.assertIn("Job Submitted", resp.text)
+        self.assertIn("p-123", resp.text)
+        
+        # Verify call
+        self.client.submit_job.assert_called_with("my-template", {"prompt": "hello world", "steps": "20"})
+
+    def test_admin_gating_deny(self):
+        # User 123 is not admin
+        req = self._req('/approvals', sender="123")
+        resp = asyncio.run(self.router.handle(req))
+        self.assertIn("Access Denied", resp.text)
+        self.client.get_approvals.assert_not_called()
+
+    def test_admin_gating_allow(self):
+        # User 999 is admin
+        req = self._req('/approvals', sender="999")
+        resp = asyncio.run(self.router.handle(req))
+        self.assertIn("Pending Approvals", resp.text)
+        self.assertIn("apr_1", resp.text)
+        self.client.get_approvals.assert_called_once()
+    
+    def test_interrupt_command(self):
+        # /stop (aliased to interrupt) requires admin
+        req = self._req('/stop', sender="999")
+        resp = asyncio.run(self.router.handle(req))
+        self.assertIn("Global Interrupt sent", resp.text)
+        self.client.interrupt_output.assert_called_once()
+
+        # Deny non-admin
+        req = self._req('/stop', sender="123")
+        resp = asyncio.run(self.router.handle(req))
+        self.assertIn("Access Denied", resp.text)
+
+    def test_complex_quotes(self):
+        # Unbalanced
+        req = self._req('/run "oops')
+        resp = asyncio.run(self.router.handle(req))
+        self.assertIn("unbalanced quotes", resp.text) # or whatever the error message is
+
+if __name__ == "__main__":
+    unittest.main()

tests/connector/test_router_phase3.py

@@ -0,0 +1,60 @@
+"""
+Unit Tests for Connector Router Phase 3 (Introspection).
+"""
+import unittest
+import asyncio
+from unittest.mock import AsyncMock, MagicMock
+from connector.config import ConnectorConfig
+from connector.router import CommandRouter
+from connector.contract import CommandRequest
+
+class TestCommandRouterPhase3(unittest.TestCase):
+    def setUp(self):
+        self.config = ConnectorConfig()
+        self.config.admin_users = ["999"]
+        self.client = MagicMock()
+        self.client.get_health = AsyncMock(return_value={"ok": True})
+        self.client.get_prompt_queue = AsyncMock(return_value={"ok": True, "data": {"exec_info": {"queue_remaining": 0}}})
+        # Mock Phase 3 Methods
+        self.client.get_history = AsyncMock(return_value={"ok": True, "data": {"status": {"status_str": "success"}}})
+        self.client.get_trace = AsyncMock(return_value={"ok": True, "data": "trace logs..."})
+        self.client.get_jobs = AsyncMock(return_value={"ok": True, "data": 5})
+
+        self.router = CommandRouter(self.config, self.client)
+
+    def _req(self, text, sender="123"):
+        return CommandRequest(
+            platform="test", sender_id=sender, channel_id="c1", 
+            username="u", message_id="m1", text=text, timestamp=0
+        )
+
+    def test_history(self):
+        # Public
+        req = self._req('/history p1', sender="123")
+        resp = asyncio.run(self.router.handle(req))
+        self.assertIn("success", resp.text)
+        self.client.get_history.assert_called_with("p1")
+
+    def test_jobs(self):
+        # Public
+        req = self._req('/jobs', sender="123")
+        resp = asyncio.run(self.router.handle(req))
+        self.assertIn("5", resp.text)
+        self.client.get_jobs.assert_called_once()
+
+    def test_trace_admin(self):
+        # Admin allow
+        req = self._req('/trace p1', sender="999")
+        resp = asyncio.run(self.router.handle(req))
+        self.assertIn("trace logs", resp.text)
+        self.client.get_trace.assert_called_with("p1")
+
+    def test_trace_deny(self):
+        # Non-admin deny
+        req = self._req('/trace p1', sender="123")
+        resp = asyncio.run(self.router.handle(req))
+        self.assertIn("Access Denied", resp.text)
+        self.client.get_trace.assert_not_called()
+
+if __name__ == "__main__":
+    unittest.main()

web/tabs/settings_tab.js

@@ -9,7 +9,17 @@ export const settingsTab = {
     id: "settings",
     title: "Settings",
     render: async (container) => {
-        container.innerHTML = "<div>Loading...</div>";
+        // IMPORTANT (UI layout): `.moltbot-content` has `overflow: hidden`.
+        // This tab MUST render its own scroll container (`.moltbot-scroll-area`),
+        // otherwise lower sections (e.g. UI Key Store) will be clipped with no way to scroll.
+        container.innerHTML = `
+            <div class="moltbot-panel">
+                <div class="moltbot-scroll-area" id="openclaw-settings-scroll">
+                    <div>Loading...</div>
+                </div>
+            </div>
+        `;
+        const scroll = container.querySelector("#openclaw-settings-scroll");
 
         const [healthRes, logRes, configRes] = await Promise.all([
             moltbotApi.getHealth(),
@@ -17,7 +27,7 @@ export const settingsTab = {
             moltbotApi.getConfig(),
         ]);
 
-        container.innerHTML = "";
+        scroll.innerHTML = "";
 
         // If everything is 404, backend routes not registered
 
@@ -43,7 +53,7 @@ export const settingsTab = {
                 </ul>
             `;
             warn.appendChild(hint);
-            container.appendChild(warn);
+            scroll.appendChild(warn);
         }
 
         // -- System Health & Diagnostics --
@@ -113,7 +123,7 @@ export const settingsTab = {
                 addRow(healthSec, "Detail", detail);
             }
         }
-        container.appendChild(healthSec);
+        scroll.appendChild(healthSec);
 
         // -- LLM Settings Section --
         const llmSec = createSection("LLM Settings");
@@ -358,7 +368,7 @@ export const settingsTab = {
             ].filter(Boolean).join(" — ");
             addRow(llmSec, "Error", detail);
         }
-        container.appendChild(llmSec);
+        scroll.appendChild(llmSec);
 
         // --- S26: Collapsible Secrets Section (always visible) ---
         if (configRes.ok) {
@@ -529,7 +539,7 @@ export const settingsTab = {
 
             secretsContent.appendChild(secretsBtnRow);
 
-            container.appendChild(secretsSec.container);
+            scroll.appendChild(secretsSec.container);
         }
 
         // -- Logs Section --
@@ -549,7 +559,7 @@ export const settingsTab = {
         }
 
         logsSec.appendChild(logView);
-        container.appendChild(logsSec);
+        scroll.appendChild(logsSec);
     },
 };