diff --git a/connector/router.py b/connector/router.py
index 1624469..4ecf61c 100644
--- a/connector/router.py
+++ b/connector/router.py
@@ -3,35 +3,35 @@ Connector Router (F29 Remediation).
 Dispatches parsed commands to handlers with AST argument parsing.
 """
 
-import logging
-import shlex
-from collections.abc import Mapping
-from typing import Any, Dict, List, Optional
-
-from .config import CommandClass, ConnectorConfig
-from .contract import CommandRequest, CommandResponse
+from .config import ConnectorConfig
+from .contract import CommandRequest as CommandRequest
+from .contract import CommandResponse as CommandResponse
+from .llm_client import LLMClient
 from .openclaw_client import OpenClawClient
+from .router_admin_handlers import RouterAdminMixin
+from .router_chat_handlers import RouterChatMixin
+from .router_dispatch import RouterDispatchMixin
+from .router_execution_handlers import RouterExecutionMixin
 from .state import ConnectorState
 
 if False:  # Type hinting only
     from .results_poller import ResultsPoller
 
 from .command_firewall import CommandFirewall
-from .jobs_summary import JobsContractError, format_jobs_summary, format_queue_fallback
-from .llm_client import LLMClient
-from .prompts import CHAT_STATUS_PROMPT, CHAT_SYSTEM_PROMPT
 from .rate_limiter import RateLimiter
-from .semantic_guard import GuardAction, SemanticGuard
-
-try:
-    from services.reasoning_redaction import sanitize_operator_payload
-except Exception:  # pragma: no cover - connector tests may stub import graph
-    sanitize_operator_payload = lambda value, **_: value  # type: ignore
-
-logger = logging.getLogger(__name__)
+from .semantic_guard import SemanticGuard
 
 
-class CommandRouter:
+class CommandRouter(
+    RouterDispatchMixin,
+    RouterExecutionMixin,
+    RouterAdminMixin,
+    RouterChatMixin,
+):
+    def _build_llm_client(self) -> LLMClient:
+        """Resolve the facade dependency at call time to preserve patch seams."""
+        return LLMClient(self.client)
+
     def __init__(
         self,
         config: ConnectorConfig,
@@ -42,7 +42,7 @@ class CommandRouter:
         self.client = client
         self.poller = poller
         self.state = ConnectorState(path=self.config.state_path)
-        self._template_meta_cache: Dict[str, Dict[str, Any]] = {}
+        self._template_meta_cache: dict[str, dict[str, object]] = {}
         # F32 WP2: Rate limiter
         self._rate_limiter = RateLimiter(
             user_rpm=self.config.rate_limit_user_rpm,
@@ -51,1010 +51,3 @@ class CommandRouter:
         # S44/R97: Semantic Guards
         self.semantic_guard = SemanticGuard()
         self.command_firewall = CommandFirewall()
-
-    async def handle(self, req: CommandRequest) -> CommandResponse:
-        """Main dispatch loop."""
-        text = req.text.strip()
-        # NOTE: Debug-only raw message logging for troubleshooting parsing issues.
-        # Enable with OPENCLAW_CONNECTOR_DEBUG=1. May include sensitive user content.
-        if self.config.debug:
-            logger.info(
-                "DEBUG raw message: platform=%s user=%s chat=%s text=%r",
-                req.platform,
-                req.sender_id,
-                req.channel_id,
-                text,
-            )
-
-        # F32 WP2: Rate limiting
-        if not self._rate_limiter.is_allowed(str(req.sender_id), str(req.channel_id)):
-            return CommandResponse(
-                text="[Rate Limited] Too many requests. Please wait a moment."
-            )
-
-        # F32 WP5: Command length limit
-        if len(text) > self.config.max_command_length:
-            return CommandResponse(
-                text=f"[Error] Command too long ({len(text)} chars). Max: {self.config.max_command_length}."
-            )
-
-        try:
-            # IMPORTANT (recurring usability bug):
-            # Do not use `shlex.split()` directly for ChatOps commands that may include natural
-            # language. In POSIX mode, `shlex` treats apostrophes (`'`) as quote delimiters, so
-            # common contractions like "She's" trigger "unbalanced quotes" failures.
-            #
-            # We therefore only treat *double quotes* (`"`) as quoting characters, so users can
-            # still do: positive_prompt="a prompt with spaces" while apostrophes remain safe.
-            lexer = shlex.shlex(text, posix=True)
-            lexer.whitespace_split = True
-            lexer.commenters = ""
-            lexer.quotes = '"'
-            parts = list(lexer)
-        except ValueError:
-            return CommandResponse(
-                text="[Error] Parsing command arguments failed (unbalanced quotes?)."
-            )
-
-        if not parts:
-            return CommandResponse(text="Empty command.")
-
-        cmd = parts[0].lower()
-        args = parts[1:]
-
-        # Telegram group commands often include the bot username suffix, e.g. `/help@mybot`.
-        # If we don't strip it, the command won't match our dispatch table and appears "dead"
-        # even though polling is working.
-        if (
-            (req.platform or "").lower() == "telegram"
-            and cmd.startswith("/")
-            and "@" in cmd
-        ):
-            cmd = cmd.split("@", 1)[0]
-
-        # Some users type `@bot /help` in group chats. Treat that as a command too.
-        if cmd.startswith("@") and args and args[0].startswith("/"):
-            cmd = args[0].lower()
-            args = args[1:]
-
-        # Dispatch Table
-        handlers = {
-            ("/status", "status"): (self._handle_status, CommandClass.PUBLIC),
-            ("/help", "help", "/start"): (self._handle_help, CommandClass.PUBLIC),
-            ("/run", "run"): (self._handle_run, CommandClass.RUN),
-            ("/interrupt", "interrupt", "/cancel", "cancel", "/stop"): (
-                self._handle_interrupt,
-                CommandClass.ADMIN,
-            ),  # Global interrupt => admin-only.
-            ("/approvals", "approvals"): (
-                self._handle_approvals_list,
-                CommandClass.ADMIN,
-            ),
-            ("/approve", "approve"): (self._handle_approve, CommandClass.ADMIN),
-            ("/reject", "reject"): (self._handle_reject, CommandClass.ADMIN),
-            ("/schedules", "schedules"): (
-                self._handle_schedules_list,
-                CommandClass.ADMIN,
-            ),
-            ("/schedule", "schedule"): (
-                self._handle_schedule_subcommand,
-                CommandClass.ADMIN,
-            ),
-            # Phase 3 Introspection
-            ("/history", "history"): (self._handle_history, CommandClass.PUBLIC),
-            ("/trace", "trace"): (self._handle_trace, CommandClass.ADMIN),  # Admin only
-            ("/jobs", "jobs", "queue"): (self._handle_jobs, CommandClass.ADMIN),
-            # F30: Chat Assistant
-            ("/chat", "chat"): (self._handle_chat, CommandClass.PUBLIC),
-        }
-
-        # Find Handler
-        handler = None
-        requires_admin = False
-
-        canonical_cmd = cmd  # Fallback
-        for aliases, (func, cmd_class) in handlers.items():
-            if cmd in aliases:
-                handler = func
-                default_class = cmd_class
-                # R80 Remediation: Use canonical command (first alias) for policy checks
-                # This prevents "run" vs "/run" bypass issues.
-                if isinstance(aliases, tuple):
-                    # Convention: first alias is canonical (e.g. "/run")
-                    canonical_cmd = aliases[0]
-                else:
-                    canonical_cmd = aliases
-                break
-
-        if not handler:
-            return CommandResponse(
-                text=f"Unknown command: {cmd}. Type /help for options."
-            )
-
-        # R80: Centralized Authorization Gate
-        # Pass canonical_cmd to ensure policy matches aliases correctly
-        if auth_err := self._check_command_authz(canonical_cmd, req, default_class):
-            return auth_err
-
-        # 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
-
-    def _delivery_context(self, req: CommandRequest) -> Dict[str, Any]:
-        context: Dict[str, Any] = {}
-        if getattr(req, "workspace_id", ""):
-            context["workspace_id"] = str(req.workspace_id)
-        if getattr(req, "thread_id", ""):
-            context["thread_id"] = str(req.thread_id)
-        return context
-
-    def _check_command_authz(
-        self, cmd: str, req: CommandRequest, default_class: CommandClass
-    ) -> Optional[CommandResponse]:
-        """
-        R80: Verify command authorization policy.
-        Returns None if allowed, or CommandResponse(text=error) if denied.
-        """
-        policy = self.config.command_policy
-
-        # 1. Resolve Effective Class (Handle per-command overrides)
-        # Note: 'cmd' here is the canonical parsed command string (lowercase), e.g., "/run" or "run"
-        # The overrides dict might use "/run" or "run", we should check both or normalize.
-        # Currently, the router logic normalized `cmd` from input (lines 90-101).
-        # We'll check exact match against the override key.
-        eff_class = policy.command_overrides.get(cmd, default_class)
-
-        # 2. Check AllowFrom List (Explicit User Allow)
-        # If an explicit AllowFrom list exists for this class, the user MUST be in it.
-        # This takes precedence over role logic.
-        allowed_users = policy.allow_from.get(eff_class)
-        if allowed_users is not None and len(allowed_users) > 0:
-            if str(req.sender_id) not in allowed_users:
-                # If explicit allow-list is active, even admins must be in it?
-                # Decision: YES, for strict compliance. If you want admins, add them to the list.
-                # However, for usability, usually admins are implied.
-                # Let's stick to "Explicit List Wins" for R80 strict mode.
-                return CommandResponse(
-                    text="[Access Denied] You are not in the allow-list for this command."
-                )
-            # If in list, proceed (bypass default role checks? No, usually allows)
-            return None
-
-        # 3. Default Role Logic
-        if eff_class == CommandClass.ADMIN:
-            if not self._is_admin(req.sender_id):
-                return CommandResponse(
-                    text="[Access Denied] This command requires Admin privileges."
-                )
-
-        # PUBLIC and RUN are allowed by default (RUN checks trust internally)
-        return None
-
-    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 ValueError:
-                uid = None
-            try:
-                cid = int(channel_id)
-            except ValueError:
-                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
-
-        if platform == "whatsapp":
-            if sender_id in self.config.whatsapp_allowed_users:
-                return True
-            return False
-
-        if platform == "wechat":
-            if sender_id in self.config.wechat_allowed_users:
-                return True
-            return False
-
-        if platform == "kakao":
-            if sender_id in self.config.kakao_allowed_users:
-                return True
-            return False
-
-        if platform == "slack":
-            if sender_id in self.config.slack_allowed_users:
-                return True
-            if channel_id in self.config.slack_allowed_channels:
-                return True
-            return False
-
-        if platform == "feishu":
-            if sender_id in self.config.feishu_allowed_users:
-                return True
-            if channel_id in self.config.feishu_allowed_chats:
-                return True
-            return False
-
-        # Unknown platform: trust only admins
-        return False
-
-    # --- 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)
-        )
-
-    def _require_admin_token_configured(self) -> Optional[CommandResponse]:
-        """
-        F32 WP3: Check if admin token is configured before running admin commands.
-        Fail-fast with clear error message instead of 403/500 later.
-
-        IMPORTANT (recurring CI failure mode):
-        - Admin-only commands are gated by BOTH:
-          (1) sender is an admin user, AND
-          (2) the connector admin token is configured (OPENCLAW_CONNECTOR_ADMIN_TOKEN).
-        - Unit tests that exercise admin command handlers MUST set `config.admin_token`,
-          otherwise they will correctly receive the config error response.
-        """
-        if not self.config.admin_token:
-            return CommandResponse(
-                text="[Error] Admin token not configured. Set OPENCLAW_CONNECTOR_ADMIN_TOKEN and restart connector."
-            )
-        return None
-
-    async def _handle_run(
-        self, req: CommandRequest, args: List[str]
-    ) -> CommandResponse:
-        if not args:
-            return CommandResponse(
-                text="Usage: /run <template_id> [prompt text] [key=value ...] [--approval]"
-            )
-
-        # Parse flags
-        explicit_approval = False
-        clean_args = []
-        for arg in args:
-            if arg in ("--require-approval", "--approval", "-a"):
-                explicit_approval = True
-            else:
-                clean_args.append(arg)
-
-        if not clean_args:
-            return CommandResponse(text="Usage: /run <template_id> ...")
-
-        template_id = clean_args[0]
-        inputs: Dict[str, str] = {}
-        free_text_parts: List[str] = []
-        for arg in clean_args[1:]:
-            if "=" in arg:
-                k, v = arg.split("=", 1)
-                inputs[k.strip()] = v.strip()
-            else:
-                free_text_parts.append(arg)
-
-        # If user provided free text without key=value, treat it as the prompt.
-        # We map it to a best-effort prompt key (prefers template metadata if available).
-        if free_text_parts:
-            prompt_key = await self._resolve_prompt_key(template_id)
-            if prompt_key not in inputs:
-                inputs[prompt_key] = " ".join(free_text_parts).strip()
-            elif self.config.debug:
-                logger.info(
-                    "DEBUG /run free-text ignored (prompt key already set): %s",
-                    prompt_key,
-                )
-
-        # NOTE: Debug-only payload logging for troubleshooting prompt mismatches.
-        # Enable with OPENCLAW_CONNECTOR_DEBUG=1 to log template_id + inputs.
-        if self.config.debug:
-            logger.info(
-                "DEBUG /run payload: template=%s inputs=%s approval_flag=%s trusted=%s",
-                template_id,
-                inputs,
-                explicit_approval,
-                self._is_trusted(req),
-            )
-
-        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
-        )
-        if res.get("ok"):
-            data = res.get("data", {})
-            trace_id = data.get("trace_id", "unknown")
-
-            if data.get("pending"):
-                approval_id = data.get("approval_id", "unknown")
-                msg = f"[Approval Requested]\nID: {approval_id}\nTrace: {trace_id}"
-                if "expires_at" in data:
-                    msg += f"\nExpires: {data['expires_at']}"
-                if self.poller:
-                    # IMPORTANT:
-                    # For untrusted users, approvals are done in the OpenClaw UI.
-                    # We must start tracking the approval_id so we can map
-                    # approval_id -> executed_prompt_id later and auto-deliver images.
-                    self.poller.track_approval(
-                        approval_id,
-                        req.platform,
-                        req.channel_id,
-                        req.sender_id,
-                        delivery_context=self._delivery_context(req),
-                    )
-                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,
-                        delivery_context=self._delivery_context(req),
-                    )
-
-                return CommandResponse(
-                    text=f"[Job Submitted]\nID: {prompt_id}\nTemplate: {template_id}\nTrace: {trace_id}"
-                )
-        else:
-            err = res.get("error", "Unknown error")
-            return CommandResponse(text=f"[Submission Failed] Reason: {err}")
-
-    async def _resolve_prompt_key(self, template_id: str) -> str:
-        """
-        Best-effort prompt key resolution.
-        Prefer template metadata (allowed_inputs), then fall back to common names.
-        """
-        meta = await self._get_template_meta(template_id)
-        allowed = meta.get("allowed_inputs") or []
-
-        # If template explicitly declares a single input, use it.
-        if isinstance(allowed, list) and len(allowed) == 1:
-            return str(allowed[0])
-
-        preferred = ("positive_prompt", "prompt", "text", "positive", "caption")
-        if isinstance(allowed, list):
-            for key in preferred:
-                if key in allowed:
-                    return key
-
-        # Default fallback
-        return "positive_prompt"
-
-    async def _get_template_meta(self, template_id: str) -> Dict[str, Any]:
-        if template_id in self._template_meta_cache:
-            return self._template_meta_cache[template_id]
-        try:
-            res = await self.client.get_templates()
-            if res.get("ok"):
-                for item in res.get("templates", []) or []:
-                    if item.get("id") == template_id:
-                        self._template_meta_cache[template_id] = item
-                        return item
-        except Exception as e:
-            if self.config.debug:
-                logger.info(f"DEBUG template meta fetch failed: {e}")
-        return {}
-
-    async def _handle_interrupt(
-        self, req: CommandRequest, args: List[str]
-    ) -> CommandResponse:
-        # F32 WP3: Guard
-        if err := self._require_admin_token_configured():
-            return err
-
-        targets = self._parse_stop_targets(args)
-        if not targets:
-            res = await self.client.interrupt_output()
-            if res.get("ok"):
-                return CommandResponse(text="[Stop] Global Interrupt sent to ComfyUI.")
-            return CommandResponse(text=f"[Stop Failed] {res.get('error')}")
-
-        if len(targets) == 1:
-            job_id = targets[0]
-            res = await self.client.cancel_job(job_id)
-            if res.get("ok"):
-                return CommandResponse(
-                    text=f"[Stop] Cancellation requested for job {job_id}."
-                )
-
-            # IMPORTANT: Targeted stops must never degrade to no-payload global
-            # interrupt. Older-host fallback is allowed only with prompt_id set.
-            if self._jobs_cancel_unsupported(res):
-                fallback = await self.client.interrupt_output(prompt_id=job_id)
-                if fallback.get("ok"):
-                    return CommandResponse(
-                        text=(
-                            f"[Stop] Targeted interrupt sent for job {job_id} "
-                            "(jobs cancel unsupported)."
-                        )
-                    )
-                return CommandResponse(text=f"[Stop Failed] {fallback.get('error')}")
-
-            return CommandResponse(text=f"[Stop Failed] {res.get('error')}")
-
-        res = await self.client.cancel_jobs(targets)
-        if res.get("ok"):
-            return CommandResponse(
-                text=f"[Stop] Cancellation requested for {len(targets)} jobs."
-            )
-        return CommandResponse(text=f"[Stop Failed] {res.get('error')}")
-
-    @staticmethod
-    def _parse_stop_targets(args: List[str]) -> List[str]:
-        targets: List[str] = []
-        for arg in args:
-            for part in str(arg).split(","):
-                target = part.strip()
-                if target:
-                    targets.append(target)
-        return targets
-
-    @staticmethod
-    def _jobs_cancel_unsupported(res: Dict[str, Any]) -> bool:
-        status = res.get("status")
-        if status in (404, 405, 501):
-            return True
-        error = str(res.get("error", "")).lower()
-        unsupported_markers = (
-            "404",
-            "not found",
-            "method not allowed",
-            "unsupported",
-            "not implemented",
-        )
-        return any(marker in error for marker in unsupported_markers)
-
-    async def _handle_approvals_list(
-        self, req: CommandRequest, args: List[str]
-    ) -> CommandResponse:
-        # F32 WP3: Guard
-        if err := self._require_admin_token_configured():
-            return err
-
-        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 = []
-        buttons = []
-        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}"
-            )
-        for i in items[:3]:
-            approval_id = i.get("approval_id") or i.get("id") or "unknown"
-            short_id = str(approval_id)[:8]
-            buttons.append(
-                {
-                    "label": f"Approve {short_id}",
-                    "value": f"/approve {approval_id}",
-                    "action_type": "approval.approve",
-                    "approval_id": approval_id,
-                    "style": "primary",
-                }
-            )
-            buttons.append(
-                {
-                    "label": f"Reject {short_id}",
-                    "value": f"/reject {approval_id}",
-                    "action_type": "approval.reject",
-                    "approval_id": approval_id,
-                    "style": "danger",
-                }
-            )
-
-        header = "Pending Approvals"
-        if isinstance(pending_count, int):
-            header += f" ({pending_count})"
-        return CommandResponse(
-            text=header + ":\n" + "\n".join(lines),
-            buttons=buttons,
-        )
-
-    async def _handle_approve(
-        self, req: CommandRequest, args: List[str]
-    ) -> CommandResponse:
-        if not args:
-            return CommandResponse(text="Usage: /approve <id>")
-
-        # F32 WP3: Guard
-        if err := self._require_admin_token_configured():
-            return err
-
-        # Assuming auto_execute=True by default for chat logic
-        res = await self.client.approve_request(args[0], auto_execute=True)
-        if not res.get("ok"):
-            return CommandResponse(text=f"[Failed] {res.get('error')}")
-
-        data = res.get("data", {})
-        msg = f"[Approved] {args[0]}"
-
-        # Phase 4: Show execution result
-        if "prompt_id" in data:
-            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,
-                    delivery_context=self._delivery_context(req),
-                )
-        elif data.get("executed") is False:
-            msg += "\n(Not Executed)"
-            if err := data.get("execution_error"):
-                msg += f"\nError: {err}"
-
-        return CommandResponse(text=msg)
-
-    async def _handle_reject(
-        self, req: CommandRequest, args: List[str]
-    ) -> CommandResponse:
-        if not args:
-            return CommandResponse(text="Usage: /reject <id> [reason]")
-
-        # F32 WP3: Guard
-        if err := self._require_admin_token_configured():
-            return err
-
-        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:
-        # F32 WP3: Guard
-        if err := self._require_admin_token_configured():
-            return err
-
-        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>")
-
-        # F32 WP3: Guard
-        if err := self._require_admin_token_configured():
-            return err
-
-        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> [prompt] [k=v] - Run a generation (trusted users auto-exec; others require approval)\n"
-                "/stop [job_id ...] - Cancel jobs by id; no args sends Global Interrupt (Admin)\n"
-                "/history <id> - Job details\n"
-                "Admin Only:\n"
-                "/jobs - Authoritative jobs summary\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>")
-
-        # F32 WP3: Guard
-        if err := self._require_admin_token_configured():
-            return err
-
-        res = await self.client.get_trace(args[0])
-        if not res.get("ok"):
-            return CommandResponse(text=f"[Error] {res.get('error')}")
-
-        # Dump trace
-        sanitized = sanitize_operator_payload(res.get("data"))
-        return CommandResponse(text=f"Trace {args[0]}: {str(sanitized)[:1000]}...")
-
-    async def _handle_jobs(
-        self, req: CommandRequest, args: List[str]
-    ) -> CommandResponse:
-        if err := self._require_admin_token_configured():
-            return err
-
-        res = await self.client.get_jobs()
-        if not isinstance(res, Mapping):
-            return CommandResponse(
-                text="[Jobs] Could not fetch the authoritative jobs snapshot."
-            )
-        if res.get("ok") is True:
-            try:
-                return CommandResponse(text=format_jobs_summary(res.get("data")))
-            except JobsContractError:
-                return CommandResponse(
-                    text="[Jobs] Malformed or unsupported jobs response."
-                )
-
-        status = res.get("status")
-        error = res.get("error")
-        access_denied = (
-            isinstance(status, int)
-            and not isinstance(status, bool)
-            and status in {401, 403}
-        )
-        if access_denied:
-            return CommandResponse(
-                text="[Jobs] Access denied. Check connector Admin authorization and token posture."
-            )
-        fallback_allowed = isinstance(error, str) and (
-            (status == 501 and error == "jobs_host_contract_unsupported")
-            or (status == 503 and error == "jobs_backend_unavailable")
-        )
-        if fallback_allowed:
-            return CommandResponse(
-                text=format_queue_fallback(await self.client.get_prompt_queue())
-            )
-        return CommandResponse(
-            text="[Jobs] Could not fetch the authoritative jobs snapshot."
-        )
-
-    # -------------------------------------------------------------------------
-    # F30: Chat LLM Assistant
-    # -------------------------------------------------------------------------
-
-    async def _handle_chat(
-        self, req: CommandRequest, args: List[str]
-    ) -> CommandResponse:
-        """
-        /chat [subcommand] <message>
-        Subcommands: run, template, status
-        Default: general chat
-
-        Security: Never auto-executes commands. Only suggests command text.
-        """
-        llm = LLMClient(self.client)
-
-        if not await llm.is_configured():
-            return CommandResponse(
-                text="[Chat Error] LLM not configured. Configure in OpenClaw Settings."
-            )
-
-        # Parse subcommand
-        if not args:
-            return CommandResponse(
-                text="Usage: /chat <message> or /chat run|template|status <request>"
-            )
-
-        subcommand = args[0].lower()
-        message = " ".join(args[1:]) if len(args) > 1 else ""
-
-        trust_level = "TRUSTED" if self._is_trusted(req) else "UNTRUSTED"
-
-        if subcommand == "run":
-            return await self._chat_run(llm, message, trust_level)
-        elif subcommand == "template":
-            return await self._chat_template(llm, message)
-        elif subcommand == "status":
-            return await self._chat_status(llm)
-        else:
-            # General chat: first word is part of message
-            full_message = " ".join(args)
-            return await self._chat_general(llm, full_message, trust_level)
-
-    async def _chat_general(
-        self, llm: LLMClient, message: str, trust_level: str
-    ) -> CommandResponse:
-        """General chat with assistant."""
-        # S44: Semantic Guard Evaluation
-        decision = self.semantic_guard.evaluate_request(message, {"trust": trust_level})
-
-        if decision.action == GuardAction.DENY:
-            return CommandResponse(
-                text=(
-                    "[Blocked] Request denied by semantic policy "
-                    f"({decision.reason}). {self._policy_kv(decision.to_contract())}"
-                )
-            )
-
-        system_prompt = CHAT_SYSTEM_PROMPT.format(trust_level=trust_level)
-        response = await llm.chat(system_prompt, message)
-
-        # S44: Output Validation + SAFE_REPLY sanitization.
-        try:
-            response = self.semantic_guard.validate_output(
-                response, "general", decision.action
-            )
-        except ValueError as e:
-            return CommandResponse(
-                text=(
-                    "[Validation Error] Assistant output invalid: "
-                    f"{e}. {self._policy_kv({'code': 'semantic_output_invalid', 'severity': 'medium', 'action': 'deny', 'reason': str(e)})}"
-                )
-            )
-
-        if decision.action == GuardAction.SAFE_REPLY:
-            safe_response = (
-                response
-                or "I can help with general guidance, but commands are restricted for this request."
-            )
-            return CommandResponse(
-                text=(
-                    f"[Safe Mode] {safe_response}\n\n"
-                    f"(Policy: {self._policy_kv(decision.to_contract())})"
-                )
-            )
-
-        return CommandResponse(text=response)
-
-    async def _chat_run(
-        self, llm: LLMClient, request: str, trust_level: str
-    ) -> CommandResponse:
-        """Suggest a /run command based on user request."""
-        if not request:
-            return CommandResponse(
-                text="Usage: /chat run <description of what you want>"
-            )
-
-        # S44: Semantic Guard Evaluation
-        decision = self.semantic_guard.evaluate_request(request, {"trust": trust_level})
-
-        if decision.action == GuardAction.DENY:
-            return CommandResponse(
-                text=(
-                    "[Blocked] Request denied by semantic policy "
-                    f"({decision.reason}). {self._policy_kv(decision.to_contract())}"
-                )
-            )
-
-        # Force Approval Override based on Risk
-        force_approval_policy = decision.action == GuardAction.FORCE_APPROVAL
-
-        # Get available templates (simplified - could fetch from API)
-        templates = "txt2img, img2img, upscale (examples)"
-
-        system_prompt = CHAT_SYSTEM_PROMPT.format(trust_level=trust_level)
-        user_prompt = f"""User wants to run a generation. Suggest a `/run` command.
-
-Request: {request}
-Available templates: {templates}
-Trust level: {trust_level}
-
-Remember: {"add --approval flag" if trust_level == "UNTRUSTED" else "no --approval needed"}.
-Output only the command in a code block."""
-
-        response = await llm.chat(system_prompt, user_prompt)
-
-        # S44: Output Structure Validation
-        try:
-            response = self.semantic_guard.validate_output(
-                response, "run", decision.action
-            )
-        except ValueError as e:
-            return CommandResponse(
-                text=(
-                    "[Validation Error] Assistant output invalid: "
-                    f"{e}. {self._policy_kv({'code': 'semantic_output_invalid', 'severity': 'high', 'action': 'deny', 'reason': str(e)})}"
-                )
-            )
-
-        # R97: Command Firewall - Extract and Validate
-        import re
-
-        cmd_match = re.search(r"```(?:bash)?\s*(.*?)\s*```", response, re.DOTALL)
-        raw_cmd = cmd_match.group(1).strip() if cmd_match else response.strip()
-
-        # Validate through Firewall
-        normalized = self.command_firewall.validate_suggestion(raw_cmd)
-
-        if not normalized.is_safe:
-            return CommandResponse(
-                text=(
-                    "[Safety Block] Assistant suggested unsafe command: "
-                    f"{normalized.safety_reason}. {self._policy_kv(normalized.to_contract())}"
-                )
-            )
-
-        # R97: Strict /run enforcement (Remediation for Medium Severity)
-        # CRITICAL: keep this check. /chat run must never emit non-/run commands.
-        if normalized.command != "/run":
-            return CommandResponse(
-                text=(
-                    "[Policy Block] Only /run commands are allowed in this mode. "
-                    f"Got: {normalized.command}. "
-                    f"{self._policy_kv({'code': 'firewall_non_run_command', 'severity': 'high', 'action': 'deny', 'reason': 'non_run_command_in_run_mode'})}"
-                )
-            )
-
-        # R97/S44: Apply Policy Overrides
-        # If risk was elevated, ensure --approval is present
-        if (
-            force_approval_policy
-            and "--approval" not in normalized.args
-            and "approval" not in normalized.flags
-        ):
-            normalized.args.append("--approval")
-
-        final_cmd = normalized.to_string()
-
-        # Return as code block for easy copy-paste (or auto-execution UI cues)
-        if force_approval_policy:
-            return CommandResponse(
-                text=(
-                    f"```\n{final_cmd}\n```\n"
-                    f"(Policy: {self._policy_kv(decision.to_contract())})"
-                )
-            )
-        return CommandResponse(text=f"```\n{final_cmd}\n```")
-
-    @staticmethod
-    def _policy_kv(contract: Dict[str, Any]) -> str:
-        ordered = ("code", "severity", "action", "reason")
-        parts = []
-        for key in ordered:
-            value = contract.get(key)
-            if value is not None:
-                parts.append(f"{key}={value}")
-        return "[" + ", ".join(parts) + "]"
-
-    async def _chat_template(self, llm: LLMClient, request: str) -> CommandResponse:
-        """Generate a template JSON suggestion."""
-        if not request:
-            return CommandResponse(text="Usage: /chat template <description>")
-
-        system_prompt = CHAT_SYSTEM_PROMPT.format(trust_level="N/A")
-        user_prompt = f"""Generate a workflow template JSON for this request:
-
-Request: {request}
-
-Output:
-1. Suggested filename
-2. Template JSON in a code block
-
-Keep it minimal."""
-
-        response = await llm.chat(system_prompt, user_prompt)
-        return CommandResponse(text=response)
-
-    async def _chat_status(self, llm: LLMClient) -> CommandResponse:
-        """Summarize system status using LLM."""
-        # Fetch status data
-        health = await self.client.get_health()
-        queue = await self.client.get_prompt_queue()
-
-        status_data = {
-            "health": health.get("data", {}) if health.get("ok") else "unavailable",
-            "jobs": "admin-only; use /jobs as an authorized operator",
-            "queue": queue.get("data", {}) if queue.get("ok") else "unavailable",
-        }
-
-        system_prompt = CHAT_SYSTEM_PROMPT.format(trust_level="N/A")
-        user_prompt = CHAT_STATUS_PROMPT.format(status_data=status_data)
-
-        response = await llm.chat(system_prompt, user_prompt)
-        return CommandResponse(text=response)
diff --git a/connector/router_admin_handlers.py b/connector/router_admin_handlers.py
new file mode 100644
index 0000000..0312775
--- /dev/null
+++ b/connector/router_admin_handlers.py
@@ -0,0 +1,332 @@
+"""Owned status, approval, schedule, and introspection command mixin."""
+
+# ruff: noqa: UP006, UP035, UP045 -- preserve the frozen public annotations.
+# mypy: disable-error-code="attr-defined"
+
+from collections.abc import Mapping
+from typing import List, Optional
+
+from .contract import CommandRequest, CommandResponse
+from .jobs_summary import JobsContractError, format_jobs_summary, format_queue_fallback
+
+try:
+    from services.reasoning_redaction import sanitize_operator_payload
+except Exception:  # pragma: no cover - connector tests may stub import graph
+
+    def sanitize_operator_payload(value, **_):  # type: ignore
+        return value
+
+
+class RouterAdminMixin:
+    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)
+        )
+
+    def _require_admin_token_configured(self) -> Optional[CommandResponse]:
+        """
+        F32 WP3: Check if admin token is configured before running admin commands.
+        Fail-fast with clear error message instead of 403/500 later.
+
+        IMPORTANT (recurring CI failure mode):
+        - Admin-only commands are gated by BOTH:
+          (1) sender is an admin user, AND
+          (2) the connector admin token is configured (OPENCLAW_CONNECTOR_ADMIN_TOKEN).
+        - Unit tests that exercise admin command handlers MUST set `config.admin_token`,
+          otherwise they will correctly receive the config error response.
+        """
+        if not self.config.admin_token:
+            return CommandResponse(
+                text="[Error] Admin token not configured. Set OPENCLAW_CONNECTOR_ADMIN_TOKEN and restart connector."
+            )
+        return None
+
+    async def _handle_approvals_list(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        # F32 WP3: Guard
+        if err := self._require_admin_token_configured():
+            return err
+
+        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 = []
+        buttons = []
+        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}"
+            )
+        for i in items[:3]:
+            approval_id = i.get("approval_id") or i.get("id") or "unknown"
+            short_id = str(approval_id)[:8]
+            buttons.append(
+                {
+                    "label": f"Approve {short_id}",
+                    "value": f"/approve {approval_id}",
+                    "action_type": "approval.approve",
+                    "approval_id": approval_id,
+                    "style": "primary",
+                }
+            )
+            buttons.append(
+                {
+                    "label": f"Reject {short_id}",
+                    "value": f"/reject {approval_id}",
+                    "action_type": "approval.reject",
+                    "approval_id": approval_id,
+                    "style": "danger",
+                }
+            )
+
+        header = "Pending Approvals"
+        if isinstance(pending_count, int):
+            header += f" ({pending_count})"
+        return CommandResponse(
+            text=header + ":\n" + "\n".join(lines),
+            buttons=buttons,
+        )
+
+    async def _handle_approve(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        if not args:
+            return CommandResponse(text="Usage: /approve <id>")
+
+        # F32 WP3: Guard
+        if err := self._require_admin_token_configured():
+            return err
+
+        # Assuming auto_execute=True by default for chat logic
+        res = await self.client.approve_request(args[0], auto_execute=True)
+        if not res.get("ok"):
+            return CommandResponse(text=f"[Failed] {res.get('error')}")
+
+        data = res.get("data", {})
+        msg = f"[Approved] {args[0]}"
+
+        # Phase 4: Show execution result
+        if "prompt_id" in data:
+            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,
+                    delivery_context=self._delivery_context(req),
+                )
+        elif data.get("executed") is False:
+            msg += "\n(Not Executed)"
+            if err := data.get("execution_error"):
+                msg += f"\nError: {err}"
+
+        return CommandResponse(text=msg)
+
+    async def _handle_reject(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        if not args:
+            return CommandResponse(text="Usage: /reject <id> [reason]")
+
+        # F32 WP3: Guard
+        if err := self._require_admin_token_configured():
+            return err
+
+        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:
+        # F32 WP3: Guard
+        if err := self._require_admin_token_configured():
+            return err
+
+        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>")
+
+        # F32 WP3: Guard
+        if err := self._require_admin_token_configured():
+            return err
+
+        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> [prompt] [k=v] - Run a generation (trusted users auto-exec; others require approval)\n"
+                "/stop [job_id ...] - Cancel jobs by id; no args sends Global Interrupt (Admin)\n"
+                "/history <id> - Job details\n"
+                "Admin Only:\n"
+                "/jobs - Authoritative jobs summary\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>")
+
+        # F32 WP3: Guard
+        if err := self._require_admin_token_configured():
+            return err
+
+        res = await self.client.get_trace(args[0])
+        if not res.get("ok"):
+            return CommandResponse(text=f"[Error] {res.get('error')}")
+
+        # Dump trace
+        sanitized = sanitize_operator_payload(res.get("data"))
+        return CommandResponse(text=f"Trace {args[0]}: {str(sanitized)[:1000]}...")
+
+    async def _handle_jobs(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        if err := self._require_admin_token_configured():
+            return err
+
+        res = await self.client.get_jobs()
+        if not isinstance(res, Mapping):
+            return CommandResponse(
+                text="[Jobs] Could not fetch the authoritative jobs snapshot."
+            )
+        if res.get("ok") is True:
+            try:
+                return CommandResponse(text=format_jobs_summary(res.get("data")))
+            except JobsContractError:
+                return CommandResponse(
+                    text="[Jobs] Malformed or unsupported jobs response."
+                )
+
+        status = res.get("status")
+        error = res.get("error")
+        access_denied = (
+            isinstance(status, int)
+            and not isinstance(status, bool)
+            and status in {401, 403}
+        )
+        if access_denied:
+            return CommandResponse(
+                text="[Jobs] Access denied. Check connector Admin authorization and token posture."
+            )
+        fallback_allowed = isinstance(error, str) and (
+            (status == 501 and error == "jobs_host_contract_unsupported")
+            or (status == 503 and error == "jobs_backend_unavailable")
+        )
+        if fallback_allowed:
+            return CommandResponse(
+                text=format_queue_fallback(await self.client.get_prompt_queue())
+            )
+        return CommandResponse(
+            text="[Jobs] Could not fetch the authoritative jobs snapshot."
+        )
+
+    # -------------------------------------------------------------------------
+    # F30: Chat LLM Assistant
+    # -------------------------------------------------------------------------
diff --git a/connector/router_chat_handlers.py b/connector/router_chat_handlers.py
new file mode 100644
index 0000000..a44d552
--- /dev/null
+++ b/connector/router_chat_handlers.py
@@ -0,0 +1,244 @@
+"""Owned chat and semantic-guard command-family mixin."""
+
+# ruff: noqa: UP006, UP035 -- preserve the frozen public annotations.
+# mypy: disable-error-code="attr-defined"
+
+from typing import Any, Dict, List
+
+from .contract import CommandRequest, CommandResponse
+from .llm_client import LLMClient
+from .prompts import CHAT_STATUS_PROMPT, CHAT_SYSTEM_PROMPT
+from .semantic_guard import GuardAction
+
+
+class RouterChatMixin:
+    async def _handle_chat(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        """
+        /chat [subcommand] <message>
+        Subcommands: run, template, status
+        Default: general chat
+
+        Security: Never auto-executes commands. Only suggests command text.
+        """
+        llm = self._build_llm_client()
+
+        if not await llm.is_configured():
+            return CommandResponse(
+                text="[Chat Error] LLM not configured. Configure in OpenClaw Settings."
+            )
+
+        # Parse subcommand
+        if not args:
+            return CommandResponse(
+                text="Usage: /chat <message> or /chat run|template|status <request>"
+            )
+
+        subcommand = args[0].lower()
+        message = " ".join(args[1:]) if len(args) > 1 else ""
+
+        trust_level = "TRUSTED" if self._is_trusted(req) else "UNTRUSTED"
+
+        if subcommand == "run":
+            return await self._chat_run(llm, message, trust_level)
+        elif subcommand == "template":
+            return await self._chat_template(llm, message)
+        elif subcommand == "status":
+            return await self._chat_status(llm)
+        else:
+            # General chat: first word is part of message
+            full_message = " ".join(args)
+            return await self._chat_general(llm, full_message, trust_level)
+
+    async def _chat_general(
+        self, llm: LLMClient, message: str, trust_level: str
+    ) -> CommandResponse:
+        """General chat with assistant."""
+        # S44: Semantic Guard Evaluation
+        decision = self.semantic_guard.evaluate_request(message, {"trust": trust_level})
+
+        if decision.action == GuardAction.DENY:
+            return CommandResponse(
+                text=(
+                    "[Blocked] Request denied by semantic policy "
+                    f"({decision.reason}). {self._policy_kv(decision.to_contract())}"
+                )
+            )
+
+        system_prompt = CHAT_SYSTEM_PROMPT.format(trust_level=trust_level)
+        response = await llm.chat(system_prompt, message)
+
+        # S44: Output Validation + SAFE_REPLY sanitization.
+        try:
+            response = self.semantic_guard.validate_output(
+                response, "general", decision.action
+            )
+        except ValueError as e:
+            return CommandResponse(
+                text=(
+                    "[Validation Error] Assistant output invalid: "
+                    f"{e}. {self._policy_kv({'code': 'semantic_output_invalid', 'severity': 'medium', 'action': 'deny', 'reason': str(e)})}"
+                )
+            )
+
+        if decision.action == GuardAction.SAFE_REPLY:
+            safe_response = (
+                response
+                or "I can help with general guidance, but commands are restricted for this request."
+            )
+            return CommandResponse(
+                text=(
+                    f"[Safe Mode] {safe_response}\n\n"
+                    f"(Policy: {self._policy_kv(decision.to_contract())})"
+                )
+            )
+
+        return CommandResponse(text=response)
+
+    async def _chat_run(
+        self, llm: LLMClient, request: str, trust_level: str
+    ) -> CommandResponse:
+        """Suggest a /run command based on user request."""
+        if not request:
+            return CommandResponse(
+                text="Usage: /chat run <description of what you want>"
+            )
+
+        # S44: Semantic Guard Evaluation
+        decision = self.semantic_guard.evaluate_request(request, {"trust": trust_level})
+
+        if decision.action == GuardAction.DENY:
+            return CommandResponse(
+                text=(
+                    "[Blocked] Request denied by semantic policy "
+                    f"({decision.reason}). {self._policy_kv(decision.to_contract())}"
+                )
+            )
+
+        # Force Approval Override based on Risk
+        force_approval_policy = decision.action == GuardAction.FORCE_APPROVAL
+
+        # Get available templates (simplified - could fetch from API)
+        templates = "txt2img, img2img, upscale (examples)"
+
+        system_prompt = CHAT_SYSTEM_PROMPT.format(trust_level=trust_level)
+        user_prompt = f"""User wants to run a generation. Suggest a `/run` command.
+
+Request: {request}
+Available templates: {templates}
+Trust level: {trust_level}
+
+Remember: {"add --approval flag" if trust_level == "UNTRUSTED" else "no --approval needed"}.
+Output only the command in a code block."""
+
+        response = await llm.chat(system_prompt, user_prompt)
+
+        # S44: Output Structure Validation
+        try:
+            response = self.semantic_guard.validate_output(
+                response, "run", decision.action
+            )
+        except ValueError as e:
+            return CommandResponse(
+                text=(
+                    "[Validation Error] Assistant output invalid: "
+                    f"{e}. {self._policy_kv({'code': 'semantic_output_invalid', 'severity': 'high', 'action': 'deny', 'reason': str(e)})}"
+                )
+            )
+
+        # R97: Command Firewall - Extract and Validate
+        import re
+
+        cmd_match = re.search(r"```(?:bash)?\s*(.*?)\s*```", response, re.DOTALL)
+        raw_cmd = cmd_match.group(1).strip() if cmd_match else response.strip()
+
+        # Validate through Firewall
+        normalized = self.command_firewall.validate_suggestion(raw_cmd)
+
+        if not normalized.is_safe:
+            return CommandResponse(
+                text=(
+                    "[Safety Block] Assistant suggested unsafe command: "
+                    f"{normalized.safety_reason}. {self._policy_kv(normalized.to_contract())}"
+                )
+            )
+
+        # R97: Strict /run enforcement (Remediation for Medium Severity)
+        # CRITICAL: keep this check. /chat run must never emit non-/run commands.
+        if normalized.command != "/run":
+            return CommandResponse(
+                text=(
+                    "[Policy Block] Only /run commands are allowed in this mode. "
+                    f"Got: {normalized.command}. "
+                    f"{self._policy_kv({'code': 'firewall_non_run_command', 'severity': 'high', 'action': 'deny', 'reason': 'non_run_command_in_run_mode'})}"
+                )
+            )
+
+        # R97/S44: Apply Policy Overrides
+        # If risk was elevated, ensure --approval is present
+        if (
+            force_approval_policy
+            and "--approval" not in normalized.args
+            and "approval" not in normalized.flags
+        ):
+            normalized.args.append("--approval")
+
+        final_cmd = normalized.to_string()
+
+        # Return as code block for easy copy-paste (or auto-execution UI cues)
+        if force_approval_policy:
+            return CommandResponse(
+                text=(
+                    f"```\n{final_cmd}\n```\n"
+                    f"(Policy: {self._policy_kv(decision.to_contract())})"
+                )
+            )
+        return CommandResponse(text=f"```\n{final_cmd}\n```")
+
+    @staticmethod
+    def _policy_kv(contract: Dict[str, Any]) -> str:
+        ordered = ("code", "severity", "action", "reason")
+        parts = []
+        for key in ordered:
+            value = contract.get(key)
+            if value is not None:
+                parts.append(f"{key}={value}")
+        return "[" + ", ".join(parts) + "]"
+
+    async def _chat_template(self, llm: LLMClient, request: str) -> CommandResponse:
+        """Generate a template JSON suggestion."""
+        if not request:
+            return CommandResponse(text="Usage: /chat template <description>")
+
+        system_prompt = CHAT_SYSTEM_PROMPT.format(trust_level="N/A")
+        user_prompt = f"""Generate a workflow template JSON for this request:
+
+Request: {request}
+
+Output:
+1. Suggested filename
+2. Template JSON in a code block
+
+Keep it minimal."""
+
+        response = await llm.chat(system_prompt, user_prompt)
+        return CommandResponse(text=response)
+
+    async def _chat_status(self, llm: LLMClient) -> CommandResponse:
+        """Summarize system status using LLM."""
+        # Fetch status data
+        health = await self.client.get_health()
+        queue = await self.client.get_prompt_queue()
+
+        status_data = {
+            "health": health.get("data", {}) if health.get("ok") else "unavailable",
+            "jobs": "admin-only; use /jobs as an authorized operator",
+            "queue": queue.get("data", {}) if queue.get("ok") else "unavailable",
+        }
+
+        system_prompt = CHAT_SYSTEM_PROMPT.format(trust_level="N/A")
+        user_prompt = CHAT_STATUS_PROMPT.format(status_data=status_data)
+
+        response = await llm.chat(system_prompt, user_prompt)
+        return CommandResponse(text=response)
diff --git a/connector/router_dispatch.py b/connector/router_dispatch.py
new file mode 100644
index 0000000..21927ee
--- /dev/null
+++ b/connector/router_dispatch.py
@@ -0,0 +1,275 @@
+"""Owned command parsing, dispatch, and authorization mixin."""
+
+# ruff: noqa: UP006, UP035, UP045 -- preserve the frozen public annotations.
+# mypy: disable-error-code="attr-defined,no-any-return"
+
+import logging
+import shlex
+from dataclasses import dataclass
+from typing import Any, Dict, Optional
+
+from .config import CommandClass
+from .contract import CommandRequest, CommandResponse
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class RouterRequestContext:
+    """Immutable dispatch values for one authorized command attempt."""
+
+    request: CommandRequest
+    parsed_command: str
+    canonical_command: str
+    args: tuple[str, ...]
+    command_class: CommandClass
+
+
+class RouterDispatchMixin:
+    async def handle(self, req: CommandRequest) -> CommandResponse:
+        """Main dispatch loop."""
+        text = req.text.strip()
+        # NOTE: Debug-only raw message logging for troubleshooting parsing issues.
+        # Enable with OPENCLAW_CONNECTOR_DEBUG=1. May include sensitive user content.
+        if self.config.debug:
+            logger.info(
+                "DEBUG raw message: platform=%s user=%s chat=%s text=%r",
+                req.platform,
+                req.sender_id,
+                req.channel_id,
+                text,
+            )
+
+        # F32 WP2: Rate limiting
+        if not self._rate_limiter.is_allowed(str(req.sender_id), str(req.channel_id)):
+            return CommandResponse(
+                text="[Rate Limited] Too many requests. Please wait a moment."
+            )
+
+        # F32 WP5: Command length limit
+        if len(text) > self.config.max_command_length:
+            return CommandResponse(
+                text=f"[Error] Command too long ({len(text)} chars). Max: {self.config.max_command_length}."
+            )
+
+        try:
+            # IMPORTANT (recurring usability bug):
+            # Do not use `shlex.split()` directly for ChatOps commands that may include natural
+            # language. In POSIX mode, `shlex` treats apostrophes (`'`) as quote delimiters, so
+            # common contractions like "She's" trigger "unbalanced quotes" failures.
+            #
+            # We therefore only treat *double quotes* (`"`) as quoting characters, so users can
+            # still do: positive_prompt="a prompt with spaces" while apostrophes remain safe.
+            lexer = shlex.shlex(text, posix=True)
+            lexer.whitespace_split = True
+            lexer.commenters = ""
+            lexer.quotes = '"'
+            parts = list(lexer)
+        except ValueError:
+            return CommandResponse(
+                text="[Error] Parsing command arguments failed (unbalanced quotes?)."
+            )
+
+        if not parts:
+            return CommandResponse(text="Empty command.")
+
+        cmd = parts[0].lower()
+        args = parts[1:]
+
+        # Telegram group commands often include the bot username suffix, e.g. `/help@mybot`.
+        # If we don't strip it, the command won't match our dispatch table and appears "dead"
+        # even though polling is working.
+        if (
+            (req.platform or "").lower() == "telegram"
+            and cmd.startswith("/")
+            and "@" in cmd
+        ):
+            cmd = cmd.split("@", 1)[0]
+
+        # Some users type `@bot /help` in group chats. Treat that as a command too.
+        if cmd.startswith("@") and args and args[0].startswith("/"):
+            cmd = args[0].lower()
+            args = args[1:]
+
+        # Dispatch Table
+        handlers = {
+            ("/status", "status"): (self._handle_status, CommandClass.PUBLIC),
+            ("/help", "help", "/start"): (self._handle_help, CommandClass.PUBLIC),
+            ("/run", "run"): (self._handle_run, CommandClass.RUN),
+            ("/interrupt", "interrupt", "/cancel", "cancel", "/stop"): (
+                self._handle_interrupt,
+                CommandClass.ADMIN,
+            ),  # Global interrupt => admin-only.
+            ("/approvals", "approvals"): (
+                self._handle_approvals_list,
+                CommandClass.ADMIN,
+            ),
+            ("/approve", "approve"): (self._handle_approve, CommandClass.ADMIN),
+            ("/reject", "reject"): (self._handle_reject, CommandClass.ADMIN),
+            ("/schedules", "schedules"): (
+                self._handle_schedules_list,
+                CommandClass.ADMIN,
+            ),
+            ("/schedule", "schedule"): (
+                self._handle_schedule_subcommand,
+                CommandClass.ADMIN,
+            ),
+            # Phase 3 Introspection
+            ("/history", "history"): (self._handle_history, CommandClass.PUBLIC),
+            ("/trace", "trace"): (self._handle_trace, CommandClass.ADMIN),  # Admin only
+            ("/jobs", "jobs", "queue"): (self._handle_jobs, CommandClass.ADMIN),
+            # F30: Chat Assistant
+            ("/chat", "chat"): (self._handle_chat, CommandClass.PUBLIC),
+        }
+
+        # Find Handler
+        handler = None
+
+        canonical_cmd = cmd  # Fallback
+        for aliases, (func, cmd_class) in handlers.items():
+            if cmd in aliases:
+                handler = func
+                default_class = cmd_class
+                # R80 Remediation: Use canonical command (first alias) for policy checks
+                # This prevents "run" vs "/run" bypass issues.
+                # Convention: first alias is canonical (e.g. "/run").
+                canonical_cmd = aliases[0] if isinstance(aliases, tuple) else aliases
+                break
+
+        if not handler:
+            return CommandResponse(
+                text=f"Unknown command: {cmd}. Type /help for options."
+            )
+
+        context = RouterRequestContext(
+            request=req,
+            parsed_command=cmd,
+            canonical_command=canonical_cmd,
+            args=tuple(args),
+            command_class=default_class,
+        )
+
+        # R80: Centralized Authorization Gate
+        # Pass canonical_cmd to ensure policy matches aliases correctly
+        if auth_err := self._check_command_authz(
+            context.canonical_command, context.request, context.command_class
+        ):
+            return auth_err
+
+        # Execute
+        try:
+            return await handler(context.request, list(context.args))
+        except Exception as e:
+            logger.exception(f"Command execution error {cmd}: {e}")
+            return CommandResponse(text=f"[Internal Error] {e!s}")
+
+    def _is_admin(self, user_id: str) -> bool:
+        return str(user_id) in self.config.admin_users
+
+    def _delivery_context(self, req: CommandRequest) -> Dict[str, Any]:
+        context: Dict[str, Any] = {}
+        if getattr(req, "workspace_id", ""):
+            context["workspace_id"] = str(req.workspace_id)
+        if getattr(req, "thread_id", ""):
+            context["thread_id"] = str(req.thread_id)
+        return context
+
+    def _check_command_authz(
+        self, cmd: str, req: CommandRequest, default_class: CommandClass
+    ) -> Optional[CommandResponse]:
+        """
+        R80: Verify command authorization policy.
+        Returns None if allowed, or CommandResponse(text=error) if denied.
+        """
+        policy = self.config.command_policy
+
+        # 1. Resolve Effective Class (Handle per-command overrides)
+        # Note: 'cmd' here is the canonical parsed command string (lowercase), e.g., "/run" or "run"
+        # The overrides dict might use "/run" or "run", we should check both or normalize.
+        # Currently, the router logic normalized `cmd` from input (lines 90-101).
+        # We'll check exact match against the override key.
+        eff_class = policy.command_overrides.get(cmd, default_class)
+
+        # 2. Check AllowFrom List (Explicit User Allow)
+        # If an explicit AllowFrom list exists for this class, the user MUST be in it.
+        # This takes precedence over role logic.
+        allowed_users = policy.allow_from.get(eff_class)
+        if allowed_users is not None and len(allowed_users) > 0:
+            if str(req.sender_id) not in allowed_users:
+                # If explicit allow-list is active, even admins must be in it?
+                # Decision: YES, for strict compliance. If you want admins, add them to the list.
+                # However, for usability, usually admins are implied.
+                # Let's stick to "Explicit List Wins" for R80 strict mode.
+                return CommandResponse(
+                    text="[Access Denied] You are not in the allow-list for this command."
+                )
+            # If in list, proceed (bypass default role checks? No, usually allows)
+            return None
+
+        # 3. Default Role Logic
+        if eff_class == CommandClass.ADMIN and not self._is_admin(req.sender_id):
+            return CommandResponse(
+                text="[Access Denied] This command requires Admin privileges."
+            )
+
+        # PUBLIC and RUN are allowed by default (RUN checks trust internally)
+        return None
+
+    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 ValueError:
+                uid = None
+            try:
+                cid = int(channel_id)
+            except ValueError:
+                cid = None
+            if uid is not None and uid in self.config.telegram_allowed_users:
+                return True
+            return cid is not None and cid in self.config.telegram_allowed_chats
+
+        if platform == "discord":
+            if sender_id in self.config.discord_allowed_users:
+                return True
+            return channel_id in self.config.discord_allowed_channels
+
+        if platform == "line":
+            if sender_id in self.config.line_allowed_users:
+                return True
+            return channel_id in self.config.line_allowed_groups
+
+        if platform == "whatsapp":
+            return sender_id in self.config.whatsapp_allowed_users
+
+        if platform == "wechat":
+            return sender_id in self.config.wechat_allowed_users
+
+        if platform == "kakao":
+            return sender_id in self.config.kakao_allowed_users
+
+        if platform == "slack":
+            if sender_id in self.config.slack_allowed_users:
+                return True
+            return channel_id in self.config.slack_allowed_channels
+
+        if platform == "feishu":
+            if sender_id in self.config.feishu_allowed_users:
+                return True
+            return channel_id in self.config.feishu_allowed_chats
+
+        # Unknown platform: trust only admins
+        return False
+
+    # --- Handlers ---
diff --git a/connector/router_execution_handlers.py b/connector/router_execution_handlers.py
new file mode 100644
index 0000000..0b76534
--- /dev/null
+++ b/connector/router_execution_handlers.py
@@ -0,0 +1,217 @@
+"""Owned run and interrupt command-family mixin."""
+
+# ruff: noqa: UP006, UP035 -- preserve the frozen public annotations.
+# mypy: disable-error-code="attr-defined,no-any-return"
+
+import logging
+from typing import Any, Dict, List
+
+from .contract import CommandRequest, CommandResponse
+
+logger = logging.getLogger(__name__)
+
+
+class RouterExecutionMixin:
+    async def _handle_run(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        if not args:
+            return CommandResponse(
+                text="Usage: /run <template_id> [prompt text] [key=value ...] [--approval]"
+            )
+
+        # Parse flags
+        explicit_approval = False
+        clean_args = []
+        for arg in args:
+            if arg in ("--require-approval", "--approval", "-a"):
+                explicit_approval = True
+            else:
+                clean_args.append(arg)
+
+        if not clean_args:
+            return CommandResponse(text="Usage: /run <template_id> ...")
+
+        template_id = clean_args[0]
+        inputs: Dict[str, str] = {}
+        free_text_parts: List[str] = []
+        for arg in clean_args[1:]:
+            if "=" in arg:
+                k, v = arg.split("=", 1)
+                inputs[k.strip()] = v.strip()
+            else:
+                free_text_parts.append(arg)
+
+        # If user provided free text without key=value, treat it as the prompt.
+        # We map it to a best-effort prompt key (prefers template metadata if available).
+        if free_text_parts:
+            prompt_key = await self._resolve_prompt_key(template_id)
+            if prompt_key not in inputs:
+                inputs[prompt_key] = " ".join(free_text_parts).strip()
+            elif self.config.debug:
+                logger.info(
+                    "DEBUG /run free-text ignored (prompt key already set): %s",
+                    prompt_key,
+                )
+
+        # NOTE: Debug-only payload logging for troubleshooting prompt mismatches.
+        # Enable with OPENCLAW_CONNECTOR_DEBUG=1 to log template_id + inputs.
+        if self.config.debug:
+            logger.info(
+                "DEBUG /run payload: template=%s inputs=%s approval_flag=%s trusted=%s",
+                template_id,
+                inputs,
+                explicit_approval,
+                self._is_trusted(req),
+            )
+
+        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
+        )
+        if res.get("ok"):
+            data = res.get("data", {})
+            trace_id = data.get("trace_id", "unknown")
+
+            if data.get("pending"):
+                approval_id = data.get("approval_id", "unknown")
+                msg = f"[Approval Requested]\nID: {approval_id}\nTrace: {trace_id}"
+                if "expires_at" in data:
+                    msg += f"\nExpires: {data['expires_at']}"
+                if self.poller:
+                    # IMPORTANT:
+                    # For untrusted users, approvals are done in the OpenClaw UI.
+                    # We must start tracking the approval_id so we can map
+                    # approval_id -> executed_prompt_id later and auto-deliver images.
+                    self.poller.track_approval(
+                        approval_id,
+                        req.platform,
+                        req.channel_id,
+                        req.sender_id,
+                        delivery_context=self._delivery_context(req),
+                    )
+                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,
+                        delivery_context=self._delivery_context(req),
+                    )
+
+                return CommandResponse(
+                    text=f"[Job Submitted]\nID: {prompt_id}\nTemplate: {template_id}\nTrace: {trace_id}"
+                )
+        else:
+            err = res.get("error", "Unknown error")
+            return CommandResponse(text=f"[Submission Failed] Reason: {err}")
+
+    async def _resolve_prompt_key(self, template_id: str) -> str:
+        """
+        Best-effort prompt key resolution.
+        Prefer template metadata (allowed_inputs), then fall back to common names.
+        """
+        meta = await self._get_template_meta(template_id)
+        allowed = meta.get("allowed_inputs") or []
+
+        # If template explicitly declares a single input, use it.
+        if isinstance(allowed, list) and len(allowed) == 1:
+            return str(allowed[0])
+
+        preferred = ("positive_prompt", "prompt", "text", "positive", "caption")
+        if isinstance(allowed, list):
+            for key in preferred:
+                if key in allowed:
+                    return key
+
+        # Default fallback
+        return "positive_prompt"
+
+    async def _get_template_meta(self, template_id: str) -> Dict[str, Any]:
+        if template_id in self._template_meta_cache:
+            return self._template_meta_cache[template_id]
+        try:
+            res = await self.client.get_templates()
+            if res.get("ok"):
+                for item in res.get("templates", []) or []:
+                    if item.get("id") == template_id:
+                        self._template_meta_cache[template_id] = item
+                        return item
+        except Exception as e:
+            if self.config.debug:
+                logger.info(f"DEBUG template meta fetch failed: {e}")
+        return {}
+
+    async def _handle_interrupt(
+        self, req: CommandRequest, args: List[str]
+    ) -> CommandResponse:
+        # F32 WP3: Guard
+        if err := self._require_admin_token_configured():
+            return err
+
+        targets = self._parse_stop_targets(args)
+        if not targets:
+            res = await self.client.interrupt_output()
+            if res.get("ok"):
+                return CommandResponse(text="[Stop] Global Interrupt sent to ComfyUI.")
+            return CommandResponse(text=f"[Stop Failed] {res.get('error')}")
+
+        if len(targets) == 1:
+            job_id = targets[0]
+            res = await self.client.cancel_job(job_id)
+            if res.get("ok"):
+                return CommandResponse(
+                    text=f"[Stop] Cancellation requested for job {job_id}."
+                )
+
+            # IMPORTANT: Targeted stops must never degrade to no-payload global
+            # interrupt. Older-host fallback is allowed only with prompt_id set.
+            if self._jobs_cancel_unsupported(res):
+                fallback = await self.client.interrupt_output(prompt_id=job_id)
+                if fallback.get("ok"):
+                    return CommandResponse(
+                        text=(
+                            f"[Stop] Targeted interrupt sent for job {job_id} "
+                            "(jobs cancel unsupported)."
+                        )
+                    )
+                return CommandResponse(text=f"[Stop Failed] {fallback.get('error')}")
+
+            return CommandResponse(text=f"[Stop Failed] {res.get('error')}")
+
+        res = await self.client.cancel_jobs(targets)
+        if res.get("ok"):
+            return CommandResponse(
+                text=f"[Stop] Cancellation requested for {len(targets)} jobs."
+            )
+        return CommandResponse(text=f"[Stop Failed] {res.get('error')}")
+
+    @staticmethod
+    def _parse_stop_targets(args: List[str]) -> List[str]:
+        targets: List[str] = []
+        for arg in args:
+            for part in str(arg).split(","):
+                target = part.strip()
+                if target:
+                    targets.append(target)
+        return targets
+
+    @staticmethod
+    def _jobs_cancel_unsupported(res: Dict[str, Any]) -> bool:
+        status = res.get("status")
+        if status in (404, 405, 501):
+            return True
+        error = str(res.get("error", "")).lower()
+        unsupported_markers = (
+            "404",
+            "not found",
+            "method not allowed",
+            "unsupported",
+            "not implemented",
+        )
+        return any(marker in error for marker in unsupported_markers)
diff --git a/scripts/verify_connector_router_contract.py b/scripts/verify_connector_router_contract.py
new file mode 100644
index 0000000..bd8a96e
--- /dev/null
+++ b/scripts/verify_connector_router_contract.py
@@ -0,0 +1,131 @@
+"""Verify the frozen R222 CommandRouter facade and command contract."""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import hashlib
+import inspect
+import json
+import sys
+import textwrap
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+CONTRACT_PATH = ROOT / "tests" / "connector_router_contract_r222.json"
+
+
+def _canonical_json(value: Any) -> str:
+    return json.dumps(value, indent=2, sort_keys=True, ensure_ascii=False) + "\n"
+
+
+def _command_table() -> list[dict[str, Any]]:
+    from connector.router import CommandRouter
+
+    tree = ast.parse(textwrap.dedent(inspect.getsource(CommandRouter.handle)))
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Assign) and any(
+            isinstance(target, ast.Name) and target.id == "handlers"
+            for target in node.targets
+        ):
+            if not isinstance(node.value, ast.Dict):
+                break
+            entries = []
+            for key, value in zip(node.value.keys, node.value.values, strict=True):
+                if not isinstance(key, ast.Tuple) or not isinstance(value, ast.Tuple):
+                    raise RuntimeError("invalid command table entry")
+                aliases = [ast.literal_eval(item) for item in key.elts]
+                handler_attr = value.elts[0]
+                command_class = value.elts[1]
+                if not isinstance(handler_attr, ast.Attribute) or not isinstance(
+                    command_class, ast.Attribute
+                ):
+                    raise RuntimeError("invalid command table target")
+                entries.append(
+                    {
+                        "aliases": aliases,
+                        "handler": handler_attr.attr,
+                        "class": command_class.attr,
+                    }
+                )
+            return entries
+    raise RuntimeError("CommandRouter.handle command table not found")
+
+
+def build_contract() -> dict[str, Any]:
+    from connector.router import CommandRouter
+
+    method_names = sorted(
+        {
+            name
+            for owner in CommandRouter.__mro__
+            if owner is not object
+            for name, value in owner.__dict__.items()
+            if callable(value)
+            and (name == "handle" or name.startswith("_"))
+            and name != "_build_llm_client"
+        }
+    )
+    digests = {}
+    for filename in (
+        "api_config_contract_r221.json",
+        "api_route_contract_r220.json",
+    ):
+        path = ROOT / "tests" / filename
+        digests[filename] = hashlib.sha256(path.read_bytes()).hexdigest()
+    return {
+        "schema_version": 1,
+        "constructor_signature": str(inspect.signature(CommandRouter)),
+        "facade_signatures": {
+            name: str(inspect.signature(getattr(CommandRouter, name)))
+            for name in method_names
+        },
+        "command_table": _command_table(),
+        "instance_ownership": [
+            "config",
+            "client",
+            "poller",
+            "state",
+            "_template_meta_cache",
+            "_rate_limiter",
+            "semantic_guard",
+            "command_firewall",
+        ],
+        "response_matrix_owners": [
+            "tests.connector.test_router_hotspot_r181",
+            "tests.connector.test_r214_jobs_command",
+            "tests.connector.test_router_admin",
+            "tests.connector.test_router_command_authz_r80",
+            "tests.connector.test_security",
+            "tests.connector.test_chat",
+            "tests.connector.test_chat_integration",
+            "tests.connector.test_media_delivery",
+            "tests.chat_connector.test_router_phase2",
+            "tests.chat_connector.test_router_phase3",
+        ],
+        "upstream_contract_digests": digests,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--write-baseline", action="store_true")
+    args = parser.parse_args()
+    actual = build_contract()
+    if args.write_baseline:
+        CONTRACT_PATH.write_text(_canonical_json(actual), encoding="utf-8")
+        print(f"CONNECTOR-ROUTER-CONTRACT-WRITTEN: {CONTRACT_PATH}")
+        return 0
+    expected = json.loads(CONTRACT_PATH.read_text(encoding="utf-8"))
+    if actual != expected:
+        print("CONNECTOR-ROUTER-CONTRACT-FAIL")
+        return 1
+    print("CONNECTOR-ROUTER-CONTRACT-PASS")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/connector_router_contract_r222.json b/tests/connector_router_contract_r222.json
new file mode 100644
index 0000000..0eebee5
--- /dev/null
+++ b/tests/connector_router_contract_r222.json
@@ -0,0 +1,172 @@
+{
+  "command_table": [
+    {
+      "aliases": [
+        "/status",
+        "status"
+      ],
+      "class": "PUBLIC",
+      "handler": "_handle_status"
+    },
+    {
+      "aliases": [
+        "/help",
+        "help",
+        "/start"
+      ],
+      "class": "PUBLIC",
+      "handler": "_handle_help"
+    },
+    {
+      "aliases": [
+        "/run",
+        "run"
+      ],
+      "class": "RUN",
+      "handler": "_handle_run"
+    },
+    {
+      "aliases": [
+        "/interrupt",
+        "interrupt",
+        "/cancel",
+        "cancel",
+        "/stop"
+      ],
+      "class": "ADMIN",
+      "handler": "_handle_interrupt"
+    },
+    {
+      "aliases": [
+        "/approvals",
+        "approvals"
+      ],
+      "class": "ADMIN",
+      "handler": "_handle_approvals_list"
+    },
+    {
+      "aliases": [
+        "/approve",
+        "approve"
+      ],
+      "class": "ADMIN",
+      "handler": "_handle_approve"
+    },
+    {
+      "aliases": [
+        "/reject",
+        "reject"
+      ],
+      "class": "ADMIN",
+      "handler": "_handle_reject"
+    },
+    {
+      "aliases": [
+        "/schedules",
+        "schedules"
+      ],
+      "class": "ADMIN",
+      "handler": "_handle_schedules_list"
+    },
+    {
+      "aliases": [
+        "/schedule",
+        "schedule"
+      ],
+      "class": "ADMIN",
+      "handler": "_handle_schedule_subcommand"
+    },
+    {
+      "aliases": [
+        "/history",
+        "history"
+      ],
+      "class": "PUBLIC",
+      "handler": "_handle_history"
+    },
+    {
+      "aliases": [
+        "/trace",
+        "trace"
+      ],
+      "class": "ADMIN",
+      "handler": "_handle_trace"
+    },
+    {
+      "aliases": [
+        "/jobs",
+        "jobs",
+        "queue"
+      ],
+      "class": "ADMIN",
+      "handler": "_handle_jobs"
+    },
+    {
+      "aliases": [
+        "/chat",
+        "chat"
+      ],
+      "class": "PUBLIC",
+      "handler": "_handle_chat"
+    }
+  ],
+  "constructor_signature": "(config: connector.config.ConnectorConfig, client: connector.openclaw_client.OpenClawClient, poller: 'ResultsPoller' = None)",
+  "facade_signatures": {
+    "__init__": "(self, config: connector.config.ConnectorConfig, client: connector.openclaw_client.OpenClawClient, poller: 'ResultsPoller' = None)",
+    "_chat_general": "(self, llm: connector.llm_client.LLMClient, message: str, trust_level: str) -> connector.contract.CommandResponse",
+    "_chat_run": "(self, llm: connector.llm_client.LLMClient, request: str, trust_level: str) -> connector.contract.CommandResponse",
+    "_chat_status": "(self, llm: connector.llm_client.LLMClient) -> connector.contract.CommandResponse",
+    "_chat_template": "(self, llm: connector.llm_client.LLMClient, request: str) -> connector.contract.CommandResponse",
+    "_check_command_authz": "(self, cmd: str, req: connector.contract.CommandRequest, default_class: connector.config.CommandClass) -> Optional[connector.contract.CommandResponse]",
+    "_delivery_context": "(self, req: connector.contract.CommandRequest) -> Dict[str, Any]",
+    "_get_template_meta": "(self, template_id: str) -> Dict[str, Any]",
+    "_handle_approvals_list": "(self, req: connector.contract.CommandRequest, args: List[str]) -> connector.contract.CommandResponse",
+    "_handle_approve": "(self, req: connector.contract.CommandRequest, args: List[str]) -> connector.contract.CommandResponse",
+    "_handle_chat": "(self, req: connector.contract.CommandRequest, args: List[str]) -> connector.contract.CommandResponse",
+    "_handle_help": "(self, req: connector.contract.CommandRequest, args: List[str]) -> connector.contract.CommandResponse",
+    "_handle_history": "(self, req: connector.contract.CommandRequest, args: List[str]) -> connector.contract.CommandResponse",
+    "_handle_interrupt": "(self, req: connector.contract.CommandRequest, args: List[str]) -> connector.contract.CommandResponse",
+    "_handle_jobs": "(self, req: connector.contract.CommandRequest, args: List[str]) -> connector.contract.CommandResponse",
+    "_handle_reject": "(self, req: connector.contract.CommandRequest, args: List[str]) -> connector.contract.CommandResponse",
+    "_handle_run": "(self, req: connector.contract.CommandRequest, args: List[str]) -> connector.contract.CommandResponse",
+    "_handle_schedule_subcommand": "(self, req: connector.contract.CommandRequest, args: List[str]) -> connector.contract.CommandResponse",
+    "_handle_schedules_list": "(self, req: connector.contract.CommandRequest, args: List[str]) -> connector.contract.CommandResponse",
+    "_handle_status": "(self, req: connector.contract.CommandRequest, args: List[str]) -> connector.contract.CommandResponse",
+    "_handle_trace": "(self, req: connector.contract.CommandRequest, args: List[str]) -> connector.contract.CommandResponse",
+    "_is_admin": "(self, user_id: str) -> bool",
+    "_is_trusted": "(self, req: connector.contract.CommandRequest) -> bool",
+    "_jobs_cancel_unsupported": "(res: Dict[str, Any]) -> bool",
+    "_parse_stop_targets": "(args: List[str]) -> List[str]",
+    "_policy_kv": "(contract: Dict[str, Any]) -> str",
+    "_require_admin_token_configured": "(self) -> Optional[connector.contract.CommandResponse]",
+    "_resolve_prompt_key": "(self, template_id: str) -> str",
+    "handle": "(self, req: connector.contract.CommandRequest) -> connector.contract.CommandResponse"
+  },
+  "instance_ownership": [
+    "config",
+    "client",
+    "poller",
+    "state",
+    "_template_meta_cache",
+    "_rate_limiter",
+    "semantic_guard",
+    "command_firewall"
+  ],
+  "response_matrix_owners": [
+    "tests.connector.test_router_hotspot_r181",
+    "tests.connector.test_r214_jobs_command",
+    "tests.connector.test_router_admin",
+    "tests.connector.test_router_command_authz_r80",
+    "tests.connector.test_security",
+    "tests.connector.test_chat",
+    "tests.connector.test_chat_integration",
+    "tests.connector.test_media_delivery",
+    "tests.chat_connector.test_router_phase2",
+    "tests.chat_connector.test_router_phase3"
+  ],
+  "schema_version": 1,
+  "upstream_contract_digests": {
+    "api_config_contract_r221.json": "496063d3a838c5cfa884952215049054cdf6ff030c98a5b02802ae7f64f2d57e",
+    "api_route_contract_r220.json": "17c804dc80f35ddf774e5a941ab4fd4404f55f1e37a8cafd3b3a9238a5e12c9e"
+  }
+}
diff --git a/tests/exception_boundary_policy.json b/tests/exception_boundary_policy.json
index 4ae4f4b..0e57c62 100644
--- a/tests/exception_boundary_policy.json
+++ b/tests/exception_boundary_policy.json
@@ -51,7 +51,7 @@
         }
       ]
     },
-    "connector/router.py": {
+    "connector/router_admin_handlers.py": {
       "coverage": "all_broad_catches",
       "broad_catches": [
         {
@@ -61,17 +61,27 @@
           "reason": "Standalone connector tests may stub the sanitizer import graph.",
           "regression_owner": "tests/connector/test_router_hotspot_r181.py",
           "review_after": "2027-01-11"
-        },
+        }
+      ]
+    },
+    "connector/router_dispatch.py": {
+      "coverage": "all_broad_catches",
+      "broad_catches": [
         {
-          "scope": "CommandRouter.handle",
+          "scope": "RouterDispatchMixin.handle",
           "expected_count": 1,
           "classification": "allowed_boundary_guard",
           "reason": "External command dispatch translates handler failures to a fixed operator response.",
           "regression_owner": "tests/connector/test_router_hotspot_r181.py",
           "review_after": "2027-01-11"
-        },
+        }
+      ]
+    },
+    "connector/router_execution_handlers.py": {
+      "coverage": "all_broad_catches",
+      "broad_catches": [
         {
-          "scope": "CommandRouter._get_template_meta",
+          "scope": "RouterExecutionMixin._get_template_meta",
           "expected_count": 1,
           "classification": "allowed_boundary_guard",
           "reason": "Template metadata lookup is best-effort and has an explicit prompt-name fallback.",
diff --git a/tests/static_analysis_policy.json b/tests/static_analysis_policy.json
index b3d362b..6f6ff9f 100644
--- a/tests/static_analysis_policy.json
+++ b/tests/static_analysis_policy.json
@@ -760,7 +760,7 @@
       "tool": "mypy",
       "path": "connector/platforms/slack_webhook.py",
       "code": "arg-type",
-      "message": "Argument 1 to \"_is_admin\" of \"CommandRouter\" has incompatible type \"CommandRequest\"; expected \"str\"",
+      "message": "Argument 1 to \"_is_admin\" of \"RouterDispatchMixin\" has incompatible type \"CommandRequest\"; expected \"str\"",
       "count": 1
     },
     {
@@ -924,20 +924,6 @@
       "message": "Incompatible default for parameter \"poller\" (default has type \"None\", parameter has type \"ResultsPoller\")",
       "count": 1
     },
-    {
-      "tool": "mypy",
-      "path": "connector/router.py",
-      "code": "no-any-return",
-      "message": "Returning Any from function declared to return \"dict[str, Any]\"",
-      "count": 1
-    },
-    {
-      "tool": "mypy",
-      "path": "connector/router.py",
-      "code": "unused-ignore",
-      "message": "Unused \"type: ignore\" comment",
-      "count": 1
-    },
     {
       "tool": "mypy",
       "path": "connector/state.py",
@@ -3563,132 +3549,6 @@
       "message": "Use `X | None` for type annotations",
       "count": 6
     },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "E731",
-      "message": "Do not assign a `lambda` expression, use a `def`",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "F841",
-      "message": "Local variable `requires_admin` is assigned to but never used",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "RUF010",
-      "message": "Use explicit conversion flag",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "SIM102",
-      "message": "Use a single `if` statement instead of nested `if` statements",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "SIM103",
-      "message": "Return the condition `channel_id in self.config.discord_allowed_channels` directly",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "SIM103",
-      "message": "Return the condition `channel_id in self.config.feishu_allowed_chats` directly",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "SIM103",
-      "message": "Return the condition `channel_id in self.config.line_allowed_groups` directly",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "SIM103",
-      "message": "Return the condition `channel_id in self.config.slack_allowed_channels` directly",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "SIM103",
-      "message": "Return the condition `sender_id in self.config.kakao_allowed_users` directly",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "SIM103",
-      "message": "Return the condition `sender_id in self.config.wechat_allowed_users` directly",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "SIM103",
-      "message": "Return the condition `sender_id in self.config.whatsapp_allowed_users` directly",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "SIM103",
-      "message": "Return the condition directly",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "SIM108",
-      "message": "Use ternary operator `canonical_cmd = aliases[0] if isinstance(aliases, tuple) else aliases` instead of `if`-`else`-block",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "UP006",
-      "message": "Use `dict` instead of `Dict` for type annotation",
-      "count": 8
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "UP006",
-      "message": "Use `list` instead of `List` for type annotation",
-      "count": 17
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "UP035",
-      "message": "`typing.Dict` is deprecated, use `dict` instead",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "UP035",
-      "message": "`typing.List` is deprecated, use `list` instead",
-      "count": 1
-    },
-    {
-      "tool": "ruff",
-      "path": "connector/router.py",
-      "code": "UP045",
-      "message": "Use `X | None` for type annotations",
-      "count": 2
-    },
     {
       "tool": "ruff",
       "path": "connector/security_profile.py",
diff --git a/tests/test_r219_exception_boundary_phase2.py b/tests/test_r219_exception_boundary_phase2.py
index 36df1de..3997cf9 100644
--- a/tests/test_r219_exception_boundary_phase2.py
+++ b/tests/test_r219_exception_boundary_phase2.py
@@ -36,7 +36,9 @@ class TestPolicyV2(unittest.TestCase):
             {
                 "api/route_handlers.py",
                 "api/route_orchestration.py",
-                "connector/router.py",
+                "connector/router_admin_handlers.py",
+                "connector/router_dispatch.py",
+                "connector/router_execution_handlers.py",
                 "services/route_bootstrap.py",
                 "api/config_projection_handlers.py",
                 "api/config_llm_handlers.py",
diff --git a/tests/test_r222_connector_router_decomposition.py b/tests/test_r222_connector_router_decomposition.py
new file mode 100644
index 0000000..fcec16c
--- /dev/null
+++ b/tests/test_r222_connector_router_decomposition.py
@@ -0,0 +1,117 @@
+"""Contract-first tests for R222 connector router decomposition."""
+
+from __future__ import annotations
+
+import importlib.util
+import inspect
+import json
+import unittest
+from dataclasses import FrozenInstanceError
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _verifier():
+    path = ROOT / "scripts" / "verify_connector_router_contract.py"
+    spec = importlib.util.spec_from_file_location("r222_router_contract", path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError("unable to load R222 verifier")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class TestR222ConnectorRouterDecomposition(unittest.TestCase):
+    def test_frozen_router_contract_matches_byte_for_byte(self):
+        verifier = _verifier()
+        fixture = (ROOT / "tests" / "connector_router_contract_r222.json").read_text(
+            encoding="utf-8"
+        )
+        self.assertEqual(verifier._canonical_json(verifier.build_contract()), fixture)
+
+    def test_owned_family_modules_are_substantive_and_one_way(self):
+        from connector import (
+            router_admin_handlers,
+            router_chat_handlers,
+            router_dispatch,
+            router_execution_handlers,
+        )
+
+        required = {
+            router_dispatch: (
+                router_dispatch.RouterDispatchMixin,
+                ("handle", "_check_command_authz"),
+            ),
+            router_execution_handlers: (
+                router_execution_handlers.RouterExecutionMixin,
+                ("_handle_run", "_handle_interrupt"),
+            ),
+            router_admin_handlers: (
+                router_admin_handlers.RouterAdminMixin,
+                ("_handle_jobs", "_handle_approvals_list"),
+            ),
+            router_chat_handlers: (
+                router_chat_handlers.RouterChatMixin,
+                ("_handle_chat", "_chat_general"),
+            ),
+        }
+        for module, (owner, names) in required.items():
+            source = inspect.getsource(module)
+            self.assertNotIn("import connector.router", source)
+            self.assertNotIn("from connector import router", source)
+            self.assertNotIn("from . import router", source)
+            for name in names:
+                self.assertGreater(
+                    len(inspect.getsource(getattr(owner, name)).splitlines()), 12
+                )
+
+    def test_command_aliases_are_unique_and_canonical_first(self):
+        contract = _verifier().build_contract()
+        seen = set()
+        for entry in contract["command_table"]:
+            self.assertTrue(entry["aliases"][0].startswith("/"))
+            for alias in entry["aliases"]:
+                self.assertNotIn(alias, seen)
+                seen.add(alias)
+
+    def test_dispatch_request_context_is_immutable(self):
+        from connector.config import CommandClass
+        from connector.contract import CommandRequest
+        from connector.router_dispatch import RouterRequestContext
+
+        request = CommandRequest(
+            platform="telegram",
+            channel_id="channel",
+            sender_id="sender",
+            username="user",
+            message_id="message",
+            text="/status",
+            timestamp=0.0,
+        )
+        context = RouterRequestContext(
+            request=request,
+            parsed_command="/status",
+            canonical_command="/status",
+            args=(),
+            command_class=CommandClass.PUBLIC,
+        )
+        self.assertIsInstance(context.args, tuple)
+        with self.assertRaises(FrozenInstanceError):
+            context.canonical_command = "/run"
+
+    def test_upstream_route_and_config_contracts_are_unchanged(self):
+        verifier = _verifier()
+        expected = json.loads(
+            (ROOT / "tests" / "connector_router_contract_r222.json").read_text(
+                encoding="utf-8"
+            )
+        )
+        self.assertEqual(
+            verifier.build_contract()["upstream_contract_digests"],
+            expected["upstream_contract_digests"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()