fix: propagate cancellation across active turns

docs/api.md

@@ -43,6 +43,8 @@ The success body contains `reply`, `session_id`, `profile`, `decision_assist`, `
 
 `POST /api/chat/stream` takes the same body and authorization as `/api/chat` and returns `text/event-stream`. Each event is one `data: <JSON>\n\n` record. Implementations may receive:
 
+An active foreground turn can be cancelled with `POST /api/chat/{session_id}/cancel`. The response reports whether an active turn accepted the cancellation request. Already-dispatched remote or external side effects may still complete; the session event journal records cancellation and that uncertainty.
+
 | `event` | Key data | Meaning |
 |---|---|---|
 | `content` | `chunk` | Plain response text delta; control/tool prefixes are filtered from user-visible chunks. |

docs/tool-inventory.md

@@ -80,6 +80,7 @@ Parameterized names are part of the contract; clients may substitute a value for
 | `POST` | `/api/auth/session` |
 | `POST` | `/api/chat` |
 | `POST` | `/api/chat/stream` |
+| `POST` | `/api/chat/{session_id}/cancel` |
 | `GET` | `/api/cost` |
 | `POST` | `/api/cost/reset` |
 | `GET` | `/api/health` |

openkyrozen/agent/delegation_runtime.py

@@ -316,12 +316,15 @@ def _invoke_delegated(self, run, *, review, feedback, coordinator, report_repair
            "cite actual tool receipt IDs. Do not claim unsupported results.")
     )
     child.interaction.set_mode("ask" if review or report_repair else "agent")
+    from openkyrozen.agent.turn import CancellationEvents
     context = ExecutionContext(model=model, provider=provider, provider_config=config,
         capability_token=issue_capability_token("subagent:" + child_id, capabilities),
         child_run_id=run["run_id"], coordinator=coordinator,
         delegation_budget=coordinator.budgets.get(run["run_id"]),
         delegation_budget_run_id=run.get("budget_id", run["run_id"]),
-        delegation_tools=frozenset(tools), max_steps=profile.max_steps)
+        delegation_tools=frozenset(tools), max_steps=profile.max_steps,
+        cancelled=CancellationEvents(self.execution_context.cancelled,
+                                     coordinator.cancelled.get(run["run_id"])))
     turn_token = self._turn_context.set(context)
     state_token = self._active_context_state.set(ContextState(model, config.context_window_tokens))
     usage_token = self._active_usage_run_id.set(child_id)

openkyrozen/agent/executor.py

@@ -545,6 +545,9 @@ def finish(outcome: ToolResult, authorized: bool = False, raw_result: Any = None
         while True:
             attempt += 1
             with self._delegation_tool_access(action, execution_args):
+                if self.execution_context.cancelled is not None and self.execution_context.cancelled.is_set():
+                    outcome = tool_failure("cancelled", "Tool call was cancelled before execution.")
+                    break
                 if callable(_pre_execution_check) and not _pre_execution_check():
                     outcome = tool_failure("cancelled", "Tool retry stopped because its task is no longer active.")
                     break

openkyrozen/agent/loop.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import asyncio
 from typing import Any
 from openkyrozen.tasks.engine import is_complete, is_terminal
 from openkyrozen.agent.modes import InteractionError, is_plan_acceptance
@@ -131,9 +132,31 @@ def _chat_turn(self, user_input: str, clear_tasks: bool = False, profile: str |
                 )
             if backend := self._fast_used_backend.get():
                 reply = f"Applied routing advice from {'Jev' if backend == 'jev' else 'Kev'}.\n\n{reply}"
-        except Exception as exc:
+        except (Exception, KeyboardInterrupt, asyncio.CancelledError) as exc:
+            from openkyrozen.providers.errors import ProviderError, ProviderErrorKind
+            cancelled = (self.execution_context.cancelled is not None
+                         and self.execution_context.cancelled.is_set()) or isinstance(
+                             exc, (KeyboardInterrupt, asyncio.CancelledError)
+                         ) or (isinstance(exc, ProviderError) and exc.kind == ProviderErrorKind.CANCELLED)
+            if cancelled and active_plan:
+                for task_id in active_plan.get("working_task_ids", []):
+                    index = next((i for i, task in enumerate(self.tasks.tasks)
+                                  if task.get("id") == task_id), None)
+                    if index is not None and self.tasks.tasks[index].get("status") not in {"succeeded", "cancelled"}:
+                        self.tasks.set_status(index, "cancelled")
+                try:
+                    self._interaction_controller.complete_plan(
+                        active_plan, status="cancelled", task_manager=self.tasks,
+                    )
+                except Exception:
+                    pass
+            if cancelled:
+                self._publish_runtime_event("turn.effect_uncertain", {
+                    "reason": "cancelled_during_execution",
+                    "message": "Remote or already-dispatched side effects may still complete.",
+                })
             self._record_conversation_entry("turn.outcome", {
-                "status": "failed", "error_type": type(exc).__name__,
+                "status": "cancelled" if cancelled else "failed", "error_type": type(exc).__name__,
             })
             from openkyrozen.agent.compaction import is_context_digest, retain_context_digests
             display = conversation.display_messages()
@@ -142,9 +165,10 @@ def _chat_turn(self, user_input: str, clear_tasks: bool = False, profile: str |
             projected = ([digest] if digest else []) + display
             self.short_term_memory = retain_context_digests(projected, limit)
             self.current_session.messages_compacted = bool(digest) or len(projected) > limit
-            if active_plan:
+            if active_plan and not cancelled:
                 self._interaction_controller.complete_plan(active_plan, status="failed")
-            self._track_fix_outcome(original_user_input, f"Turn failed: {type(exc).__name__}: {exc}")
+            if not cancelled:
+                self._track_fix_outcome(original_user_input, f"Turn failed: {type(exc).__name__}: {exc}")
             runtime.turn_end(reply="", success=False,
                              error=f"{type(exc).__name__}: {exc}", **context)
             raise

openkyrozen/agent/runtime.py

@@ -5,6 +5,7 @@
 import asyncio
 import json
 import uuid
+import threading
 from contextlib import contextmanager
 from contextvars import ContextVar
 from .models import AgentSession, WorkspaceState
@@ -283,6 +284,8 @@ def __init__(self, memory, tasks, interaction, workspace):
         self._default_session = AgentSession(memory.session_id or "surface:cli", memory, tasks, interaction, workspace)
         self._default_turn = ExecutionContext()
         self._turn_context = ContextVar("agent_turn", default=None)
+        self._active_turns = {}
+        self._active_turns_lock = threading.Lock()
         self._session_context = ContextVar("agent_session", default=None)
         self._surface_capability_override = ContextVar("agent_surface_capabilities", default=None)
         self._conversation_turn_id = ContextVar("conversation_turn_id", default=None)
@@ -405,6 +408,14 @@ def chat(self, session, message, *, clear_tasks=False, profile=None, memory_cont
                     kind, payload if isinstance(payload, dict) else {"data": payload}, task_id=task_id,
                 )
             ) if task_event_context is not None else None
+            cancellation = threading.Event()
+            turn_context = copy.copy(self.execution_context)
+            from openkyrozen.agent.turn import CancellationEvents
+            turn_context.cancelled = CancellationEvents(turn_context.cancelled, cancellation)
+            turn_context_token = self._turn_context.set(turn_context)
+            session_key = (str(getattr(getattr(session, "memory", None), "user_id", "local")), str(session.session_id))
+            with self._active_turns_lock:
+                self._active_turns.setdefault(session_key, set()).add(cancellation)
             try:
                 publisher.publish("turn.started", {})
                 try:
@@ -428,6 +439,13 @@ def chat(self, session, message, *, clear_tasks=False, profile=None, memory_cont
                     publisher.publish("turn.completed", {"status": self._turn_completion_status.get()})
                 return reply
             finally:
+                with self._active_turns_lock:
+                    active = self._active_turns.get(session_key)
+                    if active is not None:
+                        active.discard(cancellation)
+                        if not active:
+                            self._active_turns.pop(session_key, None)
+                self._turn_context.reset(turn_context_token)
                 self._stream_event_callback.reset(events)
                 self._last_response_is_native.reset(native_response)
                 self._turn_failed.reset(turn_failed)
@@ -442,6 +460,16 @@ def chat(self, session, message, *, clear_tasks=False, profile=None, memory_cont
                 self._context_history_seen.reset(history_seen)
                 self._context_history_projection_ids.reset(history_projection)
 
+    def cancel_turn(self, session_id: str, *, user_id: str = "local") -> bool:
+        """Request cancellation of the active foreground turn for one session."""
+        with self._active_turns_lock:
+            active = tuple(self._active_turns.get((str(user_id), str(session_id)), ()))
+        if not active:
+            return False
+        for cancellation in active:
+            cancellation.set()
+        return True
+
     @property
     def conversation_history(self):
         from openkyrozen.agent.conversation import ConversationHistory
@@ -543,7 +571,10 @@ def run_command(self, args):
             except (OSError, RuntimeError, ValueError) as exc:
                 return CommandResult(f"Error starting background process: {exc}", False,
                                      failure="execution_error")
-        return self.current_session.workspace.adapters.run_command(args)
+        if self.execution_context.cancelled is not None and self.execution_context.cancelled.is_set():
+            from openkyrozen.tools.models import CommandResult
+            return CommandResult("Command cancelled before it started.", False, failure="cancelled")
+        return self.current_session.workspace.adapters.run_command(args, cancelled=self.execution_context.cancelled)
 
     @legacy_text_api
     def _read_process(self, args):

openkyrozen/agent/turn.py

@@ -1,5 +1,13 @@
 """State shared by the phases of a single turn."""
 from dataclasses import dataclass
+
+
+class CancellationEvents:
+    def __init__(self, *events):
+        self.events = tuple(event for event in events if event is not None)
+
+    def is_set(self):
+        return any(event.is_set() for event in self.events)
 from typing import Any
 
 @dataclass
@@ -53,6 +61,7 @@ class ExecutionContext:
     delegation_budget_run_id: str | None = None
     delegation_tools: frozenset[str] | None = None
     max_steps: int | None = None
+    cancelled: Any = None
 
 
 def scoped_turn(function):

openkyrozen/interfaces/tui/backend.py

@@ -158,6 +158,8 @@ def stop(self) -> None:
         if self._stopping.is_set():
             return
         self._stopping.set()
+        session = runtime.current_session
+        runtime.cancel_turn(session.session_id, user_id=session.memory.user_id)
         coordinator = getattr(runtime.subagent_manager, "coordinator", None)
         if coordinator:
             coordinator.close()

openkyrozen/interfaces/tui/commands.py

@@ -18,7 +18,11 @@ def _command(self, name: str, args: Any, request_id: str) -> None:
     else:
         command = raw.lower().replace("-", "_")
         arg_text = args if isinstance(args, str) else ""
-    if command in {"/quit", "/exit", "quit", "exit", "shutdown"}:
+    if command in {"/cancel", "cancel"}:
+        session = runtime.current_session
+        cancelled = runtime.cancel_turn(session.session_id, user_id=session.memory.user_id)
+        self.emit("response", request_id, text="Cancellation requested." if cancelled else "No active turn to cancel.")
+    elif command in {"/quit", "/exit", "quit", "exit", "shutdown"}:
         self.stop()
     elif command in {"/provider", "provider"}:
         if isinstance(args, Mapping) and args.get("provider"):

openkyrozen/interfaces/web/chat.py

@@ -434,6 +434,14 @@ async def api_chat(self, request: Request):
             "interaction": session["interaction"], "context": session.get("context")}
 
 
+async def api_cancel_chat(self, session_id: str, request: Request):
+    service = self
+    session_id = service._normalise_session_id(session_id)
+    user_id = service._actor_for_request(request)
+    return {"session_id": session_id,
+            "cancelled": service._agent.cancel_turn(session_id, user_id=user_id)}
+
+
 async def api_chat_stream(self, request: Request):
     service = self
     """SSE streaming chat endpoint."""

openkyrozen/interfaces/web/service.py

@@ -68,6 +68,7 @@ def __init__(self, application=None):
         self.app.get('/api/v2/decision-assist', dependencies=[Depends(self.require_api_access)])(self.decision_assist_status)
         self.app.post('/api/chat', dependencies=[Depends(self.require_api_access)])(self.api_chat)
         self.app.post('/api/chat/stream', dependencies=[Depends(self.require_api_access)])(self.api_chat_stream)
+        self.app.post('/api/chat/{session_id}/cancel', dependencies=[Depends(self.require_api_access)])(self.api_cancel_chat)
         self.app.get('/api/memory', dependencies=[Depends(self.require_api_access)])(self.api_memory)
         self.app.get('/api/v2/memory', dependencies=[Depends(self.require_api_access)])(self.api_v2_memory)
         self.app.get('/api/v2/memory/claims', dependencies=[Depends(self.require_api_access)])(self.api_v2_memory_claims)
@@ -147,7 +148,7 @@ def CHAT_HTML(self):
 
     from openkyrozen.interfaces.web.tasks import (_execute_durable_task, _task_manager, _task_worker, _task_scopes, _recover_task_scopes, _run_task_worker_cycle, _run_scheduled_job, api_v2_tasks, api_v2_create_task, api_v2_resume_task, api_v2_schedules, api_v2_create_schedule, api_v2_disable_schedule)
 
-    from openkyrozen.interfaces.web.chat import (_normalise_session_id, _normalise_profile, _normalise_memory_actor, _normalise_memory_context, _initialise_session_defaults, _set_memory_context, _actor_for_request, _get_or_create_session, _interaction_for_session, _history_for_session, _ensure_history_baseline, _public_history_node, _apply_chat_controls, _run_session_chat, _validate_message, _server_capabilities, _allowed_server_tools, _cost_summary, _cost_report, _json_object, api_chat, api_chat_stream)
+    from openkyrozen.interfaces.web.chat import (_normalise_session_id, _normalise_profile, _normalise_memory_actor, _normalise_memory_context, _initialise_session_defaults, _set_memory_context, _actor_for_request, _get_or_create_session, _interaction_for_session, _history_for_session, _ensure_history_baseline, _public_history_node, _apply_chat_controls, _run_session_chat, _validate_message, _server_capabilities, _allowed_server_tools, _cost_summary, _cost_report, _json_object, api_chat, api_chat_stream, api_cancel_chat)
 
     from openkyrozen.interfaces.web.pages import (chat_page, fast_diagnostics, decision_assist_status, pwa_manifest)
 

openkyrozen/providers/calls.py

@@ -34,9 +34,16 @@ def _provider_timeout_seconds(self) -> float:
 
 def _child_cancellation(self):
     context = self.execution_context
+    cancelled = getattr(context, "cancelled", None)
+    child = None
     if context.child_run_id and context.coordinator:
-        return context.coordinator.cancelled.get(context.child_run_id)
-    return None
+        child = context.coordinator.cancelled.get(context.child_run_id)
+    if cancelled is not None and child is not None:
+        from openkyrozen.agent.turn import CancellationEvents
+        return CancellationEvents(cancelled, child)
+    if child is not None:
+        return child
+    return cancelled
 
 
 def _bounded_provider_call(self, callback: Any) -> Any:

openkyrozen/tools/ports.py

@@ -7,7 +7,7 @@
 class ToolPort(Protocol):
     tool_registry: ToolRegistry
     AVAILABLE_TOOLS: MutableMapping[str, Callable[[str], str]]
-    def run_command(self, args: str | dict) -> CommandResult: ...
+    def run_command(self, args: str | dict, *, cancelled=None) -> CommandResult: ...
     def new_process_manager(self, owner_id: str, workspace_id: str, session_id: str): ...
     def set_workspace_root(self, root: str) -> None: ...
     def set_project_graph(self, graph) -> None: ...

openkyrozen/tools/shell.py

@@ -63,7 +63,7 @@ def _validated_shell_options(self, args: str | dict):
     return cmd, timeout, str(working_directory), env, background
 
 
-def run_command(self, args: str | dict) -> CommandResult:
+def run_command(self, args: str | dict, *, cancelled=None) -> CommandResult:
     """
     Execute a shell command and retain its machine-readable outcome.
     """
@@ -87,7 +87,38 @@ def run_command(self, args: str | dict) -> CommandResult:
             start_new_session=(os.name != "nt"),
         )
         try:
-            out, err = process.communicate(timeout=timeout)
+            deadline = started + timeout
+            while True:
+                if cancelled is not None and cancelled.is_set():
+                    raise InterruptedError
+                try:
+                    out, err = process.communicate(timeout=min(0.1, max(0.01, deadline - time.monotonic())))
+                    break
+                except subprocess.TimeoutExpired:
+                    if time.monotonic() >= deadline:
+                        raise
+        except InterruptedError:
+            if os.name == "nt":
+                subprocess.run(["taskkill", "/PID", str(process.pid), "/T", "/F"],
+                               capture_output=True, timeout=5, check=False)
+            else:
+                try:
+                    os.killpg(process.pid, signal.SIGTERM)
+                except ProcessLookupError:
+                    pass
+            try:
+                out, err = process.communicate(timeout=2)
+            except subprocess.TimeoutExpired:
+                if os.name != "nt":
+                    try:
+                        os.killpg(process.pid, signal.SIGKILL)
+                    except ProcessLookupError:
+                        pass
+                out, err = process.communicate()
+            return CommandResult("Error: command cancelled", False, process.returncode,
+                                 "cancelled", out, err, time.monotonic() - started, True,
+                                 sandbox_backend=invocation.backend,
+                                 sandbox_reason=invocation.unavailable_reason)
         except subprocess.TimeoutExpired:
             if os.name == "nt":
                 subprocess.run(["taskkill", "/PID", str(process.pid), "/T", "/F"],

tests/test_process_lifecycle.py

@@ -3,9 +3,27 @@
 import unittest
 import shlex
 import tempfile
+import threading
 
 
 class ProcessLifecycleTests(unittest.TestCase):
+    def test_foreground_shell_cancellation_terminates_process_group(self):
+        from openkyrozen.tools import ToolAdapters
+        with tempfile.TemporaryDirectory() as root:
+            adapters = ToolAdapters(root)
+            cancelled = threading.Event()
+            result = []
+            worker = threading.Thread(target=lambda: result.append(adapters.run_command({
+                "command": shlex.join([sys.executable, "-c", "import time; time.sleep(30)"]),
+                "timeout_seconds": 30,
+            }, cancelled=cancelled)))
+            worker.start()
+            time.sleep(.2)
+            cancelled.set()
+            worker.join(5)
+            self.assertFalse(worker.is_alive())
+            self.assertEqual(result[0].failure, "cancelled")
+
     def test_runtime_exposes_session_process_tools_and_application_close_stops_process(self):
         from pathlib import Path
         from unittest.mock import patch