GodopuandClaude Opus 5
e551e3baf1
The --plan-file check landed on the one path a human is watching. The three --plan-mode loads read a report written by another agent with nobody watching, and had no content check at all: find_report only tests that the path exists. An empty or whitespace-only report therefore blanked current_plan in silence. Phase 2's `if self.current_plan` then flipped the prompt to self-planning and the run finished rc 0, having paid for a planner job, a Creator challenge and a refine job and thrown the result away. The escalation refine was worse: it has no guard, so the correction prompt asserted "follow the updated plan below" and carried nothing. All four loads share _plan_text() now - a plan is a readable regular file with text in it - and each caller says what None means. An empty initial plan stops the run instead of demoting it, and an empty refine warns and keeps the plan it already has, which is still good. Those two together leave the escalation prompt unable to claim a plan it does not carry, so it needs no new branch. Found by the Creator's challenge to the plan (job 0f63ac1d), sustained by the Planner as revision 1. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1424 lines
73 KiB
Python
1424 lines
73 KiB
Python
"""Autonomous Planning, Execution, and Peer-Review Orchestrator.
|
|
|
|
Issue #5 Phase 3: Python port of run_loop.sh into lib_py.loop.runner.
|
|
Preserves public CLI contracts, exit codes, prompt shapes, and state transitions
|
|
while using fence-aware strict verdict parsing and atomic checkpointing.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import atexit
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import shlex
|
|
import shutil
|
|
import signal
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Set, Tuple
|
|
|
|
from lib_py.loop import checkpoint, eager, parser
|
|
|
|
UNAVAILABLE_MARKER = "!!! CHANGE SET UNAVAILABLE !!!" # diff_collect.sh MAM_DIFF_UNAVAILABLE_MARKER
|
|
SELF_REVIEW_PROMPT = ("작업 완료 상태에 대해 스스로 검증(Self-Review)하여 결함이 없음을 확인하고 종결해주세요. "
|
|
"리뷰 리포트 마지막에 단독 행으로 반드시 '[VERDICT: PASS]' 혹은 '[VERDICT: NOT PASS]' 태그를 명시해주세요.")
|
|
|
|
# ANSI colors matching run_loop.sh
|
|
_COLOR_INFO = "\033[1;34m[*]\033[0m"
|
|
_COLOR_SUCCESS = "\033[1;32m[✓]\033[0m"
|
|
_COLOR_WARN = "\033[1;33m[!]\033[0m"
|
|
_COLOR_ERROR = "\033[1;31m[✗]\033[0m"
|
|
|
|
|
|
def log_info(msg: str) -> None:
|
|
print(f"{_COLOR_INFO} {msg}")
|
|
|
|
|
|
def log_success(msg: str) -> None:
|
|
print(f"{_COLOR_SUCCESS} {msg}")
|
|
|
|
|
|
def log_warn(msg: str) -> None:
|
|
print(f"{_COLOR_WARN} {msg}")
|
|
|
|
|
|
def log_error(msg: str) -> None:
|
|
print(f"{_COLOR_ERROR} {msg}", file=sys.stderr)
|
|
|
|
|
|
class LoopRunner:
|
|
"""Orchestrates planning, code implementation, and multi-agent peer reviews."""
|
|
|
|
def __init__(self, argv: Optional[List[str]] = None):
|
|
# Eagerly load all required framework modules to ensure immutable runtime (Phase 4)
|
|
eager.eager_load()
|
|
|
|
self.raw_argv = list(argv) if argv is not None else sys.argv[1:]
|
|
self.workspace_root = Path(os.environ.get("MAM_REAL_ROOT") or os.environ.get("WORKSPACE_ROOT") or Path.cwd()).resolve()
|
|
self.mam_dir = self.workspace_root / ".mam"
|
|
self.checkpoint_file = self.mam_dir / "loop-checkpoint.json"
|
|
# Issue #21: the jobs dir honours MAM_JOBS_DIR (and the wrapper is told
|
|
# the same dir explicitly). The source root (--worktree) is where the
|
|
# diff, base and head come from; everything else - wrapper, registry,
|
|
# checkpoint, loop guard, promotion, cleanup - stays on the control
|
|
# root (workspace_root).
|
|
# A relative MAM_JOBS_DIR is taken against the control root, as the
|
|
# wrapper (run with cwd=workspace_root) takes it.
|
|
self.jobs_dir = (self.workspace_root / Path(os.environ.get("MAM_JOBS_DIR") or self.mam_dir / "jobs")).resolve()
|
|
self.worktree_root: Path = self.workspace_root
|
|
self.worktree_given = False
|
|
self.base_ref: Optional[str] = None
|
|
self.base_sha = ""
|
|
self.base_source = "head"
|
|
self.review_only = False
|
|
# Issue #21 item 5: review identity (active with --worktree/--base)
|
|
self.loop_id = "loop-" + time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) + "-" + secrets.token_hex(3)
|
|
self.review_isolation = "worktree"
|
|
self.remote_review = "refuse"
|
|
self._remote_recipients: Set[str] = set()
|
|
self._review_paths: List[str] = []
|
|
self._rounds: List[Dict[str, Any]] = []
|
|
self._accepted_round: Optional[Dict[str, Any]] = None
|
|
self.created_jobs: List[str] = []
|
|
self.failed_submit_jobs: Set[str] = set()
|
|
self.current_plan = ""
|
|
self.plan_job_id: Optional[str] = None
|
|
self.base_commit = ""
|
|
self.interrupted = False
|
|
|
|
# Options populated by parse_args
|
|
self.creator: str = ""
|
|
self.task: str = ""
|
|
self.plan_mode: bool = False
|
|
self.plan_file: Optional[str] = None
|
|
self.planner_override: Optional[str] = None
|
|
self.plan_talk_turns: int = 1
|
|
self.reviewer_list: Optional[str] = None
|
|
self.all_reviewers: bool = False
|
|
self.max_loop: int = 3
|
|
self.max_rebut: int = 1
|
|
self.verbose: bool = False
|
|
self.cleanup: bool = False
|
|
|
|
def print_usage(self) -> None:
|
|
prog = "run_loop.sh"
|
|
lines = [
|
|
f"Usage: {prog} [options] --creator <agent-session-name> --task <goal-text>",
|
|
"Options:",
|
|
" --creator <name> Creator session that implements the task (required)",
|
|
" --plan Enable Planner agent intervention & design phase",
|
|
" --planner <name> Planner session (requires --plan; default: auto-resolve)",
|
|
" --plan-talk N Planner-Creator discussion limit turns (default: 1)",
|
|
" --plan-file <path> Reuse an existing plan file, relative to the control root (not with --plan)",
|
|
" --reviewer \"A,B\" Targeted reviewer session name list (comma-separated)",
|
|
" --all-reviewer Enforce PASS verdict from all active reviewer sessions",
|
|
" --max-loop N Max execution-review corrective loop runs (default: 3)",
|
|
" --max-rebut N Max rebuttal attempts per review iteration (default: 1, 0: disabled)",
|
|
" --verbose Print detailed execution timeline traces",
|
|
" --cleanup Purge temporary job directories upon success",
|
|
" --worktree <path> Source worktree for diff/base/head (control plane stays here)",
|
|
" --base <ref> Pin the review diff base (must be an ancestor of the source HEAD)",
|
|
" --review-only Skip implementation; review <base>..HEAD (requires --base)",
|
|
" --review-isolation <v> Review tree with --worktree/--base: worktree (default) or env",
|
|
" --remote-review <v> Remote reviewers with --worktree/--base: refuse (default) or diff-only",
|
|
]
|
|
print("\n".join(lines))
|
|
|
|
def parse_args(self) -> None:
|
|
args = list(self.raw_argv)
|
|
i = 0
|
|
while i < len(args):
|
|
arg = args[i]
|
|
if arg in ("-h", "--help"):
|
|
self.print_usage()
|
|
sys.exit(1)
|
|
elif arg == "--creator":
|
|
if i + 1 >= len(args):
|
|
self.print_usage()
|
|
sys.exit(1)
|
|
self.creator = args[i + 1]
|
|
i += 2
|
|
elif arg == "--target-agent":
|
|
print("ERROR: --target-agent was removed. Use --creator <session> instead.", file=sys.stderr)
|
|
sys.exit(1)
|
|
elif arg == "--task":
|
|
if i + 1 >= len(args):
|
|
self.print_usage()
|
|
sys.exit(1)
|
|
self.task = args[i + 1]
|
|
i += 2
|
|
elif arg == "--plan":
|
|
self.plan_mode = True
|
|
i += 1
|
|
elif arg == "--planner":
|
|
if i + 1 >= len(args):
|
|
self.print_usage()
|
|
sys.exit(1)
|
|
self.planner_override = args[i + 1]
|
|
i += 2
|
|
elif arg == "--plan-file":
|
|
if i + 1 >= len(args):
|
|
self.print_usage()
|
|
sys.exit(1)
|
|
self.plan_file = args[i + 1]
|
|
i += 2
|
|
elif arg == "--plan-talk":
|
|
if i + 1 >= len(args) or not re.match(r"^[0-9]+$", args[i + 1]):
|
|
print("ERROR: --plan-talk requires a positive integer.", file=sys.stderr)
|
|
sys.exit(1)
|
|
self.plan_talk_turns = int(args[i + 1])
|
|
i += 2
|
|
elif arg == "--reviewer":
|
|
if i + 1 >= len(args):
|
|
self.print_usage()
|
|
sys.exit(1)
|
|
self.reviewer_list = args[i + 1]
|
|
i += 2
|
|
elif arg == "--all-reviewer":
|
|
self.all_reviewers = True
|
|
i += 1
|
|
elif arg == "--max-loop":
|
|
if i + 1 >= len(args) or not re.match(r"^[0-9]+$", args[i + 1]) or int(args[i + 1]) <= 0:
|
|
print("ERROR: --max-loop requires a positive non-zero integer.", file=sys.stderr)
|
|
sys.exit(1)
|
|
self.max_loop = int(args[i + 1])
|
|
i += 2
|
|
elif arg == "--max-rebut":
|
|
if i + 1 >= len(args) or not re.match(r"^[0-9]+$", args[i + 1]):
|
|
print("ERROR: --max-rebut requires a non-negative integer.", file=sys.stderr)
|
|
sys.exit(1)
|
|
self.max_rebut = int(args[i + 1])
|
|
i += 2
|
|
elif arg == "--verbose":
|
|
self.verbose = True
|
|
i += 1
|
|
elif arg == "--cleanup":
|
|
self.cleanup = True
|
|
i += 1
|
|
elif arg == "--worktree":
|
|
if i + 1 >= len(args):
|
|
self.print_usage()
|
|
sys.exit(1)
|
|
self.worktree_root = Path(args[i + 1])
|
|
self.worktree_given = True
|
|
i += 2
|
|
elif arg == "--base":
|
|
if i + 1 >= len(args):
|
|
self.print_usage()
|
|
sys.exit(1)
|
|
self.base_ref = args[i + 1]
|
|
i += 2
|
|
elif arg == "--review-only":
|
|
self.review_only = True
|
|
i += 1
|
|
elif arg in ("--review-isolation", "--remote-review"):
|
|
if i + 1 >= len(args):
|
|
self.print_usage()
|
|
sys.exit(1)
|
|
allowed = ("worktree", "env") if arg == "--review-isolation" else ("refuse", "diff-only")
|
|
if args[i + 1] not in allowed:
|
|
print(f"ERROR: {arg} must be one of: {', '.join(allowed)}.", file=sys.stderr)
|
|
sys.exit(1)
|
|
if arg == "--review-isolation":
|
|
self.review_isolation = args[i + 1]
|
|
else:
|
|
self.remote_review = args[i + 1]
|
|
i += 2
|
|
else:
|
|
print(f"Unknown option: {arg}")
|
|
self.print_usage()
|
|
sys.exit(1)
|
|
|
|
if not self.creator or not self.task:
|
|
print("ERROR: --creator and --task are mandatory fields.", file=sys.stderr)
|
|
self.print_usage()
|
|
sys.exit(1)
|
|
|
|
if self.planner_override and not self.plan_mode:
|
|
print("ERROR: --planner was specified without --plan.", file=sys.stderr)
|
|
print("To enable planning, include the --plan flag:", file=sys.stderr)
|
|
print(f' run_loop.sh --creator <creator> --plan --planner <planner> --task "..."', file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
# Issue #25: a promoted plan is reused only when it is named (D2). The
|
|
# file is read here, before a single job exists, so a bad path cannot
|
|
# surface halfway through a run (D3).
|
|
if self.plan_file is not None:
|
|
if self.plan_mode:
|
|
print("ERROR: --plan-file cannot be combined with --plan.", file=sys.stderr)
|
|
print("Use --plan to have the Planner write a new plan, or --plan-file to reuse one:", file=sys.stderr)
|
|
print(' run_loop.sh --creator <creator> --plan-file <path> --task "..."', file=sys.stderr)
|
|
sys.exit(1)
|
|
resolved = Path(self.plan_file).expanduser().resolve()
|
|
text = self._plan_text(resolved)
|
|
if text is None:
|
|
log_error(f"--plan-file {resolved}: not a readable file with plan text in it.")
|
|
sys.exit(1)
|
|
self.plan_file = str(resolved)
|
|
if self.review_only:
|
|
# D6: --review-only delegates no implementation job, so there is
|
|
# nothing for a plan to steer.
|
|
log_warn(f"--plan-file is ignored with --review-only: {resolved}")
|
|
else:
|
|
self.current_plan = text
|
|
|
|
if self.all_reviewers and self.reviewer_list:
|
|
log_warn(f"--all-reviewer takes precedence; ignoring --reviewer list ('{self.reviewer_list}').")
|
|
|
|
if self.plan_talk_turns > 0 and not self.plan_mode:
|
|
log_warn("--plan-talk was specified but --plan mode is not enabled. Discussion turns will be ignored.")
|
|
|
|
# Issue #21 (D3): CLI --worktree wins over MAM_LOOP_WORKTREE. --base has
|
|
# no env fallback (D17): a stale sha left in a shell would still pass
|
|
# the ancestor check and silently widen the next loop's review range.
|
|
if not self.worktree_given and os.environ.get("MAM_LOOP_WORKTREE", "").strip():
|
|
self.worktree_root = Path(os.environ["MAM_LOOP_WORKTREE"].strip())
|
|
self.worktree_given = True
|
|
if self.worktree_given:
|
|
self.worktree_root = self._validate_worktree(self.worktree_root)
|
|
if self.review_only and self.base_ref is None:
|
|
print("ERROR: --review-only requires --base <ref>.", file=sys.stderr)
|
|
sys.exit(1)
|
|
if self.review_only and self.plan_mode:
|
|
print("ERROR: --review-only cannot be combined with --plan.", file=sys.stderr)
|
|
sys.exit(1)
|
|
if self.base_ref is not None:
|
|
self.base_sha = self._resolve_base(self.base_ref)
|
|
self.base_source = "cli"
|
|
if not self._context_active() and any(a in self.raw_argv for a in ("--review-isolation", "--remote-review")):
|
|
log_warn("--review-isolation/--remote-review only apply with --worktree or --base; ignored.")
|
|
|
|
@staticmethod
|
|
def _plan_text(path: Path) -> Optional[str]:
|
|
"""Issue #25 (D3): the one check every plan load shares - a plan is a
|
|
readable regular file that has text in it. None means 'no plan here';
|
|
each caller decides what that means (D13 stops the run, D14 keeps the
|
|
plan it already has). A file that cannot be read is not told apart from
|
|
an empty one: either way the loop would carry no plan text."""
|
|
try:
|
|
if not path.is_file():
|
|
return None
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|
except OSError:
|
|
return None
|
|
return text if text.strip() else None
|
|
|
|
@staticmethod
|
|
def _git(cwd: Path, *args: str) -> subprocess.CompletedProcess:
|
|
return subprocess.run(["git", "-C", str(cwd), *args], capture_output=True, text=True)
|
|
|
|
def _git_common_dir(self, cwd: Path) -> str:
|
|
res = self._git(cwd, "rev-parse", "--git-common-dir")
|
|
if res.returncode != 0 or not res.stdout.strip():
|
|
return ""
|
|
d = res.stdout.strip()
|
|
# the main checkout answers ".git", relative to cwd (not to this process)
|
|
return os.path.realpath(d if os.path.isabs(d) else os.path.join(str(cwd), d))
|
|
|
|
def _validate_worktree(self, p: Path) -> Path:
|
|
p = Path(os.path.abspath(str(p)))
|
|
if not p.is_dir():
|
|
log_error(f"--worktree {p}: no such directory.")
|
|
sys.exit(1)
|
|
top = self._git(p, "rev-parse", "--show-toplevel")
|
|
if top.returncode != 0 or not top.stdout.strip():
|
|
log_error(f"--worktree {p}: not a git working tree.")
|
|
sys.exit(1)
|
|
if os.path.realpath(top.stdout.strip()) != os.path.realpath(p):
|
|
log_error(f"--worktree {p}: not the top of its working tree ({top.stdout.strip()}).")
|
|
sys.exit(1)
|
|
mine, ctl = self._git_common_dir(p), self._git_common_dir(self.workspace_root)
|
|
if not mine or mine != ctl:
|
|
log_error(f"--worktree {p}: not a worktree of the repository at {self.workspace_root}.")
|
|
sys.exit(1)
|
|
return Path(os.path.realpath(p))
|
|
|
|
def _resolve_base(self, ref: str) -> str:
|
|
"""--base (D14): a full sha that is an ancestor of the source HEAD."""
|
|
res = self._git(self.worktree_root, "rev-parse", "--verify", "--quiet", f"{ref}^{{commit}}")
|
|
sha = res.stdout.strip() if res.returncode == 0 else ""
|
|
if not sha:
|
|
log_error(f"--base {ref}: not a commit in {self.worktree_root}.")
|
|
sys.exit(1)
|
|
if self._git(self.worktree_root, "merge-base", "--is-ancestor", sha, "HEAD").returncode != 0:
|
|
log_error(f"--base {ref} ({sha[:12]}) is not an ancestor of HEAD in {self.worktree_root}; "
|
|
"refusing to review an unrelated range.")
|
|
sys.exit(1)
|
|
return sha
|
|
|
|
def load_sessions_state(self) -> Dict[str, Any]:
|
|
"""Load registered sessions from MAM_STATE_JSON or agent-sessions.db/yaml."""
|
|
raw_env = os.environ.get("MAM_STATE_JSON")
|
|
if raw_env:
|
|
try:
|
|
return json.loads(raw_env)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
yaml_path = os.environ.get("AGENT_SESSIONS_YAML")
|
|
if not yaml_path:
|
|
yaml_path = str(self.mam_dir / "agent-sessions.yaml")
|
|
|
|
db_path = str(Path(yaml_path).with_suffix(".db"))
|
|
d: Dict[str, Any] = {}
|
|
if os.path.exists(db_path):
|
|
try:
|
|
conn = sqlite3.connect(db_path, timeout=10.0)
|
|
cursor = conn.execute("SELECT data FROM sessions")
|
|
sessions = [json.loads(row[0]) for row in cursor.fetchall()]
|
|
d["herdr_sessions"] = sessions
|
|
conn.close()
|
|
return d
|
|
except Exception:
|
|
pass
|
|
|
|
if os.path.exists(yaml_path):
|
|
try:
|
|
import yaml
|
|
with open(yaml_path, "r", encoding="utf-8") as f:
|
|
return yaml.safe_load(f) or {}
|
|
except Exception:
|
|
pass
|
|
|
|
return {}
|
|
|
|
def resolve_agent_type(self, name: str, state: Dict[str, Any]) -> str:
|
|
"""Resolve agent type (claude, hermes, agy, grok, opencode, codex)."""
|
|
for s in state.get("herdr_sessions", []):
|
|
if s.get("name") == name:
|
|
agent = s.get("agent") or (s.get("pane") or {}).get("cmd")
|
|
if agent:
|
|
return str(agent)
|
|
break
|
|
|
|
segments = name.split("-")
|
|
if "agy" in segments:
|
|
return "agy"
|
|
elif "hermes" in segments:
|
|
return "hermes"
|
|
elif "grok" in segments:
|
|
return "grok"
|
|
elif "opencode" in segments:
|
|
return "opencode"
|
|
elif "codex" in segments:
|
|
return "codex"
|
|
return "claude"
|
|
|
|
def session_label(self, name: str) -> str:
|
|
"""`herdr:<name>` for a pane-backed session, `remote:<name>` for a row
|
|
registered with `transport: remote` (Issue #18: the prompt then goes
|
|
over the broker and no herdr/pane/lsof/ps path is taken)."""
|
|
state = getattr(self, "_sessions_state", None)
|
|
if state is None:
|
|
state = self.load_sessions_state()
|
|
self._sessions_state = state
|
|
for s in state.get("herdr_sessions", []):
|
|
if s.get("name") == name and str(s.get("transport") or "").lower() == "remote":
|
|
return f"remote:{name}"
|
|
return f"herdr:{name}"
|
|
|
|
def contract_args(self) -> List[str]:
|
|
"""Liveness-contract knobs passed through to every delegate job
|
|
(Issue #18): MAM_HEARTBEAT_SEC (N), MAM_HEARTBEAT_MISSES (K),
|
|
MAM_ACK_TIMEOUT_S (T). The wrapper derives --idle-timeout as K x N."""
|
|
args: List[str] = []
|
|
for env_name, flag in (("MAM_HEARTBEAT_SEC", "--heartbeat"),
|
|
("MAM_HEARTBEAT_MISSES", "--heartbeat-misses"),
|
|
("MAM_ACK_TIMEOUT_S", "--ack-timeout")):
|
|
val = os.environ.get(env_name)
|
|
if val and val.isdigit() and int(val) > 0:
|
|
args += [flag, val]
|
|
return args
|
|
|
|
def resolve_all_reviewers(self, state: Dict[str, Any]) -> List[str]:
|
|
"""Resolve running reviewer sessions excluding creator."""
|
|
reviewers = []
|
|
for s in state.get("herdr_sessions", []):
|
|
role = (s.get("role") or "").lower()
|
|
if "reviewer" in role and s.get("status") == "running" and s.get("name") != self.creator:
|
|
reviewers.append(str(s.get("name")))
|
|
return reviewers
|
|
|
|
def resolve_planner_session(self, state: Dict[str, Any]) -> str:
|
|
for s in state.get("herdr_sessions", []):
|
|
role = (s.get("role") or "").lower()
|
|
if "planner" in role and s.get("status") == "running":
|
|
return str(s.get("name"))
|
|
return ""
|
|
|
|
def delegate_job(self, agent_session: str, agent_type: str, role: str, prompt: str) -> Optional[str]:
|
|
"""Submit a delegate job and extract the 32-bit hex Job ID."""
|
|
wrapper_script = self.workspace_root / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
|
if not wrapper_script.exists():
|
|
wrapper_script = Path(__file__).resolve().parent.parent / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
|
|
|
def _run_cmd(p_text: str):
|
|
cmd = [
|
|
"bash", str(wrapper_script), "submit",
|
|
"--agent-session", self.session_label(agent_session),
|
|
"--agent", agent_type,
|
|
"--type", "direct",
|
|
"--role", role,
|
|
"--registry-dir", str(self.jobs_dir),
|
|
"--prompt", p_text,
|
|
] + self.contract_args()
|
|
env = dict(os.environ)
|
|
if "DELEGATE_JOB_PYTHON" not in env:
|
|
env["DELEGATE_JOB_PYTHON"] = sys.executable
|
|
return subprocess.run(cmd, capture_output=True, text=True, cwd=str(self.workspace_root), env=env)
|
|
|
|
res = _run_cmd(prompt)
|
|
match = re.search(r"registered job:\s*([A-Za-z0-9_-]+)", res.stdout)
|
|
job_id = match.group(1) if match else None
|
|
|
|
if res.returncode != 0:
|
|
log_error(f"delegate_job submit failed (exit {res.returncode}):\n{res.stderr or res.stdout}")
|
|
if res.returncode == 5:
|
|
# rc=5 (session_busy): MAM_SESSION_BUSY_RETRY_S wait, then 1 retry
|
|
retry_s = int(os.environ.get("MAM_SESSION_BUSY_RETRY_S", "30"))
|
|
if retry_s > 0:
|
|
time.sleep(retry_s)
|
|
res_retry = _run_cmd(prompt)
|
|
match_retry = re.search(r"registered job:\s*([A-Za-z0-9_-]+)", res_retry.stdout)
|
|
if match_retry:
|
|
job_id_retry = match_retry.group(1)
|
|
self.created_jobs.append(job_id_retry)
|
|
self._record_checkpoint()
|
|
if res_retry.returncode != 0:
|
|
self.failed_submit_jobs.add(job_id_retry)
|
|
log_error(f"delegate_job retry failed (exit {res_retry.returncode}):\n{res_retry.stderr or res_retry.stdout}")
|
|
return job_id_retry
|
|
if job_id:
|
|
self.created_jobs.append(job_id)
|
|
self.failed_submit_jobs.add(job_id)
|
|
self._record_checkpoint()
|
|
return job_id
|
|
return None
|
|
|
|
if job_id:
|
|
self.created_jobs.append(job_id)
|
|
self._record_checkpoint()
|
|
return job_id
|
|
return None
|
|
|
|
def wait_for_job(self, job_id: str, max_wait: int = 3900, check_interval: float = 1.0) -> bool:
|
|
"""Blocking wait for job completion or error with safety timeout."""
|
|
if job_id in self.failed_submit_jobs:
|
|
# Submit command for this job failed (rc!=0). Cap wait to 5s to avoid 3900s hang
|
|
max_wait = min(max_wait, 5.0)
|
|
|
|
if self.verbose:
|
|
log_info(f"Monitoring job '{job_id}' for status changes (timeout: {max_wait}s)...")
|
|
|
|
job_file = self.jobs_dir / f"{job_id}.json"
|
|
start_time = time.time()
|
|
missing_count = 0
|
|
unknown_count = 0
|
|
|
|
while time.time() - start_time < max_wait:
|
|
if self.interrupted:
|
|
return False
|
|
status = "unknown"
|
|
if job_file.exists():
|
|
try:
|
|
with open(job_file, "r", encoding="utf-8") as f:
|
|
status = json.load(f).get("status", "unknown")
|
|
except Exception:
|
|
status = "unknown"
|
|
else:
|
|
missing_count += 1
|
|
if missing_count >= 5 and (time.time() - start_time) >= min(max_wait, 5.0):
|
|
log_error(f"Job record '{job_file}' not found on disk; aborting wait.")
|
|
return False
|
|
|
|
if status == "unknown":
|
|
unknown_count += 1
|
|
if unknown_count >= 5 and (time.time() - start_time) >= min(max_wait, 5.0):
|
|
log_error(f"Job '{job_id}' record unreadable / status unknown; aborting wait.")
|
|
return False
|
|
|
|
if status == "completed":
|
|
if self.verbose:
|
|
log_success(f"Job '{job_id}' completed successfully.")
|
|
return True
|
|
elif status in ("error", "cancelled"):
|
|
log_error(f"Job '{job_id}' finished with status '{status}'.")
|
|
return False
|
|
|
|
time.sleep(check_interval)
|
|
|
|
log_error(f"Job '{job_id}' timed out after {max_wait}s.")
|
|
return False
|
|
|
|
def find_report(self, job_id: str) -> Optional[Path]:
|
|
"""Locate report-final.md for a job."""
|
|
job_dir = self.jobs_dir / job_id
|
|
matches = list(job_dir.glob("*/report-final.md")) + list(job_dir.glob("report-final.md"))
|
|
if matches:
|
|
return matches[0]
|
|
return None
|
|
|
|
def collect_changes_diff(self, head_commit: Optional[str] = None) -> str:
|
|
"""Collect git diff since base commit - of the live source tree, or,
|
|
with head_commit (review identity, B13), of base..head_commit."""
|
|
# The collector is the control root's copy (never the worktree's
|
|
# unreviewed one); the tree it diffs is the source root (Issue #21).
|
|
diff_script = self.workspace_root / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "diff_collect.sh"
|
|
src = str(self.worktree_root)
|
|
if diff_script.exists():
|
|
cmd = (f"source {shlex.quote(str(diff_script))} && "
|
|
f"mam_collect_changes_diff {shlex.quote(src)} {shlex.quote(self.base_commit)}"
|
|
+ (f" {shlex.quote(head_commit)}" if head_commit else ""))
|
|
res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, cwd=src)
|
|
if res.returncode == 0:
|
|
return res.stdout
|
|
if head_commit:
|
|
# the snapshot's evidence is not replaced by anything else (B13):
|
|
# the collector's UNAVAILABLE output is passed on as it is
|
|
return res.stdout if UNAVAILABLE_MARKER in res.stdout else (
|
|
f"{UNAVAILABLE_MARKER}\nreason: diff collector failed (rc {res.returncode})\n")
|
|
# Fallback to direct git diff
|
|
try:
|
|
if head_commit:
|
|
cmd = ["git", "diff", self.base_commit, head_commit]
|
|
else:
|
|
cmd = ["git", "diff", self.base_commit] if self.base_commit else ["git", "diff", "HEAD"]
|
|
res = subprocess.run(cmd, capture_output=True, text=True, cwd=src)
|
|
if head_commit and res.returncode != 0:
|
|
return f"{UNAVAILABLE_MARKER}\nreason: git diff failed (rc {res.returncode})\n"
|
|
return res.stdout
|
|
except Exception:
|
|
return ""
|
|
|
|
def _source_head(self) -> str:
|
|
res = self._git(self.worktree_root, "rev-parse", "HEAD")
|
|
out = res.stdout if isinstance(res.stdout, str) else ""
|
|
return out.strip() if res.returncode == 0 else ""
|
|
|
|
def _context_active(self) -> bool:
|
|
return self.worktree_given or self.base_ref is not None
|
|
|
|
def _worktree_context(self, remote: bool = False, purpose: str = "work") -> str:
|
|
"""The source-root block put in front of every delegated prompt (D6);
|
|
empty without --worktree/--base, so prompts stay byte-identical.
|
|
purpose: 'work' (creator/planner), 'review' (a verdict job, whose
|
|
target is the [리뷰 identity] path) or 'adjudication' (a re-hearing)."""
|
|
if not self._context_active():
|
|
return ""
|
|
src = self.worktree_root
|
|
branch = self._git(src, "branch", "--show-current").stdout.strip() or "(detached)"
|
|
head = self._source_head()
|
|
base = self.base_commit or self.base_sha
|
|
origin = "--base" if self.base_source == "cli" else "루프 시작 시 HEAD"
|
|
count = "?"
|
|
if base and head:
|
|
r = self._git(src, "rev-list", "--count", f"{base}..HEAD")
|
|
if r.returncode == 0 and r.stdout.strip():
|
|
count = r.stdout.strip()
|
|
st = self._git(src, "status", "--porcelain", "--untracked-files=all")
|
|
dirty = "알 수 없음" if st.returncode != 0 else ("있음" if st.stdout.strip() else "없음")
|
|
# a remote reviewer (diff-only) cannot reach this host's paths (B6)
|
|
src_line = ("- 소스 경로: (오케스트레이터 호스트 전용 — 원격 리뷰어는 접근할 수 없음)\n" if remote
|
|
else f"- 소스 경로: {src}\n")
|
|
if purpose == "review" and remote:
|
|
tail = "이 잡의 검증 대상은 아래 SHA 와 diff 다.\n\n"
|
|
elif purpose == "review" and self.review_isolation == "worktree":
|
|
# one verification target per prompt (loop B review B-2)
|
|
tail = ("리뷰 검증은 아래 [리뷰 identity] 의 검증 경로에서만 한다. 소스 경로는 creator 의 작업 트리이며 "
|
|
"리뷰어는 거기서 실행·수정하지 않는다. 브리프와 리포트는 브리프에 적힌 절대경로를 그대로 쓴다.\n\n")
|
|
elif purpose == "review":
|
|
tail = ("리뷰 검증은 아래 [리뷰 identity] 의 검증 경로(= 위 소스 경로)에서 하며, 거기서 수정·커밋하지 않는다. "
|
|
"브리프와 리포트는 브리프에 적힌 절대경로를 그대로 쓴다.\n\n")
|
|
elif purpose == "adjudication" and remote:
|
|
tail = "이 잡은 아래 반론의 재심이다. 판정 근거는 직전 리뷰에서 받은 SHA·diff 와 아래 반론 본문이다.\n\n"
|
|
elif remote:
|
|
tail = "이 잡의 검증 대상은 아래 SHA 와 diff 다.\n\n"
|
|
else:
|
|
tail = "이 잡의 코드 작업·검증 대상은 위 소스 경로다. 브리프와 리포트는 브리프에 적힌 절대경로를 그대로 쓴다.\n\n"
|
|
return ("[워크트리 컨텍스트]\n"
|
|
+ src_line
|
|
+ f"- 브랜치: {branch}\n"
|
|
f"- base: {base or '(Phase 2 시작 시 HEAD 로 정해짐)'} (출처: {origin})\n"
|
|
f"- head: {head} (base..head 커밋: {count}개, 미커밋 변경: {dirty})\n"
|
|
f"- 제어 평면(잡 브리프·리포트·레지스트리): {self.workspace_root}\n"
|
|
+ tail)
|
|
|
|
def _dispatch(self, agent_session: str, agent_type: str, role: str, prompt: str,
|
|
remote: bool = False, purpose: str = "work") -> Optional[str]:
|
|
"""delegate_job with the source-root context block in front of the prompt."""
|
|
return self.delegate_job(agent_session, agent_type, role, self._worktree_context(remote, purpose) + prompt)
|
|
|
|
def _change_set_state(self) -> str:
|
|
"""'empty' | 'present' | 'unknown' for the source root against base,
|
|
judged by git exit codes only (D16)."""
|
|
if not self.base_commit:
|
|
return "unknown"
|
|
try:
|
|
d = self._git(self.worktree_root, "diff", "--quiet", self.base_commit)
|
|
if d.returncode == 1:
|
|
return "present"
|
|
if d.returncode != 0:
|
|
return "unknown"
|
|
u = self._git(self.worktree_root, "ls-files", "-o", "--exclude-standard", "-z")
|
|
if u.returncode != 0:
|
|
return "unknown"
|
|
return "present" if u.stdout else "empty"
|
|
except Exception:
|
|
return "unknown"
|
|
|
|
def _change_set_allows_review(self) -> bool:
|
|
"""Fail-closed guard (D16), on only with --worktree/--base: never ask
|
|
for a review of an empty or undeterminable change set."""
|
|
if not self._context_active():
|
|
return True
|
|
state = self._change_set_state()
|
|
if state == "empty":
|
|
log_error(f"empty change set since base {self.base_commit}; refusing to request reviews")
|
|
return False
|
|
if state != "present":
|
|
log_error(f"cannot determine the change set since base {self.base_commit or '(none)'} in "
|
|
f"{self.worktree_root}; refusing to request reviews on no evidence")
|
|
return False
|
|
return True
|
|
|
|
# ------------------------------------------------------------------
|
|
# Issue #21 item 5: review identity
|
|
# ------------------------------------------------------------------
|
|
def _is_remote(self, name: str) -> bool:
|
|
return self.session_label(name).startswith("remote:")
|
|
|
|
def _verdict_recipients(self, state: Dict[str, Any]) -> List[str]:
|
|
"""Who receives a verdict job - the same rule Phase 3 applies; the
|
|
creator when there is no reviewer (self-review)."""
|
|
if self.all_reviewers:
|
|
names = self.resolve_all_reviewers(state)
|
|
elif self.reviewer_list:
|
|
names = [r.strip() for r in self.reviewer_list.split(",") if r.strip()]
|
|
else:
|
|
names = []
|
|
return names or [self.creator]
|
|
|
|
def _check_remote_recipients(self, state: Dict[str, Any]) -> bool:
|
|
"""B14: a verdict recipient must be able to reach the evidence recorded
|
|
for it. Remote reviewers cannot reach this host's review tree: refused
|
|
by default, admitted with the snapshot diff as their only evidence
|
|
under --remote-review diff-only. Checked before any delegation."""
|
|
remote = [n for n in self._verdict_recipients(state) if self._is_remote(n)]
|
|
self._remote_recipients = set(remote)
|
|
if remote and self.remote_review != "diff-only":
|
|
log_error(f"identity review needs the review tree on this host; remote reviewer(s) {' '.join(remote)} "
|
|
"cannot verify it. Use local reviewers, or pass --remote-review diff-only")
|
|
return False
|
|
if remote:
|
|
log_warn(f"remote reviewer(s) {' '.join(remote)} admitted with the snapshot diff as their only "
|
|
"evidence (--remote-review diff-only)")
|
|
return True
|
|
|
|
def _worktree_tree(self) -> 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."""
|
|
dirty_subs = self._dirty_submodules()
|
|
if dirty_subs is None:
|
|
log_error(f"review snapshot: cannot check the submodules of {self.worktree_root}")
|
|
return None
|
|
if dirty_subs:
|
|
log_error(f"review snapshot: submodule(s) with uncommitted or untracked content: {' '.join(dirty_subs)}; "
|
|
"the snapshot records only their commit - commit or clean them first")
|
|
return None
|
|
fd, idx = tempfile.mkstemp(prefix="mam-review-index-")
|
|
os.close(fd)
|
|
os.unlink(idx)
|
|
env = dict(os.environ, GIT_INDEX_FILE=idx)
|
|
try:
|
|
for step in (("read-tree", "HEAD"), ("add", "-A"), ("write-tree",)):
|
|
r = subprocess.run(["git", "-C", str(self.worktree_root), *step],
|
|
capture_output=True, text=True, env=env)
|
|
if r.returncode != 0:
|
|
log_error(f"review snapshot: git {step[0]} failed in {self.worktree_root}: {r.stderr.strip()}")
|
|
return None
|
|
return r.stdout.strip() or None
|
|
finally:
|
|
for p in (idx, idx + ".lock"):
|
|
try:
|
|
os.unlink(p)
|
|
except OSError:
|
|
pass
|
|
|
|
def _dirty_submodules(self) -> Optional[List[str]]:
|
|
"""Checked-out submodules (recursively) with uncommitted or untracked
|
|
content; None when the check itself fails. Each submodule is asked
|
|
directly with --ignore-submodules=none and --untracked-files=all, so no
|
|
submodule.*.ignore, diff.ignoreSubmodules or status.showUntrackedFiles
|
|
setting can hide it."""
|
|
script = ('s=$(git status --porcelain --untracked-files=all --ignore-submodules=none) || exit 1; '
|
|
'[ -z "$s" ] || printf "%s\\n" "$displaypath"')
|
|
r = self._git(self.worktree_root, "submodule", "foreach", "--recursive", "--quiet", script)
|
|
if r.returncode != 0:
|
|
return None
|
|
return [line for line in r.stdout.splitlines() if line.strip()]
|
|
|
|
def _gitlinks(self, treeish: str) -> Optional[Set[str]]:
|
|
r = self._git(self.worktree_root, "ls-tree", "-r", "-z", treeish)
|
|
if r.returncode != 0:
|
|
return None
|
|
links = set()
|
|
for ent in r.stdout.split("\0"):
|
|
meta, _, path = ent.partition("\t")
|
|
if meta.split(" ")[0] == "160000":
|
|
links.add(path)
|
|
return links
|
|
|
|
def _identity_now(self) -> Tuple[str, str]:
|
|
return (self._source_head(), self._worktree_tree() or "")
|
|
|
|
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
|
|
top of HEAD. An untracked nested repository cannot be pinned (only a
|
|
gitlink would be) and is refused."""
|
|
head = self._source_head()
|
|
tree = self._worktree_tree() if head else None
|
|
if not head or not tree:
|
|
log_error(f"cannot snapshot {self.worktree_root}; refusing to request reviews")
|
|
return None
|
|
new_links = self._gitlinks(tree)
|
|
old_links = self._gitlinks(head)
|
|
if new_links is None or old_links is None:
|
|
log_error("cannot list the snapshot's entries; refusing to request reviews")
|
|
return None
|
|
untracked_repos = sorted(new_links - old_links)
|
|
if untracked_repos:
|
|
log_error(f"untracked nested git repository {' '.join(untracked_repos)} cannot be pinned into the review "
|
|
"snapshot; commit it, ignore it or move it out")
|
|
return None
|
|
dirty = tree != self._git(self.worktree_root, "rev-parse", "HEAD^{tree}").stdout.strip()
|
|
commit = head
|
|
if dirty:
|
|
ident = {"GIT_AUTHOR_NAME": "mam-loop", "GIT_AUTHOR_EMAIL": "mam-loop@localhost",
|
|
"GIT_COMMITTER_NAME": "mam-loop", "GIT_COMMITTER_EMAIL": "mam-loop@localhost"}
|
|
c = subprocess.run(["git", "-C", str(self.worktree_root), "commit-tree", tree, "-p", head,
|
|
"-m", f"mam review snapshot {self.loop_id} #{n}"],
|
|
capture_output=True, text=True, env=dict(os.environ, **ident))
|
|
if c.returncode != 0 or not c.stdout.strip():
|
|
log_error(f"cannot commit the review snapshot: {c.stderr.strip()}")
|
|
return None
|
|
commit = c.stdout.strip()
|
|
return {"head": head, "tree": tree, "dirty": dirty, "review_commit": commit}
|
|
|
|
def _open_review_tree(self, snap: Dict[str, Any], n: int) -> Optional[str]:
|
|
path = Path(tempfile.gettempdir()) / "mam-review" / self.loop_id / str(n)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
r = self._git(self.worktree_root, "worktree", "add", "--detach", str(path), snap["review_commit"])
|
|
if r.returncode != 0:
|
|
log_error(f"cannot open the review worktree {path}: {r.stderr.strip()}; refusing to request reviews "
|
|
"(the isolation is not lowered silently)")
|
|
return None
|
|
self._review_paths.append(str(path))
|
|
return str(path)
|
|
|
|
def _close_review_tree(self, path: Optional[str]) -> None:
|
|
if not path or path not in self._review_paths:
|
|
return
|
|
r = self._git(self.worktree_root, "worktree", "remove", "--force", path)
|
|
if os.path.isdir(path):
|
|
shutil.rmtree(path, ignore_errors=True)
|
|
if r.returncode != 0:
|
|
# only when remove failed: prune drops this entry (its directory is
|
|
# gone now) - and any other stale one, which is why it is not routine
|
|
self._git(self.worktree_root, "worktree", "prune")
|
|
self._review_paths.remove(path)
|
|
try:
|
|
os.rmdir(os.path.dirname(path)) # TMPDIR/mam-review/<loop_id>, once empty
|
|
except OSError:
|
|
pass
|
|
|
|
def _close_all_review_trees(self) -> None:
|
|
for p in list(self._review_paths):
|
|
self._close_review_tree(p)
|
|
|
|
def _evidence(self, name: str) -> Dict[str, str]:
|
|
if name in self._remote_recipients:
|
|
return {"transport": "remote", "evidence": "diff-only", "isolation": "none (remote)"}
|
|
if self.review_isolation == "worktree":
|
|
return {"transport": "herdr", "evidence": "review-tree", "isolation": "worktree"}
|
|
return {"transport": "herdr", "evidence": "source-tree", "isolation": "env"}
|
|
|
|
def _identity_block(self, snap: Dict[str, Any], path: Optional[str], remote: bool) -> str:
|
|
lines = ["[리뷰 identity]",
|
|
f"- loop_id: {self.loop_id}",
|
|
f"- base: {self.base_commit}",
|
|
f"- head: {snap['head']}",
|
|
f"- tree: {snap['tree']}"]
|
|
if snap["dirty"]:
|
|
lines.append(f"- 미커밋 변경 포함 스냅샷 커밋: {snap['review_commit']}")
|
|
if remote:
|
|
lines.append("당신의 호스트에는 이 트리가 없다. 자기 체크아웃이나 다른 커밋에서 실행한 결과를 판정 근거로 "
|
|
"삼지 말 것. 근거는 아래 diff 뿐이며, 리포트에 'diff 만으로 검토' 를 밝힐 것.")
|
|
else:
|
|
py = self.workspace_root / ".venv" / "bin" / "python"
|
|
py_cmd = str(py) if py.exists() else "python3"
|
|
lines.append(f"- 검증 경로: {path}")
|
|
lines.append("이 경로에서만 검증하고 커밋·브랜치 변경은 하지 말 것. 테스트는 "
|
|
f"`PYTHONDONTWRITEBYTECODE=1 {py_cmd} -m pytest -p no:cacheprovider "
|
|
"--basetemp=<검증 경로 밖의 임시 디렉터리>` 로 실행한다.")
|
|
return "\n".join(lines) + "\n\n"
|
|
|
|
def _record_round(self, n: int, snap: Dict[str, Any], path: Optional[str], jobs: Dict[str, str],
|
|
verdicts: Dict[str, str], stale: bool) -> None:
|
|
self._rounds.append({
|
|
"iteration": n, "base_sha": self.base_commit, "head_sha": snap["head"], "tree_sha": snap["tree"],
|
|
"review_commit": snap["review_commit"], "isolation": self.review_isolation,
|
|
"review_path": path or "", "jobs": jobs, "verdicts": verdicts,
|
|
"reviewers": {r: self._evidence(r) for r in jobs}, "stale": stale,
|
|
})
|
|
self._record_checkpoint(iteration=n)
|
|
|
|
def _round_is_stale(self, snap: Dict[str, Any]) -> bool:
|
|
"""B7: the source changed while the round was being reviewed."""
|
|
return self._identity_now() != (snap["head"], snap["tree"])
|
|
|
|
def _record_checkpoint(self, iteration: int = 1, review_status: Optional[Dict[str, str]] = None) -> None:
|
|
data = {
|
|
"iteration": iteration,
|
|
"created_jobs": list(self.created_jobs),
|
|
"current_plan": self.current_plan,
|
|
"plan_job_id": self.plan_job_id,
|
|
"review_status": review_status or {},
|
|
"base_commit": self.base_commit,
|
|
"base_source": self.base_source,
|
|
"review_only": self.review_only,
|
|
"worktree_root": str(self.worktree_root),
|
|
"head_commit": self._source_head() if self._context_active() else "",
|
|
"loop_id": self.loop_id,
|
|
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
}
|
|
if self._context_active():
|
|
data["review_rounds"] = self._rounds
|
|
checkpoint.save_checkpoint(self.checkpoint_file, data)
|
|
|
|
def release_guard(self) -> None:
|
|
"""Release MAM_LOOP_MARKER and clean up freeze dir if owned."""
|
|
self._close_all_review_trees()
|
|
marker = os.environ.get("MAM_LOOP_MARKER") or str(self.mam_dir / "loop-guard-active")
|
|
if os.path.isfile(marker):
|
|
try:
|
|
with open(marker, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
if f"pid={os.getpid()}" in content:
|
|
os.unlink(marker)
|
|
except OSError:
|
|
pass
|
|
|
|
freeze_dir = os.environ.get("MAM_LOOP_FREEZE_DIR")
|
|
if freeze_dir and os.environ.get("MAM_LOOP_FREEZE_OWNED") == "1":
|
|
if "mam-loop-freeze." in freeze_dir and os.path.isdir(freeze_dir):
|
|
shutil.rmtree(freeze_dir, ignore_errors=True)
|
|
|
|
def run(self) -> int:
|
|
try:
|
|
return self._run_body()
|
|
finally:
|
|
self._close_all_review_trees() # a return or exit mid-round leaves no review tree
|
|
|
|
def _run_body(self) -> int:
|
|
self.parse_args()
|
|
|
|
atexit.register(self.release_guard)
|
|
|
|
# Signal handling
|
|
def _handle_signal(signum, frame):
|
|
self.interrupted = True
|
|
log_warn(f"Received signal {signum}. Quiescing and saving checkpoint...")
|
|
self._record_checkpoint()
|
|
self.release_guard()
|
|
sys.exit(128 + signum)
|
|
|
|
signal.signal(signal.SIGINT, _handle_signal)
|
|
signal.signal(signal.SIGTERM, _handle_signal)
|
|
|
|
log_info("Initializing multi-agent-mux-loop controller...")
|
|
log_info(f"Loop id: {self.loop_id}")
|
|
log_info(f"Target Agent: {self.creator}")
|
|
log_info(f"Task Goal: {self.task}")
|
|
|
|
state = self.load_sessions_state()
|
|
|
|
# Validate target agent session
|
|
target_status = None
|
|
for s in state.get("herdr_sessions", []):
|
|
if s.get("name") == self.creator:
|
|
target_status = s.get("status")
|
|
break
|
|
|
|
if not target_status:
|
|
log_error(f"Target agent session '{self.creator}' is not registered in the session registry.")
|
|
return 1
|
|
elif target_status != "running":
|
|
log_error(f"Target agent session '{self.creator}' is not running (current status: '{target_status}'). Please start it first.")
|
|
return 1
|
|
|
|
# Resolve Planner session
|
|
planner_session = ""
|
|
if self.planner_override:
|
|
planner_session = self.planner_override
|
|
p_status = None
|
|
for s in state.get("herdr_sessions", []):
|
|
if s.get("name") == planner_session:
|
|
p_status = s.get("status")
|
|
break
|
|
if not p_status:
|
|
log_error(f"Specified planner session '{planner_session}' is not registered in the session registry.")
|
|
return 1
|
|
elif p_status != "running":
|
|
log_error(f"Specified planner session '{planner_session}' is not running (current status: '{p_status}').")
|
|
return 1
|
|
else:
|
|
planner_session = self.resolve_planner_session(state)
|
|
if self.plan_mode and not planner_session:
|
|
log_error("Planner mode enabled (--plan) but no running session with a 'planner' role was found.")
|
|
return 1
|
|
|
|
log_info(f"Resolved Planner session: {planner_session}")
|
|
|
|
# Issue #21 item 5 (B14): every verdict recipient must reach the
|
|
# evidence recorded for it - checked before anything is delegated.
|
|
if self._context_active() and not self._check_remote_recipients(state):
|
|
return 1
|
|
|
|
# =======================================================================
|
|
# PHASE 1: PLANNING & DISCUSSIONS
|
|
# =======================================================================
|
|
if self.plan_mode:
|
|
log_info("=== Phase 1: Interactive Planning Phase ===")
|
|
log_info("Requesting initial implementation plan from Planner...")
|
|
p_type = self.resolve_agent_type(planner_session, state)
|
|
plan_prompt = f"태스크 목표를 바탕으로 구체적인 구현 계획서를 작성해주세요. 목표: {self.task}"
|
|
p_jid = self._dispatch(planner_session, p_type, "Planner", plan_prompt)
|
|
if not p_jid:
|
|
log_error("Failed to register planner job.")
|
|
return 1
|
|
self.plan_job_id = p_jid
|
|
log_info(f"Planner Job ID: {p_jid}")
|
|
|
|
if not self.wait_for_job(p_jid):
|
|
log_error("Planning phase failed during initial plan design.")
|
|
return 1
|
|
|
|
p_file = self.find_report(p_jid)
|
|
if not p_file or not p_file.exists():
|
|
log_error("Final plan file not found.")
|
|
return 1
|
|
plan_text = self._plan_text(p_file)
|
|
if plan_text is None:
|
|
# D13: the run paid for a planning phase and got nothing back.
|
|
# Falling through would silently demote it to self-planning.
|
|
log_error(f"the planner's report has no plan text: {p_file}")
|
|
return 1
|
|
self.current_plan = plan_text
|
|
|
|
# Interactive turns
|
|
creator_type = self.resolve_agent_type(self.creator, state)
|
|
for turn in range(1, self.plan_talk_turns + 1):
|
|
log_info(f"Discussion Turn {turn}/{self.plan_talk_turns}: Creator challenging the plan...")
|
|
c_prompt = f"Planner가 제시한 다음 계획서를 꼼꼼히 검토하고, 실제 구현 시 마주할 수 있는 맹점이나 제약사항 1가지를 발굴하여 Planner에게 이의를 제기(Challenge)해주세요. 계획서:\n{self.current_plan}"
|
|
c_jid = self._dispatch(self.creator, creator_type, "Worker", c_prompt)
|
|
if not c_jid or not self.wait_for_job(c_jid):
|
|
log_error("Creator critique step failed.")
|
|
return 1
|
|
|
|
c_file = self.find_report(c_jid)
|
|
critique_text = c_file.read_text(encoding="utf-8", errors="replace") if c_file else ""
|
|
|
|
log_info("Planner refining plan with Creator's feedback...")
|
|
refine_prompt = f"작업자(Creator)로부터 다음 이의제기 피드백을 받았습니다. 피드백을 반영하여 계획서를 정교하게 업데이트(Refine)하여 다시 출력해주세요. 피드백:\n{critique_text}\n기존 계획서:\n{self.current_plan}"
|
|
ref_jid = self._dispatch(planner_session, p_type, "Planner", refine_prompt)
|
|
if not ref_jid or not self.wait_for_job(ref_jid):
|
|
log_error("Planner refinement step failed.")
|
|
return 1
|
|
|
|
ref_file = self.find_report(ref_jid)
|
|
refined = self._plan_text(ref_file) if ref_file else None
|
|
if refined is None:
|
|
# D14: overwriting with an empty report loses a plan that is
|
|
# still good, and Phase 2 would quietly switch prompts.
|
|
log_warn(f"the refine report has no plan text (job {ref_jid}); keeping the previous plan.")
|
|
else:
|
|
self.current_plan = refined
|
|
|
|
log_success("Interactive planning completed. Plan finalized.")
|
|
else:
|
|
log_info("=== Phase 1: Self-Planning Mode (Direct Execution) ===")
|
|
# Issue #25 (D1): no plan is loaded unless one was named. This used
|
|
# to take the newest promoted plan in the planner's report dir,
|
|
# keyed on mtime (checkout order) and on a planner session that says
|
|
# nothing about this task - so it injected other issues' plans.
|
|
if self.current_plan:
|
|
mtime = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(os.path.getmtime(self.plan_file)))
|
|
log_info(f"Using the plan given with --plan-file: {self.plan_file} (modified {mtime} UTC)")
|
|
else:
|
|
log_info("No plan loaded; the Creator plans by itself (reuse one with --plan-file <path>).")
|
|
|
|
# =======================================================================
|
|
# PHASE 2: IMPLEMENTATION (CREATION)
|
|
# =======================================================================
|
|
log_info("=== Phase 2: Code Implementation ===")
|
|
if self.base_sha:
|
|
self.base_commit = self.base_sha # --base (D14)
|
|
else:
|
|
try:
|
|
res_base = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True, cwd=str(self.worktree_root))
|
|
self.base_commit = res_base.stdout.strip()
|
|
except Exception:
|
|
self.base_commit = ""
|
|
|
|
creator_type = self.resolve_agent_type(self.creator, state)
|
|
if self.review_only:
|
|
log_info(f"Phase 2 skipped (--review-only): reviewing {self.base_commit}..{self._source_head()}")
|
|
else:
|
|
if self.current_plan:
|
|
exec_prompt = f"이미 수립된 다음 계획서에 입각하여 작업자의 판단하에 코드를 구현하고 작업 목표를 완성해주세요. 계획서:\n{self.current_plan}\n\n작업 목표: {self.task}"
|
|
else:
|
|
exec_prompt = f"계획서가 존재하지 않으므로, 작업자(Creator)의 판단하에 스스로 구현 계획 및 설계를 수립한 뒤, 이를 바탕으로 코드를 구현하고 다음 작업 목표를 완성해주세요. 작업 목표: {self.task}"
|
|
|
|
exec_jid = self._dispatch(self.creator, creator_type, "Worker", exec_prompt)
|
|
if not exec_jid:
|
|
log_error("Failed to register Creator execution job.")
|
|
return 1
|
|
log_info(f"Creator Job ID: {exec_jid}")
|
|
|
|
if not self.wait_for_job(exec_jid):
|
|
log_error("Creator execution failed.")
|
|
return 1
|
|
log_success("Initial implementation finished.")
|
|
|
|
# =======================================================================
|
|
# PHASE 3: VERIFICATION LOOP (PEER REVIEW)
|
|
# =======================================================================
|
|
log_info("=== Phase 3: Verification & Corrective Review Loop ===")
|
|
|
|
reviewers: List[str] = []
|
|
if self.all_reviewers:
|
|
reviewers = self.resolve_all_reviewers(state)
|
|
if not reviewers:
|
|
log_error("--all-reviewer was specified, but no running reviewer sessions were found.")
|
|
return 1
|
|
elif self.reviewer_list:
|
|
reviewers = [r.strip() for r in self.reviewer_list.split(",") if r.strip()]
|
|
|
|
rebut_total_budget = self.max_rebut * self.max_loop
|
|
|
|
final_passed_reviewers: Dict[str, Tuple[str, Path]] = {}
|
|
identity = self._context_active()
|
|
# 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.
|
|
self_review = identity and not reviewers
|
|
if self_review:
|
|
log_warn("No reviewers specified. Conducting Creator Self-Review...")
|
|
reviewers = [self.creator]
|
|
|
|
for loop_count in range(1, self.max_loop + 1):
|
|
log_info(f"Review Loop Iteration {loop_count}/{self.max_loop}...")
|
|
rebut_budget = self.max_rebut
|
|
self._record_checkpoint(iteration=loop_count)
|
|
if not self._change_set_allows_review():
|
|
return 1
|
|
|
|
# Issue #21 item 5: pin this round (B3), open its review tree (B4)
|
|
snap: Optional[Dict[str, Any]] = None
|
|
review_path: Optional[str] = None
|
|
if identity:
|
|
final_passed_reviewers.clear() # B8: only this round's approvals can be promoted
|
|
snap = self._snapshot(loop_count)
|
|
if snap is None:
|
|
return 1
|
|
if self.review_isolation == "worktree":
|
|
review_path = self._open_review_tree(snap, loop_count)
|
|
if review_path is None:
|
|
return 1
|
|
else:
|
|
review_path = str(self.worktree_root)
|
|
|
|
if not reviewers:
|
|
# Creator Self-Review
|
|
log_warn("No reviewers specified. Conducting Creator Self-Review...")
|
|
sr_prompt = 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 sr_report and parser.has_verdict(sr_report, "PASS") and not parser.has_verdict(sr_report, "NOT PASS"):
|
|
log_success("Self-Review PASS.")
|
|
break
|
|
else:
|
|
log_warn("Self-Review NOT PASS.")
|
|
if loop_count == self.max_loop:
|
|
log_error("Reached max loop count. Self-review loop aborted with failures.")
|
|
return 1
|
|
continue
|
|
|
|
log_info(f"Active reviewers: {' '.join(reviewers)}")
|
|
# B13: in identity mode the reviewers' diff is the snapshot's, so the
|
|
# evidence they get is exactly what the verdict is bound to.
|
|
changes_diff = self.collect_changes_diff(snap["review_commit"] if identity else None)
|
|
diff_only = [r for r in reviewers if r in self._remote_recipients]
|
|
if identity and diff_only and ("!!! DIFF TRUNCATED !!!" in changes_diff
|
|
or "!!! CHANGE SET UNAVAILABLE !!!" in changes_diff):
|
|
log_error(f"the snapshot diff is truncated/unavailable; diff-only remote reviewer(s) "
|
|
f"{' '.join(diff_only)} would have no evidence")
|
|
return 1
|
|
|
|
review_jobs: List[Tuple[str, str]] = [] # (rev, job_id)
|
|
for rev in reviewers:
|
|
log_info(f"Requesting code review from Reviewer '{rev}'...")
|
|
r_type = self.resolve_agent_type(rev, state)
|
|
if self_review:
|
|
r_prompt = f"{SELF_REVIEW_PROMPT} 변경분:\n{changes_diff}"
|
|
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
|
|
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,
|
|
purpose="review" if identity else "work")
|
|
if not r_jid:
|
|
log_error(f"Failed to register review job for '{rev}'.")
|
|
return 1
|
|
review_jobs.append((rev, r_jid))
|
|
|
|
all_passed = True
|
|
round_verdicts: Dict[str, str] = {}
|
|
fail_revs: List[str] = []
|
|
fail_reports: List[Path] = []
|
|
feedback_parts: List[str] = []
|
|
|
|
for rev, jid in review_jobs:
|
|
if not self.wait_for_job(jid):
|
|
log_warn(f"Reviewer '{rev}' job crashed.")
|
|
round_verdicts[rev] = "crashed"
|
|
all_passed = False
|
|
continue
|
|
|
|
r_file = self.find_report(jid)
|
|
if not r_file or not r_file.exists():
|
|
log_warn(f"Reviewer '{rev}' report not found. Counting as NOT PASS.")
|
|
round_verdicts[rev] = "missing"
|
|
all_passed = False
|
|
continue
|
|
|
|
content = r_file.read_text(encoding="utf-8", errors="replace")
|
|
verdict = parser.parse_verdict(content)
|
|
|
|
if verdict != "PASS":
|
|
log_warn(f"Reviewer '{rev}': NOT PASS")
|
|
round_verdicts[rev] = "NOT PASS"
|
|
all_passed = False
|
|
fail_revs.append(rev)
|
|
fail_reports.append(r_file)
|
|
feedback_parts.append(f"\n--- Reviewer ({rev}) Feedback ---\n{content}")
|
|
else:
|
|
log_success(f"Reviewer '{rev}': PASS")
|
|
round_verdicts[rev] = "PASS"
|
|
final_passed_reviewers[rev] = (jid, r_file)
|
|
|
|
if identity:
|
|
# B7: judge the round against the source as it is now
|
|
if self.review_isolation == "worktree":
|
|
self._close_review_tree(review_path)
|
|
stale = self._round_is_stale(snap)
|
|
self._record_round(loop_count, snap, review_path, dict(review_jobs), round_verdicts, stale)
|
|
if stale:
|
|
final_passed_reviewers.clear()
|
|
log_warn(f"source changed during review (head {snap['head'][:12]}, tree {snap['tree'][: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:
|
|
self._accepted_round = self._rounds[-1]
|
|
log_success("All reviewers issued [VERDICT: PASS]. Loop completed successfully.")
|
|
break
|
|
|
|
if self_review:
|
|
log_warn("Self-Review NOT PASS.")
|
|
if loop_count == self.max_loop:
|
|
log_error("Reached max loop count. Self-review loop aborted with failures.")
|
|
return 1
|
|
continue
|
|
|
|
if loop_count == self.max_loop:
|
|
log_error(f"Reached max loop count ({self.max_loop}). Review loop aborted with failures.")
|
|
return 1
|
|
|
|
feedback_aggregate = "".join(feedback_parts)
|
|
complex_fix = any(parser.has_tag_line(rf, "ESCALATE: PLANNER") for rf in fail_reports)
|
|
|
|
if self.plan_mode and complex_fix:
|
|
log_warn("Feedback involves complex code modifications. Diverting to Planner to revise plan...")
|
|
p_type = self.resolve_agent_type(planner_session, state)
|
|
refine_prompt = f"리뷰어들로부터 다음과 같이 정교한 코드 수정 피드백이 도착했습니다. 해당 피드백을 수렴하여 구현 계획서(Plan)를 갱신(Refine)하여 다시 작성해주세요. 피드백:\n{feedback_aggregate}\n기존 계획서:\n{self.current_plan}"
|
|
ref_jid = self._dispatch(planner_session, p_type, "Planner", refine_prompt)
|
|
if not ref_jid or not self.wait_for_job(ref_jid):
|
|
log_error("Failed to register plan refinement job.")
|
|
return 1
|
|
ref_file = self.find_report(ref_jid)
|
|
refined = self._plan_text(ref_file) if ref_file else None
|
|
if refined is None:
|
|
log_warn(f"the refine report has no plan text (job {ref_jid}); keeping the previous plan.")
|
|
else:
|
|
self.current_plan = refined
|
|
# D13 stops a plan-mode run whose initial plan is empty and D14
|
|
# never overwrites with an empty one, so current_plan is not
|
|
# empty here: this prompt cannot claim a plan it does not carry.
|
|
correction_prompt = f"갱신된 다음 계획서에 입각하여 지적된 오류들을 수정하고 코드를 다시 구현해주세요. 계획서:\n{self.current_plan}\n피드백 상세:\n{feedback_aggregate}"
|
|
else:
|
|
log_info("Applying straight bugfixes based on reviewer feedback...")
|
|
correction_prompt = f"리뷰어들이 지적한 다음 피드백에 입각하여 코드를 수정해주세요. 피드백:\n{feedback_aggregate}"
|
|
|
|
if self.max_rebut > 0:
|
|
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):
|
|
log_error("Failed to register Creator correction job.")
|
|
return 1
|
|
|
|
# -----------------------------------------------------------------
|
|
# REBUTTAL & ADJUDICATION ROUND
|
|
# -----------------------------------------------------------------
|
|
if self.max_rebut > 0 and rebut_budget > 0 and rebut_total_budget > 0:
|
|
correct_report = self.find_report(correct_jid)
|
|
rebut_targets: List[str] = []
|
|
if correct_report:
|
|
rebut_tags = parser.extract_tag_values(correct_report, "REBUT")
|
|
for t in rebut_tags:
|
|
if t in fail_revs and t not in rebut_targets:
|
|
rebut_targets.append(t)
|
|
|
|
if rebut_targets:
|
|
rebut_budget -= 1
|
|
rebut_total_budget -= 1
|
|
rebuttal_text = correct_report.read_text(encoding="utf-8", errors="replace") if correct_report else ""
|
|
log_warn(f"Creator filed a rebuttal against: {' '.join(rebut_targets)} (iteration budget left: {rebut_budget}, run total left: {rebut_total_budget})")
|
|
|
|
overruled_revs: List[str] = []
|
|
for rt in rebut_targets:
|
|
log_info(f"Re-adjudication: returning rebuttal to Reviewer '{rt}'...")
|
|
rt_type = self.resolve_agent_type(rt, state)
|
|
adj_prompt = f"작업자(Creator)가 귀하의 리뷰 지적 중 일부를 타당하지 않다고 판단하여 반론을 제기했습니다. 반론을 검토하고 재심(re-adjudication)하십시오. 반론이 타당하여 귀하의 지적을 철회한다면 리포트 마지막에 단독 행으로 '[ADJUDICATION: SUSTAINED]' 를, 반론이 타당하지 않아 지적을 유지한다면 '[ADJUDICATION: OVERRULED]' 를 명시하십시오. 반론 내용:\n{rebuttal_text}"
|
|
adj_jid = self._dispatch(rt, rt_type, "Reviewer", adj_prompt,
|
|
remote=rt in self._remote_recipients, purpose="adjudication")
|
|
if not adj_jid or not self.wait_for_job(adj_jid):
|
|
log_warn(f"Re-adjudication job for '{rt}' failed; objection stands.")
|
|
overruled_revs.append(rt)
|
|
continue
|
|
|
|
adj_file = self.find_report(adj_jid)
|
|
if adj_file and parser.has_tag_line(adj_file, "ADJUDICATION: SUSTAINED"):
|
|
log_success(f"Reviewer '{rt}' SUSTAINED the rebuttal; objection withdrawn.")
|
|
else:
|
|
log_warn(f"Reviewer '{rt}' OVERRULED the rebuttal; objection stands.")
|
|
overruled_revs.append(rt)
|
|
|
|
if overruled_revs:
|
|
arbitration = "REVIEWER"
|
|
if self.plan_mode and planner_session:
|
|
log_info(f"Deadlock on {' '.join(overruled_revs)}; escalating to Planner for arbitration...")
|
|
p_type = self.resolve_agent_type(planner_session, state)
|
|
arb_prompt = f"작업자(Creator)와 리뷰어의 의견이 충돌하여 교착 상태입니다. 계획서 관점에서 어느 쪽이 타당한지 재정(arbitration)하십시오. 리포트 마지막에 단독 행으로 '[ARBITRATION: CREATOR]' 또는 '[ARBITRATION: REVIEWER]' 를 명시하십시오. 리뷰어 지적:\n{feedback_aggregate}\n작업자 반론:\n{rebuttal_text}\n기존 계획서:\n{self.current_plan}"
|
|
arb_jid = self._dispatch(planner_session, p_type, "Planner", arb_prompt)
|
|
if arb_jid and self.wait_for_job(arb_jid):
|
|
arb_file = self.find_report(arb_jid)
|
|
if arb_file and parser.has_tag_line(arb_file, "ARBITRATION: CREATOR"):
|
|
arbitration = "CREATOR"
|
|
else:
|
|
log_warn("Planner arbitration job failed; defaulting to REVIEWER.")
|
|
|
|
if arbitration == "CREATOR":
|
|
log_success("Planner ruled for the Creator; objections withdrawn for this round.")
|
|
else:
|
|
log_warn("Ruling stands with the reviewer(s); Creator must comply.")
|
|
rebut_budget = 0
|
|
comply_prompt = f"귀하의 반론은 재심 결과 기각되었습니다. 더 이상 반론하지 말고 다음 리뷰 피드백을 그대로 반영하여 코드를 수정하십시오. 피드백:\n{feedback_aggregate}"
|
|
comp_jid = self._dispatch(self.creator, creator_type, "Worker", comply_prompt)
|
|
if comp_jid:
|
|
self.wait_for_job(comp_jid)
|
|
|
|
# Promotion
|
|
if self.plan_mode and self.current_plan and self.plan_job_id:
|
|
plan_dest_dir = self.workspace_root / ".agents" / "reports" / planner_session
|
|
log_info(f"Promoting final plan to durable location: {plan_dest_dir}")
|
|
plan_dest_dir.mkdir(parents=True, exist_ok=True)
|
|
tmp_plan = plan_dest_dir / f"plan-{self.plan_job_id}.md.tmp"
|
|
tmp_plan.write_text(self.current_plan, encoding="utf-8")
|
|
tmp_plan.replace(plan_dest_dir / f"plan-{self.plan_job_id}.md")
|
|
|
|
if final_passed_reviewers and identity and self._accepted_round:
|
|
# B10: an approval is promoted only for the source it approved
|
|
acc = self._accepted_round
|
|
if self._identity_now() != (acc["head_sha"], acc["tree_sha"]):
|
|
log_error("source changed after approval; not promoting")
|
|
return 1
|
|
|
|
if final_passed_reviewers:
|
|
log_info("Promoting final review reports to durable location...")
|
|
for rev, (jid, rep_file) in final_passed_reviewers.items():
|
|
dest_dir = self.workspace_root / ".agents" / "reports" / rev
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
dest_file = dest_dir / f"report-{jid}.md"
|
|
tmp_dest = dest_dir / f"report-{jid}.md.tmp"
|
|
shutil.copyfile(rep_file, tmp_dest)
|
|
tmp_dest.replace(dest_file)
|
|
if identity and self._accepted_round:
|
|
acc = self._accepted_round
|
|
ev = self._evidence(rev)
|
|
sidecar = {"loop_id": self.loop_id, "base_sha": acc["base_sha"], "head_sha": acc["head_sha"],
|
|
"tree_sha": acc["tree_sha"], "review_commit": acc["review_commit"],
|
|
"reviewer": rev, "job_id": jid, "verdict": "PASS", **ev,
|
|
"verified_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}
|
|
tmp_side = dest_dir / f"report-{jid}.identity.json.tmp"
|
|
tmp_side.write_text(json.dumps(sidecar, indent=2), encoding="utf-8")
|
|
tmp_side.replace(dest_dir / f"report-{jid}.identity.json")
|
|
log_info(f" {rev}: report-{jid}.md (evidence: {ev['evidence']})")
|
|
|
|
# Cleanup
|
|
if self.cleanup:
|
|
log_info("Cleaning up temporary job directories created during this loop run...")
|
|
for job in self.created_jobs:
|
|
job_dir = self.jobs_dir / job
|
|
if job_dir.exists():
|
|
shutil.rmtree(job_dir, ignore_errors=True)
|
|
sub_out = self.jobs_dir / f"{job}.subscriber.out"
|
|
if sub_out.exists():
|
|
sub_out.unlink(missing_ok=True)
|
|
if self.verbose:
|
|
log_info(f"Purged temp assets for job: {job}")
|
|
log_success("Cleanup complete.")
|
|
|
|
checkpoint.remove_checkpoint(self.checkpoint_file)
|
|
log_success("Mux loop finished with 100% PASS verdicts.")
|
|
return 0
|
|
|
|
|
|
def main(argv: Optional[List[str]] = None) -> int:
|
|
runner = LoopRunner(argv)
|
|
return runner.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|