tmpl/multi-agent-mux
1
0 Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d400add8413f77b28362427a5e31c05c1b107d30
multi-agent-mux/.agents/skills/lib_py/delegation/registry.py
T
GodopuandClaude Opus 5.5 b49314c3d0 fix(reset): never take another live reset's fence, and fence absent row registries (#78) 
Review round 4 (codex):
- R1: fence_registries reads an existing marker under the registry lock and
  refuses the reset, changing nothing, when another live (or undecidable)
  reset holds it - registries may be shared between workspaces; only a stale
  marker is replaced, and the reset's exit removes only its own fences.
- R2: the registries to fence come from update.jobs_dirs, the enumeration
  open_jobs now uses too, whether or not they exist; an absent row registry
  is created and fenced (a registration would create it), and one that cannot
  be created refuses the reset.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-30 08:21:40 +09:00

817 lines
35 KiB
Python
Executable File
Raw Blame History

"""Job registry for the multi-agent-mux-delegate-job skill.
A job record is the single source of truth for one delegated unit of work:
its id, prompt, owning agent session, broker connection, timeouts, and status.
Records live as ``<registry_dir>/<job_id>.json`` with an append-only event log
``<registry_dir>/<job_id>.events.log`` and a shared ``<registry_dir>/.lock``.
Concurrency is handled via the fcntl lock in :mod:`mqtt_common` (PoC). For
multi-host delegation, migrate to SQLite WAL — see references/registry.md.
Importable as a library and runnable as a CLI (``register``/``list``/``get``/
``status``/``pick``) so the ``multi-agent-mux-delegate-job`` bash wrapper can shell out.
"""
from __future__ import annotations
import argparse
import json
import logging
import os
import subprocess
import sys
import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from lib_py.delegation import mqtt_common
from lib_py.delegation.mqtt_common import (
DEFAULT_REGISTRY_DIR,
SCHEMA_VERSION,
_atomic_write_record,
_utcnow,
broker_config_from_env,
default_registry_dir,
load_job,
registry_lock,
topic_prefix_for,
)
logger = logging.getLogger("delegate_job.registry")
VALID_STATUSES = ("pending", "running", "completed", "error", "cancelled")
def generate_job_id(bits: int = 32) -> str:
"""PoC: 32-bit hex (8 chars). Production: 128-bit (full uuid4 hex)."""
if bits >= 128:
return uuid.uuid4().hex
nibbles = max(1, bits // 4)
return uuid.uuid4().hex[:nibbles]
# Issue #18: the liveness contract's defaults. An agent heartbeats (publishes
# `progress`) at least every N seconds; the orchestrator treats K missed
# heartbeats (K x N seconds since the last event) as unresponsive; `started`
# must arrive within T seconds of the prompt injection. K x N with these
# defaults equals the historical 120 s idle timeout, so a caller that never
# asked for the contract sees the same idle window as before. T is generous:
# a real agent reads the brief and reasons before it runs the start command
# (grok took ~70 s in the v4.5.0 review round), and T only bounds delivery.
DEFAULT_HEARTBEAT_SEC = 30
DEFAULT_HEARTBEAT_MISSES = 4
DEFAULT_ACK_TIMEOUT_SEC = 180
def liveness_window(heartbeat_sec: int, heartbeat_misses: int) -> int:
"""K x N: seconds without any event after which a job is unresponsive."""
return max(1, int(heartbeat_sec)) * max(1, int(heartbeat_misses))
# Issue #78: a reset of a workspace purges its agents' conversations, so no job
# may be registered for them while it runs. The reset writes RESET_MARKER into
# every registry its open-job check reads (its own .mam/jobs and MAM_JOBS_DIR,
# and a row's other control_plane.mam_jobs_dir), each under that registry's
# lock and before the check: a registration reads it under the same lock, so it
# is either written before the check or refused - by whatever path it names
# the directory. The marker carries the reset's pid/lstart (loop_lock.sh's
# format), its workspace, and for another control plane's registry the
# sessions it covers. The reset's own create (the onboarding job of an agent
# it recreates) passes the reset's pid in MAM_RESET_OWNER_PID.
RESET_MARKER = ".reset-in-progress"
RESET_OWNER_ENV = "MAM_RESET_OWNER_PID"
class ResetInProgressError(RuntimeError):
pass
def _lstart(pid: str) -> str:
try:
out = subprocess.run(["ps", "-p", pid, "-o", "lstart="], capture_output=True, text=True,
timeout=10).stdout
except (OSError, subprocess.SubprocessError):
return ""
return " ".join(out.split())
def reset_marker(registry_dir: str) -> Tuple[str, Dict[str, str]]:
"""(holder state, fields) of the registry's RESET_MARKER: "absent", or as
loop_lock.sh decides - "stale" (dead or reused pid), "live", "unknown" (ps
fails on this process; a marker that cannot be read is unknown too)."""
marker = os.path.join(registry_dir, RESET_MARKER)
try:
with open(marker, encoding="utf-8") as fh:
fields = dict(line.split("=", 1) for line in fh.read().splitlines() if "=" in line)
except FileNotFoundError:
return "absent", {}
except (OSError, ValueError) as exc:
return "unknown", {"error": str(exc)}
pid = fields.get("pid", "")
if not pid.isdigit():
return "stale", fields
if not _lstart(str(os.getpid())):
return "unknown", fields
cur = _lstart(pid)
if not cur or (fields.get("lstart") and cur != fields["lstart"]):
return "stale", fields
return "live", fields
def reset_refusal(registry_dir: str, agent_session: str) -> Optional[str]:
"""Why registering a job for ``agent_session`` in ``registry_dir`` is
refused, or None: a live (or undecidable) reset fenced this registry, for
that session, and the caller is not its own create."""
state, fields = reset_marker(registry_dir)
if state in ("absent", "stale"):
return None
if "sessions" in fields and agent_session not in fields["sessions"].split(","):
return None
if fields.get("pid") and os.environ.get(RESET_OWNER_ENV) == fields["pid"]:
return None
if state == "unknown":
return "cannot tell whether the reset fencing %s is live (%s)" % (
os.path.join(registry_dir, RESET_MARKER), fields.get("error") or "ps unavailable?")
return ("a reset of %s is in progress (pid %s); its agents are being purged and recreated;"
" register the job when it has finished" % (fields.get("workspace", "?"), fields["pid"]))
def register_job(
prompt: str,
agent: str = "claude-code",
agent_session: str = "herdr:claude",
role: str = "Worker",
broker: Optional[Dict[str, Any]] = None,
timeout_sec: int = 3600,
idle_timeout_sec: Optional[int] = None,
registry_dir: str = DEFAULT_REGISTRY_DIR,
job_id: Optional[str] = None,
expected_artifacts: Optional[List[str]] = None,
bits: int = 32,
auth_token: Optional[str] = None,
job_type: str = "direct",
reviewer: Optional[str] = None,
reviewer_session: Optional[str] = None,
max_iterations: int = 5,
heartbeat_sec: int = DEFAULT_HEARTBEAT_SEC,
heartbeat_misses: int = DEFAULT_HEARTBEAT_MISSES,
ack_timeout_sec: int = DEFAULT_ACK_TIMEOUT_SEC,
dispatch_token: Optional[str] = None,
attempt_id: int = 1,
transport: str = "herdr",
supersede_older_attempt: bool = False,
parent_job: Optional[str] = None,
dry_run: bool = False,
) -> str:
"""Create a new ``pending`` job record and return its id.
``broker`` defaults to the current environment's resolved broker block, so
the registry alone is enough for ``publish_event.py`` to connect later.
Issue #18: ``idle_timeout_sec`` left as None derives from the liveness
contract (``heartbeat_misses x heartbeat_sec``); an explicit value is kept
as given. ``dispatch_token`` / ``attempt_id`` are minted here unless a
remote runner mirrors a job the orchestrator already registered, in which
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.
Issue #50 (review 1e25cf2f F1): ``dry_run`` marks the record of a
``submit --dry-run``, which is never dispatched, so the registry admission
does not count it as an open job of its session.
"""
job_id = job_id or generate_job_id(bits)
heartbeat_sec = max(1, int(heartbeat_sec))
heartbeat_misses = max(1, int(heartbeat_misses))
if idle_timeout_sec is None:
idle_timeout_sec = liveness_window(heartbeat_sec, heartbeat_misses)
if broker is None:
broker = broker_config_from_env().to_registry_block()
if auth_token is None:
# Auto-generate token if secure broker configuration (TLS or username) is detected
if broker.get("tls") or broker.get("username"):
import secrets
auth_token = secrets.token_urlsafe(32)
now = _utcnow()
record: Dict[str, Any] = {
"schema_version": SCHEMA_VERSION,
"job_id": job_id,
"status": "pending",
"created_at": now,
"updated_at": now,
"prompt": prompt,
"agent": agent,
"agent_session": agent_session,
"role": role,
"broker": broker,
"topic_prefix": topic_prefix_for(job_id),
"timeout_sec": int(timeout_sec),
"idle_timeout_sec": int(idle_timeout_sec),
"expected_artifacts": expected_artifacts or [],
"last_seq": 0,
"auth_token": auth_token,
"job_type": job_type,
"reviewer": reviewer,
"reviewer_session": reviewer_session,
"max_iterations": int(max_iterations),
"iteration": 1,
"dispatch_token": dispatch_token or uuid.uuid4().hex,
"attempt_id": int(attempt_id),
# Issue #18 liveness contract
"heartbeat_sec": heartbeat_sec,
"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,
}
if dry_run:
record["dry_run"] = True
with registry_lock(registry_dir):
refusal = reset_refusal(registry_dir, agent_session)
if refusal:
raise ResetInProgressError(refusal)
if mqtt_common._job_path(job_id, registry_dir).exists():
if not supersede_older_attempt:
raise FileExistsError(f"job already exists: {job_id}")
# A remote runner adopting the orchestrator's next attempt (review
# round 2, codex R1). Decided under this lock, so a stale or
# conflicting dispatch can never roll the record back.
old = mqtt_common.load_job(job_id, registry_dir)
old_attempt = int(old.get("attempt_id", old.get("iteration", 1)) or 1)
if int(attempt_id) <= old_attempt:
raise FileExistsError(
f"job {job_id} is already at attempt {old_attempt}; refusing attempt {attempt_id}")
record["last_seq"] = int(old.get("last_seq", 0) or 0) # seqs stay monotonic across attempts
_atomic_write_record(job_id, registry_dir, record)
# Seed the persistent audit log (meta.json + status.json + a "registered"
# event). Best-effort inside init_job_log — never blocks registration.
mqtt_common.init_job_log(job_id, meta=record, logs_dir=mqtt_common.get_logs_dir(registry_dir))
logger.info("registered job %s (agent=%s session=%s)", job_id, agent, agent_session)
return job_id
def pick_pending(agent_session: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> Optional[str]:
"""Claim the oldest ``pending`` job for ``agent_session``, flipping it to
``running`` atomically under the lock. Returns the job id, or None if no
pending job matches. This is how each herdr session takes only its own work
without two sessions grabbing the same job."""
from lib_py.delegation import engine # same package: no sibling hop (see __init__)
if not engine.cas_enabled():
return _pick_pending_legacy(agent_session, registry_dir)
# Selection still happens under the lock, then the claim itself is committed
# by the single transition boundary with an explicit precondition, so two
# sessions still cannot both win.
with registry_lock(registry_dir):
candidates = [
record for record in _iter_records(registry_dir)
if record.get("status") == "pending" and record.get("agent_session") == agent_session
]
if not candidates:
return None
candidates.sort(key=lambda r: r.get("created_at", ""))
chosen_id = candidates[0]["job_id"]
result = engine.transition(chosen_id, "claim", registry_dir=registry_dir,
expected_status="pending", source=agent_session)
if not result.committed:
# Someone else claimed it between selection and commit; the precondition
# refused, which is exactly the behaviour the lock used to provide.
logger.info("job %s was claimed by another session first", chosen_id)
return None
logger.info("session %s picked job %s", agent_session, chosen_id)
return chosen_id
def _pick_pending_legacy(agent_session: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> Optional[str]:
"""Pre-phase-2a body, preserved for MAM_ENGINE_CAS=0 rollback."""
with registry_lock(registry_dir):
candidates = []
for record in _iter_records(registry_dir):
if record.get("status") == "pending" and record.get("agent_session") == agent_session:
candidates.append(record)
if not candidates:
return None
candidates.sort(key=lambda r: r.get("created_at", ""))
chosen = candidates[0]
chosen["status"] = "running"
chosen["updated_at"] = _utcnow()
_atomic_write_record(chosen["job_id"], registry_dir, chosen)
logger.info("session %s picked job %s", agent_session, chosen["job_id"])
job_id = chosen["job_id"]
updated_at = chosen["updated_at"]
# Mirrors the pending->running transition into the audit log outside the
# lock. This is the shape the engine now generalises to every transition.
logs_dir = mqtt_common.get_logs_dir(registry_dir)
mqtt_common.update_logged_status(job_id, "running", logs_dir=logs_dir, updated_at=updated_at)
mqtt_common.append_event(job_id, {
"event": "status_changed",
"from": "pending",
"to": "running",
"by": agent_session,
"timestamp": updated_at,
}, logs_dir=logs_dir)
return job_id
def update_status(job_id: str, registry_dir: str, status: str) -> Dict[str, Any]:
if status not in VALID_STATUSES:
raise ValueError(f"invalid status {status!r}; expected one of {VALID_STATUSES}")
return mqtt_common.update_job_status(job_id, registry_dir, status=status)
def list_jobs(registry_dir: str = DEFAULT_REGISTRY_DIR, status: Optional[str] = None) -> List[Dict[str, Any]]:
records = list(_iter_records(registry_dir))
if status:
records = [r for r in records if r.get("status") == status]
records.sort(key=lambda r: r.get("created_at", ""))
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."""
try:
Path(registry_dir).mkdir(parents=True, exist_ok=True)
log_path = Path(registry_dir) / f"{job_id}.events.log"
with open(log_path, "a", encoding="utf-8") as fh:
fh.write(json.dumps(payload, ensure_ascii=False) + "\n")
except OSError as exc: # pragma: no cover - best effort
logger.warning("could not append event for %s: %s", job_id, exc)
# convenience re-export so callers can `from registry import load_job`
__all__ = [
"register_job", "pick_pending", "update_status", "load_job",
"list_jobs", "append_event", "generate_job_id", "get_feedback",
"children_of", "descendants_of",
]
def _iter_records(registry_dir: str):
base = Path(registry_dir)
if not base.exists():
return
for path in sorted(base.glob("*.json")):
try:
with open(path, "r", encoding="utf-8") as fh:
yield json.load(fh)
except (OSError, json.JSONDecodeError) as exc:
logger.warning("skipping unreadable record %s: %s", path, exc)
def get_feedback(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR,
events: Tuple[str, ...] = ("completed", "error")) -> str:
"""Read the job's audit log or events log and return the detail of the last
event whose kind is in ``events`` (completed/error by default; Issue #34's
sub-job table adds ``cancelled`` to see why a job was never dispatched)."""
# 1) Try the unified audit log first (ndjson) since it's written synchronously by the subscriber
try:
from lib_py.delegation import mqtt_common
logs_dir = mqtt_common.get_logs_dir(registry_dir)
logged = list(mqtt_common.iter_logged_events(job_id, logs_dir))
for e in reversed(logged):
if e.get("source_event") in events:
return e.get("detail", "")
if e.get("event") in events:
return e.get("detail", "")
except Exception:
pass
# 2) Fallback to local .events.log
log_path = Path(registry_dir) / f"{job_id}.events.log"
if log_path.exists():
feedback = ""
try:
with open(log_path, "r", encoding="utf-8") as fh:
for line in fh:
if not line.strip():
continue
try:
payload = json.loads(line)
if payload.get("event") in events:
feedback = payload.get("detail", "")
except json.JSONDecodeError:
continue
except OSError:
pass
if feedback:
return feedback
return ""
# --------------------------------------------------------------------------
# CLI (so the bash wrapper can shell out without inline python)
# --------------------------------------------------------------------------
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="multi-agent-mux-delegate-job registry CLI")
parser.add_argument("--registry-dir", default=default_registry_dir())
sub = parser.add_subparsers(dest="command", required=True)
p_reg = sub.add_parser("register", help="create a pending job; prints the job id")
# Issue #36: a long prompt goes by file - one argv string is capped at
# MAX_ARG_STRLEN (131,072 bytes on Linux).
reg_prompt = p_reg.add_mutually_exclusive_group(required=True)
reg_prompt.add_argument("--prompt")
reg_prompt.add_argument("--prompt-file", help="read the prompt from this file (utf-8)")
p_reg.add_argument("--agent", default="claude-code")
p_reg.add_argument("--agent-session", default="herdr:claude")
p_reg.add_argument("--role", default="Worker", help="logical role for the delegated agent (e.g. Worker, Planner, Reviewer)")
p_reg.add_argument("--timeout", type=int, default=3600)
p_reg.add_argument("--idle-timeout", type=int, default=None,
help="seconds without an event before the job is unresponsive "
"(default: --heartbeat-misses x --heartbeat-sec)")
p_reg.add_argument("--heartbeat-sec", type=int, default=DEFAULT_HEARTBEAT_SEC,
help="agent heartbeat period N in seconds (Issue #18)")
p_reg.add_argument("--heartbeat-misses", type=int, default=DEFAULT_HEARTBEAT_MISSES,
help="missed heartbeats K before the job counts as unresponsive")
p_reg.add_argument("--ack-timeout", type=int, default=DEFAULT_ACK_TIMEOUT_SEC,
help="seconds T after injection within which `started` must arrive")
p_reg.add_argument("--transport", default="herdr", choices=["herdr", "remote"],
help="how the prompt reaches the agent")
p_reg.add_argument("--dispatch-token", default=None, help=argparse.SUPPRESS)
p_reg.add_argument("--attempt-id", type=int, default=1, help=argparse.SUPPRESS)
p_reg.add_argument("--job-id", default=None, help=argparse.SUPPRESS)
p_reg.add_argument("--bits", type=int, default=32, help="32 (PoC) or 128 (prod)")
p_reg.add_argument("--artifact", action="append", default=[], dest="artifacts")
p_reg.add_argument("--auth-token", default=None, help="HMAC auth token for the job (auto-generated if secure broker is detected)")
p_reg.add_argument("--job-type", default="direct", choices=["direct", "loop", "discuss"])
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("--dry-run", action="store_true",
help="the record of a submit --dry-run: never dispatched, not an open job for admission")
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")
p_get.add_argument("--job", required=True)
p_status = sub.add_parser("status", help="set a job status")
p_status.add_argument("--job", required=True)
p_status.add_argument("--set", required=True, dest="status")
p_update = sub.add_parser("update", help="update a job record")
p_update.add_argument("--job", required=True)
p_update.add_argument("--status", default=None)
p_update.add_argument("--agent-session", default=None)
upd_prompt = p_update.add_mutually_exclusive_group()
upd_prompt.add_argument("--prompt", default=None)
upd_prompt.add_argument("--prompt-file", default=None, help="read the prompt from this file (utf-8)")
p_update.add_argument("--iteration", type=int, default=None)
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, wall_clock_timeout or cancelled")
p_close.add_argument("--job", required=True)
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)
p_feedback = sub.add_parser("get-feedback", help="get the last feedback detail (completed/error) for a job")
p_feedback.add_argument("--job", required=True)
p_pick = sub.add_parser("pick", help="claim a pending job for a session; prints id")
p_pick.add_argument("--agent-session", default="herdr:claude")
p_logs = sub.add_parser(
"logs",
help="show the persistent audit log for a job, or --list every logged job",
)
p_logs.add_argument("job_id", nargs="?", default=None,
help="job id whose events.ndjson to print")
p_logs.add_argument("--list", action="store_true", dest="list_all",
help="summarise every job under the logs dir instead")
p_logs.add_argument("--logs-dir", default=None,
help="override the audit-log root (default: $DELEGATE_JOB_LOGS_DIR "
"or the delegate_job_logs next to --registry-dir)")
p_logs.add_argument("--tail", type=int, default=0,
help="show only the last N events (0 = all)")
p_logs.add_argument("--json", action="store_true",
help="emit raw JSON lines / records instead of a table")
p_scrub = sub.add_parser(
"scrub-secrets",
help="remove broker.password from every job record (Issue #12 migration)",
)
p_scrub.add_argument("--dry-run", action="store_true",
help="only list the records that still carry a password")
return parser
def _cmd_scrub_secrets(args) -> int:
"""Issue #12: strip broker.password from records written before
to_registry_block stopped persisting it. Every other field is kept.
Only a stored value counts. The old rule wrote `"password": null` even
for an unauthenticated broker; that empty key is no secret, and
_atomic_write_record drops it on the record's next write anyway, so
reporting or rewriting it here would list phantom secrets and hold the
registry lock over every record in a PoC registry."""
rd = args.registry_dir
found = []
with registry_lock(rd):
for record in list(_iter_records(rd)):
broker = record.get("broker")
if not (isinstance(broker, dict) and broker.get("password")):
continue
found.append(record.get("job_id", "?"))
if not args.dry_run:
broker.pop("password", None)
_atomic_write_record(record["job_id"], rd, record)
verb = "would scrub" if args.dry_run else "scrubbed"
print(f"{verb} {len(found)} record(s) under {rd}" + (": " + ", ".join(found) if found else ""))
return 0
def _operator_write(job_id: str, registry_dir: str, **fields) -> int:
"""Apply an operator-authority write and report a refusal on the exit status.
Before phase 2a this path could not be refused, so it always exited 0. Once
the CAS boundary can decline a write, exiting 0 anyway would hide it -- which
is how a refused loop re-dispatch looked like success to the wrapper.
Two things this must NOT do, both learned the hard way:
* treat an idempotent ACK as a refusal. The wrapper finalises an approved
job with `status --set completed` on a record that is already completed,
and it runs under `set -e`, so reporting failure there aborts the success
path before the validation hook and the final stdout line.
* skip the status vocabulary check. That validation used to live in
`update_status`; routing around it let an invalid status through, and in
the rollback mode it was actually written to the record.
"""
from lib_py.delegation import engine # same package: no sibling hop (see __init__)
status = fields.get("status")
if status is not None and status not in VALID_STATUSES:
print("invalid status {!r}; expected one of {}".format(status, VALID_STATUSES),
file=sys.stderr)
return 1
if not engine.cas_enabled():
mqtt_common.update_job_status(job_id, registry_dir, **fields)
return 0
result = mqtt_common.transition_job_status(
job_id, registry_dir, _source="registry_cli", **fields)
if result.committed:
return 0
if result.idempotent:
# The requested state already holds. Nothing to do, and nothing wrong.
return 0
print("refused: {}".format(result.reason), file=sys.stderr)
return 4
def _prompt_from_args(parser: argparse.ArgumentParser, args: argparse.Namespace) -> Optional[str]:
"""The prompt text: from --prompt-file (read verbatim) or from --prompt."""
if getattr(args, "prompt_file", None) is None:
return args.prompt
try:
return Path(args.prompt_file).read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError) as exc:
parser.error(f"cannot read --prompt-file {args.prompt_file}: {exc}")
def main(argv: Optional[List[str]] = None) -> int:
if not mqtt_common.load_env_or_report():
return 1
mqtt_common.setup_logging(logging.INFO)
parser = _build_parser()
args = parser.parse_args(argv)
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
try:
job_id = register_job(
prompt=_prompt_from_args(parser, args),
agent=args.agent,
agent_session=args.agent_session,
role=args.role,
timeout_sec=args.timeout,
idle_timeout_sec=args.idle_timeout,
registry_dir=rd,
expected_artifacts=args.artifacts,
bits=args.bits,
auth_token=args.auth_token,
job_type=args.job_type,
reviewer=args.reviewer,
reviewer_session=args.reviewer_session,
max_iterations=args.max_iterations,
heartbeat_sec=args.heartbeat_sec,
heartbeat_misses=args.heartbeat_misses,
ack_timeout_sec=args.ack_timeout,
transport=args.transport,
dispatch_token=args.dispatch_token,
attempt_id=args.attempt_id,
job_id=args.job_id,
parent_job=args.parent_job,
dry_run=args.dry_run,
)
except ResetInProgressError as exc:
print("ERROR: %s" % exc, file=sys.stderr)
return 1
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:
if not records:
print("(no jobs)")
for r in records:
print(f"{r['job_id']} {r.get('status','?'):10s} {r.get('agent_session','')}"
f" {r.get('prompt','')[:48]}")
return 0
if args.command == "get":
try:
print(json.dumps(load_job(args.job, rd), ensure_ascii=False, indent=2))
except FileNotFoundError as exc:
print(str(exc), file=sys.stderr)
return 1
return 0
if args.command == "status":
try:
rc = _operator_write(args.job, rd, status=args.status)
except (FileNotFoundError, ValueError) as exc:
print(str(exc), file=sys.stderr)
return 1
return rc
if args.command == "update":
fields = {}
if args.status is not None:
fields["status"] = args.status
if args.agent_session is not None:
fields["agent_session"] = args.agent_session
prompt = _prompt_from_args(parser, args)
if prompt is not None:
fields["prompt"] = prompt
if args.iteration is not None:
fields["iteration"] = args.iteration
if args.attempt_id is not None:
fields["attempt_id"] = args.attempt_id
if args.role is not None:
fields["role"] = args.role
try:
rc = _operator_write(args.job, rd, **fields)
except FileNotFoundError as exc:
print(str(exc), file=sys.stderr)
return 1
return rc
if args.command == "close":
from lib_py.delegation import engine
fields = {}
if args.detail is not None:
fields["detail"] = args.detail
if not engine.cas_enabled():
req_status = engine.REQUEST_TO_STATUS.get(args.request, "error")
fields["status"] = req_status
try:
mqtt_common.update_job_status(args.job, rd, **fields)
except FileNotFoundError as exc:
print(str(exc), file=sys.stderr)
return 1
return 0
try:
res = engine.transition(
args.job,
args.request,
registry_dir=rd,
source="delegate_wrapper",
fields=fields,
expected_dispatch_token=args.expect_dispatch_token,
)
except FileNotFoundError as exc:
print(str(exc), file=sys.stderr)
return 1
if res.committed:
return 0
if res.idempotent:
if res.reason:
print(res.reason, file=sys.stderr)
return 0
if res.reason and "conflicting" in res.reason and "after terminal" in res.reason:
print(res.reason, file=sys.stderr)
return 0
print(f"refused: {res.reason}", file=sys.stderr)
return 4
if args.command == "get-feedback":
print(get_feedback(args.job, rd))
return 0
if args.command == "pick":
job_id = pick_pending(args.agent_session, rd)
if job_id is None:
return 3 # no pending job for this session
print(job_id)
return 0
if args.command == "logs":
return _cmd_logs(args)
if args.command == "scrub-secrets":
return _cmd_scrub_secrets(args)
return 1
def _cmd_logs(args) -> int:
"""Pretty-print one job's events.ndjson, or summarise all logged jobs."""
logs_dir = args.logs_dir or mqtt_common.get_logs_dir(args.registry_dir)
if args.list_all:
jobs = mqtt_common.list_logged_jobs(logs_dir)
if args.json:
print(json.dumps(jobs, ensure_ascii=False, indent=2))
return 0
if not jobs:
print(f"(no logged jobs under {logs_dir})")
return 0
for m in jobs:
print(f"{m.get('job_id','?')} {m.get('status','?'):10s} "
f"{m.get('created_at','-'):20s} {(m.get('prompt') or '')[:48]}")
return 0
if not args.job_id:
print("logs requires a <job_id> or --list", file=sys.stderr)
return 1
events = list(mqtt_common.iter_logged_events(args.job_id, logs_dir))
if not events and not mqtt_common.job_log_dir(args.job_id, logs_dir).exists():
print(f"no audit log for job {args.job_id} under {logs_dir}", file=sys.stderr)
return 1
if args.tail and args.tail > 0:
events = events[-args.tail:]
if args.json:
for e in events:
print(json.dumps(e, ensure_ascii=False))
return 0
for e in events:
ts = e.get("logged_at") or e.get("timestamp") or "-"
extra = e.get("detail") or e.get("to") or e.get("source_event") or ""
print(f"{ts:24s} {e.get('event','?'):<16s} {extra}")
return 0
if __name__ == "__main__":
sys.exit(main())
Reference in new issue View Git Blame Copy Permalink