hope-voice-api/tests/test_background_tasks.py

"""Regression tests for background-task routing and event isolation."""
from __future__ import annotations

import asyncio
import json
import sys
import uuid
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).parents[1] / "agent"))
from task_registry import TaskRegistry  # noqa: E402
from task_router import should_force_background_task  # noqa: E402
from task_worker import TaskWorker  # noqa: E402


@pytest.mark.parametrize(
    "utterance",
    [
        "What are the top news stories today?",
        "Compare the PS5 Pro and Xbox Series X for me.",
        "Plan a weekend trip to Denver.",
        "Write a short birthday message for my sister.",
        "Write a birthday note for Sam.",
        "Look up the current price of a PlayStation 5.",
        "Research the latest solid state battery developments.",
        "Tell me about quantum computing.",
        "Look online and give me a recipe for making buttercream icing.",
    ],
)
def test_nontrivial_requests_force_dispatch(utterance):
    assert should_force_background_task(utterance)


@pytest.mark.parametrize(
    "utterance",
    [
        "What's my name?",
        "Remember that I prefer tea.",
        "Read my saved notes.",
        "What's the weather in Raleigh?",
        "What time is it?",
        "Hello there.",
    ],
)
def test_direct_requests_do_not_force_dispatch(utterance):
    assert not should_force_background_task(utterance)


def test_deferred_task_answers_are_rejected():
    assert TaskWorker._is_deferred_answer(
        "I need to perform a more targeted search. Would you like me to continue?"
    )
    assert TaskWorker._is_deferred_answer("I can refine this if you'd like.")
    assert not TaskWorker._is_deferred_answer(
        "Reuters reports that the agreement was signed today, while the BBC says voting begins Monday."
    )

    news_worker = TaskWorker("test", "Find the latest international news headlines")
    assert news_worker._is_underfilled_news_answer(
        "One flood was reported. Other sources indicate ongoing coverage of world events."
    )
    assert not news_worker._is_underfilled_news_answer(
        "Nepal began flood recovery, Venezuela signed an oil agreement, and Iran resumed talks."
    )

    recipe_worker = TaskWorker("test", "Give me a buttercream recipe")
    assert recipe_worker._is_underfilled_recipe_answer(
        "Use butter, powdered sugar, vanilla, and milk, then mix until smooth."
    )
    assert not recipe_worker._is_underfilled_recipe_answer(
        "Beat one cup butter with three cups powdered sugar and two tablespoons milk until smooth."
    )


def test_task_event_directories_are_isolated(tmp_path):
    first = TaskRegistry(tmp_path / "room-one")
    second = TaskRegistry(tmp_path / "room-two")

    task = first.create("private task")
    first.complete(task.id, "private result")

    first_events = asyncio.run(first.poll_events())
    second_events = asyncio.run(second.poll_events())

    assert [event[1]["event"] for event in first_events] == ["created", "completed"]
    assert second_events == []


def test_task_summary_reports_live_status(tmp_path):
    """task_status/list_tasks summarise live state including a stall hint."""
    import time as _time

    sys.path.insert(0, str(Path(__file__).parents[1] / "agent"))
    from task_registry import status_summary  # noqa: E402
    import task_registry as tr_mod  # noqa: E402

    # Isolate: status_summary reads the module-level registry.
    held = tr_mod.registry
    iso = tr_mod.TaskRegistry(tmp_path)
    tr_mod.registry = iso
    try:
        task = iso.create("test task")
        task_id = task.id
        iso.add_step(task_id, "thinking", "computing")
        summary = status_summary(task_id, stalled_s=45.0)
        assert "RUNNING" in summary, summary
        assert "step(s)" in summary, summary

        # Stalled: force last activity far in the past.
        task.last_activity = _time.time() - 100
        iso._save_task(task)
        assert "STALLED" in status_summary(task_id, stalled_s=45.0)

        # Completed with result.
        iso.complete(task_id, "the answer")
        done = status_summary(task_id, stalled_s=45.0)
        assert "COMPLETED" in done, done
        assert "the answer" in done, done

        # Unknown id.
        assert "was not found" in status_summary("nope", stalled_s=45.0)
    finally:
        tr_mod.registry = held


def test_registry_tracks_last_activity(tmp_path):
    reg = TaskRegistry(tmp_path)
    task = reg.create("tracked")
    before = task.last_activity
    task.last_activity = before - 10
    reg.add_step(task.id, "thinking", "step")
    data = reg.get(task.id)
    assert data["last_activity"] > before - 10, data["last_activity"]


@pytest.mark.slow
def test_dispatch_mcp_task_reaches_terminal_event():
    """The MCP request returns immediately while its child task stays alive."""
    from conftest import AGENT_DIR_IN_CONTAINER, CONTAINER_PYTHON, docker_exec_python

    task_dir = f"/tmp/task-test-{uuid.uuid4().hex}"
    code = f"""
import asyncio, json, os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    task_dir = {task_dir!r}
    os.makedirs(task_dir, exist_ok=True)
    env = {{**os.environ, "TASKS_DIR": task_dir, "TASK_TIMEOUT_SECONDS": "60"}}
    params = StdioServerParameters(
        command={CONTAINER_PYTHON!r},
        args=[{AGENT_DIR_IN_CONTAINER + '/dispatch_mcp.py'!r}],
        env=env,
    )
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool(
                "dispatch_task",
                {{"description": "Write a warm two-sentence birthday greeting for Sam."}},
            )
            assert not result.isError
            for _ in range(60):
                await asyncio.sleep(0.5)
                with open(os.path.join(task_dir, "events.jsonl"), encoding="utf-8") as f:
                    events = [json.loads(line) for line in f if line.strip()]
                terminal = [e for e in events if e.get("event") in ("completed", "failed")]
                if terminal:
                    print(json.dumps(terminal[-1]))
                    return
            raise RuntimeError("background task produced no terminal event")

asyncio.run(main())
"""
    try:
        proc = docker_exec_python(code, timeout=90, check=False)
        assert proc.returncode == 0, proc.stderr[-1200:]
        terminal = json.loads(proc.stdout.strip().splitlines()[-1])
        assert terminal["event"] == "completed", terminal
        assert terminal.get("result")
    finally:
        # Best-effort cleanup; a failed container connection should not mask the assertion.
        try:
            from conftest import docker_exec

            docker_exec("rm", "-rf", task_dir, check=False)
        except Exception:
            pass