test: add comprehensive unit tests for swarm and coordinator components

Adds 156 tests across 9 new test files covering:
- coordinator_mode: TaskNotification XML, is_coordinator_mode, WorkerConfig
- agent_definitions: AgentDefinition model, built-ins, load_agents_dir
- swarm/types: TeammateIdentity, SpawnResult, TeammateExecutor protocol
- swarm/registry: BackendRegistry register/detect/get_executor
- swarm/mailbox: TeammateMailbox write/read/mark_read/clear + factories
- swarm/in_process: InProcessBackend spawn/shutdown/send_message, contextvars
- swarm/permission_sync: create/send/poll/handle permission flow
- swarm/team_lifecycle: TeamLifecycleManager CRUD with tmp_path fixtures
- swarm/worktree: validate_worktree_slug edge cases, flatten/branch helpers

tests/test_coordinator/test_agent_definitions.py

@@ -0,0 +1,184 @@
+"""Tests for AgentDefinition model, built-in defs, and load_agents_dir."""
+
+from __future__ import annotations
+
+
+import pytest
+
+from openharness.coordinator.agent_definitions import (
+    AgentDefinition,
+    _parse_agent_frontmatter,
+    get_builtin_agent_definitions,
+    load_agents_dir,
+)
+
+
+# ---------------------------------------------------------------------------
+# AgentDefinition model
+# ---------------------------------------------------------------------------
+
+
+def test_agent_definition_required_fields():
+    agent = AgentDefinition(
+        name="my-agent",
+        description="does things",
+    )
+    assert agent.name == "my-agent"
+    assert agent.description == "does things"
+    assert agent.tools is None
+    assert agent.model is None
+    assert agent.permissions == []
+    assert agent.subagent_type == "general-purpose"
+    assert agent.source == "builtin"
+
+
+def test_agent_definition_with_tools():
+    agent = AgentDefinition(
+        name="reader",
+        description="reads files",
+        tools=["Read", "Glob", "Grep"],
+        source="user",
+    )
+    assert "Read" in agent.tools
+    assert agent.source == "user"
+
+
+def test_agent_definition_invalid_source():
+    with pytest.raises(Exception):
+        AgentDefinition(name="bad", description="desc", source="unknown")
+
+
+# ---------------------------------------------------------------------------
+# Built-in agent definitions
+# ---------------------------------------------------------------------------
+
+
+def test_get_builtin_returns_expected_names():
+    builtins = get_builtin_agent_definitions()
+    names = {a.name for a in builtins}
+    assert "general-purpose" in names
+    assert "Explore" in names
+    assert "Plan" in names
+    assert "worker" in names
+    assert "verifier" in names
+
+
+def test_builtin_agents_have_descriptions():
+    for agent in get_builtin_agent_definitions():
+        assert agent.description, f"Agent {agent.name!r} is missing a description"
+
+
+def test_builtin_explore_has_tools():
+    builtins = get_builtin_agent_definitions()
+    explore = next(a for a in builtins if a.name == "Explore")
+    assert explore.tools is not None
+    assert "Read" in explore.tools
+
+
+def test_builtin_general_purpose_has_all_tools():
+    builtins = get_builtin_agent_definitions()
+    gp = next(a for a in builtins if a.name == "general-purpose")
+    assert gp.tools is None  # None means all tools
+
+
+# ---------------------------------------------------------------------------
+# _parse_agent_frontmatter
+# ---------------------------------------------------------------------------
+
+
+def test_parse_frontmatter_with_valid_yaml():
+    content = "---\nname: my-agent\ndescription: a test agent\n---\nThis is the body."
+    fm, body = _parse_agent_frontmatter(content)
+    assert fm["name"] == "my-agent"
+    assert fm["description"] == "a test agent"
+    assert body == "This is the body."
+
+
+def test_parse_frontmatter_missing_delimiter_returns_empty():
+    content = "name: my-agent\ndescription: desc\nbody text"
+    fm, body = _parse_agent_frontmatter(content)
+    assert fm == {}
+    assert body == content
+
+
+def test_parse_frontmatter_unclosed_returns_empty():
+    content = "---\nname: agent\ndescription: desc\nbody"
+    fm, body = _parse_agent_frontmatter(content)
+    assert fm == {}
+
+
+def test_parse_frontmatter_strips_quotes():
+    content = "---\nname: 'quoted-name'\ndescription: \"also quoted\"\n---\nbody"
+    fm, _ = _parse_agent_frontmatter(content)
+    assert fm["name"] == "quoted-name"
+    assert fm["description"] == "also quoted"
+
+
+# ---------------------------------------------------------------------------
+# load_agents_dir
+# ---------------------------------------------------------------------------
+
+
+def test_load_agents_dir_empty_dir(tmp_path):
+    agents = load_agents_dir(tmp_path)
+    assert agents == []
+
+
+def test_load_agents_dir_nonexistent(tmp_path):
+    agents = load_agents_dir(tmp_path / "no_such_dir")
+    assert agents == []
+
+
+def test_load_agents_dir_single_file(tmp_path):
+    md = tmp_path / "my_agent.md"
+    md.write_text(
+        "---\nname: my-agent\ndescription: test agent\n---\nDo something useful.",
+        encoding="utf-8",
+    )
+    agents = load_agents_dir(tmp_path)
+    assert len(agents) == 1
+    assert agents[0].name == "my-agent"
+    assert agents[0].description == "test agent"
+    assert agents[0].system_prompt == "Do something useful."
+    assert agents[0].source == "user"
+
+
+def test_load_agents_dir_file_with_tools(tmp_path):
+    md = tmp_path / "explorer.md"
+    md.write_text(
+        "---\nname: explorer\ndescription: explores code\ntools: Read, Glob, Grep\n---\nExplore.",
+        encoding="utf-8",
+    )
+    agents = load_agents_dir(tmp_path)
+    assert agents[0].tools == ["Read", "Glob", "Grep"]
+
+
+def test_load_agents_dir_falls_back_to_stem_for_name(tmp_path):
+    md = tmp_path / "fallback_name.md"
+    md.write_text("---\ndescription: no name given\n---\nbody", encoding="utf-8")
+    agents = load_agents_dir(tmp_path)
+    assert agents[0].name == "fallback_name"
+
+
+def test_load_agents_dir_with_model_and_permissions(tmp_path):
+    md = tmp_path / "specialized.md"
+    md.write_text(
+        "---\nname: spec\ndescription: specialized\nmodel: claude-opus-4-6\n"
+        "permissions: allow:bash, deny:write\n---\nbody",
+        encoding="utf-8",
+    )
+    agents = load_agents_dir(tmp_path)
+    assert agents[0].model == "claude-opus-4-6"
+    assert "allow:bash" in agents[0].permissions
+    assert "deny:write" in agents[0].permissions
+
+
+def test_load_agents_dir_skips_unreadable_files(tmp_path):
+    good = tmp_path / "good.md"
+    good.write_text("---\nname: good\ndescription: fine\n---\nbody", encoding="utf-8")
+    bad = tmp_path / "bad.md"
+    bad.write_bytes(b"\xff\xfe invalid utf-32")  # not utf-8, but won't crash
+    # Should still load the good file
+    agents = load_agents_dir(tmp_path)
+    names = [a.name for a in agents]
+    assert "good" in names

tests/test_coordinator/test_coordinator_mode.py

@@ -0,0 +1,199 @@
+"""Tests for CoordinatorMode, TaskNotification XML, and WorkerConfig."""
+
+from __future__ import annotations
+
+import pytest
+
+from openharness.coordinator.coordinator_mode import (
+    TaskNotification,
+    WorkerConfig,
+    format_task_notification,
+    get_coordinator_tools,
+    get_coordinator_user_context,
+    is_coordinator_mode,
+    match_session_mode,
+    parse_task_notification,
+)
+
+
+# ---------------------------------------------------------------------------
+# TaskNotification XML round-trip
+# ---------------------------------------------------------------------------
+
+
+def test_format_and_parse_basic():
+    n = TaskNotification(task_id="t123", status="completed", summary="all done")
+    xml = format_task_notification(n)
+    assert "<task-notification>" in xml
+    assert "<task-id>t123</task-id>" in xml
+    assert "<status>completed</status>" in xml
+    assert "<summary>all done</summary>" in xml
+
+    parsed = parse_task_notification(xml)
+    assert parsed.task_id == "t123"
+    assert parsed.status == "completed"
+    assert parsed.summary == "all done"
+    assert parsed.result is None
+    assert parsed.usage is None
+
+
+def test_format_and_parse_with_result_and_usage():
+    n = TaskNotification(
+        task_id="abc",
+        status="failed",
+        summary="error occurred",
+        result="traceback here",
+        usage={"total_tokens": 42, "tool_uses": 3, "duration_ms": 1500},
+    )
+    xml = format_task_notification(n)
+    assert "<result>traceback here</result>" in xml
+    assert "<total_tokens>42</total_tokens>" in xml
+    assert "<tool_uses>3</tool_uses>" in xml
+    assert "<duration_ms>1500</duration_ms>" in xml
+
+    parsed = parse_task_notification(xml)
+    assert parsed.task_id == "abc"
+    assert parsed.status == "failed"
+    assert parsed.result == "traceback here"
+    assert parsed.usage == {"total_tokens": 42, "tool_uses": 3, "duration_ms": 1500}
+
+
+def test_parse_ignores_missing_optional_fields():
+    xml = "<task-notification><task-id>x</task-id><status>completed</status><summary>ok</summary></task-notification>"
+    parsed = parse_task_notification(xml)
+    assert parsed.task_id == "x"
+    assert parsed.result is None
+    assert parsed.usage is None
+
+
+def test_parse_partial_usage_block():
+    xml = (
+        "<task-notification>"
+        "<task-id>y</task-id><status>completed</status><summary>ok</summary>"
+        "<usage><total_tokens>100</total_tokens></usage>"
+        "</task-notification>"
+    )
+    parsed = parse_task_notification(xml)
+    assert parsed.usage == {"total_tokens": 100}
+
+
+# ---------------------------------------------------------------------------
+# is_coordinator_mode
+# ---------------------------------------------------------------------------
+
+
+def test_is_coordinator_mode_false_by_default(monkeypatch):
+    monkeypatch.delenv("CLAUDE_CODE_COORDINATOR_MODE", raising=False)
+    assert is_coordinator_mode() is False
+
+
+@pytest.mark.parametrize("value", ["1", "true", "True", "yes", "YES"])
+def test_is_coordinator_mode_true_variants(monkeypatch, value):
+    monkeypatch.setenv("CLAUDE_CODE_COORDINATOR_MODE", value)
+    assert is_coordinator_mode() is True
+
+
+def test_is_coordinator_mode_false_for_garbage(monkeypatch):
+    monkeypatch.setenv("CLAUDE_CODE_COORDINATOR_MODE", "maybe")
+    assert is_coordinator_mode() is False
+
+
+# ---------------------------------------------------------------------------
+# get_coordinator_tools
+# ---------------------------------------------------------------------------
+
+
+def test_get_coordinator_tools_returns_expected():
+    tools = get_coordinator_tools()
+    assert "agent" in tools
+    assert "send_message" in tools
+    assert "task_stop" in tools
+    assert len(tools) == 3
+
+
+# ---------------------------------------------------------------------------
+# match_session_mode
+# ---------------------------------------------------------------------------
+
+
+def test_match_session_mode_no_change_when_already_coordinator(monkeypatch):
+    monkeypatch.setenv("CLAUDE_CODE_COORDINATOR_MODE", "1")
+    result = match_session_mode("coordinator")
+    assert result is None
+    assert is_coordinator_mode() is True
+
+
+def test_match_session_mode_switches_to_coordinator(monkeypatch):
+    monkeypatch.delenv("CLAUDE_CODE_COORDINATOR_MODE", raising=False)
+    result = match_session_mode("coordinator")
+    assert result is not None
+    assert "coordinator" in result.lower()
+    assert is_coordinator_mode() is True
+
+
+def test_match_session_mode_exits_coordinator(monkeypatch):
+    monkeypatch.setenv("CLAUDE_CODE_COORDINATOR_MODE", "1")
+    result = match_session_mode("worker")
+    assert result is not None
+    assert is_coordinator_mode() is False
+
+
+def test_match_session_mode_none_returns_none(monkeypatch):
+    result = match_session_mode(None)
+    assert result is None
+
+
+# ---------------------------------------------------------------------------
+# get_coordinator_user_context
+# ---------------------------------------------------------------------------
+
+
+def test_coordinator_user_context_empty_when_not_coordinator(monkeypatch):
+    monkeypatch.delenv("CLAUDE_CODE_COORDINATOR_MODE", raising=False)
+    ctx = get_coordinator_user_context()
+    assert ctx == {}
+
+
+def test_coordinator_user_context_includes_tools(monkeypatch):
+    monkeypatch.setenv("CLAUDE_CODE_COORDINATOR_MODE", "1")
+    monkeypatch.delenv("CLAUDE_CODE_SIMPLE", raising=False)
+    ctx = get_coordinator_user_context()
+    assert "workerToolsContext" in ctx
+    assert "bash" in ctx["workerToolsContext"]
+
+
+def test_coordinator_user_context_with_mcp_clients(monkeypatch):
+    monkeypatch.setenv("CLAUDE_CODE_COORDINATOR_MODE", "1")
+    ctx = get_coordinator_user_context(mcp_clients=[{"name": "my-server"}])
+    assert "my-server" in ctx["workerToolsContext"]
+
+
+def test_coordinator_user_context_with_scratchpad(monkeypatch):
+    monkeypatch.setenv("CLAUDE_CODE_COORDINATOR_MODE", "1")
+    ctx = get_coordinator_user_context(scratchpad_dir="/tmp/scratch")
+    assert "/tmp/scratch" in ctx["workerToolsContext"]
+
+
+# ---------------------------------------------------------------------------
+# WorkerConfig dataclass
+# ---------------------------------------------------------------------------
+
+
+def test_worker_config_defaults():
+    cfg = WorkerConfig(agent_id="w1", name="coder", prompt="do stuff")
+    assert cfg.model is None
+    assert cfg.color is None
+    assert cfg.team is None
+
+
+def test_worker_config_full():
+    cfg = WorkerConfig(
+        agent_id="w2",
+        name="tester",
+        prompt="run tests",
+        model="claude-opus-4-6",
+        color="blue",
+        team="alpha",
+    )
+    assert cfg.model == "claude-opus-4-6"
+    assert cfg.team == "alpha"

tests/test_swarm/test_in_process.py

@@ -0,0 +1,182 @@
+"""Tests for InProcessBackend: spawn, shutdown, send_message, and contextvars."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from openharness.swarm.in_process import (
+    InProcessBackend,
+    TeammateContext,
+    get_teammate_context,
+    set_teammate_context,
+)
+from openharness.swarm.types import TeammateMessage, TeammateSpawnConfig
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def spawn_config():
+    return TeammateSpawnConfig(
+        name="worker",
+        team="test-team",
+        prompt="hello",
+        cwd="/tmp",
+        parent_session_id="sess-001",
+    )
+
+
+@pytest.fixture
+def backend(tmp_path, monkeypatch):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    return InProcessBackend()
+
+
+# ---------------------------------------------------------------------------
+# TeammateContext
+# ---------------------------------------------------------------------------
+
+
+def test_teammate_context_defaults():
+    ctx = TeammateContext(
+        agent_id="w@t",
+        agent_name="w",
+        team_name="t",
+    )
+    assert ctx.color is None
+    assert ctx.plan_mode_required is False
+    assert not ctx.cancel_event.is_set()
+
+
+# ---------------------------------------------------------------------------
+# ContextVar get / set
+# ---------------------------------------------------------------------------
+
+
+def test_get_teammate_context_returns_none_outside_task():
+    # Outside any async task, the contextvar should be None
+    result = get_teammate_context()
+    assert result is None
+
+
+async def test_set_and_get_teammate_context():
+    ctx = TeammateContext(agent_id="x@y", agent_name="x", team_name="y")
+    set_teammate_context(ctx)
+    assert get_teammate_context() is ctx
+
+
+# ---------------------------------------------------------------------------
+# InProcessBackend.spawn
+# ---------------------------------------------------------------------------
+
+
+async def test_spawn_returns_success_result(backend, spawn_config):
+    result = await backend.spawn(spawn_config)
+    assert result.success is True
+    assert result.agent_id == "worker@test-team"
+    assert result.backend_type == "in_process"
+    assert result.task_id.startswith("in_process_")
+
+
+async def test_spawn_duplicate_returns_failure(backend, spawn_config):
+    await backend.spawn(spawn_config)
+    # Spawn again while first is still running
+    result = await backend.spawn(spawn_config)
+    assert result.success is False
+    assert result.error is not None
+
+
+async def test_spawn_creates_active_agent(backend, spawn_config):
+    await backend.spawn(spawn_config)
+    assert backend.is_active("worker@test-team")
+
+
+# ---------------------------------------------------------------------------
+# InProcessBackend.shutdown
+# ---------------------------------------------------------------------------
+
+
+async def test_shutdown_unknown_agent_returns_false(backend):
+    result = await backend.shutdown("nonexistent@team")
+    assert result is False
+
+
+async def test_graceful_shutdown(backend, spawn_config):
+    await backend.spawn(spawn_config)
+    assert backend.is_active("worker@test-team")
+
+    result = await backend.shutdown("worker@test-team", timeout=2.0)
+    assert result is True
+    assert not backend.is_active("worker@test-team")
+
+
+async def test_force_shutdown(backend, spawn_config):
+    await backend.spawn(spawn_config)
+    result = await backend.shutdown("worker@test-team", force=True, timeout=2.0)
+    assert result is True
+
+
+# ---------------------------------------------------------------------------
+# InProcessBackend.send_message
+# ---------------------------------------------------------------------------
+
+
+async def test_send_message_writes_to_mailbox(backend, tmp_path, monkeypatch):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    config = TeammateSpawnConfig(
+        name="rcvr",
+        team="myteam",
+        prompt="wait",
+        cwd="/tmp",
+        parent_session_id="s",
+    )
+    await backend.spawn(config)
+
+    msg = TeammateMessage(text="work on it", from_agent="leader")
+    # Should not raise
+    await backend.send_message("rcvr@myteam", msg)
+
+    # Verify the message was written to mailbox
+    from openharness.swarm.mailbox import TeammateMailbox
+    mailbox = TeammateMailbox(team_name="myteam", agent_id="rcvr")
+    messages = await mailbox.read_all(unread_only=False)
+    assert any(m.payload.get("content") == "work on it" for m in messages)
+
+    await backend.shutdown("rcvr@myteam", force=True)
+
+
+async def test_send_message_invalid_agent_id_raises(backend):
+    with pytest.raises(ValueError, match="agentName@teamName"):
+        await backend.send_message("no-at-sign", TeammateMessage(text="hi", from_agent="l"))
+
+
+# ---------------------------------------------------------------------------
+# active_agents / shutdown_all
+# ---------------------------------------------------------------------------
+
+
+async def test_active_agents_lists_running(backend, spawn_config):
+    await backend.spawn(spawn_config)
+    active = backend.active_agents()
+    assert "worker@test-team" in active
+
+
+async def test_shutdown_all(backend, tmp_path, monkeypatch):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    for name in ("a", "b"):
+        cfg = TeammateSpawnConfig(
+            name=name,
+            team="t",
+            prompt="run",
+            cwd="/tmp",
+            parent_session_id="s",
+        )
+        await backend.spawn(cfg)
+
+    await backend.shutdown_all(force=True, timeout=2.0)
+    assert backend.active_agents() == []

tests/test_swarm/test_mailbox.py

@@ -0,0 +1,196 @@
+"""Tests for TeammateMailbox: write/read/mark_read/clear and factory helpers."""
+
+from __future__ import annotations
+
+import time
+from pathlib import Path
+
+import pytest
+
+from openharness.swarm.mailbox import (
+    MailboxMessage,
+    TeammateMailbox,
+    create_idle_notification,
+    create_shutdown_request,
+    create_user_message,
+)
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def mailbox(tmp_path, monkeypatch):
+    """Return a TeammateMailbox whose team directory is inside tmp_path."""
+    # Redirect the home-dir lookup so mailbox writes to tmp_path
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    return TeammateMailbox(team_name="test-team", agent_id="worker1")
+
+
+def _make_msg(sender="leader", recipient="worker1") -> MailboxMessage:
+    return MailboxMessage(
+        id="msg-001",
+        type="user_message",
+        sender=sender,
+        recipient=recipient,
+        payload={"content": "hello"},
+        timestamp=time.time(),
+    )
+
+
+# ---------------------------------------------------------------------------
+# MailboxMessage serialisation
+# ---------------------------------------------------------------------------
+
+
+def test_mailbox_message_round_trip():
+    msg = _make_msg()
+    d = msg.to_dict()
+    msg2 = MailboxMessage.from_dict(d)
+    assert msg2.id == msg.id
+    assert msg2.type == msg.type
+    assert msg2.sender == msg.sender
+    assert msg2.payload == msg.payload
+    assert msg2.read is False
+
+
+def test_mailbox_message_from_dict_defaults_read_false():
+    data = {
+        "id": "x",
+        "type": "user_message",
+        "sender": "a",
+        "recipient": "b",
+        "payload": {},
+        "timestamp": 1234.0,
+    }
+    msg = MailboxMessage.from_dict(data)
+    assert msg.read is False
+
+
+# ---------------------------------------------------------------------------
+# TeammateMailbox write / read_all
+# ---------------------------------------------------------------------------
+
+
+async def test_write_and_read_all(mailbox):
+    msg = _make_msg()
+    await mailbox.write(msg)
+    messages = await mailbox.read_all(unread_only=False)
+    assert len(messages) == 1
+    assert messages[0].id == "msg-001"
+
+
+async def test_read_all_unread_only_filters(mailbox):
+    msg = _make_msg()
+    await mailbox.write(msg)
+
+    # Mark it read directly by re-writing with read=True
+    inbox = mailbox.get_mailbox_dir()
+    for path in inbox.glob("*.json"):
+        import json as _json
+        data = _json.loads(path.read_text())
+        data["read"] = True
+        path.write_text(_json.dumps(data))
+
+    unread = await mailbox.read_all(unread_only=True)
+    assert unread == []
+
+    all_msgs = await mailbox.read_all(unread_only=False)
+    assert len(all_msgs) == 1
+
+
+async def test_write_multiple_messages_sorted_by_timestamp(mailbox):
+    for i in range(3):
+        msg = MailboxMessage(
+            id=f"msg-{i}",
+            type="user_message",
+            sender="leader",
+            recipient="worker1",
+            payload={"seq": i},
+            timestamp=1000.0 + i,
+        )
+        await mailbox.write(msg)
+
+    messages = await mailbox.read_all(unread_only=False)
+    timestamps = [m.timestamp for m in messages]
+    assert timestamps == sorted(timestamps)
+
+
+# ---------------------------------------------------------------------------
+# mark_read
+# ---------------------------------------------------------------------------
+
+
+async def test_mark_read_updates_flag(mailbox):
+    msg = _make_msg()
+    await mailbox.write(msg)
+
+    await mailbox.mark_read(msg.id)
+    all_msgs = await mailbox.read_all(unread_only=False)
+    assert all_msgs[0].read is True
+
+
+async def test_mark_read_nonexistent_id_is_noop(mailbox):
+    msg = _make_msg()
+    await mailbox.write(msg)
+    # Should not raise
+    await mailbox.mark_read("does-not-exist")
+    # Original message still unread
+    messages = await mailbox.read_all(unread_only=True)
+    assert len(messages) == 1
+
+
+# ---------------------------------------------------------------------------
+# clear
+# ---------------------------------------------------------------------------
+
+
+async def test_clear_removes_all_messages(mailbox):
+    for i in range(3):
+        msg = MailboxMessage(
+            id=f"c-{i}",
+            type="user_message",
+            sender="l",
+            recipient="w",
+            payload={},
+            timestamp=float(i),
+        )
+        await mailbox.write(msg)
+
+    await mailbox.clear()
+    messages = await mailbox.read_all(unread_only=False)
+    assert messages == []
+
+
+async def test_clear_on_empty_mailbox_is_noop(mailbox):
+    await mailbox.clear()  # should not raise
+    messages = await mailbox.read_all(unread_only=False)
+    assert messages == []
+
+
+# ---------------------------------------------------------------------------
+# Factory helpers
+# ---------------------------------------------------------------------------
+
+
+def test_create_user_message():
+    msg = create_user_message("leader", "worker1", "do stuff")
+    assert msg.type == "user_message"
+    assert msg.sender == "leader"
+    assert msg.recipient == "worker1"
+    assert msg.payload["content"] == "do stuff"
+    assert msg.id  # has a UUID
+
+
+def test_create_shutdown_request():
+    msg = create_shutdown_request("leader", "worker1")
+    assert msg.type == "shutdown"
+    assert msg.payload == {}
+
+
+def test_create_idle_notification():
+    msg = create_idle_notification("worker1", "leader", "finished task")
+    assert msg.type == "idle_notification"
+    assert msg.payload["summary"] == "finished task"

tests/test_swarm/test_permission_sync.py

@@ -0,0 +1,182 @@
+"""Tests for swarm permission sync protocol: create/send/poll/handle."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from openharness.swarm.permission_sync import (
+    SwarmPermissionResponse,
+    _is_read_only,
+    create_permission_request,
+    handle_permission_request,
+    poll_permission_response,
+    send_permission_request,
+    send_permission_response,
+)
+
+
+# ---------------------------------------------------------------------------
+# _is_read_only heuristic
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "tool_name",
+    ["Read", "Glob", "Grep", "WebFetch", "WebSearch", "TaskGet", "TaskList", "CronList"],
+)
+def test_is_read_only_true_for_safe_tools(tool_name):
+    assert _is_read_only(tool_name) is True
+
+
+@pytest.mark.parametrize("tool_name", ["Bash", "Edit", "Write", "TaskCreate"])
+def test_is_read_only_false_for_write_tools(tool_name):
+    assert _is_read_only(tool_name) is False
+
+
+# ---------------------------------------------------------------------------
+# create_permission_request
+# ---------------------------------------------------------------------------
+
+
+def test_create_permission_request_has_unique_id():
+    r1 = create_permission_request("Bash", "tu-1", {"command": "ls"})
+    r2 = create_permission_request("Bash", "tu-2", {"command": "ls"})
+    assert r1.id != r2.id
+
+
+def test_create_permission_request_fields():
+    req = create_permission_request(
+        "Edit",
+        "tu-xyz",
+        {"file_path": "/tmp/f.py"},
+        description="edit a file",
+        permission_suggestions=[{"type": "allow"}],
+    )
+    assert req.tool_name == "Edit"
+    assert req.tool_use_id == "tu-xyz"
+    assert req.description == "edit a file"
+    assert req.permission_suggestions == [{"type": "allow"}]
+
+
+def test_create_permission_request_default_suggestions():
+    req = create_permission_request("Bash", "tu-1", {})
+    assert req.permission_suggestions == []
+
+
+# ---------------------------------------------------------------------------
+# SwarmPermissionResponse
+# ---------------------------------------------------------------------------
+
+
+def test_swarm_permission_response_defaults():
+    resp = SwarmPermissionResponse(request_id="r1", allowed=True)
+    assert resp.feedback is None
+    assert resp.updated_rules == []
+
+
+# ---------------------------------------------------------------------------
+# send_permission_request writes to leader mailbox
+# ---------------------------------------------------------------------------
+
+
+async def test_send_permission_request_writes_to_leader(tmp_path, monkeypatch):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    req = create_permission_request("Bash", "tu-1", {"command": "echo hi"})
+    await send_permission_request(req, "myteam", "worker1", "leader")
+
+    from openharness.swarm.mailbox import TeammateMailbox
+    mailbox = TeammateMailbox("myteam", "leader")
+    messages = await mailbox.read_all(unread_only=False)
+    assert len(messages) == 1
+    assert messages[0].type == "permission_request"
+    assert messages[0].payload["tool_name"] == "Bash"
+    assert messages[0].payload["worker_id"] == "worker1"
+
+
+# ---------------------------------------------------------------------------
+# send_permission_response writes to worker mailbox
+# ---------------------------------------------------------------------------
+
+
+async def test_send_permission_response_writes_to_worker(tmp_path, monkeypatch):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    resp = SwarmPermissionResponse(request_id="r1", allowed=True, feedback=None)
+    await send_permission_response(resp, "myteam", "worker1", "leader")
+
+    from openharness.swarm.mailbox import TeammateMailbox
+    mailbox = TeammateMailbox("myteam", "worker1")
+    messages = await mailbox.read_all(unread_only=False)
+    assert len(messages) == 1
+    assert messages[0].type == "permission_response"
+    assert messages[0].payload["allowed"] is True
+
+
+# ---------------------------------------------------------------------------
+# handle_permission_request
+# ---------------------------------------------------------------------------
+
+
+async def test_handle_read_only_tool_auto_approved():
+    req = create_permission_request("Read", "tu-1", {"file_path": "/tmp/f.py"})
+    checker = MagicMock()
+    resp = await handle_permission_request(req, checker)
+    assert resp.allowed is True
+    checker.evaluate.assert_not_called()
+
+
+async def test_handle_write_tool_delegates_to_checker():
+    req = create_permission_request("Bash", "tu-2", {"command": "rm -rf /"})
+
+    decision = MagicMock()
+    decision.allowed = False
+    decision.reason = "dangerous command"
+    checker = MagicMock()
+    checker.evaluate.return_value = decision
+
+    resp = await handle_permission_request(req, checker)
+    assert resp.allowed is False
+    assert resp.feedback == "dangerous command"
+    checker.evaluate.assert_called_once_with(
+        "Bash", is_read_only=False, file_path=None, command="rm -rf /"
+    )
+
+
+async def test_handle_write_tool_allowed_by_checker():
+    req = create_permission_request("Edit", "tu-3", {"file_path": "/src/main.py"})
+
+    decision = MagicMock()
+    decision.allowed = True
+    decision.reason = None
+    checker = MagicMock()
+    checker.evaluate.return_value = decision
+
+    resp = await handle_permission_request(req, checker)
+    assert resp.allowed is True
+    assert resp.feedback is None
+
+
+# ---------------------------------------------------------------------------
+# poll_permission_response timeout
+# ---------------------------------------------------------------------------
+
+
+async def test_poll_permission_response_times_out(tmp_path, monkeypatch):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    result = await poll_permission_response("myteam", "worker1", "nonexistent-id", timeout=0.1)
+    assert result is None
+
+
+async def test_poll_permission_response_finds_matching_message(tmp_path, monkeypatch):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+
+    # Pre-write a response to the worker mailbox
+    resp = SwarmPermissionResponse(request_id="req-abc", allowed=True)
+    await send_permission_response(resp, "myteam", "worker1", "leader")
+
+    result = await poll_permission_response("myteam", "worker1", "req-abc", timeout=2.0)
+    assert result is not None
+    assert result.allowed is True
+    assert result.request_id == "req-abc"

tests/test_swarm/test_registry.py

@@ -0,0 +1,114 @@
+"""Tests for BackendRegistry: register, detect, and get_executor."""
+
+from __future__ import annotations
+
+import pytest
+
+from openharness.swarm.registry import BackendRegistry
+from openharness.swarm.types import TeammateExecutor
+
+
+# ---------------------------------------------------------------------------
+# Default registration
+# ---------------------------------------------------------------------------
+
+
+def test_registry_registers_subprocess_and_in_process():
+    registry = BackendRegistry()
+    available = registry.available_backends()
+    assert "subprocess" in available
+    assert "in_process" in available
+
+
+def test_get_executor_subprocess():
+    registry = BackendRegistry()
+    executor = registry.get_executor("subprocess")
+    assert executor is not None
+    assert executor.type == "subprocess"
+
+
+def test_get_executor_in_process():
+    registry = BackendRegistry()
+    executor = registry.get_executor("in_process")
+    assert executor.type == "in_process"
+
+
+def test_get_executor_unknown_raises():
+    registry = BackendRegistry()
+    with pytest.raises(KeyError, match="tmux"):
+        registry.get_executor("tmux")
+
+
+# ---------------------------------------------------------------------------
+# detect_backend
+# ---------------------------------------------------------------------------
+
+
+def test_detect_backend_returns_subprocess_when_not_in_tmux(monkeypatch):
+    monkeypatch.delenv("TMUX", raising=False)
+    registry = BackendRegistry()
+    detected = registry.detect_backend()
+    assert detected == "subprocess"
+
+
+def test_detect_backend_is_cached(monkeypatch):
+    monkeypatch.delenv("TMUX", raising=False)
+    registry = BackendRegistry()
+    first = registry.detect_backend()
+    second = registry.detect_backend()
+    assert first == second
+
+
+def test_detect_backend_reset_clears_cache(monkeypatch):
+    monkeypatch.delenv("TMUX", raising=False)
+    registry = BackendRegistry()
+    _ = registry.detect_backend()
+    assert registry._detected == "subprocess"
+    registry.reset()
+    assert registry._detected is None
+
+
+# ---------------------------------------------------------------------------
+# register_backend custom
+# ---------------------------------------------------------------------------
+
+
+def test_register_custom_backend():
+    class FakeExecutor:
+        type = "in_process"
+
+        def is_available(self):
+            return True
+
+        async def spawn(self, config):
+            ...
+
+        async def send_message(self, agent_id, message):
+            ...
+
+        async def shutdown(self, agent_id, *, force=False):
+            ...
+
+    registry = BackendRegistry()
+    fake = FakeExecutor()
+    registry.register_backend(fake)
+    assert registry.get_executor("in_process") is fake
+
+
+def test_get_executor_auto_detect_returns_executor(monkeypatch):
+    monkeypatch.delenv("TMUX", raising=False)
+    registry = BackendRegistry()
+    executor = registry.get_executor()  # auto-detect
+    assert executor is not None
+    assert isinstance(executor, TeammateExecutor)
+
+
+# ---------------------------------------------------------------------------
+# available_backends
+# ---------------------------------------------------------------------------
+
+
+def test_available_backends_sorted():
+    registry = BackendRegistry()
+    available = registry.available_backends()
+    assert available == sorted(available)

tests/test_swarm/test_team_lifecycle.py

@@ -0,0 +1,197 @@
+"""Tests for TeamLifecycleManager CRUD operations with tmp_path fixtures."""
+
+from __future__ import annotations
+
+import time
+from pathlib import Path
+
+import pytest
+
+from openharness.swarm.team_lifecycle import (
+    TeamFile,
+    TeamLifecycleManager,
+    TeamMember,
+)
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def manager(tmp_path, monkeypatch):
+    """Return a TeamLifecycleManager whose teams live inside tmp_path."""
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    return TeamLifecycleManager()
+
+
+def _make_member(agent_id: str = "worker1@alpha", name: str = "worker1") -> TeamMember:
+    return TeamMember(
+        agent_id=agent_id,
+        name=name,
+        backend_type="subprocess",
+        joined_at=time.time(),
+    )
+
+
+# ---------------------------------------------------------------------------
+# TeamMember serialization
+# ---------------------------------------------------------------------------
+
+
+def test_team_member_round_trip():
+    member = _make_member()
+    data = member.to_dict()
+    restored = TeamMember.from_dict(data)
+    assert restored.agent_id == member.agent_id
+    assert restored.name == member.name
+    assert restored.backend_type == member.backend_type
+    assert restored.status == "active"
+
+
+def test_team_member_default_status():
+    member = _make_member()
+    assert member.status == "active"
+
+
+# ---------------------------------------------------------------------------
+# TeamFile serialization
+# ---------------------------------------------------------------------------
+
+
+def test_team_file_round_trip(tmp_path):
+    tf = TeamFile(name="myteam", created_at=time.time(), description="test team")
+    path = tmp_path / "team.json"
+    tf.save(path)
+    loaded = TeamFile.load(path)
+    assert loaded.name == "myteam"
+    assert loaded.description == "test team"
+
+
+def test_team_file_load_missing_raises(tmp_path):
+    with pytest.raises(FileNotFoundError):
+        TeamFile.load(tmp_path / "nonexistent.json")
+
+
+# ---------------------------------------------------------------------------
+# TeamLifecycleManager.create_team
+# ---------------------------------------------------------------------------
+
+
+def test_create_team_persists_to_disk(manager):
+    tf = manager.create_team("alpha", "first team")
+    assert tf.name == "alpha"
+    assert tf.description == "first team"
+
+    # Verify it was written to disk
+    reloaded = manager.get_team("alpha")
+    assert reloaded is not None
+    assert reloaded.name == "alpha"
+
+
+def test_create_team_duplicate_raises(manager):
+    manager.create_team("beta")
+    with pytest.raises(ValueError, match="already exists"):
+        manager.create_team("beta")
+
+
+# ---------------------------------------------------------------------------
+# TeamLifecycleManager.get_team
+# ---------------------------------------------------------------------------
+
+
+def test_get_team_returns_none_for_missing(manager):
+    result = manager.get_team("no-such-team")
+    assert result is None
+
+
+def test_get_team_returns_team_file(manager):
+    manager.create_team("gamma")
+    tf = manager.get_team("gamma")
+    assert tf is not None
+    assert tf.name == "gamma"
+
+
+# ---------------------------------------------------------------------------
+# TeamLifecycleManager.delete_team
+# ---------------------------------------------------------------------------
+
+
+def test_delete_team_removes_from_disk(manager):
+    manager.create_team("to-delete")
+    manager.delete_team("to-delete")
+    assert manager.get_team("to-delete") is None
+
+
+def test_delete_nonexistent_team_raises(manager):
+    with pytest.raises(ValueError, match="does not exist"):
+        manager.delete_team("ghost")
+
+
+# ---------------------------------------------------------------------------
+# TeamLifecycleManager.list_teams
+# ---------------------------------------------------------------------------
+
+
+def test_list_teams_empty_initially(manager):
+    teams = manager.list_teams()
+    assert teams == []
+
+
+def test_list_teams_returns_all_sorted(manager):
+    for name in ("charlie", "alpha", "bravo"):
+        manager.create_team(name)
+    teams = manager.list_teams()
+    names = [t.name for t in teams]
+    assert names == sorted(names)
+    assert set(names) == {"alpha", "bravo", "charlie"}
+
+
+# ---------------------------------------------------------------------------
+# TeamLifecycleManager.add_member / remove_member
+# ---------------------------------------------------------------------------
+
+
+def test_add_member_persists(manager):
+    manager.create_team("delta")
+    member = _make_member()
+    updated = manager.add_member("delta", member)
+    assert member.agent_id in updated.members
+
+    reloaded = manager.get_team("delta")
+    assert reloaded is not None
+    assert member.agent_id in reloaded.members
+
+
+def test_add_member_replaces_existing(manager):
+    manager.create_team("epsilon")
+    m1 = TeamMember(
+        agent_id="w@epsilon", name="old", backend_type="subprocess", joined_at=1.0
+    )
+    manager.add_member("epsilon", m1)
+
+    m2 = TeamMember(
+        agent_id="w@epsilon", name="new", backend_type="in_process", joined_at=2.0
+    )
+    updated = manager.add_member("epsilon", m2)
+    assert updated.members["w@epsilon"].name == "new"
+
+
+def test_remove_member(manager):
+    manager.create_team("zeta")
+    member = _make_member("x@zeta", "x")
+    manager.add_member("zeta", member)
+    updated = manager.remove_member("zeta", "x@zeta")
+    assert "x@zeta" not in updated.members
+
+
+def test_remove_nonexistent_member_raises(manager):
+    manager.create_team("eta")
+    with pytest.raises(ValueError, match="not a member"):
+        manager.remove_member("eta", "ghost@eta")
+
+
+def test_add_member_to_nonexistent_team_raises(manager):
+    with pytest.raises(ValueError, match="does not exist"):
+        manager.add_member("no-team", _make_member())

tests/test_swarm/test_types.py

@@ -0,0 +1,154 @@
+"""Tests for swarm type definitions: TeammateIdentity, SpawnResult, TeammateExecutor."""
+
+from __future__ import annotations
+
+
+from openharness.swarm.types import (
+    SpawnResult,
+    TeammateExecutor,
+    TeammateIdentity,
+    TeammateMessage,
+    TeammateSpawnConfig,
+)
+
+
+# ---------------------------------------------------------------------------
+# TeammateIdentity
+# ---------------------------------------------------------------------------
+
+
+def test_teammate_identity_required_fields():
+    identity = TeammateIdentity(agent_id="coder@alpha", name="coder", team="alpha")
+    assert identity.agent_id == "coder@alpha"
+    assert identity.name == "coder"
+    assert identity.team == "alpha"
+    assert identity.color is None
+    assert identity.parent_session_id is None
+
+
+def test_teammate_identity_with_optional_fields():
+    identity = TeammateIdentity(
+        agent_id="r@t",
+        name="r",
+        team="t",
+        color="blue",
+        parent_session_id="sess-123",
+    )
+    assert identity.color == "blue"
+    assert identity.parent_session_id == "sess-123"
+
+
+# ---------------------------------------------------------------------------
+# SpawnResult
+# ---------------------------------------------------------------------------
+
+
+def test_spawn_result_success_defaults():
+    result = SpawnResult(task_id="t1", agent_id="a@b", backend_type="subprocess")
+    assert result.success is True
+    assert result.error is None
+
+
+def test_spawn_result_failure():
+    result = SpawnResult(
+        task_id="",
+        agent_id="a@b",
+        backend_type="in_process",
+        success=False,
+        error="already running",
+    )
+    assert result.success is False
+    assert result.error == "already running"
+
+
+def test_spawn_result_backend_types():
+    for bt in ("subprocess", "in_process", "tmux"):
+        r = SpawnResult(task_id="x", agent_id="a@b", backend_type=bt)
+        assert r.backend_type == bt
+
+
+# ---------------------------------------------------------------------------
+# TeammateMessage
+# ---------------------------------------------------------------------------
+
+
+def test_teammate_message_required():
+    msg = TeammateMessage(text="hello", from_agent="leader")
+    assert msg.text == "hello"
+    assert msg.from_agent == "leader"
+    assert msg.color is None
+    assert msg.timestamp is None
+    assert msg.summary is None
+
+
+def test_teammate_message_full():
+    msg = TeammateMessage(
+        text="do this",
+        from_agent="boss",
+        color="green",
+        timestamp="2026-01-01T00:00:00",
+        summary="a task",
+    )
+    assert msg.color == "green"
+    assert msg.summary == "a task"
+
+
+# ---------------------------------------------------------------------------
+# TeammateSpawnConfig
+# ---------------------------------------------------------------------------
+
+
+def test_teammate_spawn_config_defaults():
+    cfg = TeammateSpawnConfig(
+        name="worker",
+        team="myteam",
+        prompt="do work",
+        cwd="/tmp",
+        parent_session_id="sess",
+    )
+    assert cfg.model is None
+    assert cfg.system_prompt is None
+    assert cfg.color is None
+    assert cfg.permissions == []
+    assert cfg.plan_mode_required is False
+    assert cfg.allow_permission_prompts is False
+
+
+# ---------------------------------------------------------------------------
+# TeammateExecutor protocol structural check
+# ---------------------------------------------------------------------------
+
+
+def test_teammate_executor_is_protocol():
+    """TeammateExecutor is a runtime_checkable Protocol."""
+
+    class MockExecutor:
+        type = "subprocess"
+
+        def is_available(self) -> bool:
+            return True
+
+        async def spawn(self, config):
+            ...
+
+        async def send_message(self, agent_id, message):
+            ...
+
+        async def shutdown(self, agent_id, *, force=False):
+            ...
+
+    executor = MockExecutor()
+    assert isinstance(executor, TeammateExecutor)
+
+
+def test_teammate_executor_missing_method_fails_check():
+    class IncompleteExecutor:
+        type = "subprocess"
+
+        def is_available(self) -> bool:
+            return True
+
+        # Missing: spawn, send_message, shutdown
+
+    incomplete = IncompleteExecutor()
+    assert not isinstance(incomplete, TeammateExecutor)

tests/test_swarm/test_worktree.py

@@ -0,0 +1,130 @@
+"""Tests for validate_worktree_slug edge cases and WorktreeManager helpers."""
+
+from __future__ import annotations
+
+import pytest
+
+from openharness.swarm.worktree import (
+    _flatten_slug,
+    _worktree_branch,
+    validate_worktree_slug,
+)
+
+
+# ---------------------------------------------------------------------------
+# validate_worktree_slug — valid cases
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "slug",
+    [
+        "simple",
+        "with-dashes",
+        "with_underscores",
+        "alpha123",
+        "a.b.c",
+        "feature/my-task",
+        "a/b/c",
+        "A-Z_0-9.mixed",
+        "x" * 64,  # exactly 64 chars
+    ],
+)
+def test_validate_worktree_slug_valid(slug):
+    assert validate_worktree_slug(slug) == slug
+
+
+# ---------------------------------------------------------------------------
+# validate_worktree_slug — invalid cases
+# ---------------------------------------------------------------------------
+
+
+def test_validate_empty_slug_raises():
+    with pytest.raises(ValueError, match="empty"):
+        validate_worktree_slug("")
+
+
+def test_validate_too_long_slug_raises():
+    with pytest.raises(ValueError, match="64"):
+        validate_worktree_slug("x" * 65)
+
+
+def test_validate_absolute_path_raises():
+    with pytest.raises(ValueError, match="absolute"):
+        validate_worktree_slug("/absolute/path")
+
+
+def test_validate_backslash_absolute_raises():
+    with pytest.raises(ValueError, match="absolute"):
+        validate_worktree_slug("\\windows\\path")
+
+
+def test_validate_dot_segment_raises():
+    with pytest.raises(ValueError, match=r"\.|\.\."):
+        validate_worktree_slug("a/./b")
+
+
+def test_validate_dotdot_segment_raises():
+    with pytest.raises(ValueError, match=r"\.|\.\."):
+        validate_worktree_slug("a/../b")
+
+
+def test_validate_invalid_chars_raises():
+    with pytest.raises(ValueError):
+        validate_worktree_slug("has space")
+
+
+def test_validate_empty_segment_via_double_slash_raises():
+    with pytest.raises(ValueError):
+        validate_worktree_slug("a//b")
+
+
+@pytest.mark.parametrize(
+    "slug",
+    [
+        "has space",
+        "has@symbol",
+        "has!bang",
+        "has$dollar",
+        "has#hash",
+        "has%percent",
+    ],
+)
+def test_validate_various_invalid_chars(slug):
+    with pytest.raises(ValueError):
+        validate_worktree_slug(slug)
+
+
+# ---------------------------------------------------------------------------
+# _flatten_slug
+# ---------------------------------------------------------------------------
+
+
+def test_flatten_slug_replaces_slash_with_plus():
+    assert _flatten_slug("feature/my-task") == "feature+my-task"
+
+
+def test_flatten_slug_no_slash_unchanged():
+    assert _flatten_slug("simple") == "simple"
+
+
+def test_flatten_slug_multiple_slashes():
+    assert _flatten_slug("a/b/c") == "a+b+c"
+
+
+# ---------------------------------------------------------------------------
+# _worktree_branch
+# ---------------------------------------------------------------------------
+
+
+def test_worktree_branch_simple():
+    assert _worktree_branch("fix-bug") == "worktree-fix-bug"
+
+
+def test_worktree_branch_with_slash():
+    assert _worktree_branch("feature/foo") == "worktree-feature+foo"
+
+
+def test_worktree_branch_prefix():
+    branch = _worktree_branch("anything")
+    assert branch.startswith("worktree-")