diff --git a/connector/openclaw_client.py b/connector/openclaw_client.py index bcf6baf..5bf5923 100644 --- a/connector/openclaw_client.py +++ b/connector/openclaw_client.py @@ -7,6 +7,7 @@ import json import logging import uuid from typing import Optional +from urllib.parse import quote from .config import ConnectorConfig @@ -68,7 +69,7 @@ class OpenClawClient: async with session.request( method, url, headers=self.headers, json=json_data, timeout=timeout ) as resp: - result = {"ok": resp.status in (200, 201, 202)} + result = {"ok": resp.status in (200, 201, 202), "status": resp.status} try: data = await resp.json() @@ -166,9 +167,17 @@ class OpenClawClient: } return await self._request("POST", "/openclaw/triggers/fire", data) - async def interrupt_output(self) -> dict: - # Remediation: Cancel -> Interrupt (Global) - return await self._request("POST", "/api/interrupt", {}) + async def cancel_job(self, job_id: str) -> dict: + encoded_job_id = quote(str(job_id), safe="") + return await self._request("POST", f"/api/jobs/{encoded_job_id}/cancel", {}) + + async def cancel_jobs(self, job_ids: list[str]) -> dict: + return await self._request("POST", "/api/jobs/cancel", {"job_ids": job_ids}) + + async def interrupt_output(self, prompt_id: Optional[str] = None) -> dict: + # No prompt_id means explicit global interrupt. A prompt_id is targeted. + payload = {"prompt_id": str(prompt_id)} if prompt_id else {} + return await self._request("POST", "/api/interrupt", payload) async def get_view( self, filename: str, subfolder: str = "", type: str = "output" diff --git a/connector/router.py b/connector/router.py index 258460e..da7d579 100644 --- a/connector/router.py +++ b/connector/router.py @@ -498,13 +498,68 @@ class CommandRouter: if err := self._require_admin_token_configured(): return err - # Remediation: Global Interrupt - res = await self.client.interrupt_output() - if res.get("ok"): - return CommandResponse(text="[Stop] Global Interrupt sent to ComfyUI.") - else: + 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: @@ -679,7 +734,7 @@ class CommandRouter: "OpenClaw Connector\n" "/status - Check system health and queue\n" "/run