GodopuandClaude Opus 5.5
bf0f5c1014
- R1 (P1): without rules the input region runs from the last prompt line to the end of the pane; only a last line matching the adapter's footer pattern (codex: ` <model> · ~/<cwd> · …`) is dropped, so a draft on a second line keeps the nudge withheld. - R2: while the synchronous submit runs, a relay thread finds the job it registers (same session, created since the submit) and prints its stalled:/stall_watch: notices; before, they appeared only after the job. - R3: the broker-unavailable disk wait that runs into the deadline is the wall clock: wall_clock_timeout (caller rc stays 3). - R4: a state_change_seq change seen right before the judgement restarts the idle window (decision 9: S seconds of continuous idle). Regression tests are the reviewer's reproductions, each with its passing twin; the wrapper-level #30 tests now use this environment's broker instead of the #18 harness's public default. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1759 lines
92 KiB
Python
1759 lines
92 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 threading
|
|
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", "wall_clock_timeout", "agent_blocked")
|
|
SUBJOB_RULE = ("규칙: 열린(OPEN)·미확인 종결(CLOSED-UNCONFIRMED)·실패 하위 잡이 하나라도 있으면 산출물을 미완으로 보고 "
|
|
"[VERDICT: NOT PASS] 를 내십시오. creator 리포트가 해당 부분의 대체 완료를 입증하면 예외입니다. "
|
|
"CLOSED-UNCONFIRMED 는 오케스트레이터가 마감으로 닫은 잡이며 그 에이전트는 계속 일하고 있을 수 있습니다. "
|
|
"리뷰 중 소스가 바뀌면 이 회차의 승인은 무효가 됩니다.")
|
|
STRAY_RULE = ("규칙: 귀속되지 않은 열린 잡은 --parent-job 없이 등록된 이 작업의 하위 위임일 수 있습니다. 그 잡의 산출물이 "
|
|
"작업 목표에 필요하다면 산출물을 미완으로 보고 [VERDICT: NOT PASS] 를 내십시오.")
|
|
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] = []
|
|
# Issue #34: every creator job of this run (execution, corrections,
|
|
# compliance). Kept across rounds: a later creator job does not make an
|
|
# earlier one's open or failed sub-job go away - that is the reviewers' call.
|
|
self.round_roots: List[str] = []
|
|
self.wait_open_jobs = False
|
|
self.failed_submit_jobs: Set[str] = set()
|
|
self._stall_seen: Dict[str, Tuple[int, Set[str]]] = {}
|
|
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.job_timeout: Optional[int] = None
|
|
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)",
|
|
" --job-timeout <sec> Budget of each delegated job (default: $MAM_JOB_TIMEOUT_S or 3600)",
|
|
" --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",
|
|
" --wait-open-jobs Also wait for open jobs registered without --parent-job during the creator's work",
|
|
]
|
|
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 == "--job-timeout":
|
|
if i + 1 >= len(args) or not re.match(r"^[0-9]+$", args[i + 1]) or int(args[i + 1]) <= 0:
|
|
print("ERROR: --job-timeout requires a positive non-zero integer.", file=sys.stderr)
|
|
sys.exit(1)
|
|
self.job_timeout = 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 == "--wait-open-jobs":
|
|
self.wait_open_jobs = 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)
|
|
# A relative path is taken against the control root, as the usage
|
|
# line says and as jobs_dir already does (Issue #21 R2): the runner
|
|
# never chdirs to workspace_root, so resolving against the caller's
|
|
# cwd would look somewhere else entirely - and could find a
|
|
# same-named file there. `/` keeps an absolute path as it is.
|
|
resolved = (self.workspace_root / 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.
|
|
Issue #33: the job budget is always passed explicitly, so the budget
|
|
the wrapper registers is the one job_wait_s was derived from."""
|
|
args: List[str] = ["--timeout", str(self.job_budget)]
|
|
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) -> subprocess.CompletedProcess:
|
|
# Issue #36: the prompt goes by file. One argv string is capped at
|
|
# MAX_ARG_STRLEN (131,072 bytes on Linux), and a review prompt
|
|
# carries the whole diff. The wrapper blocks until the job ends,
|
|
# so nothing reads the file after it returns.
|
|
prompt_file = ""
|
|
try:
|
|
self.mam_dir.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=self.mam_dir,
|
|
prefix=f"prompt-{self.loop_id}-", suffix=".md",
|
|
delete=False) as fh:
|
|
prompt_file = fh.name
|
|
fh.write(p_text)
|
|
except OSError as e:
|
|
# The same class the shell paths handle (merge review H1): a
|
|
# partial file is removed and the submit fails like an exec
|
|
# error would, instead of the exception reaching the loop.
|
|
if prompt_file:
|
|
try:
|
|
os.unlink(prompt_file)
|
|
except OSError:
|
|
pass
|
|
msg = (f"cannot write the prompt file under {self.mam_dir}: {e} "
|
|
f"(prompt {len(p_text.encode('utf-8'))} bytes)")
|
|
return subprocess.CompletedProcess(["bash", str(wrapper_script), "submit"], 126, "", msg)
|
|
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-file", prompt_file,
|
|
] + self.contract_args()
|
|
env = dict(os.environ)
|
|
if "DELEGATE_JOB_PYTHON" not in env:
|
|
env["DELEGATE_JOB_PYTHON"] = sys.executable
|
|
stop = threading.Event()
|
|
relay = threading.Thread(target=self._relay_while_submitting,
|
|
args=(self.session_label(agent_session), job_registry._utcnow(), stop),
|
|
daemon=True)
|
|
relay.start()
|
|
try:
|
|
return subprocess.run(cmd, capture_output=True, text=True, cwd=str(self.workspace_root), env=env)
|
|
except OSError as e:
|
|
# A wrapper that cannot start is a failed submit (126, the shell's
|
|
# "cannot execute"), on the first try and on the retry alike.
|
|
msg = (f"cannot start the wrapper: {e} "
|
|
f"(prompt {len(p_text.encode('utf-8'))} bytes, wrapper {wrapper_script})")
|
|
return subprocess.CompletedProcess(cmd, 126, "", msg)
|
|
finally:
|
|
stop.set()
|
|
relay.join(timeout=5)
|
|
try:
|
|
os.unlink(prompt_file)
|
|
except OSError:
|
|
pass
|
|
|
|
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
|
|
log_error(f"delegate_job retry failed (exit {res_retry.returncode}):\n{res_retry.stderr or res_retry.stdout}")
|
|
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
|
|
|
|
@property
|
|
def job_budget(self) -> int:
|
|
"""Issue #33: --job-timeout > $MAM_JOB_TIMEOUT_S (positive integer) > 3600."""
|
|
if self.job_timeout:
|
|
return self.job_timeout
|
|
env = os.environ.get("MAM_JOB_TIMEOUT_S", "")
|
|
return int(env) if env.isdigit() and int(env) > 0 else 3600
|
|
|
|
@property
|
|
def job_wait_s(self) -> int:
|
|
"""Wait cap for a job and its sub-jobs: budget + MAM_WAIT_GRACE_S (300,
|
|
today's 3900 - 3600). The grace covers the wrapper's close/publish tail
|
|
after the wall clock and the 1 s record poll; it is not proportional
|
|
to the budget because nothing observed would justify that."""
|
|
grace = os.environ.get("MAM_WAIT_GRACE_S", "")
|
|
return self.job_budget + (int(grace) if grace.isdigit() else 300)
|
|
|
|
def wait_for_job(self, job_id: str, max_wait: Optional[float] = None, check_interval: float = 1.0) -> bool:
|
|
"""Blocking wait for job completion or error with safety timeout."""
|
|
if max_wait is None:
|
|
max_wait = self.job_wait_s
|
|
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
|
|
|
|
self._relay_stall_warnings(job_id)
|
|
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 _relay_while_submitting(self, label: str, since: str, stop: threading.Event) -> None:
|
|
"""Issue #30 (review R2): the wrapper runs until its job ends and its
|
|
output is captured, so a blocked agent would only reach the loop log
|
|
after the job. While the submit runs, find the job it registers (this
|
|
session, created since the submit started) and relay its notices."""
|
|
jid = None
|
|
while not stop.wait(1.0):
|
|
if jid is None:
|
|
try:
|
|
jobs = job_registry.list_jobs(str(self.jobs_dir))
|
|
except (OSError, ValueError):
|
|
continue
|
|
mine = [r for r in jobs if r.get("agent_session") == label
|
|
and str(r.get("created_at") or "") >= since]
|
|
if not mine:
|
|
continue
|
|
jid = str(mine[-1].get("job_id"))
|
|
self._relay_stall_warnings(jid)
|
|
|
|
def _relay_stall_warnings(self, job_id: str) -> None:
|
|
"""Issue #30: surface the stall watcher's progress events (detail
|
|
`stalled:...` / `stall_watch:...`) in the loop log, once per detail.
|
|
Reads events.log incrementally; never closes the job."""
|
|
offset, seen = self._stall_seen.get(job_id, (0, set()))
|
|
try:
|
|
with open(self.jobs_dir / f"{job_id}.events.log", "r", encoding="utf-8") as fh:
|
|
fh.seek(offset)
|
|
lines = fh.readlines()
|
|
offset = fh.tell()
|
|
except OSError:
|
|
return
|
|
for line in lines:
|
|
try:
|
|
ev = json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
detail = str(ev.get("detail") or "")
|
|
if ev.get("event") == "progress" and detail.startswith(("stalled:", "stall_watch:")) \
|
|
and detail not in seen:
|
|
seen.add(detail)
|
|
log_warn(f"job {job_id}: {detail}")
|
|
self._stall_seen[job_id] = (offset, seen)
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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."""
|
|
child_env = os.environ.get("MAM_CHILD_WAIT_S")
|
|
max_wait = float(child_env) if child_env else float(self.job_wait_s)
|
|
wait_src = "MAM_CHILD_WAIT_S" if child_env else \
|
|
f"MAM_CHILD_WAIT_S default: job budget {self.job_budget} + {self.job_wait_s - self.job_budget}"
|
|
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 ({wait_src}) 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", "-"
|
|
# merge review N1: a cancelled job's reason lives in its `cancelled` event
|
|
reason = job_registry.get_feedback(str(r.get("job_id")), str(self.jobs_dir),
|
|
events=("completed", "error", "cancelled")) or ""
|
|
if status == "completed":
|
|
return "completed", reason or "-"
|
|
# merge review R2 (codex): the remote runner phrases its deadline as
|
|
# "deadline_timeout (remote runner)", so the kind is the first word
|
|
closed_kind = reason.split(" ", 1)[0].rstrip(":") if reason else ""
|
|
if closed_kind 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 in ("session_busy", "target_not_found") 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 stray:
|
|
# merge review K1: the case where --parent-job was left out is the one
|
|
# the table cannot show, so the rule has to name it explicitly
|
|
stray_line += STRAY_RULE + "\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
|
|
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, 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.
|
|
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}")
|
|
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:
|
|
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:
|
|
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 _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
|
|
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}")
|
|
|
|
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.")
|
|
|
|
# =======================================================================
|
|
# 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()
|
|
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.
|
|
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)
|
|
subjobs = self._subjobs_block(self.round_roots)
|
|
light = None
|
|
if not identity and subjobs:
|
|
# Issue #34 (D9): the identity mode's B7, armed only when this run
|
|
# has sub-jobs - one closed by deadline may still be writing to the
|
|
# source. A loop without sub-delegation keeps its old behaviour: its
|
|
# source may hold untracked files that change on every job.
|
|
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
|
|
|
|
if not reviewers:
|
|
# Creator Self-Review
|
|
log_warn("No reviewers specified. Conducting Creator Self-Review...")
|
|
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
|
|
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
|
|
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,
|
|
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
|
|
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:
|
|
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 correct_jid:
|
|
self.round_roots.append(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
|
|
|
|
# -----------------------------------------------------------------
|
|
# 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:
|
|
# a job that was never registered is a failure, not a verdict (Issue #36)
|
|
log_error(f"Failed to register re-adjudication job for '{rt}'.")
|
|
return 1
|
|
if 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 not arb_jid:
|
|
log_error("Failed to register Planner arbitration job.")
|
|
return 1
|
|
if 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.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
|
|
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())
|