fix(code): interrupt remote runs on chat cancellation (#4234)

Pressing Esc cancelled only the local Textual worker, so the LangGraph
server could keep remote/subgraph runs active until a later state-write
conflict. Interrupt cleanup now explicitly cancels active remote runs
before persisting recovery state, while keeping the cleanup best-effort
for local agents.

Made by [Open
SWE](https://openswe.vercel.app/agents/d68b606b-4b3c-5278-0cab-51b79c3d5af4)

---------

Co-authored-by: Nick Hollon <274035459+nick-hollon-lc@users.noreply.github.com>
Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
Co-authored-by: Mason Daugherty <mason@langchain.dev>
Co-authored-by: Mason Daugherty <github@mdrxy.com>

libs/code/deepagents_code/remote_client.py

@@ -310,6 +310,22 @@ class RemoteAgent:
             )
             raise
 
+    async def acancel_active_runs(self, config: dict[str, Any]) -> None:
+        """Cancel pending/running runs on the configured thread.
+
+        Best-effort: per-run cancellation failures are swallowed by
+        `_cancel_active_runs`. Intended for proactive cancellation on
+        interrupt, before recovery-state writes.
+
+        Args:
+            config: Config with `configurable.thread_id`.
+
+        Raises:
+            ValueError: If `thread_id` is not present in `config`.
+        """  # noqa: DOC502 — raised by _require_thread_id
+        thread_id = _require_thread_id(config)
+        await _cancel_active_runs(self._get_graph(), thread_id)
+
     async def aupdate_state(
         self,
         config: dict[str, Any],

libs/code/deepagents_code/textual_adapter.py

@@ -1540,6 +1540,22 @@ async def _handle_interrupt_cleanup(
 
     await adapter._mount_message(AppMessage("Interrupted by user"))
 
+    # Proactively cancel server-side runs before persisting recovery state, so
+    # the aupdate_state writes below don't 409 against a still-busy thread. This
+    # is defense-in-depth layered on top of aupdate_state's own 409 -> cancel ->
+    # retry path (see RemoteAgent.aupdate_state); a failure here is not fatal.
+    # Absent on local agents, so this is a no-op for them.
+    cancel_active_runs = getattr(agent, "acancel_active_runs", None)
+    if cancel_active_runs is not None:
+        try:
+            await cancel_active_runs(config)
+        except Exception:
+            logger.warning(
+                "Failed to cancel active remote runs for thread %s",
+                config.get("configurable", {}).get("thread_id"),
+                exc_info=True,
+            )
+
     interrupted_msg = _build_interrupted_ai_message(
         pending_text_by_namespace,
         adapter._current_tool_messages,

libs/code/tests/unit_tests/test_remote_client.py

@@ -584,6 +584,42 @@ class TestRemoteAgentUpdateState:
         uuid.UUID(call_config["configurable"]["thread_id"])
 
 
+class TestRemoteAgentCancelActiveRuns:
+    """`acancel_active_runs` exposes best-effort remote run cancellation."""
+
+    async def test_cancels_running_and_pending_runs(self) -> None:
+        agent = RemoteAgent(url="http://localhost:8123", graph_name="agent")
+        runs_list = AsyncMock(
+            side_effect=[
+                [{"run_id": "run-1"}],
+                [{"run_id": "run-2"}],
+            ]
+        )
+        runs_cancel = AsyncMock()
+        mock_runs = MagicMock()
+        mock_runs.list = runs_list
+        mock_runs.cancel = runs_cancel
+        mock_client = MagicMock()
+        mock_client.runs = mock_runs
+        mock_graph = MagicMock()
+        mock_graph._validate_client.return_value = mock_client
+        agent._graph = mock_graph
+
+        await agent.acancel_active_runs(_config())
+
+        assert runs_list.await_count == 2
+        assert runs_cancel.await_count == 2
+        assert {call.args[1] for call in runs_cancel.await_args_list} == {
+            "run-1",
+            "run-2",
+        }
+
+    async def test_raises_when_thread_id_missing(self) -> None:
+        agent = RemoteAgent(url="http://localhost:8123", graph_name="agent")
+        with pytest.raises(ValueError, match="thread_id"):
+            await agent.acancel_active_runs({"configurable": {}})
+
+
 def _conflict_error() -> Exception:
     """Build a `ConflictError` (HTTP 409) for tests."""
     import httpx

libs/code/tests/unit_tests/test_textual_adapter.py

@@ -9,7 +9,7 @@ from importlib.metadata import PackageNotFoundError
 from io import StringIO
 from pathlib import Path
 from types import SimpleNamespace
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -39,6 +39,9 @@ from deepagents_code.widgets.messages import (
     ToolCallMessage,
 )
 
+if TYPE_CHECKING:
+    from langchain_core.runnables import RunnableConfig
+
 
 async def _mock_mount(widget: object) -> None:
     """Mock mount function for tests."""
@@ -264,6 +267,100 @@ class TestInterruptCleanup:
         sync_message_content.assert_called_once_with("asst-1", "partial response")
         assert assistant_messages == {}
 
+    async def test_interrupt_cancels_active_remote_runs_before_state_writes(
+        self,
+    ) -> None:
+        """Remote runs should be interrupted before recovery state is persisted."""
+        calls: list[str] = []
+
+        # Sync side effects are fine: the AsyncMock wrapping them is awaitable,
+        # and recording into `calls` is enough to assert relative ordering.
+        def cancel_runs(_config: object) -> None:
+            calls.append("cancel")
+
+        def update_state(_config: object, _values: dict[str, Any]) -> None:
+            calls.append("update")
+
+        adapter = TextualUIAdapter(
+            mount_message=AsyncMock(),
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+            set_spinner=AsyncMock(),
+            set_active_message=MagicMock(),
+        )
+        agent = SimpleNamespace(
+            acancel_active_runs=AsyncMock(side_effect=cancel_runs),
+            aupdate_state=AsyncMock(side_effect=update_state),
+        )
+        config: RunnableConfig = {"configurable": {"thread_id": "t-1"}}
+
+        await _handle_interrupt_cleanup(
+            adapter=adapter,
+            agent=agent,
+            config=config,
+            pending_text_by_namespace={},
+            captured_input_tokens=0,
+            captured_output_tokens=0,
+            turn_stats=SessionStats(),
+            start_time=0.0,
+        )
+
+        agent.acancel_active_runs.assert_awaited_once_with(config)
+        assert calls == ["cancel", "update"]
+
+    async def test_remote_run_cancel_failure_does_not_skip_state_writes(self) -> None:
+        """Interrupt cleanup remains best-effort when remote cancel fails."""
+        agent = SimpleNamespace(
+            acancel_active_runs=AsyncMock(side_effect=RuntimeError("down")),
+            aupdate_state=AsyncMock(),
+        )
+        adapter = TextualUIAdapter(
+            mount_message=AsyncMock(),
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+            set_spinner=AsyncMock(),
+            set_active_message=MagicMock(),
+        )
+
+        await _handle_interrupt_cleanup(
+            adapter=adapter,
+            agent=agent,
+            config={"configurable": {"thread_id": "t-1"}},
+            pending_text_by_namespace={},
+            captured_input_tokens=0,
+            captured_output_tokens=0,
+            turn_stats=SessionStats(),
+            start_time=0.0,
+        )
+
+        agent.acancel_active_runs.assert_awaited_once()
+        agent.aupdate_state.assert_awaited_once()
+
+    async def test_local_agent_without_cancel_method_still_writes_state(self) -> None:
+        """Local agents lack `acancel_active_runs`; cleanup must skip it cleanly."""
+        agent = SimpleNamespace(aupdate_state=AsyncMock())
+        assert not hasattr(agent, "acancel_active_runs")
+        adapter = TextualUIAdapter(
+            mount_message=AsyncMock(),
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+            set_spinner=AsyncMock(),
+            set_active_message=MagicMock(),
+        )
+
+        await _handle_interrupt_cleanup(
+            adapter=adapter,
+            agent=agent,
+            config={"configurable": {"thread_id": "t-1"}},
+            pending_text_by_namespace={},
+            captured_input_tokens=0,
+            captured_output_tokens=0,
+            turn_stats=SessionStats(),
+            start_time=0.0,
+        )
+
+        agent.aupdate_state.assert_awaited_once()
+
     async def test_disables_tracing_during_state_save(self) -> None:
         """Interrupt-cleanup `aupdate_state` calls must run with tracing disabled.