tests/agent/test_subagent_stop_hook.py

"""Tests for the subagent_stop hook event.

Covers wire-up from tools.delegate_tool.delegate_task:
  * fires once per child in both single-task and batch modes
  * runs on the parent thread (no re-entrancy for hook authors)
  * carries child_role when the agent exposes _delegate_role
  * carries child_role=None when _delegate_role is not set (pre-M3)
"""

from __future__ import annotations

import json
import threading
from unittest.mock import MagicMock, patch

import pytest

from tools.delegate_tool import delegate_task
from hermes_cli import plugins


def _make_parent(depth: int = 0, session_id: str = "parent-1"):
    parent = MagicMock()
    parent.base_url = "https://openrouter.ai/api/v1"
    parent.api_key = "***"
    parent.provider = "openrouter"
    parent.api_mode = "chat_completions"
    parent.model = "anthropic/claude-sonnet-4"
    parent.platform = "cli"
    parent.providers_allowed = None
    parent.providers_ignored = None
    parent.providers_order = None
    parent.provider_sort = None
    parent._session_db = None
    parent._delegate_depth = depth
    parent._active_children = []
    parent._active_children_lock = threading.Lock()
    parent._print_fn = None
    parent.tool_progress_callback = None
    parent.thinking_callback = None
    parent._memory_manager = None
    parent.session_id = session_id
    return parent


@pytest.fixture(autouse=True)
def _fresh_plugin_manager():
    """Each test gets a fresh PluginManager so hook callbacks don't
    leak between tests."""
    original = plugins._plugin_manager
    plugins._plugin_manager = plugins.PluginManager()
    yield
    plugins._plugin_manager = original


@pytest.fixture(autouse=True)
def _stub_child_builder(monkeypatch):
    """Replace _build_child_agent with a MagicMock factory so delegate_task
    never transitively imports run_agent / openai.  Keeps the test runnable
    in environments without heavyweight runtime deps installed."""
    def _fake_build_child(task_index, **kwargs):
        child = MagicMock()
        child._delegate_saved_tool_names = []
        child._credential_pool = None
        return child

    monkeypatch.setattr(
        "tools.delegate_tool._build_child_agent", _fake_build_child,
    )


def _register_capturing_hook():
    captured = []

    def _cb(**kwargs):
        kwargs["_thread"] = threading.current_thread()
        captured.append(kwargs)

    mgr = plugins.get_plugin_manager()
    mgr._hooks.setdefault("subagent_stop", []).append(_cb)
    return captured


# ── single-task mode ──────────────────────────────────────────────────────


class TestSingleTask:
    def test_fires_once(self):
        captured = _register_capturing_hook()

        with patch("tools.delegate_tool._run_single_child") as mock_run:
            mock_run.return_value = {
                "task_index": 0,
                "status": "completed",
                "summary": "Done!",
                "api_calls": 3,
                "duration_seconds": 5.0,
                "_child_role": "analyst",
            }
            delegate_task(goal="do X", parent_agent=_make_parent())

        assert len(captured) == 1
        payload = captured[0]
        assert payload["child_role"] == "analyst"
        assert payload["child_status"] == "completed"
        assert payload["child_summary"] == "Done!"
        assert payload["duration_ms"] == 5000

    def test_fires_on_parent_thread(self):
        captured = _register_capturing_hook()
        main_thread = threading.current_thread()

        with patch("tools.delegate_tool._run_single_child") as mock_run:
            mock_run.return_value = {
                "task_index": 0, "status": "completed",
                "summary": "x", "api_calls": 1, "duration_seconds": 0.1,
                "_child_role": None,
            }
            delegate_task(goal="go", parent_agent=_make_parent())

        assert captured[0]["_thread"] is main_thread

    def test_payload_includes_parent_session_id(self):
        captured = _register_capturing_hook()

        with patch("tools.delegate_tool._run_single_child") as mock_run:
            mock_run.return_value = {
                "task_index": 0, "status": "completed",
                "summary": "x", "api_calls": 1, "duration_seconds": 0.1,
                "_child_role": None,
            }
            delegate_task(
                goal="go",
                parent_agent=_make_parent(session_id="sess-xyz"),
            )

        assert captured[0]["parent_session_id"] == "sess-xyz"


# ── batch mode ────────────────────────────────────────────────────────────


class TestBatchMode:
    def test_fires_per_child(self):
        captured = _register_capturing_hook()

        with patch("tools.delegate_tool._run_single_child") as mock_run:
            mock_run.side_effect = [
                {"task_index": 0, "status": "completed",
                 "summary": "A", "api_calls": 1, "duration_seconds": 1.0,
                 "_child_role": "role-a"},
                {"task_index": 1, "status": "completed",
                 "summary": "B", "api_calls": 2, "duration_seconds": 2.0,
                 "_child_role": "role-b"},
                {"task_index": 2, "status": "completed",
                 "summary": "C", "api_calls": 3, "duration_seconds": 3.0,
                 "_child_role": "role-c"},
            ]
            delegate_task(
                tasks=[
                    {"goal": "A"}, {"goal": "B"}, {"goal": "C"},
                ],
                parent_agent=_make_parent(),
            )

        assert len(captured) == 3
        roles = sorted(c["child_role"] for c in captured)
        assert roles == ["role-a", "role-b", "role-c"]

    def test_all_fires_on_parent_thread(self):
        captured = _register_capturing_hook()
        main_thread = threading.current_thread()

        with patch("tools.delegate_tool._run_single_child") as mock_run:
            mock_run.side_effect = [
                {"task_index": 0, "status": "completed",
                 "summary": "A", "api_calls": 1, "duration_seconds": 1.0,
                 "_child_role": None},
                {"task_index": 1, "status": "completed",
                 "summary": "B", "api_calls": 2, "duration_seconds": 2.0,
                 "_child_role": None},
            ]
            delegate_task(
                tasks=[{"goal": "A"}, {"goal": "B"}],
                parent_agent=_make_parent(),
            )

        for payload in captured:
            assert payload["_thread"] is main_thread


# ── payload shape ─────────────────────────────────────────────────────────


class TestPayloadShape:
    def test_role_absent_becomes_none(self):
        captured = _register_capturing_hook()

        with patch("tools.delegate_tool._run_single_child") as mock_run:
            mock_run.return_value = {
                "task_index": 0, "status": "completed",
                "summary": "x", "api_calls": 1, "duration_seconds": 0.1,
                # Deliberately omit _child_role — pre-M3 shape.
            }
            delegate_task(goal="do X", parent_agent=_make_parent())

        assert captured[0]["child_role"] is None

    def test_result_does_not_leak_child_role_field(self):
        """The internal _child_role key must be stripped before the
        result dict is serialised to JSON."""
        _register_capturing_hook()

        with patch("tools.delegate_tool._run_single_child") as mock_run:
            mock_run.return_value = {
                "task_index": 0, "status": "completed",
                "summary": "x", "api_calls": 1, "duration_seconds": 0.1,
                "_child_role": "leaf",
            }
            raw = delegate_task(goal="do X", parent_agent=_make_parent())

        parsed = json.loads(raw)
        assert "results" in parsed
        assert "_child_role" not in parsed["results"][0]