refactor(issue-5): move delegation scripts into lib_py.delegation with package-bound shims

The four delegation scripts become a real package. Their old paths stay as
shims because those paths are a wire contract: agent instructions, the bash
wrapper, SKILL.md, BOOTSTRAP.md, deploy/install.sh and three test modules all
call them literally.

The shim binds the lib_py.delegation *package* to the tree beside it, once per
process, and never touches sys.path. Two cheaper designs were rejected with
evidence (the context matrix now lives in the test suite):

  * inserting the sibling skills dir at sys.path[0] makes the first sandbox
    copy imported under pytest shadow lib_py.agents/layout/state for the whole
    session, and the identity assertion cannot see it -- `is` proves same
    object, not same tree.
  * loading only the named module by file path leaves a module that imports
    siblings pulling them from whichever tree is on the path, so a frozen
    publish_event ran against live dependencies and lost half of B-13.

Binding the package fixes both: one tree for all delegation modules, the parent
lib_py untouched, and a sys.path append only when no lib_py exists anywhere
(true standalone). Bind-once rather than nearest-wins, because rebinding would
split existing references and break monkeypatching.

Also moves the env-file load off import (_load_dotenv -> ensure_env_loaded,
called by each process entry point including reconcile.sh's embedded
subscriber), so importing the library no longer mutates os.environ.

Verified: PV-1 identity+provenance, PV-4's ten import scenarios plus K1/K3/K6
in fresh interpreters, PV-5 shim byte-identity, an AST check that the package
imports no sibling lib_py module, and --help output diffed against the
pre-move capture.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd
This commit is contained in:
GodopuandClaude Opus 5 committed 2026-09-06 15:41:43 +09:00
1 parent 265a1f6c3d
commit fb3498e23b
12 files changed
+2187 -1656

No files matched your search

@@ -0,0 +1,18 @@
"""Delegation backplane: job registry, MQTT publish/subscribe, and the
subscriber daemon.
The implementation moved here from
``multi-agent-mux-delegate-job/scripts/*.py`` in issue #5 phase 1. Those script
paths are a wire contract (agent instructions, the bash wrapper, SKILL.md,
BOOTSTRAP.md, deploy/install.sh and three test modules call them by literal
path), so they remain as shims that bind *this package* to the tree next to
themselves, once per process, without touching ``sys.path``.
Package rules, enforced by tests:
* modules here import each other as ``from lib_py.delegation import x`` --
never by top-level name, which would collide with ``lib_py.agents.registry``.
* nothing here imports another ``lib_py`` module. Under a frozen loop the
parent ``lib_py`` may resolve to a different tree than this package, so a
sibling import would silently mix trees.
"""
+340
View File
@@ -0,0 +1,340 @@
#!/usr/bin/env python3
"""job_subscriber.py — the single entry point for observing Job events.
Subscribes to one job's ``<topic_prefix>/events`` (or, with ``--wait-any``, the
events of every running/pending job in the registry), prints one line to stdout
per accepted event, and exits on a terminal event or a timeout.
Design points (all flagged in the PLAN review):
- terminal state machine: ``completed``/``error`` is acted on exactly once per
job, so QoS-1 duplicates or an ``error``-after-``completed`` reorder are safe.
- dual timeouts: a wall-clock ``--timeout`` (total budget, started at
subscribe time so a cold start can't hang forever) AND an idle
``--idle-timeout`` (no new event for N seconds).
- defensive parsing: undecodable payloads, ``schema_version`` mismatches, and
``job_id`` values we did not subscribe for are logged and dropped.
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
"""
from __future__ import annotations
import argparse
import json
import logging
import queue
import sys
import time
from typing import Any, Dict, List, Optional, Set, Tuple
from lib_py.delegation import mqtt_common
from lib_py.delegation import registry
from lib_py.delegation.mqtt_common import (
DEFAULT_REGISTRY_DIR,
SCHEMA_VERSION,
broker_config_from_job,
load_job,
make_client,
)
logger = logging.getLogger("delegate_job.job_subscriber")
TERMINAL_EVENTS = ("completed", "error")
def _format_line(topic: str, payload: Dict[str, Any]) -> str:
source_tag = f" [{payload['source']}]" if payload.get("source") else ""
return (
f"{payload.get('timestamp','-')} "
f"job={payload.get('job_id','?')} "
f"seq={payload.get('seq','?')} "
f"{payload.get('event','?') + source_tag:<20} "
f"{payload.get('detail','')}"
)
def _check_disk_fallback(
pending: Set[str],
registry_dir: str,
terminal: Dict[str, str],
) -> None:
"""Check local on-disk registry and audit logs for terminal status (B-15)."""
for jid in list(pending):
disk_status = None
try:
job_rec = load_job(jid, registry_dir)
disk_status = job_rec.get("status")
except Exception:
pass
if not disk_status or disk_status not in ("completed", "error", "cancelled"):
try:
disk_status = mqtt_common.read_logged_status(jid, mqtt_common.get_logs_dir())
except Exception:
pass
if disk_status in ("completed", "error", "cancelled"):
synth_event = "completed" if disk_status == "completed" else "error"
synth_payload = {
"job_id": jid,
"event": synth_event,
"seq": -1,
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"detail": f"terminal status '{disk_status}' resolved via disk-fallback",
"source": "disk-fallback",
}
synth_topic = f"mam/{jid}/events"
print(_format_line(synth_topic, synth_payload), flush=True)
terminal[jid] = synth_event
pending.discard(jid)
class _Watcher:
"""Holds the shared queue + the set of job_ids we accept events for."""
def __init__(self, expected_job_ids: Set[str], expected_tokens: Dict[str, Optional[str]], expected_seqs: Dict[str, int]):
self.events: "queue.Queue[Tuple[str, Dict[str, Any]]]" = queue.Queue()
self.expected = set(expected_job_ids)
self.tokens = expected_tokens # job_id -> expected auth_token (or None)
self.last_seq = dict(expected_seqs)
def on_message(self, _client, _userdata, msg) -> None:
# --- defensive parsing -------------------------------------------
try:
payload = json.loads(msg.payload.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
logger.warning("drop unparseable payload on %s: %s", msg.topic, exc)
return
if not isinstance(payload, dict):
logger.warning("drop non-object payload on %s", msg.topic)
return
if payload.get("schema_version") != SCHEMA_VERSION:
logger.warning("drop event with schema_version=%r (expected %d)",
payload.get("schema_version"), SCHEMA_VERSION)
return
jid = payload.get("job_id")
if jid not in self.expected:
logger.warning("drop event for unexpected job_id=%r on %s", jid, msg.topic)
return
# --- production auth check: data.auth_token must match if expected ---
expected_token = self.tokens.get(jid)
if not mqtt_common.verify_hmac(payload, expected_token):
logger.warning("drop event for job %s: HMAC verify failed", jid)
return
# --- replay attack defense: check monotonic sequence ---
seq = payload.get("seq")
if seq is None or not isinstance(seq, int):
logger.warning("drop event for job %s: missing or invalid seq", jid)
return
if seq <= self.last_seq.get(jid, 0):
logger.warning("drop event for job %s: seq %d is not monotonically increasing (last %d)",
jid, seq, self.last_seq.get(jid, 0))
return
self.last_seq[jid] = seq
# Persistent audit log from the *subscriber's* vantage point: every event
# that survives defensive parsing is recorded here, including ones a
# different host published. This is the external-observer record that
# backstops the publisher's own "published" line if it never wrote one.
mqtt_common.append_event(jid, {
"event": "received",
"source_event": payload.get("event"),
"seq": payload.get("seq"),
"topic": msg.topic,
"timestamp": payload.get("timestamp"),
"detail": payload.get("detail", ""),
})
self.events.put((msg.topic, payload))
def _collect_jobs(args) -> List[Dict[str, Any]]:
"""Resolve the list of job records this invocation should watch."""
if args.wait_any:
jobs = [r for r in registry.list_jobs(args.registry_dir)
if r.get("status") in ("pending", "running")]
if not jobs:
logger.error("no pending/running jobs to wait for")
return jobs
job = load_job(args.job, args.registry_dir) # raises FileNotFoundError
return [job]
def _run_subscriber(args) -> int:
try:
jobs = _collect_jobs(args)
except FileNotFoundError as exc:
logger.error("%s", exc)
return 2
if not jobs:
return 2
expected_ids: Set[str] = {j["job_id"] for j in jobs}
tokens = {j["job_id"]: j.get("auth_token") for j in jobs}
seqs = {}
for j in jobs:
jid = j["job_id"]
last_seq = int(j.get("last_seq", 0))
seqs[jid] = max(0, last_seq - 1)
watcher = _Watcher(expected_ids, tokens, seqs)
# Resolve timeouts from CLI, falling back to the (first) job's settings.
base_job = jobs[0]
wall_timeout = args.timeout if args.timeout is not None else float(base_job.get("timeout_sec", 3600))
idle_timeout = args.idle_timeout if args.idle_timeout is not None else float(base_job.get("idle_timeout_sec", 120))
# All watched jobs share a broker in practice; connect using the first
# job's broker and subscribe to each job's events topic.
config = broker_config_from_job(base_job)
client = make_client("subscriber", config)
client.on_message = watcher.on_message
subscribed_topics = []
for job in jobs:
prefix = job.get("topic_prefix") or mqtt_common.topic_prefix_for(job["job_id"])
subscribed_topics.append(f"{prefix}/events")
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
for topic in subscribed_topics:
_c.subscribe(topic, qos=1)
logger.info("subscribed to %s", topic)
def on_disconnect(_c, _u, _flags, reason_code, _props):
rc = mqtt_common.reason_code_value(reason_code)
if rc != 0:
logger.warning("broker disconnected (rc=%s); will retry reconnect", reason_code)
def on_subscribe(_c, _u, mid, granted_qos, _props=None):
for topic in subscribed_topics:
print(f"SUBSCRIBED {topic}", flush=True)
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.on_subscribe = on_subscribe
client.reconnect_delay_set(min_delay=1, max_delay=16)
terminal: Dict[str, str] = {} # job_id -> "completed"/"error"
pending: Set[str] = set(expected_ids)
start = time.monotonic()
wall_deadline = start + wall_timeout
last_event = start
last_disk_check = 0.0
retention_checked = not args.expect_retention
connected = False
# Check if all jobs are already in terminal state on disk before connecting
_check_disk_fallback(pending, args.registry_dir, terminal)
if not pending:
if any(state == "error" for state in terminal.values()):
return 1
return 0
try:
mqtt_common.with_retry(
lambda: client.connect(config.host, config.port, config.keepalive),
attempts=3, base_delay=0.2, max_delay=1.0
)()
client.loop_start()
connected = True
except Exception as exc:
logger.warning("broker connection failed (%s); checking disk fallback", exc)
_check_disk_fallback(pending, args.registry_dir, terminal)
if not pending:
if any(state == "error" for state in terminal.values()):
return 1
return 0
logger.error("broker connection failed: %s", exc)
return 3
try:
while pending:
now = time.monotonic()
if now >= wall_deadline:
_check_disk_fallback(pending, args.registry_dir, terminal)
if not pending:
break
logger.error("wall-clock timeout (%.0fs); still pending: %s",
wall_timeout, ", ".join(sorted(pending)))
return 2
idle_left = idle_timeout - (now - last_event)
if idle_left <= 0:
_check_disk_fallback(pending, args.registry_dir, terminal)
if not pending:
break
logger.error("idle timeout (%.0fs, no events); still pending: %s",
idle_timeout, ", ".join(sorted(pending)))
return 2
wait = min(wall_deadline - now, idle_left, 1.0)
try:
topic, payload = watcher.events.get(timeout=wait)
except queue.Empty:
if not retention_checked and (now - start) > 3.0:
logger.warning("--expect-retention set but no retained "
"terminal event observed yet")
retention_checked = True
# Step 2 (B-15): Local disk fallback throttled to every 3.0s
if (now - last_disk_check) >= 3.0:
last_disk_check = now
_check_disk_fallback(pending, args.registry_dir, terminal)
continue
last_event = time.monotonic()
retention_checked = True
print(_format_line(topic, payload), flush=True)
jid = payload["job_id"]
event = payload.get("event")
if event in TERMINAL_EVENTS:
if jid in terminal:
# Already finalised: ignore duplicates / late reorders.
logger.info("ignoring duplicate terminal %s for %s", event, jid)
continue
terminal[jid] = event
pending.discard(jid)
finally:
if connected:
client.loop_stop()
try:
client.disconnect()
except Exception: # pragma: no cover
pass
# All jobs reached a terminal state. error wins over completed.
if any(state == "error" for state in terminal.values()):
return 1
return 0
def main(argv=None) -> int:
parser = argparse.ArgumentParser(description="Subscribe to Job events on MQTT")
target = parser.add_mutually_exclusive_group(required=True)
target.add_argument("--job", help="job id to watch")
target.add_argument("--wait-any", action="store_true",
help="watch every pending/running job in the registry")
parser.add_argument("--timeout", type=float, default=None,
help="wall-clock budget in seconds (default: job.timeout_sec or 3600)")
parser.add_argument("--idle-timeout", type=float, default=None,
help="max seconds with no new event (default: job.idle_timeout_sec or 120)")
parser.add_argument("--expect-retention", action="store_true",
help="warn if no retained terminal event arrives promptly")
parser.add_argument("--registry-dir", default=DEFAULT_REGISTRY_DIR)
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.WARNING)
try:
return _run_subscriber(args)
except Exception as exc:
logger.error("subscriber fatal error: %s", exc)
return 3
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,688 @@
"""Shared MQTT + registry helpers for the multi-agent-mux-delegate-job skill.
Single entry point for:
- broker configuration (env -> dataclass),
- paho client construction (auth + TLS + unique client id),
- monotonic per-job sequence counters,
- retry-with-exponential-backoff,
- atomic registry record load/update under an fcntl lock.
Requires paho-mqtt >= 2.0 (uses CallbackAPIVersion.VERSION2).
This module is the *only* place that talks to the broker config and to the
raw job record file, so PoC -> production migration touches just env/registry
values, never code (see references/mqtt-broker-setup.md).
"""
from __future__ import annotations
import functools
import hashlib
import hmac
import json
import logging
import os
import tempfile
import time
import uuid
from contextlib import contextmanager
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any, Callable, Dict, Iterable, List, Optional
import paho.mqtt.client as mqtt
logger = logging.getLogger("delegate_job.mqtt_common")
_warned_deprecated_env = False
_warned_coexistence_env = False
def _load_dotenv(workspace_dir: Optional[str] = None) -> None:
"""Load .mam.env (or .env fallback) from workspace if it exists.
Only sets vars that are not already in os.environ
(i.e. OS env takes precedence over env files).
"""
global _warned_deprecated_env, _warned_coexistence_env
# 1. Check explicit MAM_ENV_FILE override
explicit_file = os.environ.get("MAM_ENV_FILE")
if explicit_file:
if os.path.isfile(explicit_file):
_parse_env_file(explicit_file)
return
# 2. Resolve workspace directory with boundary marker check (.agents or .git)
if workspace_dir is None:
curr = os.path.dirname(os.path.abspath(__file__))
resolved_root = None
while curr and curr != os.path.dirname(curr):
if os.path.isdir(os.path.join(curr, ".agents")) or os.path.exists(os.path.join(curr, ".git")):
resolved_root = curr
break
curr = os.path.dirname(curr)
if not resolved_root:
return
d = resolved_root
else:
d = workspace_dir
mam_env_path = os.path.join(d, ".mam.env")
legacy_env_path = os.path.join(d, ".env")
has_mam = os.path.isfile(mam_env_path)
has_legacy = os.path.isfile(legacy_env_path)
if has_mam and has_legacy:
if not _warned_coexistence_env:
logger.warning(
"Both '%s' and '%s' exist. Loading '%s'. "
"Consider removing or migrating '%s'.",
mam_env_path, legacy_env_path, mam_env_path, legacy_env_path
)
_warned_coexistence_env = True
_parse_env_file(mam_env_path)
elif has_mam:
_parse_env_file(mam_env_path)
elif has_legacy:
if not _warned_deprecated_env:
logger.warning(
"Loading deprecated config file '%s'. "
"Please migrate to '.mam.env'.",
legacy_env_path
)
_warned_deprecated_env = True
_parse_env_file(legacy_env_path)
def _parse_env_file(path: str) -> None:
try:
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" in line:
key, _, val = line.partition("=")
key = key.strip()
val = val.strip().strip('"').strip("'")
if key and key not in os.environ:
os.environ[key] = val
except Exception as e:
logger.warning("Failed to parse env file %s: %s", path, e)
_env_loaded = False
def ensure_env_loaded(workspace_dir: Optional[str] = None) -> None:
"""Load the workspace env file exactly once per process (D6, issue #5 phase 1).
This used to run at import time. Importing a library must not mutate
``os.environ``, so every *process entry point* calls this instead:
``publish_event.main``, ``registry.main``, ``job_subscriber.main``,
``cli.main``, ``bench.main``, and reconcile.sh's embedded subscriber.
Production semantics are unchanged because every production path passes
through one of those, and OS env still wins over the file
(see :func:`_parse_env_file`).
"""
global _env_loaded
if _env_loaded:
return
_env_loaded = True
_load_dotenv(workspace_dir)
# --------------------------------------------------------------------------
# Constants
# --------------------------------------------------------------------------
SCHEMA_VERSION = 1
DEFAULT_REGISTRY_DIR = ".mam/jobs"
DEFAULT_TOPIC_ROOT = "python/mqtt/jobs"
LOCK_FILENAME = ".lock"
# Persistent audit-log layout: .mam/delegate_job_logs/<job_id>/{meta,events,status}.
# This is a *separate* artifact from the registry: the registry is the live job
# record (mutated in place), the audit log is an append-only history that
# survives even if the registry dir is cleaned up.
META_FILENAME = "meta.json"
EVENTS_FILENAME = "events.ndjson"
STATUS_FILENAME = "status.json"
def get_logs_dir() -> str:
"""Audit-log root, resolved at call time (B-9).
Overridable with ``DELEGATE_JOB_LOGS_DIR``; otherwise
``<cwd>/.mam/delegate_job_logs``. Resolved per call rather than at import
so a chdir after import cannot strand the audit trail in the old tree —
the same reason ``DEFAULT_REGISTRY_DIR`` stays a relative string.
"""
env = os.environ.get("DELEGATE_JOB_LOGS_DIR")
if env and env.strip():
return env
return os.path.join(os.getcwd(), ".mam", "delegate_job_logs")
def __getattr__(name: str): # PEP 562 (3.7+)
"""Keep ``mqtt_common.LOGS_DIR`` working for external consumers
(documented in registry.md) while resolving it dynamically."""
if name == "LOGS_DIR":
return get_logs_dir()
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
def __dir__(): # PEP 562 recommendation — keep dir()/tab-completion discoverable
return sorted(set(globals()) | {"LOGS_DIR"})
# --------------------------------------------------------------------------
# Broker configuration
# --------------------------------------------------------------------------
@dataclass
class BrokerConfig:
"""Resolved broker connection settings.
PoC defaults target the public HiveMQ broker. Production overrides arrive
either from environment variables or from a job record's ``broker.*`` block
(see ``broker_config_from_job``).
"""
host: str = "broker.hivemq.com"
port: int = 1883
tls: bool = False
username: Optional[str] = None
password: Optional[str] = None
client_id_prefix: str = "hermes"
# TLS material (only consulted when tls is True).
ca_certs: Optional[str] = None
certfile: Optional[str] = None
keyfile: Optional[str] = None
keepalive: int = 60
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
def to_registry_block(self) -> Dict[str, Any]:
"""The subset that gets persisted into a job record's broker block."""
return {
"host": self.host,
"port": self.port,
"tls": self.tls,
"username": self.username,
"password": self.password,
}
def _env_bool(name: str, default: bool = False) -> bool:
raw = os.environ.get(name)
if raw is None:
return default
return raw.strip().lower() in ("1", "true", "yes", "on")
def _env_int(name: str, default: int) -> int:
raw = os.environ.get(name)
if raw is None or raw.strip() == "":
return default
try:
return int(raw)
except ValueError:
logger.warning("invalid int for %s=%r; using default %d", name, raw, default)
return default
def broker_config_from_env(overrides: Optional[Dict[str, Any]] = None) -> BrokerConfig:
"""Build a :class:`BrokerConfig` from environment variables.
Recognised vars (all optional, PoC defaults shown):
MQTT_BROKER (broker.hivemq.com), MQTT_PORT (1883), MQTT_TLS (0),
MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID_PREFIX (hermes),
MQTT_CA_CERTS, MQTT_CERTFILE, MQTT_KEYFILE, MQTT_KEEPALIVE (60).
``overrides`` (e.g. a job record's broker block) wins over the env values
for any key it specifies with a non-None value.
"""
cfg = BrokerConfig(
host=os.environ.get("MQTT_BROKER", "broker.hivemq.com"),
port=_env_int("MQTT_PORT", 1883),
tls=_env_bool("MQTT_TLS", False),
username=os.environ.get("MQTT_USERNAME") or None,
password=os.environ.get("MQTT_PASSWORD") or None,
client_id_prefix=os.environ.get("MQTT_CLIENT_ID_PREFIX", "hermes"),
ca_certs=os.environ.get("MQTT_CA_CERTS") or None,
certfile=os.environ.get("MQTT_CERTFILE") or None,
keyfile=os.environ.get("MQTT_KEYFILE") or None,
keepalive=_env_int("MQTT_KEEPALIVE", 60),
)
if overrides:
for key, value in overrides.items():
if value is not None and hasattr(cfg, key):
setattr(cfg, key, value)
return cfg
def broker_config_from_job(job: Dict[str, Any]) -> BrokerConfig:
"""Resolve broker config for a job: env defaults, then the job's broker.*
block overrides. This lets ``publish_event.py`` connect from the registry
alone, while still honouring environment toggles (e.g. MQTT_TLS=1)."""
return broker_config_from_env(overrides=job.get("broker") or {})
def make_client(role: str, config: Optional[BrokerConfig] = None) -> mqtt.Client:
"""Return a configured paho ``Client`` (not yet connected).
The client id is ``f"{prefix}-{role}-{uuid8}"`` so concurrent publishers /
subscribers never collide on the broker. Auth and TLS are applied when the
config supplies them.
"""
config = config or broker_config_from_env()
client_id = f"{config.client_id_prefix}-{role}-{uuid.uuid4().hex[:8]}"
client = mqtt.Client(
callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
client_id=client_id,
)
if config.username:
client.username_pw_set(config.username, config.password)
if config.tls:
# If ca_certs is None paho uses the system trust store (good enough for
# public CAs); a private CA bundle path is passed through unchanged.
client.tls_set(
ca_certs=config.ca_certs,
certfile=config.certfile,
keyfile=config.keyfile,
)
logger.debug("built client id=%s tls=%s host=%s", client_id, config.tls, config.host)
return client
def reason_code_value(rc: Any) -> int:
"""Normalise a paho v2 connect reason code to an int.
paho-mqtt 2.x hands callbacks a ``ReasonCode`` object (not an int); older
paths may pass a plain int. ``ReasonCode`` exposes ``.value``; 0 == success.
"""
return int(getattr(rc, "value", rc))
def verify_hmac(payload: dict, auth_token: Optional[str]) -> bool:
"""Verify HMAC-SHA256 signature. Returns True if valid or no token set."""
if not auth_token:
return True # PoC mode — no auth
sig = payload.get("data", {}).get("hmac_sig")
if not sig:
return False
sign_payload = {k: v for k, v in payload.items() if k != "data"}
sign_payload["data"] = {k: v for k, v in payload.get("data", {}).items() if k != "hmac_sig"}
msg = json.dumps(sign_payload, sort_keys=True, separators=(",", ":")).encode()
expected = hmac.new(auth_token.encode(), msg, hashlib.sha256).hexdigest()
return hmac.compare_digest(sig, expected)
def topic_prefix_for(job_id: str, root: str = DEFAULT_TOPIC_ROOT) -> str:
return f"{root}/{job_id}"
def events_topic_for(job_id: str, root: str = DEFAULT_TOPIC_ROOT) -> str:
return f"{topic_prefix_for(job_id, root)}/events"
# --------------------------------------------------------------------------
# Registry primitives (single source of truth for raw record I/O)
# --------------------------------------------------------------------------
def _job_path(job_id: str, registry_dir: str) -> Path:
return Path(registry_dir) / f"{job_id}.json"
def _lock_path(registry_dir: str) -> Path:
return Path(registry_dir) / LOCK_FILENAME
@contextmanager
def registry_lock(registry_dir: str):
"""Advisory exclusive lock over the whole registry dir via fcntl.
PoC-grade single-host concurrency control. Multiple herdr sessions / scripts
serialise their read-modify-write of job records through this lock so two
sessions never claim the same pending job. For multi-host delegation move
to SQLite WAL (see references/registry.md)."""
import fcntl # POSIX only; imported lazily so import works on Windows.
Path(registry_dir).mkdir(parents=True, exist_ok=True)
lock_file = _lock_path(registry_dir)
fh = open(lock_file, "a+")
try:
fcntl.flock(fh.fileno(), fcntl.LOCK_EX)
yield
finally:
try:
fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
finally:
fh.close()
def load_job(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> Dict[str, Any]:
"""Load and parse a job record. Raises FileNotFoundError if absent."""
path = _job_path(job_id, registry_dir)
if not path.exists():
raise FileNotFoundError(f"job record not found: {path}")
with open(path, "r", encoding="utf-8") as fh:
return json.load(fh)
def _atomic_write_record(job_id: str, registry_dir: str, record: Dict[str, Any]) -> None:
"""Write a record atomically: temp file in the same dir + os.replace.
The rename is atomic on POSIX, so readers never observe a half-written
file. Callers MUST already hold ``registry_lock`` for read-modify-write
correctness."""
Path(registry_dir).mkdir(parents=True, exist_ok=True)
path = _job_path(job_id, registry_dir)
fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=f".{job_id}.", suffix=".tmp")
try:
with os.fdopen(fd, "w", encoding="utf-8") as fh:
json.dump(record, fh, ensure_ascii=False, indent=2)
fh.write("\n")
fh.flush()
os.fsync(fh.fileno())
os.replace(tmp, path)
try:
os.chmod(path, 0o600)
except Exception:
pass
except BaseException:
if os.path.exists(tmp):
os.unlink(tmp)
raise
def update_job_status(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR, **fields: Any) -> Dict[str, Any]:
"""Atomically merge ``fields`` into a job record under the registry lock.
Always refreshes ``updated_at``. Returns the new record. Raises
FileNotFoundError if the job does not exist.
This is the single chokepoint for status writes (both ``registry.update_status``
and ``publish_event.py``'s status sync route through here), so it also mirrors
any ``status`` change into the persistent audit log. We perform the log mirror
under the lock to guarantee sequential consistency in audit history."""
with registry_lock(registry_dir):
record = load_job(job_id, registry_dir)
old_status = record.get("status")
record.update(fields)
record["updated_at"] = _utcnow()
_atomic_write_record(job_id, registry_dir, record)
if "status" in fields:
new_status = record.get("status")
update_logged_status(job_id, new_status, updated_at=record["updated_at"])
if old_status != new_status:
append_event(job_id, {
"event": "status_changed",
"from": old_status,
"to": new_status,
"timestamp": record["updated_at"],
})
return record
def next_seq(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> int:
"""Return the next monotonic sequence number for a job, persisted in the
record's ``last_seq`` field so it stays consistent across process restarts.
First call returns 1."""
with registry_lock(registry_dir):
record = load_job(job_id, registry_dir)
seq = int(record.get("last_seq", 0)) + 1
record["last_seq"] = seq
record["updated_at"] = _utcnow()
_atomic_write_record(job_id, registry_dir, record)
return seq
def _utcnow() -> str:
"""ISO-8601 UTC timestamp with trailing Z (payload `timestamp` field)."""
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
def _utcnow_precise() -> str:
"""ISO-8601 UTC timestamp with millisecond resolution. Used for the audit
log's ``logged_at`` so events sort cleanly even within the same second."""
now = time.time()
base = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(now))
return f"{base}.{int((now % 1) * 1000):03d}Z"
# --------------------------------------------------------------------------
# Persistent audit log (.mam/delegate_job_logs/<job_id>/...)
#
# Every function here is idempotent, concurrency-safe, and *best-effort*: a
# logging failure is swallowed with a logger.warning and never propagated, so it
# can never break a publish, a subscribe, or a registry write. stdout is never
# touched (it is reserved for data output).
# --------------------------------------------------------------------------
def job_log_dir(job_id: str, logs_dir: Optional[str] = None) -> Path:
return Path(logs_dir or get_logs_dir()) / job_id
def job_log_path(job_id: str, kind: str, logs_dir: Optional[str] = None) -> Path:
"""Path to one audit-log file for a job. ``kind`` is a filename, e.g. the
module constants META_FILENAME / EVENTS_FILENAME / STATUS_FILENAME."""
return job_log_dir(job_id, logs_dir) / kind
@contextmanager
def _file_lock(fh):
"""Best-effort exclusive lock over a single open file via fcntl, so two
processes appending to events.ndjson never interleave a line. A no-op where
fcntl is unavailable (Windows); a short append is atomic enough there."""
try:
import fcntl
except ImportError: # pragma: no cover - non-POSIX
yield
return
fcntl.flock(fh.fileno(), fcntl.LOCK_EX)
try:
yield
finally:
fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
def _redact_dict(d: Any) -> Any:
"""Recursively mask sensitive values (passwords, secrets, tokens) inside logs."""
if isinstance(d, dict):
redacted = {}
for k, v in d.items():
if any(s in k.lower() for s in ("password", "token", "secret", "auth_token", "key")):
redacted[k] = "[REDACTED]"
else:
redacted[k] = _redact_dict(v)
return redacted
elif isinstance(d, list):
return [_redact_dict(item) for item in d]
return d
def append_event(job_id: str, event_dict: Dict[str, Any], logs_dir: Optional[str] = None) -> None:
"""Append one event as a JSON line to ``<logs>/<job_id>/events.ndjson``.
Concurrency-safe (fcntl lock over the file) and best-effort. A millisecond
``logged_at`` is stamped when the caller did not supply one."""
try:
path = job_log_path(job_id, EVENTS_FILENAME, logs_dir)
path.parent.mkdir(parents=True, exist_ok=True)
record = _redact_dict(dict(event_dict))
record.setdefault("logged_at", _utcnow_precise())
line = json.dumps(record, ensure_ascii=False) + "\n"
with open(path, "a", encoding="utf-8") as fh:
with _file_lock(fh):
fh.write(line)
fh.flush()
except Exception as exc: # pragma: no cover - best effort
logger.warning("append_event failed for job %s: %s", job_id, exc)
def update_logged_status(job_id: str, status: str, logs_dir: Optional[str] = None, **extras: Any) -> None:
"""Rewrite ``<logs>/<job_id>/status.json`` (current status for fast point
queries) atomically. Best-effort; merges any ``extras``."""
try:
path = job_log_path(job_id, STATUS_FILENAME, logs_dir)
path.parent.mkdir(parents=True, exist_ok=True)
record: Dict[str, Any] = {"job_id": job_id, "status": status, "updated_at": _utcnow()}
record.update(extras)
tmp = path.with_name(path.name + ".tmp")
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(record, fh, ensure_ascii=False, indent=2)
fh.write("\n")
os.replace(tmp, path)
except Exception as exc: # pragma: no cover - best effort
logger.warning("update_logged_status failed for job %s: %s", job_id, exc)
def init_job_log(job_id: str, meta: Dict[str, Any], logs_dir: Optional[str] = None) -> None:
"""Seed the per-job audit-log dir: write meta.json, status.json, and a first
``registered`` line in events.ndjson. Idempotent (the ``registered`` line is
written only when events.ndjson does not yet exist) and best-effort."""
try:
d = job_log_dir(job_id, logs_dir)
d.mkdir(parents=True, exist_ok=True)
meta_redacted = _redact_dict(meta)
with open(d / META_FILENAME, "w", encoding="utf-8") as fh:
json.dump(meta_redacted, fh, ensure_ascii=False, indent=2)
fh.write("\n")
status = meta.get("status", "pending")
update_logged_status(
job_id, status, logs_dir=logs_dir,
created_at=meta.get("created_at"), prompt=meta.get("prompt"),
)
events_path = d / EVENTS_FILENAME
first_time = not events_path.exists()
events_path.touch(exist_ok=True)
if first_time:
append_event(job_id, {
"event": "registered",
"status": status,
"agent": meta.get("agent"),
"agent_session": meta.get("agent_session"),
"topic_prefix": meta.get("topic_prefix"),
"timestamp": meta.get("created_at"),
}, logs_dir=logs_dir)
except Exception as exc: # pragma: no cover - best effort
logger.warning("init_job_log failed for job %s: %s", job_id, exc)
def read_logged_meta(job_id: str, logs_dir: Optional[str] = None) -> Optional[Dict[str, Any]]:
"""Return a job's audit meta.json (registration snapshot), or None."""
try:
with open(job_log_path(job_id, META_FILENAME, logs_dir), "r", encoding="utf-8") as fh:
return json.load(fh)
except (OSError, json.JSONDecodeError):
return None
def read_logged_status(job_id: str, logs_dir: Optional[str] = None) -> Optional[Dict[str, Any]]:
"""Return a job's current status.json, or None. This is the fast point-query
file (current status only), separate from the registration-time meta.json."""
try:
with open(job_log_path(job_id, STATUS_FILENAME, logs_dir), "r", encoding="utf-8") as fh:
return json.load(fh)
except (OSError, json.JSONDecodeError):
return None
def iter_logged_events(job_id: str, logs_dir: Optional[str] = None):
"""Yield each parsed event from a job's events.ndjson in file (time) order.
Malformed lines are skipped with a warning."""
path = job_log_path(job_id, EVENTS_FILENAME, logs_dir)
if not path.exists():
return
with open(path, "r", encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if not line:
continue
try:
yield json.loads(line)
except json.JSONDecodeError:
logger.warning("skipping malformed audit line in %s", path)
def list_logged_jobs(logs_dir: Optional[str] = None) -> List[Dict[str, Any]]:
"""Return one meta record per job directory under the logs root, oldest
first. Falls back to ``{"job_id": <dir>}`` when meta.json is missing."""
base = Path(logs_dir or get_logs_dir())
out: List[Dict[str, Any]] = []
if not base.exists():
return out
for d in sorted(base.iterdir()):
if not d.is_dir():
continue
meta = read_logged_meta(d.name, logs_dir) or {"job_id": d.name}
# Overlay the live status.json so the summary reflects current state, not
# the registration-time snapshot frozen in meta.json.
status = read_logged_status(d.name, logs_dir)
if status:
meta = {**meta,
"status": status.get("status", meta.get("status")),
"updated_at": status.get("updated_at", meta.get("updated_at"))}
out.append(meta)
out.sort(key=lambda m: m.get("created_at") or "")
return out
# --------------------------------------------------------------------------
# Retry helper
# --------------------------------------------------------------------------
def with_retry(
fn: Optional[Callable] = None,
*,
attempts: int = 3,
base_delay: float = 0.5,
factor: float = 2.0,
max_delay: float = 8.0,
exceptions: Iterable[type] = (Exception,),
) -> Callable:
"""Retry ``fn`` with exponential backoff.
Usable two ways::
result = with_retry(do_publish, attempts=3)() # wrap-and-call
@with_retry(attempts=5, base_delay=1.0) # decorator
def do_publish(): ...
Re-raises the last exception once ``attempts`` is exhausted.
"""
exc_tuple = tuple(exceptions)
def decorate(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
delay = base_delay
last_exc: Optional[BaseException] = None
for attempt in range(1, attempts + 1):
try:
return func(*args, **kwargs)
except exc_tuple as exc:
last_exc = exc
if attempt >= attempts:
break
logger.warning(
"attempt %d/%d failed: %s; retrying in %.1fs",
attempt, attempts, exc, delay,
)
time.sleep(delay)
delay = min(delay * factor, max_delay)
assert last_exc is not None
raise last_exc
return wrapper
if fn is not None:
return decorate(fn)
return decorate
def setup_logging(level: int = logging.WARNING) -> None:
"""Configure root logging to stderr. stdout is reserved for data output
(subscriber event lines, registry ids)."""
import sys
logging.basicConfig(
level=level,
stream=sys.stderr,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
+240
View File
@@ -0,0 +1,240 @@
#!/usr/bin/env python3
"""publish_event.py — the single entry point for emitting a Job event.
Loads the job record from the registry, resolves its broker, assigns the next
monotonic ``seq``, builds the schema-v1 JSON payload, and publishes it to
``<topic_prefix>/events`` over QoS 1 with exponential-backoff retry.
Silent by design: nothing is printed to stdout. Diagnostics go to stderr via
logging. Terminal events (``completed``/``error``) publish with retain=True so
a late subscriber still observes the final state (production hardening).
Exit codes:
0 published successfully
1 parameter / registry error (bad args, unknown job, no pending job)
2 publish failed after retries (network / broker / ACK timeout)
Usage:
publish_event.py --job <id> --event started [--detail "..."] [--data '{...}']
publish_event.py --pick-pending --agent-session herdr:claude --event completed
publish_event.py --job <id> --event completed --retained
"""
from __future__ import annotations
import argparse
import hashlib
import hmac
import json
import logging
import sys
import time
from typing import Any, Dict, Optional
from lib_py.delegation import mqtt_common
from lib_py.delegation import registry
from lib_py.delegation.mqtt_common import (
DEFAULT_REGISTRY_DIR,
SCHEMA_VERSION,
broker_config_from_job,
events_topic_for,
load_job,
make_client,
next_seq,
with_retry,
)
logger = logging.getLogger("delegate_job.publish_event")
VALID_EVENTS = ("started", "permission_required", "progress", "completed", "error", "cancelled")
TERMINAL_EVENTS = ("completed", "error", "cancelled")
# event -> registry status to sync as a best-effort side effect
EVENT_TO_STATUS = {
"started": "running",
"completed": "completed",
"error": "error",
"cancelled": "cancelled",
}
CONNECT_ACK_TIMEOUT = 10 # seconds to wait for CONNACK
PUBLISH_ACK_TIMEOUT = 5 # seconds to wait for QoS-1 PUBACK
def build_payload(
job_id: str,
seq: int,
event: str,
detail: str,
data: Optional[Dict[str, Any]],
auth_token: Optional[str],
) -> Dict[str, Any]:
payload: Dict[str, Any] = {
"schema_version": SCHEMA_VERSION,
"seq": seq,
"job_id": job_id,
"event": event,
"timestamp": mqtt_common._utcnow(),
"detail": detail,
"data": dict(data) if data else {},
}
# Production: carry the per-job HMAC-SHA256 signature in `data.hmac_sig` so
# the subscriber can verify the publisher without exposing the secret token.
# The signature is calculated over the entire payload (with `data.hmac_sig` excluded).
if auth_token:
sign_payload = {k: v for k, v in payload.items() if k != "data"}
sign_payload["data"] = {k: v for k, v in payload.get("data", {}).items() if k != "hmac_sig"}
msg = json.dumps(sign_payload, sort_keys=True, separators=(",", ":")).encode()
sig = hmac.new(auth_token.encode(), msg, hashlib.sha256).hexdigest()
payload["data"]["hmac_sig"] = sig
return payload
def _publish_once(config, topic: str, body: bytes, retain: bool) -> None:
"""Connect, publish one QoS-1 message, wait for the broker ACK, disconnect.
Raises on any failure so ``with_retry`` can re-run the whole sequence (a
fresh connection per attempt is the robust choice for a PoC)."""
client = make_client("publisher", config)
connected = {"rc": None}
def on_connect(_c, _u, _flags, reason_code, _props):
connected["rc"] = reason_code
client.on_connect = on_connect
client.connect(config.host, config.port, config.keepalive)
client.loop_start()
try:
# Wait for CONNACK so we fail fast on auth/TLS errors.
deadline = time.monotonic() + CONNECT_ACK_TIMEOUT
while connected["rc"] is None and time.monotonic() < deadline:
time.sleep(0.05)
if connected["rc"] is None:
raise TimeoutError("no CONNACK from broker")
if mqtt_common.reason_code_value(connected["rc"]) != 0:
raise ConnectionError(f"broker refused connection: rc={connected['rc']}")
info = client.publish(topic, payload=body, qos=1, retain=retain)
info.wait_for_publish(timeout=PUBLISH_ACK_TIMEOUT)
if not info.is_published():
raise TimeoutError("publish not acknowledged within timeout")
finally:
client.loop_stop()
try:
client.disconnect()
except Exception: # pragma: no cover - disconnect best effort
pass
def _resolve_job_id(args) -> Optional[str]:
if args.pick_pending:
return registry.pick_pending(args.agent_session, args.registry_dir)
return args.job
def main(argv=None) -> int:
parser = argparse.ArgumentParser(description="Publish a Job event to MQTT")
target = parser.add_mutually_exclusive_group(required=True)
target.add_argument("--job", help="job id to publish for")
target.add_argument("--pick-pending", action="store_true",
help="auto-select a pending job for --agent-session")
parser.add_argument("--agent-session", default="herdr:claude",
help="session label used with --pick-pending")
parser.add_argument("--event", default="progress", choices=VALID_EVENTS)
parser.add_argument("--detail", default="")
parser.add_argument("--data", default=None, help="optional JSON object string")
parser.add_argument("--retained", action="store_true",
help="force retain=True (auto for completed/error)")
parser.add_argument("--registry-dir", default=DEFAULT_REGISTRY_DIR)
parser.add_argument("--attempts", type=int, default=3)
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.WARNING)
# --- parse optional data JSON (parameter error -> exit 1) ---
data: Optional[Dict[str, Any]] = None
if args.data:
try:
data = json.loads(args.data)
if not isinstance(data, dict):
raise ValueError("--data must be a JSON object")
except (ValueError, json.JSONDecodeError) as exc:
logger.error("invalid --data: %s", exc)
return 1
job_id = _resolve_job_id(args)
if not job_id:
logger.error("no job to publish for (unknown --job or no pending job)")
return 1
try:
job = load_job(job_id, args.registry_dir)
except FileNotFoundError as exc:
logger.error("%s", exc)
return 1
config = broker_config_from_job(job)
topic = job.get("topic_prefix")
topic = f"{topic}/events" if topic else events_topic_for(job_id)
seq = next_seq(job_id, args.registry_dir)
payload = build_payload(
job_id=job_id,
seq=seq,
event=args.event,
detail=args.detail,
data=data,
auth_token=job.get("auth_token"),
)
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
retain = args.retained or args.event in TERMINAL_EVENTS
publish = with_retry(
_publish_once,
attempts=args.attempts,
exceptions=(OSError, TimeoutError, ConnectionError, ValueError),
)
publish_ok = True
publish_error: Optional[str] = None
try:
publish(config, topic, body, retain)
except Exception as exc:
publish_ok = False
publish_error = str(exc)
logger.error("publish failed after %d attempts: %s", args.attempts, exc)
# Persistent audit log: record the exact payload we put on the wire (or intended to).
# Best-effort (isolated inside append_event) — never fails the publish.
# Policy Note: Seq consumption on failure is intentionally maintained for monotonic replay
# defense (> highest accepted seq), with failure documented explicitly in the audit record.
mqtt_common.append_event(job_id, {
"event": "published",
"source_event": args.event,
"seq": seq,
"topic": topic,
"retain": retain,
"timestamp": payload["timestamp"],
"detail": args.detail,
"payload": payload,
"published": publish_ok,
"publish_error": publish_error,
})
# Best-effort side effects: registry status sync + (debug) event log. Never
# fail the publish on these.
registry.append_event(job_id, args.registry_dir, payload)
new_status = EVENT_TO_STATUS.get(args.event)
if new_status:
try:
mqtt_common.update_job_status(job_id, args.registry_dir, status=new_status)
except Exception as exc: # pragma: no cover - best effort
logger.warning("status sync failed: %s", exc)
if not publish_ok:
return 2
logger.info("published %s seq=%d job=%s retain=%s", args.event, seq, job_id, retain)
return 0
if __name__ == "__main__":
sys.exit(main())
+427
View File
@@ -0,0 +1,427 @@
"""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 sys
import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional
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,
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]
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: int = 120,
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,
) -> 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.
"""
job_id = job_id or generate_job_id(bits)
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,
}
with registry_lock(registry_dir):
if mqtt_common._job_path(job_id, registry_dir).exists():
raise FileExistsError(f"job already exists: {job_id}")
_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)
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."""
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"]
# pick_pending writes the record directly (not via update_job_status), so it
# mirrors the pending->running transition into the audit log here. Best-effort.
mqtt_common.update_logged_status(job_id, "running", updated_at=updated_at)
mqtt_common.append_event(job_id, {
"event": "status_changed",
"from": "pending",
"to": "running",
"by": agent_session,
"timestamp": updated_at,
})
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 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",
]
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) -> str:
"""Read the job's audit log or events log and return the detail of the last completed/error event."""
# 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()
events = list(mqtt_common.iter_logged_events(job_id, logs_dir))
for e in reversed(events):
if e.get("source_event") in ("completed", "error"):
return e.get("detail", "")
if e.get("event") in ("completed", "error"):
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 ("completed", "error"):
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")
p_reg.add_argument("--prompt", required=True)
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=120)
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_list = sub.add_parser("list", help="list jobs (optionally by status)")
p_list.add_argument("--status", default=None)
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)
p_update.add_argument("--prompt", default=None)
p_update.add_argument("--iteration", type=int, default=None)
p_update.add_argument("--role", 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 <cwd>/.mam/delegate_job_logs)")
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")
return parser
def main(argv: Optional[List[str]] = None) -> int:
mqtt_common.ensure_env_loaded()
mqtt_common.setup_logging(logging.INFO)
args = _build_parser().parse_args(argv)
rd = args.registry_dir
if args.command == "register":
job_id = register_job(
prompt=args.prompt,
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,
)
print(job_id)
return 0
if args.command == "list":
records = list_jobs(rd, status=args.status)
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:
update_status(args.job, rd, args.status)
except (FileNotFoundError, ValueError) as exc:
print(str(exc), file=sys.stderr)
return 1
return 0
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
if args.prompt is not None:
fields["prompt"] = args.prompt
if args.iteration is not None:
fields["iteration"] = args.iteration
if args.role is not None:
fields["role"] = args.role
try:
mqtt_common.update_job_status(args.job, rd, **fields)
except FileNotFoundError as exc:
print(str(exc), file=sys.stderr)
return 1
return 0
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)
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()
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())
@@ -1,339 +1,37 @@
#!/usr/bin/env python3
"""job_subscriber.py — the single entry point for observing Job events.
"""Compatibility shim -- the implementation lives in lib_py.delegation.job_subscriber.
Subscribes to one job's ``<topic_prefix>/events`` (or, with ``--wait-any``, the
events of every running/pending job in the registry), prints one line to stdout
per accepted event, and exits on a terminal event or a timeout.
This file's path is a wire contract (agent instructions, the bash wrapper,
SKILL.md, BOOTSTRAP.md, deploy/install.sh, tests). It binds the
``lib_py.delegation`` *package* to the tree next to this file, once per process,
without touching ``sys.path``; ``import job_subscriber`` then yields the very same module
object as ``lib_py.delegation.job_subscriber``.
Design points (all flagged in the PLAN review):
- terminal state machine: ``completed``/``error`` is acted on exactly once per
job, so QoS-1 duplicates or an ``error``-after-``completed`` reorder are safe.
- dual timeouts: a wall-clock ``--timeout`` (total budget, started at
subscribe time so a cold start can't hang forever) AND an idle
``--idle-timeout`` (no new event for N seconds).
- defensive parsing: undecodable payloads, ``schema_version`` mismatches, and
``job_id`` values we did not subscribe for are logged and dropped.
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
Why the package and not this one module: a per-module bind leaves a non-leaf
module (one that imports its siblings) importing dependencies from whichever tree
happens to be on the path, mixing a frozen copy with a live one. Why bind-once and
never rebind: replacing a bound package would split existing references and
break monkeypatching. See docs/contracts/issue-5/ (D2) for the context matrix.
"""
from __future__ import annotations
import argparse
import json
import logging
import queue
import sys
import time
from typing import Any, Dict, List, Optional, Set, Tuple
import mqtt_common
import registry
from mqtt_common import (
DEFAULT_REGISTRY_DIR,
SCHEMA_VERSION,
broker_config_from_job,
load_job,
make_client,
)
logger = logging.getLogger("delegate_job.job_subscriber")
TERMINAL_EVENTS = ("completed", "error")
def _format_line(topic: str, payload: Dict[str, Any]) -> str:
source_tag = f" [{payload['source']}]" if payload.get("source") else ""
return (
f"{payload.get('timestamp','-')} "
f"job={payload.get('job_id','?')} "
f"seq={payload.get('seq','?')} "
f"{payload.get('event','?') + source_tag:<20} "
f"{payload.get('detail','')}"
)
def _check_disk_fallback(
pending: Set[str],
registry_dir: str,
terminal: Dict[str, str],
) -> None:
"""Check local on-disk registry and audit logs for terminal status (B-15)."""
for jid in list(pending):
disk_status = None
try:
job_rec = load_job(jid, registry_dir)
disk_status = job_rec.get("status")
except Exception:
pass
if not disk_status or disk_status not in ("completed", "error", "cancelled"):
try:
disk_status = mqtt_common.read_logged_status(jid, mqtt_common.get_logs_dir())
except Exception:
pass
if disk_status in ("completed", "error", "cancelled"):
synth_event = "completed" if disk_status == "completed" else "error"
synth_payload = {
"job_id": jid,
"event": synth_event,
"seq": -1,
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"detail": f"terminal status '{disk_status}' resolved via disk-fallback",
"source": "disk-fallback",
}
synth_topic = f"mam/{jid}/events"
print(_format_line(synth_topic, synth_payload), flush=True)
terminal[jid] = synth_event
pending.discard(jid)
class _Watcher:
"""Holds the shared queue + the set of job_ids we accept events for."""
def __init__(self, expected_job_ids: Set[str], expected_tokens: Dict[str, Optional[str]], expected_seqs: Dict[str, int]):
self.events: "queue.Queue[Tuple[str, Dict[str, Any]]]" = queue.Queue()
self.expected = set(expected_job_ids)
self.tokens = expected_tokens # job_id -> expected auth_token (or None)
self.last_seq = dict(expected_seqs)
def on_message(self, _client, _userdata, msg) -> None:
# --- defensive parsing -------------------------------------------
try:
payload = json.loads(msg.payload.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
logger.warning("drop unparseable payload on %s: %s", msg.topic, exc)
return
if not isinstance(payload, dict):
logger.warning("drop non-object payload on %s", msg.topic)
return
if payload.get("schema_version") != SCHEMA_VERSION:
logger.warning("drop event with schema_version=%r (expected %d)",
payload.get("schema_version"), SCHEMA_VERSION)
return
jid = payload.get("job_id")
if jid not in self.expected:
logger.warning("drop event for unexpected job_id=%r on %s", jid, msg.topic)
return
# --- production auth check: data.auth_token must match if expected ---
expected_token = self.tokens.get(jid)
if not mqtt_common.verify_hmac(payload, expected_token):
logger.warning("drop event for job %s: HMAC verify failed", jid)
return
# --- replay attack defense: check monotonic sequence ---
seq = payload.get("seq")
if seq is None or not isinstance(seq, int):
logger.warning("drop event for job %s: missing or invalid seq", jid)
return
if seq <= self.last_seq.get(jid, 0):
logger.warning("drop event for job %s: seq %d is not monotonically increasing (last %d)",
jid, seq, self.last_seq.get(jid, 0))
return
self.last_seq[jid] = seq
# Persistent audit log from the *subscriber's* vantage point: every event
# that survives defensive parsing is recorded here, including ones a
# different host published. This is the external-observer record that
# backstops the publisher's own "published" line if it never wrote one.
mqtt_common.append_event(jid, {
"event": "received",
"source_event": payload.get("event"),
"seq": payload.get("seq"),
"topic": msg.topic,
"timestamp": payload.get("timestamp"),
"detail": payload.get("detail", ""),
})
self.events.put((msg.topic, payload))
def _collect_jobs(args) -> List[Dict[str, Any]]:
"""Resolve the list of job records this invocation should watch."""
if args.wait_any:
jobs = [r for r in registry.list_jobs(args.registry_dir)
if r.get("status") in ("pending", "running")]
if not jobs:
logger.error("no pending/running jobs to wait for")
return jobs
job = load_job(args.job, args.registry_dir) # raises FileNotFoundError
return [job]
def _run_subscriber(args) -> int:
import importlib, importlib.util, os, sys
_SKILLS = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
_PKG = "lib_py.delegation"
if _PKG not in sys.modules: # bind once per process; never rebind
try:
jobs = _collect_jobs(args)
except FileNotFoundError as exc:
logger.error("%s", exc)
return 2
if not jobs:
return 2
expected_ids: Set[str] = {j["job_id"] for j in jobs}
tokens = {j["job_id"]: j.get("auth_token") for j in jobs}
seqs = {}
for j in jobs:
jid = j["job_id"]
last_seq = int(j.get("last_seq", 0))
seqs[jid] = max(0, last_seq - 1)
watcher = _Watcher(expected_ids, tokens, seqs)
# Resolve timeouts from CLI, falling back to the (first) job's settings.
base_job = jobs[0]
wall_timeout = args.timeout if args.timeout is not None else float(base_job.get("timeout_sec", 3600))
idle_timeout = args.idle_timeout if args.idle_timeout is not None else float(base_job.get("idle_timeout_sec", 120))
# All watched jobs share a broker in practice; connect using the first
# job's broker and subscribe to each job's events topic.
config = broker_config_from_job(base_job)
client = make_client("subscriber", config)
client.on_message = watcher.on_message
subscribed_topics = []
for job in jobs:
prefix = job.get("topic_prefix") or mqtt_common.topic_prefix_for(job["job_id"])
subscribed_topics.append(f"{prefix}/events")
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
for topic in subscribed_topics:
_c.subscribe(topic, qos=1)
logger.info("subscribed to %s", topic)
def on_disconnect(_c, _u, _flags, reason_code, _props):
rc = mqtt_common.reason_code_value(reason_code)
if rc != 0:
logger.warning("broker disconnected (rc=%s); will retry reconnect", reason_code)
def on_subscribe(_c, _u, mid, granted_qos, _props=None):
for topic in subscribed_topics:
print(f"SUBSCRIBED {topic}", flush=True)
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.on_subscribe = on_subscribe
client.reconnect_delay_set(min_delay=1, max_delay=16)
terminal: Dict[str, str] = {} # job_id -> "completed"/"error"
pending: Set[str] = set(expected_ids)
start = time.monotonic()
wall_deadline = start + wall_timeout
last_event = start
last_disk_check = 0.0
retention_checked = not args.expect_retention
connected = False
# Check if all jobs are already in terminal state on disk before connecting
_check_disk_fallback(pending, args.registry_dir, terminal)
if not pending:
if any(state == "error" for state in terminal.values()):
return 1
return 0
try:
mqtt_common.with_retry(
lambda: client.connect(config.host, config.port, config.keepalive),
attempts=3, base_delay=0.2, max_delay=1.0
)()
client.loop_start()
connected = True
except Exception as exc:
logger.warning("broker connection failed (%s); checking disk fallback", exc)
_check_disk_fallback(pending, args.registry_dir, terminal)
if not pending:
if any(state == "error" for state in terminal.values()):
return 1
return 0
logger.error("broker connection failed: %s", exc)
return 3
try:
while pending:
now = time.monotonic()
if now >= wall_deadline:
_check_disk_fallback(pending, args.registry_dir, terminal)
if not pending:
break
logger.error("wall-clock timeout (%.0fs); still pending: %s",
wall_timeout, ", ".join(sorted(pending)))
return 2
idle_left = idle_timeout - (now - last_event)
if idle_left <= 0:
_check_disk_fallback(pending, args.registry_dir, terminal)
if not pending:
break
logger.error("idle timeout (%.0fs, no events); still pending: %s",
idle_timeout, ", ".join(sorted(pending)))
return 2
wait = min(wall_deadline - now, idle_left, 1.0)
try:
topic, payload = watcher.events.get(timeout=wait)
except queue.Empty:
if not retention_checked and (now - start) > 3.0:
logger.warning("--expect-retention set but no retained "
"terminal event observed yet")
retention_checked = True
# Step 2 (B-15): Local disk fallback throttled to every 3.0s
if (now - last_disk_check) >= 3.0:
last_disk_check = now
_check_disk_fallback(pending, args.registry_dir, terminal)
continue
last_event = time.monotonic()
retention_checked = True
print(_format_line(topic, payload), flush=True)
jid = payload["job_id"]
event = payload.get("event")
if event in TERMINAL_EVENTS:
if jid in terminal:
# Already finalised: ignore duplicates / late reorders.
logger.info("ignoring duplicate terminal %s for %s", event, jid)
continue
terminal[jid] = event
pending.discard(jid)
finally:
if connected:
client.loop_stop()
try:
client.disconnect()
except Exception: # pragma: no cover
pass
# All jobs reached a terminal state. error wins over completed.
if any(state == "error" for state in terminal.values()):
return 1
return 0
def main(argv=None) -> int:
parser = argparse.ArgumentParser(description="Subscribe to Job events on MQTT")
target = parser.add_mutually_exclusive_group(required=True)
target.add_argument("--job", help="job id to watch")
target.add_argument("--wait-any", action="store_true",
help="watch every pending/running job in the registry")
parser.add_argument("--timeout", type=float, default=None,
help="wall-clock budget in seconds (default: job.timeout_sec or 3600)")
parser.add_argument("--idle-timeout", type=float, default=None,
help="max seconds with no new event (default: job.idle_timeout_sec or 120)")
parser.add_argument("--expect-retention", action="store_true",
help="warn if no retained terminal event arrives promptly")
parser.add_argument("--registry-dir", default=DEFAULT_REGISTRY_DIR)
parser.add_argument("-v", "--verbose", action="store_true")
args = parser.parse_args(argv)
mqtt_common.setup_logging(logging.DEBUG if args.verbose else logging.WARNING)
try:
return _run_subscriber(args)
except Exception as exc:
logger.error("subscriber fatal error: %s", exc)
return 3
if __name__ == "__main__":
sys.exit(main())
import lib_py # parent resolved normally; shims never rebind it
except ModuleNotFoundError:
sys.path.append(_SKILLS) # standalone only: no lib_py exists anywhere to shadow
import lib_py
_d = os.path.join(_SKILLS, "lib_py", "delegation")
_spec = importlib.util.spec_from_file_location(
_PKG, os.path.join(_d, "__init__.py"), submodule_search_locations=[_d])
_pkg = importlib.util.module_from_spec(_spec)
sys.modules[_PKG] = _pkg # register BEFORE exec so __init__'s own imports resolve here
_spec.loader.exec_module(_pkg)
lib_py.delegation = _pkg
_impl = importlib.import_module(_PKG + ".job_subscriber") # resolved through the bound package's __path__
if __name__ != "__main__":
sys.modules[__name__] = _impl # `import job_subscriber` is lib_py.delegation.job_subscriber
else:
_main = getattr(_impl, "main", None) # a module without a CLI keeps exit status 0
sys.exit(_main() if _main is not None else 0)
+33 -666
View File
@@ -1,670 +1,37 @@
"""Shared MQTT + registry helpers for the multi-agent-mux-delegate-job skill.
#!/usr/bin/env python3
"""Compatibility shim -- the implementation lives in lib_py.delegation.mqtt_common.
Single entry point for:
- broker configuration (env -> dataclass),
- paho client construction (auth + TLS + unique client id),
- monotonic per-job sequence counters,
- retry-with-exponential-backoff,
- atomic registry record load/update under an fcntl lock.
This file's path is a wire contract (agent instructions, the bash wrapper,
SKILL.md, BOOTSTRAP.md, deploy/install.sh, tests). It binds the
``lib_py.delegation`` *package* to the tree next to this file, once per process,
without touching ``sys.path``; ``import mqtt_common`` then yields the very same module
object as ``lib_py.delegation.mqtt_common``.
Requires paho-mqtt >= 2.0 (uses CallbackAPIVersion.VERSION2).
This module is the *only* place that talks to the broker config and to the
raw job record file, so PoC -> production migration touches just env/registry
values, never code (see references/mqtt-broker-setup.md).
Why the package and not this one module: a per-module bind leaves a non-leaf
module (one that imports its siblings) importing dependencies from whichever tree
happens to be on the path, mixing a frozen copy with a live one. Why bind-once and
never rebind: replacing a bound package would split existing references and
break monkeypatching. See docs/contracts/issue-5/ (D2) for the context matrix.
"""
from __future__ import annotations
import functools
import hashlib
import hmac
import json
import logging
import os
import tempfile
import time
import uuid
from contextlib import contextmanager
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any, Callable, Dict, Iterable, List, Optional
import paho.mqtt.client as mqtt
logger = logging.getLogger("delegate_job.mqtt_common")
_warned_deprecated_env = False
_warned_coexistence_env = False
def _load_dotenv(workspace_dir: Optional[str] = None) -> None:
"""Load .mam.env (or .env fallback) from workspace if it exists.
Only sets vars that are not already in os.environ
(i.e. OS env takes precedence over env files).
"""
global _warned_deprecated_env, _warned_coexistence_env
# 1. Check explicit MAM_ENV_FILE override
explicit_file = os.environ.get("MAM_ENV_FILE")
if explicit_file:
if os.path.isfile(explicit_file):
_parse_env_file(explicit_file)
return
# 2. Resolve workspace directory with boundary marker check (.agents or .git)
if workspace_dir is None:
curr = os.path.dirname(os.path.abspath(__file__))
resolved_root = None
while curr and curr != os.path.dirname(curr):
if os.path.isdir(os.path.join(curr, ".agents")) or os.path.exists(os.path.join(curr, ".git")):
resolved_root = curr
break
curr = os.path.dirname(curr)
if not resolved_root:
return
d = resolved_root
else:
d = workspace_dir
mam_env_path = os.path.join(d, ".mam.env")
legacy_env_path = os.path.join(d, ".env")
has_mam = os.path.isfile(mam_env_path)
has_legacy = os.path.isfile(legacy_env_path)
if has_mam and has_legacy:
if not _warned_coexistence_env:
logger.warning(
"Both '%s' and '%s' exist. Loading '%s'. "
"Consider removing or migrating '%s'.",
mam_env_path, legacy_env_path, mam_env_path, legacy_env_path
)
_warned_coexistence_env = True
_parse_env_file(mam_env_path)
elif has_mam:
_parse_env_file(mam_env_path)
elif has_legacy:
if not _warned_deprecated_env:
logger.warning(
"Loading deprecated config file '%s'. "
"Please migrate to '.mam.env'.",
legacy_env_path
)
_warned_deprecated_env = True
_parse_env_file(legacy_env_path)
def _parse_env_file(path: str) -> None:
import importlib, importlib.util, os, sys
_SKILLS = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
_PKG = "lib_py.delegation"
if _PKG not in sys.modules: # bind once per process; never rebind
try:
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" in line:
key, _, val = line.partition("=")
key = key.strip()
val = val.strip().strip('"').strip("'")
if key and key not in os.environ:
os.environ[key] = val
except Exception as e:
logger.warning("Failed to parse env file %s: %s", path, e)
_load_dotenv()
# --------------------------------------------------------------------------
# Constants
# --------------------------------------------------------------------------
SCHEMA_VERSION = 1
DEFAULT_REGISTRY_DIR = ".mam/jobs"
DEFAULT_TOPIC_ROOT = "python/mqtt/jobs"
LOCK_FILENAME = ".lock"
# Persistent audit-log layout: .mam/delegate_job_logs/<job_id>/{meta,events,status}.
# This is a *separate* artifact from the registry: the registry is the live job
# record (mutated in place), the audit log is an append-only history that
# survives even if the registry dir is cleaned up.
META_FILENAME = "meta.json"
EVENTS_FILENAME = "events.ndjson"
STATUS_FILENAME = "status.json"
def get_logs_dir() -> str:
"""Audit-log root, resolved at call time (B-9).
Overridable with ``DELEGATE_JOB_LOGS_DIR``; otherwise
``<cwd>/.mam/delegate_job_logs``. Resolved per call rather than at import
so a chdir after import cannot strand the audit trail in the old tree —
the same reason ``DEFAULT_REGISTRY_DIR`` stays a relative string.
"""
env = os.environ.get("DELEGATE_JOB_LOGS_DIR")
if env and env.strip():
return env
return os.path.join(os.getcwd(), ".mam", "delegate_job_logs")
def __getattr__(name: str): # PEP 562 (3.7+)
"""Keep ``mqtt_common.LOGS_DIR`` working for external consumers
(documented in registry.md) while resolving it dynamically."""
if name == "LOGS_DIR":
return get_logs_dir()
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
def __dir__(): # PEP 562 recommendation — keep dir()/tab-completion discoverable
return sorted(set(globals()) | {"LOGS_DIR"})
# --------------------------------------------------------------------------
# Broker configuration
# --------------------------------------------------------------------------
@dataclass
class BrokerConfig:
"""Resolved broker connection settings.
PoC defaults target the public HiveMQ broker. Production overrides arrive
either from environment variables or from a job record's ``broker.*`` block
(see ``broker_config_from_job``).
"""
host: str = "broker.hivemq.com"
port: int = 1883
tls: bool = False
username: Optional[str] = None
password: Optional[str] = None
client_id_prefix: str = "hermes"
# TLS material (only consulted when tls is True).
ca_certs: Optional[str] = None
certfile: Optional[str] = None
keyfile: Optional[str] = None
keepalive: int = 60
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
def to_registry_block(self) -> Dict[str, Any]:
"""The subset that gets persisted into a job record's broker block."""
return {
"host": self.host,
"port": self.port,
"tls": self.tls,
"username": self.username,
"password": self.password,
}
def _env_bool(name: str, default: bool = False) -> bool:
raw = os.environ.get(name)
if raw is None:
return default
return raw.strip().lower() in ("1", "true", "yes", "on")
def _env_int(name: str, default: int) -> int:
raw = os.environ.get(name)
if raw is None or raw.strip() == "":
return default
try:
return int(raw)
except ValueError:
logger.warning("invalid int for %s=%r; using default %d", name, raw, default)
return default
def broker_config_from_env(overrides: Optional[Dict[str, Any]] = None) -> BrokerConfig:
"""Build a :class:`BrokerConfig` from environment variables.
Recognised vars (all optional, PoC defaults shown):
MQTT_BROKER (broker.hivemq.com), MQTT_PORT (1883), MQTT_TLS (0),
MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID_PREFIX (hermes),
MQTT_CA_CERTS, MQTT_CERTFILE, MQTT_KEYFILE, MQTT_KEEPALIVE (60).
``overrides`` (e.g. a job record's broker block) wins over the env values
for any key it specifies with a non-None value.
"""
cfg = BrokerConfig(
host=os.environ.get("MQTT_BROKER", "broker.hivemq.com"),
port=_env_int("MQTT_PORT", 1883),
tls=_env_bool("MQTT_TLS", False),
username=os.environ.get("MQTT_USERNAME") or None,
password=os.environ.get("MQTT_PASSWORD") or None,
client_id_prefix=os.environ.get("MQTT_CLIENT_ID_PREFIX", "hermes"),
ca_certs=os.environ.get("MQTT_CA_CERTS") or None,
certfile=os.environ.get("MQTT_CERTFILE") or None,
keyfile=os.environ.get("MQTT_KEYFILE") or None,
keepalive=_env_int("MQTT_KEEPALIVE", 60),
)
if overrides:
for key, value in overrides.items():
if value is not None and hasattr(cfg, key):
setattr(cfg, key, value)
return cfg
def broker_config_from_job(job: Dict[str, Any]) -> BrokerConfig:
"""Resolve broker config for a job: env defaults, then the job's broker.*
block overrides. This lets ``publish_event.py`` connect from the registry
alone, while still honouring environment toggles (e.g. MQTT_TLS=1)."""
return broker_config_from_env(overrides=job.get("broker") or {})
def make_client(role: str, config: Optional[BrokerConfig] = None) -> mqtt.Client:
"""Return a configured paho ``Client`` (not yet connected).
The client id is ``f"{prefix}-{role}-{uuid8}"`` so concurrent publishers /
subscribers never collide on the broker. Auth and TLS are applied when the
config supplies them.
"""
config = config or broker_config_from_env()
client_id = f"{config.client_id_prefix}-{role}-{uuid.uuid4().hex[:8]}"
client = mqtt.Client(
callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
client_id=client_id,
)
if config.username:
client.username_pw_set(config.username, config.password)
if config.tls:
# If ca_certs is None paho uses the system trust store (good enough for
# public CAs); a private CA bundle path is passed through unchanged.
client.tls_set(
ca_certs=config.ca_certs,
certfile=config.certfile,
keyfile=config.keyfile,
)
logger.debug("built client id=%s tls=%s host=%s", client_id, config.tls, config.host)
return client
def reason_code_value(rc: Any) -> int:
"""Normalise a paho v2 connect reason code to an int.
paho-mqtt 2.x hands callbacks a ``ReasonCode`` object (not an int); older
paths may pass a plain int. ``ReasonCode`` exposes ``.value``; 0 == success.
"""
return int(getattr(rc, "value", rc))
def verify_hmac(payload: dict, auth_token: Optional[str]) -> bool:
"""Verify HMAC-SHA256 signature. Returns True if valid or no token set."""
if not auth_token:
return True # PoC mode — no auth
sig = payload.get("data", {}).get("hmac_sig")
if not sig:
return False
sign_payload = {k: v for k, v in payload.items() if k != "data"}
sign_payload["data"] = {k: v for k, v in payload.get("data", {}).items() if k != "hmac_sig"}
msg = json.dumps(sign_payload, sort_keys=True, separators=(",", ":")).encode()
expected = hmac.new(auth_token.encode(), msg, hashlib.sha256).hexdigest()
return hmac.compare_digest(sig, expected)
def topic_prefix_for(job_id: str, root: str = DEFAULT_TOPIC_ROOT) -> str:
return f"{root}/{job_id}"
def events_topic_for(job_id: str, root: str = DEFAULT_TOPIC_ROOT) -> str:
return f"{topic_prefix_for(job_id, root)}/events"
# --------------------------------------------------------------------------
# Registry primitives (single source of truth for raw record I/O)
# --------------------------------------------------------------------------
def _job_path(job_id: str, registry_dir: str) -> Path:
return Path(registry_dir) / f"{job_id}.json"
def _lock_path(registry_dir: str) -> Path:
return Path(registry_dir) / LOCK_FILENAME
@contextmanager
def registry_lock(registry_dir: str):
"""Advisory exclusive lock over the whole registry dir via fcntl.
PoC-grade single-host concurrency control. Multiple herdr sessions / scripts
serialise their read-modify-write of job records through this lock so two
sessions never claim the same pending job. For multi-host delegation move
to SQLite WAL (see references/registry.md)."""
import fcntl # POSIX only; imported lazily so import works on Windows.
Path(registry_dir).mkdir(parents=True, exist_ok=True)
lock_file = _lock_path(registry_dir)
fh = open(lock_file, "a+")
try:
fcntl.flock(fh.fileno(), fcntl.LOCK_EX)
yield
finally:
try:
fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
finally:
fh.close()
def load_job(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> Dict[str, Any]:
"""Load and parse a job record. Raises FileNotFoundError if absent."""
path = _job_path(job_id, registry_dir)
if not path.exists():
raise FileNotFoundError(f"job record not found: {path}")
with open(path, "r", encoding="utf-8") as fh:
return json.load(fh)
def _atomic_write_record(job_id: str, registry_dir: str, record: Dict[str, Any]) -> None:
"""Write a record atomically: temp file in the same dir + os.replace.
The rename is atomic on POSIX, so readers never observe a half-written
file. Callers MUST already hold ``registry_lock`` for read-modify-write
correctness."""
Path(registry_dir).mkdir(parents=True, exist_ok=True)
path = _job_path(job_id, registry_dir)
fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=f".{job_id}.", suffix=".tmp")
try:
with os.fdopen(fd, "w", encoding="utf-8") as fh:
json.dump(record, fh, ensure_ascii=False, indent=2)
fh.write("\n")
fh.flush()
os.fsync(fh.fileno())
os.replace(tmp, path)
try:
os.chmod(path, 0o600)
except Exception:
pass
except BaseException:
if os.path.exists(tmp):
os.unlink(tmp)
raise
def update_job_status(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR, **fields: Any) -> Dict[str, Any]:
"""Atomically merge ``fields`` into a job record under the registry lock.
Always refreshes ``updated_at``. Returns the new record. Raises
FileNotFoundError if the job does not exist.
This is the single chokepoint for status writes (both ``registry.update_status``
and ``publish_event.py``'s status sync route through here), so it also mirrors
any ``status`` change into the persistent audit log. We perform the log mirror
under the lock to guarantee sequential consistency in audit history."""
with registry_lock(registry_dir):
record = load_job(job_id, registry_dir)
old_status = record.get("status")
record.update(fields)
record["updated_at"] = _utcnow()
_atomic_write_record(job_id, registry_dir, record)
if "status" in fields:
new_status = record.get("status")
update_logged_status(job_id, new_status, updated_at=record["updated_at"])
if old_status != new_status:
append_event(job_id, {
"event": "status_changed",
"from": old_status,
"to": new_status,
"timestamp": record["updated_at"],
})
return record
def next_seq(job_id: str, registry_dir: str = DEFAULT_REGISTRY_DIR) -> int:
"""Return the next monotonic sequence number for a job, persisted in the
record's ``last_seq`` field so it stays consistent across process restarts.
First call returns 1."""
with registry_lock(registry_dir):
record = load_job(job_id, registry_dir)
seq = int(record.get("last_seq", 0)) + 1
record["last_seq"] = seq
record["updated_at"] = _utcnow()
_atomic_write_record(job_id, registry_dir, record)
return seq
def _utcnow() -> str:
"""ISO-8601 UTC timestamp with trailing Z (payload `timestamp` field)."""
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
def _utcnow_precise() -> str:
"""ISO-8601 UTC timestamp with millisecond resolution. Used for the audit
log's ``logged_at`` so events sort cleanly even within the same second."""
now = time.time()
base = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(now))
return f"{base}.{int((now % 1) * 1000):03d}Z"
# --------------------------------------------------------------------------
# Persistent audit log (.mam/delegate_job_logs/<job_id>/...)
#
# Every function here is idempotent, concurrency-safe, and *best-effort*: a
# logging failure is swallowed with a logger.warning and never propagated, so it
# can never break a publish, a subscribe, or a registry write. stdout is never
# touched (it is reserved for data output).
# --------------------------------------------------------------------------
def job_log_dir(job_id: str, logs_dir: Optional[str] = None) -> Path:
return Path(logs_dir or get_logs_dir()) / job_id
def job_log_path(job_id: str, kind: str, logs_dir: Optional[str] = None) -> Path:
"""Path to one audit-log file for a job. ``kind`` is a filename, e.g. the
module constants META_FILENAME / EVENTS_FILENAME / STATUS_FILENAME."""
return job_log_dir(job_id, logs_dir) / kind
@contextmanager
def _file_lock(fh):
"""Best-effort exclusive lock over a single open file via fcntl, so two
processes appending to events.ndjson never interleave a line. A no-op where
fcntl is unavailable (Windows); a short append is atomic enough there."""
try:
import fcntl
except ImportError: # pragma: no cover - non-POSIX
yield
return
fcntl.flock(fh.fileno(), fcntl.LOCK_EX)
try:
yield
finally:
fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
def _redact_dict(d: Any) -> Any:
"""Recursively mask sensitive values (passwords, secrets, tokens) inside logs."""
if isinstance(d, dict):
redacted = {}
for k, v in d.items():
if any(s in k.lower() for s in ("password", "token", "secret", "auth_token", "key")):
redacted[k] = "[REDACTED]"
else:
redacted[k] = _redact_dict(v)
return redacted
elif isinstance(d, list):
return [_redact_dict(item) for item in d]
return d
def append_event(job_id: str, event_dict: Dict[str, Any], logs_dir: Optional[str] = None) -> None:
"""Append one event as a JSON line to ``<logs>/<job_id>/events.ndjson``.
Concurrency-safe (fcntl lock over the file) and best-effort. A millisecond
``logged_at`` is stamped when the caller did not supply one."""
try:
path = job_log_path(job_id, EVENTS_FILENAME, logs_dir)
path.parent.mkdir(parents=True, exist_ok=True)
record = _redact_dict(dict(event_dict))
record.setdefault("logged_at", _utcnow_precise())
line = json.dumps(record, ensure_ascii=False) + "\n"
with open(path, "a", encoding="utf-8") as fh:
with _file_lock(fh):
fh.write(line)
fh.flush()
except Exception as exc: # pragma: no cover - best effort
logger.warning("append_event failed for job %s: %s", job_id, exc)
def update_logged_status(job_id: str, status: str, logs_dir: Optional[str] = None, **extras: Any) -> None:
"""Rewrite ``<logs>/<job_id>/status.json`` (current status for fast point
queries) atomically. Best-effort; merges any ``extras``."""
try:
path = job_log_path(job_id, STATUS_FILENAME, logs_dir)
path.parent.mkdir(parents=True, exist_ok=True)
record: Dict[str, Any] = {"job_id": job_id, "status": status, "updated_at": _utcnow()}
record.update(extras)
tmp = path.with_name(path.name + ".tmp")
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(record, fh, ensure_ascii=False, indent=2)
fh.write("\n")
os.replace(tmp, path)
except Exception as exc: # pragma: no cover - best effort
logger.warning("update_logged_status failed for job %s: %s", job_id, exc)
def init_job_log(job_id: str, meta: Dict[str, Any], logs_dir: Optional[str] = None) -> None:
"""Seed the per-job audit-log dir: write meta.json, status.json, and a first
``registered`` line in events.ndjson. Idempotent (the ``registered`` line is
written only when events.ndjson does not yet exist) and best-effort."""
try:
d = job_log_dir(job_id, logs_dir)
d.mkdir(parents=True, exist_ok=True)
meta_redacted = _redact_dict(meta)
with open(d / META_FILENAME, "w", encoding="utf-8") as fh:
json.dump(meta_redacted, fh, ensure_ascii=False, indent=2)
fh.write("\n")
status = meta.get("status", "pending")
update_logged_status(
job_id, status, logs_dir=logs_dir,
created_at=meta.get("created_at"), prompt=meta.get("prompt"),
)
events_path = d / EVENTS_FILENAME
first_time = not events_path.exists()
events_path.touch(exist_ok=True)
if first_time:
append_event(job_id, {
"event": "registered",
"status": status,
"agent": meta.get("agent"),
"agent_session": meta.get("agent_session"),
"topic_prefix": meta.get("topic_prefix"),
"timestamp": meta.get("created_at"),
}, logs_dir=logs_dir)
except Exception as exc: # pragma: no cover - best effort
logger.warning("init_job_log failed for job %s: %s", job_id, exc)
def read_logged_meta(job_id: str, logs_dir: Optional[str] = None) -> Optional[Dict[str, Any]]:
"""Return a job's audit meta.json (registration snapshot), or None."""
try:
with open(job_log_path(job_id, META_FILENAME, logs_dir), "r", encoding="utf-8") as fh:
return json.load(fh)
except (OSError, json.JSONDecodeError):
return None
def read_logged_status(job_id: str, logs_dir: Optional[str] = None) -> Optional[Dict[str, Any]]:
"""Return a job's current status.json, or None. This is the fast point-query
file (current status only), separate from the registration-time meta.json."""
try:
with open(job_log_path(job_id, STATUS_FILENAME, logs_dir), "r", encoding="utf-8") as fh:
return json.load(fh)
except (OSError, json.JSONDecodeError):
return None
def iter_logged_events(job_id: str, logs_dir: Optional[str] = None):
"""Yield each parsed event from a job's events.ndjson in file (time) order.
Malformed lines are skipped with a warning."""
path = job_log_path(job_id, EVENTS_FILENAME, logs_dir)
if not path.exists():
return
with open(path, "r", encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if not line:
continue
try:
yield json.loads(line)
except json.JSONDecodeError:
logger.warning("skipping malformed audit line in %s", path)
def list_logged_jobs(logs_dir: Optional[str] = None) -> List[Dict[str, Any]]:
"""Return one meta record per job directory under the logs root, oldest
first. Falls back to ``{"job_id": <dir>}`` when meta.json is missing."""
base = Path(logs_dir or get_logs_dir())
out: List[Dict[str, Any]] = []
if not base.exists():
return out
for d in sorted(base.iterdir()):
if not d.is_dir():
continue
meta = read_logged_meta(d.name, logs_dir) or {"job_id": d.name}
# Overlay the live status.json so the summary reflects current state, not
# the registration-time snapshot frozen in meta.json.
status = read_logged_status(d.name, logs_dir)
if status:
meta = {**meta,
"status": status.get("status", meta.get("status")),
"updated_at": status.get("updated_at", meta.get("updated_at"))}
out.append(meta)
out.sort(key=lambda m: m.get("created_at") or "")
return out
# --------------------------------------------------------------------------
# Retry helper
# --------------------------------------------------------------------------
def with_retry(
fn: Optional[Callable] = None,
*,
attempts: int = 3,
base_delay: float = 0.5,
factor: float = 2.0,
max_delay: float = 8.0,
exceptions: Iterable[type] = (Exception,),
) -> Callable:
"""Retry ``fn`` with exponential backoff.
Usable two ways::
result = with_retry(do_publish, attempts=3)() # wrap-and-call
@with_retry(attempts=5, base_delay=1.0) # decorator
def do_publish(): ...
Re-raises the last exception once ``attempts`` is exhausted.
"""
exc_tuple = tuple(exceptions)
def decorate(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
delay = base_delay
last_exc: Optional[BaseException] = None
for attempt in range(1, attempts + 1):
try:
return func(*args, **kwargs)
except exc_tuple as exc:
last_exc = exc
if attempt >= attempts:
break
logger.warning(
"attempt %d/%d failed: %s; retrying in %.1fs",
attempt, attempts, exc, delay,
)
time.sleep(delay)
delay = min(delay * factor, max_delay)
assert last_exc is not None
raise last_exc
return wrapper
if fn is not None:
return decorate(fn)
return decorate
def setup_logging(level: int = logging.WARNING) -> None:
"""Configure root logging to stderr. stdout is reserved for data output
(subscriber event lines, registry ids)."""
import sys
logging.basicConfig(
level=level,
stream=sys.stderr,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
import lib_py # parent resolved normally; shims never rebind it
except ModuleNotFoundError:
sys.path.append(_SKILLS) # standalone only: no lib_py exists anywhere to shadow
import lib_py
_d = os.path.join(_SKILLS, "lib_py", "delegation")
_spec = importlib.util.spec_from_file_location(
_PKG, os.path.join(_d, "__init__.py"), submodule_search_locations=[_d])
_pkg = importlib.util.module_from_spec(_spec)
sys.modules[_PKG] = _pkg # register BEFORE exec so __init__'s own imports resolve here
_spec.loader.exec_module(_pkg)
lib_py.delegation = _pkg
_impl = importlib.import_module(_PKG + ".mqtt_common") # resolved through the bound package's __path__
if __name__ != "__main__":
sys.modules[__name__] = _impl # `import mqtt_common` is lib_py.delegation.mqtt_common
else:
_main = getattr(_impl, "main", None) # a module without a CLI keeps exit status 0
sys.exit(_main() if _main is not None else 0)
@@ -1,239 +1,37 @@
#!/usr/bin/env python3
"""publish_event.py — the single entry point for emitting a Job event.
"""Compatibility shim -- the implementation lives in lib_py.delegation.publish_event.
Loads the job record from the registry, resolves its broker, assigns the next
monotonic ``seq``, builds the schema-v1 JSON payload, and publishes it to
``<topic_prefix>/events`` over QoS 1 with exponential-backoff retry.
This file's path is a wire contract (agent instructions, the bash wrapper,
SKILL.md, BOOTSTRAP.md, deploy/install.sh, tests). It binds the
``lib_py.delegation`` *package* to the tree next to this file, once per process,
without touching ``sys.path``; ``import publish_event`` then yields the very same module
object as ``lib_py.delegation.publish_event``.
Silent by design: nothing is printed to stdout. Diagnostics go to stderr via
logging. Terminal events (``completed``/``error``) publish with retain=True so
a late subscriber still observes the final state (production hardening).
Exit codes:
0 published successfully
1 parameter / registry error (bad args, unknown job, no pending job)
2 publish failed after retries (network / broker / ACK timeout)
Usage:
publish_event.py --job <id> --event started [--detail "..."] [--data '{...}']
publish_event.py --pick-pending --agent-session herdr:claude --event completed
publish_event.py --job <id> --event completed --retained
Why the package and not this one module: a per-module bind leaves a non-leaf
module (one that imports its siblings) importing dependencies from whichever tree
happens to be on the path, mixing a frozen copy with a live one. Why bind-once and
never rebind: replacing a bound package would split existing references and
break monkeypatching. See docs/contracts/issue-5/ (D2) for the context matrix.
"""
from __future__ import annotations
import argparse
import hashlib
import hmac
import json
import logging
import sys
import time
from typing import Any, Dict, Optional
import mqtt_common
import registry
from mqtt_common import (
DEFAULT_REGISTRY_DIR,
SCHEMA_VERSION,
broker_config_from_job,
events_topic_for,
load_job,
make_client,
next_seq,
with_retry,
)
logger = logging.getLogger("delegate_job.publish_event")
VALID_EVENTS = ("started", "permission_required", "progress", "completed", "error", "cancelled")
TERMINAL_EVENTS = ("completed", "error", "cancelled")
# event -> registry status to sync as a best-effort side effect
EVENT_TO_STATUS = {
"started": "running",
"completed": "completed",
"error": "error",
"cancelled": "cancelled",
}
CONNECT_ACK_TIMEOUT = 10 # seconds to wait for CONNACK
PUBLISH_ACK_TIMEOUT = 5 # seconds to wait for QoS-1 PUBACK
def build_payload(
job_id: str,
seq: int,
event: str,
detail: str,
data: Optional[Dict[str, Any]],
auth_token: Optional[str],
) -> Dict[str, Any]:
payload: Dict[str, Any] = {
"schema_version": SCHEMA_VERSION,
"seq": seq,
"job_id": job_id,
"event": event,
"timestamp": mqtt_common._utcnow(),
"detail": detail,
"data": dict(data) if data else {},
}
# Production: carry the per-job HMAC-SHA256 signature in `data.hmac_sig` so
# the subscriber can verify the publisher without exposing the secret token.
# The signature is calculated over the entire payload (with `data.hmac_sig` excluded).
if auth_token:
sign_payload = {k: v for k, v in payload.items() if k != "data"}
sign_payload["data"] = {k: v for k, v in payload.get("data", {}).items() if k != "hmac_sig"}
msg = json.dumps(sign_payload, sort_keys=True, separators=(",", ":")).encode()
sig = hmac.new(auth_token.encode(), msg, hashlib.sha256).hexdigest()
payload["data"]["hmac_sig"] = sig
return payload
def _publish_once(config, topic: str, body: bytes, retain: bool) -> None:
"""Connect, publish one QoS-1 message, wait for the broker ACK, disconnect.
Raises on any failure so ``with_retry`` can re-run the whole sequence (a
fresh connection per attempt is the robust choice for a PoC)."""
client = make_client("publisher", config)
connected = {"rc": None}
def on_connect(_c, _u, _flags, reason_code, _props):
connected["rc"] = reason_code
client.on_connect = on_connect
client.connect(config.host, config.port, config.keepalive)
client.loop_start()
import importlib, importlib.util, os, sys
_SKILLS = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
_PKG = "lib_py.delegation"
if _PKG not in sys.modules: # bind once per process; never rebind
try:
# Wait for CONNACK so we fail fast on auth/TLS errors.
deadline = time.monotonic() + CONNECT_ACK_TIMEOUT
while connected["rc"] is None and time.monotonic() < deadline:
time.sleep(0.05)
if connected["rc"] is None:
raise TimeoutError("no CONNACK from broker")
if mqtt_common.reason_code_value(connected["rc"]) != 0:
raise ConnectionError(f"broker refused connection: rc={connected['rc']}")
info = client.publish(topic, payload=body, qos=1, retain=retain)
info.wait_for_publish(timeout=PUBLISH_ACK_TIMEOUT)
if not info.is_published():
raise TimeoutError("publish not acknowledged within timeout")
finally:
client.loop_stop()
try:
client.disconnect()
except Exception: # pragma: no cover - disconnect best effort
pass
def _resolve_job_id(args) -> Optional[str]:
if args.pick_pending:
return registry.pick_pending(args.agent_session, args.registry_dir)
return args.job
def main(argv=None) -> int:
parser = argparse.ArgumentParser(description="Publish a Job event to MQTT")
target = parser.add_mutually_exclusive_group(required=True)
target.add_argument("--job", help="job id to publish for")
target.add_argument("--pick-pending", action="store_true",
help="auto-select a pending job for --agent-session")
parser.add_argument("--agent-session", default="herdr:claude",
help="session label used with --pick-pending")
parser.add_argument("--event", default="progress", choices=VALID_EVENTS)
parser.add_argument("--detail", default="")
parser.add_argument("--data", default=None, help="optional JSON object string")
parser.add_argument("--retained", action="store_true",
help="force retain=True (auto for completed/error)")
parser.add_argument("--registry-dir", default=DEFAULT_REGISTRY_DIR)
parser.add_argument("--attempts", type=int, default=3)
parser.add_argument("-v", "--verbose", action="store_true")
args = parser.parse_args(argv)
mqtt_common.setup_logging(logging.DEBUG if args.verbose else logging.WARNING)
# --- parse optional data JSON (parameter error -> exit 1) ---
data: Optional[Dict[str, Any]] = None
if args.data:
try:
data = json.loads(args.data)
if not isinstance(data, dict):
raise ValueError("--data must be a JSON object")
except (ValueError, json.JSONDecodeError) as exc:
logger.error("invalid --data: %s", exc)
return 1
job_id = _resolve_job_id(args)
if not job_id:
logger.error("no job to publish for (unknown --job or no pending job)")
return 1
try:
job = load_job(job_id, args.registry_dir)
except FileNotFoundError as exc:
logger.error("%s", exc)
return 1
config = broker_config_from_job(job)
topic = job.get("topic_prefix")
topic = f"{topic}/events" if topic else events_topic_for(job_id)
seq = next_seq(job_id, args.registry_dir)
payload = build_payload(
job_id=job_id,
seq=seq,
event=args.event,
detail=args.detail,
data=data,
auth_token=job.get("auth_token"),
)
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
retain = args.retained or args.event in TERMINAL_EVENTS
publish = with_retry(
_publish_once,
attempts=args.attempts,
exceptions=(OSError, TimeoutError, ConnectionError, ValueError),
)
publish_ok = True
publish_error: Optional[str] = None
try:
publish(config, topic, body, retain)
except Exception as exc:
publish_ok = False
publish_error = str(exc)
logger.error("publish failed after %d attempts: %s", args.attempts, exc)
# Persistent audit log: record the exact payload we put on the wire (or intended to).
# Best-effort (isolated inside append_event) — never fails the publish.
# Policy Note: Seq consumption on failure is intentionally maintained for monotonic replay
# defense (> highest accepted seq), with failure documented explicitly in the audit record.
mqtt_common.append_event(job_id, {
"event": "published",
"source_event": args.event,
"seq": seq,
"topic": topic,
"retain": retain,
"timestamp": payload["timestamp"],
"detail": args.detail,
"payload": payload,
"published": publish_ok,
"publish_error": publish_error,
})
# Best-effort side effects: registry status sync + (debug) event log. Never
# fail the publish on these.
registry.append_event(job_id, args.registry_dir, payload)
new_status = EVENT_TO_STATUS.get(args.event)
if new_status:
try:
mqtt_common.update_job_status(job_id, args.registry_dir, status=new_status)
except Exception as exc: # pragma: no cover - best effort
logger.warning("status sync failed: %s", exc)
if not publish_ok:
return 2
logger.info("published %s seq=%d job=%s retain=%s", args.event, seq, job_id, retain)
return 0
if __name__ == "__main__":
sys.exit(main())
import lib_py # parent resolved normally; shims never rebind it
except ModuleNotFoundError:
sys.path.append(_SKILLS) # standalone only: no lib_py exists anywhere to shadow
import lib_py
_d = os.path.join(_SKILLS, "lib_py", "delegation")
_spec = importlib.util.spec_from_file_location(
_PKG, os.path.join(_d, "__init__.py"), submodule_search_locations=[_d])
_pkg = importlib.util.module_from_spec(_spec)
sys.modules[_PKG] = _pkg # register BEFORE exec so __init__'s own imports resolve here
_spec.loader.exec_module(_pkg)
lib_py.delegation = _pkg
_impl = importlib.import_module(_PKG + ".publish_event") # resolved through the bound package's __path__
if __name__ != "__main__":
sys.modules[__name__] = _impl # `import publish_event` is lib_py.delegation.publish_event
else:
_main = getattr(_impl, "main", None) # a module without a CLI keeps exit status 0
sys.exit(_main() if _main is not None else 0)
@@ -1,426 +1,37 @@
"""Job registry for the multi-agent-mux-delegate-job skill.
#!/usr/bin/env python3
"""Compatibility shim -- the implementation lives in lib_py.delegation.registry.
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``.
This file's path is a wire contract (agent instructions, the bash wrapper,
SKILL.md, BOOTSTRAP.md, deploy/install.sh, tests). It binds the
``lib_py.delegation`` *package* to the tree next to this file, once per process,
without touching ``sys.path``; ``import registry`` then yields the very same module
object as ``lib_py.delegation.registry``.
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.
Why the package and not this one module: a per-module bind leaves a non-leaf
module (one that imports its siblings) importing dependencies from whichever tree
happens to be on the path, mixing a frozen copy with a live one. Why bind-once and
never rebind: replacing a bound package would split existing references and
break monkeypatching. See docs/contracts/issue-5/ (D2) for the context matrix.
"""
from __future__ import annotations
import argparse
import json
import logging
import sys
import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional
import mqtt_common
from mqtt_common import (
DEFAULT_REGISTRY_DIR,
SCHEMA_VERSION,
_atomic_write_record,
_utcnow,
broker_config_from_env,
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]
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: int = 120,
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,
) -> 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.
"""
job_id = job_id or generate_job_id(bits)
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,
}
with registry_lock(registry_dir):
if mqtt_common._job_path(job_id, registry_dir).exists():
raise FileExistsError(f"job already exists: {job_id}")
_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)
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."""
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"]
# pick_pending writes the record directly (not via update_job_status), so it
# mirrors the pending->running transition into the audit log here. Best-effort.
mqtt_common.update_logged_status(job_id, "running", updated_at=updated_at)
mqtt_common.append_event(job_id, {
"event": "status_changed",
"from": "pending",
"to": "running",
"by": agent_session,
"timestamp": updated_at,
})
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 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."""
import importlib, importlib.util, os, sys
_SKILLS = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
_PKG = "lib_py.delegation"
if _PKG not in sys.modules: # bind once per process; never rebind
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",
]
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) -> str:
"""Read the job's audit log or events log and return the detail of the last completed/error event."""
# 1) Try the unified audit log first (ndjson) since it's written synchronously by the subscriber
try:
import mqtt_common
logs_dir = mqtt_common.get_logs_dir()
events = list(mqtt_common.iter_logged_events(job_id, logs_dir))
for e in reversed(events):
if e.get("source_event") in ("completed", "error"):
return e.get("detail", "")
if e.get("event") in ("completed", "error"):
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 ("completed", "error"):
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")
p_reg.add_argument("--prompt", required=True)
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=120)
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_list = sub.add_parser("list", help="list jobs (optionally by status)")
p_list.add_argument("--status", default=None)
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)
p_update.add_argument("--prompt", default=None)
p_update.add_argument("--iteration", type=int, default=None)
p_update.add_argument("--role", 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 <cwd>/.mam/delegate_job_logs)")
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")
return parser
def main(argv: Optional[List[str]] = None) -> int:
mqtt_common.setup_logging(logging.INFO)
args = _build_parser().parse_args(argv)
rd = args.registry_dir
if args.command == "register":
job_id = register_job(
prompt=args.prompt,
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,
)
print(job_id)
return 0
if args.command == "list":
records = list_jobs(rd, status=args.status)
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:
update_status(args.job, rd, args.status)
except (FileNotFoundError, ValueError) as exc:
print(str(exc), file=sys.stderr)
return 1
return 0
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
if args.prompt is not None:
fields["prompt"] = args.prompt
if args.iteration is not None:
fields["iteration"] = args.iteration
if args.role is not None:
fields["role"] = args.role
try:
mqtt_common.update_job_status(args.job, rd, **fields)
except FileNotFoundError as exc:
print(str(exc), file=sys.stderr)
return 1
return 0
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)
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()
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())
import lib_py # parent resolved normally; shims never rebind it
except ModuleNotFoundError:
sys.path.append(_SKILLS) # standalone only: no lib_py exists anywhere to shadow
import lib_py
_d = os.path.join(_SKILLS, "lib_py", "delegation")
_spec = importlib.util.spec_from_file_location(
_PKG, os.path.join(_d, "__init__.py"), submodule_search_locations=[_d])
_pkg = importlib.util.module_from_spec(_spec)
sys.modules[_PKG] = _pkg # register BEFORE exec so __init__'s own imports resolve here
_spec.loader.exec_module(_pkg)
lib_py.delegation = _pkg
_impl = importlib.import_module(_PKG + ".registry") # resolved through the bound package's __path__
if __name__ != "__main__":
sys.modules[__name__] = _impl # `import registry` is lib_py.delegation.registry
else:
_main = getattr(_impl, "main", None) # a module without a CLI keeps exit status 0
sys.exit(_main() if _main is not None else 0)
@@ -116,6 +116,7 @@ else:
import mqtt_common
import registry
mqtt_common.ensure_env_loaded() # D6: env now loads at entry points, not at import
# Executed INSIDE lib.sh::atomic_dump_yaml (system python3 + PyYAML), under the
# YAML flock with schema-validate + .bak (review item 5). Marks matching running
+126
View File
@@ -0,0 +1,126 @@
#!/usr/bin/env python3
"""Shim context harness (PV-4, issue #5 phase 1).
Runs ONE import scenario in a fresh interpreter and prints a JSON report of
which tree every relevant module resolved to. Kept out of the test module on
purpose: each scenario needs a virgin ``sys.modules``, so it must be a
subprocess.
Usage: shim_contexts.py <scenario> <real> <sandbox> <sandbox2> <freeze>
Tree arguments are skills-directory roots, each containing ``lib_py/`` and
``multi-agent-mux-delegate-job/scripts/``. The scenarios mirror the four real
execution contexts: pytest in-process (A/A_pe/B/B2/D/E), standalone with no
ambient tree (C1/C1_pe), and a frozen copy invoked while PYTHONPATH still points
at the live tree (C2/C2_pe) -- the shape every delegated agent job actually runs.
"""
import json
import os
import sys
SCENARIO = sys.argv[1]
TREES = {
"real": os.path.abspath(sys.argv[2]),
"sandbox": os.path.abspath(sys.argv[3]),
"sandbox2": os.path.abspath(sys.argv[4]),
"freeze": os.path.abspath(sys.argv[5]),
}
def scripts(tree):
return os.path.join(TREES[tree], "multi-agent-mux-delegate-job", "scripts")
def which_tree(path):
if not path:
return None
path = os.path.abspath(path)
for name, root in TREES.items():
if path == root or path.startswith(root + os.sep):
return name
return "OUTSIDE:" + path
def report():
out = {"scenario": SCENARIO}
mods = sys.modules
def note(key, mod, attr="__file__"):
m = mods.get(mod)
if m is None:
out[key] = None
return
if attr == "__path__":
out[key] = which_tree(list(getattr(m, "__path__", [None]))[0])
else:
out[key] = which_tree(getattr(m, attr, None))
note("alias_mqtt_common", "mqtt_common")
note("alias_publish_event", "publish_event")
note("pkg_delegation", "lib_py.delegation", "__path__")
note("pkg_mqtt_common", "lib_py.delegation.mqtt_common")
note("pkg_publish_event", "lib_py.delegation.publish_event")
note("lib_py", "lib_py", "__path__")
note("lib_py_agents", "lib_py.agents", "__path__")
# identity: the alias and the package attribute must be one object
ident = {}
for name in ("mqtt_common", "publish_event"):
alias = mods.get(name)
pkg_mod = mods.get("lib_py.delegation." + name)
ident[name] = None if alias is None else (alias is pkg_mod)
out["identity"] = ident
# PV-3 shape: sys.path entries that would contribute a lib_py package
out["lib_py_path_entries"] = [
which_tree(e) for e in sys.path
if e and os.path.isdir(os.path.join(e, "lib_py"))
]
return out
if SCENARIO in ("A", "A_pe", "B", "B2", "D", "E"):
sys.path.insert(0, TREES["real"]) # models conftest.py inserting the real tree
if SCENARIO == "A":
sys.path.insert(0, scripts("sandbox"))
import mqtt_common, publish_event # noqa: F401
elif SCENARIO == "A_pe":
sys.path.insert(0, scripts("sandbox"))
import publish_event # noqa: F401
elif SCENARIO == "B":
import lib_py.agents # noqa: F401
sys.path.insert(0, scripts("sandbox"))
import mqtt_common, publish_event # noqa: F401
elif SCENARIO == "B2":
import lib_py.delegation.mqtt_common # noqa: F401
sys.path.insert(0, scripts("sandbox"))
import mqtt_common, publish_event # noqa: F401
elif SCENARIO == "D":
sys.path.insert(0, scripts("sandbox"))
import mqtt_common # noqa: F401
sys.path.insert(0, scripts("sandbox2"))
import publish_event # noqa: F401
elif SCENARIO == "E":
sys.path.insert(0, scripts("sandbox"))
import mqtt_common # noqa: F401
from lib_py.delegation import publish_event # noqa: F401
elif SCENARIO == "C1":
sys.path.insert(0, scripts("freeze"))
import mqtt_common, publish_event # noqa: F401
elif SCENARIO == "C1_pe":
sys.path.insert(0, scripts("freeze"))
import publish_event # noqa: F401
elif SCENARIO == "C2":
sys.path.insert(0, scripts("freeze")) # PYTHONPATH=real set by caller
import mqtt_common, publish_event # noqa: F401
elif SCENARIO == "C2_pe":
sys.path.insert(0, scripts("freeze")) # PYTHONPATH=real set by caller
import publish_event # noqa: F401
elif SCENARIO == "K3":
sys.path.insert(0, scripts("real"))
import mqtt_common, publish_event # noqa: F401
else:
raise SystemExit("unknown scenario: " + SCENARIO)
print(json.dumps(report()))
+217
View File
@@ -17,6 +17,7 @@ Sections
"""
from __future__ import annotations
import json
import os
import re
import signal
@@ -390,3 +391,219 @@ def test_reconcile_terminal_mutation_literals_are_frozen():
"if _event in ('completed', 'cancelled')",
):
assert literal in text, literal
# ==============================================================================
# 4. P1-01 — package binding, provenance and context matrix (PV-1, PV-4, PV-5)
# ==============================================================================
DELEGATION_MODULES = ("mqtt_common", "registry", "publish_event", "job_subscriber")
PKG_DIR = SKILLS / "lib_py" / "delegation"
SHIM_DIR = SKILLS / "multi-agent-mux-delegate-job" / "scripts"
def test_pv5_all_four_shims_share_one_bootstrap_block():
"""PV-5: the four shims differ only by the module name they bind."""
normalised = {}
for mod in DELEGATION_MODULES:
text = (SHIM_DIR / (mod + ".py")).read_text(encoding="utf-8")
normalised[mod] = text.replace(mod, "{mod}")
base = normalised["mqtt_common"]
for mod, text in normalised.items():
assert text == base, "shim {} diverged from the shared bootstrap".format(mod)
assert "sys.path.insert" not in base, "the shim must never take sys.path precedence"
assert "sys.path.append(_SKILLS)" in base, "standalone fallback must remain"
def test_pv1_identity_and_provenance_of_every_alias(mam_sandbox):
"""PV-1: each alias IS the package module, and all four share one tree."""
import lib_py.delegation
paths = set()
for mod in DELEGATION_MODULES:
alias = _import_delegation(mam_sandbox, mod)
pkg_mod = getattr(lib_py.delegation, mod, None)
if pkg_mod is None:
import importlib
pkg_mod = importlib.import_module("lib_py.delegation." + mod)
assert alias is pkg_mod, "{}: alias is not the package module".format(mod)
paths.add(str(Path(alias.__file__).resolve().parent))
assert len(paths) == 1, "delegation modules span multiple trees: {}".format(paths)
bound = Path(paths.pop())
real = PKG_DIR.resolve()
basetemp_ok = "pytest-" in str(bound)
assert bound == real or basetemp_ok, (
"delegation bound to an unexpected tree: {}".format(bound)
)
# the parent package must never be dragged along with it
assert Path(lib_py.__file__).resolve().parent == (SKILLS / "lib_py").resolve()
def test_delegation_package_imports_no_sibling_lib_py_modules():
"""D1(b): a frozen delegation package must not reach into the parent tree."""
import ast
offenders = []
for path in sorted(PKG_DIR.glob("*.py")):
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
for node in ast.walk(tree):
names = []
if isinstance(node, ast.Import):
names = [a.name for a in node.names]
elif isinstance(node, ast.ImportFrom) and node.module:
names = [node.module]
for name in names:
if name in DELEGATION_MODULES:
offenders.append("{}:{} top-level `{}`".format(path.name, node.lineno, name))
if name == "lib_py" or (
name.startswith("lib_py.") and not name.startswith("lib_py.delegation")
):
offenders.append("{}:{} sibling `{}`".format(path.name, node.lineno, name))
assert not offenders, "forbidden imports inside lib_py.delegation:\n " + "\n ".join(offenders)
def test_importing_delegation_does_not_mutate_environment(mam_sandbox):
"""D6: importing the library must leave os.environ alone; env loads at entry points."""
import importlib
mqtt_common = _import_delegation(mam_sandbox, "mqtt_common")
before = set(os.environ)
importlib.reload(mqtt_common)
assert set(os.environ) == before, "library import injected environment variables"
# idempotent: the second call is a no-op
mqtt_common.ensure_env_loaded()
keys = set(os.environ)
mqtt_common.ensure_env_loaded()
assert set(os.environ) == keys
def test_shim_help_output_matches_pre_move_capture():
"""The shim must be indistinguishable from the pre-move script at the CLI."""
for mod in ("publish_event", "registry", "job_subscriber"):
golden = (FIXTURES / "help-{}.txt".format(mod)).read_text(encoding="utf-8")
res = subprocess.run(
[sys.executable, str(SHIM_DIR / (mod + ".py")), "--help"],
capture_output=True, text=True, timeout=60,
)
assert res.returncode == 0, res.stderr
assert res.stdout + res.stderr == golden, "--help drifted for {}".format(mod)
# --- PV-4: the context matrix, each scenario in a fresh interpreter ------------
@pytest.fixture(scope="module")
def shim_trees(tmp_path_factory):
"""Four copies of the live skills tree: real / sandbox / sandbox2 / freeze."""
import shutil
root = tmp_path_factory.mktemp("shim_trees")
for name in ("real", "sandbox", "sandbox2", "freeze"):
dest = root / name
(dest / "multi-agent-mux-delegate-job").mkdir(parents=True)
shutil.copytree(SKILLS / "lib_py", dest / "lib_py")
shutil.copytree(SHIM_DIR, dest / "multi-agent-mux-delegate-job" / "scripts")
return root
def _run_scenario(shim_trees, scenario, pythonpath=None):
env = dict(os.environ)
env.pop("PYTHONPATH", None)
if pythonpath:
env["PYTHONPATH"] = str(shim_trees / pythonpath)
res = subprocess.run(
[sys.executable, str(FIXTURES / "shim_contexts.py"), scenario]
+ [str(shim_trees / t) for t in ("real", "sandbox", "sandbox2", "freeze")],
capture_output=True, text=True, env=env, timeout=120,
)
assert res.returncode == 0, "scenario {} failed:\n{}".format(scenario, res.stderr)
return json.loads(res.stdout)
def _delegation_trees(report):
keys = ("alias_mqtt_common", "alias_publish_event", "pkg_delegation",
"pkg_mqtt_common", "pkg_publish_event")
return {report[k] for k in keys if report[k] is not None}
@pytest.mark.parametrize("scenario,expected_delegation", [
("A", "sandbox"), # sandbox shim binds first
("A_pe", "sandbox"), # non-leaf module alone: dependencies must follow it
("B", "sandbox"), # real lib_py.agents already imported: delegation still binds to shim tree
("B2", "real"), # real delegation already bound: bind-once keeps it
("D", "sandbox"), # a second sandbox must not rebind
("E", "sandbox"), # direct package import after a shim resolves to the bound tree
])
def test_pv4_pytest_family_binds_one_tree_and_spares_the_parent(
shim_trees, scenario, expected_delegation
):
rep = _run_scenario(shim_trees, scenario)
assert _delegation_trees(rep) == {expected_delegation}, rep
assert rep["lib_py"] == "real", "parent lib_py must stay on the real tree: {}".format(rep)
if rep["lib_py_agents"] is not None:
assert rep["lib_py_agents"] == "real", rep
for name, same in rep["identity"].items():
if same is not None:
assert same, "identity broken for {} in {}".format(name, scenario)
assert "sandbox" not in rep["lib_py_path_entries"], rep
assert "sandbox2" not in rep["lib_py_path_entries"], rep
@pytest.mark.parametrize("scenario", ["C1", "C1_pe"])
def test_pv4_standalone_bootstraps_from_its_own_tree(shim_trees, scenario):
"""K4: no ambient lib_py anywhere, so the shim may append its own tree."""
rep = _run_scenario(shim_trees, scenario)
assert _delegation_trees(rep) == {"freeze"}, rep
assert rep["lib_py"] == "freeze", rep
@pytest.mark.parametrize("scenario", ["C2", "C2_pe"])
def test_pv4_frozen_copy_wins_over_a_live_pythonpath(shim_trees, scenario):
"""K2: the shape every delegated job runs -- frozen script, live PYTHONPATH.
B-13 requires the frozen code to execute; the parent lib_py may stay live
because D1(b) forbids the delegation package from importing siblings."""
rep = _run_scenario(shim_trees, scenario, pythonpath="real")
assert _delegation_trees(rep) == {"freeze"}, rep
assert rep["lib_py"] == "real", rep
for name, same in rep["identity"].items():
if same is not None:
assert same, name
def test_pv4_k3_live_helpers_resolve_to_the_live_tree(shim_trees):
rep = _run_scenario(shim_trees, "K3")
assert _delegation_trees(rep) == {"real"}, rep
assert rep["lib_py"] == "real", rep
def test_k1_module_execution_follows_pythonpath(shim_trees):
"""K1: `python -m lib_py.delegation.<mod>` is resolved by PYTHONPATH, not by a shim."""
for tree in ("real", "freeze"):
env = dict(os.environ)
env["PYTHONPATH"] = str(shim_trees / tree)
res = subprocess.run(
[sys.executable, "-c",
"import lib_py.delegation as d; print(list(d.__path__)[0])"],
capture_output=True, text=True, env=env, timeout=60,
)
assert res.returncode == 0, res.stderr
assert res.stdout.strip() == str(shim_trees / tree / "lib_py" / "delegation"), res.stdout
def test_k6_subprocess_executes_the_copy_it_was_given(shim_trees):
"""K6: e2e-style subprocess execution runs the tree the path points at."""
env = dict(os.environ)
env.pop("PYTHONPATH", None)
shim = shim_trees / "sandbox" / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"
res = subprocess.run([sys.executable, str(shim), "--help"],
capture_output=True, text=True, env=env, timeout=60)
assert res.returncode == 0, res.stderr
probe = subprocess.run(
[sys.executable, "-c",
"import sys; sys.path.insert(0, {!r}); import publish_event;"
" print(publish_event.__file__)".format(str(shim.parent))],
capture_output=True, text=True, env=env, timeout=60,
)
assert probe.returncode == 0, probe.stderr
assert str(shim_trees / "sandbox") in probe.stdout, probe.stdout