GodopuandClaude Opus 5.5
75936f11c6
`ps -o lstart=` pads a one-digit day ("Oct 1"); loop_lock.sh's mam_lstart
squeezes the spaces before comparing, but A8 wrote the raw value, so the
live holder's marker looked stale and the second loop was not refused on
the 1st-9th of every month. The test now normalises like the product (the
product is unchanged) and also runs with a `ps` shim that always prints a
padded lstart, so the padded case is checked on every date.
The other tests that record a live lstart (test_o3_scoped_guard
_get_lstart) already normalise; test_o2's markers carry fixed stale values.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGpJXhSZasJLU38ksDghe
308 lines
14 KiB
Python
308 lines
14 KiB
Python
"""Issue #21 item 7: the eight acceptance cases (A1-A8) of the worktree
|
|
creator model, one test (or pair) per case.
|
|
|
|
Everything runs against mock herdr / stubbed delegation inside mam_sandbox or
|
|
tmp_path; nothing touches the real .mam. The real herdr pane (whether a pane
|
|
sees the --env values) and the trial operation belong to stage C - A2 pins
|
|
what the scripts hand to herdr.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from test_issue21_loop_b import LOCAL_ROWS, _drive, _reviews, _sidecars, _state
|
|
from test_issue21_phase2_loop_a import (REM, _calls, _prepare, _resume, _rows, _run, _skill, _wt, # noqa: F401
|
|
ctl)
|
|
from test_issue21_worktree_prereqs import (CP_ENV, WRAPPER, _assert_no_env_in_agent_args, _binding, _dry_submit,
|
|
_env, _env_flags, _injected, _kill_timer, _only_job_id)
|
|
from test_issue21_worktree_prereqs import _git as _g
|
|
from test_issue21_worktree_prereqs import wt_repo # noqa: F401
|
|
|
|
CREATE = ("multi-agent-mux-create", "scripts", "create_session.sh")
|
|
HS = "multi-agent-mux" # listed as running by mock herdr: an existing server
|
|
|
|
|
|
def _raw_calls(state_file) -> list:
|
|
return json.loads(Path(state_file).read_text()).get("calls", [])
|
|
|
|
|
|
def _sibling(main: Path, state_file):
|
|
"""A1 setup: a session on the existing server in the control root, then a
|
|
sibling worktree's creator on the same server. Returns (wt, since, rows before)."""
|
|
res = _run(main, CREATE, "--workspace", str(main), "--agent", "claude", "--role", "reviewer",
|
|
"--herdr-session", HS, "--no-onboard")
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
before = _rows(main)
|
|
since = len(_raw_calls(state_file))
|
|
wt = _wt(main)
|
|
res = _prepare(main, wt, "--creator-agent", "claude", "--herdr-session", HS)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
return wt, since, before
|
|
|
|
|
|
# A1 -------------------------------------------------------------------------
|
|
|
|
def test_a1_sibling_worktree_on_an_existing_herdr_server(ctl, mock_herdr, mock_agents):
|
|
main = ctl
|
|
wt, since, before = _sibling(main, mock_herdr)
|
|
new = _raw_calls(mock_herdr)[since:]
|
|
assert not [c for c in new if c[-1:] == ["server"]], "no server bootstrap on an existing server"
|
|
rows = _rows(main)
|
|
assert rows[0] == before[0], "the existing row is untouched"
|
|
creator = [r for r in rows if r.get("role") == "creator"]
|
|
assert len(rows) == 2 and len(creator) == 1
|
|
assert rows[0].get("herdr_session") == creator[0].get("herdr_session") == HS
|
|
assert os.path.realpath(creator[0]["pane"]["cwd"]) == os.path.realpath(wt)
|
|
assert creator[0]["control_plane"]["workspace_root"] == os.path.realpath(main)
|
|
|
|
|
|
# A2 -------------------------------------------------------------------------
|
|
|
|
def test_a2_pane_env_carries_the_binding_not_the_agent_args(ctl, mock_herdr, mock_agents):
|
|
main = ctl
|
|
wt, since, _ = _sibling(main, mock_herdr)
|
|
flags = _env_flags(mock_herdr, since)
|
|
for k, v in zip(CP_ENV, _binding(main).values()):
|
|
assert f"{k}={v}" in flags, flags
|
|
spawns = [c for c in _calls(mock_herdr) if c[:2] in (["pane", "split"], ["workspace", "create"])]
|
|
assert spawns and any("--env" in c for c in spawns), "the binding rides on the pane-creating call"
|
|
_assert_no_env_in_agent_args(mock_herdr, since)
|
|
|
|
|
|
# A3 -------------------------------------------------------------------------
|
|
|
|
def test_a3_brief_start_and_report_route_to_the_control_root(tmp_path):
|
|
main, wt = tmp_path / "main", tmp_path / "wt"
|
|
main.mkdir()
|
|
_g(main, "init", "-q")
|
|
(main / "a.txt").write_text("a\n")
|
|
_g(main, "add", "a.txt")
|
|
_g(main, "commit", "-q", "-m", "init")
|
|
_g(main, "worktree", "add", "-q", "-b", "feat", str(wt))
|
|
real = os.path.realpath(main)
|
|
res = _dry_submit(main, _env(tmp_path, {"WORKSPACE_ROOT": main}), "--workdir", str(main))
|
|
assert res.returncode == 0, res.stderr
|
|
reg = main / ".mam" / "jobs"
|
|
jid = _only_job_id(reg)
|
|
assert f"{real}/.mam/jobs/{jid}/brief.md" in res.stdout
|
|
start = _injected(res.stdout, "start")
|
|
assert f"--registry-dir {real}/.mam/jobs " in start and start.startswith(f"DELEGATE_JOB_LOGS_DIR={real}/")
|
|
# the brief (not written by --dry-run) derives its report path from that same
|
|
# absolute registry dir, so the report lands under MAIN as well
|
|
wrapper = WRAPPER.read_text()
|
|
assert "- **Output Report Path**: $REGISTRY_DIR/$JOB_ID/$AGENT-reports/report-final.md" in wrapper
|
|
try:
|
|
r1 = subprocess.run(["bash", "-c", start + " --attempts 1"], capture_output=True, text=True,
|
|
env=_env(tmp_path), cwd=str(wt), timeout=120) # no control env, broker unreachable
|
|
assert r1.returncode == 2, r1.stderr
|
|
assert json.loads((reg / f"{jid}.json").read_text())["status"] != "pending"
|
|
finally:
|
|
_kill_timer(reg, jid)
|
|
assert not (wt / ".mam").exists(), "nothing lands in the worktree"
|
|
|
|
|
|
# A4 -------------------------------------------------------------------------
|
|
|
|
def _main_state(main: Path) -> tuple:
|
|
branch = _g(main, "branch", "--show-current")
|
|
return (_g(main, "rev-parse", "HEAD"), branch, _g(main, "status", "--porcelain"),
|
|
_g(main, "rev-parse", f"refs/heads/{branch}"), _g(main, "worktree", "list", "--porcelain"))
|
|
|
|
|
|
def test_a4_main_checkout_is_unchanged_by_the_whole_cycle(ctl, monkeypatch):
|
|
main = ctl
|
|
# what prepare adds to info/exclude is added up front, so status compares like for like
|
|
excl = main / ".git" / "info" / "exclude"
|
|
excl.write_text(excl.read_text() + "/.venv\n.pytest_cache/\n")
|
|
scratch = main.parent / f"{main.name}-a4"
|
|
scratch.mkdir()
|
|
pre = _main_state(main)
|
|
wt = _wt(main)
|
|
res = _prepare(main, wt)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
base = _g(wt, "rev-parse", "HEAD")
|
|
(wt / "impl.txt").write_text("impl line\n")
|
|
_g(wt, "add", "impl.txt")
|
|
_g(wt, "commit", "-q", "-m", "impl")
|
|
_state(monkeypatch, LOCAL_ROWS)
|
|
rc, calls, r = _drive(monkeypatch, scratch, ["--worktree", str(wt), "--base", base, "--review-only"])
|
|
assert rc == 0 and len(_reviews(calls)) == 1
|
|
res = _run(main, REM, "--path", str(wt))
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert _main_state(main) == pre, "HEAD, branch, status, branch tip and worktree list are unchanged"
|
|
assert "mam-review" not in _g(main, "worktree", "list", "--porcelain")
|
|
|
|
|
|
# A5 -------------------------------------------------------------------------
|
|
|
|
def _review_path(prompt: str) -> Path:
|
|
m = re.search(r"^- 검증 경로: (.+)$", prompt, re.M)
|
|
assert m, prompt[:800]
|
|
return Path(m.group(1))
|
|
|
|
|
|
def _commit_h2(wt: Path) -> str:
|
|
(wt / "impl.txt").write_text("impl line\nsecond commit\n")
|
|
_g(wt, "commit", "-q", "-am", "h2")
|
|
return _g(wt, "rev-parse", "HEAD")
|
|
|
|
|
|
def test_a5_fixed_sha_review_while_the_creator_keeps_committing(wt_repo, monkeypatch, tmp_path):
|
|
main, wt = wt_repo
|
|
base = _g(wt, "rev-parse", "HEAD")
|
|
(wt / "impl.txt").write_text("impl line\n")
|
|
_g(wt, "add", "impl.txt")
|
|
_g(wt, "commit", "-q", "-m", "h1")
|
|
h1 = _g(wt, "rev-parse", "HEAD")
|
|
seen = {}
|
|
|
|
def reviewer(r, n, prompt):
|
|
if seen:
|
|
return
|
|
path = _review_path(prompt)
|
|
seen["head"] = _g(path, "rev-parse", "HEAD")
|
|
seen["symref"] = subprocess.run(["git", "-C", str(path), "symbolic-ref", "-q", "HEAD"]).returncode
|
|
seen["content"] = (path / "impl.txt").read_text()
|
|
seen["h2"] = _commit_h2(wt)
|
|
|
|
rc, calls, r = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", base, "--review-only"],
|
|
on_review=reviewer, max_loop=2)
|
|
assert rc == 0
|
|
assert (seen["head"], seen["symref"], seen["content"]) == (h1, 1, "impl line\n"), "round 1 saw H1, detached"
|
|
assert [role for _, role, _ in calls] == ["Reviewer", "Reviewer"], "no correction job for a stale round"
|
|
assert f"- head: {seen['h2']}" in _reviews(calls)[1][1]
|
|
rounds = r._rounds
|
|
assert rounds[0]["stale"] is True and rounds[1]["head_sha"] == seen["h2"]
|
|
(side,) = _sidecars(main)
|
|
data = json.loads(side.read_text())
|
|
assert (data["head_sha"], data["evidence"]) == (seen["h2"], "review-tree")
|
|
|
|
|
|
def test_a5b_a_stale_last_round_is_not_accepted(wt_repo, monkeypatch, tmp_path, capsys):
|
|
main, wt = wt_repo
|
|
base = _g(wt, "rev-parse", "HEAD")
|
|
(wt / "impl.txt").write_text("impl line\n")
|
|
_g(wt, "add", "impl.txt")
|
|
_g(wt, "commit", "-q", "-m", "h1")
|
|
rc, calls, r = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", base, "--review-only"],
|
|
on_review=lambda *_: _commit_h2(wt), max_loop=1)
|
|
assert rc == 1 and "changed during review" in capsys.readouterr().err
|
|
assert _sidecars(main) == []
|
|
|
|
|
|
# A6 -------------------------------------------------------------------------
|
|
|
|
def test_a6_row_recovery_and_resume_only_at_the_original_cwd(ctl, mock_herdr, mock_agents):
|
|
main = ctl
|
|
wt = _wt(main)
|
|
res = _prepare(main, wt, "--creator-agent", "claude")
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
(row,) = _rows(main)
|
|
name, binding = row["name"], row["control_plane"]
|
|
# the creator died: gone from herdr, the row terminated (reconcile drift A)
|
|
st = json.loads(Path(mock_herdr).read_text())
|
|
st["agents"] = {}
|
|
Path(mock_herdr).write_text(json.dumps(st))
|
|
from test_issue21_phase2_loop_a import _mutate
|
|
_mutate(main, "d['herdr_sessions'][0]['status'] = 'terminated'")
|
|
res = _resume(main, main, name)
|
|
assert res.returncode == 2 and (str(wt) in res.stderr or os.path.realpath(wt) in res.stderr), res.stderr
|
|
since = len(_raw_calls(mock_herdr))
|
|
res = _resume(main, wt, name)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
(row,) = _rows(main)
|
|
assert row["status"] == "running" and row["control_plane"] == binding
|
|
flags = _env_flags(mock_herdr, since)
|
|
assert f"WORKSPACE_ROOT={binding['workspace_root']}" in flags, flags
|
|
|
|
|
|
# A7 -------------------------------------------------------------------------
|
|
|
|
def test_a7_reviewer_test_output_stays_out_of_the_source_and_the_sandbox(wt_repo, monkeypatch, tmp_path):
|
|
main, wt = wt_repo
|
|
base = _g(wt, "rev-parse", "HEAD")
|
|
(wt / "impl.txt").write_text("impl line\n")
|
|
status0 = _g(wt, "status", "--porcelain")
|
|
paths = []
|
|
|
|
def reviewer_runs_tests(r, n, prompt):
|
|
path = _review_path(prompt)
|
|
paths.append(path)
|
|
(path / ".pytest_cache").mkdir()
|
|
(path / ".pytest_cache" / "v").write_text("x\n")
|
|
(path / "__pycache__").mkdir()
|
|
(path / "__pycache__" / "x.pyc").write_bytes(b"\0")
|
|
(path / "tmp-output.txt").write_text("output\n")
|
|
|
|
rc, calls, r = _drive(monkeypatch, tmp_path, ["--worktree", str(wt), "--base", base, "--review-only"],
|
|
on_review=reviewer_runs_tests)
|
|
assert rc == 0 and len(paths) == 1
|
|
assert _g(wt, "status", "--porcelain") == status0, "the source worktree did not change"
|
|
assert not paths[0].exists() and str(paths[0]) not in _g(main, "worktree", "list", "--porcelain")
|
|
prompt = _reviews(calls)[0][1]
|
|
assert "PYTHONDONTWRITEBYTECODE=1" in prompt and "-p no:cacheprovider" in prompt
|
|
assert not (wt / ".mam").exists() and not (wt / ".agents").exists(), "state stays in the control root"
|
|
assert len(_sidecars(main)) == 1
|
|
|
|
|
|
# A8 -------------------------------------------------------------------------
|
|
|
|
# `ps -o lstart=` pads a one-digit day with a second space ("Oct 1"); this
|
|
# stands in for it so the padded case runs on every date, not only on the 1st-9th.
|
|
_PADDED_PS = """#!/bin/sh
|
|
case " $* " in
|
|
*" lstart= "*) out=$(/bin/ps "$@") || exit $?; [ -n "$out" ] && echo "Thu Oct 1 08:02:03 2026"; exit 0 ;;
|
|
esac
|
|
exec /bin/ps "$@"
|
|
"""
|
|
|
|
|
|
def _marker_lstart(pid: int, env: dict) -> str:
|
|
"""The lstart a marker records, normalised as loop_lock.sh's mam_lstart does
|
|
(`tr -s ' '`, ends trimmed) - the raw value keeps the padding and never matches."""
|
|
out = subprocess.run(["ps", "-p", str(pid), "-o", "lstart="], capture_output=True, text=True, env=env).stdout
|
|
return " ".join(out.split())
|
|
|
|
|
|
@pytest.mark.parametrize("ps", ["real", "padded"])
|
|
def test_a8_second_loop_is_refused_before_a_scheduler_exists(ctl, tmp_path, ps):
|
|
main = ctl
|
|
wt2 = _wt(main, "wt2")
|
|
_g(main, "worktree", "add", "-q", "--detach", str(wt2))
|
|
env = {**os.environ, "WORKSPACE_ROOT": str(main), "MAM_LOOP_NO_FREEZE": "1"}
|
|
env.pop("MAM_LOOP_PY", None)
|
|
if ps == "padded":
|
|
shim = tmp_path / "ps-shim"
|
|
shim.mkdir()
|
|
(shim / "ps").write_text(_PADDED_PS)
|
|
(shim / "ps").chmod(0o755)
|
|
env["PATH"] = f"{shim}{os.pathsep}{env['PATH']}"
|
|
holder = subprocess.Popen(["sleep", "60"])
|
|
try:
|
|
lstart = _marker_lstart(holder.pid, env)
|
|
assert lstart, "ps found no holder"
|
|
if ps == "padded":
|
|
assert lstart == "Thu Oct 1 08:02:03 2026", f"the shim was not used: {lstart!r}"
|
|
marker = main.parent / f"{main.name}-loop-marker"
|
|
content = f"pid={holder.pid}\nlstart={lstart}\nstarted={time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())}\n"
|
|
marker.write_text(content)
|
|
env["MAM_LOOP_MARKER"] = str(marker)
|
|
res = subprocess.run(["bash", _skill(main, "multi-agent-mux-loop", "scripts", "run_loop.sh"),
|
|
"--creator", "c", "--task", "t", "--worktree", str(wt2), "--base", "HEAD",
|
|
"--review-only", "--reviewer", "r"],
|
|
capture_output=True, text=True, env=env, cwd=str(main), timeout=120)
|
|
assert res.returncode == 1 and "already running" in (res.stdout + res.stderr), res.stdout + res.stderr
|
|
assert list((main / ".mam" / "jobs").glob("*.json")) == [], "no job was registered"
|
|
assert marker.read_text() == content, "the live holder's marker is kept"
|
|
finally:
|
|
holder.kill()
|
|
holder.wait()
|