"""Tests for task and team tools.""" from __future__ import annotations import asyncio from pathlib import Path import pytest from openharness.coordinator.coordinator_mode import get_team_registry from openharness.tasks import get_task_manager from openharness.tools.agent_tool import AgentTool, AgentToolInput from openharness.tools.base import ToolExecutionContext from openharness.tools.task_create_tool import TaskCreateTool, TaskCreateToolInput from openharness.tools.task_output_tool import TaskOutputTool, TaskOutputToolInput from openharness.tools.task_update_tool import TaskUpdateTool, TaskUpdateToolInput from openharness.tools.team_create_tool import TeamCreateTool, TeamCreateToolInput async def _wait_for_terminal_task(task_id: str, *, timeout_seconds: float = 2.0) -> None: deadline = asyncio.get_running_loop().time() + timeout_seconds manager = get_task_manager() while asyncio.get_running_loop().time() < deadline: task = manager.get_task(task_id) if task is not None and task.status in {"completed", "failed", "killed"}: return await asyncio.sleep(0.05) raise AssertionError(f"Task {task_id} did not reach a terminal status in time") @pytest.mark.asyncio async def test_task_create_and_output_tool(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) context = ToolExecutionContext(cwd=tmp_path) create_result = await TaskCreateTool().execute( TaskCreateToolInput( type="local_bash", description="echo", command="printf 'tool task'", ), context, ) assert create_result.is_error is False task_id = create_result.output.split()[2] manager = get_task_manager() for _ in range(20): if "tool task" in manager.read_task_output(task_id): break await asyncio.sleep(0.1) output_result = await TaskOutputTool().execute( TaskOutputToolInput(task_id=task_id), context, ) assert "tool task" in output_result.output @pytest.mark.asyncio async def test_team_create_tool(tmp_path: Path): result = await TeamCreateTool().execute( TeamCreateToolInput(name="demo", description="test"), ToolExecutionContext(cwd=tmp_path), ) assert result.is_error is False assert "Created team demo" == result.output @pytest.mark.asyncio async def test_task_update_tool_updates_metadata(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) context = ToolExecutionContext(cwd=tmp_path) create_result = await TaskCreateTool().execute( TaskCreateToolInput( type="local_bash", description="updatable", command="printf 'tool task'", ), context, ) task_id = create_result.output.split()[2] update_result = await TaskUpdateTool().execute( TaskUpdateToolInput( task_id=task_id, progress=60, status_note="waiting on verification", description="renamed task", ), context, ) assert update_result.is_error is False task = get_task_manager().get_task(task_id) assert task is not None assert task.description == "renamed task" assert task.metadata["progress"] == "60" assert task.metadata["status_note"] == "waiting on verification" @pytest.mark.asyncio async def test_agent_tool_uses_subprocess_backend_and_task_is_pollable( tmp_path: Path, monkeypatch ): """Regression test for #59 / PR #60. AgentTool must use the subprocess backend so the returned task_id is registered in BackgroundTaskManager and is queryable by the task tools. Before the fix, AgentTool hardcoded in_process first. On macOS/Linux that backend is always registered (supports_swarm_mailbox=True), so spawn() returned IDs like "in_process_3f7a9b1c2d4e" that BackgroundTaskManager never saw — every poll attempt raised ValueError. """ monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) context = ToolExecutionContext(cwd=tmp_path) result = await AgentTool().execute( AgentToolInput( description="backend regression check", prompt="hello", subagent_type="test-worker", # command echoes one line and exits — minimal subprocess command='python -u -c "import sys; print(sys.stdin.readline().strip())"', ), context, ) assert not result.is_error, f"AgentTool failed: {result.output}" # 1. Backend reported in output must be subprocess, not in_process. assert "backend=subprocess" in result.output, ( f"Expected backend=subprocess in output, got: {result.output}" ) # 2. task_id must NOT be an in-process ID. assert "in_process_" not in result.output, ( f"task_id must not be an in-process ID, got: {result.output}" ) # 3. The task_id must be registered in BackgroundTaskManager so task tools # can query it without raising ValueError. # Parse task_id from "Spawned agent X (task_id=Y, backend=Z)" import re m = re.search(r"task_id=(\S+?)[,)]", result.output) assert m, f"Could not parse task_id from output: {result.output}" task_id = m.group(1) manager = get_task_manager() record = manager.get_task(task_id) assert record is not None, ( f"task_id {task_id!r} not found in BackgroundTaskManager — " "task tools (TaskGet, TaskOutput, etc.) would have failed" ) assert record.command == 'python -u -c "import sys; print(sys.stdin.readline().strip())"' assert record.type == "local_agent" await _wait_for_terminal_task(task_id) @pytest.mark.asyncio async def test_send_message_swarm_path_uses_subprocess_backend( tmp_path: Path, monkeypatch ): """SendMessageTool._send_swarm_message must route via SubprocessBackend. Before the fix, _send_swarm_message also hardcoded in_process, so even the name@team routing path would fail to find agents spawned by AgentTool. """ from unittest.mock import AsyncMock, patch monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) context = ToolExecutionContext(cwd=tmp_path) from openharness.tools.send_message_tool import SendMessageTool with patch( "openharness.swarm.subprocess_backend.SubprocessBackend.send_message", new_callable=AsyncMock, ) as mock_send: await SendMessageTool().execute( __import__( "openharness.tools.send_message_tool", fromlist=["SendMessageToolInput"], ).SendMessageToolInput( task_id="worker@default", message="ping", ), context, ) # send_message may raise ValueError because no agent was spawned yet # (no _agent_tasks entry), but the key assertion is that SubprocessBackend # was called — not InProcessBackend. mock_send.assert_called_once() agent_id_arg = mock_send.call_args[0][0] assert agent_id_arg == "worker@default" @pytest.mark.asyncio async def test_agent_tool_creates_missing_team_when_team_argument_is_provided(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) get_team_registry()._teams.clear() context = ToolExecutionContext(cwd=tmp_path) result = await AgentTool().execute( AgentToolInput( description="team auto-create regression", prompt="ready", subagent_type="test-worker-team", team="design-qa-loop", command="python -u -c \"import sys; print(sys.stdin.readline().strip())\"", ), context, ) assert result.is_error is False teams = {team.name: team for team in get_team_registry().list_teams()} assert "design-qa-loop" in teams assert len(teams["design-qa-loop"].agents) == 1 @pytest.mark.asyncio async def test_agent_tool_supports_remote_and_teammate_modes(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) context = ToolExecutionContext(cwd=tmp_path) for i, mode in enumerate(("remote_agent", "in_process_teammate")): result = await AgentTool().execute( AgentToolInput( description=f"{mode} smoke", prompt="ready", mode=mode, subagent_type=f"test-worker-{i}", command="python -u -c \"import sys; print(sys.stdin.readline().strip())\"", ), context, ) assert result.is_error is False import re match = re.search(r"task_id=(\S+?)[,)]", result.output) assert match, result.output task_id = match.group(1) record = get_task_manager().get_task(task_id) assert record is not None assert record.type == mode await _wait_for_terminal_task(task_id)