"""Issue #17 item 4: the delegate-job wrapper's decisions. The bash wrapper (multi-agent-mux-delegate-job) spawns, waits, kills and injects; everything it used to decide in bash is here, and it branches only on what a subcommand prints or returns: sub-status --job J --registry-dir R subscriber rc -> job status (was _map_sub_rc) subscriber-up [--print] exit 0: a subscriber that ended before SUBSCRIBED still lets the job proceed exit-code the wrapper's exit code (was _rc_for, with MAM_DELEGATE_FAIL_CLOSED=0 -> 0) wait-rc `wait`'s rc (either clock is a timeout, 2) stall-watch [--print] exit 0: start the stall watcher (Issue #30) The two yes/no decisions answer by exit code (0 yes, 1 no); with --print they print `yes`/`no` and exit 0, so a failure to decide (any other exit code, or no answer) cannot pass for "no". The wrapper uses --print. finalize-begin / finalize-end ... finalize_unfinished, split around the one herdr diagnostic probe bash still runs Issue #50 (the registry, not herdr's agent_status, says whether a session is free, and create --onboard waits for the onboarding to end): wait-terminal --job J --registry-dir R --timeout T [--watch-pid P] prints the job's terminal status, or `timeout` / `missing` onboard-exit create's exit code for it: 0, 44, 45 open-jobs --job J --agent-session S --registry-dir R diagnostic: the live open jobs of S wait-open-jobs ... --wait-s W --start-time T0 --timeout T [--watch-pid P] [--dispatch-token K] [--hold] prints `clear`, `busy `, or `superseded` (the record is not this attempt's pending one any more) hold-admission --job J --agent-session S --dispatch-token K --registry-dir R --timeout T [--start-time T0] [--watch-pid P] the detached hold `clear` starts with --hold finalize-end answers on fd 3 (stdout when fd 3 is closed): its stdout carries registry.py's close output to the wrapper's stdout, as before. It runs as a child of the wrapper and must not outlive it (review R1): with --watch-pid it checks that the wrapper is still alive throughout both waits and before each side effect, and leaves without closing or publishing once it is gone - as the same loops in bash ended with the wrapper. This module imports only the standard library and runs on Python 3.9 (macOS /usr/bin/python3). The wrapper runs it as `python -m lib_py.delegation.wrapper`, which imports the package first: lib_py/delegation/__init__.py loads the service and with it paho-mqtt, so the wrapper's $PY needs paho - as it always did for the subscriber and publish_event.py. registry.py and publish_event.py run as the processes the wrapper ran, with its interpreter. """ from __future__ import annotations import argparse import calendar import json import os import subprocess import sys import time from typing import List, Optional # job_subscriber.py's RC_* (a test keeps them equal; importing it would load paho). SUB_RC_COMPLETED, SUB_RC_ERROR, SUB_RC_NO_EVENT, SUB_RC_BROKER_UNAVAILABLE, SUB_RC_WALL_CLOCK = 0, 1, 2, 3, 4 TERMINAL = ("completed", "error", "cancelled") EXIT_CODES = {"completed": 0, "error": 1, "timeout": 2, "broker_unavailable": 3, "cancelled": 5} # Issue #50: create --onboard's exit code for how the onboarding job ended. # 44 = the agent ended it with error (or it was cancelled), 45 = it did not end # within MAM_ONBOARD_WAIT_S (or its record is gone). The session is kept either way. ONBOARD_EXIT_CODES = {"completed": 0, "error": 44, "cancelled": 44, "timeout": 45, "missing": 45} DEFAULT_IDLE_TIMEOUT_SEC = 120 # registry.py's default, as a last resort # Issue #50: a pending record's liveness is its dispatcher's signal - written # every DISPATCH_SIGNAL_EVERY s while it waits and, once admitted, while it # holds the session - and it stays live for EVERY x MISSES s after the last # one (Issue #18's heartbeat x misses, on the dispatcher's side). The window # is not ack_timeout_sec: that is the agent's budget to answer a prompt and # goes down to 1 s, below any cadence a second-resolution signal can keep # (review aa1bff92 F2). DISPATCH_SIGNAL_FIELD = "dispatch_wait_at" DISPATCH_SIGNAL_EVERY = 10 DISPATCH_SIGNAL_MISSES = 3 DISPATCH_SIGNAL_WINDOW = DISPATCH_SIGNAL_EVERY * DISPATCH_SIGNAL_MISSES # The admission (review aa1bff92 F1): `clear` is persisted on the admitted # record, scoped to its attempt by the dispatch token, so a later candidate - # a lower id in the same second, a redispatched older loop job - waits for it # until it ends or its signal goes stale. ADMITTED_AT_FIELD = "dispatch_admitted_at" ADMITTED_TOKEN_FIELD = "dispatch_admitted_token" DEFAULT_ACK_TIMEOUT_SEC = 180 # registry.DEFAULT_ACK_TIMEOUT_SEC (a test keeps them equal) def _now() -> int: return int(time.time()) # bash's $(date +%s) def disk_status(registry_dir: str, job_id: str) -> str: """The record's status, "" when it is missing or unreadable.""" try: with open(os.path.join(registry_dir, f"{job_id}.json")) as fh: return str(json.load(fh).get("status", "")) except Exception: return "" def sub_status(rc: int, registry_dir: str, job_id: str) -> str: if rc == SUB_RC_COMPLETED: return "completed" if rc == SUB_RC_ERROR: return "error" if rc == SUB_RC_BROKER_UNAVAILABLE: disk = disk_status(registry_dir, job_id) return disk if disk in TERMINAL else "broker_unavailable" return "timeout" # SUB_RC_NO_EVENT, SUB_RC_WALL_CLOCK (and any other code) def subscriber_up(rc: int) -> bool: return rc in (SUB_RC_COMPLETED, SUB_RC_BROKER_UNAVAILABLE) def fail_closed() -> bool: return (os.environ.get("MAM_DELEGATE_FAIL_CLOSED") or "1") != "0" def exit_code(status: str) -> int: return EXIT_CODES.get(status, 1) if fail_closed() else 0 def wait_rc(rc: int) -> int: return SUB_RC_NO_EVENT if rc == SUB_RC_WALL_CLOCK else rc def _bare(session: str) -> str: return session[len("herdr:"):] if session.startswith("herdr:") else session def stall_watch(session: str, agent: str) -> bool: """A remote or human job has no local pane to watch.""" if (os.environ.get("MAM_STALL_WATCH") or "1") == "0": return False if session.startswith("remote:"): print("stall watch: remote job, no local pane - stall detection off", file=sys.stderr) return False if agent == "human" or _bare(session) == "human": print("stall watch: human job, no local pane - stall detection off", file=sys.stderr) return False return True def recent_disk_heartbeat(registry_dir: str, job_id: str, window: float) -> bool: """ISSUE-14: publish_event appends every event to the job's events.log before its wire publish, so a heartbeat the subscriber never saw is visible here; true when the last one is at most `window` s old.""" try: last = "" with open(os.path.join(registry_dir, f"{job_id}.events.log")) as fh: for line in fh: if line.strip(): last = line seen = calendar.timegm(time.strptime(json.loads(last)["timestamp"], "%Y-%m-%dT%H:%M:%SZ")) except Exception: return False return time.time() - seen <= window class OwnerGone(Exception): """The wrapper that ran finalize-end is gone.""" def _alive(pid: int) -> bool: """A running process. An exited one its caller has not reaped yet is a zombie that still answers kill(0) (review R1, round 2): `ps -o stat=` (POSIX, no /proc) tells it apart, as tests/conftest.py::_pid_alive does.""" try: os.kill(pid, 0) except ProcessLookupError: return False except PermissionError: return True try: state = subprocess.run(["ps", "-o", "stat=", "-p", str(pid)], capture_output=True, text=True, timeout=10).stdout.strip() except (OSError, subprocess.SubprocessError): return True # no ps answer: kill(0) said it exists return bool(state) and not state.startswith("Z") def _check_owner(a: argparse.Namespace) -> None: if a.watch_pid and not _alive(a.watch_pid): raise OwnerGone(a.watch_pid) def _sleep(a: argparse.Namespace, seconds: float) -> None: """time.sleep that looks at the wrapper every 0.5 s. The span is measured on the monotonic clock, as bash's `sleep N` is: a wall clock stepped back during the wait must not stretch it (review R1, codex). Deadlines stay on the wall clock (_now).""" end = time.monotonic() + seconds while True: _check_owner(a) left = end - time.monotonic() if left <= 0: return time.sleep(min(left, 0.5)) def _tail(path: str, n: int) -> List[str]: with open(path, errors="replace") as fh: return fh.read().splitlines()[-n:] # --------------------------------------------------------------------------- # Issue #50 A: create --onboard waits for the onboarding job to end # --------------------------------------------------------------------------- def onboard_exit(status: str) -> int: return ONBOARD_EXIT_CODES.get(status, 45) def wait_terminal(a: argparse.Namespace) -> str: """The job's terminal status once it has one; `timeout` after --timeout s, `missing` when the record is gone. Reads the disk every second.""" deadline = _now() + a.timeout record = os.path.join(a.registry_dir, f"{a.job}.json") while True: disk = disk_status(a.registry_dir, a.job) if disk in TERMINAL: return disk if not os.path.isfile(record): return "missing" if _now() >= deadline: return "timeout" _sleep(a, 1) # --------------------------------------------------------------------------- # Issue #50 C: the registry says whether a session is free # --------------------------------------------------------------------------- def _parse_ts(text: str) -> Optional[int]: try: return calendar.timegm(time.strptime(str(text), "%Y-%m-%dT%H:%M:%SZ")) except (TypeError, ValueError): return None def _load_record(registry_dir: str, job_id: str) -> Optional[dict]: try: with open(os.path.join(registry_dir, f"{job_id}.json")) as fh: rec = json.load(fh) except Exception: return None return rec if isinstance(rec, dict) else None def _int_field(rec: dict, key: str, default: int) -> int: try: return max(1, int(rec.get(key) or default)) except (TypeError, ValueError): return default def open_jobs(registry_dir: str, session: str, job_id: str, now: Optional[int] = None): """(live, stale) among the other non-terminal records of `session`. live: [(created_at, job_id, status)] - open jobs the Issue #18 contract still calls alive: `running` with an event in events.log within idle_timeout_sec; `pending` with a waiter's signal (`dispatch_wait_at`) within DISPATCH_SIGNAL_WINDOW, or - never signalled - created within its own ack_timeout_sec: a producer that registers and injects without a waiter (create --onboard) waits that long for `started` (review 55219be1 R2). Such a never-signalled record is reported as `injected` and a live pending record whose admission names its current dispatch token as `admitted`: both hold the session as a running job does, whatever their place in the FIFO - the prompt is already in the pane (review 5a09a13f F-A: in the FIFO an `injected` record of the same second lost to a lower job id). A waiter's own record is unsignalled only until its first scan (wait_open_jobs signals first). stale: [(job_id, reason)] - non-terminal records nothing has touched for longer than that: a wrapper killed before or while dispatching, a zombie's unclosed record, an onboarding never acknowledged. They are reported, never waited for, so a dead record cannot block a session forever.""" now = _now() if now is None else now sess = _bare(session) live, stale = [], [] try: names = sorted(os.listdir(registry_dir)) except OSError: return live, stale for name in names: if not name.endswith(".json"): continue other = name[:-len(".json")] if other == job_id: continue rec = _load_record(registry_dir, other) if rec is None or _bare(str(rec.get("agent_session") or "")) != sess: continue status = rec.get("status") # a submit --dry-run is never dispatched (review 1e25cf2f F1); once a # pick-pending claims it, it is running work like any other (211037ab F1) if rec.get("dry_run") and status == "pending": continue if status == "running": idle = _int_field(rec, "idle_timeout_sec", DEFAULT_IDLE_TIMEOUT_SEC) if recent_disk_heartbeat(registry_dir, other, idle): live.append((str(rec.get("created_at") or ""), other, status)) else: stale.append((other, f"running, no event for {idle}s")) elif status == "pending": if rec.get(DISPATCH_SIGNAL_FIELD): seen, window = _parse_ts(rec.get(DISPATCH_SIGNAL_FIELD)), DISPATCH_SIGNAL_WINDOW why = f"no dispatch signal for {window}s" else: seen, window = _parse_ts(rec.get("created_at")), _int_field(rec, "ack_timeout_sec", DEFAULT_ACK_TIMEOUT_SEC) why = f"no dispatch signal and no started within {window}s" if seen is not None and now - seen <= window: if not rec.get(DISPATCH_SIGNAL_FIELD): kind = "injected" elif rec.get(ADMITTED_TOKEN_FIELD) and rec.get(ADMITTED_TOKEN_FIELD) == rec.get("dispatch_token"): kind = "admitted" else: kind = status live.append((str(rec.get("created_at") or ""), other, kind)) else: stale.append((other, f"pending, {why}")) return live, stale def _first_in_line(live, mine) -> bool: """No live running, admitted or injected job, and `mine` earliest among the live pending ones (FIFO by (created_at, job_id)).""" if any(st != "pending" for _c, _j, st in live): return False return min([(c, j) for c, j, _st in live] + [mine]) == mine def _own(rec: Optional[dict], token: Optional[str], session: str) -> bool: """The record is this attempt's pending one, still bound to the session the waiter scans - the identity the waiter and its hold are pinned to for their whole life (review 55219be1 R1): the token the wrapper read for this attempt and the session it asked about, never re-read from the record. A redispatch (new token), a move to another session or a close by someone else ends them without a write.""" return bool(rec) and bool(token) and rec.get("status") == "pending" and rec.get("dispatch_token") == token \ and _bare(str(rec.get("agent_session") or "")) == _bare(session) def _started_own(rec: Optional[dict], token: Optional[str], session: str) -> bool: """This attempt's record, in this session, already `running`: its own `started` landed before the prompt went out (the registry ack of Issue #14 wins over an unreachable broker; a test that acknowledges at registration). Nothing was redispatched, moved or closed by someone else - there is just nothing left to wait for.""" return bool(rec) and bool(token) and rec.get("status") == "running" and rec.get("dispatch_token") == token \ and _bare(str(rec.get("agent_session") or "")) == _bare(session) def _merge_own(registry_dir: str, job_id: str, token: Optional[str], session: str, source: str, fields: dict, guard=None) -> bool: """One merge through the engine, committed only while the record re-read under the registry lock is still this attempt's (_own) and `guard`, if any, agrees. False when it is not, or the record is gone or unwritable.""" from lib_py.delegation import engine def ok(record): return _own(record, token, session) and (guard is None or guard(record)) try: return engine.transition(job_id, "merge", registry_dir=registry_dir, source=source, guard=ok, fields=fields).committed except Exception as exc: # the record is gone or unwritable: nothing written, say so print(f"dispatch wait: could not write {source} on {job_id}: {exc}", file=sys.stderr) return False def _admit(registry_dir: str, session: str, job_id: str, token: Optional[str], mine) -> bool: """`clear`, persisted: one merge through the engine whose guard re-reads the session's open jobs under the registry lock, so two waiters cannot both be admitted. Scoped to this attempt and session (_own: a redispatch voids it); the signal is renewed with it. False when the record is not this attempt's pending one any more or another waiter was admitted first - the caller keeps waiting and sees that one.""" from lib_py.delegation import mqtt_common def first(_record): live, _stale = open_jobs(registry_dir, session, job_id) return _first_in_line(live, mine) now = mqtt_common._utcnow() return _merge_own(registry_dir, job_id, token, session, "dispatch_admit", {DISPATCH_SIGNAL_FIELD: now, ADMITTED_AT_FIELD: now, ADMITTED_TOKEN_FIELD: token}, first) def _start_hold(a: argparse.Namespace, token: str) -> None: """The waiter answers `clear` and ends; a detached hold keeps the admitted record's signal fresh until the record leaves pending (the agent's `started`, a close), its attempt changes, the wrapper is gone, or the wall clock ends (review aa1bff92 F2). It is given the waiter's pinned identity, not the record's. Its output is not the answer.""" cmd = [sys.executable, "-m", "lib_py.delegation.wrapper", "hold-admission", "--job", a.job, "--agent-session", a.agent_session, "--dispatch-token", token, "--registry-dir", a.registry_dir, "--timeout", str(a.timeout)] if a.start_time is not None: cmd += ["--start-time", str(a.start_time)] if a.watch_pid: cmd += ["--watch-pid", str(a.watch_pid)] subprocess.Popen(cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, env=dict(os.environ, PYTHONPATH=_pythonpath())) def _pythonpath() -> str: skills = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) extra = os.environ.get("PYTHONPATH", "") return skills + (os.pathsep + extra if extra else "") def hold_admission(a: argparse.Namespace) -> str: """`released` once the record is not the admitted pending one of the attempt and session the hold was given (a token that changed before it started counts), `timeout` at the wall clock; the signal is renewed every DISPATCH_SIGNAL_EVERY s in between, under the same identity (_merge_own). OwnerGone (the wrapper) ends it.""" token = a.dispatch_token deadline = (a.start_time if a.start_time is not None else _now()) + a.timeout last_signal = _now() while True: _sleep(a, 1) rec = _load_record(a.registry_dir, a.job) if not _own(rec, token, a.agent_session) or rec.get(ADMITTED_TOKEN_FIELD) != token: return "released" now = _now() if now >= deadline: return "timeout" if now - last_signal >= DISPATCH_SIGNAL_EVERY: _signal_dispatch_wait(a.registry_dir, a.job, token, a.agent_session, admitted=True) last_signal = now def _signal_dispatch_wait(registry_dir: str, job_id: str, token: Optional[str], session: str, admitted: bool = False) -> bool: """The waiter's liveness: a merge of dispatch_wait_at into its own pending record through the registry's one transition boundary (status untouched, no event in events.log - that log is the agent's), only while the record is still this attempt's (_own); for the hold, only while its admission stands. False when nothing was written.""" from lib_py.delegation import mqtt_common # loads paho, as the package import did already guard = (lambda record: record.get(ADMITTED_TOKEN_FIELD) == token) if admitted else None return _merge_own(registry_dir, job_id, token, session, "dispatch_wait", {DISPATCH_SIGNAL_FIELD: mqtt_common._utcnow()}, guard) def wait_open_jobs(a: argparse.Namespace) -> str: """`clear` when no live running or admitted job of the session remains, this job is the earliest live pending one (FIFO by (created_at, job_id)) and its admission is committed (_admit); `busy ` when the bound runs out first; `superseded` once the record is not this attempt's pending one in this session any more (redispatched, moved, closed by someone else) - nothing is written or closed then, the record is its owner's; `clear` at once, with no admission, when it is this attempt's and already running (_started_own). The bound is min(--wait-s, this job's remaining wall clock). The identity is pinned at the start: --dispatch-token (the token the wrapper read for this attempt), else the record's token read once. The waiter signals on its own record before its first scan - an unsignalled live record is one a producer injected without a waiter (open_jobs: `injected`), and this one is not - and every DISPATCH_SIGNAL_EVERY s after, so other waiters see it alive; a waiter whose wrapper is gone stops (OwnerGone) and its signal goes stale. With --hold the admission is kept alive after the answer (_start_hold).""" me = _load_record(a.registry_dir, a.job) or {} token = getattr(a, "dispatch_token", None) or me.get("dispatch_token") mine = (str(me.get("created_at") or ""), a.job) remaining = (a.start_time if a.start_time is not None else _now()) + a.timeout - _now() deadline = _now() + max(0, min(a.wait_s, remaining)) reported = set() last_signal = None while True: now = _now() rec = _load_record(a.registry_dir, a.job) if not _own(rec, token, a.agent_session): return "clear" if _started_own(rec, token, a.agent_session) else "superseded" if last_signal is None or now - last_signal >= DISPATCH_SIGNAL_EVERY: _signal_dispatch_wait(a.registry_dir, a.job, token, a.agent_session) last_signal = now live, stale = open_jobs(a.registry_dir, a.agent_session, a.job, now) for other, reason in stale: if other not in reported: reported.add(other) print(f"stale open job {other} ignored ({reason})", file=sys.stderr) if _first_in_line(live, mine) and _admit(a.registry_dir, a.agent_session, a.job, token, mine): if getattr(a, "hold", False): _start_hold(a, token) return "clear" if now >= deadline: return "busy " + " ".join(j for _c, j, _st in live) _sleep(a, 1) def finalize_begin(a: argparse.Namespace) -> str: """`keep` (FINAL_JOB_STATUS untouched), `done `, or `go|probe `.""" if not fail_closed() or a.status == "completed": return "keep" stall_out = os.path.join(a.registry_dir, f"{a.job}.stall.out") if os.path.isfile(stall_out) and os.path.getsize(stall_out) > 0: # Issue #30: what the stall watcher saw goes with an unfinished job's close. print(f"stall watch log for {a.job} (last 3 lines):", file=sys.stderr) try: for line in _tail(stall_out, 3): print(line, file=sys.stderr) except OSError: pass disk = disk_status(a.registry_dir, a.job) if disk == "completed": return "done completed" final = disk if disk in TERMINAL else a.status now = _now() deadline = (a.start_time if a.start_time is not None else now) + a.timeout evidence_until = now + a.idle_timeout sess = _bare(a.session) # herdr status and the pane are a diagnostic only (ISSUE-14 / ISSUE-18): # bash runs that probe, bounded by the deadline, when asked to. probe = (a.status == "timeout" and a.agent != "human" and sess != "human" and sess != "" and not a.session.startswith("remote:")) return f"{'probe' if probe else 'go'} {final} {evidence_until} {deadline}" def _run(a: argparse.Namespace, script: str, args: List[str], merge: bool) -> int: sys.stdout.flush() cmd = [a.python, os.path.join(a.scripts_dir, script), "--registry-dir", a.registry_dir, *args] return subprocess.call(cmd, stderr=subprocess.STDOUT if merge else subprocess.DEVNULL) def finalize_end(a: argparse.Namespace) -> str: """The rest of finalize_unfinished; returns FINAL_JOB_STATUS.""" job, reg, status, final = a.job, a.registry_dir, a.status, a.final clock = "" deadline = (a.start_time if a.start_time is not None else _now()) + a.timeout if status == "broker_unavailable": # Re-observe the disk at 3 s intervals until this dispatch's wall-clock # deadline, read off the clock on every pass. while True: left = deadline - _now() if left <= 0: break _sleep(a, min(left, 3)) disk = disk_status(reg, job) if disk in TERMINAL: status = final = disk if disk == "completed": return final break if _now() >= deadline: # Issue #33 (review R3): a wait into the deadline is the wall clock clock = "wall" if status == "timeout": # ISSUE-14 / ISSUE-18: keep re-observing the disk while the contract # says the agent is alive - an event in events.log at most IDLE_TIMEOUT # (K x N) old. Entering this grace opens one window and every disk event # re-opens it; evidence never extends the wall clock. evidence_until = a.evidence_until while _now() < deadline: disk = disk_status(reg, job) if disk in TERMINAL: status = final = disk if disk == "completed": return final break if recent_disk_heartbeat(reg, job, a.idle_timeout): evidence_until = _now() + a.idle_timeout elif _now() >= evidence_until: break if deadline - _now() <= 0: break _sleep(a, 1) if _now() >= deadline: clock = "wall" # Issue #33: the subscriber's rc says which clock ran out (4 wall clock, # 2 no event); a grace above that ends on the deadline is the wall clock too. clock = clock or "idle" if a.sub_rc == SUB_RC_WALL_CLOCK: clock = "wall" close_req = "wall_clock_timeout" if clock == "wall" and a.phase != "inject" else "deadline_timeout" detail = "" if a.phase in ("preflight", "idle_wait", "registry_wait"): close_req, detail = "cancelled", "session_busy" # Issue #34: a target the preflight could not see was never busy if a.phase == "preflight" and a.preflight_reason: detail = a.preflight_reason if close_req != "cancelled": print(f"finalize: {job} closed by {'wall_clock' if clock == 'wall' else 'no_event'} clock " f"(budget {a.timeout}s / idle {a.idle_timeout}s)", file=sys.stderr) tok = ["--expect-dispatch-token", a.dispatch_token] if a.dispatch_token else [] _check_owner(a) close_args = ["close", "--job", job, "--request", close_req] + (["--detail", detail] if detail else []) + tok if _run(a, "registry.py", close_args, merge=True) == 4: # a dispatch token mismatch (stale attempt): no secondary event (CAS/ABA) print("finalize_unfinished: dispatch token mismatch (stale attempt); suppressing event emission", file=sys.stderr) return final if disk_status(reg, job) == "completed": return "completed" # direct path only: publish the error/cancelled event (loop/discuss forbidden: O-1) if a.type == "direct": _check_owner(a) if close_req == "cancelled": _run(a, "publish_event.py", ["--job", job, "--event", "cancelled", "--detail", detail] + tok, merge=False) elif a.sub_rc != SUB_RC_ERROR or a.phase == "inject": err_detail = "agent_blocked" if a.phase == "inject" else close_req _run(a, "publish_event.py", ["--job", job, "--event", "error", "--detail", err_detail] + tok, merge=False) return final def _answer(text: str) -> None: try: os.write(3, (text + "\n").encode()) except OSError: print(text) def _parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(prog="python -m lib_py.delegation.wrapper") sub = p.add_subparsers(dest="cmd", required=True) s = sub.add_parser("sub-status") s.add_argument("rc", type=int) s.add_argument("--job", required=True) s.add_argument("--registry-dir", required=True) s = sub.add_parser("subscriber-up") s.add_argument("rc", type=int) s.add_argument("--print", action="store_true", help="print yes/no and exit 0") sub.add_parser("exit-code").add_argument("status") sub.add_parser("wait-rc").add_argument("rc", type=int) s = sub.add_parser("stall-watch") s.add_argument("session") s.add_argument("agent") s.add_argument("--print", action="store_true", help="print yes/no and exit 0") s = sub.add_parser("wait-terminal") s.add_argument("--job", required=True) s.add_argument("--registry-dir", required=True) s.add_argument("--timeout", type=int, required=True) s.add_argument("--watch-pid", type=int, default=None, help="the caller's pid; gone -> stop waiting") sub.add_parser("onboard-exit").add_argument("status") for name in ("open-jobs", "wait-open-jobs"): s = sub.add_parser(name) s.add_argument("--job", required=True) s.add_argument("--agent-session", required=True) s.add_argument("--registry-dir", required=True) if name == "wait-open-jobs": s.add_argument("--wait-s", type=int, required=True, help="MAM_DISPATCH_WAIT_S") s.add_argument("--start-time", type=int, default=None) s.add_argument("--timeout", type=int, required=True, help="the job's wall-clock budget") s.add_argument("--watch-pid", type=int, default=None, help="the wrapper's pid; gone -> stop waiting") s.add_argument("--dispatch-token", default=None, help="this attempt's token (pinned; else read once)") s.add_argument("--hold", action="store_true", help="keep the admission's signal fresh after `clear`") s = sub.add_parser("hold-admission") s.add_argument("--job", required=True) s.add_argument("--agent-session", required=True) s.add_argument("--dispatch-token", required=True, help="the waiter's pinned token") s.add_argument("--registry-dir", required=True) s.add_argument("--start-time", type=int, default=None) s.add_argument("--timeout", type=int, required=True) s.add_argument("--watch-pid", type=int, default=None, help="the wrapper's pid; gone -> stop holding") for name in ("finalize-begin", "finalize-end"): s = sub.add_parser(name) s.add_argument("--job", required=True) s.add_argument("--registry-dir", required=True) s.add_argument("--status", required=True) s.add_argument("--start-time", type=int, default=None) s.add_argument("--timeout", type=int, required=True) s.add_argument("--idle-timeout", type=int, required=True) if name == "finalize-begin": s.add_argument("--agent", default="") s.add_argument("--session", default="") else: s.add_argument("--sub-rc", type=int, required=True) s.add_argument("--phase", default="") s.add_argument("--type", default="direct") s.add_argument("--final", required=True) s.add_argument("--evidence-until", type=int, default=0) s.add_argument("--dispatch-token", default="") s.add_argument("--preflight-reason", default="") s.add_argument("--python", default=sys.executable) s.add_argument("--scripts-dir", required=True) s.add_argument("--watch-pid", type=int, default=None, help="the wrapper's pid; gone -> no close, no event") return p def _yes_no(answer: bool, printed: bool) -> int: if printed: print("yes" if answer else "no") return 0 return 0 if answer else 1 def main(argv: Optional[List[str]] = None) -> int: a = _parser().parse_args(argv) if a.cmd == "sub-status": print(sub_status(a.rc, a.registry_dir, a.job)) elif a.cmd == "subscriber-up": return _yes_no(subscriber_up(a.rc), a.print) elif a.cmd == "exit-code": print(exit_code(a.status)) elif a.cmd == "wait-rc": print(wait_rc(a.rc)) elif a.cmd == "stall-watch": return _yes_no(stall_watch(a.session, a.agent), a.print) elif a.cmd == "finalize-begin": print(finalize_begin(a)) elif a.cmd == "wait-terminal": try: print(wait_terminal(a)) except OwnerGone: print(f"wait-terminal: caller pid {a.watch_pid} is gone; stopped waiting on job {a.job}", file=sys.stderr) elif a.cmd == "onboard-exit": print(onboard_exit(a.status)) elif a.cmd == "open-jobs": live, stale = open_jobs(a.registry_dir, a.agent_session, a.job) for created, job, status in live: print(f"{job} {status:8s} created {created}") for job, reason in stale: print(f"{job} stale {reason}") elif a.cmd == "wait-open-jobs": try: print(wait_open_jobs(a)) except OwnerGone: print(f"wait-open-jobs: wrapper pid {a.watch_pid} is gone; stopped waiting on job {a.job}", file=sys.stderr) elif a.cmd == "hold-admission": try: print(hold_admission(a)) except OwnerGone: print(f"hold-admission: wrapper pid {a.watch_pid} is gone; stopped holding job {a.job}", file=sys.stderr) else: try: _answer(finalize_end(a)) except OwnerGone: print(f"finalize: wrapper pid {a.watch_pid} is gone; job {a.job} left unclosed", file=sys.stderr) return 0 if __name__ == "__main__": sys.exit(main())