完善 Core Runner 取消与群聊控制命令 (#2821)

* feat(core): add generic runner cancellation control * fix(core): 完善 Runner 取消与群聊控制命令 * fix(core): 支持多并发会话及时停止任务
2026-11-01 21:35:25 +08:00 · 2026-10-31 21:52:25 +08:00
parent f750bb96ac
commit 5be3b03e77
13 changed files with 967 additions and 13 deletions
@@ -286,6 +286,7 @@ class AgentRunOrchestrator:
                bot_id=event.bot_id,
                workspace_id=event.workspace_id,
                thread_id=event.thread_id,
+                actor_id=getattr(getattr(event, 'actor', None), 'actor_id', None),
                state_policy={
                    'enable_state': binding.state_policy.enable_state,
                    'state_scopes': list(binding.state_policy.state_scopes),
@@ -580,6 +581,59 @@ class AgentRunOrchestrator:
        """Resolve runner ID for telemetry/logging without full execution."""
        return self.query_bridge.resolve_runner_id_for_telemetry(query)

+    async def cancel_active_from_query(
+        self,
+        query: pipeline_query.Query,
+        *,
+        allow_other_actors: bool = False,
+    ) -> str | None:
+        """Request cancellation and stop the active Runner task for this Query conversation."""
+
+        plan = self.query_bridge.build_plan(query)
+        event = plan.event
+        binding = plan.binding
+        if event.event_type != 'message.received' or not event.conversation_id:
+            return None
+        descriptor = await self.registry.get(
+            get_query_execution_context(query),
+            binding.runner_id,
+            plan.bound_plugins,
+        )
+        actor = getattr(event, 'actor', None)
+        actor_id = getattr(actor, 'actor_id', None)
+        if not allow_other_actors and not actor_id:
+            return None
+        target_run_id = await self._session_registry.find_active_target(
+            conversation_id=event.conversation_id,
+            runner_id=descriptor.id,
+            bot_id=event.bot_id,
+            workspace_id=event.workspace_id,
+            thread_id=event.thread_id,
+            actor_id=None if allow_other_actors else actor_id,
+        )
+        if target_run_id is None and not allow_other_actors:
+            other_actor_run_id = await self._session_registry.find_active_target(
+                conversation_id=event.conversation_id,
+                runner_id=descriptor.id,
+                bot_id=event.bot_id,
+                workspace_id=event.workspace_id,
+                thread_id=event.thread_id,
+            )
+            if other_actor_run_id is not None:
+                query.variables['_stop_command_denied'] = True
+        if target_run_id is None:
+            return None
+        if not await self._session_registry.cancel_task(target_run_id, reason='stop_command'):
+            return None
+        from .run_ledger_store import RunLedgerStore
+
+        store = RunLedgerStore(self.ap.persistence_mgr.get_db_engine())
+        await store.request_cancel(
+            run_id=target_run_id,
+            status_reason='cancelled by stop command',
+        )
+        return target_run_id
+
    async def try_claim_steering_from_query(
        self,
        query: pipeline_query.Query,
@@ -475,7 +475,7 @@ class RunLedgerStore:
        now = _utc_now()
        async with self._session_factory() as session:
            run = await self._get_run_row(session, run_id)
-            if run is None:
+            if run is None or run.status in TERMINAL_STATUSES:
                return None
            run.cancel_requested_at = as_naive_utc(now)
            run.updated_at = as_naive_utc(now)
@@ -45,6 +45,7 @@ class RunAuthorizationSnapshot(typing.TypedDict):
    bot_id: str | None
    workspace_id: str | None
    thread_id: str | None
+    actor_id: str | None
    state_policy: dict[str, typing.Any]
    state_context: dict[str, typing.Any]
    platform_context: dict[str, typing.Any]
@@ -79,6 +80,7 @@ class AgentRunSession(typing.TypedDict):
    status: AgentRunSessionStatus
    reply_streams: typing.Any
    steering_queue: list[SteeringQueueItem]
+    task: asyncio.Task[typing.Any] | None


 class AgentRunSessionRegistry:
@@ -111,6 +113,7 @@ class AgentRunSessionRegistry:
        bot_id: str | None = None,
        workspace_id: str | None = None,
        thread_id: str | None = None,
+        actor_id: str | None = None,
        available_apis: dict[str, bool] | None = None,
        state_policy: dict[str, typing.Any] | None = None,
        state_context: dict[str, typing.Any] | None = None,
@@ -130,6 +133,7 @@ class AgentRunSessionRegistry:
            bot_id: Bot UUID for history/event access
            workspace_id: Workspace ID for history/event access
            thread_id: Thread ID for history/event access
+            actor_id: Actor ID that started the run
            available_apis: Run-scoped pull APIs exposed in RunnerContext
            state_policy: State policy from binding (enable_state, state_scopes)
            state_context: Context for state API (scope_keys, binding_identity, etc.)
@@ -157,6 +161,7 @@ class AgentRunSessionRegistry:
            'bot_id': bot_id,
            'workspace_id': workspace_id,
            'thread_id': thread_id,
+            'actor_id': actor_id,
            'state_policy': copy.deepcopy(state_policy),
            'state_context': copy.deepcopy(state_context),
            'platform_context': copy.deepcopy(platform_context or {}),
@@ -177,6 +182,7 @@ class AgentRunSessionRegistry:
                'last_activity_at': now,
            },
            'steering_queue': [],
+            'task': asyncio.current_task(),
        }

        async with self._lock:
@@ -296,6 +302,56 @@ class AgentRunSessionRegistry:
            candidates.sort(key=lambda item: item[0])
            return candidates[0][1]

+    async def find_active_target(
+        self,
+        *,
+        conversation_id: str,
+        runner_id: str,
+        bot_id: str | None = None,
+        workspace_id: str | None = None,
+        thread_id: str | None = None,
+        actor_id: str | None = None,
+    ) -> str | None:
+        """Find the oldest active run for one conversation without requiring steering."""
+
+        async with self._lock:
+            candidates: list[tuple[int, str]] = []
+            for run_id, session in self._sessions.items():
+                authorization = session['authorization']
+                if session.get('runner_id') != runner_id:
+                    continue
+                if authorization.get('conversation_id') != conversation_id:
+                    continue
+                if authorization.get('bot_id') != bot_id:
+                    continue
+                if authorization.get('workspace_id') != workspace_id:
+                    continue
+                if authorization.get('thread_id') != thread_id:
+                    continue
+                if actor_id is not None and authorization.get('actor_id') != actor_id:
+                    continue
+                candidates.append((session['status'].get('started_at', 0), run_id))
+            if not candidates:
+                return None
+            candidates.sort(key=lambda item: item[0])
+            return candidates[0][1]
+
+    async def cancel_task(self, run_id: str, *, reason: str | None = None) -> bool:
+        """Cancel the Host task that owns one active run."""
+
+        async with self._lock:
+            session = self._sessions.get(run_id)
+            task = session.get('task') if session else None
+            if task is None or task.done():
+                return False
+            execution_query = session.get('execution_query') if session else None
+            if reason and execution_query is not None:
+                variables = getattr(execution_query, 'variables', None)
+                if isinstance(variables, dict):
+                    variables['_runner_cancel_reason'] = reason
+            task.cancel()
+            return True
+
    async def enqueue_steering(
        self,
        run_id: str,
@@ -88,15 +88,8 @@ class CommandManager:
        else:
            yield command_context.CommandReturn(error=command_errors.CommandNotFoundError(context.command))

-    async def execute(
-        self,
-        command_text: str,
-        full_command_text: str,
-        query: pipeline_query.Query,
-        session: provider_session.Session,
-    ) -> typing.AsyncGenerator[command_context.CommandReturn, None]:
-        """执行命令"""
-
+    async def resolve_privilege(self, query: pipeline_query.Query) -> int:
+        """Resolve the command privilege for one platform query."""
        privilege = 1

        import sqlalchemy as _sa
@@ -113,6 +106,19 @@ class CommandManager:
        if _admins.first() is not None:
            privilege = 2

+        return privilege
+
+    async def execute(
+        self,
+        command_text: str,
+        full_command_text: str,
+        query: pipeline_query.Query,
+        session: provider_session.Session,
+    ) -> typing.AsyncGenerator[command_context.CommandReturn, None]:
+        """执行命令"""
+
+        privilege = await self.resolve_privilege(query)
+
        ctx = command_context.ExecuteContext(
            instance_uuid=query.instance_uuid,
            workspace_uuid=query.workspace_uuid,
@@ -24,6 +24,49 @@ class Controller:
        self.ap = ap
        self.semaphore = asyncio.Semaphore(self.ap.instance_config.data['concurrency']['pipeline'])

+    def _command_name(self, query: pipeline_query.Query) -> str | None:
+        """Return a command name while ignoring non-text platform components."""
+
+        command_config = self.ap.instance_config.data.get('command', {})
+        if not command_config.get('enable', True):
+            return None
+        message_chain = getattr(query, 'message_chain', None)
+        try:
+            components = list(message_chain or [])
+        except TypeError:
+            components = []
+        text_parts = [str(getattr(component, 'text', '')) for component in components]
+        message_text = ''.join(text_parts).strip() or str(message_chain or '').strip()
+        for prefix in command_config.get('prefix', []):
+            if message_text.startswith(prefix):
+                command_text = message_text[len(prefix) :].strip()
+                return command_text.split(maxsplit=1)[0].lower() if command_text else None
+        return None
+
+    async def _handle_stop_command(
+        self,
+        query: pipeline_query.Query,
+        original_query: pipeline_query.Query,
+    ) -> None:
+        """Cancel an authorized active run and retain the result for CommandHandler."""
+
+        if original_query.variables.get('_stop_command_attempted'):
+            return
+        query.variables['_stop_command_attempted'] = True
+        original_query.variables['_stop_command_attempted'] = True
+        allow_other_actors = False
+        if getattr(getattr(query, 'launcher_type', None), 'value', None) == 'group':
+            allow_other_actors = await self.ap.cmd_mgr.resolve_privilege(original_query) >= 2
+        run_id = await self.ap.agent_run_orchestrator.cancel_active_from_query(
+            query,
+            allow_other_actors=allow_other_actors,
+        )
+        if query.variables.get('_stop_command_denied'):
+            original_query.variables['_stop_command_denied'] = True
+        if run_id:
+            query.variables['_stop_command_run_id'] = run_id
+            original_query.variables['_stop_command_run_id'] = run_id
+
    async def _assert_query_execution_active(
        self,
        query: pipeline_query.Query,
@@ -46,6 +89,7 @@ class Controller:
        """Offer follow-up input to an active Runner before it queues."""

        try:
+            original_query = query
            pipeline_uuid = query.pipeline_uuid
            if not pipeline_uuid:
                return False
@@ -60,6 +104,12 @@ class Controller:
            query.pipeline_config = pipeline.pipeline_entity.config
            query.variables['_pipeline_bound_plugins'] = pipeline.bound_plugins
            query.variables['_pipeline_bound_mcp_servers'] = pipeline.bound_mcp_servers
+            command = self._command_name(query)
+            if command == 'stop':
+                await self._handle_stop_command(query, original_query)
+                # Control commands bypass ordinary group response rules while a
+                # run owns the session slot, then reach CommandHandler normally.
+                return False
            if getattr(getattr(query, 'launcher_type', None), 'value', None) == 'group':
                from .resprule.resprule import GroupRespondRuleCheckStage
                from .entities import ResultType
@@ -76,6 +126,14 @@ class Controller:
                if result.result_type == ResultType.INTERRUPT:
                    return False
                query = candidate
+                command = self._command_name(query)
+                if command == 'stop':
+                    await self._handle_stop_command(query, original_query)
+                    return False
+            if command:
+                # Commands must reach CommandHandler after the active run releases
+                # the session slot. They must never become Runner steering input.
+                return False
            return await self.ap.agent_run_orchestrator.try_claim_steering_from_query(query)
        except Exception as exc:
            self.ap.logger.warning(
@@ -175,6 +233,13 @@ class Controller:
                                break
                            continue

+                        # A stop command must cancel an active run even when the
+                        # session has spare capacity (for example, concurrency=2).
+                        # Otherwise it would occupy the second slot as a normal
+                        # Pipeline query and never reach the cancellation branch.
+                        if self._command_name(query) == 'stop' and not query.variables.get('_stop_command_attempted'):
+                            await self._try_claim_steering_before_session_slot(query)
+
                        if not session._semaphore.locked():
                            selected_query = query
                            selected_session = session
@@ -13,6 +13,7 @@ from ..entity.persistence import pipeline as persistence_pipeline
 from . import stage
 import langbot_plugin.api.entities.builtin.platform.message as platform_message
 import langbot_plugin.api.entities.builtin.platform.events as platform_events
+import langbot_plugin.api.entities.builtin.provider.message as provider_message
 import langbot_plugin.api.entities.events as events
 from ..utils import importutil
 from ..api.http.authz import WorkspaceRequiredError
@@ -20,7 +21,7 @@ from ..api.http.context import ExecutionContext, PrincipalContext, PrincipalType
 from ..workspace.errors import WorkspaceError, WorkspaceInvariantError
 from .config_coercion import coerce_pipeline_config
 from .pool import get_query_execution_context
-from ..i18n import language_scope
+from ..i18n import language_scope, t
 from ..i18n.workspace import get_workspace_language
 from ..agent.runner.config_resolver import RunnerConfigResolver
 from .extension_preferences import normalize_extension_preferences
@@ -464,6 +465,10 @@ class RuntimePipeline:
                    except asyncio.CancelledError:
                        lane_outcome = 'cancelled'
                        lane_error = 'cancelled'
+                        user_requested = (query.variables or {}).get('_runner_cancel_reason') == 'stop_command'
+                        await self._finalize_cancelled_response(query, user_visible=user_requested)
+                        if user_requested:
+                            return
                        raise
                    except BaseException as exc:
                        lane_outcome = 'failed'
@@ -497,6 +502,57 @@ class RuntimePipeline:
        finally:
            processing_mode.reset(token)

+    async def _finalize_cancelled_response(
+        self,
+        query: pipeline_query.Query,
+        *,
+        user_visible: bool,
+    ) -> None:
+        """Close the originating platform response after an active run is cancelled."""
+
+        last_chunk = next(
+            (
+                item
+                for item in reversed(getattr(query, 'resp_messages', []) or [])
+                if isinstance(item, provider_message.MessageChunk)
+            ),
+            None,
+        )
+        text = t('Task was cancelled') if user_visible else ''
+        if not user_visible and last_chunk is not None:
+            text = str(getattr(last_chunk, 'all_content', None) or getattr(last_chunk, 'content', None) or '')
+        message = platform_message.MessageChain([platform_message.Plain(text=text)])
+        quote_origin = query.pipeline_config.get('output', {}).get('misc', {}).get('quote-origin', False)
+        try:
+            if (
+                await query.adapter.is_stream_output_supported()
+                and last_chunk is not None
+                and last_chunk.resp_message_id
+            ):
+                chunk = provider_message.MessageChunk(
+                    role='assistant',
+                    content=text,
+                    all_content=text,
+                    resp_message_id=last_chunk.resp_message_id,
+                    msg_sequence=int(getattr(last_chunk, 'msg_sequence', 0) or 0) + 1,
+                    is_final=True,
+                )
+                await query.adapter.reply_message_chunk(
+                    message_source=query.message_event,
+                    bot_message=chunk,
+                    message=message,
+                    quote_origin=quote_origin,
+                    is_final=True,
+                )
+            elif user_visible:
+                await query.adapter.reply_message(
+                    message_source=query.message_event,
+                    message=message,
+                    quote_origin=quote_origin,
+                )
+        except Exception as exc:
+            self.ap.logger.warning('Failed to finalize cancelled platform response: %s', exc)
+
    async def _process_query(self, query: pipeline_query.Query):
        """Close monitoring on every exit, including cancellation and plugin veto."""
        from .monitoring_helper import MonitoringHelper
@@ -37,6 +37,21 @@ class GroupRespondRuleCheckStage(stage.PipelineStage):
        if query.launcher_type.value != 'group':  # 只处理群消息
            return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)

+        text_parts = [str(getattr(component, 'text', '')) for component in query.message_chain or []]
+        message_text = ''.join(text_parts).strip() or str(query.message_chain or '').strip()
+        command_config = self.ap.instance_config.data.get('command', {})
+        is_stop_command = False
+        if command_config.get('enable', True):
+            for prefix in command_config.get('prefix', []):
+                if not message_text.startswith(prefix):
+                    continue
+                command_text = message_text[len(prefix) :].strip()
+                command_name = command_text.split(maxsplit=1)[0].lower() if command_text else None
+                is_stop_command = command_name == 'stop'
+                break
+        if is_stop_command:
+            return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
+
        # Interaction submissions and routed platform form callbacks resume the
        # original run without another mention or prefix. Ordinary routing only
        # selects a pipeline and must still obey its group response rules.
@@ -5,7 +5,7 @@ from __future__ import annotations
 import asyncio
 import datetime
 import types
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, Mock, patch

 import pytest
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
@@ -407,6 +407,7 @@ async def test_orchestrator_runs_fake_plugin_with_authorized_context(clean_agent
    assert context['context']['available_apis']['run_claim'] is False
    assert context['context']['available_apis']['run_renew_claim'] is False
    assert context['context']['available_apis']['run_release_claim'] is False
+
    assert context['context']['available_apis']['runtime_register'] is False
    assert context['context']['available_apis']['runtime_heartbeat'] is False
    assert context['context']['available_apis']['runtime_list'] is False
@@ -432,6 +433,165 @@ async def test_orchestrator_runs_fake_plugin_with_authorized_context(clean_agent
    assert await get_session_registry().get(context['run_id']) is None


+@pytest.mark.asyncio
+async def test_cancel_active_from_query_rejects_non_message_event(clean_agent_state):
+    orchestrator = AgentRunOrchestrator(
+        FakeApplication(FakePluginConnector(), clean_agent_state),
+        FakeRegistry(make_descriptor()),
+    )
+    orchestrator.query_bridge = types.SimpleNamespace(
+        build_plan=Mock(
+            return_value=types.SimpleNamespace(
+                event=types.SimpleNamespace(event_type='timer', conversation_id='conv'),
+                binding=types.SimpleNamespace(runner_id=RUNNER_ID),
+                bound_plugins=[],
+            )
+        )
+    )
+
+    assert await orchestrator.cancel_active_from_query(make_query()) is None
+
+
+@pytest.mark.asyncio
+async def test_cancel_active_from_query_returns_none_without_active_run(clean_agent_state):
+    orchestrator = AgentRunOrchestrator(
+        FakeApplication(FakePluginConnector(), clean_agent_state),
+        FakeRegistry(make_descriptor()),
+    )
+    orchestrator.query_bridge = types.SimpleNamespace(
+        build_plan=Mock(
+            return_value=types.SimpleNamespace(
+                event=types.SimpleNamespace(
+                    event_type='message.received',
+                    conversation_id='conv',
+                    bot_id='bot',
+                    workspace_id='workspace',
+                    thread_id=None,
+                    actor=types.SimpleNamespace(actor_id='actor'),
+                ),
+                binding=types.SimpleNamespace(runner_id=RUNNER_ID),
+                bound_plugins=[],
+            )
+        )
+    )
+    orchestrator._session_registry = types.SimpleNamespace(
+        find_active_target=AsyncMock(return_value=None),
+        cancel_task=AsyncMock(),
+    )
+
+    assert await orchestrator.cancel_active_from_query(make_query()) is None
+    orchestrator._session_registry.cancel_task.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_cancel_active_from_query_records_and_cancels_run(clean_agent_state):
+    orchestrator = AgentRunOrchestrator(
+        FakeApplication(FakePluginConnector(), clean_agent_state),
+        FakeRegistry(make_descriptor()),
+    )
+    orchestrator.query_bridge = types.SimpleNamespace(
+        build_plan=Mock(
+            return_value=types.SimpleNamespace(
+                event=types.SimpleNamespace(
+                    event_type='message.received',
+                    conversation_id='conv',
+                    bot_id='bot',
+                    workspace_id='workspace',
+                    thread_id='thread',
+                    actor=types.SimpleNamespace(actor_id='actor'),
+                ),
+                binding=types.SimpleNamespace(runner_id=RUNNER_ID),
+                bound_plugins=['langbot-team/LocalAgent'],
+            )
+        )
+    )
+    orchestrator._session_registry = types.SimpleNamespace(
+        find_active_target=AsyncMock(return_value='run-active'),
+        cancel_task=AsyncMock(return_value=True),
+    )
+    store = types.SimpleNamespace(request_cancel=AsyncMock())
+
+    with patch('langbot.pkg.agent.runner.run_ledger_store.RunLedgerStore', return_value=store):
+        result = await orchestrator.cancel_active_from_query(make_query())
+
+    assert result == 'run-active'
+    store.request_cancel.assert_awaited_once_with(
+        run_id='run-active',
+        status_reason='cancelled by stop command',
+    )
+    orchestrator._session_registry.cancel_task.assert_awaited_once_with('run-active', reason='stop_command')
+
+
+@pytest.mark.asyncio
+async def test_cancel_active_from_query_returns_none_when_run_finishes_before_cancel(clean_agent_state):
+    orchestrator = AgentRunOrchestrator(
+        FakeApplication(FakePluginConnector(), clean_agent_state),
+        FakeRegistry(make_descriptor()),
+    )
+    orchestrator.query_bridge = types.SimpleNamespace(
+        build_plan=Mock(
+            return_value=types.SimpleNamespace(
+                event=types.SimpleNamespace(
+                    event_type='message.received',
+                    conversation_id='conv',
+                    bot_id='bot',
+                    workspace_id='workspace',
+                    thread_id=None,
+                    actor=types.SimpleNamespace(actor_id='actor'),
+                ),
+                binding=types.SimpleNamespace(runner_id=RUNNER_ID),
+                bound_plugins=[],
+            )
+        )
+    )
+    orchestrator._session_registry = types.SimpleNamespace(
+        find_active_target=AsyncMock(return_value='run-finished'),
+        cancel_task=AsyncMock(return_value=False),
+    )
+    store = types.SimpleNamespace(request_cancel=AsyncMock())
+
+    with patch('langbot.pkg.agent.runner.run_ledger_store.RunLedgerStore', return_value=store):
+        result = await orchestrator.cancel_active_from_query(make_query())
+
+    assert result is None
+    store.request_cancel.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_cancel_active_from_query_marks_other_actor_run_as_denied(clean_agent_state):
+    orchestrator = AgentRunOrchestrator(
+        FakeApplication(FakePluginConnector(), clean_agent_state),
+        FakeRegistry(make_descriptor()),
+    )
+    orchestrator.query_bridge = types.SimpleNamespace(
+        build_plan=Mock(
+            return_value=types.SimpleNamespace(
+                event=types.SimpleNamespace(
+                    event_type='message.received',
+                    conversation_id='group',
+                    bot_id='bot',
+                    workspace_id='workspace',
+                    thread_id=None,
+                    actor=types.SimpleNamespace(actor_id='member-2'),
+                ),
+                binding=types.SimpleNamespace(runner_id=RUNNER_ID),
+                bound_plugins=[],
+            )
+        )
+    )
+    orchestrator._session_registry = types.SimpleNamespace(
+        find_active_target=AsyncMock(side_effect=[None, 'run-member-1']),
+        cancel_task=AsyncMock(),
+    )
+    query = make_query()
+
+    result = await orchestrator.cancel_active_from_query(query)
+
+    assert result is None
+    assert query.variables['_stop_command_denied'] is True
+    orchestrator._session_registry.cancel_task.assert_not_awaited()
+
+
@pytest.mark.asyncio
 async def test_orchestrator_consumes_interaction_request_before_message_output(clean_agent_state):
    """Whitelisted interaction actions are delivered and never emitted as messages."""
@@ -127,6 +127,23 @@ async def test_finalize_without_metadata_preserves_legacy_snapshot(store, legacy
    assert run['metadata'] == legacy_run


+@pytest.mark.asyncio
+async def test_request_cancel_does_not_mutate_terminal_run(store):
+    await store.create_run(
+        run_id='already-completed',
+        event_id='event',
+        binding_id='binding',
+        runner_id='runner',
+    )
+    completed = await store.finalize_run(run_id='already-completed', status='completed')
+
+    assert await store.request_cancel(run_id='already-completed', status_reason='late stop') is None
+    current = await store.get_run('already-completed')
+    assert current['status'] == 'completed'
+    assert current['cancel_requested_at'] is None
+    assert current['status_reason'] == completed['status_reason']
+
+
@pytest.mark.asyncio
 async def test_metadata_update_still_rejects_new_inline_images(store, legacy_run):
    with pytest.raises(sqlalchemy.exc.StatementError, match='Inline images cannot be persisted'):
@@ -50,6 +50,107 @@ class TestSessionRegistryBasic:
        assert 'permissions' not in result
        assert '_authorized_ids' not in result

+    @pytest.mark.asyncio
+    async def test_cancel_task_stops_registered_run_owner(self):
+        """Cancel the task captured when the run session was registered."""
+
+        registry = AgentRunSessionRegistry()
+        registered = asyncio.Event()
+        blocker = asyncio.Event()
+
+        async def run_owner():
+            await registry.register(
+                run_id='run_cancel',
+                runner_id='plugin:test/my-runner/default',
+                query_id=1,
+                plugin_identity='test/my-runner',
+                resources=make_resources(),
+                conversation_id='person_1',
+                bot_id='bot_1',
+                workspace_id='workspace_1',
+            )
+            registered.set()
+            await blocker.wait()
+
+        task = asyncio.create_task(run_owner())
+        await registered.wait()
+
+        execution_query = type('Query', (), {'variables': {}})()
+        (await registry.get('run_cancel'))['execution_query'] = execution_query
+
+        assert await registry.cancel_task('run_cancel', reason='stop_command') is True
+        assert execution_query.variables['_runner_cancel_reason'] == 'stop_command'
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert await registry.cancel_task('run_cancel') is False
+        await registry.unregister('run_cancel')
+
+    @pytest.mark.asyncio
+    async def test_find_active_target_selects_oldest_matching_scope(self):
+        registry = AgentRunSessionRegistry()
+        cases = [
+            ('wrong_runner', 'plugin:test/other/default', 'conv', 'bot', 'workspace', 'thread'),
+            ('wrong_conversation', 'plugin:test/runner/default', 'other', 'bot', 'workspace', 'thread'),
+            ('wrong_bot', 'plugin:test/runner/default', 'conv', 'other', 'workspace', 'thread'),
+            ('wrong_workspace', 'plugin:test/runner/default', 'conv', 'bot', 'other', 'thread'),
+            ('wrong_thread', 'plugin:test/runner/default', 'conv', 'bot', 'workspace', 'other'),
+            ('newer', 'plugin:test/runner/default', 'conv', 'bot', 'workspace', 'thread'),
+            ('oldest', 'plugin:test/runner/default', 'conv', 'bot', 'workspace', 'thread'),
+        ]
+        for run_id, runner_id, conversation_id, bot_id, workspace_id, thread_id in cases:
+            await registry.register(
+                run_id=run_id,
+                runner_id=runner_id,
+                query_id=1,
+                plugin_identity='test/runner',
+                resources=make_resources(),
+                conversation_id=conversation_id,
+                bot_id=bot_id,
+                workspace_id=workspace_id,
+                thread_id=thread_id,
+                actor_id='actor' if run_id != 'oldest' else 'owner',
+            )
+        (await registry.get('newer'))['status']['started_at'] = 20
+        (await registry.get('oldest'))['status']['started_at'] = 10
+
+        assert (
+            await registry.find_active_target(
+                conversation_id='conv',
+                runner_id='plugin:test/runner/default',
+                bot_id='bot',
+                workspace_id='workspace',
+                thread_id='thread',
+            )
+            == 'oldest'
+        )
+        assert (
+            await registry.find_active_target(
+                conversation_id='conv',
+                runner_id='plugin:test/runner/default',
+                bot_id='bot',
+                workspace_id='workspace',
+                thread_id='thread',
+                actor_id='actor',
+            )
+            == 'newer'
+        )
+        assert (
+            await registry.find_active_target(
+                conversation_id='missing',
+                runner_id='plugin:test/runner/default',
+                bot_id='bot',
+                workspace_id='workspace',
+                thread_id='thread',
+            )
+            is None
+        )
+
+    @pytest.mark.asyncio
+    async def test_cancel_task_rejects_unknown_run(self):
+        registry = AgentRunSessionRegistry()
+
+        assert await registry.cancel_task('missing') is False
+
    @pytest.mark.asyncio
    async def test_register_keeps_host_execution_query_by_identity(self):
        """The runtime registry keeps the Host Query view in memory only."""
@@ -14,7 +14,9 @@ from langbot.pkg.pipeline.pool import QueryPool

 def make_app():
    app = SimpleNamespace()
-    app.instance_config = SimpleNamespace(data={'concurrency': {'pipeline': 10}})
+    app.instance_config = SimpleNamespace(
+        data={'concurrency': {'pipeline': 10}, 'command': {'enable': True, 'prefix': ['!', '！']}}
+    )
    app.logger = MagicMock()
    app.pipeline_mgr = SimpleNamespace()
    app.pipeline_mgr.get_pipeline_by_uuid = AsyncMock()
@@ -22,6 +24,8 @@ def make_app():
    app.sess_mgr.get_session = AsyncMock(return_value=SimpleNamespace())
    app.agent_run_orchestrator = SimpleNamespace()
    app.agent_run_orchestrator.try_claim_steering_from_query = AsyncMock()
+    app.agent_run_orchestrator.cancel_active_from_query = AsyncMock()
+    app.cmd_mgr = SimpleNamespace(resolve_privilege=AsyncMock(return_value=1))
    return app


@@ -81,6 +85,42 @@ async def test_try_claim_steering_sets_pipeline_context_before_claiming():
    app.agent_run_orchestrator.try_claim_steering_from_query.assert_awaited_once_with(query)


+@pytest.mark.asyncio
+async def test_busy_command_is_not_claimed_as_steering():
+    from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain
+
+    app = make_app()
+    app.pipeline_mgr.get_pipeline_by_uuid.return_value = make_pipeline()
+    query = make_query(3, 'pipeline-003')
+    query.message_chain = MessageChain([Plain(text='!list')])
+
+    claimed = await Controller(app)._try_claim_steering_before_session_slot(query)
+
+    assert claimed is False
+    app.agent_run_orchestrator.try_claim_steering_from_query.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_busy_stop_requests_active_run_cancellation():
+    from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain
+
+    app = make_app()
+    app.pipeline_mgr.get_pipeline_by_uuid.return_value = make_pipeline()
+    app.agent_run_orchestrator.cancel_active_from_query.return_value = 'run-stop'
+    query = make_query(4, 'pipeline-004')
+    query.message_chain = MessageChain([Plain(text='!stop')])
+
+    claimed = await Controller(app)._try_claim_steering_before_session_slot(query)
+
+    assert claimed is False
+    app.agent_run_orchestrator.cancel_active_from_query.assert_awaited_once_with(
+        query,
+        allow_other_actors=False,
+    )
+    assert query.variables['_stop_command_run_id'] == 'run-stop'
+    app.agent_run_orchestrator.try_claim_steering_from_query.assert_not_awaited()
+
+
@pytest.mark.asyncio
@pytest.mark.parametrize('text,expected', [('ordinary chat', False), ('qa: hello', True)])
@pytest.mark.parametrize('claim_result', [True, False])
@@ -114,6 +154,173 @@ async def test_busy_group_steering_obeys_response_rules(sample_query, text, expe
        app.agent_run_orchestrator.try_claim_steering_from_query.assert_not_awaited()


+@pytest.mark.asyncio
+async def test_busy_group_stop_bypasses_response_rule_gate(sample_query):
+    from langbot_plugin.api.entities.builtin.provider.session import LauncherTypes, Session
+    from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain
+
+    app = make_app()
+    app.sess_mgr.get_session.return_value = Session(
+        instance_uuid='test-instance',
+        workspace_uuid='test-workspace',
+        placement_generation=1,
+        launcher_type=LauncherTypes.GROUP,
+        launcher_id=sample_query.launcher_id,
+    )
+    pipeline = make_pipeline()
+    pipeline.pipeline_entity.config['trigger'] = {
+        'group-respond-rules': {'at': True, 'prefix': [], 'regexp': [], 'random': 0}
+    }
+    app.pipeline_mgr.get_pipeline_by_uuid.return_value = pipeline
+    app.agent_run_orchestrator.cancel_active_from_query.return_value = 'run-group-stop'
+    sample_query.launcher_type = LauncherTypes.GROUP
+    sample_query.message_chain = MessageChain([Plain(text='!stop')])
+
+    claimed = await Controller(app)._try_claim_steering_before_session_slot(sample_query)
+
+    assert claimed is False
+    app.cmd_mgr.resolve_privilege.assert_awaited_once_with(sample_query)
+    app.agent_run_orchestrator.cancel_active_from_query.assert_awaited_once_with(
+        sample_query,
+        allow_other_actors=False,
+    )
+    assert sample_query.variables['_stop_command_run_id'] == 'run-group-stop'
+    app.agent_run_orchestrator.try_claim_steering_from_query.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_busy_group_admin_can_stop_another_members_run(sample_query):
+    from langbot_plugin.api.entities.builtin.provider.session import LauncherTypes, Session
+    from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain
+
+    app = make_app()
+    app.cmd_mgr.resolve_privilege.return_value = 2
+    app.sess_mgr.get_session.return_value = Session(
+        instance_uuid='test-instance',
+        workspace_uuid='test-workspace',
+        placement_generation=1,
+        launcher_type=LauncherTypes.GROUP,
+        launcher_id=sample_query.launcher_id,
+    )
+    app.pipeline_mgr.get_pipeline_by_uuid.return_value = make_pipeline()
+    app.agent_run_orchestrator.cancel_active_from_query.return_value = 'run-admin-stop'
+    sample_query.launcher_type = LauncherTypes.GROUP
+    sample_query.message_chain = MessageChain([Plain(text='!stop')])
+
+    claimed = await Controller(app)._try_claim_steering_before_session_slot(sample_query)
+
+    assert claimed is False
+    app.agent_run_orchestrator.cancel_active_from_query.assert_awaited_once_with(
+        sample_query,
+        allow_other_actors=True,
+    )
+    assert sample_query.variables['_stop_command_run_id'] == 'run-admin-stop'
+
+
+@pytest.mark.asyncio
+async def test_busy_group_command_is_recognized_after_trigger_prefix_is_removed(sample_query):
+    from langbot_plugin.api.entities.builtin.provider.session import LauncherTypes, Session
+    from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain
+
+    app = make_app()
+    app.sess_mgr.get_session.return_value = Session(
+        instance_uuid='test-instance',
+        workspace_uuid='test-workspace',
+        placement_generation=1,
+        launcher_type=LauncherTypes.GROUP,
+        launcher_id=sample_query.launcher_id,
+    )
+    pipeline = make_pipeline()
+    pipeline.pipeline_entity.config['trigger'] = {
+        'group-respond-rules': {'at': False, 'prefix': ['qa:'], 'regexp': [], 'random': 0}
+    }
+    app.pipeline_mgr.get_pipeline_by_uuid.return_value = pipeline
+    sample_query.launcher_type = LauncherTypes.GROUP
+    sample_query.message_chain = MessageChain([Plain(text='qa: !list')])
+
+    claimed = await Controller(app)._try_claim_steering_before_session_slot(sample_query)
+
+    assert claimed is False
+    app.agent_run_orchestrator.try_claim_steering_from_query.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_stop_is_handled_before_an_available_second_session_slot():
+    from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain
+
+    app = make_app()
+    app.query_pool = QueryPool()
+    session = SimpleNamespace(_semaphore=asyncio.Semaphore(2))
+    app.sess_mgr.get_session = AsyncMock(return_value=session)
+    app.persistence_mgr = SimpleNamespace(mode=SimpleNamespace(value='oss'))
+    app.workspace_service = SimpleNamespace(
+        get_execution_binding=AsyncMock(
+            return_value=SimpleNamespace(
+                instance_uuid='instance-test',
+                workspace_uuid='workspace-test',
+                placement_generation=1,
+            )
+        )
+    )
+    runtime_pipeline = SimpleNamespace(run=AsyncMock())
+    app.pipeline_mgr.get_pipeline_by_uuid = AsyncMock(side_effect=[make_pipeline(), runtime_pipeline])
+    app.agent_run_orchestrator.cancel_active_from_query.return_value = 'run-already-active'
+
+    worker_tasks = []
+    task_created = asyncio.Event()
+
+    def create_task(coro, **_kwargs):
+        task = asyncio.create_task(coro)
+        worker_tasks.append(task)
+        task_created.set()
+        return task
+
+    app.task_mgr = SimpleNamespace(create_task=create_task)
+    query = Mock()
+    query.query_id = 5
+    query.bot_uuid = 'bot-test'
+    query.pipeline_uuid = 'pipeline-stop'
+    query.message_chain = MessageChain([Plain(text='!stop')])
+    query.variables = {}
+    query._execution_context = ExecutionContext(
+        instance_uuid='instance-test',
+        workspace_uuid='workspace-test',
+        placement_generation=1,
+        pipeline_uuid='pipeline-stop',
+    )
+
+    with patch('langbot.pkg.pipeline.pool.pipeline_query.Query', return_value=query):
+        await app.query_pool.add_query(
+            bot_uuid='bot-test',
+            launcher_type=Mock(),
+            launcher_id='launcher-test',
+            sender_id='sender-test',
+            message_event=Mock(),
+            message_chain=query.message_chain,
+            adapter=Mock(),
+            pipeline_uuid='pipeline-stop',
+            execution_context=query._execution_context,
+        )
+
+    controller = Controller(app)
+    consumer_task = asyncio.create_task(controller.consumer())
+    try:
+        await asyncio.wait_for(task_created.wait(), timeout=1)
+        await asyncio.wait_for(worker_tasks[0], timeout=1)
+    finally:
+        consumer_task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await consumer_task
+
+    app.agent_run_orchestrator.cancel_active_from_query.assert_awaited_once_with(
+        query,
+        allow_other_actors=False,
+    )
+    assert query.variables['_stop_command_run_id'] == 'run-already-active'
+    runtime_pipeline.run.assert_awaited_once_with(query)
+    assert session._semaphore._value == 2
+
+
@pytest.mark.asyncio
 async def test_consumer_transfers_query_from_queue_to_running_task():
    app = make_app()
@@ -65,6 +65,75 @@ async def test_group_message_no_match(mock_app, sample_query):
    mock_rule.match.assert_called_once()


+@pytest.mark.asyncio
+async def test_group_stop_command_bypasses_response_rules(mock_app, sample_query):
+    resprule, entities, rule, _ = get_modules()
+    mock_app.instance_config.data['command'] = {'enable': True, 'prefix': ['!', '！']}
+    sample_query.launcher_type = provider_session.LauncherTypes.GROUP
+    sample_query.launcher_id = '12345'
+    sample_query.message_chain = platform_message.MessageChain([platform_message.Plain(text='!stop')])
+    sample_query.pipeline_config = {
+        'trigger': {'group-respond-rules': {'at': True, 'prefix': [], 'regexp': [], 'random': 0}}
+    }
+    mock_rule = Mock(spec=rule.GroupRespondRule)
+    mock_rule.match = AsyncMock()
+    stage = resprule.GroupRespondRuleCheckStage(mock_app)
+    await stage.initialize(sample_query.pipeline_config)
+    stage.rule_matchers = [mock_rule]
+
+    result = await stage.process(sample_query, 'GroupRespondRuleCheckStage')
+
+    assert result.result_type == entities.ResultType.CONTINUE
+    mock_rule.match.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_other_group_commands_still_obey_response_rules(mock_app, sample_query):
+    resprule, entities, rule, rule_entities = get_modules()
+    mock_app.instance_config.data['command'] = {'enable': True, 'prefix': ['!', '！']}
+    sample_query.launcher_type = provider_session.LauncherTypes.GROUP
+    sample_query.launcher_id = '12345'
+    sample_query.message_chain = platform_message.MessageChain([platform_message.Plain(text='!list')])
+    sample_query.pipeline_config = {
+        'trigger': {'group-respond-rules': {'at': True, 'prefix': [], 'regexp': [], 'random': 0}}
+    }
+    mock_rule = Mock(spec=rule.GroupRespondRule)
+    mock_rule.match = AsyncMock(
+        return_value=rule_entities.RuleJudgeResult(matching=False, replacement=sample_query.message_chain)
+    )
+    stage = resprule.GroupRespondRuleCheckStage(mock_app)
+    await stage.initialize(sample_query.pipeline_config)
+    stage.rule_matchers = [mock_rule]
+
+    result = await stage.process(sample_query, 'GroupRespondRuleCheckStage')
+
+    assert result.result_type == entities.ResultType.INTERRUPT
+    mock_rule.match.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_group_stop_obeys_response_rules_when_commands_are_disabled(mock_app, sample_query):
+    resprule, entities, rule, rule_entities = get_modules()
+    mock_app.instance_config.data['command'] = {'enable': False, 'prefix': ['!']}
+    sample_query.launcher_type = provider_session.LauncherTypes.GROUP
+    sample_query.message_chain = platform_message.MessageChain([platform_message.Plain(text='!stop')])
+    sample_query.pipeline_config = {
+        'trigger': {'group-respond-rules': {'at': True, 'prefix': [], 'regexp': [], 'random': 0}}
+    }
+    mock_rule = Mock(spec=rule.GroupRespondRule)
+    mock_rule.match = AsyncMock(
+        return_value=rule_entities.RuleJudgeResult(matching=False, replacement=sample_query.message_chain)
+    )
+    stage = resprule.GroupRespondRuleCheckStage(mock_app)
+    await stage.initialize(sample_query.pipeline_config)
+    stage.rule_matchers = [mock_rule]
+
+    result = await stage.process(sample_query, 'GroupRespondRuleCheckStage')
+
+    assert result.result_type == entities.ResultType.INTERRUPT
+    mock_rule.match.assert_awaited_once()
+
+
@pytest.mark.asyncio
 async def test_random_response_draw_is_reused_across_queue_rechecks(sample_query, mock_app, monkeypatch):
    from langbot.pkg.pipeline.resprule.rules.random import RandomRespRule
@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 import types
 from importlib import import_module
+from unittest.mock import AsyncMock, Mock

 import pytest

@@ -118,6 +119,7 @@ def build_query(**overrides):
        query_id=1,
        pipeline_config={},
        variables={},
+        resp_messages=[],
        adapter=None,
        launcher_type=types.SimpleNamespace(value='private'),
        launcher_id='user',
@@ -211,6 +213,152 @@ async def test_lane_step_owns_the_apis_and_runner_it_calls():
    assert shapes[4] == ('pipeline', 'run', lane_node, event_node)


+async def test_cancelled_lane_finalizes_the_originating_stream():
+    pipelinemgr, _execution = get_modules()
+    from langbot_plugin.api.entities.builtin.provider.message import MessageChunk
+
+    counters, ap = build_counters()
+    adapter = types.SimpleNamespace(
+        is_stream_output_supported=AsyncMock(return_value=True),
+        reply_message_chunk=AsyncMock(),
+    )
+    query = build_query(
+        adapter=adapter,
+        message_event=object(),
+        pipeline_config={'output': {'misc': {'quote-origin': True}}},
+        variables={'_runner_cancel_reason': 'stop_command'},
+        resp_messages=[
+            MessageChunk(
+                role='assistant',
+                content='partial',
+                all_content='partial',
+                resp_message_id='stream-1',
+                msg_sequence=3,
+                is_final=False,
+            )
+        ],
+    )
+    lane = build_lane(pipelinemgr, ap)
+
+    async def cancelled(_query):
+        raise asyncio.CancelledError
+
+    lane._process_query = cancelled
+
+    await lane.process_query(query)
+    await stop_counters(counters)
+
+    adapter.reply_message_chunk.assert_awaited_once()
+    call = adapter.reply_message_chunk.await_args.kwargs
+    assert call['is_final'] is True
+    assert call['bot_message'].is_final is True
+    assert call['bot_message'].content == 'Task was cancelled'
+    assert call['bot_message'].resp_message_id == 'stream-1'
+    assert call['bot_message'].msg_sequence == 4
+
+
+async def test_cancelled_lane_finalizes_non_streaming_response():
+    pipelinemgr, _execution = get_modules()
+    counters, ap = build_counters()
+    adapter = types.SimpleNamespace(
+        is_stream_output_supported=AsyncMock(return_value=False),
+        reply_message=AsyncMock(),
+    )
+    query = build_query(
+        adapter=adapter,
+        message_event=object(),
+        pipeline_config={'output': {'misc': {'quote-origin': False}}},
+        variables={'_runner_cancel_reason': 'stop_command'},
+    )
+    lane = build_lane(pipelinemgr, ap)
+
+    async def cancelled(_query):
+        raise asyncio.CancelledError
+
+    lane._process_query = cancelled
+
+    await lane.process_query(query)
+    await stop_counters(counters)
+
+    adapter.reply_message.assert_awaited_once()
+    assert adapter.reply_message.await_args.kwargs['quote_origin'] is False
+
+
+async def test_cancelled_lane_logs_finalization_failure():
+    pipelinemgr, _execution = get_modules()
+    from langbot_plugin.api.entities.builtin.provider.message import MessageChunk
+
+    counters, ap = build_counters()
+    ap.logger = types.SimpleNamespace(warning=Mock())
+    adapter = types.SimpleNamespace(
+        is_stream_output_supported=AsyncMock(side_effect=RuntimeError('delivery failed')),
+    )
+    query = build_query(
+        adapter=adapter,
+        message_event=object(),
+        pipeline_config={'output': {'misc': {'quote-origin': False}}},
+        variables={'_runner_cancel_reason': 'stop_command'},
+        resp_messages=[MessageChunk(role='assistant', content='partial', resp_message_id='stream-1')],
+    )
+    lane = build_lane(pipelinemgr, ap)
+
+    async def cancelled(_query):
+        raise asyncio.CancelledError
+
+    lane._process_query = cancelled
+
+    await lane.process_query(query)
+    await stop_counters(counters)
+
+    ap.logger.warning.assert_called_once()
+    assert ap.logger.warning.call_args.args[0] == 'Failed to finalize cancelled platform response: %s'
+    assert isinstance(ap.logger.warning.call_args.args[1], RuntimeError)
+
+
+async def test_lifecycle_cancel_closes_existing_stream_without_user_notice():
+    pipelinemgr, _execution = get_modules()
+    from langbot_plugin.api.entities.builtin.provider.message import MessageChunk
+
+    counters, ap = build_counters()
+    adapter = types.SimpleNamespace(
+        is_stream_output_supported=AsyncMock(return_value=True),
+        reply_message_chunk=AsyncMock(),
+        reply_message=AsyncMock(),
+    )
+    query = build_query(
+        adapter=adapter,
+        message_event=object(),
+        pipeline_config={'output': {'misc': {'quote-origin': False}}},
+        resp_messages=[
+            MessageChunk(
+                role='assistant',
+                content='partial',
+                all_content='partial response',
+                resp_message_id='stream-life',
+                msg_sequence=2,
+                is_final=False,
+            )
+        ],
+    )
+    lane = build_lane(pipelinemgr, ap)
+
+    async def cancelled(_query):
+        raise asyncio.CancelledError
+
+    lane._process_query = cancelled
+
+    with pytest.raises(asyncio.CancelledError):
+        await lane.process_query(query)
+    await stop_counters(counters)
+
+    adapter.reply_message_chunk.assert_awaited_once()
+    call = adapter.reply_message_chunk.await_args.kwargs
+    assert call['bot_message'].content == 'partial response'
+    assert call['bot_message'].resp_message_id == 'stream-life'
+    assert call['is_final'] is True
+    adapter.reply_message.assert_not_awaited()
+
+
 async def test_lane_node_nests_under_the_platform_route_that_called_it():
    pipelinemgr, execution = get_modules()
    trace = get_trace()