plugins/tools/stop_subagent.py

"""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 {}


@tool
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"]