diff --git a/.agents/skills/lib_py/loop/runner.py b/.agents/skills/lib_py/loop/runner.py index 2300e5d..9b1606c 100644 --- a/.agents/skills/lib_py/loop/runner.py +++ b/.agents/skills/lib_py/loop/runner.py @@ -24,9 +24,20 @@ import time from pathlib import Path from typing import Any, Dict, List, Optional, Set, Tuple +from lib_py.delegation import registry as job_registry from lib_py.loop import checkpoint, eager, parser UNAVAILABLE_MARKER = "!!! CHANGE SET UNAVAILABLE !!!" # diff_collect.sh MAM_DIFF_UNAVAILABLE_MARKER +# Issue #34: what a finished sub-job's record does and does not say. A job the +# orchestrator closed (deadline, no ack) says nothing about its agent, which is +# not interrupted and may still be writing; a job that never reached its agent +# was cancelled before injection. +TERMINAL = ("completed", "error", "cancelled") +ORCHESTRATOR_CLOSED = ("deadline_timeout", "agent_blocked") +SUBJOB_RULE = ("규칙: 열린(OPEN)·미확인 종결(CLOSED-UNCONFIRMED)·실패 하위 잡이 하나라도 있으면 산출물을 미완으로 보고 " + "[VERDICT: NOT PASS] 를 내십시오. creator 리포트가 해당 부분의 대체 완료를 입증하면 예외입니다. " + "CLOSED-UNCONFIRMED 는 오케스트레이터가 마감으로 닫은 잡이며 그 에이전트는 계속 일하고 있을 수 있습니다. " + "리뷰 중 소스가 바뀌면 이 회차의 승인은 무효가 됩니다.") SELF_REVIEW_PROMPT = ("작업 완료 상태에 대해 스스로 검증(Self-Review)하여 결함이 없음을 확인하고 종결해주세요. " "리뷰 리포트 마지막에 단독 행으로 반드시 '[VERDICT: PASS]' 혹은 '[VERDICT: NOT PASS]' 태그를 명시해주세요.") @@ -87,6 +98,9 @@ class LoopRunner: self._rounds: List[Dict[str, Any]] = [] self._accepted_round: Optional[Dict[str, Any]] = None self.created_jobs: List[str] = [] + # Issue #34: the creator jobs whose sub-job trees this round's reviewers see + self.round_roots: List[str] = [] + self.wait_open_jobs = False self.failed_submit_jobs: Set[str] = set() self.current_plan = "" self.plan_job_id: Optional[str] = None @@ -128,6 +142,7 @@ class LoopRunner: " --review-only Skip implementation; review ..HEAD (requires --base)", " --review-isolation Review tree with --worktree/--base: worktree (default) or env", " --remote-review Remote reviewers with --worktree/--base: refuse (default) or diff-only", + " --wait-open-jobs Also wait for open jobs registered without --parent-job during the creator's work", ] print("\n".join(lines)) @@ -202,6 +217,9 @@ class LoopRunner: elif arg == "--cleanup": self.cleanup = True i += 1 + elif arg == "--wait-open-jobs": + self.wait_open_jobs = True + i += 1 elif arg == "--worktree": if i + 1 >= len(args): self.print_usage() @@ -599,6 +617,113 @@ class LoopRunner: log_error(f"Job '{job_id}' timed out after {max_wait}s.") return False + # ------------------------------------------------------------------ + # Issue #34: a creator job's sub-jobs (registered with --parent-job) + # ------------------------------------------------------------------ + def _created_at(self, job_id: str) -> str: + try: + return str(job_registry.load_job(job_id, str(self.jobs_dir)).get("created_at") or "") + except (OSError, ValueError): + return "" + + def unattributed_open_jobs(self, since: str) -> List[Dict[str, Any]]: + """Open jobs registered without a parent since ``since`` that this loop + did not create - most likely a sub-delegation that left out --parent-job.""" + return [r for r in job_registry.list_jobs(str(self.jobs_dir)) + if r.get("status") in ("pending", "running") and not r.get("parent_job") + and r.get("job_id") not in self.created_jobs and str(r.get("created_at") or "") >= since] + + def wait_for_descendants(self, root: str, interval: float = 3.0) -> bool: + """Wait until every record in root's sub-job tree is terminal. The tree + is re-read on each poll, so a grandchild registered meanwhile is waited + for too. Bounded by MAM_CHILD_WAIT_S (from now) and MAM_JOB_TREE_MAX. + A terminal record is not a stopped agent (a deadline close does not + interrupt it); that is left to the sub-job table and the stale-review + guard, since nothing signals such an agent's end.""" + max_wait = float(os.environ.get("MAM_CHILD_WAIT_S", "3900")) + tree_max = int(os.environ.get("MAM_JOB_TREE_MAX", "50")) + since = self._created_at(root) + start = time.time() + last = None + while True: + if self.interrupted: + return False + tree = job_registry.descendants_of(root, str(self.jobs_dir)) + if len(tree) > tree_max: + log_error(f"job {root} has {len(tree)} sub-jobs (over MAM_JOB_TREE_MAX={tree_max}); not waiting for them") + return False + open_jobs = [r for r in tree if r.get("status") not in TERMINAL] + if self.wait_open_jobs: + open_jobs += self.unattributed_open_jobs(since) + if not open_jobs: + return True + now = [(r.get("job_id"), r.get("status")) for r in open_jobs] + if now != last: + log_info(f"waiting for {len(open_jobs)} sub-job(s) of {root}: " + + " ".join(f"{r.get('job_id')}({r.get('agent_session', '?')}, {r.get('status')})" for r in open_jobs)) + last = now + if time.time() - start >= max_wait: + log_warn(f"sub-job wait for {root} ended after {int(max_wait)}s (MAM_CHILD_WAIT_S) with " + f"{len(open_jobs)} open; reviewers see them as OPEN") + return False + time.sleep(interval) + + def wait_for_job_tree(self, root: str) -> bool: + """wait_for_job, then (when the root succeeded) its sub-job tree. The + result is the root's own: sub-job outcomes are review evidence.""" + if not self.wait_for_job(root): + return False + log_info(f"Creator job {root} finished; waiting for its sub-jobs...") + self.wait_for_descendants(root) + return True + + def _subjob_state(self, r: Dict[str, Any]) -> Tuple[str, str]: + """(state, reason) for the sub-job table.""" + status = str(r.get("status") or "?") + if status not in TERMINAL: + return "OPEN", "-" + reason = job_registry.get_feedback(str(r.get("job_id")), str(self.jobs_dir)) or "" + if status == "completed": + return "completed", reason or "-" + if reason in ORCHESTRATOR_CLOSED or (status == "error" and not reason): + # no recorded reason is not the agent's own report either + return "CLOSED-UNCONFIRMED", reason or "-" + if status == "cancelled" and (reason == "session_busy" or reason.startswith("rolled back")): + return "NOT-DISPATCHED", reason + return status, reason or "-" + + def _subjobs_block(self, roots: List[str]) -> str: + """The [하위 위임 잡] evidence put in front of this round's review prompts.""" + if not roots: + return "" + rows: List[str] = [] + counts = {"open": 0, "unconfirmed": 0, "failed": 0, "completed": 0} + for root in roots: + for r in job_registry.descendants_of(root, str(self.jobs_dir)): + state, reason = self._subjob_state(r) + key = ("open" if state == "OPEN" else "unconfirmed" if state == "CLOSED-UNCONFIRMED" + else "completed" if state == "completed" else "failed") + counts[key] += 1 + if state == "CLOSED-UNCONFIRMED": + log_warn(f"sub-job {r.get('job_id')} was closed by the orchestrator ({reason}); its agent may still " + f"be writing to {self.worktree_root} - this round's approvals are void if the source changes") + rows.append(f"| {r.get('job_id')} | {r.get('agent_session', '-')} | {r.get('role', '-')} | {state} | " + f"{reason} | {r.get('depth')} | {r.get('created_at', '-')} | {r.get('updated_at', '-')} |") + since = min((self._created_at(j) for j in roots), default="") + stray = self.unattributed_open_jobs(since) + stray_text = (" ".join(f"{r.get('job_id')}({r.get('agent_session', '?')}, {r.get('status')})" for r in stray) + if stray else "없음") + stray_line = f"[귀속되지 않은 열린 잡] (부모 미표기, 이 작업 중 등록됨): {stray_text}\n" + if not rows and not stray: + return "" # no sub-delegation: the prompts stay byte-identical (Issue #34 requirement 3) + if not rows: + return f"[하위 위임 잡] 없음 (creator 잡 {' '.join(roots)})\n{stray_line}\n" + return (f"[하위 위임 잡] (creator 잡 {' '.join(roots)} 의 자손, 깊이 순)\n" + "| job | 세션 | 역할 | 상태 | 사유 | 깊이 | 등록 | 갱신 |\n" + "|---|---|---|---|---|---|---|---|\n" + "\n".join(rows) + "\n" + f"열린 {counts['open']}, 미확인 종결 {counts['unconfirmed']}, 실패 {counts['failed']}, " + f"완료 {counts['completed']}.\n{SUBJOB_RULE}\n{stray_line}\n") + def find_report(self, job_id: str) -> Optional[Path]: """Locate report-final.md for a job.""" job_dir = self.jobs_dir / job_id @@ -763,11 +888,12 @@ class LoopRunner: "evidence (--remote-review diff-only)") return True - def _worktree_tree(self) -> Optional[str]: + def _worktree_tree(self, exclude: Tuple[str, ...] = ()) -> Optional[str]: """The source worktree's state as a tree, through a temporary index (the real index, status and refs are untouched); None on failure. A dirty submodule is a failure: the tree records only its gitlink, so - its uncommitted content would be neither reviewed nor pinned.""" + its uncommitted content would be neither reviewed nor pinned. + exclude: source-relative paths left out of `add -A` (Issue #34).""" dirty_subs = self._dirty_submodules() if dirty_subs is None: log_error(f"review snapshot: cannot check the submodules of {self.worktree_root}") @@ -781,7 +907,8 @@ class LoopRunner: os.unlink(idx) env = dict(os.environ, GIT_INDEX_FILE=idx) try: - for step in (("read-tree", "HEAD"), ("add", "-A"), ("write-tree",)): + add = ("add", "-A") + (("--", ".") + tuple(f":(exclude){p}" for p in exclude) if exclude else ()) + for step in (("read-tree", "HEAD"), add, ("write-tree",)): r = subprocess.run(["git", "-C", str(self.worktree_root), *step], capture_output=True, text=True, env=env) if r.returncode != 0: @@ -822,6 +949,27 @@ class LoopRunner: def _identity_now(self) -> Tuple[str, str]: return (self._source_head(), self._worktree_tree() or "") + def _fingerprint(self) -> Optional[Tuple[str, str]]: + """Issue #34 (D9): the source's head and tree for the stale-review guard + without --worktree/--base, where the control plane (.mam, the jobs and + audit-log dirs) usually lives inside the source and changes on every + job; it is left out. None when the source cannot be fingerprinted.""" + head = self._source_head() + if not head: + return None + root = self.worktree_root.resolve() + logs = Path(os.environ.get("DELEGATE_JOB_LOGS_DIR") or self.mam_dir / "delegate_job_logs") + exclude = [] + for d in (self.mam_dir, self.jobs_dir, self.workspace_root / logs): + try: + rel = d.resolve().relative_to(root) + except ValueError: + continue + if str(rel) != ".": + exclude.append(str(rel)) + tree = self._worktree_tree(tuple(exclude)) + return (head, tree) if tree else None + def _snapshot(self, n: int) -> Optional[Dict[str, Any]]: """B3: pin the source state of review round n. Uncommitted and untracked (not ignored) files go into a ref-less snapshot commit on @@ -1137,7 +1285,8 @@ class LoopRunner: return 1 log_info(f"Creator Job ID: {exec_jid}") - if not self.wait_for_job(exec_jid): + self.round_roots = [exec_jid] + if not self.wait_for_job_tree(exec_jid): log_error("Creator execution failed.") return 1 log_success("Initial implementation finished.") @@ -1160,6 +1309,8 @@ class LoopRunner: final_passed_reviewers: Dict[str, Tuple[str, Path]] = {} identity = self._context_active() + light: Optional[Tuple[str, str]] = None # Issue #34: this round's source without --worktree/--base + guard_warned = False # Identity mode: a creator self-review is a verdict like any other, so it # takes the reviewer path - snapshot diff, evidence check, B10 re-check # and sidecar (loop B review B-R2/B-R3). A NOT PASS re-reviews, as before. @@ -1189,17 +1340,32 @@ class LoopRunner: return 1 else: review_path = str(self.worktree_root) + else: + # Issue #34 (D9): the identity mode's B7 for every loop - a sub-job + # closed by deadline may still be writing to the source + light = self._fingerprint() + if light is None and not guard_warned: + log_warn(f"cannot fingerprint the source at {self.worktree_root}; " + "the stale-review guard is off for this run") + guard_warned = True + subjobs = self._subjobs_block(self.round_roots) if not reviewers: # Creator Self-Review log_warn("No reviewers specified. Conducting Creator Self-Review...") - sr_prompt = SELF_REVIEW_PROMPT + sr_prompt = subjobs + SELF_REVIEW_PROMPT sr_jid = self._dispatch(self.creator, creator_type, "Reviewer", sr_prompt) if not sr_jid: log_error("Failed to register Self-Review job.") return 1 self.wait_for_job(sr_jid) sr_report = self.find_report(sr_jid) + if light is not None and self._fingerprint() != light: + log_warn("source changed during review; approvals discarded, re-reviewing") + if loop_count == self.max_loop: + log_error("the source changed during review in the last iteration; no approval to accept") + return 1 + continue if sr_report and parser.has_verdict(sr_report, "PASS") and not parser.has_verdict(sr_report, "NOT PASS"): log_success("Self-Review PASS.") break @@ -1230,6 +1396,7 @@ class LoopRunner: else: r_prompt = f"다음 구현 사항(작업 목표: {self.task}) 및 누적 변경분(git diff)에 대해 린트, 동작성, 유실 등의 관점에서 교차 코드 리뷰를 수행해주세요. 확인 후 최종 Verdict로 '[VERDICT: PASS]' 혹은 '[VERDICT: NOT PASS]' 태그를 리뷰 리포트 마지막에 단독 행으로 명시적으로 작성해주세요. 만약 단순 버그 수정으로는 부족하고 설계 변경/재작업 수준의 재계획이 필요하다고 판단되면, 리포트 아무 곳에나 단독 행으로 '[ESCALATE: PLANNER]' 태그도 함께 남겨주세요. 변경분:\n{changes_diff}" r_remote = rev in self._remote_recipients + r_prompt = subjobs + r_prompt if identity: r_prompt = self._identity_block(snap, review_path, r_remote) + r_prompt r_jid = self._dispatch(rev, r_type, "Reviewer", r_prompt, remote=r_remote, @@ -1288,6 +1455,14 @@ class LoopRunner: log_error("the source changed during review in the last iteration; no approval to accept") return 1 continue + elif light is not None and self._fingerprint() != light: + final_passed_reviewers.clear() + log_warn(f"source changed during review (head {light[0][:12]}, tree {light[1][:12]}); " + "approvals discarded, re-reviewing") + if loop_count == self.max_loop: + log_error("the source changed during review in the last iteration; no approval to accept") + return 1 + continue if all_passed: if identity: @@ -1335,7 +1510,9 @@ class LoopRunner: correction_prompt += f"\n\n[이의제기 채널] 위 피드백 중 타당하지 않다고 판단되는 항목이 있다면, 그 항목은 구현하지 말고 반론을 제기하십시오. 반론 시 리포트에 단독 행으로 '[REBUT: <리뷰어_세션명>]' 태그를 남기고, 해당 리뷰어의 지적 중 어느 항목을 왜 거부하는지 근거를 함께 서술하십시오. 타당한 지적은 정상적으로 반영하십시오. 반론할 항목이 없으면 태그를 남기지 마십시오. 대상 리뷰어 세션명: {' '.join(fail_revs) if fail_revs else '(없음)'}" correct_jid = self._dispatch(self.creator, creator_type, "Worker", correction_prompt) - if not correct_jid or not self.wait_for_job(correct_jid): + if correct_jid: + self.round_roots = [correct_jid] + if not correct_jid or not self.wait_for_job_tree(correct_jid): log_error("Failed to register Creator correction job.") return 1 @@ -1404,10 +1581,17 @@ class LoopRunner: rebut_budget = 0 comply_prompt = f"귀하의 반론은 재심 결과 기각되었습니다. 더 이상 반론하지 말고 다음 리뷰 피드백을 그대로 반영하여 코드를 수정하십시오. 피드백:\n{feedback_aggregate}" comp_jid = self._dispatch(self.creator, creator_type, "Worker", comply_prompt) - if not comp_jid or not self.wait_for_job(comp_jid): + if comp_jid: + self.round_roots.append(comp_jid) + if not comp_jid or not self.wait_for_job_tree(comp_jid): log_error("Failed to register Creator compliance job.") return 1 + if light is not None and self._fingerprint() != light: + # Issue #34: B10 without --worktree/--base, before anything is promoted + log_error("source changed after approval; not promoting") + return 1 + # Promotion if self.plan_mode and self.current_plan and self.plan_job_id: plan_dest_dir = self.workspace_root / ".agents" / "reports" / planner_session diff --git a/.agents/skills/multi-agent-mux-loop/scripts/run_loop.sh b/.agents/skills/multi-agent-mux-loop/scripts/run_loop.sh index 437a5f8..5b2f2d7 100644 --- a/.agents/skills/multi-agent-mux-loop/scripts/run_loop.sh +++ b/.agents/skills/multi-agent-mux-loop/scripts/run_loop.sh @@ -38,6 +38,7 @@ LOOP_BASE="" LOOP_REVIEW_ONLY=0 LOOP_REVIEW_ISOLATION="" LOOP_REMOTE_REVIEW="" +LOOP_WAIT_OPEN_JOBS=0 # Print usage instructions usage() { @@ -59,6 +60,7 @@ usage() { echo " --review-only Skip implementation; review ..HEAD (requires --base)" echo " --review-isolation Review tree with --worktree/--base: worktree (default) or env" echo " --remote-review Remote reviewers with --worktree/--base: refuse (default) or diff-only" + echo " --wait-open-jobs Also wait for open jobs registered without --parent-job during the creator's work" exit 1 } @@ -101,6 +103,7 @@ while [[ "$#" -gt 0 ]]; do --review-only) LOOP_REVIEW_ONLY=1; shift ;; --review-isolation) LOOP_REVIEW_ISOLATION="$2"; shift 2 ;; --remote-review) LOOP_REMOTE_REVIEW="$2"; shift 2 ;; + --wait-open-jobs) LOOP_WAIT_OPEN_JOBS=1; shift ;; -h|--help) usage ;; *) echo "Unknown option: $1"; usage ;; esac @@ -130,8 +133,8 @@ fi if [ "${MAM_LOOP_PY:-1}" = "0" ] && { [ -n "$LOOP_WORKTREE" ] || [ -n "$LOOP_BASE" ] \ || [ "$LOOP_REVIEW_ONLY" = "1" ] || [ -n "${MAM_LOOP_WORKTREE:-}" ] \ || [ -n "$LOOP_REVIEW_ISOLATION" ] || [ -n "$LOOP_REMOTE_REVIEW" ] \ - || [ -n "$LOOP_PLAN_FILE" ]; }; then - echo "ERROR: --worktree/--base/--review-only/--review-isolation/--remote-review/--plan-file require the Python loop (MAM_LOOP_PY=1)" >&2 + || [ -n "$LOOP_PLAN_FILE" ] || [ "$LOOP_WAIT_OPEN_JOBS" = "1" ]; }; then + echo "ERROR: --worktree/--base/--review-only/--review-isolation/--remote-review/--plan-file/--wait-open-jobs require the Python loop (MAM_LOOP_PY=1)" >&2 exit 1 fi diff --git a/tests/test_issue34_job_tree.py b/tests/test_issue34_job_tree.py new file mode 100644 index 0000000..2a399af --- /dev/null +++ b/tests/test_issue34_job_tree.py @@ -0,0 +1,485 @@ +"""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" + 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