Spaces:
Running
Running
| """stop_subagent — cooperatively ask a running sub-agent to stop.""" | |
| from __future__ import annotations | |
| import logging | |
| from typing import Any | |
| from frontier_agent.components.agent_bus import AgentBus | |
| from frontier_agent.components.agent_bus.stop_signal import get_stop_registry | |
| from frontier_agent.core.execution_context import get_current_execution_scope | |
| from frontier_agent.core.runtime import registry | |
| from frontier_agent.core.tool import tool | |
| from plugins.tools._bus_scope import resolve_bus_task_id | |
| logger = logging.getLogger(__name__) | |
| def _current_metadata(bus: AgentBus, session_id: str) -> dict[str, Any]: | |
| """Metadata of the session's running job, or ``{}`` if it cannot be read. | |
| The existence check is ``hasattr`` and the call is made on ``bus`` directly. | |
| Binding the attribute first — ``reader = getattr(bus, ..., None)`` then | |
| ``callable(reader)`` — narrows it to a bare ``(...) -> object``, which makes | |
| the ``dict()`` around the result untypeable; going through ``bus`` keeps the | |
| declared signature. ``hasattr`` also beats catching ``AttributeError``, which | |
| would swallow one raised *inside* a real implementation. | |
| A bus-shaped stand-in that predates the accessor is the case being tolerated | |
| (the tests use one). Any other failure to read metadata degrades to "not a | |
| publish job" rather than turning a stop request into an exception, and says | |
| so in the log. | |
| """ | |
| if not hasattr(bus, "current_job_metadata"): | |
| return {} | |
| try: | |
| return dict(bus.current_job_metadata(session_id) or {}) | |
| except Exception: | |
| logger.warning( | |
| "stop_subagent: could not read running-job metadata for %s", | |
| session_id, exc_info=True, | |
| ) | |
| return {} | |
| async def stop_subagent(agent_name: str, force: bool = False) -> str: | |
| """Ask a running sub-agent to stop exploring as soon as possible. | |
| Use this when a sub-agent is clearly going off-track, looping on the same | |
| searches, no longer needed, or burning budget. The sub-agent is told to | |
| stop and MAY submit a brief report if it already has valuable findings — | |
| otherwise it just stops. The sub-agent is NOT destroyed: you can assign it | |
| new work later. Pick up whatever it returns with ``collect_reports``. | |
| A sub-agent running a PUBLISH task is refused unless ``force`` is set: that | |
| task writes the run's deliverable, and stopping it throws the deliverable | |
| away. | |
| Args: | |
| agent_name: Name of the sub-agent to stop (the name you created / | |
| assigned it with). | |
| force: Stop even a publish task. Only when the publisher is genuinely | |
| wedged and you accept losing what it was writing. | |
| Returns: | |
| A short status line describing what happened. | |
| """ | |
| scope = get_current_execution_scope() | |
| if scope is None: | |
| return ( | |
| "Error: stop_subagent can only be called inside an active " | |
| "ReAct execution." | |
| ) | |
| name = (agent_name or "").strip() | |
| if not name: | |
| return "Error: `agent_name` was empty." | |
| bus = registry.get(AgentBus) | |
| task_id = resolve_bus_task_id(scope) | |
| session_id = f"{task_id}::{name}" | |
| session = bus.get_session(session_id) | |
| if session is None: | |
| return f"Error: sub-agent {name!r} not found (was it created?)." | |
| if session.current_job_id is None: | |
| return ( | |
| f"{name} has no task running right now — nothing to stop. " | |
| "(It may have already finished; call collect_reports.)" | |
| ) | |
| # A publish task is the one job whose whole value is the file it writes, and | |
| # ``StopSignalObserver`` stops by POPPING the LLM turn it just produced — | |
| # so a stop landing between "decided to write" and "wrote" discards the | |
| # deliverable and leaves nothing behind. Seen for real: a coordinator | |
| # dispatched a publish task and stopped it ~5 s later, four times running, | |
| # burning its publisher's whole task budget while its finished answer never | |
| # reached /outputs. Publish tasks are short by construction — waiting is | |
| # almost always right, so make the override explicit. | |
| if force is not True and _current_metadata(bus, session_id).get("can_publish") is True: | |
| return ( | |
| f"Refusing to stop {name}: it is running the PUBLISH task, which " | |
| "writes the final deliverable, and stopping it now would throw " | |
| "that write away. Publish tasks are short — call collect_reports " | |
| "to wait for it. Pass force=true only if the publisher is truly " | |
| "wedged and you accept losing what it was writing." | |
| ) | |
| # Non-blocking: just register the signal and return. We do NOT wait for | |
| # the sub-agent to actually stop — it reacts on its own next LLM response. | |
| # Tag the request with the running job so a stop that outlives its job | |
| # can't roll back a later task on this (reusable) session. | |
| get_stop_registry().request_stop(session_id, session.current_job_id or "") | |
| logger.info("stop_subagent: stop signal queued for %s", session_id) | |
| return ( | |
| f"Stop signal sent to {name}; it will stop exploring shortly and " | |
| "may submit a brief report if it has valuable findings." | |
| ) | |
| __all__ = ["stop_subagent"] | |