Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ed293dcef4 | ||
|
|
1d4f338b4c | ||
|
|
3fdbb626de | ||
|
|
8690ef4bd2 | ||
|
|
30c8c7ebab | ||
|
|
ec663a93c8 |
No files matched your search
@@ -151,6 +151,7 @@ TMUX 환경에서 실행되는 에이전트가 화면 스크롤 한계로 인해
|
||||
- **작업 위임**:
|
||||
- *수동 경로*: 상세 사양과 계획 수립 등의 복잡한 작업 지시는 먼저 로컬 마크다운 파일(예: `.mam/reports/brief-<job_id>.md` 또는 지정된 워크스페이스 경로)로 작성한 후, 에이전트에게 `"Read <파일경로> and execute."` 라는 실행 명령만 전달하십시오.
|
||||
- *자동 경로*: 자동화 잡 런너(`multi-agent-mux-delegate-job submit`)는 잡 등록 시 `.mam/jobs/<job_id>/brief.md` 디렉터리에 지시서를 자동 집필하고 단일 라인 포인터 프롬프트만 에이전트 세션에 인가합니다.
|
||||
- *하위 위임*: 자기 잡의 일부를 다시 위임하는 에이전트는 `submit` 에 `--parent-job <자기 job_id>` 를 붙입니다. 그래야 루프가 그 하위 잡을 기다리고 리뷰어에게 보여 줍니다(Issue #34).
|
||||
- **결과 안내 및 피드백**:
|
||||
- *수동/영구 리뷰*: 상세 리뷰 결과, 설계 제안서 등은 `.mam/reports/<tmux_session_name>/report-<job_id>.md` 경로에 저장합니다.
|
||||
- *자동화 잡 보고서*: 자동 위임된 비동기 작업의 완료 결과는 잡 디렉터리 하위인 `.mam/jobs/<job_id>/<agent_name>-reports/report-final.md` (루프/Discuss 위임 시에는 `<clean_session_name>-reports/`) 경로에 기록해야 합니다.
|
||||
|
||||
@@ -151,6 +151,7 @@ To ensure that agents running in TMUX environments do not lose debug logs or pre
|
||||
- **Task Delegation**:
|
||||
- *Manual path*: Detailed task briefs may be written to a local Markdown file (e.g., `.mam/reports/brief-<job_id>.md` or a workspace path) first. The sender then issues a simple trigger command: `"Read <file_path> and execute."`
|
||||
- *Automated path*: The automated job runner (`multi-agent-mux-delegate-job submit`) automatically provisions the brief at `.mam/jobs/<job_id>/brief.md` and sends a short pointer instruction to the agent.
|
||||
- *Sub-delegation*: an agent that delegates part of its job passes `--parent-job <its job_id>` to `submit`, so the loop waits for the sub-job and shows it to reviewers (Issue #34).
|
||||
- **Result Reporting & Feedback**:
|
||||
- *Manual/Durable reviews*: Detailed reviews, design proposals, or audit reports must be saved under `.mam/reports/<tmux_session_name>/report-<job_id>.md`.
|
||||
- *Automated job reports*: Automated execution results are saved directly to `.mam/jobs/<job_id>/<agent_name>-reports/report-final.md` (or `<clean_session_name>-reports/` for loops) as transient files.
|
||||
|
||||
@@ -1446,6 +1446,31 @@ _herdr_call_is_cross_server() {
|
||||
return 1
|
||||
}
|
||||
|
||||
# _herdr_scope_is_pane_inherited <herdr argv...>
|
||||
#
|
||||
# ISSUE-34: true for a call that names an agent (agent / has-session /
|
||||
# kill-session) from inside a pane whose own workspace is the exported
|
||||
# HERDR_WORKSPACE_ID. A pane exports it together with HERDR_PANE_ID
|
||||
# "<workspace>:<pane>"; when the two agree the id says where the caller lives,
|
||||
# not where to look - a worktree creator in its own workspace must reach a
|
||||
# reviewer in mam-agent on the same server. Live agent names are unique per
|
||||
# server (herdr --skill: "Names must ... be unique among live agents", "IDs and
|
||||
# live agent names are scoped to one server"), so a name lookup without that
|
||||
# filter cannot pick a same-named agent of another workspace: the ISSUE-3 /
|
||||
# F-2b hazard exists only across servers (_herdr_call_is_cross_server). An id
|
||||
# that disagrees with the pane, or one set outside any pane, is still a
|
||||
# deliberate scope (H-18/H-21/H-23); new-session and pane calls keep it, so a
|
||||
# pane caller still creates panes in its own workspace.
|
||||
_herdr_scope_is_pane_inherited() {
|
||||
while [ "${1:-}" = "-L" ]; do shift 2 || return 1; done
|
||||
case "${1:-}" in
|
||||
agent|has-session|kill-session) ;;
|
||||
*) return 1 ;;
|
||||
esac
|
||||
local pane="${HERDR_PANE_ID:-}"
|
||||
[ -n "${HERDR_WORKSPACE_ID:-}" ] && [ -n "$pane" ] && [ "${pane%%:*}" = "$HERDR_WORKSPACE_ID" ]
|
||||
}
|
||||
|
||||
mam_herdr() {
|
||||
_init_herdr_isolation
|
||||
# Built as an argv array and decided inside `if`: this runs under
|
||||
@@ -1457,6 +1482,11 @@ mam_herdr() {
|
||||
if [ "${MAM_SHIM_DEBUG:-0}" = "1" ]; then
|
||||
echo "mam_herdr: dropping caller pane scope ids (caller server '$(_herdr_caller_server)' != target '${HERDR_SESSION_NAME:-}')" >&2
|
||||
fi
|
||||
elif _herdr_scope_is_pane_inherited "$@"; then
|
||||
_scope_strip=(env -u HERDR_WORKSPACE_ID)
|
||||
if [ "${MAM_SHIM_DEBUG:-0}" = "1" ]; then
|
||||
echo "mam_herdr: '$1' looks up by name; the caller pane's own workspace ${HERDR_WORKSPACE_ID} is not a scope" >&2
|
||||
fi
|
||||
fi
|
||||
${_scope_strip[@]+"${_scope_strip[@]}"} "$WORKSPACE_ROOT/.mam/shim/herdr" "$@"
|
||||
}
|
||||
|
||||
@@ -89,6 +89,7 @@ def register_job(
|
||||
attempt_id: int = 1,
|
||||
transport: str = "herdr",
|
||||
supersede_older_attempt: bool = False,
|
||||
parent_job: Optional[str] = None,
|
||||
) -> str:
|
||||
"""Create a new ``pending`` job record and return its id.
|
||||
|
||||
@@ -102,6 +103,9 @@ def register_job(
|
||||
case it passes the orchestrator's so the events it publishes carry the
|
||||
identity the orchestrator's subscriber expects. ``transport`` records how
|
||||
the prompt reaches the agent (``herdr`` pane or ``remote`` dispatch topic).
|
||||
|
||||
Issue #34: ``parent_job`` names the job this one was delegated from, so a
|
||||
loop waiting on the parent can find and wait for it. None is a root job.
|
||||
"""
|
||||
job_id = job_id or generate_job_id(bits)
|
||||
heartbeat_sec = max(1, int(heartbeat_sec))
|
||||
@@ -145,6 +149,8 @@ def register_job(
|
||||
"heartbeat_misses": heartbeat_misses,
|
||||
"ack_timeout_sec": max(1, int(ack_timeout_sec)),
|
||||
"transport": transport,
|
||||
# Issue #34: the job this one was delegated from (None = a root job)
|
||||
"parent_job": parent_job,
|
||||
}
|
||||
with registry_lock(registry_dir):
|
||||
if mqtt_common._job_path(job_id, registry_dir).exists():
|
||||
@@ -245,6 +251,36 @@ def list_jobs(registry_dir: str = DEFAULT_REGISTRY_DIR, status: Optional[str] =
|
||||
return records
|
||||
|
||||
|
||||
def children_of(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> List[Dict[str, Any]]:
|
||||
"""Issue #34: the records whose ``parent_job`` is ``job_id``, oldest first."""
|
||||
return [r for r in list_jobs(registry_dir) if r.get("parent_job") == job_id]
|
||||
|
||||
|
||||
def descendants_of(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> List[Dict[str, Any]]:
|
||||
"""Issue #34: every job delegated from ``job_id``, directly or through its
|
||||
children, breadth first. Each record gets a ``depth`` (1 = a child). One
|
||||
scan of the registry; a record is visited once, so a cycle cannot loop."""
|
||||
by_parent: Dict[str, List[Dict[str, Any]]] = {}
|
||||
for r in list_jobs(registry_dir):
|
||||
if r.get("parent_job"):
|
||||
by_parent.setdefault(r["parent_job"], []).append(r)
|
||||
out: List[Dict[str, Any]] = []
|
||||
seen = {job_id}
|
||||
frontier, depth = [job_id], 0
|
||||
while frontier:
|
||||
depth += 1
|
||||
nxt = []
|
||||
for pid in frontier:
|
||||
for r in by_parent.get(pid, []):
|
||||
if r.get("job_id") in seen:
|
||||
continue
|
||||
seen.add(r["job_id"])
|
||||
out.append(dict(r, depth=depth))
|
||||
nxt.append(r["job_id"])
|
||||
frontier = nxt
|
||||
return out
|
||||
|
||||
|
||||
def append_event(job_id: str, registry_dir: str, payload: Dict[str, Any]) -> None:
|
||||
"""Append one event payload as a JSON line to the job's events log. Best
|
||||
effort, debug-only; failures are logged but never raised to the caller."""
|
||||
@@ -261,6 +297,7 @@ def append_event(job_id: str, registry_dir: str, payload: Dict[str, Any]) -> Non
|
||||
__all__ = [
|
||||
"register_job", "pick_pending", "update_status", "load_job",
|
||||
"list_jobs", "append_event", "generate_job_id", "get_feedback",
|
||||
"children_of", "descendants_of",
|
||||
]
|
||||
|
||||
|
||||
@@ -353,9 +390,12 @@ def _build_parser() -> argparse.ArgumentParser:
|
||||
p_reg.add_argument("--reviewer", default=None)
|
||||
p_reg.add_argument("--reviewer-session", default=None)
|
||||
p_reg.add_argument("--max-iterations", type=int, default=5)
|
||||
p_reg.add_argument("--parent-job", default=None,
|
||||
help="the job this one is delegated from (Issue #34); must exist in this registry")
|
||||
|
||||
p_list = sub.add_parser("list", help="list jobs (optionally by status)")
|
||||
p_list.add_argument("--status", default=None)
|
||||
p_list.add_argument("--parent", default=None, help="only the direct children of this job")
|
||||
p_list.add_argument("--json", action="store_true")
|
||||
|
||||
p_get = sub.add_parser("get", help="print one job record as JSON")
|
||||
@@ -497,6 +537,11 @@ def main(argv: Optional[List[str]] = None) -> int:
|
||||
rd = args.registry_dir
|
||||
|
||||
if args.command == "register":
|
||||
# Issue #34: a mistyped parent must not silently make a root job - the
|
||||
# loop would then not wait for it, which is the bug this field fixes.
|
||||
if args.parent_job and not mqtt_common._job_path(args.parent_job, rd).exists():
|
||||
print(f"unknown parent job {args.parent_job}", file=sys.stderr)
|
||||
return 1
|
||||
job_id = register_job(
|
||||
prompt=_prompt_from_args(parser, args),
|
||||
agent=args.agent,
|
||||
@@ -519,12 +564,15 @@ def main(argv: Optional[List[str]] = None) -> int:
|
||||
dispatch_token=args.dispatch_token,
|
||||
attempt_id=args.attempt_id,
|
||||
job_id=args.job_id,
|
||||
parent_job=args.parent_job,
|
||||
)
|
||||
print(job_id)
|
||||
return 0
|
||||
|
||||
if args.command == "list":
|
||||
records = list_jobs(rd, status=args.status)
|
||||
if args.parent:
|
||||
records = [r for r in records if r.get("parent_job") == args.parent]
|
||||
if args.json:
|
||||
print(json.dumps(records, ensure_ascii=False, indent=2))
|
||||
else:
|
||||
|
||||
@@ -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,11 @@ class LoopRunner:
|
||||
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.current_plan = ""
|
||||
self.plan_job_id: Optional[str] = None
|
||||
@@ -128,6 +144,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 +219,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 +619,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 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 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 +890,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 +909,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 +951,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 +1287,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 +1311,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 +1342,35 @@ class LoopRunner:
|
||||
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 = 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 +1401,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 +1460,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 +1515,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.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
|
||||
|
||||
@@ -1404,10 +1586,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
|
||||
|
||||
@@ -642,7 +642,8 @@ On progress (optional): $pub --event progress --detail '<short status>'.
|
||||
On success run: $pub --event completed --detail '<one-line summary>'.
|
||||
On failure run: $pub --event error --detail '<one-line reason>'.
|
||||
|
||||
Task: read the complete task in $task_brief and follow it."
|
||||
Task: read the complete task in $task_brief and follow it.
|
||||
If you delegate any part of it with multi-agent-mux-delegate-job submit, add --parent-job $DELEGATE_JOB_ID so the orchestrator waits for it."
|
||||
|
||||
# Inject instructions into the herdr pane
|
||||
rc=0
|
||||
|
||||
@@ -46,6 +46,11 @@ multi-agent-mux-delegate-job submit \
|
||||
# must arrive within T s of the injection or the inject fails closed.
|
||||
# --prompt-file <path> replaces --prompt for long prompts (Issue #36): one
|
||||
# argv string is capped at 131,072 bytes on Linux. The loop always uses it.
|
||||
# --parent-job <id> (Issue #34): this job is delegated from job <id>, which
|
||||
# must exist in the registry (else exit 1). A loop waiting on <id> waits for
|
||||
# it too and shows it to reviewers. Default: $MAM_JOB_ID (the remote runner
|
||||
# sets it), ignored with a warning when the registry lacks that job.
|
||||
# `registry.py list --parent <id>` lists a job's direct children.
|
||||
|
||||
# 1b) Same, for an agent on another host (no herdr pane, no shared disk):
|
||||
# the brief travels over the broker as a signed retained dispatch and a
|
||||
@@ -165,6 +170,7 @@ Job lifecycle execution events are persistently mirrored to an append-only log u
|
||||
## Best Practices and Pitfalls
|
||||
|
||||
- **Subscribe-Before-Publish**: The subscriber must be running before the agent starts publishing. The `submit` command handles this automatically by launching the subscriber in the background first.
|
||||
- **Sub-delegation carries `--parent-job` (Issue #34)**: an agent that hands part of its job to another agent must pass `--parent-job <its own job_id>` (the injected instructions and `brief.md` say so). Without it the loop does not wait for the sub-job; reviewers see it only as an unattributed open job. From a git worktree the worktree's copy of this wrapper runs; the pane's `MAM_JOBS_DIR` keeps the registry the same one. The target may live in another herdr workspace of the same server: inside a pane, the pane's own `HERDR_WORKSPACE_ID` (the one matching `HERDR_PANE_ID`) is not used as a scope for name lookups, since live agent names are unique per server; a `HERDR_WORKSPACE_ID` you set yourself still scopes. A target the preflight cannot see is cancelled with detail `target_not_found` (exit 5), not `session_busy`.
|
||||
- **Fresh job_id Propagation**: Make sure the worker agent receives the correct `JOB_ID` generated for the current run, rather than reusing stale IDs from previous sessions.
|
||||
- **Brief delivery via file path**: For long or complex prompts, write the instructions to a file (e.g. `/tmp/task-brief.md`) and pass a short prompt pointing to the file path to prevent terminal buffer overflows.
|
||||
- **Prompts injected into a live agent session MUST be English, ASCII-only, and short** — this is exactly what `--prompt`/the `instructions` string sent to `run_agent()` end up as. Any Korean (or other non-ASCII) content the task needs to convey must go in a markdown brief file (e.g. `.mam/jobs/<id>/brief.md`, written in Korean is fine) that the injected prompt merely tells the agent to read. Two independent bugs in `send_keys_safe`'s paste-verification (in `lib.sh`) made this matter in practice: (a) its marker was taken with a byte-based `tail -c 24`, which can slice a multi-byte UTF-8 (e.g. Korean) character in half; (b) the rendered pane soft-wraps long lines at the terminal width, which can split the marker across two visual lines. Both are now fixed at the source (character-safe truncation + newline-stripped matching before comparison), but keeping injected prompts short/English/file-referencing is still the cheapest way to avoid ever exercising this edge case at all — it's also simply what `submit`'s own default instruction template already does (see `Core Commands` above).
|
||||
|
||||
@@ -98,11 +98,12 @@ _pub_cmd() {
|
||||
}
|
||||
|
||||
# _instructions <pub> [iteration]: the prompt injected into the agent (direct
|
||||
# and loop jobs share it; test18 pins parts of this wording).
|
||||
# and loop jobs share it; test18 pins parts of this wording, so Issue #34's
|
||||
# sub-delegation sentence goes at the end).
|
||||
_instructions() {
|
||||
local pub="$1" iter="${2:-}" brief="$REGISTRY_DIR/$JOB_ID/brief.md"
|
||||
printf 'Your job_id is "%s". Detailed task requirements, instructions, and target output paths%s are documented in the task brief file at: %s. Please READ and follow %s to complete your work. Commands: start='"'"'%s --event started --heartbeat %s'"'"' (run it FIRST, within %s s; it also starts an automatic heartbeat timer), success='"'"'%s --event completed --detail <summary>'"'"', error='"'"'%s --event error --detail <reason>'"'"'.' \
|
||||
"$JOB_ID" "${iter:+ for iteration $iter}" "$brief" "$brief" "$pub" "$HEARTBEAT_SEC" "$ACK_TIMEOUT" "$pub" "$pub"
|
||||
printf 'Your job_id is "%s". Detailed task requirements, instructions, and target output paths%s are documented in the task brief file at: %s. Please READ and follow %s to complete your work. Commands: start='"'"'%s --event started --heartbeat %s'"'"' (run it FIRST, within %s s; it also starts an automatic heartbeat timer), success='"'"'%s --event completed --detail <summary>'"'"', error='"'"'%s --event error --detail <reason>'"'"'. If you delegate any part of this work with multi-agent-mux-delegate-job submit, add --parent-job %s so the orchestrator waits for it.' \
|
||||
"$JOB_ID" "${iter:+ for iteration $iter}" "$brief" "$brief" "$pub" "$HEARTBEAT_SEC" "$ACK_TIMEOUT" "$pub" "$pub" "$JOB_ID"
|
||||
}
|
||||
|
||||
usage() {
|
||||
@@ -115,7 +116,11 @@ multi-agent-mux-delegate-job <command> [options]
|
||||
[--registry-dir <dir>] [--dry-run] [--role <role_name>]
|
||||
[--type <direct|loop|discuss>] [--reviewer <reviewer_agent>]
|
||||
[--reviewer-session <reviewer_session>] [--max-iterations <count>]
|
||||
[--counterpart-role <role_name>] [--strict-role-check]
|
||||
[--counterpart-role <role_name>] [--strict-role-check] [--parent-job <id>]
|
||||
# --parent-job names the job this one is delegated from, so a loop
|
||||
# waiting on that job waits for this one too (Issue #34). It must
|
||||
# exist in the registry. Default: $MAM_JOB_ID (set by the remote
|
||||
# runner), ignored with a warning when the registry lacks it.
|
||||
# --prompt-file passes the prompt by file: one argv string is capped
|
||||
# at 131,072 bytes on Linux (Issue #36). The file is read, not kept.
|
||||
# --agent-session herdr:<name> injects into a local herdr pane;
|
||||
@@ -144,6 +149,7 @@ COUNTERPART_ROLE="$DEFAULT_COUNTERPART_ROLE"
|
||||
STRICT_ROLE_CHECK=0
|
||||
ROLE_ALIASES_JSON='{"worker": ["worker", "creator"], "planner": ["planner"], "reviewer": ["reviewer"]}'
|
||||
COUNTERPART_ROLE_EXPLICIT=0
|
||||
PARENT_JOB=""
|
||||
|
||||
parse_opts() {
|
||||
while [[ $# -gt 0 ]]; do
|
||||
@@ -169,6 +175,7 @@ parse_opts() {
|
||||
--max-iterations) MAX_ITERATIONS="$2"; shift 2;;
|
||||
--counterpart-role) COUNTERPART_ROLE="$2"; COUNTERPART_ROLE_EXPLICIT=1; shift 2;;
|
||||
--strict-role-check) STRICT_ROLE_CHECK=1; shift;;
|
||||
--parent-job) PARENT_JOB="$2"; shift 2;;
|
||||
*) echo "unknown option: $1" >&2; usage; exit 1;;
|
||||
esac
|
||||
done
|
||||
@@ -204,6 +211,20 @@ cmd_submit() {
|
||||
_abs_registry_dir
|
||||
LOGS_DIR_ABS="$(_logs_dir_abs 2>/dev/null || true)"
|
||||
|
||||
# Issue #34: an explicit --parent-job fails closed in the registry; the
|
||||
# $MAM_JOB_ID default may name a job of another host's registry (a remote
|
||||
# mirror), so it is only used when this registry has it.
|
||||
local reg_parent=()
|
||||
if [[ -n "$PARENT_JOB" ]]; then
|
||||
reg_parent=(--parent-job "$PARENT_JOB")
|
||||
elif [[ -n "${MAM_JOB_ID:-}" ]]; then
|
||||
if [[ -f "$REGISTRY_DIR/$MAM_JOB_ID.json" ]]; then
|
||||
PARENT_JOB="$MAM_JOB_ID"; reg_parent=(--parent-job "$PARENT_JOB")
|
||||
else
|
||||
echo "WARNING: MAM_JOB_ID=$MAM_JOB_ID is not in $REGISTRY_DIR; registering without a parent job" >&2
|
||||
fi
|
||||
fi
|
||||
|
||||
# 1) register job (prints the new job id)
|
||||
JOB_ID="$("$PY" "$SCRIPT_DIR/scripts/registry.py" --registry-dir "$REGISTRY_DIR" register \
|
||||
"${reg_prompt[@]}" --agent "$AGENT" --agent-session "$AGENT_SESSION" --role "$DELEGATE_ROLE" \
|
||||
@@ -211,7 +232,7 @@ cmd_submit() {
|
||||
--heartbeat-sec "$HEARTBEAT_SEC" --heartbeat-misses "$HEARTBEAT_MISSES" --ack-timeout "$ACK_TIMEOUT" \
|
||||
--transport "$(_is_remote "$AGENT_SESSION" && echo remote || echo herdr)" \
|
||||
--job-type "$TYPE" --reviewer "$REVIEWER" --reviewer-session "$REVIEWER_SESSION" \
|
||||
--max-iterations "$MAX_ITERATIONS")"
|
||||
--max-iterations "$MAX_ITERATIONS" ${reg_parent[@]+"${reg_parent[@]}"})"
|
||||
echo "registered job: $JOB_ID"
|
||||
|
||||
# 1-1) Provision job directory and write direct brief.md (MAM Job Restructuring)
|
||||
@@ -227,6 +248,7 @@ cmd_submit() {
|
||||
- **Timeout**: $TIMEOUT s (Idle: $IDLE_TIMEOUT s)
|
||||
- **Liveness contract**: heartbeat every $HEARTBEAT_SEC s (automatic once you publish \`started --heartbeat $HEARTBEAT_SEC\`), unresponsive after $HEARTBEAT_MISSES missed heartbeats, \`started\` expected within $ACK_TIMEOUT s
|
||||
- **Output Report Path**: $REGISTRY_DIR/$JOB_ID/$AGENT-reports/report-final.md
|
||||
- **Parent for sub-delegations**: pass \`--parent-job $JOB_ID\` to \`multi-agent-mux-delegate-job submit\`
|
||||
|
||||
## 🔎 Task Description
|
||||
$PROMPT
|
||||
@@ -402,6 +424,7 @@ EOF
|
||||
- **Iteration**: $iteration
|
||||
- **Liveness contract**: heartbeat every $HEARTBEAT_SEC s (automatic once you publish \`started --heartbeat $HEARTBEAT_SEC\`), unresponsive after $HEARTBEAT_MISSES missed heartbeats, \`started\` expected within $ACK_TIMEOUT s
|
||||
- **Output Report Path**: $REGISTRY_DIR/$JOB_ID/${clean_session}-reports/report-final.md
|
||||
- **Parent for sub-delegations**: pass \`--parent-job $JOB_ID\` to \`multi-agent-mux-delegate-job submit\`
|
||||
|
||||
## 🔎 Task Description
|
||||
$current_prompt
|
||||
@@ -869,6 +892,8 @@ finalize_unfinished() {
|
||||
if [ "$phase" = "preflight" ] || [ "$phase" = "idle_wait" ]; then
|
||||
close_req="cancelled"
|
||||
detail="session_busy"
|
||||
# Issue #34: a target the preflight could not see was never busy
|
||||
[ "$phase" = "preflight" ] && [ -n "${PREFLIGHT_REASON:-}" ] && detail="$PREFLIGHT_REASON"
|
||||
fi
|
||||
|
||||
local close_args=(--job "$job_id" --request "$close_req")
|
||||
@@ -898,7 +923,7 @@ finalize_unfinished() {
|
||||
local disp_arg=()
|
||||
[ -n "${_disp_tok:-}" ] && disp_arg=(--expect-dispatch-token "$_disp_tok")
|
||||
if [ "$close_req" = "cancelled" ]; then
|
||||
"$PY" "$SCRIPT_DIR/scripts/publish_event.py" --registry-dir "$REGISTRY_DIR" --job "$job_id" --event cancelled --detail "session_busy" ${disp_arg[@]+"${disp_arg[@]}"} 2>/dev/null || true
|
||||
"$PY" "$SCRIPT_DIR/scripts/publish_event.py" --registry-dir "$REGISTRY_DIR" --job "$job_id" --event cancelled --detail "$detail" ${disp_arg[@]+"${disp_arg[@]}"} 2>/dev/null || true
|
||||
elif [ "$sub_rc" -ne 1 ] || [ "$phase" = "inject" ]; then
|
||||
local err_detail="deadline_timeout"
|
||||
[ "$phase" = "inject" ] && err_detail="agent_blocked"
|
||||
@@ -924,6 +949,7 @@ preflight_agent() {
|
||||
export HERDR_SESSION_NAME="$(resolve_herdr_session "$sess" "$WORKDIR")"
|
||||
|
||||
if ! herdr has-session -t "$sess" 2>/dev/null; then
|
||||
PREFLIGHT_REASON="target_not_found" # Issue #34: not a busy session
|
||||
echo "ERROR: 에이전트 세션 '$sess'이 존재하지 않습니다. 작업을 위임하기 전에 먼저 에이전트 세션을 기동해 주세요." >&2
|
||||
echo " 팁: 'multi-agent-mux-resume' 또는 'multi-agent-mux-create'를 통해 에이전트를 먼저 생성할 수 있습니다." >&2
|
||||
return 1
|
||||
|
||||
@@ -211,12 +211,17 @@ sequenceDiagram
|
||||
| **Diff base** | `--base <ref>` (CLI 전용, env 없음) | 리뷰 diff 의 base 를 고정합니다. 소스 HEAD 의 조상이어야 하며 아니면 exit 1. 생략 시 Phase 2 시작 시점의 소스 HEAD 입니다. |
|
||||
| **Review only** | `--review-only` | Phase 2(구현 위임)를 건너뛰고 `<base>..HEAD` 를 리뷰합니다. `--base` 필수, `--plan` 과 함께 쓸 수 없습니다. 수정·순응 잡은 여전히 creator 에게 갑니다. |
|
||||
| **Review isolation** | `--review-isolation worktree\|env` | `--worktree`/`--base` 에서만 의미. `worktree`(기본): 라운드마다 스냅샷의 detached 리뷰 워크트리(TMPDIR)에서 검증합니다. `env`: 소스 루트에서 `PYTHONDONTWRITEBYTECODE=1 … -p no:cacheprovider --basetemp=<밖>` 지침으로 검증합니다. 리뷰 워크트리를 열 수 없으면 조용히 낮추지 않고 exit 1. `env` 에서 리뷰어가 ignore 되지 않는 산출물을 소스 루트에 남기면 라운드가 stale 로 판정돼 재리뷰가 반복됩니다(fail-closed) — 산출물은 `--basetemp` 등으로 소스 밖에 둡니다. |
|
||||
| **Sub-jobs** | `--wait-open-jobs` | creator 잡의 하위 잡(아래 "하위 위임 잡") 외에, creator 작업 중 `--parent-job` 없이 등록된 열린 잡도 기다립니다. 기본 off: 같은 제어 평면의 무관한 잡이 루프를 붙잡을 수 있기 때문입니다. Python 루프 전용. |
|
||||
| **Remote reviewers** | `--remote-review refuse\|diff-only` | `--worktree`/`--base` 에서만 의미. 원격 리뷰어(`transport: remote`)는 이 호스트의 리뷰 트리에 닿을 수 없으므로 기본으로 **어떤 잡도 위임하기 전에** 거부(exit 1)합니다. `diff-only` 는 스냅샷 SHA 와 diff 만을 근거로 받아들이고 그렇게 기록합니다. |
|
||||
|
||||
`--worktree` 또는 `--base` 가 있으면 매 리뷰 라운드(셀프 리뷰 포함) 전에 소스 루트의 변경분을 git 종료 코드로 판정합니다. 변경분이 비었거나 판정할 수 없으면 리뷰를 요청하지 않고 exit 1 로 끝납니다(fail-closed). 둘 다 없으면 기존 동작 그대로입니다.
|
||||
|
||||
**리뷰 identity (Issue #21 항목 5).** `--worktree`/`--base` 가 있으면 매 리뷰 라운드를 `(loop_id, base, head, tree)` 에 묶습니다. 소스의 미커밋·untracked 상태까지 담은 스냅샷(ref 없는 커밋)을 만들고, 리뷰어가 받는 diff 도 그 스냅샷에서 만듭니다(`base..snapshot`). 라운드가 끝난 뒤 소스가 바뀌었으면 그 라운드의 PASS 는 버리고 수정 잡 없이 다시 리뷰합니다(마지막 반복이면 exit 1). 승격 직전에도 한 번 더 확인하며, 승격된 리포트 옆에 `report-<jid>.identity.json`(loop_id·base·head·tree·스냅샷 커밋·리뷰어·`transport`·`evidence`: `review-tree`/`source-tree`/`diff-only`)을 둡니다. 추적되지 않는 중첩 git 저장소는 스냅샷에 담을 수 없어 거부합니다. 서브모듈 안의 미커밋·untracked 내용도 스냅샷에는 gitlink 커밋만 남으므로 거부합니다(라운드 시작 시 exit 1, 리뷰 중에 생기면 그 라운드는 stale). `submodule.*.ignore` 설정으로 가려지지 않습니다. 리뷰어가 없으면 creator 셀프 리뷰가 같은 리뷰어 경로(스냅샷 diff·증거 검사·승격 직전 재확인·사이드카)를 거치고, NOT PASS 면 수정 잡 없이 다시 리뷰합니다. 리뷰 잡의 `[워크트리 컨텍스트]` 는 검증 대상을 `[리뷰 identity]` 의 검증 경로 하나로만 지시합니다.
|
||||
|
||||
**하위 위임 잡 (Issue #34).** creator 가 작업 일부를 `delegate-job submit --parent-job <자기 job_id>` 로 맡기면, 루프는 creator 잡(실행·수정·순응)이 끝난 뒤 그 잡의 자손(자식·손자…, 폴링마다 다시 읽음)의 레코드가 모두 종결될 때까지 기다린 다음 리뷰로 넘어갑니다. 깊이 제한은 없고 상한은 둘입니다: `MAM_CHILD_WAIT_S`(기본 3900 s, creator 잡 종결부터; 각 하위 잡은 자기 등록 + `timeout_sec`(#33, 기본 3600 s)에 닫히므로 보통 그 전에 끝납니다)와 `MAM_JOB_TREE_MAX`(기본 50 개, 넘으면 기다리지 않음). 그만큼 루프 벽시계가 늘어납니다. 루프를 잡으로 감싸 돌리면 바깥 잡의 3600 s 가 먼저 끝날 수 있습니다. 자식의 결과는 루프를 멈추지 않고 리뷰 근거가 됩니다: 하위 잡이나 귀속되지 않은 열린 잡이 있으면 모든 리뷰 프롬프트(셀프 리뷰 포함) 앞에 `[하위 위임 잡]` 표(상태·사유·깊이)와 `[귀속되지 않은 열린 잡]`(부모 미표기로 이 작업 중 등록된 열린 잡) 줄이 붙고(둘 다 없으면 프롬프트는 전과 같습니다), 리뷰어는 `OPEN`·`CLOSED-UNCONFIRMED`·실패(`error`/`cancelled`/`NOT-DISPATCHED`) 하위 잡이 있으면 creator 리포트가 대체 완료를 입증하지 않는 한 NOT PASS 를 냅니다. 표는 이 실행의 모든 creator 잡(앞 회차의 실행·수정·순응 잡 포함)의 자손을 매 회차 다시 읽어 만듭니다 — 수정 잡이 새로 생겼다고 앞 잡의 열린·실패 하위 잡이 사라지지 않습니다(대기는 새 creator 잡의 자손만 하므로 대기 상한이 회차마다 다시 쌓이지 않습니다). `CLOSED-UNCONFIRMED` 는 오케스트레이터가 마감(`deadline_timeout`/`agent_blocked`)으로 닫았거나 사유가 없는 `error` 입니다 — 레코드만 닫혔고 에이전트는 인터럽트되지 않으므로 계속 쓰고 있을 수 있습니다. creator 잡 자체의 실패는 전처럼 exit 1 입니다.
|
||||
|
||||
**리뷰 중 소스 변경 = 승인 무효 (하위 잡이 있을 때, Issue #34).** `--worktree`/`--base` 가 없어도 이 실행에 하위 잡(또는 귀속되지 않은 열린 잡)이 있는 라운드마다 리뷰 요청 직전 소스(= 제어 루트)의 head 와 미커밋·untracked 를 포함한 tree 를 기록하고, 판정 수집 뒤와 승격 직전에 다시 계산합니다. 바뀌었으면 위 identity 모드와 같이 그 라운드의 PASS 를 버리고 수정 잡 없이 다시 리뷰합니다(마지막 반복이면 exit 1). 제어 평면(`.mam`, 잡·감사 로그 디렉터리)은 비교에서 뺍니다. 리뷰어가 ignore 되지 않는 산출물을 소스에 남기면 라운드가 stale 이 되니 산출물은 소스 밖에 둡니다. 하위 잡이 없는 루프는 이 검사를 하지 않습니다(전과 같음) — 소스에 잡마다 바뀌는 untracked 파일이 있을 수 있기 때문입니다. git 저장소가 아니거나 서브모듈이 dirty 해 지문을 만들 수 없으면 경고(`stale-review guard is off`) 후 기존 동작입니다. 루프가 끝난 **뒤**의 늦은 쓰기는 막을 수 없으며 다음 루프의 diff 에 미커밋 변경으로 드러납니다.
|
||||
|
||||
구현이 루프 밖(워크트리의 creator 에게 `delegate-job --type direct`)에서 끝났다면, **구현 위임 전에** base 를 기록해 두고 리뷰만 돌립니다:
|
||||
|
||||
```bash
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -0,0 +1,598 @@
|
||||
"""Issue #34: the loop waits for the creator job's sub-jobs, reviewers see them.
|
||||
|
||||
A creator that hands part of its work to another agent (delegate-job submit)
|
||||
used to end its own job while the sub-job was still running; the loop went
|
||||
straight to review and the reviewers passed half-finished work. Jobs now record
|
||||
their parent (`submit --parent-job`), the loop waits for the creator job's
|
||||
sub-job tree, the review prompt carries a sub-job table, and a round whose
|
||||
source changed during review is void in every mode (a sub-job closed by its
|
||||
deadline is not interrupted and may still be writing).
|
||||
|
||||
Sub-jobs are registered by the real wrapper (`submit --dry-run`, which returns
|
||||
after the real registry wrote the record) and closed by the real producers:
|
||||
`publish_event.py`, and for a deadline the wrapper's own sequence
|
||||
(`registry.py close --request deadline_timeout`, then `publish_event --event
|
||||
error --detail deadline_timeout`). The loop is a real `python -m lib_py.loop`
|
||||
process; only the loop's own agents are simulated, by an executable behind a
|
||||
wrapper at the control root (the #36 harness).
|
||||
|
||||
T34_SKILLS points the harness at another copy of .agents/skills (used once to
|
||||
run T3/T5b against the base runner). Every test writes only under tmp_path.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
SKILLS = Path(os.environ.get("T34_SKILLS") or REPO_ROOT / ".agents" / "skills")
|
||||
WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
||||
REGISTRY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "registry.py"
|
||||
PUBLISH = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"
|
||||
|
||||
if str(SKILLS) not in sys.path:
|
||||
sys.path.insert(0, str(SKILLS))
|
||||
|
||||
UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}
|
||||
GIT_ID = {"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@example.invalid",
|
||||
"GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@example.invalid"}
|
||||
|
||||
|
||||
def _env(home: Path, extra=None, drop=()) -> dict:
|
||||
env = dict(os.environ)
|
||||
env["DELEGATE_JOB_PYTHON"] = sys.executable
|
||||
env["AGENT_PYTHON_BIN"] = sys.executable
|
||||
env["HOME"] = str(home)
|
||||
env["MAM_ENV_FILE"] = str(home / ".mam.env.absent")
|
||||
env["PYTHONPATH"] = str(SKILLS)
|
||||
for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD",
|
||||
"MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR", "WORKSPACE_ROOT", "MAM_REAL_ROOT", "MAM_JOB_ID",
|
||||
"MAM_CHILD_WAIT_S", "MAM_JOB_TREE_MAX"):
|
||||
env.pop(k, None)
|
||||
env.update(UNREACHABLE)
|
||||
env.update(GIT_ID)
|
||||
env["DELEGATE_JOB_LOGS_DIR"] = str(home / ".mam-logs")
|
||||
env["WORKSPACE_ROOT"] = str(home)
|
||||
if extra:
|
||||
env.update({k: str(v) for k, v in extra.items()})
|
||||
for k in drop:
|
||||
env.pop(k, None)
|
||||
return env
|
||||
|
||||
|
||||
def _git(cwd: Path, *args: str) -> str:
|
||||
return subprocess.run(["git", "-C", str(cwd), *args], capture_output=True, text=True,
|
||||
check=True, env={**os.environ, **GIT_ID}).stdout.strip()
|
||||
|
||||
|
||||
def _repo(path: Path) -> Path:
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
_git(path, "init", "-q")
|
||||
(path / "a.txt").write_text("a\n")
|
||||
_git(path, "add", "a.txt")
|
||||
_git(path, "commit", "-q", "-m", "init")
|
||||
return path
|
||||
|
||||
|
||||
def _submit(env: dict, reg: Path, *extra: str) -> subprocess.CompletedProcess:
|
||||
"""The real wrapper; --dry-run returns right after the real registration."""
|
||||
return subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--agent-session", "herdr:img",
|
||||
"--dry-run", "--registry-dir", str(reg), "--prompt", "draw the figures", *extra],
|
||||
capture_output=True, text=True, env=env, cwd=env["HOME"], timeout=120)
|
||||
|
||||
|
||||
def _jid(res: subprocess.CompletedProcess) -> str:
|
||||
assert res.returncode == 0, res.stdout + res.stderr
|
||||
m = re.search(r"registered job:\s*(\S+)", res.stdout)
|
||||
assert m, res.stdout
|
||||
return m.group(1)
|
||||
|
||||
|
||||
def _record(reg: Path, jid: str) -> dict:
|
||||
return json.loads((reg / f"{jid}.json").read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def _publish(env: dict, reg: Path, jid: str, event: str, detail: str) -> None:
|
||||
"""The real producer, with the dispatch token the agent is given. The
|
||||
broker is unreachable here: publish_event exits 2 after the record and
|
||||
the events log were already written."""
|
||||
tok = _record(reg, jid)["dispatch_token"]
|
||||
res = subprocess.run([sys.executable, str(PUBLISH), "--registry-dir", str(reg), "--job", jid,
|
||||
"--event", event, "--detail", detail, "--expect-dispatch-token", tok],
|
||||
capture_output=True, text=True, env=env, timeout=120)
|
||||
assert res.returncode in (0, 2), res.stderr
|
||||
if event in ("completed", "error"):
|
||||
assert _record(reg, jid)["status"] in ("completed", "error"), res.stderr
|
||||
|
||||
|
||||
def _deadline_close(env: dict, reg: Path, jid: str) -> None:
|
||||
"""What the wrapper's finalize_unfinished does at a job's deadline."""
|
||||
tok = _record(reg, jid)["dispatch_token"]
|
||||
res = subprocess.run([sys.executable, str(REGISTRY), "--registry-dir", str(reg), "close", "--job", jid,
|
||||
"--request", "deadline_timeout", "--expect-dispatch-token", tok],
|
||||
capture_output=True, text=True, env=env, timeout=120)
|
||||
assert res.returncode == 0, res.stderr
|
||||
_publish(env, reg, jid, "error", "deadline_timeout")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# T1 - a job records its parent
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_t1_submit_records_the_parent_job(tmp_path):
|
||||
env = _env(tmp_path)
|
||||
reg = tmp_path / "reg"
|
||||
p_res = _submit(env, reg)
|
||||
p = _jid(p_res)
|
||||
assert _record(reg, p)["parent_job"] is None
|
||||
# the injected instructions tell the agent how to pass itself on
|
||||
assert f"add --parent-job {p} so the orchestrator waits for it." in p_res.stdout
|
||||
|
||||
c = _jid(_submit(env, reg, "--parent-job", p))
|
||||
assert _record(reg, c)["parent_job"] == p
|
||||
|
||||
bad = _submit(env, reg, "--parent-job", "nope1234")
|
||||
assert bad.returncode == 1 and "unknown parent job nope1234" in bad.stderr
|
||||
assert "registered job:" not in bad.stdout
|
||||
|
||||
# the remote runner's MAM_JOB_ID is the default ...
|
||||
e = _jid(_submit(_env(tmp_path, {"MAM_JOB_ID": p}), reg))
|
||||
assert _record(reg, e)["parent_job"] == p
|
||||
# ... ignored with a warning when this registry lacks it (another host's mirror)
|
||||
stray = _submit(_env(tmp_path, {"MAM_JOB_ID": "feedf00d"}), reg)
|
||||
assert _record(reg, _jid(stray))["parent_job"] is None
|
||||
assert "MAM_JOB_ID=feedf00d is not in" in stray.stderr
|
||||
# ... and the flag wins over it
|
||||
f = _jid(_submit(_env(tmp_path, {"MAM_JOB_ID": p}), reg, "--parent-job", c))
|
||||
assert _record(reg, f)["parent_job"] == c
|
||||
|
||||
listed = subprocess.run([sys.executable, str(REGISTRY), "--registry-dir", str(reg), "list", "--parent", p],
|
||||
capture_output=True, text=True, env=env, timeout=60)
|
||||
assert listed.returncode == 0
|
||||
assert {line.split()[0] for line in listed.stdout.splitlines()} == {c, e}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# T2 / T7 - LoopRunner waits for the tree (real records, real events)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _runner(tmp_path, monkeypatch, **env):
|
||||
monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
|
||||
monkeypatch.setenv("MAM_JOBS_DIR", str(tmp_path / "reg"))
|
||||
monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / ".mam-logs"))
|
||||
monkeypatch.delenv("MAM_REAL_ROOT", raising=False)
|
||||
for k in ("MAM_CHILD_WAIT_S", "MAM_JOB_TREE_MAX"):
|
||||
monkeypatch.delenv(k, raising=False)
|
||||
for k, v in env.items():
|
||||
monkeypatch.setenv(k, str(v))
|
||||
from lib_py.loop import runner as lr
|
||||
return lr.LoopRunner(["--creator", "c", "--task", "t"])
|
||||
|
||||
|
||||
def test_t2_the_runner_waits_for_children_and_grandchildren(tmp_path, monkeypatch):
|
||||
env = _env(tmp_path)
|
||||
reg = tmp_path / "reg"
|
||||
p = _jid(_submit(env, reg))
|
||||
c = _jid(_submit(env, reg, "--parent-job", p))
|
||||
g = _jid(_submit(env, reg, "--parent-job", c))
|
||||
|
||||
def finish():
|
||||
time.sleep(2)
|
||||
_publish(env, reg, g, "completed", "grandchild done")
|
||||
time.sleep(2)
|
||||
_publish(env, reg, c, "completed", "child done")
|
||||
|
||||
t = threading.Thread(target=finish)
|
||||
start = time.time()
|
||||
t.start()
|
||||
try:
|
||||
# the cap only keeps a broken helper thread from hanging the suite
|
||||
assert _runner(tmp_path, monkeypatch, MAM_CHILD_WAIT_S=60).wait_for_descendants(p, interval=0.5) is True
|
||||
elapsed = time.time() - start
|
||||
finally:
|
||||
t.join()
|
||||
assert elapsed >= 4
|
||||
assert _record(reg, c)["status"] == "completed" and _record(reg, g)["status"] == "completed"
|
||||
|
||||
|
||||
def test_t2b_the_wait_is_bounded_in_time_and_in_size(tmp_path, monkeypatch, capsys):
|
||||
env = _env(tmp_path)
|
||||
reg = tmp_path / "reg"
|
||||
p = _jid(_submit(env, reg))
|
||||
c = _jid(_submit(env, reg, "--parent-job", p))
|
||||
g = _jid(_submit(env, reg, "--parent-job", c))
|
||||
|
||||
start = time.time()
|
||||
assert _runner(tmp_path, monkeypatch, MAM_CHILD_WAIT_S=2).wait_for_descendants(p, interval=0.5) is False
|
||||
assert 2 <= time.time() - start < 10
|
||||
out = capsys.readouterr().out
|
||||
assert f"waiting for 2 sub-job(s) of {p}" in out and c in out and g in out
|
||||
assert "with 2 open" in out
|
||||
|
||||
start = time.time()
|
||||
assert _runner(tmp_path, monkeypatch, MAM_JOB_TREE_MAX=1).wait_for_descendants(p, interval=0.5) is False
|
||||
assert time.time() - start < 2
|
||||
assert "over MAM_JOB_TREE_MAX=1" in capsys.readouterr().err
|
||||
|
||||
|
||||
def test_t7_a_job_without_children_is_waited_for_as_before(tmp_path, monkeypatch):
|
||||
env = _env(tmp_path)
|
||||
reg = tmp_path / "reg"
|
||||
p = _jid(_submit(env, reg))
|
||||
_publish(env, reg, p, "completed", "done")
|
||||
r = _runner(tmp_path, monkeypatch)
|
||||
start = time.time()
|
||||
assert r.wait_for_job_tree(p) is True
|
||||
assert time.time() - start < 2 # one registry scan on top of wait_for_job
|
||||
assert r.wait_for_job(p) is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Loop e2e: a real loop process; the loop's own agents are simulated. The
|
||||
# simulated creator registers its sub-job with the REAL wrapper.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
FAKE_AGENT = r"""
|
||||
import json, os, re, secrets, subprocess, sys, time
|
||||
from pathlib import Path
|
||||
opts = dict(zip(sys.argv[2::2], sys.argv[3::2])) # argv[1] = "submit"
|
||||
prompt = Path(opts["--prompt-file"]).read_text(encoding="utf-8")
|
||||
jobs = Path(opts["--registry-dir"])
|
||||
jobs.mkdir(parents=True, exist_ok=True)
|
||||
role = opts["--role"]
|
||||
now = lambda: time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
||||
jid = "f" + secrets.token_hex(4)
|
||||
rec = {"job_id": jid, "status": "running", "created_at": now(), "updated_at": now(), "created_ts": time.time(),
|
||||
"agent_session": opts["--agent-session"], "role": role}
|
||||
path = jobs / (jid + ".json")
|
||||
path.write_text(json.dumps(rec))
|
||||
late = False
|
||||
if role == "Worker":
|
||||
act = os.environ.get("FAKE_WORKER", "none")
|
||||
marker = Path(".mam/child-id")
|
||||
if act in ("child", "stray") and not marker.exists(): # the implementation job only
|
||||
cmd = ["bash", os.environ["FAKE_REAL_WRAPPER"], "submit", "--agent", "human", "--agent-session",
|
||||
"herdr:img", "--dry-run", "--registry-dir", str(jobs), "--prompt", "draw the figures"]
|
||||
if act == "child":
|
||||
cmd += ["--parent-job", jid]
|
||||
out = subprocess.run(cmd, capture_output=True, text=True)
|
||||
if out.returncode != 0:
|
||||
print(out.stdout + out.stderr, file=sys.stderr)
|
||||
sys.exit(2)
|
||||
marker.write_text(re.search(r"registered job:\s*(\S+)", out.stdout).group(1))
|
||||
report = "done\n"
|
||||
else:
|
||||
counter = Path(".mam/review-count")
|
||||
n = int(counter.read_text()) + 1 if counter.exists() else 1
|
||||
counter.write_text(str(n))
|
||||
Path(".mam/review-prompt-%d.md" % n).write_text(prompt, encoding="utf-8")
|
||||
rule = os.environ.get("FAKE_REVIEW", "pass")
|
||||
unfinished = re.search(r"(열린|미확인 종결|실패) [1-9]", prompt)
|
||||
verdict = "NOT PASS" if rule == "rule" and unfinished else "PASS"
|
||||
if os.environ.get("FAKE_FIRST_NOT_PASS") and n == 1: # forces a correction round
|
||||
verdict = "NOT PASS"
|
||||
report = "reviewed\n\n[VERDICT: %s]\n" % verdict
|
||||
late = os.environ.get("FAKE_LATE_WRITE") == str(n)
|
||||
dest = jobs / jid / "agent-reports"
|
||||
dest.mkdir(parents=True)
|
||||
(dest / "report-final.md").write_text(report, encoding="utf-8")
|
||||
rec.update(status="completed", updated_at=now())
|
||||
path.write_text(json.dumps(rec))
|
||||
if late: # stands in for a sub-job agent that keeps writing after its deadline close
|
||||
Path("late.txt").write_text("written during review\n")
|
||||
print("registered job: " + jid)
|
||||
"""
|
||||
|
||||
|
||||
def _loop_ctl(tmp_path: Path, git: bool = True) -> Path:
|
||||
ctl = tmp_path / "ctl"
|
||||
if git:
|
||||
_repo(ctl)
|
||||
else:
|
||||
ctl.mkdir(parents=True)
|
||||
agent = ctl / "fake-agent.py"
|
||||
agent.write_text(FAKE_AGENT, encoding="utf-8")
|
||||
w = ctl / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
||||
w.parent.mkdir(parents=True, exist_ok=True)
|
||||
w.write_text(f'#!/usr/bin/env bash\nexec "{sys.executable}" "{agent}" "$@"\n', encoding="utf-8")
|
||||
w.chmod(0o755)
|
||||
(ctl / ".mam").mkdir(exist_ok=True)
|
||||
if git:
|
||||
(ctl / "plan.md").write_text("# plan\n\ndo it\n", encoding="utf-8")
|
||||
_git(ctl, "add", "-A")
|
||||
_git(ctl, "commit", "-q", "-m", "harness")
|
||||
else:
|
||||
(ctl / "plan.md").write_text("# plan\n\ndo it\n", encoding="utf-8")
|
||||
return ctl
|
||||
|
||||
|
||||
def _loop_env(ctl: Path, **extra) -> dict:
|
||||
return _env(ctl, extra={
|
||||
"WORKSPACE_ROOT": ctl,
|
||||
"MAM_JOBS_DIR": ctl / ".mam" / "jobs",
|
||||
"DELEGATE_JOB_LOGS_DIR": ctl / ".mam" / "logs",
|
||||
"MAM_SESSION_BUSY_RETRY_S": "0",
|
||||
"MAM_CHILD_WAIT_S": "60", # a broken helper thread fails the test instead of hanging it
|
||||
"FAKE_REAL_WRAPPER": WRAPPER,
|
||||
"MAM_STATE_JSON": json.dumps({"herdr_sessions": [
|
||||
{"name": "c", "agent": "claude", "role": "creator", "status": "running"},
|
||||
{"name": "r", "agent": "claude", "role": "reviewer", "status": "running"}]}),
|
||||
**extra,
|
||||
}, drop=("MAM_LOOP_WORKTREE", "MAM_LOOP_BASE", "MAM_VERSION"))
|
||||
|
||||
|
||||
def _run_loop(ctl: Path, env: dict, *argv: str, on_child=None) -> subprocess.CompletedProcess:
|
||||
"""Run the loop; on_child(child_id) runs in a thread once the simulated
|
||||
creator registered its sub-job."""
|
||||
marker = ctl / ".mam" / "child-id"
|
||||
stop = threading.Event()
|
||||
seen = {}
|
||||
|
||||
def watch():
|
||||
while not stop.is_set():
|
||||
if marker.exists() and marker.read_text().strip():
|
||||
seen["id"] = marker.read_text().strip()
|
||||
on_child(seen["id"])
|
||||
return
|
||||
time.sleep(0.2)
|
||||
|
||||
t = threading.Thread(target=watch) if on_child else None
|
||||
if t:
|
||||
t.start()
|
||||
try:
|
||||
return subprocess.run([sys.executable, "-m", "lib_py.loop", "--creator", "c", "--task", "t34",
|
||||
"--reviewer", "r", "--plan-file", str(ctl / "plan.md"), *argv],
|
||||
capture_output=True, text=True, env=env, cwd=str(ctl), timeout=300)
|
||||
finally:
|
||||
stop.set()
|
||||
if t:
|
||||
t.join()
|
||||
|
||||
|
||||
def _review_prompt(ctl: Path, n: int = 1) -> str:
|
||||
return (ctl / ".mam" / f"review-prompt-{n}.md").read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def _review_records(ctl: Path) -> list:
|
||||
recs = [json.loads(p.read_text()) for p in (ctl / ".mam" / "jobs").glob("f*.json")]
|
||||
return sorted((r for r in recs if r.get("role") == "Reviewer"), key=lambda r: r["created_ts"])
|
||||
|
||||
|
||||
def test_t3_the_loop_waits_for_the_sub_job_before_review(tmp_path):
|
||||
ctl = _loop_ctl(tmp_path)
|
||||
env = _loop_env(ctl, FAKE_WORKER="child")
|
||||
jobs = ctl / ".mam" / "jobs"
|
||||
done = {}
|
||||
|
||||
def finish_later(cid):
|
||||
time.sleep(5)
|
||||
done["ts"] = time.time()
|
||||
_publish(env, jobs, cid, "completed", "figures drawn")
|
||||
|
||||
res = _run_loop(ctl, env, "--max-loop", "1", on_child=finish_later)
|
||||
out = res.stdout + res.stderr
|
||||
assert res.returncode == 0, out
|
||||
cid = (ctl / ".mam" / "child-id").read_text().strip()
|
||||
assert _record(jobs, cid)["parent_job"] # registered by the real wrapper, under the creator job
|
||||
assert "waiting for 1 sub-job(s)" in res.stdout
|
||||
# the loop registered the review only after the sub-job was completed
|
||||
assert _review_records(ctl)[0]["created_ts"] > done["ts"]
|
||||
prompt = _review_prompt(ctl)
|
||||
assert "[하위 위임 잡]" in prompt
|
||||
assert re.search(rf"\| {cid} \| herdr:img \| Worker \| completed \| figures drawn \| 1 \|", prompt), prompt
|
||||
assert "열린 0, 미확인 종결 0, 실패 0, 완료 1." in prompt
|
||||
assert "100% PASS" in res.stdout
|
||||
|
||||
|
||||
def test_t4_an_open_sub_job_is_evidence_and_the_review_blocks(tmp_path):
|
||||
ctl = _loop_ctl(tmp_path)
|
||||
env = _loop_env(ctl, FAKE_WORKER="child", FAKE_REVIEW="rule", MAM_CHILD_WAIT_S="3")
|
||||
res = _run_loop(ctl, env, "--max-loop", "1")
|
||||
out = res.stdout + res.stderr
|
||||
assert res.returncode == 1, out
|
||||
assert "with 1 open; reviewers see them as OPEN" in res.stdout
|
||||
cid = (ctl / ".mam" / "child-id").read_text().strip()
|
||||
prompt = _review_prompt(ctl)
|
||||
assert re.search(rf"\| {cid} \| .* \| OPEN \|", prompt) and "열린 1," in prompt
|
||||
assert "100% PASS" not in res.stdout
|
||||
# the success pair is T3: the same harness and rule pass once the child completed
|
||||
|
||||
|
||||
def test_t5_a_failed_sub_job_reaches_the_review_and_blocks_it(tmp_path):
|
||||
ctl = _loop_ctl(tmp_path)
|
||||
env = _loop_env(ctl, FAKE_WORKER="child", FAKE_REVIEW="rule")
|
||||
jobs = ctl / ".mam" / "jobs"
|
||||
res = _run_loop(ctl, env, "--max-loop", "1",
|
||||
on_child=lambda cid: _publish(env, jobs, cid, "error", "simulated"))
|
||||
out = res.stdout + res.stderr
|
||||
assert res.returncode == 1, out
|
||||
assert "Requesting code review" in res.stdout # a failed child does not stop the loop
|
||||
cid = (ctl / ".mam" / "child-id").read_text().strip()
|
||||
prompt = _review_prompt(ctl)
|
||||
assert re.search(rf"\| {cid} \| .* \| error \| simulated \|", prompt), prompt
|
||||
assert "실패 1," in prompt
|
||||
assert "100% PASS" not in res.stdout
|
||||
|
||||
|
||||
@pytest.mark.parametrize("late_write", [True, False])
|
||||
def test_t5b_a_deadline_closed_sub_job_is_unconfirmed_and_a_late_write_voids_the_round(tmp_path, late_write):
|
||||
ctl = _loop_ctl(tmp_path)
|
||||
extra = {"FAKE_WORKER": "child", "FAKE_REVIEW": "pass"}
|
||||
if late_write:
|
||||
extra["FAKE_LATE_WRITE"] = "1"
|
||||
env = _loop_env(ctl, **extra)
|
||||
jobs = ctl / ".mam" / "jobs"
|
||||
res = _run_loop(ctl, env, "--max-loop", "1", on_child=lambda cid: _deadline_close(env, jobs, cid))
|
||||
out = res.stdout + res.stderr
|
||||
cid = (ctl / ".mam" / "child-id").read_text().strip()
|
||||
assert _record(jobs, cid)["status"] == "error"
|
||||
prompt = _review_prompt(ctl)
|
||||
assert re.search(rf"\| {cid} \| .* \| CLOSED-UNCONFIRMED \| deadline_timeout \|", prompt), prompt
|
||||
assert "its agent may still be writing" in res.stdout
|
||||
if late_write:
|
||||
assert res.returncode == 1, out
|
||||
assert "source changed during review" in res.stdout
|
||||
assert "100% PASS" not in res.stdout
|
||||
else: # the guard reacts to a source change only
|
||||
assert res.returncode == 0, out
|
||||
assert "source changed" not in out
|
||||
assert "100% PASS" in res.stdout
|
||||
|
||||
|
||||
def test_t6_an_unattributed_open_job_is_shown_and_waited_for_only_on_request(tmp_path):
|
||||
ctl = _loop_ctl(tmp_path)
|
||||
env = _loop_env(ctl, FAKE_WORKER="stray")
|
||||
res = _run_loop(ctl, env, "--max-loop", "1")
|
||||
assert res.returncode == 0, res.stdout + res.stderr
|
||||
uid = (ctl / ".mam" / "child-id").read_text().strip()
|
||||
assert "sub-job(s) of" not in res.stdout # nothing to wait for by default
|
||||
prompt = _review_prompt(ctl)
|
||||
assert "[하위 위임 잡] 없음" in prompt
|
||||
assert re.search(rf"\[귀속되지 않은 열린 잡\] .*: {uid}\(herdr:img, pending\)", prompt), prompt
|
||||
|
||||
ctl2 = _loop_ctl(tmp_path / "strict")
|
||||
res2 = _run_loop(ctl2, _loop_env(ctl2, FAKE_WORKER="stray", MAM_CHILD_WAIT_S="3"),
|
||||
"--max-loop", "1", "--wait-open-jobs")
|
||||
uid2 = (ctl2 / ".mam" / "child-id").read_text().strip()
|
||||
assert f"{uid2}(herdr:img, pending)" in res2.stdout and "waiting for 1 sub-job(s)" in res2.stdout
|
||||
assert "ended after 3s" in res2.stdout
|
||||
|
||||
|
||||
def test_t8_the_stale_review_guard_without_worktree_or_base(tmp_path):
|
||||
"""Armed only when the run has sub-jobs. Without them the loop behaves as
|
||||
before - its source may hold untracked files that change on every job
|
||||
(test_tier4_e2e scenario 5: the mock herdr's state file)."""
|
||||
ctl = _loop_ctl(tmp_path)
|
||||
res = _run_loop(ctl, _loop_env(ctl, FAKE_LATE_WRITE="1"), "--max-loop", "1")
|
||||
assert res.returncode == 0, res.stdout + res.stderr
|
||||
assert "source changed" not in res.stdout and "100% PASS" in res.stdout
|
||||
|
||||
ctl2 = _loop_ctl(tmp_path / "with-child")
|
||||
env2 = _loop_env(ctl2, FAKE_WORKER="child", FAKE_LATE_WRITE="1")
|
||||
res = _run_loop(ctl2, env2, "--max-loop", "2",
|
||||
on_child=lambda cid: _publish(env2, ctl2 / ".mam" / "jobs", cid, "completed", "done"))
|
||||
out = res.stdout + res.stderr
|
||||
assert res.returncode == 0, out
|
||||
# once: round 2 is not stale although .mam (jobs, logs, checkpoint) inside
|
||||
# the source kept changing - the control plane is not part of the fingerprint
|
||||
assert res.stdout.count("approvals discarded, re-reviewing") == 1
|
||||
assert len(_review_records(ctl2)) == 2 and "100% PASS" in res.stdout
|
||||
|
||||
# no repository: no fingerprint, the guard is off and says so; behaviour as before
|
||||
plain = _loop_ctl(tmp_path / "plain", git=False)
|
||||
env3 = _loop_env(plain, FAKE_WORKER="child", FAKE_LATE_WRITE="1")
|
||||
res = _run_loop(plain, env3, "--max-loop", "1",
|
||||
on_child=lambda cid: _publish(env3, plain / ".mam" / "jobs", cid, "completed", "done"))
|
||||
assert res.returncode == 0, res.stdout + res.stderr
|
||||
assert "the stale-review guard is off for this run" in res.stdout
|
||||
assert "100% PASS" in res.stdout
|
||||
|
||||
|
||||
@pytest.mark.parametrize("child_end, rc", [("open", 1), ("error", 1), ("completed", 0)])
|
||||
def test_t9_a_correction_round_keeps_the_earlier_creator_jobs_sub_jobs(tmp_path, child_end, rc):
|
||||
"""Review of round 2 (after a correction job with no sub-job of its own)
|
||||
still sees the implementation job's sub-job; the correction does not claim
|
||||
to have replaced it. Success pair: the completed child passes in round 2."""
|
||||
ctl = _loop_ctl(tmp_path)
|
||||
env = _loop_env(ctl, FAKE_WORKER="child", FAKE_REVIEW="rule", FAKE_FIRST_NOT_PASS="1",
|
||||
MAM_CHILD_WAIT_S="0")
|
||||
jobs = ctl / ".mam" / "jobs"
|
||||
on_child = {"open": None,
|
||||
"error": lambda cid: _publish(env, jobs, cid, "error", "simulated"),
|
||||
"completed": lambda cid: _publish(env, jobs, cid, "completed", "figures drawn")}[child_end]
|
||||
if child_end != "open":
|
||||
env["MAM_CHILD_WAIT_S"] = "60" # wait for the helper to close it
|
||||
res = _run_loop(ctl, env, "--max-loop", "2", on_child=on_child)
|
||||
out = res.stdout + res.stderr
|
||||
assert res.returncode == rc, out
|
||||
assert "Applying straight bugfixes" in res.stdout # a correction round ran
|
||||
cid = (ctl / ".mam" / "child-id").read_text().strip()
|
||||
second = _review_prompt(ctl, 2)
|
||||
state = {"open": "OPEN", "error": "error", "completed": "completed"}[child_end]
|
||||
assert re.search(rf"\| {cid} \| .* \| {state} \|", second), second
|
||||
assert ("100% PASS" in res.stdout) == (rc == 0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# T10-T12 - a pane caller reaches an agent of another workspace on the same
|
||||
# server (mock herdr; the pane env is the one measured in creator-claude-issue34)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
SAME_SERVER_PANE = { # herdr 0.9.1, creator in w6 of server multi-agent-mux
|
||||
"HERDR_SESSION": "multi-agent-mux",
|
||||
"HERDR_SOCKET_PATH": "/home/someone/.config/herdr/sessions/multi-agent-mux/herdr.sock",
|
||||
"HERDR_SESSION_NAME": "multi-agent-mux",
|
||||
"HERDR_WORKSPACE_ID": "w6", "HERDR_PANE_ID": "w6:p1", "HERDR_TAB_ID": "w6:t1",
|
||||
}
|
||||
|
||||
|
||||
def _run_lib(sandbox: Path, body: str, env: dict) -> subprocess.CompletedProcess:
|
||||
"""Source the sandbox lib.sh with exactly these HERDR_* (as test_issue23_24)."""
|
||||
script = 'source "%s"\n_init_herdr_isolation\n%s\n' % (sandbox / ".agents" / "skills" / "lib.sh", body)
|
||||
run_env = {k: v for k, v in os.environ.items() if not k.startswith("HERDR_")}
|
||||
run_env.update(env)
|
||||
return subprocess.run(["bash", "-c", script], capture_output=True, text=True, cwd=str(sandbox), env=run_env)
|
||||
|
||||
|
||||
def _seed_herdr_agent(state_file: Path, sandbox: Path, name: str, ws: str, pane: str) -> None:
|
||||
state = json.loads(state_file.read_text())
|
||||
if not any(w.get("workspace_id") == ws for w in state.setdefault("workspaces", [])):
|
||||
state["workspaces"].append({"workspace_id": ws, "label": ws, "cwd": str(sandbox)})
|
||||
state.setdefault("agents", {})[name] = {
|
||||
"name": name, "agent": "opencode", "status": "running", "cwd": str(sandbox), "workspace_id": ws,
|
||||
"pane_id": pane, "pid": 9999, "command": "opencode", "buffer": "Ready"}
|
||||
state.setdefault("panes", []).append({"pane_id": pane, "workspace_id": ws, "label": name, "buffer": "Ready"})
|
||||
state_file.write_text(json.dumps(state, indent=2))
|
||||
|
||||
|
||||
def test_t10_a_pane_caller_reaches_an_agent_of_another_workspace_on_its_server(
|
||||
mam_sandbox, mock_herdr, mock_agents):
|
||||
_seed_herdr_agent(mock_herdr, mam_sandbox, "reviewer-opencode-01", "w2", "w2:p1")
|
||||
res = _run_lib(mam_sandbox, "herdr has-session -t reviewer-opencode-01", SAME_SERVER_PANE)
|
||||
assert res.returncode == 0, res.stdout + res.stderr
|
||||
before = len(json.loads(mock_herdr.read_text()).get("calls", []))
|
||||
res = _run_lib(mam_sandbox, 'herdr agent prompt reviewer-opencode-01 "hello from w6"', SAME_SERVER_PANE)
|
||||
assert res.returncode == 0, res.stdout + res.stderr
|
||||
calls = json.loads(mock_herdr.read_text())["calls"][before:]
|
||||
assert any(c[-4:] == ["agent", "prompt", "reviewer-opencode-01", "hello from w6"] for c in calls), calls
|
||||
|
||||
|
||||
def test_t11_a_workspace_id_the_caller_chose_still_scopes(mam_sandbox, mock_herdr, mock_agents):
|
||||
"""The ISSUE-3 / F-2b contract: an id that is not the pane's own (the pane
|
||||
id says w6, the caller asks for w5), or one set outside any pane, hides an
|
||||
agent of another workspace."""
|
||||
_seed_herdr_agent(mock_herdr, mam_sandbox, "reviewer-opencode-01", "w2", "w2:p1")
|
||||
chosen = dict(SAME_SERVER_PANE, HERDR_WORKSPACE_ID="w5")
|
||||
assert _run_lib(mam_sandbox, "herdr has-session -t reviewer-opencode-01", chosen).returncode == 1
|
||||
outside = {"HERDR_SESSION_NAME": "multi-agent-mux", "HERDR_WORKSPACE_ID": "w6"}
|
||||
assert _run_lib(mam_sandbox, "herdr has-session -t reviewer-opencode-01", outside).returncode == 1
|
||||
# the pane's own id is still what new panes are created in: only name lookups drop it
|
||||
res = _run_lib(mam_sandbox, '_herdr_scope_is_pane_inherited pane list && echo strip || echo keep',
|
||||
SAME_SERVER_PANE)
|
||||
assert res.stdout.strip() == "keep", res.stdout + res.stderr
|
||||
|
||||
|
||||
def test_t12_a_target_the_preflight_cannot_see_is_not_recorded_as_busy(mam_sandbox, mock_herdr, mock_agents):
|
||||
reg = mam_sandbox / ".mam" / "jobs"
|
||||
env = {k: v for k, v in os.environ.items() if not k.startswith("HERDR_")}
|
||||
env.update(UNREACHABLE, DELEGATE_JOB_PYTHON=sys.executable, MAM_JOB_ID="",
|
||||
DELEGATE_JOB_LOGS_DIR=str(mam_sandbox / ".mam" / "logs"))
|
||||
res = subprocess.run(["bash", str(mam_sandbox / ".agents" / "skills" / "multi-agent-mux-delegate-job"
|
||||
/ "multi-agent-mux-delegate-job"),
|
||||
"submit", "--agent", "opencode", "--agent-session", "herdr:no-such-agent",
|
||||
"--registry-dir", str(reg), "--prompt", "x"],
|
||||
capture_output=True, text=True, env=env, cwd=str(mam_sandbox), timeout=120)
|
||||
assert res.returncode == 5, res.stdout + res.stderr
|
||||
jid = re.search(r"registered job:\s*(\S+)", res.stdout).group(1)
|
||||
assert _record(reg, jid)["status"] == "cancelled"
|
||||
events = (reg / f"{jid}.events.log").read_text()
|
||||
assert '"detail": "target_not_found"' in events and "session_busy" not in events, events
|
||||
Reference in new issue
Block a user