handler/tests/test_control_spawn.py

"""Control-layer spawn: the hard test-task gate, settings generation, identity env."""

from __future__ import annotations

import json

import pytest

from handler.control import spawn
from handler.db import repository as repo
from handler.db.engine import get_engine


def _register_project(root):
    with get_engine().begin() as conn:
        repo.create_project(conn, "proj", str(root))


def _write_mise(root, with_test=True):
    root.mkdir(parents=True, exist_ok=True)
    body = "[tasks.lint]\nrun = 'ruff check .'\n"
    if with_test:
        body = "[tasks.test]\nrun = 'pytest'\n" + body
    (root / ".mise.toml").write_text(body)


def test_spawn_refuses_without_test_task(env, fake_tmux):
    root = env["tmp"] / "proj"
    _write_mise(root, with_test=False)
    _register_project(root)
    with pytest.raises(spawn.SpawnError, match="no \\[tasks.test\\]"):
        spawn.spawn("proj", "api")
    assert fake_tmux["calls"]["new_session"] == []


def test_spawn_refuses_without_mise_file(env, fake_tmux):
    root = env["tmp"] / "proj"
    root.mkdir(parents=True, exist_ok=True)
    _register_project(root)
    with pytest.raises(spawn.SpawnError, match="no mise config"):
        spawn.spawn("proj", "api")


def test_spawn_accepts_dotless_mise_toml(env, fake_tmux):
    # mise also reads `mise.toml` (no leading dot); the gate must honor it too.
    root = env["tmp"] / "proj"
    root.mkdir(parents=True, exist_ok=True)
    (root / "mise.toml").write_text("[tasks.test]\nrun = 'pytest'\n")
    _register_project(root)

    agent = spawn.spawn("proj", "api")
    with get_engine().begin() as conn:
        assert repo.get_agent_by_name(conn, "proj", "api")["id"] == agent["id"]


def test_spawn_creates_agent_settings_and_session(env, fake_tmux):
    root = env["tmp"] / "proj"
    _write_mise(root, with_test=True)
    _register_project(root)

    agent = spawn.spawn("proj", "api", task="build the thing")

    # Agent row created.
    with get_engine().begin() as conn:
        assert repo.get_agent_by_name(conn, "proj", "api")["id"] == agent["id"]

    # settings.json wires all four hook events.
    settings = json.loads((root / ".claude" / "settings.json").read_text())
    assert set(settings["hooks"]) == {"Stop", "SessionEnd", "PreToolUse", "Notification"}
    pre = settings["hooks"]["PreToolUse"][0]
    assert pre["matcher"] == "AskUserQuestion|Bash"
    assert "handler.hooks pre_tool_use" in pre["hooks"][0]["command"]

    # tmux session named project__agent, with identity + DATABASE_URL in env.
    call = fake_tmux["calls"]["new_session"][0]
    assert call["name"] == "proj__api"
    assert call["env"]["HANDLER_PROJECT_ID"] == "proj"
    assert call["env"]["HANDLER_AGENT_NAME"] == "api"
    assert call["env"]["HANDLER_AGENT_ID"] == str(agent["id"])
    assert call["env"]["DATABASE_URL"] == env["url"]


def test_spawn_mise_init_skips_test_gate_and_marks_env(env, fake_tmux):
    # A repo with no .mise.toml at all: the normal gate would refuse, but the mise-init
    # bootstrap agent must launch anyway (creating that file is its whole job).
    root = env["tmp"] / "proj"
    root.mkdir(parents=True, exist_ok=True)
    _register_project(root)

    agent = spawn.spawn("proj", "mise-init", require_tests=False, mise_init=True)

    with get_engine().begin() as conn:
        assert repo.get_agent_by_name(conn, "proj", "mise-init")["id"] == agent["id"]
    # The launched session carries HANDLER_MISE_INIT so its hooks enforce commit + push.
    call = fake_tmux["calls"]["new_session"][0]
    assert call["env"]["HANDLER_MISE_INIT"] == "1"


def test_spawn_still_gates_without_mise_init_flag(env, fake_tmux):
    root = env["tmp"] / "proj"
    root.mkdir(parents=True, exist_ok=True)
    _register_project(root)
    # require_tests defaults on, so a normal spawn against a mise-less repo still refuses.
    with pytest.raises(spawn.SpawnError, match="no mise config"):
        spawn.spawn("proj", "api")
    assert fake_tmux["calls"]["new_session"] == []


def test_kill_sets_done_and_kills_session(env, fake_tmux):
    root = env["tmp"] / "proj"
    _write_mise(root, with_test=True)
    _register_project(root)
    spawn.spawn("proj", "api")

    spawn.kill("proj", "api")
    assert "proj__api" in fake_tmux["calls"]["kill_session"]
    with get_engine().begin() as conn:
        assert repo.get_agent_by_name(conn, "proj", "api")["status"] == "done"


def test_resume_sends_answer_to_live_session(env, fake_tmux):
    root = env["tmp"] / "proj"
    _write_mise(root, with_test=True)
    _register_project(root)
    agent = spawn.spawn("proj", "api")

    ok, detail = spawn.resume(agent, "use Postgres")
    assert ok is True
    assert fake_tmux["calls"]["send_keys"][0] == {"name": "proj__api", "keys": "use Postgres"}