From ec9586bb9b7f10a6eacef360c4066fbffaeef842 Mon Sep 17 00:00:00 2001 From: Godopu Date: Fri, 25 Sep 2026 11:43:10 +0000 Subject: [PATCH] feat(issue-33): the job budget travels and the two clocks carry different labels - wrapper: TIMEOUT falls back to MAM_JOB_TIMEOUT_S (--timeout > env > 3600); an invalid value warns and keeps 3600. - loop: --job-timeout (run_loop.sh passes it on; refused in the legacy shell loop). The runner always hands the wrapper --timeout, so the budget the wrapper registers is the one its wait caps come from. - runner: wait_for_job and the MAM_CHILD_WAIT_S default are budget + 300 s (MAM_WAIT_GRACE_S), today's 3900 with no option; an explicit MAM_CHILD_WAIT_S still wins. - subscriber: the wall-clock budget returns 4 (RC_WALL_CLOCK), silence keeps 2 (RC_NO_EVENT). The wrapper maps both to caller rc 2 and labels the close from the rc: wall_clock_timeout vs deadline_timeout; the disk-heartbeat grace that ends on the deadline is the wall clock too. - registry close accepts wall_clock_timeout (engine: -> error); the remote runner's deadline and command-timeout details lead with wall_clock_timeout; the sub-job table files it CLOSED-UNCONFIRMED (first token, colon stripped). Test expectations changed only where the contract changed: the loop's contract_args now starts with --timeout, the subscriber's exit-code docstring names rc 2/4, and the registry --help capture lists the new close kind. Co-Authored-By: Claude Opus 5.5 --- .agents/skills/lib_py/delegation/engine.py | 3 +- .../lib_py/delegation/job_subscriber.py | 14 +- .agents/skills/lib_py/delegation/registry.py | 4 +- .agents/skills/lib_py/delegation/remote.py | 4 +- .agents/skills/lib_py/loop/runner.py | 46 ++- .../multi-agent-mux-delegate-job | 22 +- .../multi-agent-mux-loop/scripts/run_loop.sh | 7 +- .../fixtures/issue_5/phase1/help-registry.txt | 2 +- tests/test_issue18_liveness_contract.py | 4 +- tests/test_issue33_budget.py | 324 ++++++++++++++++++ tests/test_issue5_phase1_delegation.py | 2 +- 11 files changed, 409 insertions(+), 23 deletions(-) create mode 100644 tests/test_issue33_budget.py diff --git a/.agents/skills/lib_py/delegation/engine.py b/.agents/skills/lib_py/delegation/engine.py index 62c3cd9..46e541f 100644 --- a/.agents/skills/lib_py/delegation/engine.py +++ b/.agents/skills/lib_py/delegation/engine.py @@ -47,7 +47,8 @@ REQUEST_TO_STATUS = { "completed": "completed", "error": "error", "cancelled": "cancelled", - "deadline_timeout": "error", + "deadline_timeout": "error", # no event for K x N (subscriber rc 2) + "wall_clock_timeout": "error", # the job budget ran out (subscriber rc 4, Issue #33) "merge": None, # Operator-authority requests. The wrapper drives multi-iteration delegation # (`submit --type loop|discuss`) by re-dispatching a finished job to the next diff --git a/.agents/skills/lib_py/delegation/job_subscriber.py b/.agents/skills/lib_py/delegation/job_subscriber.py index fa5eea9..f4244de 100755 --- a/.agents/skills/lib_py/delegation/job_subscriber.py +++ b/.agents/skills/lib_py/delegation/job_subscriber.py @@ -19,7 +19,9 @@ stdout = event lines only. Diagnostics go to stderr via logging. Exit codes: 0 all watched jobs reached ``completed`` 1 any watched job reached ``error`` - 2 timed out (wall-clock or idle) before all jobs finished + 2 no event for the idle window (K x N) - the wrapper labels it deadline_timeout + 3 broker unreachable + 4 the wall-clock budget (--timeout) ran out - the wrapper labels it wall_clock_timeout """ from __future__ import annotations @@ -47,6 +49,12 @@ logger = logging.getLogger("delegate_job.job_subscriber") TERMINAL_EVENTS = ("completed", "error") TERMINAL_STATUSES = ("completed", "error", "cancelled") STATE_EVENT = "state" +# Issue #33: which clock ended the wait. The wrapper maps both to a timeout +# (caller rc 2) and labels the close from them: rc 2 -> deadline_timeout +# (no event for the idle window, K x N), rc 4 -> wall_clock_timeout (the job +# budget, --timeout, ran out while events may still have been arriving). +RC_NO_EVENT = 2 +RC_WALL_CLOCK = 4 def _format_line(topic: str, payload: Dict[str, Any]) -> str: @@ -473,7 +481,7 @@ def _run_subscriber(args) -> int: break logger.error("wall-clock timeout (%.0fs); still pending: %s", wall_timeout, ", ".join(sorted(pending))) - return 2 + return RC_WALL_CLOCK if not subscribed: last_event = now idle_left = (idle_timeout if first_event_seen else idle_timeout + ack_timeout) - (now - last_event) @@ -483,7 +491,7 @@ def _run_subscriber(args) -> int: break logger.error("idle timeout (%.0fs, no events); still pending: %s", idle_timeout, ", ".join(sorted(pending))) - return 2 + return RC_NO_EVENT wait = min(wall_deadline - now, idle_left, 1.0) try: topic, payload = watcher.events.get(timeout=wait) diff --git a/.agents/skills/lib_py/delegation/registry.py b/.agents/skills/lib_py/delegation/registry.py index eb4e842..0beff05 100755 --- a/.agents/skills/lib_py/delegation/registry.py +++ b/.agents/skills/lib_py/delegation/registry.py @@ -419,9 +419,9 @@ def _build_parser() -> argparse.ArgumentParser: p_update.add_argument("--attempt-id", type=int, default=None) p_update.add_argument("--role", default=None) - p_close = sub.add_parser("close", help="close a job with deadline_timeout or cancelled") + p_close = sub.add_parser("close", help="close a job with deadline_timeout, wall_clock_timeout or cancelled") p_close.add_argument("--job", required=True) - p_close.add_argument("--request", required=True, choices=["deadline_timeout", "cancelled"]) + p_close.add_argument("--request", required=True, choices=["deadline_timeout", "wall_clock_timeout", "cancelled"]) p_close.add_argument("--detail", default=None) p_close.add_argument("--expect-dispatch-token", default=None) diff --git a/.agents/skills/lib_py/delegation/remote.py b/.agents/skills/lib_py/delegation/remote.py index cdbd0eb..0866689 100644 --- a/.agents/skills/lib_py/delegation/remote.py +++ b/.agents/skills/lib_py/delegation/remote.py @@ -331,7 +331,7 @@ def handle_dispatch(d: Dict[str, Any], registry_dir: str, session: str, timeout = float(job.get("timeout_sec", 3600)) rc, out, err, timed_out = _run_agent_command(exec_cmd, env, timeout) if timed_out: - detail = "agent command exceeded timeout_sec {}".format(int(timeout)) + detail = "wall_clock_timeout: agent command exceeded timeout_sec {}".format(int(timeout)) else: tail = (out or "").strip().splitlines() detail = tail[-1][:200] if tail else "exit {}".format(rc) @@ -349,7 +349,7 @@ def handle_dispatch(d: Dict[str, Any], registry_dir: str, session: str, if cur.get("status") in ("completed", "error", "cancelled"): return 0 if cur.get("status") == "completed" else 1 time.sleep(1.0) - _publish(jid, registry_dir, "error", "deadline_timeout (remote runner)", d) + _publish(jid, registry_dir, "error", "wall_clock_timeout (remote runner)", d) return 2 diff --git a/.agents/skills/lib_py/loop/runner.py b/.agents/skills/lib_py/loop/runner.py index 97cd4e6..8d1081a 100644 --- a/.agents/skills/lib_py/loop/runner.py +++ b/.agents/skills/lib_py/loop/runner.py @@ -33,7 +33,7 @@ UNAVAILABLE_MARKER = "!!! CHANGE SET UNAVAILABLE !!!" # diff_collect.sh MAM_DIF # 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") +ORCHESTRATOR_CLOSED = ("deadline_timeout", "wall_clock_timeout", "agent_blocked") SUBJOB_RULE = ("규칙: 열린(OPEN)·미확인 종결(CLOSED-UNCONFIRMED)·실패 하위 잡이 하나라도 있으면 산출물을 미완으로 보고 " "[VERDICT: NOT PASS] 를 내십시오. creator 리포트가 해당 부분의 대체 완료를 입증하면 예외입니다. " "CLOSED-UNCONFIRMED 는 오케스트레이터가 마감으로 닫은 잡이며 그 에이전트는 계속 일하고 있을 수 있습니다. " @@ -121,6 +121,7 @@ class LoopRunner: self.reviewer_list: Optional[str] = None self.all_reviewers: bool = False self.max_loop: int = 3 + self.job_timeout: Optional[int] = None self.max_rebut: int = 1 self.verbose: bool = False self.cleanup: bool = False @@ -139,6 +140,7 @@ class LoopRunner: " --all-reviewer Enforce PASS verdict from all active reviewer sessions", " --max-loop N Max execution-review corrective loop runs (default: 3)", " --max-rebut N Max rebuttal attempts per review iteration (default: 1, 0: disabled)", + " --job-timeout Budget of each delegated job (default: $MAM_JOB_TIMEOUT_S or 3600)", " --verbose Print detailed execution timeline traces", " --cleanup Purge temporary job directories upon success", " --worktree Source worktree for diff/base/head (control plane stays here)", @@ -209,6 +211,12 @@ class LoopRunner: sys.exit(1) self.max_loop = int(args[i + 1]) i += 2 + elif arg == "--job-timeout": + if i + 1 >= len(args) or not re.match(r"^[0-9]+$", args[i + 1]) or int(args[i + 1]) <= 0: + print("ERROR: --job-timeout requires a positive non-zero integer.", file=sys.stderr) + sys.exit(1) + self.job_timeout = int(args[i + 1]) + i += 2 elif arg == "--max-rebut": if i + 1 >= len(args) or not re.match(r"^[0-9]+$", args[i + 1]): print("ERROR: --max-rebut requires a non-negative integer.", file=sys.stderr) @@ -454,8 +462,10 @@ class LoopRunner: def contract_args(self) -> List[str]: """Liveness-contract knobs passed through to every delegate job (Issue #18): MAM_HEARTBEAT_SEC (N), MAM_HEARTBEAT_MISSES (K), - MAM_ACK_TIMEOUT_S (T). The wrapper derives --idle-timeout as K x N.""" - args: List[str] = [] + MAM_ACK_TIMEOUT_S (T). The wrapper derives --idle-timeout as K x N. + Issue #33: the job budget is always passed explicitly, so the budget + the wrapper registers is the one job_wait_s was derived from.""" + args: List[str] = ["--timeout", str(self.job_budget)] for env_name, flag in (("MAM_HEARTBEAT_SEC", "--heartbeat"), ("MAM_HEARTBEAT_MISSES", "--heartbeat-misses"), ("MAM_ACK_TIMEOUT_S", "--ack-timeout")): @@ -572,8 +582,27 @@ class LoopRunner: return job_id return None - def wait_for_job(self, job_id: str, max_wait: int = 3900, check_interval: float = 1.0) -> bool: + @property + def job_budget(self) -> int: + """Issue #33: --job-timeout > $MAM_JOB_TIMEOUT_S (positive integer) > 3600.""" + if self.job_timeout: + return self.job_timeout + env = os.environ.get("MAM_JOB_TIMEOUT_S", "") + return int(env) if env.isdigit() and int(env) > 0 else 3600 + + @property + def job_wait_s(self) -> int: + """Wait cap for a job and its sub-jobs: budget + MAM_WAIT_GRACE_S (300, + today's 3900 - 3600). The grace covers the wrapper's close/publish tail + after the wall clock and the 1 s record poll; it is not proportional + to the budget because nothing observed would justify that.""" + grace = os.environ.get("MAM_WAIT_GRACE_S", "") + return self.job_budget + (int(grace) if grace.isdigit() else 300) + + def wait_for_job(self, job_id: str, max_wait: Optional[float] = None, check_interval: float = 1.0) -> bool: """Blocking wait for job completion or error with safety timeout.""" + if max_wait is None: + max_wait = self.job_wait_s if job_id in self.failed_submit_jobs: # Submit command for this job failed (rc!=0). Cap wait to 5s to avoid 3900s hang max_wait = min(max_wait, 5.0) @@ -644,7 +673,10 @@ class LoopRunner: 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")) + child_env = os.environ.get("MAM_CHILD_WAIT_S") + max_wait = float(child_env) if child_env else float(self.job_wait_s) + wait_src = "MAM_CHILD_WAIT_S" if child_env else \ + f"MAM_CHILD_WAIT_S default: job budget {self.job_budget} + {self.job_wait_s - self.job_budget}" tree_max = int(os.environ.get("MAM_JOB_TREE_MAX", "50")) since = self._created_at(root) start = time.time() @@ -667,7 +699,7 @@ class LoopRunner: + " ".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 " + log_warn(f"sub-job wait for {root} ended after {int(max_wait)}s ({wait_src}) with " f"{len(open_jobs)} open; reviewers see them as OPEN") return False time.sleep(interval) @@ -693,7 +725,7 @@ class LoopRunner: return "completed", reason or "-" # merge review R2 (codex): the remote runner phrases its deadline as # "deadline_timeout (remote runner)", so the kind is the first word - closed_kind = reason.split(" ", 1)[0] if reason else "" + closed_kind = reason.split(" ", 1)[0].rstrip(":") if reason else "" if closed_kind in ORCHESTRATOR_CLOSED or (status == "error" and not reason): # no recorded reason is not the agent's own report either return "CLOSED-UNCONFIRMED", reason or "-" diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index e3a3a5f..66c08aa 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job @@ -111,7 +111,7 @@ usage() { multi-agent-mux-delegate-job [options] submit --agent (--prompt | --prompt-file ) [--workdir ] [--agent-session