feat: 中断 router 分发（✖ on thinking chip）

acp_runtime.py

@@ -7,12 +7,28 @@
 from __future__ import annotations
 
 import os
+import signal
 import subprocess
+import threading
 
-from agent_runtime import AgentError, clean
+from agent_runtime import AgentError, clean, _killpg_safe
 
 ACP_AGENT_TIMEOUT = int(os.environ.get("OPENFORGE_ACP_AGENT_TIMEOUT", "240"))
 
+# chip_post_id → pgid for in-flight ACP CLI invocations. Used by the
+# cancel endpoint to SIGTERM the subprocess group on ✖ click.
+_acp_chip_to_pgid: dict[str, int] = {}
+_acp_chip_to_pgid_lock = threading.Lock()
+
+
+def cancel_acp_chip_subprocess(chip_post_id: str) -> bool:
+    with _acp_chip_to_pgid_lock:
+        pgid = _acp_chip_to_pgid.get(chip_post_id)
+    if pgid is None:
+        return False
+    _killpg_safe(pgid, signal.SIGTERM)
+    return True
+
 
 def _argv_for_agent(agent_id: str, prompt: str) -> list[str]:
     if agent_id == "codex":
@@ -33,6 +49,7 @@ def call_acp_agent(
     prompt: str,
     extra_env: dict[str, str] | None = None,
     cwd: str | None = None,
+    chip_post_id: str | None = None,
 ) -> str:
     """Invoke an ACP-backed CLI employee in oneshot mode.
 
@@ -47,26 +64,45 @@ def call_acp_agent(
         if cleaned:
             spawn_env = {**os.environ, **cleaned}
     argv = _argv_for_agent(agent_id, prompt)
+    proc: subprocess.Popen | None = None
+    pgid: int | None = None
     try:
-        proc = subprocess.run(
-            argv,
-            cwd=run_cwd,
-            env=spawn_env,
-            capture_output=True,
-            text=True,
-            timeout=ACP_AGENT_TIMEOUT,
-        )
-    except FileNotFoundError:
-        raise AgentError(f"ACP CLI not found for {agent_id}: {argv[0]}") from None
-    except subprocess.TimeoutExpired:
-        raise AgentError(f"ACP {agent_id} timeout after {ACP_AGENT_TIMEOUT}s") from None
-
-    if proc.returncode != 0:
-        tail = (proc.stderr or proc.stdout or "").strip().splitlines()[-3:]
-        raise AgentError(
-            f"ACP {agent_id} exited {proc.returncode}: " + " | ".join(tail)
-        )
-    out = clean(proc.stdout or "")
-    if not out:
-        raise AgentError(f"ACP {agent_id} produced no output")
-    return out
+        try:
+            proc = subprocess.Popen(
+                argv,
+                cwd=run_cwd,
+                env=spawn_env,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                start_new_session=True,
+            )
+        except FileNotFoundError:
+            raise AgentError(f"ACP CLI not found for {agent_id}: {argv[0]}") from None
+        pgid = proc.pid
+        if chip_post_id:
+            with _acp_chip_to_pgid_lock:
+                _acp_chip_to_pgid[chip_post_id] = pgid
+        try:
+            stdout, stderr = proc.communicate(timeout=ACP_AGENT_TIMEOUT)
+        except subprocess.TimeoutExpired:
+            _killpg_safe(pgid, signal.SIGTERM)
+            try:
+                proc.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                _killpg_safe(pgid, signal.SIGKILL)
+            raise AgentError(f"ACP {agent_id} timeout after {ACP_AGENT_TIMEOUT}s") from None
+        returncode = proc.returncode
+        if returncode != 0:
+            tail = (stderr or stdout or "").strip().splitlines()[-3:]
+            raise AgentError(
+                f"ACP {agent_id} exited {returncode}: " + " | ".join(tail)
+            )
+        out = clean(stdout or "")
+        if not out:
+            raise AgentError(f"ACP {agent_id} produced no output")
+        return out
+    finally:
+        if chip_post_id:
+            with _acp_chip_to_pgid_lock:
+                _acp_chip_to_pgid.pop(chip_post_id, None)

agent_runtime.py

@@ -177,6 +177,43 @@ class AgentError(RuntimeError):
 _active_procs: dict[int, subprocess.Popen] = {}
 _active_procs_lock = _threading.Lock()
 
+# Map chip_post_id → pgid for in-flight agent turns. Lets the cancel
+# endpoint (POST /api/.../posts/<chip_id>/cancel) SIGTERM the specific
+# subprocess group when a user clicks ✖ on a thinking/running chip.
+_chip_to_pgid: dict[str, int] = {}
+_chip_to_pgid_lock = _threading.Lock()
+
+
+def _bind_chip(chip_post_id: str, pgid: int) -> None:
+    if not chip_post_id:
+        return
+    with _chip_to_pgid_lock:
+        _chip_to_pgid[chip_post_id] = pgid
+
+
+def _unbind_chip(chip_post_id: str) -> None:
+    if not chip_post_id:
+        return
+    with _chip_to_pgid_lock:
+        _chip_to_pgid.pop(chip_post_id, None)
+
+
+def cancel_chip_subprocess(chip_post_id: str) -> bool:
+    """SIGTERM the subprocess group bound to ``chip_post_id`` if any.
+
+    Returns True iff a live process group was signalled. The worker thread
+    itself owns lifecycle cleanup (unregister + chip patching) — we only
+    nudge the subprocess so the worker's ``communicate()`` returns fast and
+    the cancellation can be observed by the router (which then drops the
+    reply). Idempotent: missing/already-gone groups are silent.
+    """
+    with _chip_to_pgid_lock:
+        pgid = _chip_to_pgid.get(chip_post_id)
+    if pgid is None:
+        return False
+    _killpg_safe(pgid, signal.SIGTERM)
+    return True
+
 
 def _register_active(pgid: int, proc: subprocess.Popen) -> None:
     with _active_procs_lock:
@@ -271,7 +308,7 @@ def _killpg_safe(pgid: int, sig: int) -> None:
         pass
 
 
-def call_agent(agent_id: str, session_id: str, prompt: str, extra_env: dict[str, str] | None = None) -> str:
+def call_agent(agent_id: str, session_id: str, prompt: str, extra_env: dict[str, str] | None = None, chip_post_id: str | None = None) -> str:
     """Invoke `openclaw agent --local --json`. Raises AgentError on failure.
 
     --local keeps the run fully sandboxed in a subprocess so
@@ -298,7 +335,7 @@ def call_agent(agent_id: str, session_id: str, prompt: str, extra_env: dict[str,
     from forge_employees import acp_employee_ids
     if agent_id in acp_employee_ids():
         from acp_runtime import call_acp_agent
-        return call_acp_agent(agent_id, session_id, prompt, extra_env)
+        return call_acp_agent(agent_id, session_id, prompt, extra_env, chip_post_id=chip_post_id)
 
     argv = [
         OPENCLAW_BIN, "agent",
@@ -327,10 +364,13 @@ def call_agent(agent_id: str, session_id: str, prompt: str, extra_env: dict[str,
 
     pgid = proc.pid  # equals process group id thanks to start_new_session
     _register_active(pgid, proc)
+    if chip_post_id:
+        _bind_chip(chip_post_id, pgid)
     try:
         stdout, stderr = proc.communicate(timeout=AGENT_TIMEOUT + 30)
     except subprocess.TimeoutExpired:
         _unregister_active(pgid)
+        _unbind_chip(chip_post_id or "")
         _killpg_safe(pgid, signal.SIGTERM)
         try:
             proc.wait(timeout=_GROUP_KILL_GRACE_SECONDS)
@@ -352,6 +392,7 @@ def call_agent(agent_id: str, session_id: str, prompt: str, extra_env: dict[str,
         ) from None
 
     _unregister_active(pgid)
+    _unbind_chip(chip_post_id or "")
     if proc.returncode != 0:
         tail = (stderr or stdout or "").strip().splitlines()[-3:]
         raise AgentError(

post_router.py

@@ -58,7 +58,7 @@
 
 # ─── config ──────────────────────────────────────────────────────────
 ROUTER_SPEAKER_FALLBACK = "__router__"
-STATUS_PHASES = {"thinking", "running", "done", "failed", "skipped"}
+STATUS_PHASES = {"thinking", "running", "done", "failed", "skipped", "cancelled"}
 ERROR_TAIL_LIMIT = 2048
 
 # Concurrency cap across all in-flight agent subprocesses. --local agent
@@ -72,6 +72,39 @@
 _inflight: set[tuple[str, str]] = set()
 _inflight_lock = threading.Lock()
 
+# Set of chip_post_ids that have been cancelled by the user. The worker
+# checks this BEFORE writing the agent's reply post and, if present,
+# discards the reply (so a hung subprocess that finally returns can't
+# pollute the thread after cancel).
+_cancelled_chips: set[str] = set()
+_cancelled_chips_lock = threading.Lock()
+
+
+def mark_cancelled(chip_post_id: str) -> None:
+    if not chip_post_id:
+        return
+    with _cancelled_chips_lock:
+        _cancelled_chips.add(chip_post_id)
+
+
+def _consume_cancelled(chip_post_id: str) -> bool:
+    with _cancelled_chips_lock:
+        return _cancelled_chips.discard(chip_post_id) or False  # noqa
+
+
+def is_cancelled(chip_post_id: str) -> bool:
+    if not chip_post_id:
+        return False
+    with _cancelled_chips_lock:
+        return chip_post_id in _cancelled_chips
+
+
+def _clear_cancelled(chip_post_id: str) -> None:
+    if not chip_post_id:
+        return
+    with _cancelled_chips_lock:
+        _cancelled_chips.discard(chip_post_id)
+
 # Set to True by drain_and_terminate() when forge is shutting down.
 # Once set, enqueue_if_needed() refuses to spawn new workers — those
 # triggers will be picked up as orphan placeholders on the next boot
@@ -529,13 +562,34 @@ def _route_to_agent(thread_id: str, agent_id: str, trigger: dict,
                 _patch_chip(thread_id, placeholder_id, phase="running")
             if agent_id in forge_employees.acp_employee_ids():
                 prompt = _render_acp_preamble(thread_id, agent_id, trigger) + prompt
-            reply = call_agent(agent_id, session_id, prompt, extra_env=spawn_env)
+            reply = call_agent(agent_id, session_id, prompt, extra_env=spawn_env, chip_post_id=placeholder_id)
         except AgentError as e:
+            # If the user cancelled this chip mid-flight, the SIGTERM we
+            # sent surfaces here as a non-zero exit. Convert to phase
+            # 'cancelled' so the UI matches intent (no misleading 'failed').
+            if is_cancelled(placeholder_id or ""):
+                if placeholder_id:
+                    _patch_chip(thread_id, placeholder_id,
+                                phase="cancelled",
+                                duration_ms=_duration_ms(started))
+                _clear_cancelled(placeholder_id or "")
+                return
             if placeholder_id:
                 _patch_chip(thread_id, placeholder_id,
                             phase="failed", error=_error_tail(e),
                             duration_ms=_duration_ms(started))
             return
+        # Late cancel: subprocess produced a reply but the user clicked
+        # the cancel button while we were rendering. Drop the reply
+        # silently — router is the single source of truth for what lands
+        # in the thread, so a hung agent can never pollute it post-cancel.
+        if is_cancelled(placeholder_id or ""):
+            if placeholder_id:
+                _patch_chip(thread_id, placeholder_id,
+                            phase="cancelled",
+                            duration_ms=_duration_ms(started))
+            _clear_cancelled(placeholder_id or "")
+            return
         reply = clean(reply)
         if is_empty(reply):
             if placeholder_id:

server.py

@@ -98,7 +98,7 @@
 SQUAD_ROUTE_RE = r"([\w-]{1,32})"
 THREAD_ROUTE_RE = r"(th_[0-9a-f]+_[0-9a-f]+)"
 POST_ID_ROUTE_RE = r"(p_[A-Za-z0-9_]+)"
-STATUS_PHASES = {"thinking", "running", "done", "failed", "skipped"}
+STATUS_PHASES = {"thinking", "running", "done", "failed", "skipped", "cancelled"}
 
 
 def _is_local(host: str) -> bool:
@@ -1316,6 +1316,83 @@ def do_POST(self):
             self._json(_serializable_post(updated))
             return
 
+        m = re.match(rf"^/api/threads/{THREAD_ROUTE_RE}/posts/{POST_ID_ROUTE_RE}/cancel$", url.path)
+        if m:
+            tid = m.group(1)
+            pid = m.group(2)
+            thread = store.project_thread(tid)
+            if thread is None:
+                self._json({"error": "unknown thread"}, 404)
+                return
+            post = (thread.get("posts_by_id") or {}).get(pid)
+            if post is None:
+                self._json({"error": "unknown post"}, 404)
+                return
+            if post.get("post_type") != "status_chip":
+                self._json({"error": "post must be status_chip"}, 400)
+                return
+            current_phase = post.get("phase")
+            # Idempotent: already cancelled is a no-op success; terminal
+            # phases (done/failed/skipped/cancelled) cannot be cancelled.
+            if current_phase == "cancelled":
+                self._json(_serializable_post(post))
+                return
+            if current_phase in ("done", "failed", "skipped"):
+                self._json({
+                    "error": "already_completed",
+                    "phase": current_phase,
+                }, 409)
+                return
+            # 1) Mark the chip as cancelled so the worker discards any
+            #    reply the subprocess may still emit.
+            post_router.mark_cancelled(pid)
+            # 2) Best-effort SIGTERM the subprocess group so the worker
+            #    returns fast. Either side independently lands the
+            #    cancellation; router-side suppression is the guarantee.
+            try:
+                from agent_runtime import cancel_chip_subprocess
+                cancel_chip_subprocess(pid)
+            except Exception:
+                pass
+            try:
+                from acp_runtime import cancel_acp_chip_subprocess
+                cancel_acp_chip_subprocess(pid)
+            except Exception:
+                pass
+            # 3) Patch chip phase to 'cancelled' immediately. The worker,
+            #    when it returns, will see is_cancelled() and noop on
+            #    chip patching/reply writing.
+            try:
+                updated = store.patch_post(tid, pid, {
+                    "phase": "cancelled",
+                    "error": None,
+                })
+            except ValueError as e:
+                self._json({"error": str(e)}, 400)
+                return
+            # 4) Audit post so anyone reading the thread later understands
+            #    why no reply appeared.
+            try:
+                agent_id = post.get("agent_id")
+                if not agent_id:
+                    import re as _re
+                    mm = _re.match(r"^([a-z][a-z0-9_-]*)\s+thinking$",
+                                   (post.get("content") or "").strip(),
+                                   _re.IGNORECASE)
+                    if mm:
+                        agent_id = mm.group(1)
+                trigger_pid = post.get("trigger_post_id") or post.get("parent_post_id")
+                display = agent_id or "agent"
+                store.add_thread_post(
+                    tid, post_router.ROUTER_SPEAKER_FALLBACK,
+                    f"⛔ {display} 的本轮回复已被中断。",
+                    parent_post_id=trigger_pid,
+                )
+            except Exception as e:
+                print(f"⚠️  cancel audit-post failed: {e!r}", flush=True)
+            self._json(_serializable_post(updated))
+            return
+
         m = re.match(rf"^/api/threads/{THREAD_ROUTE_RE}/posts/{POST_ID_ROUTE_RE}/reactions$", url.path)
         if m:
             tid = m.group(1)