GodopuandClaude Opus 5.5
ec9586bb9b
- wrapper: TIMEOUT falls back to MAM_JOB_TIMEOUT_S (--timeout > env > 3600); an invalid value warns and keeps 3600. - loop: --job-timeout <sec> (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 <noreply@anthropic.com>
482 lines
22 KiB
Python
482 lines
22 KiB
Python
"""remote.py — deliver a job to an agent on another host over the broker (Issue #18).
|
|
|
|
The herdr transport injects the prompt into a pane on this machine and reads
|
|
the agent's `started`/`progress`/`completed` events back over the broker. A
|
|
remote agent shares neither the pane nor the orchestrator's disk, so the
|
|
prompt itself has to travel over the wire too:
|
|
|
|
orchestrator --retained `<topic_prefix>/dispatch`--> remote runner
|
|
remote runner --`<topic_prefix>/events` + `/state`--> orchestrator's subscriber
|
|
|
|
`publish_dispatch` (wrapper side) publishes one retained dispatch message
|
|
carrying the brief, the job's auth_token / dispatch_token / attempt_id and the
|
|
liveness contract (heartbeat N, misses K, ack T). `run_runner` (remote side)
|
|
subscribes for its session name, mirrors the job into its own registry,
|
|
publishes `started` with the automatic heartbeat, runs the agent command, and
|
|
publishes the terminal event. Every event it publishes is signed with the
|
|
job's auth_token and carries the orchestrator's dispatch identity, so the
|
|
orchestrator's unchanged subscriber accepts it; the subscriber's `--remote`
|
|
mode mirrors those events into the orchestrator's registry.
|
|
|
|
Trust: a dispatch makes a runner execute a command, so it is signed with a
|
|
pre-shared secret (`MAM_DISPATCH_SECRET`, from each host's own .mam.env) and a
|
|
runner ignores anything unsigned or mis-signed. The auth_token travels inside
|
|
that signed message; use this only on a private broker (see MESSAGING.md).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import logging
|
|
import os
|
|
import queue
|
|
import shlex
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from typing import Any, Dict, Optional
|
|
|
|
from lib_py.delegation import mqtt_common, registry
|
|
from lib_py.delegation.mqtt_common import (
|
|
DEFAULT_TOPIC_ROOT,
|
|
SCHEMA_VERSION,
|
|
broker_config_from_env,
|
|
broker_config_from_job,
|
|
default_registry_dir,
|
|
load_job,
|
|
make_client,
|
|
)
|
|
from lib_py.delegation.publisher import OneShotPublisher
|
|
|
|
logger = logging.getLogger("delegate_job.remote")
|
|
|
|
DISPATCH_KIND = "dispatch"
|
|
SECRET_ENV = "MAM_DISPATCH_SECRET"
|
|
SESSION_PREFIX = "remote:"
|
|
|
|
|
|
def is_remote_session(agent_session: Optional[str]) -> bool:
|
|
return bool(agent_session) and str(agent_session).startswith(SESSION_PREFIX)
|
|
|
|
|
|
def session_name(agent_session: str) -> str:
|
|
return agent_session[len(SESSION_PREFIX):] if is_remote_session(agent_session) else agent_session
|
|
|
|
|
|
def dispatch_topic_for(job: Dict[str, Any]) -> str:
|
|
prefix = job.get("topic_prefix") or mqtt_common.topic_prefix_for(job["job_id"])
|
|
return f"{prefix}/dispatch"
|
|
|
|
|
|
def _canonical(payload: Dict[str, Any]) -> bytes:
|
|
body = {k: v for k, v in payload.items() if k != "sig"}
|
|
return json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
|
|
|
|
|
|
def sign_dispatch(payload: Dict[str, Any], secret: str) -> Dict[str, Any]:
|
|
payload = dict(payload)
|
|
payload["sig"] = hmac.new(secret.encode("utf-8"), _canonical(payload), hashlib.sha256).hexdigest()
|
|
return payload
|
|
|
|
|
|
def verify_dispatch(payload: Dict[str, Any], secret: Optional[str]) -> bool:
|
|
"""True when the dispatch carries a valid signature for ``secret``. With
|
|
no secret configured on the runner side nothing is accepted."""
|
|
if not secret:
|
|
return False
|
|
sig = payload.get("sig")
|
|
if not isinstance(sig, str):
|
|
return False
|
|
expected = hmac.new(secret.encode("utf-8"), _canonical(payload), hashlib.sha256).hexdigest()
|
|
return hmac.compare_digest(sig, expected)
|
|
|
|
|
|
def build_dispatch(job: Dict[str, Any], session: str, instructions: str) -> Dict[str, Any]:
|
|
return {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"kind": DISPATCH_KIND,
|
|
"job_id": job["job_id"],
|
|
"session": session,
|
|
"instructions": instructions,
|
|
"brief": job.get("prompt", ""),
|
|
"role": job.get("role"),
|
|
"agent": job.get("agent"),
|
|
"auth_token": job.get("auth_token"),
|
|
"dispatch_token": job.get("dispatch_token"),
|
|
"attempt_id": job.get("attempt_id", job.get("iteration", 1)),
|
|
"timeout_sec": job.get("timeout_sec", 3600),
|
|
"idle_timeout_sec": job.get("idle_timeout_sec"),
|
|
"heartbeat_sec": job.get("heartbeat_sec", registry.DEFAULT_HEARTBEAT_SEC),
|
|
"heartbeat_misses": job.get("heartbeat_misses", registry.DEFAULT_HEARTBEAT_MISSES),
|
|
"ack_timeout_sec": job.get("ack_timeout_sec", registry.DEFAULT_ACK_TIMEOUT_SEC),
|
|
"topic_prefix": job.get("topic_prefix") or mqtt_common.topic_prefix_for(job["job_id"]),
|
|
"origin": socket.gethostname(),
|
|
"timestamp": mqtt_common._utcnow(),
|
|
}
|
|
|
|
|
|
def publish_dispatch(job_id: str, registry_dir: str, session: str, instructions: str,
|
|
secret: Optional[str] = None, allow_unsigned: bool = False) -> str:
|
|
"""Publish the retained dispatch for ``job_id``; returns the topic."""
|
|
job = load_job(job_id, registry_dir)
|
|
secret = secret or os.environ.get(SECRET_ENV)
|
|
payload = build_dispatch(job, session, instructions)
|
|
if secret:
|
|
payload = sign_dispatch(payload, secret)
|
|
elif not allow_unsigned:
|
|
raise RuntimeError("{} is not set: refusing to publish an unsigned dispatch "
|
|
"(set it in .mam.env on both hosts, or pass --allow-unsigned)".format(SECRET_ENV))
|
|
topic = dispatch_topic_for(job)
|
|
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
mqtt_common.with_retry(OneShotPublisher().publish_raw, attempts=3,
|
|
exceptions=(OSError, TimeoutError, ConnectionError, ValueError))(
|
|
broker_config_from_job(job), topic, body, True)
|
|
mqtt_common.append_event(job_id, {"event": "dispatched", "topic": topic, "session": session,
|
|
"attempt_id": payload["attempt_id"]})
|
|
logger.info("dispatched job %s to remote session %s on %s", job_id, session, topic)
|
|
return topic
|
|
|
|
|
|
def clear_dispatch(config, topic: str) -> None:
|
|
"""Retire a retained dispatch (empty retained payload) so a runner that
|
|
restarts does not execute it again."""
|
|
try:
|
|
OneShotPublisher().publish_raw(config, topic, b"", True)
|
|
except Exception as exc: # pragma: no cover - best effort
|
|
logger.warning("could not clear dispatch %s: %s", topic, exc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# remote side
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _attempt_of(record: Dict[str, Any]) -> int:
|
|
try:
|
|
return int(record.get("attempt_id", record.get("iteration", 1)) or 1)
|
|
except (TypeError, ValueError):
|
|
return 1
|
|
|
|
|
|
def mirror_job(d: Dict[str, Any], registry_dir: str, session: str) -> Optional[str]:
|
|
"""Create the local job record a remote agent publishes against, or refuse
|
|
the dispatch (None).
|
|
|
|
The record carries the orchestrator's job_id, auth_token, dispatch_token
|
|
and attempt_id, so `publish_event` on this host produces events the
|
|
orchestrator's subscriber accepts. Admission (review round 2, codex R1): a
|
|
valid signature proves where a dispatch came from, not that it is current,
|
|
so
|
|
|
|
* the same dispatch_token is idempotent while its attempt is open, and is
|
|
never run again once that attempt reached a terminal;
|
|
* a lower attempt, or the same attempt under another token, is a stale or
|
|
conflicting copy: refused before anything is written, so the record and
|
|
its events.log stay as they are;
|
|
* only a strictly newer attempt replaces the record, and that check and
|
|
the write are one step under the registry lock
|
|
(``register_job(supersede_older_attempt=True)``). The events.log is kept:
|
|
the history of earlier attempts is not erased.
|
|
"""
|
|
jid = d["job_id"]
|
|
try:
|
|
incoming = int(d.get("attempt_id") or 1)
|
|
except (TypeError, ValueError):
|
|
logger.warning("drop dispatch for %s: attempt_id %r is not a number", jid, d.get("attempt_id"))
|
|
return None
|
|
path = mqtt_common._job_path(jid, registry_dir)
|
|
if path.exists():
|
|
old = load_job(jid, registry_dir)
|
|
if old.get("dispatch_token") == d.get("dispatch_token"):
|
|
if old.get("status") in ("completed", "error", "cancelled"):
|
|
return None # this attempt already ran to its end: a re-sent dispatch is not run again
|
|
return jid # same attempt, already mirrored (duplicate delivery)
|
|
if incoming <= _attempt_of(old):
|
|
logger.warning("refusing dispatch for %s attempt %s: this runner already holds attempt %s "
|
|
"(a stale or conflicting copy)", jid, incoming, _attempt_of(old))
|
|
return None
|
|
try:
|
|
registry.register_job(
|
|
prompt=d.get("brief", ""),
|
|
agent=d.get("agent") or "remote",
|
|
agent_session=SESSION_PREFIX + session,
|
|
role=d.get("role") or "Worker",
|
|
timeout_sec=int(d.get("timeout_sec") or 3600),
|
|
idle_timeout_sec=d.get("idle_timeout_sec"),
|
|
registry_dir=registry_dir,
|
|
job_id=jid,
|
|
auth_token=d.get("auth_token"),
|
|
heartbeat_sec=int(d.get("heartbeat_sec") or registry.DEFAULT_HEARTBEAT_SEC),
|
|
heartbeat_misses=int(d.get("heartbeat_misses") or registry.DEFAULT_HEARTBEAT_MISSES),
|
|
ack_timeout_sec=int(d.get("ack_timeout_sec") or registry.DEFAULT_ACK_TIMEOUT_SEC),
|
|
dispatch_token=d.get("dispatch_token"),
|
|
attempt_id=incoming,
|
|
transport="remote",
|
|
supersede_older_attempt=True,
|
|
)
|
|
except FileExistsError as exc: # lost the race to a newer (or equal) attempt under the lock
|
|
logger.warning("refusing dispatch for %s: %s", jid, exc)
|
|
return None
|
|
brief_dir = os.path.join(registry_dir, jid)
|
|
os.makedirs(brief_dir, exist_ok=True)
|
|
with open(os.path.join(brief_dir, "brief.md"), "w", encoding="utf-8") as fh:
|
|
fh.write(d.get("brief", ""))
|
|
return jid
|
|
|
|
|
|
def _publish(jid: str, registry_dir: str, event: str, detail: str, d: Dict[str, Any],
|
|
heartbeat: Optional[int] = None) -> int:
|
|
from lib_py.delegation import publish_event
|
|
argv = ["--registry-dir", registry_dir, "--job", jid, "--event", event, "--detail", detail]
|
|
if d.get("dispatch_token"):
|
|
argv += ["--expect-dispatch-token", d["dispatch_token"]]
|
|
if d.get("attempt_id") is not None:
|
|
argv += ["--attempt-id", str(d["attempt_id"])]
|
|
if heartbeat:
|
|
# The runner publishes in-process, so the timer must die with the runner.
|
|
argv += ["--heartbeat", str(heartbeat), "--watch-pid", str(os.getpid())]
|
|
return publish_event.main(argv)
|
|
|
|
|
|
def _group_alive(pgid: int) -> bool:
|
|
try:
|
|
os.killpg(pgid, 0)
|
|
except OSError:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _terminate_group(proc: "subprocess.Popen", grace: float = 5.0) -> None:
|
|
"""TERM, then KILL, the agent command's own process group (it runs in its
|
|
own session) and wait until the group is gone. Only that owned group is
|
|
signalled (review round 2, codex R2)."""
|
|
for sig, wait in ((signal.SIGTERM, grace), (signal.SIGKILL, 2.0)):
|
|
try:
|
|
os.killpg(proc.pid, sig)
|
|
except OSError:
|
|
return # the group is already gone
|
|
deadline = time.monotonic() + wait
|
|
while time.monotonic() < deadline:
|
|
proc.poll() # reap the shell itself
|
|
if not _group_alive(proc.pid):
|
|
return
|
|
time.sleep(0.1)
|
|
|
|
|
|
def _run_agent_command(exec_cmd: str, env: Dict[str, str], timeout: float):
|
|
"""Run the agent command in its own session; returns (rc, stdout, stderr,
|
|
timed_out). At the deadline - or when the runner itself is interrupted -
|
|
the whole process group is terminated before this returns, so no
|
|
descendant keeps working past the job's timeout or overlaps the next
|
|
dispatch (review round 2, codex R2: `subprocess.run(shell=True,
|
|
timeout=...)` killed only the shell)."""
|
|
proc = subprocess.Popen(exec_cmd, shell=True, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
text=True, cwd=os.getcwd(), start_new_session=True)
|
|
try:
|
|
out, err = proc.communicate(timeout=timeout)
|
|
return proc.returncode, out, err, False
|
|
except subprocess.TimeoutExpired:
|
|
_terminate_group(proc)
|
|
try:
|
|
out, err = proc.communicate(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
# a descendant that left the group still holds the pipes: stop reading
|
|
for stream in (proc.stdout, proc.stderr):
|
|
try:
|
|
stream.close()
|
|
except Exception:
|
|
pass
|
|
out, err = "", ""
|
|
return 124, out, err, True
|
|
except BaseException:
|
|
_terminate_group(proc)
|
|
raise
|
|
|
|
|
|
def handle_dispatch(d: Dict[str, Any], registry_dir: str, session: str,
|
|
exec_cmd: Optional[str], python: Optional[str] = None) -> int:
|
|
"""Run one dispatched job to its terminal event. Returns the agent rc."""
|
|
jid = mirror_job(d, registry_dir, session)
|
|
if jid is None:
|
|
logger.info("dispatch for %s attempt %s not admitted (already finished here, or a stale or "
|
|
"conflicting copy); not running it", d.get("job_id"), d.get("attempt_id"))
|
|
return 0
|
|
heartbeat = int(d.get("heartbeat_sec") or registry.DEFAULT_HEARTBEAT_SEC)
|
|
rc_started = _publish(jid, registry_dir, "started", "remote runner {} claimed the job".format(session),
|
|
d, heartbeat=heartbeat)
|
|
if rc_started not in (0, 2):
|
|
logger.error("could not publish started for %s (rc %s)", jid, rc_started)
|
|
return 1
|
|
job = load_job(jid, registry_dir)
|
|
rc = 0
|
|
if exec_cmd:
|
|
py = python or sys.executable
|
|
skills_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
env = dict(os.environ)
|
|
env["PYTHONPATH"] = skills_root + (os.pathsep + env["PYTHONPATH"] if env.get("PYTHONPATH") else "")
|
|
env.update({
|
|
"JOB_ID": jid,
|
|
"MAM_JOB_ID": jid,
|
|
"MAM_JOB_REGISTRY_DIR": registry_dir,
|
|
"MAM_JOB_BRIEF": os.path.join(registry_dir, jid, "brief.md"),
|
|
"MAM_JOB_INSTRUCTIONS": d.get("instructions", ""),
|
|
"MAM_PUBLISH_CMD": " ".join(shlex.quote(x) for x in [
|
|
py, "-m", "lib_py.delegation.publish_event", "--registry-dir", registry_dir, "--job", jid]
|
|
+ (["--expect-dispatch-token", d["dispatch_token"]] if d.get("dispatch_token") else [])
|
|
+ (["--attempt-id", str(d["attempt_id"])] if d.get("attempt_id") is not None else [])),
|
|
})
|
|
timeout = float(job.get("timeout_sec", 3600))
|
|
rc, out, err, timed_out = _run_agent_command(exec_cmd, env, timeout)
|
|
if timed_out:
|
|
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)
|
|
if rc != 0:
|
|
errs = (err or "").strip().splitlines()
|
|
detail = "exit {}: {}".format(rc, (errs[-1] if errs else detail)[:200])
|
|
cur = load_job(jid, registry_dir)
|
|
if cur.get("status") not in ("completed", "error", "cancelled"):
|
|
_publish(jid, registry_dir, "completed" if rc == 0 else "error", detail, d)
|
|
return rc
|
|
# No command: an agent on this host publishes the terminal event itself.
|
|
deadline = time.monotonic() + float(job.get("timeout_sec", 3600))
|
|
while time.monotonic() < deadline:
|
|
cur = load_job(jid, registry_dir)
|
|
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", "wall_clock_timeout (remote runner)", d)
|
|
return 2
|
|
|
|
|
|
def run_runner(session: str, registry_dir: str, exec_cmd: Optional[str] = None, once: bool = False,
|
|
secret: Optional[str] = None, allow_unsigned: bool = False,
|
|
timeout: Optional[float] = None, python: Optional[str] = None) -> int:
|
|
secret = secret or os.environ.get(SECRET_ENV)
|
|
if not secret and not allow_unsigned:
|
|
logger.error("%s is not set: a runner must not execute unsigned dispatches "
|
|
"(pass --allow-unsigned only on a private broker)", SECRET_ENV)
|
|
return 1
|
|
cfg = broker_config_from_env()
|
|
client = make_client("remote-runner", cfg)
|
|
inbox: "queue.Queue[Dict[str, Any]]" = queue.Queue()
|
|
topic = f"{DEFAULT_TOPIC_ROOT}/+/dispatch"
|
|
|
|
def on_connect(_c, _u, _flags, reason_code, _props):
|
|
if mqtt_common.reason_code_value(reason_code) != 0:
|
|
logger.error("broker connection failed: rc=%s", reason_code)
|
|
return
|
|
_c.subscribe(topic, qos=1)
|
|
print(f"SUBSCRIBED {topic}", flush=True)
|
|
|
|
def on_message(_c, _u, msg):
|
|
if not msg.payload:
|
|
return # a cleared (retired) dispatch
|
|
try:
|
|
d = json.loads(msg.payload.decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError):
|
|
logger.warning("drop unparseable dispatch on %s", msg.topic)
|
|
return
|
|
if not isinstance(d, dict) or d.get("kind") != DISPATCH_KIND or d.get("schema_version") != SCHEMA_VERSION:
|
|
return
|
|
if d.get("session") != session:
|
|
return
|
|
if secret and not verify_dispatch(d, secret):
|
|
logger.warning("drop dispatch for %s: bad or missing signature", d.get("job_id"))
|
|
return
|
|
if not secret and not allow_unsigned:
|
|
return
|
|
d["_topic"] = msg.topic
|
|
inbox.put(d)
|
|
|
|
client.on_connect = on_connect
|
|
client.on_message = on_message
|
|
client.reconnect_delay_set(min_delay=1, max_delay=16)
|
|
mqtt_common.with_retry(lambda: client.connect(cfg.host, cfg.port, cfg.keepalive),
|
|
attempts=3, base_delay=0.2, max_delay=1.0)()
|
|
# Review round 2 (codex R2): SIGTERM unwinds the runner like an interrupt,
|
|
# so a command still running is terminated together with its process group
|
|
# (_run_agent_command) instead of being left behind as an orphan.
|
|
def _on_term(signum, _frame):
|
|
raise SystemExit(128 + signum)
|
|
|
|
try:
|
|
previous_term = signal.signal(signal.SIGTERM, _on_term)
|
|
except ValueError: # not the main thread (a caller driving the runner in-process)
|
|
previous_term = None
|
|
client.loop_start()
|
|
deadline = time.monotonic() + timeout if timeout else None
|
|
handled = 0
|
|
try:
|
|
while True:
|
|
wait = 1.0 if deadline is None else max(0.0, min(1.0, deadline - time.monotonic()))
|
|
try:
|
|
d = inbox.get(timeout=wait)
|
|
except queue.Empty:
|
|
if deadline is not None and time.monotonic() >= deadline:
|
|
logger.warning("runner timeout after %d job(s)", handled)
|
|
return 0 if handled else 2
|
|
continue
|
|
print("DISPATCH job={} attempt={} from={}".format(d.get("job_id"), d.get("attempt_id"), d.get("origin")),
|
|
flush=True)
|
|
# Retire the retained dispatch before running so a restart cannot replay it.
|
|
clear_dispatch(cfg, d["_topic"])
|
|
rc = handle_dispatch(d, registry_dir, session, exec_cmd, python=python)
|
|
handled += 1
|
|
print("DONE job={} rc={}".format(d.get("job_id"), rc), flush=True)
|
|
if once:
|
|
return 0
|
|
finally:
|
|
if previous_term is not None:
|
|
signal.signal(signal.SIGTERM, previous_term)
|
|
client.loop_stop()
|
|
try:
|
|
client.disconnect()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
parser = argparse.ArgumentParser(description="remote transport for delegated jobs (Issue #18)")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
p_d = sub.add_parser("dispatch", help="publish the retained dispatch for a registered job")
|
|
p_d.add_argument("--job", required=True)
|
|
p_d.add_argument("--registry-dir", default=default_registry_dir())
|
|
p_d.add_argument("--session", required=True, help="remote session name (without the remote: prefix)")
|
|
p_d.add_argument("--instructions", default="")
|
|
p_d.add_argument("--allow-unsigned", action="store_true")
|
|
p_r = sub.add_parser("run", help="run a remote session: execute dispatches addressed to it")
|
|
p_r.add_argument("--session", required=True)
|
|
p_r.add_argument("--registry-dir", default=default_registry_dir())
|
|
p_r.add_argument("--exec", dest="exec_cmd", default=None,
|
|
help="shell command run per job (env: JOB_ID, MAM_JOB_BRIEF, MAM_PUBLISH_CMD, ...)")
|
|
p_r.add_argument("--once", action="store_true", help="exit after one job")
|
|
p_r.add_argument("--timeout", type=float, default=None, help="exit after this many seconds")
|
|
p_r.add_argument("--allow-unsigned", action="store_true")
|
|
p_r.add_argument("--python", default=None)
|
|
parser.add_argument("-v", "--verbose", action="store_true")
|
|
args = parser.parse_args(argv)
|
|
mqtt_common.ensure_env_loaded()
|
|
mqtt_common.setup_logging(logging.DEBUG if args.verbose else logging.INFO)
|
|
if args.command == "dispatch":
|
|
try:
|
|
print(publish_dispatch(args.job, args.registry_dir, args.session, args.instructions,
|
|
allow_unsigned=args.allow_unsigned))
|
|
except (FileNotFoundError, RuntimeError) as exc:
|
|
logger.error("%s", exc)
|
|
return 1
|
|
except Exception as exc:
|
|
logger.error("dispatch publish failed: %s", exc)
|
|
return 2
|
|
return 0
|
|
return run_runner(args.session, args.registry_dir, args.exec_cmd, args.once,
|
|
allow_unsigned=args.allow_unsigned, timeout=args.timeout, python=args.python)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|