fix(issue-34): the loop waits for the creator job's descendants, hands reviewers the sub-job table with closing reasons, and voids a round whose source changed during review in every mode

- wait_for_job_tree at the execution, correction and compliance waits;
  bounded by MAM_CHILD_WAIT_S (3900) and MAM_JOB_TREE_MAX (50)
- [하위 위임 잡] table: OPEN / CLOSED-UNCONFIRMED (deadline_timeout,
  agent_blocked, or error with no reason) / NOT-DISPATCHED / error /
  completed, plus unattributed open jobs; omitted when there are none so
  option-less prompts stay byte-identical
- --wait-open-jobs also waits for unattributed open jobs
- stale-review guard without --worktree/--base: head + tree (control
  plane excluded) before review, re-checked after verdicts and before
  promotion; off with a warning when the source cannot be fingerprinted
- tests/test_issue34_job_tree.py: real wrapper registrations, real
  publish_event / deadline close, real loop process

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
GodopuandClaude Opus 5.5 committed 2026-09-24 13:37:05 +00:00
1 parent ec663a93c8
commit 30c8c7ebab
3 files changed
+681 -9

No files matched your search

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