bench/tests/test_self_hosting.py

"""Layout resolution: the manager must start cleanly both vendored one
level inside a host repo (.task-manager/) and self-hosted, where this repo
is simultaneously the distribution and the project.

Stdlib only, like everything else here:

    python3 -m unittest discover -s tests -v
"""

from __future__ import annotations

import json
import os
import shutil
import socket
import subprocess
import sys
import tempfile
import time
import unittest
import urllib.request
from pathlib import Path

REPO = Path(__file__).resolve().parents[1]
ADAPTER_FILES = ("wire", "emit.py")

EMIT_SUFFIX = "manager/core/adapters/claude/emit.py"
OLD_EMIT_CMD = ('python3 "$CLAUDE_PROJECT_DIR/.task-manager/'
                'manager/core/adapters/claude/emit.py"')


class TempDirTestCase(unittest.TestCase):
    def setUp(self):
        # resolve(): macOS tempdirs live behind the /var → /private/var
        # symlink, and both wire and git report resolved paths.
        self.tmp = Path(tempfile.mkdtemp(prefix="bench-test-")).resolve()
        self.addCleanup(shutil.rmtree, self.tmp, True)


def make_manager(tm_root: Path) -> None:
    """A minimal manager tree at tm_root: install.py plus the claude
    adapter's wire and emit.py, copied from this repo."""
    adapter = tm_root / "manager" / "core" / "adapters" / "claude"
    adapter.mkdir(parents=True)
    for name in ADAPTER_FILES:
        shutil.copy(REPO / "manager" / "core" / "adapters" / "claude" / name,
                    adapter / name)
    shutil.copy(REPO / "install.py", tm_root / "install.py")


def run(cmd: list, cwd: Path) -> subprocess.CompletedProcess:
    env = {k: v for k, v in os.environ.items() if not k.startswith("BOARD_")}
    return subprocess.run([str(a) for a in cmd], cwd=str(cwd), env=env,
                          capture_output=True, text=True)


def wire(tm_root: Path, project: Path) -> subprocess.CompletedProcess:
    script = tm_root / "manager" / "core" / "adapters" / "claude" / "wire"
    return run([sys.executable, script, project], cwd=project)


def settings(project: Path) -> dict:
    return json.loads((project / ".claude" / "settings.json").read_text())


def our_hook_commands(cfg: dict) -> set:
    return {
        hook["command"]
        for groups in cfg.get("hooks", {}).values()
        for group in groups
        for hook in group.get("hooks", [])
        if "emit.py" in hook.get("command", "")
    }


class WireLayouts(TempDirTestCase):
    def test_self_hosted_paths_have_no_task_manager_segment(self):
        project = self.tmp / "bench"
        make_manager(project)
        (project / ".claude").mkdir()

        result = wire(project, project)

        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        cfg = settings(project)
        self.assertEqual(cfg["plansDirectory"], "./plans")
        self.assertEqual(
            our_hook_commands(cfg),
            {f'python3 "$CLAUDE_PROJECT_DIR/{EMIT_SUFFIX}"'})

    def test_vendored_paths_keep_task_manager_segment(self):
        host = self.tmp / "host"
        make_manager(host / ".task-manager")
        (host / ".claude").mkdir()

        result = wire(host / ".task-manager", host)

        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        cfg = settings(host)
        self.assertEqual(cfg["plansDirectory"], "./.task-manager/plans")
        self.assertEqual(
            our_hook_commands(cfg),
            {f'python3 "$CLAUDE_PROJECT_DIR/.task-manager/{EMIT_SUFFIX}"'})

    def test_stale_hardcoded_path_is_repaired_in_place(self):
        project = self.tmp / "bench"
        make_manager(project)
        (project / ".claude").mkdir()
        stale_hook = {"type": "command", "command": OLD_EMIT_CMD, "timeout": 5}
        (project / ".claude" / "settings.json").write_text(json.dumps({
            "plansDirectory": "./.task-manager/plans",
            "hooks": {"SessionStart": [{"hooks": [stale_hook]}],
                      "PreToolUse": [{"matcher": "Bash", "hooks": [stale_hook]}]},
        }))

        result = wire(project, project)

        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        cfg = settings(project)
        self.assertEqual(cfg["plansDirectory"], "./plans")
        self.assertEqual(
            our_hook_commands(cfg),
            {f'python3 "$CLAUDE_PROJECT_DIR/{EMIT_SUFFIX}"'})
        hooks = [hook for group in cfg["hooks"]["SessionStart"]
                 for hook in group["hooks"]]
        self.assertEqual(len(hooks), 1, "stale hook must be replaced, not kept")

    def test_second_run_is_a_no_op(self):
        for project, tm_root in (
            (self.tmp / "bench", self.tmp / "bench"),
            (self.tmp / "host", self.tmp / "host" / ".task-manager"),
        ):
            with self.subTest(project=project.name):
                make_manager(tm_root)
                (project / ".claude").mkdir()
                wire(tm_root, project)
                before = (project / ".claude" / "settings.json").read_text()

                result = wire(tm_root, project)

                self.assertEqual(result.returncode, 0)
                self.assertIn("nothing to do", result.stdout)
                self.assertEqual(
                    (project / ".claude" / "settings.json").read_text(), before)

    def test_refuses_project_without_claude_dir(self):
        project = self.tmp / "bench"
        make_manager(project)

        result = wire(project, project)

        self.assertEqual(result.returncode, 1)
        self.assertIn("not a .claude-initialised project", result.stdout)


class InstallRootResolution(TempDirTestCase):
    def _git_init(self, path: Path) -> None:
        subprocess.run(["git", "init", "--quiet", str(path)], check=True,
                       capture_output=True)

    def test_vendored_resolves_host_repo_root(self):
        host = self.tmp / "host"
        make_manager(host / ".task-manager")
        self._git_init(host)
        (host / ".claude").mkdir()

        result = run([sys.executable, host / ".task-manager" / "install.py"],
                     cwd=host)

        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        self.assertEqual(settings(host)["plansDirectory"],
                         "./.task-manager/plans")

    def test_self_hosted_resolves_own_repo_not_parent(self):
        project = self.tmp / "bench"
        make_manager(project)
        self._git_init(project)
        (project / ".claude").mkdir()

        result = run([sys.executable, project / "install.py"], cwd=project)

        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        self.assertEqual(settings(project)["plansDirectory"], "./plans")
        self.assertFalse((self.tmp / ".claude").exists(),
                         "must not wire the parent of the repo")

    def test_no_git_falls_back_to_parent(self):
        host = self.tmp / "host"
        make_manager(host / ".task-manager")
        (host / ".claude").mkdir()

        result = run([sys.executable, host / ".task-manager" / "install.py"],
                     cwd=host)

        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        self.assertEqual(settings(host)["plansDirectory"],
                         "./.task-manager/plans")

    def test_refusal_names_the_resolved_root(self):
        project = self.tmp / "bench"
        make_manager(project)
        self._git_init(project)

        result = run([sys.executable, project / "install.py"], cwd=project)

        self.assertEqual(result.returncode, 1)
        self.assertIn(f"{project} is not a .claude-initialised project",
                      result.stdout)


class VirginBoot(TempDirTestCase):
    """board.py must boot and serve from a checkout with no local/state/
    and no wiring at all — start.sh tolerates a failed wire on purpose."""

    def test_board_boots_without_state_dirs_or_wiring(self):
        tm_root = self.tmp / "bench"
        shutil.copytree(REPO / "manager" / "core",
                        tm_root / "manager" / "core",
                        ignore=shutil.ignore_patterns("__pycache__"))
        (tm_root / "manager" / "local").mkdir()
        for stage in ("backlog", "to-do", "in-progress", "review", "done"):
            (tm_root / "tasks" / stage).mkdir(parents=True)

        probe = socket.socket()
        probe.bind(("127.0.0.1", 0))
        port = probe.getsockname()[1]
        probe.close()

        env = {k: v for k, v in os.environ.items() if not k.startswith("BOARD_")}
        proc = subprocess.Popen(
            [sys.executable, str(tm_root / "manager" / "core" / "board.py"),
             "--port", str(port), "--no-open"],
            env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        try:
            state = None
            for _ in range(50):
                if proc.poll() is not None:
                    break
                try:
                    with urllib.request.urlopen(
                            f"http://127.0.0.1:{port}/api/state",
                            timeout=1) as response:
                        state = json.load(response)
                    break
                except OSError:
                    time.sleep(0.2)

            if proc.poll() is not None:
                out, err = proc.communicate()
                self.fail(f"board died (rc={proc.returncode}):\n{out}\n{err}")
            self.assertIsNotNone(state, "board never answered /api/state")
            local_state = tm_root / "manager" / "local" / "state"
            self.assertTrue((local_state / "sessions").is_dir())
            self.assertTrue((local_state / "agent").is_dir())
        finally:
            proc.terminate()
            proc.wait(timeout=10)
            for stream in (proc.stdout, proc.stderr):
                if stream:
                    stream.close()


if __name__ == "__main__":
    unittest.main()