From 5be3b03e77c400aa8daff6e29779a0e36a7e9659 Mon Sep 17 00:00:00 2001 From: Martin <46433701+Nikarhub@users.noreply.github.com> Date: Sat, 31 Oct 2026 21:52:25 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E5=96=84=20Core=20Runner=20=E5=8F=96?= =?UTF-8?q?=E6=B6=88=E4=B8=8E=E7=BE=A4=E8=81=8A=E6=8E=A7=E5=88=B6=E5=91=BD?= =?UTF-8?q?=E4=BB=A4=20(#2821)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(core): add generic runner cancellation control * fix(core): 完善 Runner 取消与群聊控制命令 * fix(core): 支持多并发会话及时停止任务 --- src/langbot/pkg/agent/runner/orchestrator.py | 54 +++++ .../pkg/agent/runner/run_ledger_store.py | 2 +- .../pkg/agent/runner/session_registry.py | 56 +++++ src/langbot/pkg/command/cmdmgr.py | 24 +- src/langbot/pkg/pipeline/controller.py | 65 ++++++ src/langbot/pkg/pipeline/pipelinemgr.py | 58 ++++- src/langbot/pkg/pipeline/resprule/resprule.py | 15 ++ .../agent/test_orchestrator_integration.py | 162 +++++++++++++- .../unit_tests/agent/test_run_ledger_store.py | 17 ++ .../unit_tests/agent/test_session_registry.py | 101 +++++++++ tests/unit_tests/pipeline/test_controller.py | 209 +++++++++++++++++- tests/unit_tests/pipeline/test_resprule.py | 69 ++++++ .../telemetry/test_pipeline_lane.py | 148 +++++++++++++ 13 files changed, 967 insertions(+), 13 deletions(-) diff --git a/src/langbot/pkg/agent/runner/orchestrator.py b/src/langbot/pkg/agent/runner/orchestrator.py index 0e04cc67a..fab4b8ef6 100644 --- a/src/langbot/pkg/agent/runner/orchestrator.py +++ b/src/langbot/pkg/agent/runner/orchestrator.py @@ -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, diff --git a/src/langbot/pkg/agent/runner/run_ledger_store.py b/src/langbot/pkg/agent/runner/run_ledger_store.py index ade45de91..4aa87c3a2 100644 --- a/src/langbot/pkg/agent/runner/run_ledger_store.py +++ b/src/langbot/pkg/agent/runner/run_ledger_store.py @@ -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) diff --git a/src/langbot/pkg/agent/runner/session_registry.py b/src/langbot/pkg/agent/runner/session_registry.py index e7c12716b..0e63df5b2 100644 --- a/src/langbot/pkg/agent/runner/session_registry.py +++ b/src/langbot/pkg/agent/runner/session_registry.py @@ -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, diff --git a/src/langbot/pkg/command/cmdmgr.py b/src/langbot/pkg/command/cmdmgr.py index 31f28dfdf..1ac8db113 100644 --- a/src/langbot/pkg/command/cmdmgr.py +++ b/src/langbot/pkg/command/cmdmgr.py @@ -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, diff --git a/src/langbot/pkg/pipeline/controller.py b/src/langbot/pkg/pipeline/controller.py index bb042b591..65a2dd05e 100644 --- a/src/langbot/pkg/pipeline/controller.py +++ b/src/langbot/pkg/pipeline/controller.py @@ -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 diff --git a/src/langbot/pkg/pipeline/pipelinemgr.py b/src/langbot/pkg/pipeline/pipelinemgr.py index 7383032b7..194f61478 100644 --- a/src/langbot/pkg/pipeline/pipelinemgr.py +++ b/src/langbot/pkg/pipeline/pipelinemgr.py @@ -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 diff --git a/src/langbot/pkg/pipeline/resprule/resprule.py b/src/langbot/pkg/pipeline/resprule/resprule.py index 3e8c16a96..c541880e4 100644 --- a/src/langbot/pkg/pipeline/resprule/resprule.py +++ b/src/langbot/pkg/pipeline/resprule/resprule.py @@ -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. diff --git a/tests/unit_tests/agent/test_orchestrator_integration.py b/tests/unit_tests/agent/test_orchestrator_integration.py index 602b7a93c..e50fc9984 100644 --- a/tests/unit_tests/agent/test_orchestrator_integration.py +++ b/tests/unit_tests/agent/test_orchestrator_integration.py @@ -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.""" diff --git a/tests/unit_tests/agent/test_run_ledger_store.py b/tests/unit_tests/agent/test_run_ledger_store.py index 3f3b2b601..6d120221b 100644 --- a/tests/unit_tests/agent/test_run_ledger_store.py +++ b/tests/unit_tests/agent/test_run_ledger_store.py @@ -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'): diff --git a/tests/unit_tests/agent/test_session_registry.py b/tests/unit_tests/agent/test_session_registry.py index 6b476f56b..0da11409c 100644 --- a/tests/unit_tests/agent/test_session_registry.py +++ b/tests/unit_tests/agent/test_session_registry.py @@ -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.""" diff --git a/tests/unit_tests/pipeline/test_controller.py b/tests/unit_tests/pipeline/test_controller.py index 19f4e7f7a..33ca50d45 100644 --- a/tests/unit_tests/pipeline/test_controller.py +++ b/tests/unit_tests/pipeline/test_controller.py @@ -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() diff --git a/tests/unit_tests/pipeline/test_resprule.py b/tests/unit_tests/pipeline/test_resprule.py index b66814e31..7edea001e 100644 --- a/tests/unit_tests/pipeline/test_resprule.py +++ b/tests/unit_tests/pipeline/test_resprule.py @@ -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 diff --git a/tests/unit_tests/telemetry/test_pipeline_lane.py b/tests/unit_tests/telemetry/test_pipeline_lane.py index 80e635436..1a7424e0e 100644 --- a/tests/unit_tests/telemetry/test_pipeline_lane.py +++ b/tests/unit_tests/telemetry/test_pipeline_lane.py @@ -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()