istosandClaude Fable 5
cc931a9174
Three layout defects broke ./start.sh in the self-hosted repo, one of which (missing local/state/) broke every fresh vendored install too: - install.py resolved the project as TM.parent, a hardcoded vendored- layout assumption. It now asks git for the toplevel from the manager's directory (same resolution as config._repo_root), falling back to the parent when git is unavailable — vendored installs still find the host repo, self-hosted bench finds itself instead of its parent. - adapters/claude/wire hardcoded ".task-manager/" into the emit hook command and plansDirectory. Both are now derived from the manager's path relative to the project root, so vendored installs keep the .task-manager/ prefix and self-hosted bench gets prefix-free paths. _is_ours also recognises the emit.py suffix, so settings wired with the old literal path count as stale and are repaired idempotently. - board.py and state.py created the sessions/agent dirs without parents=True; local/state/ is gitignored and ships empty, so a virgin checkout died with FileNotFoundError before serving. Boot now creates the whole chain, wiring or no wiring. tests/test_self_hosting.py (stdlib unittest) covers both layouts' wiring, stale-path repair, idempotent re-runs, refusal without .claude/, root resolution with and without git, and an integration boot of board.py from a scratch checkout with no local/state/. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
265 lines
9.7 KiB
Python
265 lines
9.7 KiB
Python
"""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()
|