feat(executions): propagate cancelled terminal status end-to-end (#679)

docker/base-image/agent_server/routers/chat.py

@@ -157,12 +157,49 @@ async def execute_task(request: ParallelTaskRequest):
             persist_session=bool(request.persist_session),  # Session tab: write JSONL for future --resume
             images=request.images,  # Vision images from channel adapters (#562)
         )
+    except HTTPException as exc:
+        # #679 (F3): a terminated turn that surfaces a non-auth/non-rate terminal
+        # is a user cancel, not a failure. SIGINT→graceful-exit-0 with no output,
+        # or SIGKILL escalation, lands here as a 504 (signal exits) / 502 (empty
+        # result) / 500 — relabel ALL of them to a `cancelled` 200, mirroring the
+        # async result-callback's `_is_auth_or_rate` guard. 503/429/auth and any
+        # unterminated terminal re-raise unchanged (Issue 6/C6) so SUB-003 + the
+        # AUTH dispatch breaker still fire. The label is what matters — the cancel
+        # is non-billable, so we drop metadata (dict detail → empty response).
+        is_cancel = (
+            exc.status_code not in (503, 429)
+            and bool(request.execution_id)
+            and get_process_registry().was_terminated(request.execution_id)
+        )
+        # #679 (F4): a cancel is neutral for the failure counter (never trips the
+        # dispatch breaker); a genuine failure still increments it.
+        agent_state.record_task_finish(success=None if is_cancel else False)
+        if is_cancel:
+            logger.info(f"[Task] Task {request.execution_id} cancelled by user (status {exc.status_code})")
+            return {
+                "response": exc.detail if isinstance(exc.detail, str) else "",
+                "execution_log": [],
+                "metadata": {},
+                "session_id": None,
+                "status": "cancelled",
+                "timestamp": datetime.now().isoformat(),
+            }
+        raise
     except BaseException:
         agent_state.record_task_finish(success=False)
         raise
-    agent_state.record_task_finish(success=True)
 
-    logger.info(f"[Task] Task {session_id} completed successfully")
+    # #679 graceful path: Claude catches SIGINT, emits a final message, and exits
+    # 0 — the return code can't distinguish that from genuine success. Cross-check
+    # the cancel marker, keyed off the backend `execution_id` (NEVER the returned
+    # session_id, which can differ on a resumed/forked turn). Compute BEFORE
+    # record_task_finish so the cancel is recorded neutrally (F4).
+    cancelled = bool(request.execution_id) and get_process_registry().was_terminated(request.execution_id)
+    agent_state.record_task_finish(success=None if cancelled else True)
+    if cancelled:
+        logger.info(f"[Task] Task {request.execution_id} cancelled by user")
+    else:
+        logger.info(f"[Task] Task {session_id} completed successfully")
 
     # raw_messages contains the full Claude Code JSON stream (init, assistant, user, result)
     # This is the complete execution transcript showing thinking, tool calls, and results
@@ -171,6 +208,7 @@ async def execute_task(request: ParallelTaskRequest):
         "execution_log": raw_messages,  # Full JSON transcript from Claude Code
         "metadata": metadata.model_dump(),
         "session_id": session_id,
+        "status": "cancelled" if cancelled else "success",
         "timestamp": datetime.now().isoformat()
     }
 

docker/base-image/agent_server/services/process_registry.py

@@ -31,6 +31,13 @@
 # with comfortable headroom for backend slowness.
 RECENTLY_COMPLETED_TTL_SECONDS = 300  # 5 minutes
 
+# Issue #679: window during which a just-terminated execution is still surfaced
+# as "cancelled by user" to the chat handler / async result callback. Mirrors
+# RECENTLY_COMPLETED_TTL_SECONDS — the marker is best-effort observability (the
+# DB CANCELLED write is the durable authority), so it just needs to comfortably
+# outlast the gap between a SIGINT send and the turn's graceful exit + finalize.
+TERMINATED_TTL_SECONDS = 300  # 5 minutes
+
 
 class ProcessRegistry:
     """
@@ -59,6 +66,12 @@ def __init__(self):
         # read. Capped indirectly by traffic — at agent's ~10 concurrent
         # max with 5 min retention this is a few dozen entries at peak.
         self._recently_completed: Dict[str, float] = {}
+        # Issue #679: execution_id -> unix timestamp when terminate() sent a
+        # kill-signal. Read (not popped) by was_terminated() so the chat handler
+        # and async result callback can label a graceful-exit-0 / SIGKILL→504
+        # turn as "cancelled". Lazy TTL eviction on read; cleared on register()
+        # so a reused execution_id (#678 in-line retry) can't inherit the label.
+        self._terminated: Dict[str, float] = {}
 
     def register(self, execution_id: str, process: subprocess.Popen, metadata: dict = None):
         """
@@ -78,6 +91,10 @@ def register(self, execution_id: str, process: subprocess.Popen, metadata: dict
             # Initialize log streaming structures
             self._log_subscribers[execution_id] = []
             self._log_buffers[execution_id] = []
+            # Issue #679 (C10): clear any stale cancel marker so an execution_id
+            # reused by the #678 in-line reader-race retry can't inherit the
+            # previous attempt's "cancelled" label.
+            self._terminated.pop(execution_id, None)
             logger.info(f"[ProcessRegistry] Registered execution {execution_id}")
 
     def unregister(self, execution_id: str):
@@ -150,6 +167,15 @@ def terminate(self, execution_id: str, graceful_timeout: int = 5) -> dict:
             logger.info(f"[ProcessRegistry] Sending SIGINT to execution {execution_id} (process group)")
             _signal_process_tree(process, signal.SIGINT, pgid=pgid)
 
+            # Issue #679: record the cancel marker immediately after a successful
+            # SIGINT send — still-running branch only (NOT already_finished /
+            # not_found, handled above; NOT on signal-failure, which raises into
+            # the `except` below and skips this). The send is causally before the
+            # subprocess's graceful exit, so was_terminated() is set before the
+            # chat handler observes execute_headless returning — race-free.
+            with self._lock:
+                self._terminated[execution_id] = time.time()
+
             try:
                 process.wait(timeout=graceful_timeout)
                 logger.info(f"[ProcessRegistry] Execution {execution_id} terminated gracefully")
@@ -300,6 +326,22 @@ def list_recently_completed_ids(self) -> List[str]:
                 del self._recently_completed[eid]
             return list(self._recently_completed.keys())
 
+    def was_terminated(self, execution_id: str) -> bool:
+        """Issue #679: True if terminate() sent a kill-signal for this execution
+        within the last TERMINATED_TTL_SECONDS.
+
+        Read-only — it does NOT consume the marker, so the graceful-exit relabel
+        path and a later SIGKILL→504 check both observe it. Expired entries are
+        dropped lazily here (mirrors list_recently_completed_ids); no separate
+        sweeper needed.
+        """
+        cutoff = time.time() - TERMINATED_TTL_SECONDS
+        with self._lock:
+            expired = [eid for eid, ts in self._terminated.items() if ts < cutoff]
+            for eid in expired:
+                del self._terminated[eid]
+            return execution_id in self._terminated
+
     def cleanup_finished(self) -> int:
         """
         Remove entries for finished processes.

docker/base-image/agent_server/services/result_callback.py

@@ -30,12 +30,13 @@
 import re
 import time
 from pathlib import Path
-from typing import Any, Dict, Set
+from typing import Any, Dict, Optional, Set
 
 import httpx
 from fastapi import HTTPException
 
 from ..state import agent_state
+from .process_registry import get_process_registry
 
 logger = logging.getLogger(__name__)
 
@@ -174,6 +175,33 @@ def _envelope_from_http_exception(exc: HTTPException) -> Dict:
     }
 
 
+# ---------------------------------------------------------------------------
+# #679 cancel relabel
+# ---------------------------------------------------------------------------
+def _cancelled_override(envelope: Dict) -> Dict:
+    """Relabel a terminal envelope as a user cancel (#679).
+
+    Keeps whatever response/metadata/session_id/execution_log the turn produced
+    (Claude emits a graceful final message on SIGINT) and drops ``error_code`` so
+    the backend doesn't treat the cancel as an agent failure. The backend maps
+    ``status:"cancelled"`` → ``CANCELLED`` (never a billable success)."""
+    overridden = dict(envelope)
+    overridden["status"] = "cancelled"
+    overridden["terminal_reason"] = "cancelled"
+    overridden["error_code"] = None
+    return overridden
+
+
+def _is_auth_or_rate(envelope: Dict) -> bool:
+    """Issue 6 / C6: an auth (503) or rate-limit (429) terminal must NOT be
+    relabelled cancelled even if the execution was terminated — the backend's
+    AUTH dispatch breaker / SUB-003 still need to record it on the async path."""
+    return (
+        envelope.get("error_code") == "auth"
+        or envelope.get("terminal_reason") in ("auth", "rate_limit")
+    )
+
+
 # ---------------------------------------------------------------------------
 # Delivery
 # ---------------------------------------------------------------------------
@@ -253,12 +281,12 @@ async def _run_and_report(request, backend_url: str, mcp_key: str, dispatch_mono
             images=request.images,
         )
         envelope = _success_envelope(response_text, raw_messages, metadata, session_id)
-        agent_state.record_task_finish(success=True)
+        finish_success: Optional[bool] = True
     except HTTPException as exc:
-        agent_state.record_task_finish(success=False)
+        finish_success = False
         envelope = _envelope_from_http_exception(exc)
     except BaseException as exc:  # noqa: BLE001 — any failure must still report a terminal
-        agent_state.record_task_finish(success=False)
+        finish_success = False
         envelope = {
             "status": "failed",
             "error": str(exc)[:500] or type(exc).__name__,
@@ -267,6 +295,26 @@ async def _run_and_report(request, backend_url: str, mcp_key: str, dispatch_mono
             "metadata": {},
         }
 
+    # #679: a graceful exit-0 OR a SIGKILL→504 for a turn the operator cancelled
+    # must be relabelled `cancelled` so the backend writes CANCELLED — never a
+    # billable success, never a breaker-counting failure. Auth/rate envelopes are
+    # excluded (C6) so the AUTH dispatch breaker + SUB-003 still record on async.
+    if (
+        execution_id
+        and get_process_registry().was_terminated(execution_id)
+        and not _is_auth_or_rate(envelope)
+    ):
+        envelope = _cancelled_override(envelope)
+        # #679 (F4) parity with the sync path: a cancel is NEUTRAL for the
+        # consecutive_failures health counter — never trips or resets the
+        # dispatch breaker (#526). Recorded after the relabel decision so the
+        # graceful-cancel (was True) and the 504/502-cancel (was False) agree.
+        finish_success = None
+
+    # Recorded once, after the cancel decision, so the health counter matches the
+    # final terminal label.
+    agent_state.record_task_finish(success=finish_success)
+
     _persist(execution_id, {"agent_name": agent_name, "envelope": envelope})
 
     # Deadline = dispatch + (agent timeout + buffer) = the slot-lease TTL window.

docker/base-image/agent_server/state.py

@@ -76,14 +76,22 @@ def record_task_start(self) -> None:
             self.active_task_count += 1
             self.last_task_at = datetime.now(timezone.utc).isoformat()
 
-    def record_task_finish(self, success: bool) -> None:
-        """Mark an execution as finished. Resets the consecutive-failure
-        counter on success, increments it on failure — this is the signal the
-        dispatch circuit breaker (#526) consumes."""
+    def record_task_finish(self, success: Optional[bool]) -> None:
+        """Mark an execution as finished. Resets the consecutive-failure counter
+        on success, increments it on failure — this is the signal the dispatch
+        circuit breaker (#526) consumes.
+
+        #679 (F4): ``success=None`` is a NEUTRAL finish (a user cancel) — the
+        task still finishes (decrement the active count, stamp last_task_at) but
+        the failure counter is left untouched. A cancel is neither a failure (it
+        must not push a healthy agent toward an open breaker) nor a success (it
+        proves nothing about agent health), so it must not reset OR increment."""
         with self._health_lock:
             if self.active_task_count > 0:
                 self.active_task_count -= 1
             self.last_task_at = datetime.now(timezone.utc).isoformat()
+            if success is None:
+                return
             if success:
                 self.consecutive_failures = 0
             else:

docs/security-reports/cso-diff-2026-06-22-679.md

@@ -0,0 +1,108 @@
+# CSO Security Audit
+
+**Mode**: diff
+**Scope**: branch `AndriiPasternak31/autoplan-issue-679` (working-tree changes vs `origin/dev`)
+**Issue**: #679 — agent chat cancellation (propagate a `cancelled` terminal status end-to-end)
+**Date**: 2026-06-22
+
+## Summary
+
+| Category | CRITICAL | HIGH | MEDIUM | LOW |
+|----------|----------|------|--------|-----|
+| Secrets | 0 | 0 | 0 | 0 |
+| Dependencies | 0 | 0 | 0 | 0 |
+| Auth Boundaries | 0 | 0 | 0 | 0 |
+| Injection | 0 | 0 | 0 | 0 |
+| Platform Patterns | 0 | 0 | 0 | 2 |
+| Configuration | 0 | 0 | 0 | 0 |
+
+## What the change does
+
+Introduces a third terminal outcome — `cancelled` — alongside `success`/`failed`, so an
+operator-initiated cancel is no longer mis-recorded as either a billable success or an agent
+failure:
+
+- **Agent server** (`agent_server/`): `ProcessRegistry` records a `_terminated[execution_id]`
+  marker on a successful SIGINT send; `was_terminated()` (read-only, 300s lazy TTL, cleared on
+  `register()`) lets the chat handler + async result callback relabel a graceful-exit-0 or a
+  SIGKILL→504 turn as `status:"cancelled"`.
+- **Backend**: `ExecutionResultEnvelope.status` 3-way map (`success`→SUCCESS,
+  `cancelled`→CANCELLED, else→FAILED) in the async callback (`routers/agents.py`) and the sync
+  applier (`task_execution_service.py`); consumers (`message_router`, `chat`, `paid`, `public`,
+  `validation_service`) treat `cancelled` as non-delivery.
+
+## Findings
+
+### CRITICAL
+None.
+
+### HIGH
+None.
+
+### MEDIUM
+None.
+
+### LOW / INFORMATIONAL
+
+1. **Paid cancel returns partial response (no charge) — not exploitable by the payer.**
+   `routers/paid.py` now skips settlement for a `cancelled` turn (correctly fixing the
+   charge-on-cancel money bug) and still returns `exec_result.response` in the 200 body. A
+   theoretical free-ride ("let the agent work, then cancel to dodge payment") is **not reachable**:
+   the terminate endpoint is owner-only (`get_authorized_agent` + `get_current_user`), so the x402
+   payer cannot self-trigger a cancel. Only the agent owner can cancel, and returning their own
+   agent's partial output to the payer for free is benign. No action required.
+
+2. **Auth/rate-vs-cancel relabel guard is agent-side only (defense-in-depth gap, within trust model).**
+   `result_callback._is_auth_or_rate()` correctly prevents an honest agent from relabelling a 503
+   auth / 429 rate terminal as `cancelled` (so the AUTH dispatch breaker + SUB-003 still fire). The
+   backend callback (`agent_execution_result`) mapped `status:"cancelled"`→CANCELLED **unconditionally**
+   — it did not re-apply that guard. A compromised agent could therefore POST
+   `status:"cancelled", error_code:"auth"` to dodge the breaker. This grants **no new capability**:
+   the same authenticated endpoint (agent's own MCP key) already lets a compromised agent self-report
+   SUCCESS/FAILED, and cross-tenant isolation is intact (ownership 404 + agent-scoped-key gate). The
+   dispatch breaker's threat model is honest-agent fault containment, not a malicious agent.
+
+   **RESOLVED (2026-06-22, confirmed worth doing by Codex consult):** the backend callback now
+   mirrors `_is_auth_or_rate` — a `cancelled` payload carrying `error_code == "auth"` or a
+   `terminal_reason` of `auth`/`rate_limit` is mapped to FAILED (not CANCELLED), preserving the
+   invariant "an auth/rate terminal is never reclassified as cancellation" regardless of caller
+   image. The win is operational (buggy/mixed-version agents can't silently disarm the breaker /
+   SUB-003), not anti-malicious. Tests added in `tests/unit/test_679_callback_cancel.py`
+   (3 guard cases + a genuine-cancel regression guard); full #679 suite green (50 passed).
+   *Scope note:* the sync path (`task_execution_service.py`) was intentionally left unchanged —
+   the agent-server only emits `cancelled` there on a 504-after-terminate or graceful exit-0
+   (auth/rate re-raises as 503/429 and never reaches a `cancelled` label), so there is no body
+   signal to guard. Finding 1 (paid partial-response) was assessed and **SKIPPED** — not
+   payer-triggerable (terminate is owner-only), so it is an owner choice, not a platform issue.
+
+## Verifications (clean)
+
+- **Secrets**: no hardcoded credentials in added lines or new `test_679_*` files. No `.env`/manifest changes.
+- **Dependencies**: no `requirements*/package*.json` changes — zero new packages.
+- **Auth boundaries**:
+  - `terminate_agent_execution` — owner-only (`get_authorized_agent` + `get_current_user`); unchanged.
+  - `agent_execution_result` (async callback) — agent's own MCP key (`authorize_heartbeat`,
+    `track_usage=False`) + ownership(404) + `dispatched_async` marker(409) + body-size(413) all preserved;
+    the new CANCELLED branch sits **after** every gate.
+  - Replay gate: `_AUTHORITATIVE_TERMINALS` includes `CANCELLED`, so a late callback on a cancelled
+    row short-circuits to `{replayed:true}` with no write.
+- **CAS contract** (`db/schedules.py:update_execution_status`): CANCELLED is a non-success terminal
+  write guarded against overwriting any already-terminal row; SUCCESS is blocked over an existing
+  CANCELLED (#671). Both cancel races (terminate-first / callback-first) resolve to a single
+  authoritative CANCELLED. Verified consistent with the new 3-way maps.
+- **Injection**: `execution_id` is used only as a dict key (process registry) and a parameterized
+  SQLAlchemy lookup (`db.get_execution`); no string-built SQL, no `subprocess`/`os.system`/`eval`
+  with user input, no path traversal. Unknown future `status` values fall through to FAILED (safe default).
+- **DoS / memory**: `_terminated` dict bounded by traffic (300s lazy TTL eviction, cleared on
+  `register`) — mirrors the existing `_recently_completed` map; no unbounded growth.
+- **Exception hygiene**: no new broad `except: pass` / silent swallows; the agent-server handler
+  re-raises 503/429/auth and unterminated 504s unchanged.
+- **Configuration**: no new routes, no CORS wildcard, no debug flags.
+
+## Recommendation
+
+**CLEAR** — no CRITICAL or HIGH findings. The two LOW items were taken to a Codex consult for an
+independent worth-resolving judgment: **Finding 2 RESOLVED** (backend callback now mirrors the
+auth/rate-vs-cancel guard; tests added, suite green), **Finding 1 SKIPPED** (owner-only cancel,
+not payer-triggerable — an owner choice, not a platform issue). The cancel-billing change fixes
+(does not introduce) a money bug. Safe to proceed.

src/backend/adapters/message_router.py

@@ -436,9 +436,11 @@ async def _run_agent_task(
                 images=image_data or None,
             )
 
-            if result.status == "failed":
+            if result.status in ("failed", "cancelled"):
+                # #679: a CANCELLED turn is non-delivery — surface the cancel
+                # notice instead of posting the empty response as if it succeeded.
                 error_msg = result.error or "Unknown error"
-                logger.error(f"[ROUTER:{channel}] Step 9 - task failed: {error_msg}")
+                logger.info(f"[ROUTER:{channel}] Step 9 - task {result.status}: {error_msg}")
                 await adapter.indicate_done(message)
 
                 # Reply with the actual error if available, otherwise generic message