hermes-agent/tests/tools/test_finite_delegation_outcomes.py

"""Finite dispatch returns sibling outcomes, not an orphaned async handle.

Only child construction/conversation is synthetic: dispatch, parallel executor,
child timeout/status conversion, interrupt propagation and cleanup are real.
"""
from __future__ import annotations

import json
import threading
from types import SimpleNamespace

import pytest

from agent.interrupt_control import InterruptControlMixin
from gateway import session_context as sc
from tools import async_delegation, delegate_tool as dt
from tools.delegate_tool_child_run import _attach_child
from tools.process_registry import process_registry


class _Parent(InterruptControlMixin, SimpleNamespace):
    pass


class _Child:
    def __init__(self, outcome="completed"):
        self.outcome = outcome
        self.tool_progress_callback = None
        self._credential_pool = None
        self._delegate_saved_tool_names = []
        self._delegate_role = "leaf"
        self._delegate_depth = 1
        self._subagent_id = None
        self.model = "test-model"
        self.session_prompt_tokens = self.session_completion_tokens = 0
        self.session_estimated_cost_usd = 0.0
        self.session_cost_status = "unknown"
        self.started = threading.Event()
        self.interrupted = threading.Event()
        self.unwinding = threading.Event()
        self.allow_finish = threading.Event()
        self.finished = threading.Event()
        self.closed = threading.Event()
        self.close_while_running = False
        self.worker = None

    def run_conversation(self, **_kwargs):
        self.worker = threading.current_thread()
        self.started.set()
        try:
            if self.outcome == "slow":
                assert self.interrupted.wait(5), "child never received stop"
                self.unwinding.set()
                assert self.allow_finish.wait(5), "child teardown was not released"
                return {"final_response": "partial", "completed": False,
                        "interrupted": True, "api_calls": 1, "messages": []}
            if self.outcome == "error":
                raise RuntimeError("synthetic child exception")
            if self.outcome == "failed":
                return {"final_response": "provider rejection", "completed": False,
                        "failed": True, "error": "synthetic provider rejection",
                        "api_calls": 1, "messages": []}
            return {"final_response": "sibling evidence", "completed": True,
                    "api_calls": 1, "messages": []}
        finally:
            self.finished.set()

    def hard_interrupt(self, _reason=None, **_kwargs):
        self.interrupted.set()

    def get_activity_summary(self):
        return {"api_call_count": 1}

    def close(self):
        self.close_while_running |= not self.finished.is_set()
        self.closed.set()


@pytest.fixture
def harness(monkeypatch, tmp_path):
    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    monkeypatch.setenv("HERMES_SINGLE_QUERY_SESSION", "1")
    tokens = [(v, v.set(sc._UNSET)) for v in sc._VAR_MAP.values()]
    tokens += [(v, v.set(sc._UNSET)) for v in
               (sc._SESSION_ASYNC_DELIVERY, sc._SESSION_HISTORY_DELIVERY)]
    async_delegation._reset_for_tests()
    parent = _Parent(session_id="finite-outcomes", _delegate_depth=0,
                     _current_task_id=None, _interrupt_requested=False,
                     _execution_thread_id=None, quiet_mode=True,
                     _active_children=[], _active_children_lock=threading.Lock())
    children = []

    def build_child(task_index, parent_agent, **_kwargs):
        child = children[task_index]
        _attach_child(parent_agent, child)
        return child

    monkeypatch.setattr(dt, "_build_child_agent", build_child)
    monkeypatch.setattr(dt, "_load_config", lambda: {})
    monkeypatch.setattr(dt, "_get_max_concurrent_children", lambda: 2)
    monkeypatch.setattr(dt, "_get_worktree_isolation", lambda: False)
    monkeypatch.setattr(dt, "_get_child_timeout", lambda: 4)
    monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *_a, **_k: {
        "model": "test-model", "provider": None, "base_url": None,
        "api_key": None, "api_mode": None, "command": None, "args": None,
    })

    def dispatch(*outcomes):
        children.extend(_Child(outcome) for outcome in outcomes)
        return json.loads(dt.delegate_task(
            tasks=[{"goal": f"Exercise child outcome {i}"} for i in range(len(children))],
            background=True, parent_agent=parent,
        ))

    yield parent, children, dispatch
    # Release on assertion failure too. Never leave conversation workers behind.
    for child in children:
        child.interrupted.set()
        child.allow_finish.set()
    async_delegation._reset_for_tests()
    for child in children:
        if child.started.is_set():
            assert child.finished.wait(5)
            assert child.closed.wait(5)
            child.worker.join(5)
            assert not child.worker.is_alive()
            assert not child.close_while_running
    assert not parent._active_children
    while not process_registry.completion_queue.empty():
        process_registry.completion_queue.get_nowait()
    for var, token in reversed(tokens):
        var.reset(token)


def _joined(result, statuses):
    assert result.get("status") != "dispatched", result
    assert [entry["task_index"] for entry in result["results"]] == [0, 1]
    assert [entry["status"] for entry in result["results"]] == statuses
    assert result["results"][0]["summary"] == "sibling evidence"
    assert process_registry.completion_queue.empty()


@pytest.mark.parametrize("outcome", ["error", "failed"])
def test_finite_batch_returns_success_and_child_failure(harness, outcome):
    _parent, children, dispatch = harness
    result = dispatch("completed", outcome)
    _joined(result, ["completed", outcome])
    assert "synthetic" in result["results"][1]["error"]
    assert result["results"][1]["exit_reason"] == "error"
    assert all(child.closed.is_set() for child in children)


def test_finite_batch_returns_timeout_while_child_unwinds(harness, monkeypatch):
    _parent, children, dispatch = harness
    monkeypatch.setattr(dt, "_get_child_timeout", lambda: 0.5)
    result = dispatch("completed", "slow")
    _joined(result, ["completed", "timeout"])
    assert "timed out" in result["results"][1]["error"]
    assert result["results"][1]["exit_reason"] == "timeout"
    slow = children[1]
    assert slow.unwinding.wait(1)
    assert not slow.finished.is_set()
    assert not slow.closed.is_set(), "timeout must not close an unwinding child"
    slow.allow_finish.set()
    assert slow.closed.wait(2)


def test_finite_batch_returns_parent_interruption(harness):
    parent, children, dispatch = harness
    children.extend([_Child(), _Child("slow")])
    slow = children[1]
    stop_errors = []

    def stop_parent():
        try:
            assert slow.started.wait(3)
            assert children[0].closed.wait(3)
            parent.hard_interrupt("test parent stop")
        except Exception as exc:
            stop_errors.append(exc)
        finally:
            slow.allow_finish.set()

    stopper = threading.Thread(target=stop_parent)
    stopper.start()
    try:
        result = dispatch()
        _joined(result, ["completed", "interrupted"])
        assert parent._interrupt_requested is True
        assert slow.interrupted.is_set()
    finally:
        slow.allow_finish.set()
        stopper.join(5)
        assert not stopper.is_alive()
        assert not stop_errors


def test_finite_marker_overrides_api_history_continuation(harness):
    _parent, _children, dispatch = harness
    sc.set_session_vars(platform="api_server", chat_id="history-session",
                        session_key="history-session", session_id="history-session",
                        session_history_delivery="1", async_delivery=False)
    assert sc.session_history_delivery_supported()
    result = dispatch("completed", "error")
    _joined(result, ["completed", "error"])


@pytest.mark.parametrize("marker", [None, "0", "false"])
@pytest.mark.parametrize("api_history", [False, True])
def test_nonfinite_marker_preserves_background_dispatch(harness, monkeypatch, marker, api_history):
    _parent, children, dispatch = harness
    if marker is None:
        monkeypatch.delenv("HERMES_SINGLE_QUERY_SESSION")
    else:
        monkeypatch.setenv("HERMES_SINGLE_QUERY_SESSION", marker)
    if api_history:
        sc.set_session_vars(platform="api_server", chat_id="history-session",
                            session_key="history-session", session_id="history-session",
                            session_history_delivery="1", async_delivery=False)
    result = dispatch("completed", "completed")
    assert result["status"] == "dispatched", result
    assert result["mode"] == "background"
    assert "results" not in result
    event = process_registry.completion_queue.get(timeout=5)
    assert event["type"] == "async_delegation"
    if api_history:
        assert event["origin_session_id"] == "history-session"
    assert all(child.closed.wait(2) for child in children)