diff --git a/docs/api.md b/docs/api.md index 41de429..dc70bfd 100644 --- a/docs/api.md +++ b/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: \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. | diff --git a/docs/tool-inventory.md b/docs/tool-inventory.md index d9e9b9c..0a1cf46 100644 --- a/docs/tool-inventory.md +++ b/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` | diff --git a/openkyrozen/agent/delegation_runtime.py b/openkyrozen/agent/delegation_runtime.py index bb7fdf2..f230740 100644 --- a/openkyrozen/agent/delegation_runtime.py +++ b/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) diff --git a/openkyrozen/agent/executor.py b/openkyrozen/agent/executor.py index f406bc4..8cfd211 100644 --- a/openkyrozen/agent/executor.py +++ b/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 diff --git a/openkyrozen/agent/loop.py b/openkyrozen/agent/loop.py index 87f13a3..ccc9b3c 100644 --- a/openkyrozen/agent/loop.py +++ b/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 diff --git a/openkyrozen/agent/runtime.py b/openkyrozen/agent/runtime.py index 2a5a60b..ea6fc0a 100644 --- a/openkyrozen/agent/runtime.py +++ b/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): diff --git a/openkyrozen/agent/turn.py b/openkyrozen/agent/turn.py index 7d79abd..df53ccf 100644 --- a/openkyrozen/agent/turn.py +++ b/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): diff --git a/openkyrozen/interfaces/tui/backend.py b/openkyrozen/interfaces/tui/backend.py index 20dc021..3c95cbb 100644 --- a/openkyrozen/interfaces/tui/backend.py +++ b/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() diff --git a/openkyrozen/interfaces/tui/commands.py b/openkyrozen/interfaces/tui/commands.py index 4cb856d..f46182c 100644 --- a/openkyrozen/interfaces/tui/commands.py +++ b/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"): diff --git a/openkyrozen/interfaces/web/chat.py b/openkyrozen/interfaces/web/chat.py index 6c8568d..40d492a 100644 --- a/openkyrozen/interfaces/web/chat.py +++ b/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.""" diff --git a/openkyrozen/interfaces/web/service.py b/openkyrozen/interfaces/web/service.py index 043df87..ff34ee6 100644 --- a/openkyrozen/interfaces/web/service.py +++ b/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) diff --git a/openkyrozen/providers/calls.py b/openkyrozen/providers/calls.py index 50f5757..fa120aa 100644 --- a/openkyrozen/providers/calls.py +++ b/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: diff --git a/openkyrozen/tools/ports.py b/openkyrozen/tools/ports.py index 51126a3..ddcfe00 100644 --- a/openkyrozen/tools/ports.py +++ b/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: ... diff --git a/openkyrozen/tools/shell.py b/openkyrozen/tools/shell.py index 1147f03..5f186c2 100644 --- a/openkyrozen/tools/shell.py +++ b/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"], diff --git a/tests/test_process_lifecycle.py b/tests/test_process_lifecycle.py index 94d159a..a00d674 100644 --- a/tests/test_process_lifecycle.py +++ b/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 diff --git a/tests/test_runtime_events.py b/tests/test_runtime_events.py index 447d1f1..b2467a7 100644 --- a/tests/test_runtime_events.py +++ b/tests/test_runtime_events.py @@ -229,6 +229,39 @@ def test_chat_emits_cancelled_terminal_event_for_async_cancellation(self): runtime.chat(runtime.current_session, "hi", on_runtime_event=events.append) self.assertEqual([event.kind for event in events], ["turn.started", "turn.cancelled"]) + def test_cancel_turn_sets_only_the_active_session_signal(self): + import threading + runtime = self.app.runtime + session = runtime.current_session + entered = threading.Event() + observed = [] + + def blocked_turn(*_args, **_kwargs): + entered.set() + cancellation = runtime.execution_context.cancelled + while not cancellation.is_set(): + time.sleep(.01) + observed.append(cancellation.is_set()) + from openkyrozen.providers.errors import ProviderError, ProviderErrorKind + raise ProviderError(ProviderErrorKind.CANCELLED) + + def run_chat(): + try: + runtime.chat(session, "hi") + except Exception: + pass + + with patch.object(runtime, "_chat_turn", side_effect=blocked_turn): + worker = threading.Thread(target=run_chat) + worker.start() + self.assertTrue(entered.wait(2)) + self.assertFalse(runtime.cancel_turn("other-session", user_id=session.memory.user_id)) + self.assertTrue(runtime.cancel_turn(session.session_id, user_id=session.memory.user_id)) + worker.join(2) + self.assertFalse(worker.is_alive()) + self.assertEqual(observed, [True]) + + def test_native_response_text_does_not_enter_legacy_control_parser(self): runtime = self.app.runtime text = "Action: ordinary prose\nTaskList: ordinary prose" diff --git a/tui/slash.go b/tui/slash.go index eb65a09..8971806 100644 --- a/tui/slash.go +++ b/tui/slash.go @@ -19,6 +19,7 @@ var commands = []command{ {name: "agent", aliases: nil, description: "Switch auto, coder, or researcher mode"}, {name: "permissions", aliases: []string{"permission"}, description: "Choose Ask, protected full access, or full access"}, {name: "agents", description: "Inspect parallel agents, results, and independent reviews"}, + {name: "cancel", aliases: nil, description: "Cancel the active foreground turn"}, {name: "new", aliases: nil, description: "Start a new chat in the current project"}, {name: "project", aliases: nil, description: "Create or open a directory as a new project chat"}, {name: "sessions", aliases: nil, description: "Show project and chat navigation"},