"""Issue #34: the loop waits for the creator job's sub-jobs, reviewers see them. A creator that hands part of its work to another agent (delegate-job submit) used to end its own job while the sub-job was still running; the loop went straight to review and the reviewers passed half-finished work. Jobs now record their parent (`submit --parent-job`), the loop waits for the creator job's sub-job tree, the review prompt carries a sub-job table, and a round whose source changed during review is void in every mode (a sub-job closed by its deadline is not interrupted and may still be writing). Sub-jobs are registered by the real wrapper (`submit --dry-run`, which returns after the real registry wrote the record) and closed by the real producers: `publish_event.py`, and for a deadline the wrapper's own sequence (`registry.py close --request deadline_timeout`, then `publish_event --event error --detail deadline_timeout`). The loop is a real `python -m lib_py.loop` process; only the loop's own agents are simulated, by an executable behind a wrapper at the control root (the #36 harness). T34_SKILLS points the harness at another copy of .agents/skills (used once to run T3/T5b against the base runner). Every test writes only under tmp_path. """ from __future__ import annotations import json import os import re import subprocess import sys import threading import time from pathlib import Path import pytest REPO_ROOT = Path(__file__).resolve().parent.parent SKILLS = Path(os.environ.get("T34_SKILLS") or REPO_ROOT / ".agents" / "skills") WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job" REGISTRY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "registry.py" PUBLISH = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py" if str(SKILLS) not in sys.path: sys.path.insert(0, str(SKILLS)) UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"} GIT_ID = {"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@example.invalid", "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@example.invalid"} def _env(home: Path, extra=None, drop=()) -> dict: env = dict(os.environ) env["DELEGATE_JOB_PYTHON"] = sys.executable env["AGENT_PYTHON_BIN"] = sys.executable env["HOME"] = str(home) env["MAM_ENV_FILE"] = str(home / ".mam.env.absent") env["PYTHONPATH"] = str(SKILLS) for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD", "MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR", "WORKSPACE_ROOT", "MAM_REAL_ROOT", "MAM_JOB_ID", "MAM_CHILD_WAIT_S", "MAM_JOB_TREE_MAX"): env.pop(k, None) env.update(UNREACHABLE) env.update(GIT_ID) env["DELEGATE_JOB_LOGS_DIR"] = str(home / ".mam-logs") env["WORKSPACE_ROOT"] = str(home) if extra: env.update({k: str(v) for k, v in extra.items()}) for k in drop: env.pop(k, None) return env def _git(cwd: Path, *args: str) -> str: return subprocess.run(["git", "-C", str(cwd), *args], capture_output=True, text=True, check=True, env={**os.environ, **GIT_ID}).stdout.strip() def _repo(path: Path) -> Path: path.mkdir(parents=True, exist_ok=True) _git(path, "init", "-q") (path / "a.txt").write_text("a\n") _git(path, "add", "a.txt") _git(path, "commit", "-q", "-m", "init") return path def _submit(env: dict, reg: Path, *extra: str) -> subprocess.CompletedProcess: """The real wrapper; --dry-run returns right after the real registration.""" return subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--agent-session", "herdr:img", "--dry-run", "--registry-dir", str(reg), "--prompt", "draw the figures", *extra], capture_output=True, text=True, env=env, cwd=env["HOME"], timeout=120) def _jid(res: subprocess.CompletedProcess) -> str: assert res.returncode == 0, res.stdout + res.stderr m = re.search(r"registered job:\s*(\S+)", res.stdout) assert m, res.stdout return m.group(1) def _record(reg: Path, jid: str) -> dict: return json.loads((reg / f"{jid}.json").read_text(encoding="utf-8")) def _publish(env: dict, reg: Path, jid: str, event: str, detail: str) -> None: """The real producer, with the dispatch token the agent is given. The broker is unreachable here: publish_event exits 2 after the record and the events log were already written.""" tok = _record(reg, jid)["dispatch_token"] res = subprocess.run([sys.executable, str(PUBLISH), "--registry-dir", str(reg), "--job", jid, "--event", event, "--detail", detail, "--expect-dispatch-token", tok], capture_output=True, text=True, env=env, timeout=120) assert res.returncode in (0, 2), res.stderr if event in ("completed", "error"): assert _record(reg, jid)["status"] in ("completed", "error"), res.stderr def _deadline_close(env: dict, reg: Path, jid: str) -> None: """What the wrapper's finalize_unfinished does at a job's deadline.""" tok = _record(reg, jid)["dispatch_token"] res = subprocess.run([sys.executable, str(REGISTRY), "--registry-dir", str(reg), "close", "--job", jid, "--request", "deadline_timeout", "--expect-dispatch-token", tok], capture_output=True, text=True, env=env, timeout=120) assert res.returncode == 0, res.stderr _publish(env, reg, jid, "error", "deadline_timeout") # --------------------------------------------------------------------------- # T1 - a job records its parent # --------------------------------------------------------------------------- def test_t1_submit_records_the_parent_job(tmp_path): env = _env(tmp_path) reg = tmp_path / "reg" p_res = _submit(env, reg) p = _jid(p_res) assert _record(reg, p)["parent_job"] is None # the injected instructions tell the agent how to pass itself on assert f"add --parent-job {p} so the orchestrator waits for it." in p_res.stdout c = _jid(_submit(env, reg, "--parent-job", p)) assert _record(reg, c)["parent_job"] == p bad = _submit(env, reg, "--parent-job", "nope1234") assert bad.returncode == 1 and "unknown parent job nope1234" in bad.stderr assert "registered job:" not in bad.stdout # the remote runner's MAM_JOB_ID is the default ... e = _jid(_submit(_env(tmp_path, {"MAM_JOB_ID": p}), reg)) assert _record(reg, e)["parent_job"] == p # ... ignored with a warning when this registry lacks it (another host's mirror) stray = _submit(_env(tmp_path, {"MAM_JOB_ID": "feedf00d"}), reg) assert _record(reg, _jid(stray))["parent_job"] is None assert "MAM_JOB_ID=feedf00d is not in" in stray.stderr # ... and the flag wins over it f = _jid(_submit(_env(tmp_path, {"MAM_JOB_ID": p}), reg, "--parent-job", c)) assert _record(reg, f)["parent_job"] == c listed = subprocess.run([sys.executable, str(REGISTRY), "--registry-dir", str(reg), "list", "--parent", p], capture_output=True, text=True, env=env, timeout=60) assert listed.returncode == 0 assert {line.split()[0] for line in listed.stdout.splitlines()} == {c, e} # --------------------------------------------------------------------------- # T2 / T7 - LoopRunner waits for the tree (real records, real events) # --------------------------------------------------------------------------- def _runner(tmp_path, monkeypatch, **env): monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path)) monkeypatch.setenv("MAM_JOBS_DIR", str(tmp_path / "reg")) monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / ".mam-logs")) monkeypatch.delenv("MAM_REAL_ROOT", raising=False) for k in ("MAM_CHILD_WAIT_S", "MAM_JOB_TREE_MAX"): monkeypatch.delenv(k, raising=False) for k, v in env.items(): monkeypatch.setenv(k, str(v)) from lib_py.loop import runner as lr return lr.LoopRunner(["--creator", "c", "--task", "t"]) def test_t2_the_runner_waits_for_children_and_grandchildren(tmp_path, monkeypatch): env = _env(tmp_path) reg = tmp_path / "reg" p = _jid(_submit(env, reg)) c = _jid(_submit(env, reg, "--parent-job", p)) g = _jid(_submit(env, reg, "--parent-job", c)) def finish(): time.sleep(2) _publish(env, reg, g, "completed", "grandchild done") time.sleep(2) _publish(env, reg, c, "completed", "child done") t = threading.Thread(target=finish) start = time.time() t.start() try: # the cap only keeps a broken helper thread from hanging the suite assert _runner(tmp_path, monkeypatch, MAM_CHILD_WAIT_S=60).wait_for_descendants(p, interval=0.5) is True elapsed = time.time() - start finally: t.join() assert elapsed >= 4 assert _record(reg, c)["status"] == "completed" and _record(reg, g)["status"] == "completed" def test_t2b_the_wait_is_bounded_in_time_and_in_size(tmp_path, monkeypatch, capsys): env = _env(tmp_path) reg = tmp_path / "reg" p = _jid(_submit(env, reg)) c = _jid(_submit(env, reg, "--parent-job", p)) g = _jid(_submit(env, reg, "--parent-job", c)) start = time.time() assert _runner(tmp_path, monkeypatch, MAM_CHILD_WAIT_S=2).wait_for_descendants(p, interval=0.5) is False assert 2 <= time.time() - start < 10 out = capsys.readouterr().out assert f"waiting for 2 sub-job(s) of {p}" in out and c in out and g in out assert "with 2 open" in out start = time.time() assert _runner(tmp_path, monkeypatch, MAM_JOB_TREE_MAX=1).wait_for_descendants(p, interval=0.5) is False assert time.time() - start < 2 assert "over MAM_JOB_TREE_MAX=1" in capsys.readouterr().err def test_t7_a_job_without_children_is_waited_for_as_before(tmp_path, monkeypatch): env = _env(tmp_path) reg = tmp_path / "reg" p = _jid(_submit(env, reg)) _publish(env, reg, p, "completed", "done") r = _runner(tmp_path, monkeypatch) start = time.time() assert r.wait_for_job_tree(p) is True assert time.time() - start < 2 # one registry scan on top of wait_for_job assert r.wait_for_job(p) is True # --------------------------------------------------------------------------- # Loop e2e: a real loop process; the loop's own agents are simulated. The # simulated creator registers its sub-job with the REAL wrapper. # --------------------------------------------------------------------------- FAKE_AGENT = r""" import json, os, re, secrets, subprocess, sys, time from pathlib import Path opts = dict(zip(sys.argv[2::2], sys.argv[3::2])) # argv[1] = "submit" prompt = Path(opts["--prompt-file"]).read_text(encoding="utf-8") jobs = Path(opts["--registry-dir"]) jobs.mkdir(parents=True, exist_ok=True) role = opts["--role"] now = lambda: time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) jid = "f" + secrets.token_hex(4) rec = {"job_id": jid, "status": "running", "created_at": now(), "updated_at": now(), "created_ts": time.time(), "agent_session": opts["--agent-session"], "role": role} path = jobs / (jid + ".json") path.write_text(json.dumps(rec)) late = False if role == "Worker": act = os.environ.get("FAKE_WORKER", "none") marker = Path(".mam/child-id") if act in ("child", "stray") and not marker.exists(): # the implementation job only cmd = ["bash", os.environ["FAKE_REAL_WRAPPER"], "submit", "--agent", "human", "--agent-session", "herdr:img", "--dry-run", "--registry-dir", str(jobs), "--prompt", "draw the figures"] if act == "child": cmd += ["--parent-job", jid] out = subprocess.run(cmd, capture_output=True, text=True) if out.returncode != 0: print(out.stdout + out.stderr, file=sys.stderr) sys.exit(2) marker.write_text(re.search(r"registered job:\s*(\S+)", out.stdout).group(1)) report = "done\n" else: counter = Path(".mam/review-count") n = int(counter.read_text()) + 1 if counter.exists() else 1 counter.write_text(str(n)) Path(".mam/review-prompt-%d.md" % n).write_text(prompt, encoding="utf-8") rule = os.environ.get("FAKE_REVIEW", "pass") unfinished = re.search(r"(열린|미확인 종결|실패) [1-9]", prompt) verdict = "NOT PASS" if rule == "rule" and unfinished else "PASS" if os.environ.get("FAKE_FIRST_NOT_PASS") and n == 1: # forces a correction round verdict = "NOT PASS" report = "reviewed\n\n[VERDICT: %s]\n" % verdict late = os.environ.get("FAKE_LATE_WRITE") == str(n) dest = jobs / jid / "agent-reports" dest.mkdir(parents=True) (dest / "report-final.md").write_text(report, encoding="utf-8") rec.update(status="completed", updated_at=now()) path.write_text(json.dumps(rec)) if late: # stands in for a sub-job agent that keeps writing after its deadline close Path("late.txt").write_text("written during review\n") print("registered job: " + jid) """ def _loop_ctl(tmp_path: Path, git: bool = True) -> Path: ctl = tmp_path / "ctl" if git: _repo(ctl) else: ctl.mkdir(parents=True) agent = ctl / "fake-agent.py" agent.write_text(FAKE_AGENT, encoding="utf-8") w = ctl / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job" w.parent.mkdir(parents=True, exist_ok=True) w.write_text(f'#!/usr/bin/env bash\nexec "{sys.executable}" "{agent}" "$@"\n', encoding="utf-8") w.chmod(0o755) (ctl / ".mam").mkdir(exist_ok=True) if git: (ctl / "plan.md").write_text("# plan\n\ndo it\n", encoding="utf-8") _git(ctl, "add", "-A") _git(ctl, "commit", "-q", "-m", "harness") else: (ctl / "plan.md").write_text("# plan\n\ndo it\n", encoding="utf-8") return ctl def _loop_env(ctl: Path, **extra) -> dict: return _env(ctl, extra={ "WORKSPACE_ROOT": ctl, "MAM_JOBS_DIR": ctl / ".mam" / "jobs", "DELEGATE_JOB_LOGS_DIR": ctl / ".mam" / "logs", "MAM_SESSION_BUSY_RETRY_S": "0", "MAM_CHILD_WAIT_S": "60", # a broken helper thread fails the test instead of hanging it "FAKE_REAL_WRAPPER": WRAPPER, "MAM_STATE_JSON": json.dumps({"herdr_sessions": [ {"name": "c", "agent": "claude", "role": "creator", "status": "running"}, {"name": "r", "agent": "claude", "role": "reviewer", "status": "running"}]}), **extra, }, drop=("MAM_LOOP_WORKTREE", "MAM_LOOP_BASE", "MAM_VERSION")) def _run_loop(ctl: Path, env: dict, *argv: str, on_child=None) -> subprocess.CompletedProcess: """Run the loop; on_child(child_id) runs in a thread once the simulated creator registered its sub-job.""" marker = ctl / ".mam" / "child-id" stop = threading.Event() seen = {} def watch(): while not stop.is_set(): if marker.exists() and marker.read_text().strip(): seen["id"] = marker.read_text().strip() on_child(seen["id"]) return time.sleep(0.2) t = threading.Thread(target=watch) if on_child else None if t: t.start() try: return subprocess.run([sys.executable, "-m", "lib_py.loop", "--creator", "c", "--task", "t34", "--reviewer", "r", "--plan-file", str(ctl / "plan.md"), *argv], capture_output=True, text=True, env=env, cwd=str(ctl), timeout=300) finally: stop.set() if t: t.join() def _review_prompt(ctl: Path, n: int = 1) -> str: return (ctl / ".mam" / f"review-prompt-{n}.md").read_text(encoding="utf-8") def _review_records(ctl: Path) -> list: recs = [json.loads(p.read_text()) for p in (ctl / ".mam" / "jobs").glob("f*.json")] return sorted((r for r in recs if r.get("role") == "Reviewer"), key=lambda r: r["created_ts"]) def test_t3_the_loop_waits_for_the_sub_job_before_review(tmp_path): ctl = _loop_ctl(tmp_path) env = _loop_env(ctl, FAKE_WORKER="child") jobs = ctl / ".mam" / "jobs" done = {} def finish_later(cid): time.sleep(5) done["ts"] = time.time() _publish(env, jobs, cid, "completed", "figures drawn") res = _run_loop(ctl, env, "--max-loop", "1", on_child=finish_later) out = res.stdout + res.stderr assert res.returncode == 0, out cid = (ctl / ".mam" / "child-id").read_text().strip() assert _record(jobs, cid)["parent_job"] # registered by the real wrapper, under the creator job assert "waiting for 1 sub-job(s)" in res.stdout # the loop registered the review only after the sub-job was completed assert _review_records(ctl)[0]["created_ts"] > done["ts"] prompt = _review_prompt(ctl) assert "[하위 위임 잡]" in prompt assert re.search(rf"\| {cid} \| herdr:img \| Worker \| completed \| figures drawn \| 1 \|", prompt), prompt assert "열린 0, 미확인 종결 0, 실패 0, 완료 1." in prompt assert "100% PASS" in res.stdout def test_t4_an_open_sub_job_is_evidence_and_the_review_blocks(tmp_path): ctl = _loop_ctl(tmp_path) env = _loop_env(ctl, FAKE_WORKER="child", FAKE_REVIEW="rule", MAM_CHILD_WAIT_S="3") res = _run_loop(ctl, env, "--max-loop", "1") out = res.stdout + res.stderr assert res.returncode == 1, out assert "with 1 open; reviewers see them as OPEN" in res.stdout cid = (ctl / ".mam" / "child-id").read_text().strip() prompt = _review_prompt(ctl) assert re.search(rf"\| {cid} \| .* \| OPEN \|", prompt) and "열린 1," in prompt assert "100% PASS" not in res.stdout # the success pair is T3: the same harness and rule pass once the child completed def test_t5_a_failed_sub_job_reaches_the_review_and_blocks_it(tmp_path): ctl = _loop_ctl(tmp_path) env = _loop_env(ctl, FAKE_WORKER="child", FAKE_REVIEW="rule") jobs = ctl / ".mam" / "jobs" res = _run_loop(ctl, env, "--max-loop", "1", on_child=lambda cid: _publish(env, jobs, cid, "error", "simulated")) out = res.stdout + res.stderr assert res.returncode == 1, out assert "Requesting code review" in res.stdout # a failed child does not stop the loop cid = (ctl / ".mam" / "child-id").read_text().strip() prompt = _review_prompt(ctl) assert re.search(rf"\| {cid} \| .* \| error \| simulated \|", prompt), prompt assert "실패 1," in prompt assert "100% PASS" not in res.stdout @pytest.mark.parametrize("late_write", [True, False]) def test_t5b_a_deadline_closed_sub_job_is_unconfirmed_and_a_late_write_voids_the_round(tmp_path, late_write): ctl = _loop_ctl(tmp_path) extra = {"FAKE_WORKER": "child", "FAKE_REVIEW": "pass"} if late_write: extra["FAKE_LATE_WRITE"] = "1" env = _loop_env(ctl, **extra) jobs = ctl / ".mam" / "jobs" res = _run_loop(ctl, env, "--max-loop", "1", on_child=lambda cid: _deadline_close(env, jobs, cid)) out = res.stdout + res.stderr cid = (ctl / ".mam" / "child-id").read_text().strip() assert _record(jobs, cid)["status"] == "error" prompt = _review_prompt(ctl) assert re.search(rf"\| {cid} \| .* \| CLOSED-UNCONFIRMED \| deadline_timeout \|", prompt), prompt assert "its agent may still be writing" in res.stdout if late_write: assert res.returncode == 1, out assert "source changed during review" in res.stdout assert "100% PASS" not in res.stdout else: # the guard reacts to a source change only assert res.returncode == 0, out assert "source changed" not in out assert "100% PASS" in res.stdout def test_t6_an_unattributed_open_job_is_shown_and_waited_for_only_on_request(tmp_path): ctl = _loop_ctl(tmp_path) env = _loop_env(ctl, FAKE_WORKER="stray") res = _run_loop(ctl, env, "--max-loop", "1") assert res.returncode == 0, res.stdout + res.stderr uid = (ctl / ".mam" / "child-id").read_text().strip() assert "sub-job(s) of" not in res.stdout # nothing to wait for by default prompt = _review_prompt(ctl) assert "[하위 위임 잡] 없음" in prompt assert re.search(rf"\[귀속되지 않은 열린 잡\] .*: {uid}\(herdr:img, pending\)", prompt), prompt ctl2 = _loop_ctl(tmp_path / "strict") res2 = _run_loop(ctl2, _loop_env(ctl2, FAKE_WORKER="stray", MAM_CHILD_WAIT_S="3"), "--max-loop", "1", "--wait-open-jobs") uid2 = (ctl2 / ".mam" / "child-id").read_text().strip() assert f"{uid2}(herdr:img, pending)" in res2.stdout and "waiting for 1 sub-job(s)" in res2.stdout assert "ended after 3s" in res2.stdout def test_t8_the_stale_review_guard_without_worktree_or_base(tmp_path): ctl = _loop_ctl(tmp_path) res = _run_loop(ctl, _loop_env(ctl, FAKE_LATE_WRITE="1"), "--max-loop", "2") out = res.stdout + res.stderr assert res.returncode == 0, out assert res.stdout.count("approvals discarded, re-reviewing") == 1 assert len(_review_records(ctl)) == 2 assert "100% PASS" in res.stdout # the control plane inside the source (.mam: jobs, logs, checkpoint) is not a change res_quiet = _run_loop(ctl, _loop_env(ctl), "--max-loop", "1") assert res_quiet.returncode == 0 and "source changed" not in res_quiet.stdout # no repository: no fingerprint, the guard is off and says so; behaviour as before plain = _loop_ctl(tmp_path / "plain", git=False) res_plain = _run_loop(plain, _loop_env(plain, FAKE_LATE_WRITE="1"), "--max-loop", "1") assert res_plain.returncode == 0, res_plain.stdout + res_plain.stderr assert "the stale-review guard is off for this run" in res_plain.stdout assert "100% PASS" in res_plain.stdout @pytest.mark.parametrize("child_end, rc", [("open", 1), ("error", 1), ("completed", 0)]) def test_t9_a_correction_round_keeps_the_earlier_creator_jobs_sub_jobs(tmp_path, child_end, rc): """Review of round 2 (after a correction job with no sub-job of its own) still sees the implementation job's sub-job; the correction does not claim to have replaced it. Success pair: the completed child passes in round 2.""" ctl = _loop_ctl(tmp_path) env = _loop_env(ctl, FAKE_WORKER="child", FAKE_REVIEW="rule", FAKE_FIRST_NOT_PASS="1", MAM_CHILD_WAIT_S="0") jobs = ctl / ".mam" / "jobs" on_child = {"open": None, "error": lambda cid: _publish(env, jobs, cid, "error", "simulated"), "completed": lambda cid: _publish(env, jobs, cid, "completed", "figures drawn")}[child_end] if child_end != "open": env["MAM_CHILD_WAIT_S"] = "60" # wait for the helper to close it res = _run_loop(ctl, env, "--max-loop", "2", on_child=on_child) out = res.stdout + res.stderr assert res.returncode == rc, out assert "Applying straight bugfixes" in res.stdout # a correction round ran cid = (ctl / ".mam" / "child-id").read_text().strip() second = _review_prompt(ctl, 2) state = {"open": "OPEN", "error": "error", "completed": "completed"}[child_end] assert re.search(rf"\| {cid} \| .* \| {state} \|", second), second assert ("100% PASS" in res.stdout) == (rc == 0)