137 lines
4.8 KiB
Python
137 lines
4.8 KiB
Python
"""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."
|
|
)
|
|
|
|
|
|
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 == []
|
|
|
|
|
|
@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
|